FastMCP语音栈第二部分：用于语音智能体的LangGraph编排技术
第二部分：LangGraph状态、条件边、工具以及检查机制驱动FastMCP服务器实现从语音输入到语音输出的全流程，同时对比了LangChain与LangGraph的特点。
从MCP与语音输入输出到智能体循环
第一部分介绍了MCP、FastMCP工具服务器、语音转文本与文本转语音功能，以及一个无需大语言模型的语音演示。本部分则加入了LangChain和LangGraph，使助手能够进行推理、调用这些工具并保持短期记忆——同时尽可能使用免费的API套餐。
一页看懂LangChain与LangGraph的区别
LangChain将模型（OpenAI、Anthropic、Groq、Ollama等）与数据、工具及工作流相连。LangGraph则是用于以图形式表示的状态化、长时间运行、多步骤智能体的编排层。
一个实用的智能体必须具备思考、搜索、总结、编写文件和发送邮件的能力。这些功能分别对应一个节点，即一个Python函数。节点是点，而连接节点的线则代表关系。
a. Search the web for the latest AI news
b. Summarize it
c. Save it to a file
d. Email file
将这套组件安装到现有的第一部分环境中：
cmd>> uv pip install langgraph langchain langchain_groq python-dotenv
# if using OpenAI API use langchain_openai instead of langchain_groq
普通的LangChain辅助工具难以实现持久化内存、复杂的分支结构、重试机制以及需要人工干预的暂停功能。LangGraph则通过明确的状态和边来解决这些问题。
词汇表
图：整个工作流程。节点：一个任务函数。边：下一步要执行的内容。状态：所有节点均可读取和更新的共享数据。START / END：起始点和结束点。compile()用于构建可执行的图；invoke()则使用初始状态来运行该图。
问候语生成图
一个适合在笔记本中使用的简单示例：
#!/usr/bin/python3
from typing import TypedDict # we will use this to store our state
from langgraph.graph import StateGraph, START, END
# ---Step 1: Define the state Schema
#- State holds shared data passed between nodes
#- Each node updates the field in our case now it is message
class AgentState(TypedDict):
# holds our message---> read on pydantic to understand the state better
message: str
# ----Step 2: Create a node
# - Node a python function that performs one task
# - The node receives the current state we defined above, does some work, and returns the updated state
def greet(state: AgentState)-> AgentState:
"""This simple node adds a greeting message"""
state['message']='Hey ' + state['message'] +' , how are you doing?'
# updated state returned
return state
# ---Step 3: create graph schema
graph=StateGraph(AgentState)
# add the node to our graph
graph.add_node("greeter", greet)
# Set starting point
# Remember: To connect the nodes together we use edges
graph.add_edge(START, "greeter") # the START is where execution of the graph begins
# Set the endpoint
graph.add_edge("greeter", END)
# compile our graph to an executable
app=graph.compile()
#---Step 4: invoke to run the graph
# test our graph using the invoke keyword
result=app.invoke({"message":"Allan"})
# view the current updated graph state
print(f"Current Graph State:\n{result['message']}")
"""
===========Ignore this part: Is my result on CMD=======
cmd>> python greetings.py
Current Graph State:
Hey Allan , how are you doing?
"""
状态为何重要
想象一个多页的应用表单：后面的页面仍然可以查看之前的回答。LangGraph中的状态就是那个共享的数据容器——每个节点都会对其进行修改，而边则负责将更新结果传递下去。
Name: Allan
Age: 25
Country: Kenya
Course: Information Technology
条件边
真正的智能体并不会遵循单一路径。路由函数会检查当前状态并选择下一个节点——例如在矩形计算器中就会在区域与边界之间做出选择：
#!/usr/bin/python3
from typing import TypedDict
from langgraph.graph import StateGraph
# State Schema: Shared amoung all nodes
class AgentState(TypedDict):
length: int
width: int
area: int
perimeter: int
operation: str # we will use it in our router function
# Node--> Function that performs an action
def get_area(state: AgentState)-> AgentState:
"""Calculate the area"""
state['area']=state['length'] * state ['width']
return state
# Another Node for calculating perimeter
def get_perimeter(state: AgentState)-> AgentState:
"""Calculate the Perimeter"""
state['perimeter']=2*(state['length'] + state['width'])
return state
# Router---> Decides what function to run. Either get_area() or get_perimeter().
def decide_next_node(state: AgentState)-> AgentState:
"""Decide the next node to run"""
# u will see this when we invoke the graph changing the operation value in state.
if state['operation']=='a':
return "area_operation"
elif state['operation']=='p':
return "perimeter_operation"
# Graph Schema
graph=StateGraph(AgentState)
# add our two nodes
graph.add_node('area', get_area)
graph.add_node('perimeter', get_perimeter)
graph.add_node('router', lambda state:state) # pass through function. Initial state passes through here
# Below is a quicker way to set the start of the graph without using START being imported
graph.set_entry_point('router')
# Now add conditional edge--> Selects the node to run using our router(decide_next_node)
graph.add_conditional_edges(
'router', #the node we added
decide_next_node, # the function to run inorder to decide what to execute next
{
# the function decide_next_node returns strings that we match with the names given to our nodes
"area_operation":"area",
"perimeter_operation":"perimeter"
}
)
# Connect both our nodes to the END
# Only one node will be run by our conditional edge. Either we get perimeter or area
# this is another way to connect nodes to the END without importing END
graph.set_finish_point("area")
graph.set_finish_point("perimeter")
# Compile--> make our graph executable
app=graph.compile()
print("---------------We will demonstrate Area fisrt--------")
result=app.invoke({'length':15, 'width':6, 'operation':'a'})
print(f"The Area is: {result['area']}")
print('\n---------------------------------------------------')
"""
=======Dont Add this part: this is my cmd result==============
area_perimeter>python area_perimeter_rect.py
---------------We will demonstrate Area fisrt--------
The Area is: 90
---------------------------------------------------
"""
print("---------------We will demonstrate Perimeter--------")
result=app.invoke({'length':20, 'width':10, 'operation':'p'})
print(f"The Perimeter is: {result['perimeter']}")
print('\n---------------------------------------------------')
"""
======Dont Add this part: this is my cmd result========
area_perimeter>python area_perimeter_rect.py
---------------We will demonstrate Area fisrt--------
The Area is: 90
---------------------------------------------------
---------------We will demonstrate Perimeter--------
The Perimeter is: 60
---------------------------------------------------
"""
状态检查工具
状态检查工具会捕获当前状态，以便后续能够回溯之前的操作步骤（演示时使用内存存储；实际应用中则采用SQLite或Postgres）。工具则是模型可以调用的功能——比如计算器、天气查询、搜索等功能。应为这些工具编写清晰的文档字符串；ToolNode会执行模型请求的功能，而无需为每个工具节点都手动编写代码。
from langchain_core.tools import tool
@tool
def current time():
"""Return the current local time"""
return datetime.now().strftime("%H:%M")
tools=[
current_time,
]
# Provide tools to LLM
llm=ChatOpenAI(model="gpt-4.1")
llm_with_tools=llm.bind_tools(tools)
一种结合了工具与记忆功能的智能体：
#!/usr/bin/python3
from datetime import datetime
from dotenv import load_dotenv
import os
from langchain_groq import ChatGroq
#from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph
from langgraph.graph import add_messages
from langgraph.prebuilt import ToolNode
from typing import Annotated
from typing_extensions import TypedDict
# import a helper conditional routing
from langgraph.prebuilt import tools_condition
# add memory using
from langgraph.checkpoint.memory import MemorySaver
# load our .env file to this file
load_dotenv()
# Our state schema
class AgentState(TypedDict):
# messages returned by the individual nodes are appended here
messages: Annotated[list, add_messages]
# add the tools.
# '@' decorator turns our function to a tool
@tool
def current_time():
"""Return the current local time."""
return datetime.now().strftime("%H:%M:%S")
# add our tools to a list in order to bind them to the llm
tools=[
current_time, # we have one tool
]
# set up our Groq LLM
#- remember the groq API we used in building vision.py in Part 1. We can use that
#- Add the API key to a .env file like we did
llm=ChatGroq(
temperature=0.3,
groq_api_key=os.getenv("GROQ_API_KEY"),
model_name="openai/gpt-oss-120b" # our model in here we pick this.
)
"""
llm = ChatOpenAI(
model="gpt-5-nano",
openai_api_key=os.getenv("GPT_5")
)
"""
# Bind our tools to the model
llm_with_tools=llm.bind_tools(tools)
# Add a Node
def chatbot(state:AgentState)-> AgentState:
"""Send in messages to our LLM"""
# call the LLM
response=llm_with_tools.invoke(state["messages"])
# append our LLM response to our state
return {
"messages":[response]
}
# Buiild the graph blueprint
graph=StateGraph(AgentState)
# add nodes
graph.add_node("chatbot", chatbot)
graph.add_node("tools", ToolNode(tools)) # we covered this
# Add edges--> Tells the graph what to execute next.
graph.set_entry_point('chatbot')
# Decides if the LLM needs a tool
graph.add_conditional_edges(
'chatbot',
tools_condition, # we imported this
)
# Bring back execution to the LLM after running chatbot
graph.add_edge('tools', 'chatbot')
graph.set_finish_point('chatbot')
# ---NOW add memory to our graph
memory=MemorySaver()
# turn graph to an executable
app=graph.compile(
checkpointer=memory,
)
# After adding memory we need to set up the threat_id
# Each conversation has a different thread_id and LangGraph will
# call each one with its unique id
config={
"configurable":{
"thread_id":"user1"
}
}
# lets test if our agent responds
while True:
user_input=input("\nYOU: ")
# check if exit or quit and terminate program
if user_input =='exit' or user_input == 'quit':
print("\nGoodbye!")
break
elif user_input=='':
# if user input is empty restart the infinity loop
continue
# invoke our graph
response=app.invoke(
{
"messages":[
HumanMessage(content=user_input)
]
},
config=config # add our checkpointer here
)
# display output
print(f"AI🤖: {response['messages'][-1].content}") # retrieve the last message in our dictionary
"""
============Do not Add: This is my CMD results===================
agent_mem_tool>python agent_memory_tool.py
YOU: Hey, I am Fred a CS student. I love skating on Rollerblades.
AI🤖: Hey Fred! 👋 Great to meet a fellow CS enthusiast who also loves skating on Rollerblades.
**CS side:** What areas are you diving into right now—algorithms, AI, systems, graphics, something else? If you ever want project ideas that blend coding with skating (e.g., a skate‑track visualizer, a sensor‑based trick analyzer, or a simple game), just let me know!
YOU:
YOU: Can You tell me the time right now and date
AI🤖: It’s currently **00:11:01** on **2026‑08‑28** (local time).
YOU: me in one statement
AI🤖: Fred is a computer‑science student who fuels his creativity and focus by roller‑blading whenever he can.
"""
最终项目：智能语音助手
可复用第一部分中的Groq模型。在智能体循环中建议使用强大的文本处理模型；图像相关功能则继续使用视觉处理模块。整体结构如下：
cmd>> dir
tools main.py mcp_server.py stt.py tts.py .env .gitignore
语音转文本模块
stt.py在本地硬件性能较差时可以调用部署在Groq上的Whisper，而在有GPU可用时则可在本地运行Whisper。
#!/usr/bin/python3
import os
import speech_recognition as sr
from groq import Groq
from dotenv import load_dotenv
load_dotenv()
client = Groq(api_key=os.environ.get("GROQ_API_KEY"))
# Audio to Groq's whisper model for transcription
def listen():
# create the speech recognizer object
recognizer = sr.Recognizer()
# open the default microphone
with sr.Microphone() as source:
#Notify the user recording has started.
print("\n🎤Listening...")
# Listen until the user stops speaking
# recorded audio is stored in a variable 'audio'
audio = recognizer.listen(source)
try:
# speech_recognition gives us the captured audio as WAV bytes,
# which we can send straight to Groq's transcription endpoint
wav_bytes = audio.get_wav_data()
# send the recorded audio to Groq's Whisper model
# and convert the spoken words into text
result = client.audio.transcriptions.create(
file=("mic_input.wav", wav_bytes),
model="whisper-large-v3",
response_format="text",
language="en",
)
text = result.strip()
# return the text in lowercase
return text.lower()
# Exception when nothing usable was returned (empty/failed transcription)
except Exception as e:
print(f"Transcription error: {e}")
# Return empty string, not the error text itself - the caller
return ""
文本转语音模块
tts.py可在本地驱动Piper。先进行安装并下载语音文件，然后进行语音合成：
cmd>> uv pip install piper-tts
cmd>> mkdir piper_voices
cmd>> cd piper_voices
# Get the voice from hugging face with corresponding json
cmd>> curl -L -o en_US-amy-medium.onnx https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/amy/medium/en_US-amy-medium.onnx
cmd>> curl -L -o en_US-amy-medium.onnx.json https://huggingface.co/rhasspy/piper-voices/resolve/main/en/en_US/amy/medium/en_US-amy-medium.onnx.json
cmd>> cd ..
# when you get back to root directory you should have
cmd>> dir
tools main.py mcp_server.py stt.py tts.py .env .gitignore piper_voices
#!/usr/bin/python3
from piper import PiperVoice
import sounddevice as sd
import numpy as np
MODEL_PATH="piper_voices/en_US-amy-medium.onnx"
# Load once at startup — reuse across calls, loading the model is the slow part
voice = PiperVoice.load(MODEL_PATH)
def speak(text: str):
"""Synthesize text to speech and play it immediately."""
audio_chunks = []
for chunk in voice.synthesize(text):
audio_chunks.append(np.frombuffer(chunk.audio_int16_bytes, dtype=np.int16))
audio = np.concatenate(audio_chunks)
sd.play(audio, samplerate=voice.config.sample_rate)
sd.wait() # block until playback finishes
在main.py中的协调处理
将第一部分的MCP服务器作为工具加载，导入STT/TTS相关函数，用检查点机制构建LangGraph智能体（为提升持久性可改用SQLite），然后循环执行：监听 → 使用工具推理 → 生成语音。
#!/usr/bin/python3
import asyncio
from typing import Annotated, TypedDict
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.checkpoint.memory import MemorySaver
from langchain_groq import ChatGroq
# from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
import os
# This are our user defined libraries from tts.py and stt.py
from stt import listen
from tts import speak
# our env variables loaded here
load_dotenv()
# Set up our LLM-> If you have GPT API comment this out
llm=ChatGroq(
temperature=0.3,
groq_api_key=os.getenv("GROQ_API_KEY"),
model_name="openai/gpt-oss-120b",
reasoning_effort="low"
)
# Comment out GPT if you dont have the paid API and use the free Groq.
"""
llm = ChatOpenAI(
model="gpt-5-nano",
openai_api_key=os.getenv("GPT_5")
)
"""
# We need this of Groq to prevent the model from hallucinating. GPT will work fine with a smaller prompt
SYSTEM_PROMPT = """
You are Called Botnet, a concise voice assistant. Your replies are spoken aloud, not read.
Tools: pick the single right tool, follow its schema exactly, never invent arguments, never repeat a successful call, never fabricate results. If a tool fails and can't be fixed, say so.
Speaking style: talk like a person, not a document. No emojis, no markdown, no symbols like * # - _ / \\ | ~ < > or brackets. Say numbers and dates the way you'd speak them. Use only periods, commas, and question marks. Answer only what was asked, in 1 to 3 sentences unless more detail is truly needed.
"""
# -----Our State: 'add_messages' is the reducer that appends new messages rather
# than overwritingt the list on each update.
class AgentState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
# Remember Part 1 communicaion to the MCP server is asynchronous communication
async def main():
# 1. Configure MCP server connections
# We are using stdio because it is running locally
client=MultiServerMCPClient(
{
"tools":{
"transport":"stdio",
"command":"python",
"args":["mcp_server.py"],
},
# If we had a remote one:
# "wikipedia":{
# "transport":"streamable_http",
# "url":"https://<domain/ServerIP>/mcp",
#},
}
)
# 2. Discover tools exposed by the connected MCP server
# Each MCP tools is automatically wrapped as a LangChain BaseTool,
# with name/description/schema taken from the server's tool spec.
tools=await client.get_tools()
tools_by_name={t.name: t for t in tools}
#display on screen the loaded tools
print(f"Loaded {len(tools)} tools:{[t.name for t in tools]}\n")
speak("Welcome Sir!!")
# Bind up our model
llm_with_tools=llm.bind_tools(tools)
# Node--> a python function to carry one task.
# Get user Speech Transcribed
def get_input(state:AgentState)-> AgentState:
"""User inputs his command via microphone and they are transcribed to text"""
text=listen()
print(f"YOU Said: {text}")
# the add_reducer function returns only the new messsage
return {"messages":[HumanMessage(content=text)]}
# Takes the last message now, the user input and executed using the LLM+MCP
async def call_model(state: AgentState):
"""Call the LLM passing the user question"""
messages=[SystemMessage(content=SYSTEM_PROMPT)]+state["messages"]
response=await llm_with_tools.ainvoke(messages)
# append our results to our state--> we are just adding new messages remember
return {"messages":[response]}
# The speaking node to take the last AI message and speak it
def talk_back(state: AgentState):
"""Return the message as output voice"""
text=state["messages"][-1].content
print(f"AI: {text}")
speak(text)
#Node functions should return a state update even when empty to prevent error
return {}
# add our MCP server as a tool Node
tool_node=ToolNode(tools)
# Build our graph schema
graph=StateGraph(AgentState)
# Add our nodes
graph.add_node("llm", call_model)
graph.add_node("tools", tool_node)
graph.add_node("listening", get_input)
graph.add_node("speaking", talk_back)
# add the edges-->control the next node to execute in the graph
graph.set_entry_point("listening")
graph.add_edge("listening", "llm")
# here we have a conditional edge when our LLM needs a tool
# tools_condition we imported will return "tools" if LLM asked for tools
# or END otherwise. Mapping END to speaking
graph.add_conditional_edges(
"llm",
tools_condition,
{"tools":"tools", END:"speaking"}
)
# after tools run we loop back to LLM with the result
graph.add_edge("tools", "llm")
# end execution
graph.set_finish_point("speaking")
# The memory we will use
memory=MemorySaver()
# compile the graph
app=graph.compile(
checkpointer=memory
)
# set the the thread id since we are using checkpointer memory
config={
"configurable":{
"thread_id":"user1"
}
}
while True:
await app.ainvoke({"messages":[]}, config=config)
if __name__=="__main__":
asyncio.run(main())
"""
===========Do Not add this Part: It is my CMD Output(Using Groq Free API)===========
final_Project>python main.py
Loaded 7 tools:['current_time', 'calculate', 'image_info', 'internet_search', 'scrape_webpage', 'weather_forecast', 'system_info']
🎤Listening...
YOU Said: hello, i am Fred.
AI: Hello Fred. Nice to meet you. How can I help you today?
🎤Listening...
YOU Said: introduce yourself in three words.
AI: Helpful. Concise. Friendly.
🎤Listening...
YOU Said: i need to check the weather of a particular city.
AI: Sure, which city would you like the forecast for?
🎤Listening...
YOU Said: nairobi.
AI: The current weather in Nairobi is about twenty three degrees Celsius with broken clouds and moderate humidity.Let me know if you need a longer forecast or anything else.
🎤Listening...
YOU Said: okay, thanks.
AI: You’re welcome. Anything else I can do for you?
"""
增强安全性的方法包括：完全离线运行Whisper、Piper以及Ollama模型；加强免费套餐的速率限制机制；在不改变语音界面的前提下扩展MCP工具。
总结
该系列的路径为MCP工具 → 语音输入输出 → LangGraph编排。选取最终项目，在本地运行并不断迭代。一旦框架可行，更多关于LangGraph的资料便能轻松获取。
“The real problem is not whether machines think but whether men do”
— John McCarthy (founded the term “Artificial Intelligence”)
最终语音智能体的集成注意事项
当MCP服务器、文本转语音、语音转文本功能以及LangGraph循环最终整合在一起后，首先进行纯文本调用测试。将输入的字符串发送到日后麦克风会传输数据的相同节点中。只有在工具路由逻辑正常之后，才引入音频功能——否则任何故障都会被误认为是Whisper的问题。
线程ID与检查点机制需要明确的政策规范。如果语音助手忘记之前的对话内容，就会显得功能异常；而将线程A的工具泄露到线程B中的情况则更为严重。默认情况下，每次会话都应使用新的线程，并将持久性数据存储到由MCP支持的配置存储中，而不仅仅是保存在检查点历史记录里。
免费版Groq（或类似服务）的速率限制会导致演示效果不稳定。应对麦克风输入的语音进行排队处理，给出“正在处理中”的语音反馈，并避免因突然插入的音频而引发模型调用冲突。当使用Ollama、Whisper和Piper完全离线运行时，需重新测试CPU环境下的延迟情况：用户在麦克风处于开启状态并等待回复时，对缓慢的响应容忍度会更低。
系列项目中的“完成”标准
明确需要实现的检查项：智能体可通过MCP计算器解答口述的算术题；当缺少天气相关参数时，它应能够拒绝处理或给出解释；在同一个流程的两次调用之间，检查点机制需能记住用户的姓名；Piper应完整输出最终答案而不会被截断。除此之外的一切都属于待处理任务。在追求更佳的语音效果或更大规模的模型之前，先把这些检查项写下来。一个通过简单垂直测试的方案，远胜于一次演示就出问题的复杂横向系统。在分享项目时，需提供准确的uv锁输入值及模型编号，以免他人误对模型进行无声替换。其教育价值在于整个整合后的流程——包括MCP、语音功能及图形处理模块——而非某个脆弱的单点组件。
如果语音循环出现延迟，应在更换模型前先分析性能指标：文本转语音的时间、首个工具的响应延迟以及文本转文字的时间往往远超过令牌生成时间。应解决你测得的那个慢环节，而非最容易调整的环节。 先推出最小可行版本，再逐步优化。