专栏第8篇第五篇我们讲了工具调用的四层防御第六篇讲了 MCP 协议第七篇讲了 LangChain 的 LCEL。但这些都有一个共同点单次调用。真正的 Agent 需要循环——LLM 决定调用工具 → 执行 → 结果返回 → LLM 再决定…直到生成最终回答。今天我们要讲的 LangGraph就是专门解决这个循环工作流问题的状态机框架。目录一、为什么需要 LangGraph手动管理循环太累了二、StateGraph 三要素State、Node、Edge三、条件路由Agent 自己决定下一步走哪四、状态持久化对话记忆与多用户隔离五、流式执行实时查看每个节点的输出六、实战80 行代码的客服机器人七、总结一、为什么需要 LangGraph手动管理循环太累了第五篇和第六篇讲的工具调用都是一问一答模式用户提问 → LLM 决定调用 → 执行工具 → 返回结果 → 结束。但现实中的 Agent 往往不是一次就能搞定的用户帮我查一下北京和上海的人口再算一下人均 GDP 第1轮 LLM → 需要调用 get_population(北京)、get_population(上海) 执行 → 返回两个结果 第2轮 LLM → 还需要调用 get_gdp(北京)、get_gdp(上海) 执行 → 返回两个结果 第3轮 LLM → 现在有所有数据了计算人均 GDP 执行 → 返回计算结果 第4轮 LLM → 生成最终回答1.1 手动管理的问题如果不用框架代码会写成这样# 第1轮response1llm.invoke(messages)ifresponse1.tool_calls:results1execute_tools(response1.tool_calls)messages.extend([response1]results1)# 第2轮response2llm.invoke(messages)ifresponse2.tool_calls:results2execute_tools(response2.tool_calls)messages.extend([response2]results2)# 第3轮...第N轮# 代码层层嵌套无法扩展问题很明显代码嵌套几轮调用就嵌套几层状态管理混乱messages 列表手动维护容易出错没有条件分支无法根据结果走不同的流程无法持久化重启后对话历史丢失1.2 LangGraph 的解决思路LangGraph 把上面的循环抽象成了状态机是否STARTLLM节点需要工具?工具节点END核心思想用State保存整个工作流的数据对话历史、中间结果用Node定义每个处理步骤LLM 决策、工具执行用Edge定义节点之间的流转关系用条件路由让 LLM 自己决定走哪条路二、StateGraph 三要素State、Node、Edge2.1 State共享数据容器State 是整个工作流的内存所有节点都能读写。用TypedDict定义fromtypingimportTypedDict,Annotatedfromlanggraph.graph.messageimportadd_messagesclassAgentState(TypedDict):# Annotated[list, add_messages] → 追加模式新消息添加到末尾messages:Annotated[list,add_messages]# 普通字段 → 覆盖模式iteration:int关键设计messages用Annotated[list, add_messages]包装表示这条字段是追加更新的——新消息添加到列表末尾而不是替换。这保证了对话历史的完整性。2.2 Node处理函数Node 是工作流中的处理步骤。每个 Node 接收 State返回需要更新的字段defllm_node(state:AgentState)-dict:LLM 决策节点llmcreate_llm_instance()llm_with_toolsllm.bind_tools(tools)responsellm_with_tools.invoke(state[messages])return{messages:[response],# 追加到 messagesiteration:state[iteration]1# 覆盖 iteration}deftool_node(state:AgentState)-dict:工具执行节点last_msgstate[messages][-1]tool_messages[]fortcinlast_msg.tool_calls:resulttool_map[tc[name]].invoke(tc[args])tool_messages.append(ToolMessage(contentstr(result),tool_call_idtc[id]))return{messages:tool_messages}# 追加工具结果Node 的返回值规则只返回需要更新的字段未返回的字段保持原值。2.3 Edge节点连接Edge 定义数据怎么流fromlanggraph.graphimportStateGraph,START,END# 1. 创建图workflowStateGraph(AgentState)# 2. 添加节点workflow.add_node(llm,llm_node)workflow.add_node(tools,tool_node)# 3. 添加边固定流转workflow.add_edge(START,llm)# 入口workflow.add_edge(tools,llm)# 工具执行后回到 LLM# 4. 编译appworkflow.compile()# 5. 运行resultapp.invoke({messages:[HumanMessage(content搜索Python信息)],iteration:0})三、条件路由Agent 自己决定下一步走哪前面的例子中工具执行后总是回到 LLM。但如果 LLM 已经不需要工具了呢我们需要条件路由。3.1 条件边的定义defshould_continue(state:AgentState)-str:判断是继续调用工具还是结束last_msgstate[messages][-1]iterationstate[iteration]# 安全阀最多5轮防止死循环ifiteration5:returnEND# 如果 LLM 返回了 tool_calls继续执行工具iflast_msg.tool_calls:returntools# 否则结束returnEND# 添加条件边workflow.add_conditional_edges(llm,# 从哪个节点出发should_continue,# 路由函数{# 路由映射tools:tools,# 返回tools → 走到 tools 节点END:END# 返回 END → 结束})3.2 完整工作流图有tool_calls无tool_callsSTARTLLM节点should_continue工具节点END3.3 条件路由的执行流程第1轮 用户提问 → LLM 节点 → 返回 tool_calls → 条件路由 → 走 tools 节点 第2轮 tools 节点 → 工具结果追加到 messages → 回到 LLM 节点 LLM 再次判断 → 如果还需要工具 → 继续循环 LLM 再次判断 → 如果不需要了 → 走 END 第N轮 达到最大迭代次数 → 强制 END防止死循环四、状态持久化对话记忆与多用户隔离客服机器人需要记住之前的对话。StateGraph 通过Checkpointer实现状态持久化。4.1 MemorySaver内存级持久化fromlanggraph.checkpoint.memoryimportMemorySaver# 编译时加入 CheckpointermemoryMemorySaver()appworkflow.compile(checkpointermemory)# 用 thread_id 区分不同会话config{configurable:{thread_id:user-001}}# 第1轮r1app.invoke({messages:[HumanMessage(content我叫张三)]},configconfig)# 第2轮同一会话自动携带历史r2app.invoke({messages:[HumanMessage(content我叫什么名字)]},configconfig)# AI 会回答你叫张三4.2 多会话隔离# 用户A的会话config_a{configurable:{thread_id:user-A}}# 用户B的会话config_b{configurable:{thread_id:user-B}}# 两个会话互不干扰同一个 StateGraph 实例服务多个用户4.3 生产环境的持久化方案方案特点适用场景MemorySaver内存存储重启丢失开发调试SQLite轻量级文件数据库小规模部署PostgreSQL关系型数据库生产环境Redis内存数据库高并发场景五、流式执行实时查看每个节点的输出除了invoke()等待全部完成StateGraph 还支持stream()逐步返回每个节点的输出# 流式执行foreventinapp.stream({messages:[HumanMessage(content什么是AI Agent)],iteration:0}):fornode_name,node_outputinevent.items():print(f节点{node_name}输出{node_output})# 输出# 节点 llm 输出{messages: [AIMessage], iteration: 1}# 节点 tools 输出{messages: [ToolMessage]}# 节点 llm 输出{messages: [AIMessage], iteration: 2}stream 的价值实时展示 Agent 的思考过程“正在搜索…”、“正在计算…”便于调试能看到每个节点的中间状态生产环境中配合 SSE 实现流式响应六、实战80 行代码的客服机器人把前面的知识串起来实现一个完整的客服机器人fromtypingimportTypedDict,Annotatedfromlangchain_core.messagesimportHumanMessage,AIMessage,ToolMessagefromlangchain_core.toolsimporttoolfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.checkpoint.memoryimportMemorySaver# 1. 定义工具 tooldefsearch_knowledge(query:str)-str:搜索知识库输入搜索关键词mock_data{Python:安装步骤1.访问官网下载 2.运行安装程序 3.勾选Add to PATH 4.点击Install,Git:使用步骤1.git init初始化 2.git add添加文件 3.git commit提交 4.git push推送到远程,}forkey,valueinmock_data.items():ifkeyinquery:returnvaluereturnf未找到与{query}相关的信息tooldefcalculate(expression:str)-str:计算数学表达式try:returnf计算结果{eval(expression)}except:return计算错误tools[search_knowledge,calculate]tool_map{t.name:tfortintools}# 2. 定义状态 classChatState(TypedDict):messages:Annotated[list,add_messages]iteration:int# 3. 定义节点 defllm_node(state:ChatState)-dict:llmcreate_llm_instance(temperature0)llm_with_toolsllm.bind_tools(tools)responsellm_with_tools.invoke(state[messages])return{messages:[response],iteration:state.get(iteration,0)1}deftool_node(state:ChatState)-dict:last_msgstate[messages][-1]tool_messages[]fortcinlast_msg.tool_calls:resulttool_map[tc[name]].invoke(tc[args])tool_messages.append(ToolMessage(contentstr(result),tool_call_idtc[id]))return{messages:tool_messages}# 4. 条件路由 defshould_continue(state:ChatState)-str:last_msgstate[messages][-1]iterationstate.get(iteration,0)ifiteration5:returnENDifisinstance(last_msg,AIMessage)andlast_msg.tool_calls:returntoolsreturnEND# 5. 构建工作流 workflowStateGraph(ChatState)workflow.add_node(llm,llm_node)workflow.add_node(tools,tool_node)workflow.add_edge(START,llm)workflow.add_conditional_edges(llm,should_continue,{tools:tools,END:END})workflow.add_edge(tools,llm)# 6. 编译带记忆 memoryMemorySaver()appworkflow.compile(checkpointermemory)# 7. 运行 config{configurable:{thread_id:user-001}}resultapp.invoke({messages:[HumanMessage(contentPython怎么安装然后帮我算一下1250*8)]},configconfig)# 提取最终回答formsginreversed(result[messages]):ifisinstance(msg,AIMessage)andmsg.content:print(fAI{msg.content})break执行流程用户Python怎么安装然后帮我算一下1250*8 LLM 节点 → 返回 2 个 tool_calls → search_knowledge(queryPython安装) → calculate(expression1250*8) 工具节点 → 并行执行 → 返回 2 个 ToolMessage LLM 节点 → 基于工具结果 → 生成最终回答 AIPython安装步骤是... 1250*8 的计算结果是 10000七、总结本文从手动管理循环的痛点出发梳理了 LangGraph 的核心机制为什么需要 LangGraph手动管理循环会导致代码嵌套、状态混乱、无法扩展StateGraph 三要素State共享数据、Node处理函数、Edge流转方向条件路由add_conditional_edges让 Agent 动态决定下一步走向防止死循环状态持久化Checkpointer thread_id 实现对话记忆和多会话隔离流式执行stream()逐步获取每个节点输出适合实时展示实战代码80 行客服机器人支持工具调用、循环决策、状态记忆LangGraph 的定位它不是替代 LangChain而是在 LangChain 之上增加状态管理能力。LCEL 负责组件组合StateGraph 负责流程编排。参考资源LangGraph Documentation: StateGraph APILangChain Documentation: Function CallingLangGraph Cookbook: Building Agents