LangGraph实战：从零构建可控的Agent流程

如果你正在做 Agent 开发或者准备从 LangChain 进入 Agent 工程化最近一定频繁听到 LangGraph 这个名字。但真正动手用过之后很多人会有一种共同的困惑它不过是一个“画流程图”的框架为什么社区把它捧得这么高我的判断是LangGraph 真正改变的不是“能不能写 Agent”的问题而是“Agent 流程能不能被控制”的问题。它把 Agent 从一段不可控的提示词循环变成了一个有状态、可路由、可恢复的图执行引擎。这篇文章不打算讲太多情怀直接从最小示例开始带你跑通节点、状态、条件路由和工具调用最后给出一份能直接改造成生产项目的 Agent 骨架。文章会围绕五个关键点展开LangGraph 和 LangChain 到底是什么关系State、Node、Edge 三个核心概念如何理解条件路由怎么做一个带工具调用的天气 Agent 怎么实现以及最容易踩坑的几个工程问题怎么排查。读完你至少能独立写完一个可运行的 LangGraph Agent并知道下一步该往哪些方向深入。1. 这篇文章真正要解决的问题很多开发者是带着 LangChain 的使用经验转过来的。在 LangChain 里写一个简单的 Agent 确实很愉快定义工具、绑定模型、调用 AgentExecutor三步就能跑通。可一旦项目规模变大问题就会出现。第一个痛点是流程不可控。AgentExecutor 内部是一个封装好的 ReAct 循环开发者在外面只能看到输入和输出。中间模型调用了哪些工具、每一步进了哪个分支、为什么多调了一次工具出了问题很难快速定位。第二个痛点是状态难管理。多轮对话场景下历史消息、临时变量、工具返回结果混在一起容易出现串话和消息爆炸。第三个痛点是扩展困难。想在某个步骤中间插入一个人工审批节点或者做成“先检索、再判断、后回答”的复杂流程AgentExecutor 的黑盒机制很难支持。LangGraph 解决的正是这三个问题。它把业务流程定义成一张有向图每个节点是一个函数每条边决定下一步去哪State 负责在节点之间传递数据。流程的所有转折点都摆在明面上开发者可以精确控制每一条路径也可以在任意节点暂停、恢复注入外部输入。如果你是这几类读者这篇文章会比较有价值已经写过一些 LangChain 脚本但对 Agent 内部机制仍然模糊正在做 AI 应用开发想了解 Agent 框架选型团队想把 Agent 流程工程化需要可控的状态管理和分支逻辑准备面试大模型应用开发岗位需要系统梳理 LangGraph 的知识框架。如果你完全不了解 LangGraph也能从零开始跟练如果你已经看过官方文档则可以重点看后面的工程实践和踩坑部分。2. LangGraph 核心概念它和 LangChain、Agent 究竟是什么关系2.1 一句话区分 LangChain 和 LangGraphLangChain 是一个大模型应用工具箱提供模型封装、Prompt 模板、文档加载、向量存储接口等能力。LangGraph 是一个流程编排框架它用图的思路来组织 Agent 的执行逻辑。用工厂流水线来类比LangChain 提供的是各种机器零件——电机、传送带、机械臂。LangGraph 解决的是流水线怎么架设——先放哪个机器物料怎么流转什么条件下走哪条支线整个产线怎么启停和检修。更直白的理解是在 LangGraph 里调用大模型通常还是要借助 LangChain 的 ChatOpenAI 等接口但 LangGraph 本身只负责“调度”不负责“生成”。它定义好谁先执行、下一步去哪、状态如何更新真正的大模型推理发生在你写的节点函数内部。2.2 Agent 在 LangGraph 眼里是什么传统观点认为 Agent 是“能自动调用工具的模型”。LangGraph 的观点更工程化Agent 是一张图图上每个节点承担一种职责。典型流程如下agent 节点让大模型思考判断要不要调用工具tools 节点执行工具调用把结果返回给模型条件边根据模型输出决定下一步是继续调用工具还是直接结束。这个循环本质上就是 ReAct 范式思考Thought、行动Action、观察Observation。LangGraph 最大的贡献是把这个循环从“框架内部的黑盒”变成了“开发者可以逐节点控制和观察的图”。2.3 与其他 Agent 框架的对比现在主流的 Agent 框架思路并不相同。AutoGen 走的是多智能体对话路线强调多个 Agent 之间通过消息协作CrewAI 偏向角色分工让不同角色按流程配合LangGraph 则采用显式图结构把所有流程画在明面上。框架核心思路适用场景特点LangChain AgentExecutor封装好的 ReAct 循环快速验证 Demo上手快控制力弱LangGraph有向状态图编排生产级流程控制状态可控适合复杂流程AutoGen多智能体对话多角色协作研究对话驱动调度相对隐式CrewAI角色分工任务队列固定流程团队协作表达直观灵活度有限MetaGPTSOP 流程驱动软件公司式多角色协作强流程偏重上游设计从材料看LangGraph 在工程可控性和复杂流程表达能力上更突出。如果你的需求是“一个能上线、出问题能排查、流程能拆解”的 Agent它显然更合适。2.4 三个核心概念State、Node、EdgeLangGraph 的概念层非常少掌握三个就够State 是全局状态。所有节点共享同一个状态对象每个节点从 State 读取输入通过返回值更新 State。它是连接所有节点的数据总线。Node 是执行单元。每个节点就是一个 Python 函数接收当前 State返回一个字典表示要对 State 做的更新。Edge 是节点之间的连接。普通边表示“执行完 A 一定去 B”条件边表示“根据当前状态判断去 B 还是去 C”。节点返回的状态更新默认是覆盖旧值。如果希望合并而不是覆盖需要用Annotated配合operator.add定义合并行为。这是 LangGraph 新手最容易忽略的地方后面会重点演示。3. 环境准备与基础安装LangGraph 是一个 Python 库环境准备相对简单。建议使用 Python 3.10 及以上版本并使用虚拟环境隔离项目依赖。mkdir langgraph-demo cd langgraph-demo python -m venv .venv source .venv/bin/activate # Windows 用户执行: .venv\Scripts\activate安装依赖pip install --upgrade pip pip install langgraph langchain-openai python-dotenv如果你打算使用本地模型替代 OpenAI 接口可以额外安装pip install langchain-ollama安装完成后在项目根目录创建.env文件OPENAI_API_KEY你的_API_Key注意生产环境不要将 Key 提交到代码仓库本地调试也建议把.env加入.gitignore。验证安装是否成功python -c from langgraph.graph import StateGraph; print(LangGraph OK)如果输出LangGraph OK说明环境已经就绪。4. 第一个 LangGraph 应用理解 State、Node、Edge先不引入大模型写一个纯 Python 的最小图跑通 LangGraph 的执行机制。创建文件demo_graph.py# 文件路径demo_graph.py from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class DemoState(TypedDict): count: int messages: Annotated[list[str], operator.add] def add_one(state: DemoState) - dict: return {count: state[count] 1, messages: [add_one 节点执行完毕]} def double_node(state: DemoState) - dict: return {count: state[count] * 2, messages: [double_node 节点执行完毕]} # 构建图 graph StateGraph(DemoState) # 添加节点 graph.add_node(add_one, add_one) graph.add_node(double_node, double_node) # 添加边 graph.add_edge(START, add_one) graph.add_edge(add_one, double_node) graph.add_edge(double_node, END) # 编译 app graph.compile() # 执行 result app.invoke({count: 1, messages: []}) print(result)这段代码做的事情很直观从START进入add_one节点count 变为 2然后进入double_nodecount 变为 4最后到达END。运行方式python demo_graph.py预期输出{count: 4, messages: [add_one 节点执行完毕, double_node 节点执行完毕]}这里有两个关键细节需要留意。第一messages字段使用了Annotated[list[str], operator.add]。如果没有这行声明后面的节点返回值会直接覆盖前面的消息加了operator.add之后LangGraph 会把多次返回的列表合并起来。在 Agent 场景里我们需要累积每一轮的 AI 消息和工具消息这个合并机制非常重要。第二节点函数返回的是字典里面只写需要更新的字段不需要返回完整 State。LangGraph 会以当前 State 为基础把返回值合并进去再传给下一个节点。尝试一个错误示例如果把第一段代码里count想成“先翻倍再加一”但实际执行顺序却是按边定义的顺序。这提醒我们LangGraph 的执行顺序由图的拓扑结构决定不是由节点的定义顺序决定。这也是图编排和普通函数调用的核心差异。5. 条件路由让 Agent 学会自己选择下一步LangGraph 最常用的能力之一是条件路由。为什么要做条件路由因为在真实 Agent 场景中模型不是每次都需要调用工具。用户问“今天北京天气怎么样”可能需要工具用户说“你好”则可以直接回答。如果流程写死“必须调用工具”既浪费费用也影响响应速度。条件路由通过add_conditional_edges实现。它的作用是根据当前 State 的内容动态决定下一步执行哪个节点。继续扩展前面的示例编写route_graph.py# 文件路径route_graph.py from typing import TypedDict, Annotated, Literal import operator from langgraph.graph import StateGraph, START, END class RouteState(TypedDict): question: str use_tool: bool final_answer: str messages: Annotated[list[str], operator.add] def router_node(state: RouteState) - dict: # 这里仅用于演示真实项目会调用大模型判断 needs_tool 天气 in state[question] or 查询 in state[question] return {use_tool: needs_tool, messages: [f路由判断需要工具{needs_tool}]} def call_tool_node(state: RouteState) - dict: return { final_answer: 北京今天 26 度多云, messages: [工具节点返回天气数据], } def direct_answer_node(state: RouteState) - dict: return { final_answer: 你好我是 AI 助手, messages: [直接回答节点无需调用工具], } # 路由函数返回值必须匹配映射表中的 key def route_decision(state: RouteState) - Literal[call_tool_node, direct_answer_node]: if state[use_tool]: return call_tool_node return direct_answer_node graph StateGraph(RouteState) graph.add_node(router_node, router_node) graph.add_node(call_tool_node, call_tool_node) graph.add_node(direct_answer_node, direct_answer_node) graph.add_edge(START, router_node) graph.add_conditional_edges( router_node, route_decision, { call_tool_node: call_tool_node, direct_answer_node: direct_answer_node, }, ) graph.add_edge(call_tool_node, END) graph.add_edge(direct_answer_node, END) app graph.compile() # 测试两条路径 result1 app.invoke({question: 北京今天天气怎么样, messages: []}) print(result1[final_answer]) result2 app.invoke({question: 你好, messages: []}) print(result2[final_answer])add_conditional_edges有三个关键参数起始节点即从哪个节点之后开始判断路由函数接收当前 State返回一个字符串映射表把路由函数的返回值映射到目标节点名。执行结果路由判断需要工具True 北京今天 26 度多云 路由判断需要工具False 你好我是 AI 助手这个例子的核心价值在于流程终于在“执行前”有了分支判断能力。实际 Agent 的业务流程本质上就是一个不断在“模型回答”和“工具执行”之间做条件路由的循环。6. 完整实战构建一个带工具调用的天气 Agent现在把前几章的知识点串起来实现一个真正的 Agent大模型根据用户问题决定是否调用天气工具工具返回结果后大模型再组织最终回答。这里使用langchain-openai接入大模型langchain_core.tools定义工具langgraph负责编排循环。准备工作已经在前面的环境章节完成。创建weather_agent.py# 文件路径weather_agent.py import json import operator from typing import TypedDict, Annotated, Literal from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import MemorySaver from langgraph.graph import StateGraph, START, END # 1. 定义工具 tool def get_weather(city: str) - str: 查询指定城市的实时天气情况。 weather_map { 北京: 26 度多云空气质量良好, 上海: 28 度小雨记得带伞, 广州: 31 度晴较热, } return weather_map.get(city, f暂无 {city} 的天气数据请稍后重试) tools [get_weather] tool_map {tool.name: tool for tool in tools} # 2. 定义 State class AgentState(TypedDict): messages: Annotated[list, operator.add] next_step: str # 3. 初始化模型并绑定工具 model ChatOpenAI(modelgpt-4o-mini, temperature0) model_with_tools model.bind_tools(tools) # 4. agent 节点让模型思考并决定是否调用工具 def agent_node(state: AgentState) - dict: response model_with_tools.invoke(state[messages]) has_tools len(response.tool_calls) 0 return { messages: [response], next_step: tools if has_tools else end, } # 5. tools 节点执行模型要求的工具调用 def tools_node(state: AgentState) - dict: last_message state[messages][-1] tool_messages [] for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] tool_result tool_map[tool_name].invoke(tool_args) tool_messages.append({ role: tool, tool_call_id: tool_call[id], content: tool_result, }) return {messages: tool_messages} # 6. 条件路由函数 def should_continue(state: AgentState) - Literal[tools, end]: return state[next_step] # 7. 构建图 graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tools_node) graph.add_edge(START, agent) # 关键agent 节点之后根据 next_step 决定去 tools 还是 END graph.add_conditional_edges( agent, should_continue, { tools: tools, end: END, }, ) # 工具执行完必须回到 agent让模型根据工具结果生成最终回答 graph.add_edge(tools, agent) # 8. 编译并配置检查点 checkpointer MemorySaver() app graph.compile(checkpointercheckpointer) # 9. 执行测试 def run_agent(user_input: str, thread_id: str thread-1): config {configurable: {thread_id: thread_id}} result app.invoke( {messages: [{role: user, content: user_input}]}, configconfig, ) return result[messages][-1].content if __name__ __main__: print( 第一轮对话 ) print(run_agent(北京今天天气怎么样)) print(\n 第二轮对话同一会话 ) print(run_agent(广州呢, thread_idthread-1))这段代码拆开看核心难点是循环结构。普通图是“有向无环”的但这张图里有一个环agent - tools - agent。为什么允许环因为 Agent 本身就是循环模型思考 - 调用工具 - 看到结果 - 再思考 - 可能再调用下一个工具 - 最终生成回答。LangGraph 通过条件路由来控制这个环什么时候退出。agent_node里如果模型没有产生tool_calls就把next_step设为end条件边会让流程走到END如果还有工具调用就走进tools节点执行完后回到agent。运行程序python weather_agent.py预期输出大致如下 第一轮对话 北京今天 26 度多云空气质量良好。 第二轮对话同一会话 广州今天 31 度晴较热。第二轮的输入只有“广州呢”但模型能从同一thread_id的历史消息里知道你说的是“广州今天天气怎么样”的后续这就是MemorySaver检查点机制的作用。每一轮对话的执行状态都会被保存下来新输入进来时LangGraph 会从断点恢复把之前的历史消息一起喂给模型。如果去掉checkpointer不传thread_id这种多轮追问就做不到了模型会丢失上下文。这是 LangGraph 在生产项目和本地 Demo 之间一个很大的区别。需要提醒的是这里使用了MemorySaver它只把状态保存在内存中适合学习和小规模演示。生产环境应该换成持久化的检查点存储比如 SQLite 或 PostgreSQL后面最佳实践部分会展开说明。7. 运行验证与调试手段7.1 用 stream 观察中间过程直接看最终输出很难判断中间走了哪些节点。LangGraph 提供stream方法可以逐步打印每个节点的输出。将上面的运行部分临时替换为config {configurable: {thread_id: thread-debug}} for event in app.stream( {messages: [{role: user, content: 上海天气怎么样}]}, configconfig, ): for node_name, node_output in event.items(): print(f--- 节点: {node_name} ---) print(node_output)执行后你能看到类似这样的过程--- 节点: agent --- {messages: [AIMessage(content, tool_calls[...])], next_step: tools} --- 节点: tools --- {messages: [ToolMessage(content28 度小雨记得带伞)]} --- 节点: agent --- {messages: [AIMessage(content上海今天 28 度小雨记得带伞。)], next_step: end}通过这段输出可以快速确认流程是否正确模型是否产生了工具调用、工具返回了什么、最终是否走到了结束条件。调试 Agent 的时候这个观察能力比重试猜错要高效得多。7.2 常见调试切入顺序如果程序没按预期运行建议按以下顺序定位第一步检查图结构。代码里定义的节点名和边是否一致路由函数的返回值是否都出现在映射表里。少写一个 key运行时会直接报错。第二步打印关键节点返回值。在agent_node和tools_node里加print确认当前 State 里 messages 的内容是否完整。很多问题出在消息列表累积不正确。第三步观察模型输出。检查response.tool_calls是否为空模型有没有正确生成工具调用格式。如果模型始终不调用工具大概率是模型本身不支持工具调用或者bind_tools没有生效。第四步检查编译和检查点配置。如果多轮对话串话优先确认thread_id是否正确区分了会话如果状态不更新检查是否有Annotated合并声明。7.3 非流式输出容易被忽略的点在run_agent函数中我们取的是result[messages][-1].content。这里必须明确如果最后一轮是 ToolMessage直接取 content 取到的是工具返回不是最终回答。正因为我们设计的图在工具执行后总会回到agent节点所以最终结果里最后一条消息一定是 AI 的最终回答。如果后续改造图结构要重新审视这个取值逻辑。8. 常见问题与排查思路结合 LangGraph 社区和实际开发经验下面这些问题出现频率最高。问题现象可能原因排查方式解决方案程序报错找不到节点边或条件边映射到了不存在的节点名检查所有add_edge和add_conditional_edges的字符串统一节点命名使用常量管理节点名流程不结束一直循环条件路由缺少“结束”分支或next_step永远不是 end打印每个节点的next_step值确保条件边映射表包含 END并让模型在没有工具调用时明确返回 end多轮对话串话未使用检查点或所有会话共用同一个 thread_id检查是否传入了 configthread_id 是否唯一接入 MemorySaver 等检查点每轮会话创建独立 thread_id模型始终不调用工具模型本身不支持 function calling或未完成 bind_tools打印response.tool_calls查看内容更换支持工具调用的模型确认调用链完整工具返回结果后模型仍重复调用工具返回内容格式异常或模型上下文信息不足打印工具返回的 ToolMessage 内容检查工具函数返回是否为字符串并保持提示词简洁清晰调用远程模型超时网络波动或单次请求时间过长查看错误堆栈确认是否提示类似 response in time 的报错在模型调用处增加重试机制适当增加超时时间必要时换更快的模型状态字段被覆盖而不是累加消息列表字段未使用operator.add合并检查 State 类型定义中 messages 的声明将 messages 定义为Annotated[list, operator.add]生产环境重启后对话丢失MemorySaver 只存在内存中确认检查点存储类型换成 SQLite/Postgres 等持久化 checkpointer其中需要重点强调的是循环问题。LangGraph 本身是允许循环的这是 Agent 正常工作的基础。但如果条件边没有任何一个分支指向 END或者模型持续产生工具调用图就会变成死循环。设计图的时候务必保证从任意一个节点出发都存在通往 END 的路径。另一个容易被忽略的点是 ToolCall 的格式。不同大模型返回的 tool_call 字段结构可能略有差异取name、args、id的时候建议先打印一次结构再写解析逻辑不要凭记忆硬写。9. 最佳实践与工程建议9.1 State 设计遵循最小化原则State 会随着图的分支传递给所有节点字段越多越难排查问题。建议把消息流单独放在messages里把临时变量和流程控制变量拆成独立字段比如next_step、retry_count。不要让节点往 State 里写大段中间结果确需保存时用命名规范区分。9.2 节点命名与职责边界节点名使用动词短语例如agent、tools、summarize、review。每个节点只做一件事。如果某个节点函数超过 50 行通常说明应该拆成多个节点。尤其是“先调用模型、再处理结果、再决定分支”这种组合逻辑放在同一个节点里会丧失图的可观察性。9.3 条件路由必须收敛每次写add_conditional_edges都要在心里画一遍状态机所有分支最终是否能到达 END上一步节点的返回值是否覆盖了所有可能性建议路由函数返回Literal类型Python 类型检查能在早期发现遗漏。9.4 工具设计要幂等和可重试Agent 的工具调用可能因为网络或超时而失败也可能被模型以不同参数多次调用。工具函数应该做到相同参数返回稳定结果失败时返回明确错误信息而不是抛异常涉及数据库写入、文件修改等操作时必须保证操作可重试且不会产生副作用。这是 Agent 生产化的重要边界。9.5 消息历史要做窗口管理如果不加控制多轮对话后 messages 会无限增长。一方面消耗 Token另一方面可能超出模型上下文窗口。常见做法是在agent_node前增加一个消息裁剪节点只保留最近 N 轮消息同时把早期对话压缩成摘要再放回上下文。9.6 检查点要从内存迁移到持久化存储MemorySaver适用于示例生产环境建议使用langgraph-checkpoint-postgres或langgraph-checkpoint-sqlite。持久化检查点不仅能保存对话状态还能让服务重启后恢复未完成的 Agent 流程这是工程化部署的关键能力。9.7 安全边界与权限审批如果 Agent 的工具涉及删除数据、修改配置、发送消息等高权限操作不要只靠模型提示词约束而应在图中设计审批节点。例如模型决策结果为“执行高危操作”时把流程路由到人工审批节点由审核完成后才继续。权限控制必须落在代码层级而不是依赖模型判断。9.8 准备测试集与确认性验证Agent 的运行有随机性建议把模型参数设为temperature0并用固定问题集做回归测试。工具调用分支、直接回答分支、多轮追问分支都要覆盖。调试时可以使用固定 mock 返回避免外部服务不稳定影响判断。10. 总结与后续学习方向回到开头的问题LangGraph 值得学吗从我的判断看值得。但值得的不是因为它名字热而是因为它把 Agent 的流程控制权还给了开发者。State、Node、Edge 三个概念不难真正的难点在于怎么把业务需求拆解成图结构以及怎么处理条件路由和状态累积这些工程细节。这篇文章的核心信息可以浓缩成三条第一LangGraph 是 LangChain 生态里的流程编排层二者不是替代关系而是工具箱和调度系统的关系。第二Agent 的本质是图上的一次循环理解好条件路由就理解了 Agent 的退出机制。第三生产级 Agent 的关键不在图本身而在状态管理、检查点持久化、工具幂等性和安全边界。建议的下一步路径先把这个天气 Agent 改成你自己场景里的工具比如查询数据库、调用内部 API、检索知识库然后加入消息历史裁剪和人工审批节点最后把检查点从内存切换成持久化存储做一个可重启恢复的服务。多轮对话、子图拆分、并行分支这些进阶话题等基础图结构熟练之后再深入研究即可。如果你正在规划 AI 大模型应用开发的学习路线LangGraph 会是一个承上启下的核心节点。它上承 Prompt 工程和模型调用下接 Agent 工程化和生产部署。先把这一层打牢再去看多智能体协作或复杂工作流编排会顺很多。