如果你最近在关注 AI 大模型应用开发一定绕不开“智能体Agent”这个词。很多人以为智能体就是用 Prompt 套一个大模型接口实际做起来才发现工具调用、多轮记忆、条件分支、异常重试这些逻辑一旦叠加代码很快就变成一团乱麻。LangGraph 就是用来解决这类问题的图编排框架它把智能体流程拆成“节点”和“边”让状态流转变得可控制、可复用。本文会从零开始讲清楚 LangGraph 的核心概念、环境安装、关键组件再通过一个带天气查询能力的智能客服案例把条件路由、工具调用、状态管理完整串起来。无论你是刚入门 AI 应用开发还是已经用过 LangChain这篇文章都能帮你快速建立一套可以落地的 LangGraph 实战思路。1. 智能体开发为什么需要 LangGraph1.1 从“链”到“图”的思维转变先看一个简单的场景用户问“北京今天天气怎么样”一个智能客服需要做几件事判断用户意图确定需要调用天气查询工具。调用工具获取北京的天气数据。把工具返回的数据整理成自然语言回复用户。如果业务流程是固定线性的用传统的代码顺序调用也能实现。但真实业务往往不是线性流程同一轮对话里用户可能问完天气又问股票智能体需要判断调用哪个工具。工具返回异常时可能需要重试或者换一个工具。多轮对话中用户说“那上海呢”智能体要能识别省略指代继续查天气。有的业务并行执行多个查询再把结果合并。这种场景下“链式调用”很难表达分支和循环。LangGraph 的核心思路是把整个流程建模成一张图每个加工步骤是一个节点Node节点之间通过边Edge连接运行时会沿着边在节点间传递状态State。1.2 LangGraph 与 LangChain 的区别LangChain 是最早火起来的 LLM 应用开发框架之一提供了大量工具集成、Chain 封装、RAG 组件、Memory 等能力。它解决的是“大模型和外部系统怎么连接”的问题。LangGraph 则更聚焦在“流程怎么编排”这一层。它的前身是 LangChain 生态内的一个实验组件后来独立成单独的项目定位是构建有状态、可观察、可恢复的 Agent 工作流。两者的关系可以这样理解维度LangChainLangGraph核心抽象Chain链Graph图流程形态线性为主分支能力弱支持分支、循环、并行、子图状态管理依赖外部 Memory内置 State 机制适用场景RAG、简单的链式调用复杂 Agent、多工具协作、条件路由灵活度封装程度高上手快更贴近底层可控性强实际项目中两者不是二选一。LangGraph 的节点函数内部完全可以调用 LangChain 的组件比如 ChatOpenAI、PromptTemplate、文档加载器等。很多项目是 LangChain 提供“零件”LangGraph 负责“组装流水线”。1.3 LangGraph 能解决哪些具体问题从工程角度看LangGraph 最实用的价值有三个第一是状态统一管理。每个节点都能读取和更新全局 State不用手动维护一堆局部变量多轮对话的历史消息、上下文变量都放在 State 里。第二是条件路由能力。通过add_conditional_edges可以根据模型输出、工具返回结果或自定义规则动态决定下一步走向这是实现 ReAct 模式 Agent 的基础。第三是支持持久化与断点恢复。配合 Checkpointer可以把每一步的中间状态保存下来进程重启后还能从断点继续执行这对长任务和人工审核场景非常关键。2. 环境准备与安装配置2.1 环境要求LangGraph 是基于 Python 的框架建议使用 Python 3.9 及以上版本。如果你同时使用 LangChain 生态的组件不同版本的 API 会有差异建议先创建一个干净的虚拟环境避免把系统 Python 环境搞乱。操作系统方面Windows、macOS、Linux 都支持。示例中大部分命令在三个平台都能直接使用Windows 用户注意把python命令换成python -m的写法或者用 Anaconda Prompt 执行。2.2 安装 LangGraph 及依赖使用 pip 安装命令很简单pip install langgraph一个完整的智能体项目通常还会用到 LangChain 的大模型封装层和工具库所以建议一起安装pip install langchain-core langchain-openai如果你需要把对话状态持久化到本地 SQLite可以安装官方提供的内存或 SQLite 检查点实现pip install langgraph-checkpoint-sqlite版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路。LangGraph 的接口迭代比较快遇到 API 变化时以你安装版本的官方文档为准。2.3 验证安装是否成功安装完成后进入 Python 交互环境执行下面两行代码确认导入无异常import langgraph import langchain_core print(langgraph imported successfully) print(langchain_core imported successfully)如果输出两条成功信息说明基础依赖已经就绪。接下来在编译图谱时如果遇到langgraph.prebuilt导入失败说明你安装的版本较新或较旧可以先升级到最新版pip install --upgrade langgraph2.4 推荐的项目结构一个小型智能体项目不需要复杂的目录结构但建议从一开始就保持清晰agent_project/ ├── .env # 存放 API Key ├── requirements.txt # 依赖清单 ├── agent/ │ ├── __init__.py │ ├── graph.py # 搭建图谱 │ ├── nodes.py # 定义节点函数 │ ├── state.py # 定义 State │ ├── tools.py # 定义工具 │ └── config.py # 读取配置 └── run.py # 入口脚本API Key 不要直接写在代码里建议通过环境变量或者.env文件管理。3. LangGraph 核心组件拆解3.1 State贯穿全图的共享状态State 是 LangGraph 里最基础也最容易忽略的概念。它本质上是一个 Python TypedDict定义了智能体运行过程中需要维护的所有数据。from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] weather_info: str这里有两个关键点第一Annotated[list, add_messages]中的add_messages是一个 reducer 函数。普通字段如果直接用list后一个节点返回的新列表会直接覆盖旧列表加了 reducer 之后新消息会追加到已有消息列表中而不是覆盖。第二State 是整个图运行期间的“全局变量”。每个节点函数接收当前的 State返回一个 dictdict 里只写需要更新的字段LangGraph 会自动把返回值合并进 State。3.2 Node最小的执行单元Node 是图中的节点本质就是一个 Python 函数接收 State 作为输入返回一个 dict 作为输出。def my_node(state: AgentState): # 拿到当前状态 last_user_message state[messages][-1] # 处理业务逻辑 result do_something(last_user_message) # 返回需要更新的字段 return {messages: [result]}节点函数的返回值不一定包含所有字段只需要返回需要更新的部分。这种设计让每个节点职责单一方便测试和复用。在图中注册节点的方法from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(my_node, my_node) builder.add_edge(START, my_node) builder.add_edge(my_node, END)START是图的入口节点END是终止节点。所有流程都必须从START开始最终到达END。3.3 Edge 与条件路由边决定节点之间的流转方向。普通边是无条件的比如add_edge(node_a, node_b)表示 nA 执行完后一定执行 nB。条件路由通过add_conditional_edges实现。你需要提供一个路由函数返回值决定下一步进入哪个节点。def route_after_agent(state: AgentState): # 根据最后一条消息判断 last_message state[messages][-1] if hasattr(last_message, tool_calls) and last_message.tool_calls: return tools return finish builder.add_conditional_edges( agent, route_after_agent, { tools: tools, finish: END, }, )条件路由是 LangGraph 实现 Agent 自动决策的关键。大模型返回的内容中包含tool_calls时就说明它想调用某个工具没有工具调用意图时就可以直接结束或进入回答生成节点。3.4 Checkpointer持久化与记忆默认情况下图执行完一次就结束了状态只存在于内存中。如果要做多轮对话或者需要从失败中断点恢复就需要引入 Checkpointer。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(checkpoints.sqlite) app builder.compile(checkpointermemory)运行图时通过config传入thread_id来区分不同的会话config {configurable: {thread_id: user-001}} result app.invoke( {messages: [{role: user, content: 北京天气怎么样}]}, configconfig, )相同thread_id的消息会被连续保存在同一个会话上下文中。后续再问“那上海呢”模型能看到前面的对话历史。4. 代码实战构建一个会调用工具的智能客服4.1 案例需求分析下面我们做一个完整可运行的案例一个带天气查询能力的智能客服。需求拆解用户可以正常聊天智能体直接回答。用户询问天气时智能体判断需要调用天气工具。工具返回天气数据后智能体整理成自然语言回复。多轮对话中智能体能记住之前的上下文。对应的图结构如下START - agent - 有工具调用? - tools - agent | ----- 无工具调用 - END也就是经典的 ReAct 循环Agent 决定是否调用工具如果调用把工具结果放回消息列表再次交给 Agent 组织答案。4.2 定义 Agent 状态from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages]这个 State 只有一个字段messages专门用来承载对话历史。add_messagesreducer 保证消息可以持续追加。4.3 创建天气查询工具工具函数使用tool装饰器定义LangChain 会自动把函数的 docstring 和参数解析成模型能够识别的工具描述。from langchain_core.tools import tool tool def get_weather(city: str) - str: 查询指定城市的实时天气。 Args: city: 城市名称例如北京、上海。 Returns: 包含天气、温度、风力等信息的字符串。 weather_map { 北京: 晴26℃风力2级, 上海: 多云28℃风力3级, 广州: 阵雨30℃风力2级, 深圳: 雷阵雨29℃风力3级, } return weather_map.get(city, f{city}暂无天气数据)实际项目中这里可以替换成真实天气 API 的调用比如拼接城市参数请求气象服务端的接口解析返回的 JSON再整理成字符串。4.4 初始化模型并绑定工具按下面方式创建大模型对象并把工具绑定到模型上。from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, temperature0, api_keyYOUR_API_KEY, # 换成你自己的 Key base_urlhttps://api.openai.com/v1, ) tools [get_weather] llm_with_tools llm.bind_tools(tools)bind_tools是关键一步。绑定之后模型在回答时如果判断需要查天气返回的内容中就会包含tool_calls字段里面写明了要调用哪个工具、传什么参数。大模型本身不执行工具只是“请求”调用工具真正执行是由程序来完成。如果没有 OpenAI 的 Key也可以换成其他兼容 OpenAI 接口格式的服务只要把base_url和model改为自己的配置即可。4.5 定义节点函数主节点是agent_node它的职责是接收整个消息列表让模型决定下一步动作。def agent_node(state: AgentState): system_prompt ( 你是智能客服助手。当用户询问天气时你必须调用 get_weather 工具。 工具返回结果后再根据结果组织回答。其他问题可直接回答。 ) messages [{role: system, content: system_prompt}] messages state[messages] response llm_with_tools.invoke(messages) return {messages: [response]}注意这里把系统提示词放在消息列表头部再拼接历史消息。模型返回的response是一个AIMessage对象如果它决定调用工具这个对象上会带有tool_calls属性。4.6 搭建图谱并配置条件路由接下来把节点和边组装起来。工具节点直接用 LangGraph 提供的ToolNode它会自动执行模型请求的工具。from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode builder StateGraph(AgentState) builder.add_node(agent, agent_node) builder.add_node(tools, ToolNode(tools)) builder.add_edge(START, agent) def should_continue(state: AgentState): last_message state[messages][-1] if hasattr(last_message, tool_calls) and last_message.tool_calls: return tools return end builder.add_conditional_edges( agent, should_continue, { tools: tools, end: END, }, ) builder.add_edge(tools, agent) app builder.compile()整个流程可以这样理解从START进入agent节点。agent节点调用模型模型可能给出普通回复也可能请求调用工具。should_continue检查最后一条AIMessage如果存在工具调用意图进入tools节点。tools节点执行真实工具把工具返回的消息追加到历史里然后回到agent。模型拿到工具结果后组织最终回复此时没有新的工具调用流程走向END。4.7 运行并观察输出写一个简单的入口脚本调用编译好的图。if __name__ __main__: config {configurable: {thread_id: demo-001}} result app.invoke( {messages: [{role: user, content: 北京今天天气怎么样}]}, configconfig, ) print(result[messages][-1].content)第一次运行后你可以接着发起第二轮对话测试多轮记忆能力result app.invoke( {messages: [{role: user, content: 那上海呢}]}, configconfig, ) print(result[messages][-1].content)因为两次使用了相同的thread_id第二次提问时模型能根据历史记录判断“那上海呢”指的是查询上海的天气而不会把这句话当成闲聊。预期输出大致为北京今天晴气温26摄氏度风力2级。适合外出活动。 上海今天多云气温28摄氏度风力3级。体感较热注意补水。如果你的模型没有正确触发工具调用大概率是system_prompt指令不够明确或者模型本身不支持 function calling。4.8 简单并行分支示例LangGraph 还支持同时执行多个互不依赖的分支。比如用户既想知道天气又想知道当前时间可以并行调用两个工具。一个简化写法如下from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolNode class MultiState(TypedDict): messages: Annotated[list, add_messages] weather_result: str time_result: str builder StateGraph(MultiState) builder.add_node(agent, agent_node) builder.add_node(weather, ToolNode([get_weather])) builder.add_node(time, ToolNode([get_current_time])) builder.add_edge(START, agent) # 根据工具调用情况可以同时进入多个分支 builder.add_edge(agent, weather) builder.add_edge(agent, time) builder.add_edge(weather, merge) builder.add_edge(time, merge)并行分支适合工具之间没有依赖关系的场景能明显降低整体响应耗时。不过分支越多状态合并的复杂度也会上升建议只有在性能确实成为瓶颈时再用。5. MCP 在智能体中的应用5.1 MCP 协议到底是什么MCP 全称 Model Context Protocol即模型上下文协议是一套用于大模型应用与外部数据、工具交互的开放协议。在没有 MCP 之前每个工具都要写一套自定义的接口封装调用天气 API 要自己构造 HTTP 请求访问数据库要自己写连接代码对接内部系统要处理各自的鉴权方式。工具数量一多集成成本非常高。MCP 的出现把“工具接入”变成了标准化操作。它定义了一套基于 JSON-RPC 的消息格式把外部能力抽象为三类Tools可被模型调用的函数。Resources可读取的上下文数据。Prompts可复用的提示词模板。只要服务端实现了 MCP 协议客户端就能通过一套统一的方式发现并调用它的能力。5.2 为什么智能体需要 MCP对于单个智能体项目直接在 LangGraph 里用tool注册函数完全够用。但到了中大型项目问题就变了企业内可能有多个团队各自维护一套工具服务。同一个工具可能被多个 Agent 复用。工具上线、下线、鉴权需要统一管理。MCP 的价值在于把工具从“代码里的函数”变成“独立运行的服务”。团队 A 发布一个天气 MCP Server团队 B 和团队 C 的智能体都能直接接入不需要重新实现一遍调用逻辑。在 LangGraph 生态中LangChain 官方提供了 MCP 适配器可以把 MCP Server 暴露的工具转换成 LangChain 的 Tool 对象再绑定给模型。具体接入方式会因为协议实现和版本不同而变化建议以官方文档中的适配器说明为准。5.3 Agent Skill 与 MCP 的区别最近智能体领域还经常听到一个词Agent Skill。它和 MCP 容易混淆简单区分一下MCP 解决的是“工具如何标准化暴露和调用”的问题偏底层协议。Agent Skill 偏向上层封装通常包含“使用某项能力的完整技能描述”比如一段特定的提示词、若干工具调用策略、处理流程甚至可以是一段可执行的小程序。一个 Skill 内部可能调用多个 MCP ToolMCP 更像基础能力Skill 更像组合技能。实际选型时不用纠结二选一它们处于不同抽象层级可以配合使用。6. 常见问题与排查思路问题现象常见原因解决思路安装后无法导入 langgraphPython 版本过低或包被破坏升级 Python 到 3.9删除虚拟环境后重新安装图编译报错Invalid reducerState 字段没有正确使用 Annotated 包装检查 TypedDict 中的字段定义需要追加时用 add_messages模型返回的内容没有 tool_calls模型不支持 function calling或提示词不够明确换用支持工具调用的模型强化 system prompt 中的指令工具执行后流程没有回到 agent缺少add_edge(tools, agent)确认条件路由每个分支都有终点并且工具节点后面接回主节点多轮对话丢失上下文没有配置 Checkpointer 或 thread_id 不统一使用 SqliteSaver 并保证同一会话传入相同 thread_id工具结果格式不匹配返回内容不是字符串模型解析困难在工具函数里把结果整理成清晰的自然语言或 JSON 字符串并发执行时状态互相覆盖访问了共享数据结构不要把可变对象直接放进 State尽量用不可变对象或深度拷贝模型反复调用同一个工具不结束工具返回结果不明确导致死循环在提示词中限制工具调用次数或在循环节点中设置最大轮数遇到运行时问题排查顺序建议是先确认模型是否真的返回了tool_calls能直接把问题和流程的核心原因定位到“模型层”还是“流程层”。再检查节点函数的返回值是否正确State 字段是否被 reducer 正确合并。最后检查边的连接尤其是条件路由的映射表是否写全防止某个分支没有出口导致流程卡死。7. 最佳实践与工程建议7.1 State 设计原则State 是整张图的“公共总线”设计得越精简问题越少。第一不要把临时变量全塞进 State。节点内部自己能计算的中间结果就应该留在节点内部不暴露到全局。只有需要在多个节点间共享的数据才放进 State。第二注意字段更新语义。对于messages这类需要累积的字段必须用Annotated[list, add_messages]提供合并策略对于普通字段后写入的节点会直接覆盖容易造成数据丢失。第三State 中的对象尽量是不可变的或者在返回时重新创建新对象避免多个节点共享同一个可变对象引发并发修改问题。7.2 节点与路由的职责划分一个节点只做一件事。比如“判断是否调用工具”和“组织最终回复”可以拆成两个节点虽然有时候可以合并在同一个agent_node里但职责分开更利于调试。路由函数要写得像一个纯函数不执行任何副作用的业务逻辑只读 State返回一个分支名称。这样你可以单独写单元测试覆盖各种分支情况。条件路由的映射表建议把 key 定义成常量避免在add_conditional_edges和路由函数里各写一遍字符串减少拼写错误。7.3 错误处理与重试机制工具调用涉及外部系统任何网络请求都可能失败。工具节点内部要做好异常捕获和超时控制把错误信息返回给模型让模型有机会修正参数或者改用其他工具。对于重试逻辑LangGraph 的add_node支持传入重试配置也可以自己在节点函数里实现循环重试。无论用哪种方式都要限制最大重试次数防止死循环拖垮服务。7.4 日志与可观测性智能体项目最难排查的就是“模型为什么这么决策”。所以从一开始就要把日志打全每次调用模型时记录输入的 messages 数量、token 使用量。模型返回后记录是否有tool_calls调用的是什么工具、传了什么参数。工具执行后记录耗时和返回结果摘要。路由决策时记录走了哪个分支。这样即使线上出了问题也能顺着日志还原当时的决策链路。生产环境建议接入 OpenTelemetry 之类的链路追踪体系LangGraph 本身也提供了 Tracing 能力配置后可以在控制台查看每一步的执行详情。7.5 安全与权限边界智能体工具能调用的权限一定要遵循最小权限原则。每个工具只赋予它完成业务所需的最小数据权限不要图省事给一个“万能工具”。特别是在工具涉及文件读写、数据库操作、对外发送消息时建议增加人工确认步骤。LangGraph 的interrupt机制可以暂停流程等待人工审核适合高危操作场景。对用户的输入也保持基本的防护意识不要在提示词里直接拼接不可信的 SQL、Shell 命令或者 HTML 片段。虽然大模型本身有一定过滤能力但业务层仍然要自己做参数校验。7.6 测试策略智能体项目的测试比传统业务代码难因为模型输出有随机性。建议这样分层路由函数做纯单元测试输入构造好的 State断言返回分支正确。工具函数单独做单元测试重点验证超时、异常、边界值。节点函数用 mock 的模型响应测试不真正调用大模型。集成测试只跑固定的几条业务链路用相同的 temperature 和 prompt保证结果基本稳定。回归测试时把历史问题整理成回归用例集防止升级依赖或修改提示词后老问题重新出现。8. 总结与学习路线这篇文章从智能体开发的实际痛点出发介绍了 LangGraph 的定位、核心组件和完整实战流程。你已经掌握的关键点包括LangGraph 与 LangChain 的关系LangChain 提供组件LangGraph 负责图编排。三个核心概念State 管理共享数据Node 封装执行逻辑Edge 控制流转方向。条件路由是智能体的核心机制模型返回的工具调用意图触发分支流转。用 Checkpointer 实现跨轮次记忆和多会话隔离。MCP 协议用于标准化工具接入LangGraph 生态提供了对应的适配能力。下一步可以继续深入学习的方向ReAct 模式的完整实现研究模型如何“思考-调用-观察-总结”。子图Subgraph的使用把复杂业务拆成多个可复用的小图。人工审核节点interrupt的用法实现高危操作审批流程。LangGraph 与 RAG 的结合在节点中注入检索结果。最后给一个建议不要一上来就追求复杂的并行编排和多 Agent 协作先把单个节点的职责写清楚把一条主流程跑通再逐步增加分支和异常处理。框架的复杂度应该由业务复杂度驱动而不是反过来。如果本文对你有帮助可以收藏备用也欢迎动手把示例代码跑起来改成你自己的业务场景试一遍。