一、引言为什么现成的 AgentExecutor 不够用2023 年 LangChain 的initialize_agentAgentExecutor几乎成了 ReAct 智能体的代名词。但只要在真实业务里跑过一周就会遇到这几个绕不开的痛点第一控制流是黑盒。AgentExecutor把 Thought → Action → Observation 的循环封在while里你无法在“工具调用前”插入权限校验、在“工具返回后”插入结果脱敏、在“连续三次调用同一工具”时强制打断。第二记忆与推理耦合。memory参数把历史消息拼进 prompt但历史裁剪、摘要压缩、跨会话持久化这些需求一旦想定制就得重写整个agent链路。第三调试困难。循环卡死时你只能看到一个max_iterations报错看不到中间状态是怎么演化的。LangGraph 的解法很直接把智能体显式建模成一张有向图。节点是计算单元调 LLM、跑工具边是控制流状态是显式声明的 TypedDict。ReAct 循环不再藏在框架里而是你自己连出来的两条边。这篇文章就从零把这张图连出来并解决工具调用与多轮记忆这两个核心工程问题。二、核心原理2.1 ReAct 的本质是一个二状态机ReActReasoning Acting论文的核心思想是让模型交替产出“推理轨迹”和“动作”。在工程上它只对应两个状态Agent 节点把当前消息历史喂给 LLM模型要么直接给出最终答案要么输出一个带tool_calls字段的AIMessage。Tool 节点解析tool_calls执行对应函数把结果包装成ToolMessage追加回历史。控制流的唯一分支逻辑是最后一条消息有没有tool_calls有就跳 Tool没有就结束。如果你读过 ReAct 原论文会发现它其实描述了更复杂的“思考-行动-观察”循环但在 LLM 原生支持 function calling 之后推理过程已经内化到模型权重里我们不再需要显式输出 “Thought: …”。工程实现上二状态机完全等价于原论文的 ReAct 循环而且更简洁、更可控。与 Plan-and-Execute 相比ReAct 是“走一步看一步”适合动态环境与 Reflexion 相比ReAct 没有自我反思步骤但可以通过在 Agent 节点里追加反思提示词来模拟。理解这一点你就能根据业务需求灵活裁剪。2.2 LangGraph 的三个关键抽象抽象作用StateGraph声明状态类型与节点编译成可执行图Reducer定义“节点返回值如何合并进全局状态”Checkpointer在每个 superstep 后快照状态支撑多轮记忆与时间旅行其中Reducer 是最容易被忽略、也最关键的一环。默认语义是“覆盖”而消息列表必须用“追加”这就是add_messages存在的意义——它不仅追加还会按id去重更新避免重放时消息重复。具体来说add_messages的合并逻辑是对于新消息列表中的每一条如果它的id已经存在于旧列表中就替换旧消息否则追加到末尾。这个设计让“更新已有消息”和“插入新消息”可以统一处理。例如当你使用graph.update_state修改某条消息时只要保持id不变reducer 就会正确覆盖。StateGraph的编译过程会做几件事验证节点和边的合法性、把 reducer 绑定到每个状态键、生成 Pregel 执行计划。编译后的CompiledGraph才是可调用对象。Checkpointer则是在每个 superstep可以理解为图的一轮执行结束后把当前状态序列化写入存储。MemorySaver是内存版适合开发调试生产环境可以用SqliteSaver、PostgresSaver或RedisSaver。这些 Checkpointer 都实现了相同的接口切换成本极低。2.3 多轮记忆的两个层次很多人把“多轮记忆”简单理解成“把历史塞进 prompt”实际上要拆成两层会话内短期记忆同一thread_id下的消息历史。由Checkpointer自动持久化graph.invoke时只需传新消息历史会自动加载。跨会话长期记忆用户偏好、事实性知识。存在BaseStore里通过工具按需检索不占用上下文窗口。下面我们先把第一层做扎实。短期记忆的关键是thread_id的设计。它可以是用户 ID、会话 ID 的组合比如user-42-session-1。同一个thread_id下的所有调用共享历史不同thread_id完全隔离。长期记忆则需要BaseStore配合例如用InMemoryStore保存用户偏好然后在 Agent 节点里根据用户 ID 检索并注入系统提示词。这部分我们会在后续进阶文章中展开。三、实战从零构建3.1 环境与依赖pipinstalllanggraph langchain-openai langchain-coreexportOPENAI_API_KEYsk-...建议使用 Python 3.10因为TypedDict和Annotated的语法在旧版本上可能不兼容。如果使用其他模型比如 Claude 或本地 Ollama只需替换ChatOpenAI为对应的ChatAnthropic或ChatOllama并确保模型支持 tool calling。3.2 定义工具docstring 就是 Promptfromlangchain_core.toolsimporttooltooldefquery_order(order_id:str)-str:根据订单号查询订单的当前状态与物流信息。 Args: order_id: 订单号形如 ORD20240512001必须由用户提供。 # 真实场景这里走内部 APIfake_db{ORD20240512001:已发货顺丰 SF1234567890预计 5 月 14 日送达,}returnfake_db.get(order_id,f未找到订单{order_id}请确认订单号是否正确)tooldefcalc_shipping(distance_km:float,weight_kg:float)-str:按距离和重量估算运费元。returnf预估运费{5distance_km*0.8weight_kg*2:.2f}元TOOLS[query_order,calc_shipping]这里有个新手常踩的坑工具函数的 docstring 会被tool装饰器提取为 JSON Schema 的description直接进入模型的 function calling 上下文。写“查询订单”和写清参数格式、使用前提模型选对工具的概率能差出几十个百分点。除了 docstring参数类型注解也会被解析成 JSON Schema。比如order_id: str会生成{type: string}。如果你需要更复杂的参数结构可以使用 Pydantic 模型。另外工具函数的返回值最好是字符串如果是字典或列表LangChain 会尝试序列化但可能丢失格式信息。建议在工具内部就把结果整理成人类可读的文本。工具描述里还可以加入“使用示例”例如“当用户说‘查一下订单’时必须先从用户输入中提取订单号再调用本工具。” 这能显著降低模型漏调参数的概率。如果工具调用失败比如订单号格式错误工具应该返回明确的错误信息而不是抛出异常因为异常会被ToolNode捕获并变成错误消息但自定义错误提示对模型更友好。3.3 状态、节点与图的完整实现fromtypingimportAnnotated,Literalfromtyping_extensionsimportTypedDictfromlangchain_core.messagesimportBaseMessage,SystemMessagefromlangchain_openaiimportChatOpenAIfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.graph.messageimportadd_messagesfromlanggraph.prebuiltimportToolNodefromlanggraph.checkpoint.memoryimportMemorySaver SYSTEM_PROMPT(你是一个电商客服助手。涉及订单状态必须调用 query_order 查询不要凭记忆编造。计算运费使用 calc_shipping。信息不足时主动追问。)# ---------- 1. 状态定义 ----------classAgentState(TypedDict):# 关键用 add_messages 作为 reducer实现“追加 按 id 更新”messages:Annotated[list[BaseMessage],add_messages]# ---------- 2. 模型绑定工具 ----------llmChatOpenAI(modelgpt-4o-mini,temperature0)llm_with_toolsllm.bind_tools(TOOLS)# ---------- 3. 节点实现 ----------defagent_node(state:AgentState)-dict:LLM 推理节点产出 AIMessage可能含 tool_callsmsgs[SystemMessage(contentSYSTEM_PROMPT)]state[messages]responsellm_with_tools.invoke(msgs)return{messages:[response]}# ToolNode 是 LangGraph 内置节点自动读最后一条 AIMessage 的 tool_calls# 并发执行工具并把结果包装成带 tool_call_id 的 ToolMessage 返回tool_nodeToolNode(TOOLS,handle_tool_errorsTrue)# ---------- 4. 条件边ReAct 的唯一分支逻辑 ----------defshould_continue(state:AgentState)-Literal[tools,__end__]:laststate[messages][-1]ifgetattr(last,tool_calls,None):returntoolsreturnEND# ---------- 5. 连图 ----------builderStateGraph(AgentState)builder.add_node(agent,agent_node)builder.add_node(tools,tool_node)builder.add_edge(START,agent)builder.add_conditional_edges(agent,should_continue,{tools:tools,END:END},)builder.add_edge(tools,agent)# ★ 回边ReAct 循环的灵魂graphbuilder.compile(checkpointerMemorySaver())整个 ReAct 循环就落在add_edge(tools, agent)这一行上。工具执行完必须回到 agent 重新推理否则模型永远拿不到 Observation。逐段解释一下AgentState只定义了一个messages键并用Annotated标注了 reducer。这意味着任何节点返回的{messages: [...]}都会经过add_messages合并到全局状态。agent_node里我们把系统提示词拼在历史消息前面注意这里每次都重新拼而不是把系统消息存进状态这样系统提示词不会被 Checkpointer 快照修改后立即生效。llm_with_tools.invoke返回的AIMessage可能包含tool_calls列表每个元素有name、args、id。ToolNode会自动读取最后一条AIMessage的tool_calls按name查找工具用args调用然后把结果包装成ToolMessage其中tool_call_id就是对应的id。handle_tool_errorsTrue表示如果工具执行抛异常会把异常信息作为ToolMessage返回而不是让整个图崩溃。条件边函数should_continue的返回值必须是节点名的字符串或END。我们这里用Literal做类型提示编译时会校验目标节点是否存在。add_conditional_edges的第三个参数是路径映射把返回值映射到实际节点名。最后compile时传入checkpointer图就具备了持久化能力。3.4 多轮记忆thread_id 即会话隔离键MemorySaver会在每个 superstep 后自动快照状态。调用时通过config[configurable][thread_id]指定会话LangGraph 会按thread_id加载历史状态用add_messagesreducer 把新传入的消息追加到历史尾部执行完成后再次落盘。fromlangchain_core.messagesimportHumanMessage config{configurable:{thread_id:user-42-session-1},recursion_limit:25,# 防止工具调用死循环}defchat(text:str):forchunkingraph.stream({messages:[HumanMessage(contenttext)]},config,stream_modevalues,):pass# 中间态可在此做可观测性埋点finalchunk[messages][-1]print(AI:,final.content)returnfinal chat(帮我查一下订单 ORD20240512001)# → 触发 query_orderObservation 回填后给出答案chat(那从北京寄 300 公里、2 公斤要多少钱)# → 第二轮无需重复上文模型能看到完整历史直接调 calc_shippingchat(刚才那个订单号是多少来着)# → 不调工具直接从历史消息中回忆 → 验证记忆生效第三个问题是最有说服力的验证如果Checkpointer没生效模型不可能知道上一轮的订单号。你还可以用graph.get_state(config)查看当前完整状态包括所有消息和下一步要执行的节点。如果想回滚到某个历史检查点可以用graph.get_state_history(config)遍历所有快照然后graph.update_state恢复到指定checkpoint_id。这就是“时间旅行”调试对排查复杂对话非常有用。另外thread_id的设计要考虑并发。如果多个用户同时使用同一个thread_id状态会互相污染。通常用user_id session_id组合并在服务端做好鉴权。MemorySaver只适合单机开发生产环境建议换成PostgresSaver它支持事务和并发控制。四、踩坑与优化建议4.1 循环失控recursion_limit是安全阀不是万能药recursion_limit默认 25单位是superstep图步数不是对话轮次。一次完整的“Agent → Tool → Agent”要消耗 2 步所以 25 步其实只够 12 次工具调用。常见死循环场景是模型反复调用同一个工具却拿不到满意结果——正确做法是在条件边里加业务级熔断defshould_continue(state:AgentState)-Literal[tools,__end__]:laststate[messages][-1]ifnotgetattr(last,tool_calls,None):returnEND# 统计本轮对话中同一工具被调用的次数calls[mforminstate[messages]ifgetattr(m,tool_calls,None)]iflen(calls)6:returnEND# 强制收敛让模型基于已有信息作答returntools更精细的做法是只统计某个特定工具的调用次数比如query_order连续调用 3 次就熔断。你还可以在熔断时追加一条系统消息提示模型“已多次查询未果请直接告知用户稍后重试”。4.2 上下文膨胀消息裁剪必须放在 Agent 节点内部随着多轮对话推进state[messages]会无限增长成本和延迟线性上升。注意不要把裁剪写进 reducer——那样会破坏 Checkpointer 的完整快照。正确姿势是在agent_node里对喂给 LLM 的列表做临时裁剪fromlangchain_core.messagesimporttrim_messages trimmertrim_messages(max_tokens3000,strategylast,token_counterllm,include_systemTrue,start_onhuman,# 保证裁剪后首条是 HumanMessage避免 API 报错)defagent_node(state:AgentState)-dict:historytrim_messages(state[messages],max_tokens3000,strategylast,token_counterllm,include_systemTrue,start_onhuman,)msgs[SystemMessage(contentSYSTEM_PROMPT)]historyreturn{messages:[llm_with_tools.invoke(msgs)]}如果对话本身很长更好的方案是加一个Summarize 节点当消息数超过阈值时把早期消息压缩成一段摘要写回状态再删除原始消息。摘要可以保留关键实体订单号、金额、时间丢弃寒暄。注意摘要节点也要返回{messages: [...]}但要用RemoveMessage删除旧消息否则 reducer 会继续追加。4.3 工具返回必须携带tool_call_id如果你手动构造ToolMessage比如在自定义 ToolNode 里做结果脱敏必须把对应的tool_call_id原样带上否则 OpenAI API 会直接返回 400fromlangchain_core.messagesimportToolMessagedefcustom_tool_node(state:AgentState)-dict:laststate[messages][-1]outputs[]forcallinlast.tool_calls:resultTOOLS_BY_NAME[call[name]].invoke(call[args])outputs.append(ToolMessage(contentstr(result),tool_call_idcall[id],# ★ 缺这个必报错namecall[name],))return{messages:outputs}这个错误在开发时很常见因为ToolNode内部帮你处理了一旦自己写就容易忘。记住AIMessage里的每个tool_call都有一个唯一id对应的ToolMessage必须用同一个id这样模型才能把结果和请求关联起来。4.4 并行工具调用的副作用问题ToolNode对同一轮里的多个tool_calls是并发执行的。查询类工具没问题但写操作类工具下单、发消息必须串行化或加锁否则可能重复下单。解决方案有三种在工具内部加幂等键让模型传入唯一的request_id工具内部用数据库唯一约束去重。自定义 ToolNode串行执行把ToolNode替换成自己的节点用for循环逐个调用而不是并发。在系统提示词里禁止并行写操作明确告诉模型“一次只能调用一个写操作工具”但这种方法不可靠。推荐第二种代码简单且可控defserial_tool_node(state:AgentState)-dict:laststate[messages][-1]outputs[]forcallinlast.tool_calls:tool_fnTOOLS_BY_NAME[call[name]]try:resulttool_fn.invoke(call[args])exceptExceptionase:resultf工具执行失败{e}outputs.append(ToolMessage(contentstr(result),tool_call_idcall[id],namecall[name],))return{messages:outputs}4.5 流式输出与可观测性graph.stream支持多种stream_mode。values返回每个 superstep 后的完整状态适合调试updates只返回节点更新的增量适合前端实时展示messages会逐 token 流式输出 LLM 的回复。生产环境建议组合使用并把中间状态推送到 LangSmith 或自建日志系统。LangSmith 可以自动追踪每次工具调用的输入输出、耗时和 token 消耗定位问题非常方便。4.6 工具调用错误处理ToolNode(handle_tool_errorsTrue)会把异常转成ToolMessage但默认错误信息对模型不够友好。你可以传入自定义错误处理函数defhandle_error(e:Exception)-str:returnf工具调用出错请检查参数格式。错误详情{type(e).__name__}tool_nodeToolNode(TOOLS,handle_tool_errorshandle_error)这样模型收到错误后有机会自我修正参数并重试。4.7 系统提示词工程系统提示词对工具选择的影响极大。除了明确“必须调用”还要给出反例。例如“如果用户只是问‘你们发什么快递’不需要调用 query_order直接回答即可。” 另外把工具描述写得具体比如“订单号格式为 ORD8位数字”能减少模型编造参数。4.8 测试与评估对智能体的测试不能只看最终回复还要断言工具调用序列。可以用graph.get_state(config)取出所有消息检查tool_calls的name和args是否符合预期。建议为每个工具写单元测试再用集成测试覆盖多轮对话。评估指标包括工具调用准确率、参数提取准确率、任务完成率、平均步数。五、常见问题 FAQQ1为什么模型不调用工具直接回答A检查工具描述是否清晰、系统提示词是否强制要求调用。可以尝试在提示词中加入“必须调用工具否则视为错误”。另外模型温度设为 0 能提高稳定性。Q2如何限制工具调用总次数A在条件边里统计tool_calls数量超过阈值就返回END。也可以设置recursion_limit作为最后防线。Q3如何持久化到数据库A把MemorySaver换成SqliteSaver或PostgresSaver它们都实现了BaseCheckpointSaver接口。安装对应依赖后用from langgraph.checkpoint.sqlite import SqliteSaver即可。Q4多用户并发时 thread_id 怎么设计A用user_id:session_id格式确保每个用户会话独立。如果同一用户多设备可以用user_id:device_id区分。Q5如何让智能体记住用户偏好A使用BaseStore实现长期记忆。在 Agent 节点里根据user_id检索偏好拼接到系统提示词中。写入偏好可以通过一个专门的工具完成。六、总结与下一步本文从零构建了一个基于 LangGraph 的 ReAct 智能体核心只有两个节点、一条条件边和一条回边却完整实现了工具调用与多轮记忆。更多硬核网安与AI工具包请扫码获取完整源码相比AgentExecutor你获得了完全透明的控制流、可插拔的 Checkpointer 和显式的状态管理。下一步可以探索用SendAPI 实现动态工具选择、用子图拆分复杂业务、用BaseStore做长期记忆以及接入 LangSmith 做生产级可观测性。把这些组合起来你就能手搓出一个真正可用的智能体系统。