如果你是从传统 LLM 应用开发转过来的,最近一定有一个很强烈的感受:LangChain 的教程变了,代码写法也变了,过去把几个 Prompt、一个模型、一个 Python 函数串起来的 Chain 方式,正在被一种叫 LangGraph 的图结构替代。
真正容易让人困惑的点在于:既然已经学了 LangChain,为什么还要再学 LangGraph?Chain 和 Graph 到底有什么区别?新手到底应该从哪套开始?这些问题如果只看碎片化资料,很容易越看越乱。
这篇文章不会从概念堆砌开始,而是从一条主线展开:带你从零构建一个能调用工具、能做条件判断、能记住多轮对话上下文的 AI-Agent。在这个过程中,我会把 LangChain 与 LangGraph 的关系讲清楚,解释 MCP 协议在真实项目里解决什么问题,也会拆解智能体记忆 Memory 的落地方式。
读完这篇文章,你能跑通一个最小可用的 Agent 项目,并且知道再往生产环境走,真正的坑在哪里。
1. 为什么现在必须重新理解 LangChain 与 LangGraph
1.1 从“管线时代”到“状态机时代”
LangChain 早期最核心的用法,是把 LLM 调用封装成PromptTemplate + LLM + OutputParser,然后用Chain把它们按顺序串起来。这种模式非常适合固定流程的应用,比如一个简单的 RAG 问答、一个文本摘要服务。
但到了 Agent 时代,应用需求发生了变化。Agent 不再只是“按顺序执行”,它需要根据模型的输出决定下一步做什么:是调用工具查天气,还是直接给出最终答案;是继续追问用户,还是结束对话。这种“下一步执行什么”是运行时动态决定的,固定写死的 Chain 处理不了这种分支和循环。
所以 LangGraph 做的事情很直接:把执行流程变成一张有向图,图的节点是函数,图的边是连接关系。通过add_conditional_edges条件边,节点之间可以形成循环、分支、跳转,这才符合 Agent 的真实运行逻辑。
1.2 LangGraph 不是替代 LangChain,而是替代 Chain 编排
很多初学者问一个问题:LangGraph 出来了,是不是就不用学 LangChain 了?
这个理解不准确。LangGraph 并没有替换掉 LangChain 里的模型调用接口、工具封装、提示词模板这些基础能力,它替换的是“编排层”。你在 LangGraph 的节点里,依然会使用ChatOpenAI、bind_tools、tool装饰器这些 LangChain 经典组件。
换句话说:
- LangChain 提供的是“零件”:模型、工具、解析器、向量库封装。
- LangGraph 提供的是“组装逻辑”:把零件放在一个状态机里,由状态和条件决定执行路径。
- Agent 则是这套组合想实现的高级形态:让模型作为“大脑”,在循环中决定调用哪些工具、什么时候给出结果。
1.3 一个比较明确的判断
如果你现在要新起一个项目,尤其是涉及工具调用、多轮决策、MCP、复杂记忆的项目,直接选 LangGraph,不要再用旧 Chain 方式堆流水线。如果项目只是一个没有分支的固定流程,Chain 也够用,但从长期维护和扩展角度看,LangGraph 的改造成本并不高。
后面我会用一个非常小的人事流程示例来演示 LangGraph 的直觉模型,你会发现它并没有想象中复杂。
2. 核心概念扫盲:State、Node、Edge、Agent、MCP、Memory
在开始写代码之前,先把几个高频概念一次性说清楚。理解这些概念,后面看代码会轻松很多。
2.1 State:图中流转的状态对象
State 是整个 LangGraph 运行期间保存数据的对象,在 Python 里通常是一个TypedDict。每个节点函数接收当前 State,然后返回一个字典,这个字典里的字段会合并到 State 中。
这是 LangGraph 最重要的设计之一:节点函数不直接修改外部变量,而是通过返回值更新状态。这样做的好处是流程可追踪、可回放,也更容易排查问题。
2.2 Node 与 Edge:节点与边
节点就是普通的 Python 函数。一个节点做一件事,比如调用一次模型、执行一个工具、写一段摘要。
边表示节点之间的执行顺序。普通边用add_edge,表示“执行完 A 一定执行 B”。条件边用add_conditional_edges,表示“根据函数返回值,动态选择下一步去哪个节点”。Agent 的决策循环,本质上就是条件边的应用。
2.3 Agent:具备决策循环的智能体
可以这样理解 Agent:它不是一个单独的 Python 类,而是“模型 + 工具 + 循环”的组合。最经典的实现模式是 ReAct:
- 模型观察当前问题
- 决定需要调用哪个工具
- 工具返回结果
- 模型继续推理
- 直到模型认为可以给出最终答案
这种循环在 LangGraph 里非常自然:Agent 节点调用模型,如果模型返回了工具调用请求,就走工具节点,执行完工具再回到 Agent 节点;如果没有工具调用,就走向结束节点。
2.4 MCP:连接 Agent 与外部系统的开放协议
MCP 全称 Model Context Protocol,是一个把工具、资源、提示词统一成标准接口的开放协议。它的目的是解决一个实际问题:以前每接入一个外部系统,就要写一套适配代码;现在只要对方提供 MCP Server,Agent 就能通过 MCP Client 加载暴露出来的工具。
可以把 MCP 理解为“USB-C 接口”。以前你要为不同设备准备不同充电线,现在设备都提供统一接口,只要线缆支持这个标准就能直接连上。
2.5 Memory:短期记忆与长期记忆
Memory 在 Agent 系统里不是一个单一组件,而是分层的概念:
- 短期记忆:在一个会话内让 Agent 记住上下文。LangGraph 里通过 State 中的消息列表实现,如果加上 Checkpointer,还能把状态持久化下来,实现跨多轮、跨请求的记忆。
- 长期记忆:跨会话记住用户偏好、历史信息。通常需要借助外部存储,比如 Redis、MySQL、向量数据库。
这两个概念非常容易被混在一起,后面第 7 章会结合代码演示。
下表可以快速对照这些概念:
| 概念 | 解决什么问题 | 典型实现 |
|---|---|---|
| State | 图运行期间的状态共享 | TypedDict、Annotated |
| Node | 图中的一个执行单元 | 普通 Python 函数 |
| Edge | 节点之间的执行顺序 | add_edge、add_conditional_edges |
| Agent | 模型决策 + 工具调用的循环 | ReAct 模式、条件边 |
| MCP | 工具与外部系统的标准化连接 | MCP Server / MCP Client |
| Checkpointer | 状态持久化与多轮会话记忆 | MemorySaver、SqliteSaver |
3. 环境准备与前置条件
3.1 版本与环境要求
LangGraph 目前支持 Python 3.9 以上版本,推荐使用 Python 3.11 或 3.12。演示代码依赖 LangChain 和 LangGraph,安装的时候最好在虚拟环境里进行,避免污染系统 Python。
3.2 创建虚拟环境
python -m venv .venv source .venv/bin/activate # Windows 环境执行:.venv\Scripts\activate pip install --upgrade pip3.3 安装依赖
pip install langgraph langchain langchain-openai langchain-core如果后面要做 MCP 接入,还需要额外安装 MCP 相关包:
pip install mcp langchain-mcp-adapters版本说明:LangChain 生态迭代速度很快,本文示例基于当前主流 API 写法。建议你安装后先运行代码验证,如果发现 API 有变化,以官方文档和代码提示为准。具体版本不建议盲追最新,生产环境更要锁定版本。
3.4 模型配置
本文示例使用 OpenAI 兼容接口。你可以在环境变量中配置 API Key:
export OPENAI_API_KEY="sk-xxxx"如果你本地部署了 Ollama 或 DeepSeek 这类支持 OpenAI 接口的模型,也可以通过base_url参数接入,不绑定具体厂商。
4. 从 Chain 到 Graph:LangGraph 基础构建
4.1 理解 State 的合并机制
先解决一个很常见的问题:在 LangGraph 的节点函数里,怎么修改 State 的值?
答案不是原地修改,而是返回一个字典。LangGraph 会自动把返回值按 key 合并到 State。例如当前 State 有topic字段,节点函数返回{"outline": "xxx"},运行后 State 会同时包含原字段和新字段。
4.2 最小示例:两个顺序节点
下面这个例子非常小,但包含了 LangGraph 的全部核心要素:定义 State、添加节点、连接边、编译、调用。
# file: simple_graph.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class BlogState(TypedDict): topic: str outline: str content: str def write_outline(state: BlogState): # 根据 topic 生成大纲 return {"outline": f"{state['topic']} 的完整大纲"} def write_content(state: BlogState): # 根据 outline 生成正文 return {"content": f"{state['outline']}\n\n这里是正文内容……"} g = StateGraph(BlogState) g.add_node("write_outline", write_outline) g.add_node("write_content", write_content) g.add_edge(START, "write_outline") g.add_edge("write_outline", "write_content") g.add_edge("write_content", END) app = g.compile() result = app.invoke({"topic": "LangGraph入门"}) print(result)运行命令:
python simple_graph.py预期输出:
{'topic': 'LangGraph入门', 'outline': 'LangGraph入门 的完整大纲', 'content': 'LangGraph入门 的完整大纲\n\n这里是正文内容……'}这个示例说明了一个关键点:LangGraph 的执行流程就是“节点函数依次被调用,每个节点返回的字段不断累积到 State 中”。不需要中间变量满天飞,所有数据都从 State 里拿。
4.3 条件边:让 Graph 拥有决策能力
上面这个例子还是顺序执行,和 Chain 相比优势不大。LangGraph 真正的能力来自条件边。
基本用法是:
g.add_conditional_edges( "agent", should_continue, { "tools": "tools", "end": END, }, )should_continue是一个函数,接收当前 State,返回tools或end,LangGraph 根据返回值选中对应的目标节点。这个机制就是后面 Agent 循环的核心。
4.4 子图、并行分支与循环检测
LangGraph 还支持子图(Subgraph)、并行分支和循环检测。子图适合把一段可复用的流程抽出来,并行分支适合同时执行多个工具调用。实际项目里经常会出现“一个节点里并行调用多个工具”的需求,新建子图的场景更多是为了复用基础流程。
不过,对初学者来说,先专注把状态、节点、条件边掌握好,已经能解决 80% 的 Agent 场景。子图和并行会在后面的进阶文章中单独展开。
5. AI-Agent 实战:让 LLM 使用工具做决策循环
5.1 Agent 的最简结构
一个最小可用的 Agent 需要四部分:
- 一个支持工具调用的模型。
- 一组工具函数。
- 一个 Agent 节点,负责调用模型并判断下一步。
- 一个工具执行节点,负责执行模型指定的工具。
模型通过bind_tools(tools)感知到工具的存在,返回结构化工具调用指令。LangGraph 根据模型返回内容,决定走工具节点还是直接结束。
5.2 定义一个天气查询工具
这里先定义一个简单的工具,方便演示机制。
from langchain_core.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市当前天气,输入城市名,返回天气描述。""" # 真实项目这里应该调用天气 API 或数据库 return f"{city}今天晴,25摄氏度"注意,@tool装饰器会把函数变成 LangChain Tool 对象,函数名、参数、docstring 都会作为模型理解的元数据。
5.3 完整 Agent 示例代码
下面这个例子实现了一个 ReAct 风格的 Agent:模型判断需要天气时调用get_weather,工具返回结果后模型继续推理,最后给出答案。
# file: basic_agent.py from typing import TypedDict, Literal from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, START, END @tool def get_weather(city: str) -> str: """查询指定城市当前天气,输入城市名,返回天气描述。""" return f"{city}今天晴,25摄氏度" class AgentState(TypedDict): messages: list model = ChatOpenAI(model="gpt-4o-mini", temperature=0) tools = [get_weather] model_with_tools = model.bind_tools(tools) def call_agent(state: AgentState): response = model_with_tools.invoke(state["messages"]) return {"messages": state["messages"] + [response]} def call_tool(state: AgentState): last_message = state["messages"][-1] new_messages = [] for tool_call in last_message.tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] if tool_name == "get_weather": result = get_weather.invoke(tool_args) else: result = f"未知工具: {tool_name}" new_messages.append({ "role": "tool", "tool_call_id": tool_call["id"], "content": result, }) return {"messages": state["messages"] + new_messages} def should_continue(state: AgentState) -> Literal["tools", "end"]: last_message = state["messages"][-1] if last_message.tool_calls: return "tools" return "end" g = StateGraph(AgentState) g.add_node("agent", call_agent) g.add_node("tools", call_tool) g.add_edge(START, "agent") g.add_conditional_edges("agent", should_continue, { "tools": "tools", "end": END, }) g.add_edge("tools", "agent") app = g.compile() result = app.invoke({ "messages": [{"role": "user", "content": "帮我查一下杭州天气"}] }) print(result["messages"][-1].content)5.4 运行与验证
运行前确保已经配置好模型 API Key:
python basic_agent.py如果一切正常,控制台会输出类似:
杭州今天晴,25摄氏度。这个例子虽然简单,但已经把 Agent 的完整循环跑通了。你可以改成别的工具,比如查数据库、调用 HTTP API、读文件,机制都一样。
5.5 对新手最重要的三个认知
第一,bind_tools(tools)这一步决定了模型是否知道工具的存在。如果模型返回结果中没有tool_calls,说明它没有识别到需要调用工具,问题可能出在 prompt 或工具描述上。
第二,条件边是 Agent 循环的核心。should_continue判断最后一次模型消息是否包含工具调用,包含就走工具节点,不包含就结束。
第三,节点函数修改 State 的正确姿势是返回新字典、拼接新列表,而不是直接原地修改旧列表。这样能避免并行执行时出现数据竞争。
6. 接入 MCP:让 Agent 连接外部系统
6.1 MCP 解决的核心问题
在真实项目中,Agent 不会只调用一个天气函数。它可能需要查数据库、读文件、操作网页、调用内部 API。如果没有统一标准,每接一个系统就要写一套工具适配代码,维护成本很高。
MCP 的思路是:把外部能力封装成标准化的 MCP Server,Agent 通过 MCP Client 连接协议、发现工具、调用工具。对 Agent 开发者来说,工具来源变了:以前是自己写函数,现在可以加载远端 MCP Server 暴露出来的工具。
6.2 Agent Skill 与 MCP 的区别
搜索热词里经常有人问“Agent Skill 和 MCP 有什么区别”。这里做个明确区分:
- Skill 是 Agent 侧的“能力包”,它包含提示词、调用模板、预设操作流程。它解决的是 Agent “会做什么、按什么方式做”的问题。
- MCP 是工具通信协议,解决的是 Agent “怎么访问外部系统”的问题。它更偏底层通信标准化。
可以理解为:Skill 更像人的工作经验总结,MCP 更像拿在手里的标准接口插头。两者不是替代关系,实际项目中常常一起出现。
6.3 MCP Server 配置示例
MCP 客户端通常通过一个配置文件来定义如何启动 Server。下面是一个常见格式:
{ "mcpServers": { "sqlite": { "command": "uvx", "args": ["mcp-server-sqlite", "--db-path", "./test.db"] } } }这个配置的含义是:客户端会启动command指定的命令,通过标准输入输出与这个进程通信。mcpServers下面可以有多个 Server,每个 Server 暴露自己的工具集。
这种command + args的启动方式,本质上让 Agent 团队可以“即插即用”各种外部能力,不需要为每个系统单独开发一套长连接服务。
6.4 在 LangGraph Agent 中加载 MCP 工具
下面的代码演示了如何用 MCP SDK 启动一个本地 Server,并加载它暴露出来的工具。
# file: load_mcp_tools_example.py import asyncio from langchain_mcp_adapters.tools import load_mcp_tools from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def load_tools(): server_params = StdioServerParameters( command="python", args=["my_mcp_server.py"], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: return await load_mcp_tools(session) if __name__ == "__main__": tools = asyncio.run(load_tools()) print(tools)这些返回的工具可以直接作为 LangChain Tool 使用,绑定给模型后,前面第 5 章的 Agent 循环不必改动太多,工具来源从本地函数变成了 MCP Server。
需要提醒的是,langchain-mcp-adapters的具体导入路径和 API 可能会随版本变化。安装后建议先运行一次官方示例确认接口,再集成到项目里。
6.5 MCP 使用的安全边界
使用 MCP 时要特别注意一点:一个 MCP Server 可能暴露文件读取、命令执行、网络请求等高危工具。生产环境中必须只连接可信的 Server,并且要做权限控制。Agent 在无人监督时自动执行工具带来的风险比代码调用大得多,因为 Agent 可能基于异常输入触发危险操作。建议在工具层增加白名单、参数校验和审计日志。
7. 智能体记忆 Memory:让 Agent 记住上下文
7.1 短期记忆:Memory Channel 与 Checkpointer
LangGraph 里经常会看到Memory Channel这个说法。它并不神秘,本质就是 State 中用于保存消息列表的通道,节点之间通过这个通道传递对话历史。没有这个 Channel,Agent 每轮之间就是孤立的。
但仅靠 State 还不够:默认情况下,一次invoke结束,状态就丢了。要实现“下一次调用还能记住上一次对话”,需要 Checkpointer。它负责把每一步状态持久化,可以用内存版,也可以用 SQLite、Redis 等外部存储。
7.2 用 MemorySaver 实现多轮记忆
在之前的basic_agent.py基础上,只需要改编译部分:
# 在 basic_agent.py 的基础上,将 app 编译部分改为: from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() app = g.compile(checkpointer=memory) config = {"configurable": {"thread_id": "user-123"}} first = app.invoke( {"messages": [{"role": "user", "content": "我叫张三,请记住"}]}, config, ) second = app.invoke( {"messages": [{"role": "user", "content": "我叫什么名字?"}]}, config, ) print(second["messages"][-1].content)这里关键点是thread_id。相同thread_id的多次调用共享同一个状态,相当于同一个会话;不同thread_id之间互相隔离。
如果运行时发现模型不记得上一轮内容,优先排查是不是没有传config,或者thread_id是否一致。
7.3 长期记忆:外部存储
短期记忆解决的是会话内上下文。长期记忆需要把信息放到外部系统,比如 Redis、MySQL、向量数据库。典型做法是:
- 用户第一次使用时,Agent 提取关键信息(偏好、称呼、禁止事项)写入存储。
- 下次会话开始时,先根据
user_id加载用户画像,放进 State 的初始字段。 - Agent 生成回答时可以参考这些长期信息。
# 伪代码示意 def load_user_profile(user_id: str) -> str: # 从 Redis / MySQL / 向量库读取用户画像 return "用户偏好:喜欢简洁回答,对价格敏感" def save_user_profile(user_id: str, summary: str) -> None: # 写入外部存储 pass state = { "messages": [], "user_profile": load_user_profile("user-123"), } # 调用 Agent 时有意识地把 user_profile 拼进 system prompt长期记忆的难点不在于写代码,而在于设计“什么时候写入、什么时候更新、什么时候删除”。不是所有对话内容都值得长期保存,盲目存储反而会导致隐私合规风险。
7.4 记忆设计的三条建议
第一,短期记忆使用 LangGraph 的 Checkpointer 就够了,不要自己造轮子。第二,长期记忆要从业务需求出发,明确哪些信息需要跨会话保留。第三,长期保存的信息必须提供用户查看和删除的入口,这是工程底线,不只是功能问题。
8. 常见问题与排查思路
LangGraph 项目最常见的错误,集中在依赖版本、工具调用、状态隔离、MCP 连接这几类。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 安装时依赖冲突 | langgraph 与 langchain 版本不匹配 | 执行pip check查看依赖树 | 使用虚拟环境,统一锁版本 |
| 模型完全不调用工具 | 没有执行 bind_tools,或工具描述不清 | 打印模型原始返回内容 | 绑定工具,改进工具描述 |
| 报错 last_message.tool_calls 不存在 | 模型返回普通文本,没有结构化工具调用 | 检查模型是否支持 function calling | 更换支持工具调用的模型 |
| 多轮对话中模型不记得上文 | 每次 invoke 没有传 thread_id | 检查 config 是否传递 | 使用相同 thread_id |
| MCP 工具加载为空 | Server 启动失败或命令路径错误 | 先单独启动 Server 看日志 | 检查 command、args、工作目录 |
| 并发调用时状态互相污染 | 多个请求共用了同一个 MemorySaver 实例 | 检查是否复用了变量 | 按会话隔离图实例或合理使用线程隔离 |
| 本地模型工具返回格式不对 | 模型厂商实现不兼容 tool_calls 标准 | 查看工具返回消息格式 | 增加适配层或换模型 |
这里特别强调第一条:LangChain 生态版本更新很快,很多“网上看着能跑,本地跑不起来”的问题,根因都是依赖版本不一致。建议创建项目时使用虚拟环境,写一个requirements.txt把关键包版本固定下来。
9. 最佳实践与工程建议
9.1 版本锁定
LangChain、LangGraph、LangChain OpenAI 适配器、MCP SDK,这些包的 API 都可能小版本升级后发生变化。生产环境不要依赖“最新版”,要使用锁文件或固定版本号。
pip freeze > requirements.txt9.2 State 设计要小
State 里的字段越多,LangGraph 每一步持久化的成本越高,也越容易出错。不要把大段文件内容、完整原始响应都塞进 State。能存 ID 就存 ID,能存摘要就存摘要,需要细节时再通过工具获取。
9.3 超时与重试
Agent 是多步调用,不是一次 HTTP 请求。任何一个工具调用都可能慢、失败、卡住。建议在工具节点加上超时控制,在关键节点加上重试逻辑。
9.4 人类介入
生产级别的 Agent 不应完全无人值守。LangGraph 支持中断和恢复机制,可以在执行到关键步骤前“暂停”,等人工确认后再继续。涉及支付、删除、外发消息等操作时,这个能力几乎必备。
9.5 可观测性
Agent 系统比普通后端系统更难排查,因为最终结果由模型的多步决策决定。建议把每一轮的输入输出、工具调用参数、返回结果、耗时全部记录下来。调试时回放完整决策路径,比只看最终输出有用的多。
9.6 安全边界
涉及外部工具调用时,先问三个问题:这个工具会不会修改数据?会不会发起对外请求?参数是不是用户可控?如果都是“否”,可以交给无人值守流程;只要有一个“是”,都要加权限和人工确认。
10. 总结与后续学习方向
这篇文章从 Chain 到 Graph 的演进讲起,通过一个最小示例让你理解了 State、Node、Edge 和条件边,然后实现了一个真正的 ReAct Agent,最后讲解了 MCP 和记忆 Memory 的落地方式。如果你能把第 5 章的示例代码完整跑通,再自己换一个其他类型工具,基本上就算入门了 LLM Agent 开发。
接下来可以按这个顺序深入:先自己扩展工具集,比如加一个 HTTP 请求工具,让 Agent 能查真实 API;然后给 Agent 接入 Checkpointer,实现多轮记忆;之后再了解并行分支和子图;最后再研究生产部署,包括权限、监控、人工介入和模型降级。
LangGraph 的优势不在于一步到位解决所有问题,而在于它把复杂的 Agent 流程变成了一张清晰可见的图。你在图上加节点、加边、加状态,逻辑始终是可控的。这也是我建议新项目直接选择 LangGraph 的原因。建议把示例代码保存下来,之后写自己的 Agent 时直接在上面修改,能少走很多弯路。