从单节点聊天机器人到LangGraph中的MCP驱动智能体
逐层构建 LangGraph 应用：状态与 reducer、边、工具循环、检查点线程、三种流式模式，以及通过 MCP 提供的工具。
来源: Building AI Applications with LangGraph: From a Basic Chatbot to Tools, Memory, Streaming and MCP
大多数大型语言模型原型最初都是通过单次调用实现的：发送提示语，然后输出回复。但一旦应用需要网页搜索、决定下一步该执行什么操作、保留之前的对话记录，或是需要在用户界面中实时显示进度，这种简单结构就会失效。LangGraph通过将控制流转化为基于共享状态的显式图结构来解决这一问题。本指南展示如何将一个仅包含单节点的聊天机器人逐步发展成能够使用工具的ReAct智能体，该智能体具备检查点存储功能、流式处理能力，以及由独立MCP服务器提供的工具支持，同时还会指出人们在实现过程中容易遇到的问题。
思维模型：状态通过边在节点间流动
普通的LLM集成流程是从提示词直接传递到模型并得到答案的线性过程。而LangGraph要求以循环方式思考：一个状态对象进入图结构，节点读取该状态并返回变更结果，边决定下一个由哪个节点处理，最终更新后的状态会输出出来。因此，其三个基本构成要素为：
- 状态：在各个步骤之间传递的数据，例如消息历史记录。
- 节点：执行特定任务并返回部分更新结果的普通函数。
- 边：连接节点的规则，包括流程的起始点和结束点。
最简单的图结构在入口点和出口点之间只有一个节点：
┌─────────────┐
│ State │
└──────┬──────┘
│
▼
┌─────────────┐
│ Node │
│ chatbot │
└──────┬──────┘
│
▼
┌─────────────┐
│ END │
└─────────────┘
将“存在哪些数据”与“以何种顺序执行操作”分开是核心设计思路。你可以更改执行流程而不影响数据的存储方式，也可以在任意时刻查看状态，而无需反编译函数调用链。
为何选择图结构而非调用链
最简单的聊天机器人只是用户与模型之间的一条直接通道：
User
↓
LLM
↓
Response
现在加入实际需求：网页搜索、多种工具、分支逻辑、内存管理、人工审批步骤、多个智能体、流式输出，以及根据模型返回结果的不同而采取不同行为。这些需求都会增加分支或循环，而且围绕 API 调用嵌套的 if 语句很快就会变得难以理解。即便是最常见的情况——“如果模型要求则调用工具”——也 already 包含循环结构：
┌─────────────┐
│ LLM │
└──────┬──────┘
│
Need a tool?
/ \
Yes No
↓ ↓
Tool Response
│
↓
LLM
一旦流程被表示为图结构，分支和循环就会变成可以命名、测试和绘制的可见对象，而不再隐藏在辅助函数中的控制流中。
第一步：使用StateGraph创建单节点聊天机器人
第一个版本被刻意设计得非常简单，仅在起始标记和结束标记之间连接了一个节点：
START
↓
Chatbot Node
↓
END
定义状态及其还原函数
对于聊天机器人而言，状态主要需要存储对话历史。最初的尝试是使用包含messages列表的TypedDict：
from typing import TypedDict
class State(TypedDict):
messages: list
从概念上讲，该列表用于记录人类与人工智能交替发言的内容：
State
└── messages
├── HumanMessage
├── AIMessage
├── HumanMessage
└── AIMessage
这里有一个非常重要的细节。在 LangGraph 中，状态中的每个键都可以拥有一个reducer，用于决定如何将节点的返回值合并到现有值中。普通的 list 注解没有 reducer，因此返回 {"messages": [reply]} 的节点会替换整个列表而非将其追加到其中。对于聊天记录，你几乎总是需要使用 add_messages reducer，它被定义为 Annotated[list, add_messages]（或继承自预构建的 MessagesState）。使用该 reducer 后，返回的消息会被追加，而具有匹配 ID 的消息则会直接被更新。上面的示例仅用于辅助理解，实际代码中应使用带注解的形式。
编写聊天机器人节点
节点只是一个函数，它接收当前状态并返回需要修改的键。聊天机器人节点会将历史记录传递给模型，然后返回回复：
def chatbot(state: State):
return {
"messages": [
llm.invoke(state["messages"])
]
}
从内到外阅读代码。该函数会接收到整个状态：
state
它提取出目前的对话内容：
state["messages"]
它将这段历史记录发送给模型：
llm.invoke(state["messages"])
然后它用字段名作为键，将模型的回复封装在字典中。需要注意的是，该节点不会直接修改状态；它返回的是部分更新结果，由LangGraph负责合并：
{
"messages": [
llm.invoke(state["messages"])
]
}
从数据流的角度来看，该节点位于同一状态的两个版本之间：
State
│
│ messages
▼
┌─────────────┐
│ Chatbot │
│ Node │
└──────┬──────┘
│
│ LLM Response
▼
State
注册节点
一旦该函数在构建器中以特定名称被注册，它就会成为图结构的一部分：
graph_builder.add_node(
"llmchatbot",
chatbot
)
第一个参数是节点的标识符，边和日志都将以此来引用该节点：
"llmchatbot"
第二个参数是每当执行流程到达该节点时就会运行的可调用函数：
chatbot
步骤2：连接边并编译
单个节点本身永远不会被执行；边用于告诉LangGraph如何到达该节点以及之后应前往何处。这个图需要两条边：第一条边将特殊的START标记与聊天机器人相连：
graph_builder.add_edge(
START,
"llmchatbot"
)
以图表形式表示，执行流程进入图后直接到达该节点：
START
↓
llmchatbot
第二条边将控制权从聊天机器人传递到END，从而结束执行流程：
graph_builder.add_edge(
"llmchatbot",
END
)
这对应于：
llmchatbot
↓
END
这两条边共同构成了一个虽然简单但完整的图：
START
│
▼
┌────────────────┐
│ llmchatbot │
│ Node │
└────────┬───────┘
│
▼
END
编译与调用
编译过程会验证结构（例如确保每条边都指向已知的节点），并生成一个可运行的对象：
graph = graph_builder.compile()
通过传入初始状态来运行它。普通的角色/内容字典会被接受并转换为LangChain消息对象：
result = graph.invoke({
"messages": [
{
"role": "user",
"content": "What is LangGraph?"
}
]
})
在内部，输入状态从START节点开始，聊天机器人节点会调用模型，回复会被合并到状态中，随后运行结束：
Input State
↓
START
↓
llmchatbot
↓
LLM
↓
Updated State
↓
END
invoke的返回值即为最终状态，因此模型的回答就是result["messages"]中的最后一个条目。
步骤3：绘制图表
已编译的图表可以自行渲染。在Jupyter笔记本中，IPython的显示工具会展示由图表的Mermaid描述生成的PNG图片：
from IPython.display import Image, display
display(
Image(
graph.get_graph().draw_mermaid_png()
)
)
对于这个图表，生成的图片较短：
┌───────┐
│ START │
└───┬───┘
│
▼
┌──────────────┐
│ llmchatbot │
└──────┬───────┘
│
▼
┌───────┐
│ END │
└───────┘
三个框时数值并不明显，但会迅速增长。一旦出现条件边、工具节点和循环，渲染出的图表是确认所构建的图确实符合预期最快的方法。需要注意的是，默认情况下draw_mermaid_png()会通过外部Mermaid服务进行渲染，因此需要网络连接；如果无法访问网络，则draw_mermaid()会返回Mermaid文本，可将其在其他地方进行渲染。
第4步：为模型添加搜索工具
聊天机器人只能根据模型在训练期间学到的知识来回答问题。对于有关代理型AI最新发展的疑问，需要最新的信息，因此下一步就是添加网络搜索工具。示例中使用了Tavily Search，其工作流程如下：
User
↓
LLM
↓
Search required?
↓
Tavily Search
↓
Search Results
↓
LLM
↓
Final Answer
绑定工具与运行工具并非同一回事
这项工作的前半部分是工具绑定：告诉模型有哪些工具以及它们需要哪些参数。
llm_with_tools = llm.bind_tools(
[tavily_search]
)
完成绑定后，模型可以以结构化的工具调用形式（“用此查询调用 tavily_search”）来响应，而非纯文本。模型本身无法执行该调用，必须由应用程序中的某些组件检测到工具调用、运行相应工具，将结果作为 ToolMessage 添加进去，再将其全部反馈给模型。仅让模型能够使用工具并不会使其成为智能体，真正起作用的是周围的执行流程，而图结构正是用来定义这种流程的。
第5步：将其转化为ReAct循环
ReAct结合了推理与行动：模型先思考任务，判断是否需要采取行动，再通过工具执行该动作，观察结果后继续下一步。用图表示的话，之前的直线：
START
↓
LLM
↓
END
会变成一个包含决策点以及通向模型的路径的图：
START
│
▼
Agent
│
▼
┌─────────┐
│ LLM │
└────┬────┘
│
Need a tool?
/ \
Yes No
│ │
▼ ▼
Tool END
│
▼
Agent
│
▼
LLM
在 LangGraph 代码中，这一决策体现为条件边。预构建的 ToolNode 会执行最后一条 AI 消息中指定的工具功能，而 tools_condition 会在存在工具调用时将流程导向该节点，否则则导向 END 节点。从工具节点返回模型的边则构成了循环结构。预构建的 create_react_agent 辅助函数会为你自动生成相同的结构；由于预构建的代理辅助函数的位置会随时间在不同包之间变动，建议查阅最新的 LangGraph 文档以获取其准确位置。
追踪一次循环过程
假设用户询问关于 LangGraph 的最新消息，那么一次循环执行后会产生如下消息与步骤序列：
HumanMessage
↓
Agent
↓
LLM
↓
"I need current information."
↓
Tool Call
↓
Tavily Search
↓
Search Results
↓
ToolMessage
↓
Agent
↓
LLM
↓
Final Answer
重要的结构差异在于循环机制。最初的聊天机器人采用的是线性流程：
START → LLM → END
ReAct 图可以根据需要多次重新调用模型：
START
↓
Agent
↓
LLM
↓
Tool?
┌───────┐
│ │
Yes No
│ │
↓ ↓
Tool END
│
↓
Agent
由于循环在原理上可能无限运行，LangGraph 会对每次运行设置递归限制（可通过 recursion_limit 参数进行配置）。如果模型持续请求工具，系统会报错而不会产生无限的费用。
步骤 6：使用 MemorySaver 记录对话
到目前为止，每次调用 invoke 都是从头开始。假设有两个对话轮次：首先用户自我介绍：
User:
My name is Himanshi.
然后在另一条请求中提出进一步的问题：
User:
What is my name?
只有当图中仍存在第一轮对话时，才能回应第二轮。LangGraph通过检查指针来解决这一问题：每执行一步后都会保存当前状态的快照，下次处理同一对话时就会从该快照继续。最简单的实现方式是MemorySaver：
┌─────────────┐
│ Agent │
└──────┬──────┘
│
▼
┌───────────┐
│ State │
└─────┬─────┘
│
▼
┌─────────────┐
│ MemorySaver │
└─────────────┘
可在编译时通过graph_builder.compile(checkpointer=memory)来启用它。这样一来，每次调用不再是独立的执行过程，而是属于同一个对话序列，其状态会在多次运行之间被保留下来。
线程用于分隔不同对话
将各次运行关联起来的关键在于线程ID，该ID以{"configurable": {"thread_id": "..."}}的形式在配置中传递。每个线程都有自己独立的历史记录和当前状态：
Thread 1
├── Message 1
├── Message 2
├── Message 3
└── Current State
Thread 2
├── Message 1
├── Message 2
└── Current State
因此，运行在两个线程中的两个用户永远无法看到对方的消息：
User A → Thread A → State A
User B → Thread B → State B
有两个需要注意的点。MemorySaver会将所有数据保存在进程内存中，所以重启后所有历史记录都会消失，且无法在多个服务器实例之间共享；而生产环境部署则会使用诸如 Postgres 或 SQLite 这样的持久化检查点机制。在最近的版本中，该类也被命名为 InMemorySaver。如需详细了解检查点实际存储的内容，请参阅InMemorySaver 中检查点、写入数据及二进制数据是如何协同工作的。
第7步：在图结构运行过程中流式处理状态
以上所有内容都使用了阻塞式调用：
graph.invoke(...)
它会等待整个图计算完成后再返回结果。对于多步骤的智能体来说，这意味着屏幕上会数秒没有任何显示，让用户感觉体验不佳。此外，编译后的图还会暴露stream接口，在执行过程中就会持续输出结果。
stream_mode="values"：每一步之后的完整状态
首先需要了解的模式是：
stream_mode="values"
在循环中使用时，它会在每一步之后输出整个状态的快照：
for chunk in graph.stream(
input_state,
stream_mode="values"
):
print(chunk)
每个片段都展示着当前时刻的完整状态，因此你可以看到随着节点和工具的运行，状态是如何逐渐变化的：
START
↓
State Update #1
↓
Node Execution
↓
State Update #2
↓
Tool Execution
↓
State Update #3
↓
END
当状态规模较小，且你只想每次渲染“最新版本”时，这种模式非常方便。
stream_mode="updates"；：仅显示每个节点的变更内容
第二种模式是：
stream_mode="updates"
两者的区别在于每个数据块包含的内容。在一种模式下，你会收到完整的状态：
values
而在另一种模式下，你只会收到差分信息，这些信息以生成它的节点名称作为标识：
updates
记住这一区别的简单方法：
- values 可以回答“当前状态是怎样的？”
- updates 可以回答“这个特定节点刚刚改变了什么？”
updates 通常更适合作为调试和日志记录的选择，因为它能告诉你每个变化是由哪个节点引起的，并且在状态较大时能使数据块保持较小规模。这两种模式都不会逐个传输模型中的标记；若需要按标记逐一输出，还有单独的 messages 流模式，你可以传入多个模式列表以同时接收多种类型的数据。
第8步：使用astream_events实现精细化的可见性控制
状态快照和节点差异能告诉你发生了什么变化，但无法呈现所有细节。astream_events则进一步深入，为运行过程中每个组件的每一个生命周期节点都会触发一个事件：
Graph Execution
│
├── Graph Started
│
├── Node Started
│
├── LLM Started
│
├── LLM Streaming
│
├── LLM Completed
│
├── Tool Started
│
├── Tool Completed
│
├── Node Completed
│
└── Graph Completed
你需要以异步方式处理这些事件。注意，下面代码片段中的print语句需要在循环体内部缩进；否则如所示，Python会因缩进错误而拒绝执行：
async for event in graph.astream_events(
input_state,
version="v2"
):
print(event)
每个事件都包含类型（例如on_chat_model_stream或on_tool_start）、组件名称及其数据。这使得事件流能够用于：
- 调试与结构化日志记录
- 监控智能体运行的整体情况
- 追踪每一次大语言模型调用和工具调用
具有后端背景的工程师会认出这种模式：它更接近于追踪分布式请求，而非读取单个函数的返回值。问题从“代理是否正常工作？”转变为“代理具体做了什么、按什么顺序执行，以及每个步骤耗时多久？”
各组件的功能定位
回顾整个过程，每一步都在相同的框架基础上增添了一项功能。最初只是简单的模型调用：
START
↓
LLM
↓
END
工具的加入增加了与外部世界的往返通信：
START
↓
LLM
↓
Tool
↓
LLM
↓
END
检查点的引入使循环能够记住状态：
START
↓
Agent
↓
LLM ↔ Tools
↓
Memory / Checkpoint
↓
END
流式处理和事件机制则让运行过程能够展示在用户界面中并便于监控：
Graph
↓
Events
↓
UI / Logs / Observability
这些步骤均无需重写之前的内容；每个步骤都是在同一图结构中新增的节点、边、编译选项或方法调用。
第9步：将工具置于MCP服务器之后
到目前为止，这些工具都存在于应用程序内部。而模型上下文协议（MCP）采用了不同的方式：通过独立的服务器来提供工具（及资源），任何MCP 客户端都可以发现并调用这些工具。这样一来，应用程序就不再需要包含所有的工具实现：
┌──────────────┐
│ MCP Client │
└──────┬───────┘
│
│
▼
┌──────────────┐
│ MCP Server │
└──────┬───────┘
│
┌──────────┼──────────┐
↓ ↓ ↓
add() multiply() weather()
包含三个工具的简易服务器
一个小型学习用服务器就足以展示这种架构。它提供了一个加法工具：
add(a, b)
一个乘法工具：
multiply(a, b)
以及一个能够接收城市名称的天气查询工具：
weather(city)
该天气工具并未调用真实的天气服务，而是返回固定的字符串，这使得关注点集中在协议本身而非 API 密钥上：
"It's sunny in Mumbai."
重点在于数据往返过程，而非天气预报内容：
Client
↓
MCP Server
↓
Tool
↓
Result
↓
Client
工具不再需要嵌入到使用它们的应用程序中
更宏观的理念是，工具可以独立于使用它们的应用来部署。应用程序与 MCP 客户端通信，客户端通过协议与服务器交互，而具体的实现则由服务器负责：
Application
│
▼
MCP Client
│
MCP Protocol
│
▼
MCP Server
/ | \
/ | \
↓ ↓ ↓
Add Multiply Weather
这样一来，可重用的工具服务器便成为可能。某个团队可以运行一个能够提供某一类功能的服务器：
Database Tools
File Tools
Git Tools
Search Tools
Weather Tools
Internal APIs
多个不同的代理、聊天客户端或 IDE 集成都可以使用同一个服务器，而无需重复编写代码。
步骤 10：结合 LangGraph 和 MCP
这两种技术分别解决问题的不同方面。LangGraph负责管理工作流程：状态、路由、循环、内存以及流处理。MCP则规范了工具与资源的暴露方式。在结合使用的设计中，智能体的工具节点会与MCP客户端通信，而非调用本地函数：
User
│
▼
LangGraph Agent
│
▼
LLM
│
Need a tool?
│
▼
MCP Client
│
▼
MCP Server
/ | \
↓ ↓ ↓
Math Search Weather
智能体无需了解工具的具体实现方式，只需知道有哪些功能以及如何调用它们，其余工作由MCP服务器处理。在Python中，langchain-mcp-adapters包是一种常见的方法，可用于将MCP工具加载为LangChain工具，从而将其绑定到模型并在ToolNode中使用。现有文章探讨了如何利用MCP在LangGraph中实现工具集成，内容更为深入。
完整的思维模型
综合来看，LangGraph智能体的工作流程如下：用户的输入转化为状态，节点对这些状态进行操作，模型判断是否需要使用工具，工具的返回结果再反馈给模型，最终以响应形式结束整个流程：
USER
│
▼
┌────────────┐
│ LangGraph │
│ Agent │
└──────┬─────┘
│
▼
STATE
│
┌─────────────┼─────────────┐
│ │ │
▼ ▼ ▼
NODE NODE NODE
│ │ │
└─────────────┼─────────────┘
│
▼
LLM
│
Need a tool?
/ \
Yes No
│ │
▼ ▼
TOOL RESPONSE
│
▼
RESULT
│
└──────► LLM
│
▼
END
在那个核心循环周围，是根据需求启用的各种功能：
┌──────────────────────────────┐
│ Capabilities │
│ │
│ Memory / Checkpointing │
│ Streaming │
│ Event Tracking │
│ MCP │
│ Tools │
└──────────────────────────────┘
LangChain 的终点与 LangGraph 的起点
由于 LangGraph 通常是在 LangChain 之后学习的，因此将两者的角色区分开来很有帮助。LangChain 为大型语言模型应用提供了基础组件：模型封装器、消息、工具、提示词以及结构化输出。而 LangGraph 则将这些组件整合成具有明确状态的工作流。二者是互补的关系而非竞争关系，大多数 LangGraph 代码在其节点中都会使用 LangChain 的组件。
可应用于更大规模图结构的经验
从这个小型示例中总结出的几条原则同样适用于规模更大的系统。
状态是核心
与其通过函数参数传递数据，不如让每个节点读取和写入同一个共享状态，而 reducer 则负责定义如何合并这些更新：
State
↓
Node
↓
Updated State
↓
Next Node
节点是工作单元
每个节点应执行一项可明确识别的任务，这样就能更轻松地分解和测试复杂的工作流程：
chatbot node
search node
tool node
validation node
human approval node
边负责编码工作流程
一条边回答一个问题：当前节点处理完成后，执行流程将走向何处？普通边表示顺序执行，条件边则表示分支和循环。
图表比代码更易理解
对于包含大量节点和条件路径的图，渲染后的图表比阅读构建代码能更快地传达其架构结构：
Node → Decision → Tool → Agent → END
检查点不仅仅是聊天记录
检查点会保存完整的图结构状态，而不仅仅是消息内容。这不仅支持多轮对话，也是实现长时间运行的智能体、在人工干预下暂停以及故障后恢复功能的基础。
流式处理同时惠及用户与开发者
流式处理提升了系统的响应速度，同样的输出也有助于调试和观察运行过程。
一次智能体运行包含众多操作
通过astream_events可以清楚地看到，一次智能体调用实际上是由图操作、节点操作、模型操作和工具操作组成的嵌套序列：
Graph
↓
Node
↓
LLM
↓
Tool
↓
LLM
↓
Node
↓
Graph
应用程序越复杂，就越需要这种可视化能力。
MCP将工具与应用程序解耦
通过MCP暴露的工具可以在不同应用程序之间共享，并且可以独立于调用它们的智能体进行发展。
下一步学习方向
这一基础为多个值得探索的方向打开了大门：条件边与并行分支、人工干预机制、长期记忆与持久化检查点、图结构内的检索增强生成、多智能体系统、更深入的MCP集成、智能体的可观测性与评估、生产环境部署，以及具备重试功能的容错机制。
其共同点在于，人工智能应用并不仅仅是生成文本的模型，而是一个围绕该模型设计的系统——它能够接收请求、做出决策、执行操作、观察结果、将其存储并继续后续流程：
Understand
↓
Decide
↓
Act
↓
Observe
↓
Remember
↓
Continue
图结构是表达这种循环结构的自然方式。
总结
整个发展过程始于最简单的图结构：
START → Chatbot → END
随后逐层扩展，最终形成包含工具、智能体循环、内存、流处理及MCP功能的完整系统：
┌─────────────┐
│ LLM │
└──────┬──────┘
│
▼
┌─────────────┐
│ Tools │
└──────┬──────┘
│
▼
┌─────────────┐
│ Agent │
└──────┬──────┘
│
┌──────────┼──────────┐
↓ ↓ ↓
Memory Streaming MCP
│ │ │
└──────────┼──────────┘
↓
AI Application
将这些特性放在一起看，它们就不再像是一组列表，而更像是一种架构：
- 状态负责存储数据以及通过检查点保存内存；消息列表则由
add_messagesreducer来处理。 - 节点负责执行任务并返回部分更新结果。
- 边用于定义数据流，而条件边则可将流程转化为循环。
- 模型提供推理能力和语言支持；工具则使其能够与外部世界交互。
- 流处理和事件使得从界面更新到追踪信息在内的所有运行过程都可被观察。
- MCP标准化了工具的暴露方式，从而使它们能够在使用它们的智能体之外独立存在。