企业官网建设流程全解析

一、引言：为什么现成的 AgentExecutor 不够用

2023 年 LangChain 的initialize_agent+AgentExecutor几乎成了 ReAct 智能体的代名词。

但只要在真实业务里跑过一周，就会遇到这几个绕不开的痛点：

第一，控制流是黑盒。AgentExecutor把 Thought → Action → Observation 的循环封在while里，你无法在“工具调用前”插入权限校验、在“工具返回后”插入结果脱敏、在“连续三次调用同一工具”时强制打断。

第二，记忆与推理耦合。memory参数把历史消息拼进 prompt，但历史裁剪、摘要压缩、跨会话持久化这些需求，一旦想定制就得重写整个agent链路。

第三，调试困难。循环卡死时，你只能看到一个max_iterations报错，看不到中间状态是怎么演化的。

LangGraph 的解法很直接：把智能体显式建模成一张有向图。节点是计算单元（调 LLM、跑工具），边是控制流，状态是显式声明的 TypedDict。ReAct 循环不再藏在框架里，而是你自己连出来的两条边。这篇文章就从零把这张图连出来，并解决工具调用与多轮记忆这两个核心工程问题。

二、核心原理

2.1 ReAct 的本质是一个二状态机

ReAct（Reasoning + 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 的三个关键抽象

其中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”，实际上要拆成两层：

1. 会话内短期记忆：同一thread_id下的消息历史。由Checkpointer自动持久化，graph.invoke时只需传新消息，历史会自动加载。
2. 跨会话长期记忆：用户偏好、事实性知识。存在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_KEY="sk-..."

建议使用 Python 3.10+，因为TypedDict和Annotated的语法在旧版本上可能不兼容。如果使用其他模型，比如 Claude 或本地 Ollama，只需替换ChatOpenAI为对应的ChatAnthropic或ChatOllama，并确保模型支持 tool calling。

3.2 定义工具：docstring 就是 Prompt

fromlangchain_core.toolsimporttool@tooldefquery_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"预估运费{5+distance_km*0.8+weight_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. 模型绑定工具 ----------llm=ChatOpenAI(model="gpt-4o-mini",temperature=0)llm_with_tools=llm.bind_tools(TOOLS)# ---------- 3. 节点实现 ----------defagent_node(state:AgentState)->dict:"""LLM 推理节点：产出 AIMessage（可能含 tool_calls）"""msgs=[SystemMessage(content=SYSTEM_PROMPT)]+state["messages"]response=llm_with_tools.invoke(msgs)return{"messages":[response]}# ToolNode 是 LangGraph 内置节点：自动读最后一条 AIMessage 的 tool_calls，# 并发执行工具，并把结果包装成带 tool_call_id 的 ToolMessage 返回tool_node=ToolNode(TOOLS,handle_tool_errors=True)# ---------- 4. 条件边：ReAct 的唯一分支逻辑 ----------defshould_continue(state:AgentState)->Literal["tools","__end__"]:last=state["messages"][-1]ifgetattr(last,"tool_calls",None):return"tools"returnEND# ---------- 5. 连图 ----------builder=StateGraph(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 循环的灵魂graph=builder.compile(checkpointer=MemorySaver())

整个 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_errors=True表示如果工具执行抛异常，会把异常信息作为ToolMessage返回，而不是让整个图崩溃。

条件边函数should_continue的返回值必须是节点名的字符串或END。我们这里用Literal做类型提示，编译时会校验目标节点是否存在。add_conditional_edges的第三个参数是路径映射，把返回值映射到实际节点名。最后compile时传入checkpointer，图就具备了持久化能力。

3.4 多轮记忆：thread_id 即会话隔离键

MemorySaver会在每个 superstep 后自动快照状态。调用时通过config["configurable"]["thread_id"]指定会话，LangGraph 会：

1. 按thread_id加载历史状态；
2. 用add_messagesreducer 把新传入的消息追加到历史尾部；
3. 执行完成后再次落盘。

fromlangchain_core.messagesimportHumanMessage config={"configurable":{"thread_id":"user-42-session-1"},"recursion_limit":25,# 防止工具调用死循环}defchat(text:str):forchunkingraph.stream({"messages":[HumanMessage(content=text)]},config,stream_mode="values",):pass# 中间态可在此做可观测性埋点final=chunk["messages"][-1]print("AI:",final.content)returnfinal chat("帮我查一下订单 ORD20240512001")# → 触发 query_order，Observation 回填后给出答案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__"]:last=state["messages"][-1]ifnotgetattr(last,"tool_calls",None):returnEND# 统计本轮对话中同一工具被调用的次数calls=[mforminstate["messages"]ifgetattr(m,"tool_calls",None)]iflen(calls)>=6:returnEND# 强制收敛，让模型基于已有信息作答return"tools"

更精细的做法是只统计某个特定工具的调用次数，比如query_order连续调用 3 次就熔断。你还可以在熔断时追加一条系统消息，提示模型“已多次查询未果，请直接告知用户稍后重试”。

4.2 上下文膨胀：消息裁剪必须放在 Agent 节点内部

随着多轮对话推进，state["messages"]会无限增长，成本和延迟线性上升。注意不要把裁剪写进 reducer——那样会破坏 Checkpointer 的完整快照。正确姿势是在agent_node里对喂给 LLM 的列表做临时裁剪：

fromlangchain_core.messagesimporttrim_messages trimmer=trim_messages(max_tokens=3000,strategy="last",token_counter=llm,include_system=True,start_on="human",# 保证裁剪后首条是 HumanMessage，避免 API 报错)defagent_node(state:AgentState)->dict:history=trim_messages(state["messages"],max_tokens=3000,strategy="last",token_counter=llm,include_system=True,start_on="human",)msgs=[SystemMessage(content=SYSTEM_PROMPT)]+historyreturn{"messages":[llm_with_tools.invoke(msgs)]}

如果对话本身很长，更好的方案是加一个Summarize 节点：当消息数超过阈值时，把早期消息压缩成一段摘要写回状态，再删除原始消息。摘要可以保留关键实体（订单号、金额、时间），丢弃寒暄。注意摘要节点也要返回{"messages": [...]}，但要用RemoveMessage删除旧消息，否则 reducer 会继续追加。

4.3 工具返回必须携带tool_call_id

如果你手动构造ToolMessage（比如在自定义 ToolNode 里做结果脱敏），必须把对应的tool_call_id原样带上，否则 OpenAI API 会直接返回 400：

fromlangchain_core.messagesimportToolMessagedefcustom_tool_node(state:AgentState)->dict:last=state["messages"][-1]outputs=[]forcallinlast.tool_calls:result=TOOLS_BY_NAME[call["name"]].invoke(call["args"])outputs.append(ToolMessage(content=str(result),tool_call_id=call["id"],# ★ 缺这个必报错name=call["name"],))return{"messages":outputs}

这个错误在开发时很常见，因为ToolNode内部帮你处理了，一旦自己写就容易忘。记住：AIMessage里的每个tool_call都有一个唯一id，对应的ToolMessage必须用同一个id，这样模型才能把结果和请求关联起来。

4.4 并行工具调用的副作用问题

ToolNode对同一轮里的多个tool_calls是并发执行的。查询类工具没问题，但写操作类工具（下单、发消息）必须串行化或加锁，否则可能重复下单。解决方案有三种：

1. 在工具内部加幂等键：让模型传入唯一的request_id，工具内部用数据库唯一约束去重。
2. 自定义 ToolNode，串行执行：把ToolNode替换成自己的节点，用for循环逐个调用，而不是并发。
3. 在系统提示词里禁止并行写操作：明确告诉模型“一次只能调用一个写操作工具”，但这种方法不可靠。

推荐第二种，代码简单且可控：

defserial_tool_node(state:AgentState)->dict:last=state["messages"][-1]outputs=[]forcallinlast.tool_calls:tool_fn=TOOLS_BY_NAME[call["name"]]try:result=tool_fn.invoke(call["args"])exceptExceptionase:result=f"工具执行失败：{e}"outputs.append(ToolMessage(content=str(result),tool_call_id=call["id"],name=call["name"],))return{"messages":outputs}

4.5 流式输出与可观测性

graph.stream支持多种stream_mode。"values"返回每个 superstep 后的完整状态，适合调试；"updates"只返回节点更新的增量，适合前端实时展示；"messages"会逐 token 流式输出 LLM 的回复。生产环境建议组合使用，并把中间状态推送到 LangSmith 或自建日志系统。LangSmith 可以自动追踪每次工具调用的输入输出、耗时和 token 消耗，定位问题非常方便。

4.6 工具调用错误处理

ToolNode(handle_tool_errors=True)会把异常转成ToolMessage，但默认错误信息对模型不够友好。你可以传入自定义错误处理函数：

defhandle_error(e:Exception)->str:returnf"工具调用出错，请检查参数格式。错误详情：{type(e).__name__}"tool_node=ToolNode(TOOLS,handle_tool_errors=handle_error)

这样模型收到错误后，有机会自我修正参数并重试。

4.7 系统提示词工程

系统提示词对工具选择的影响极大。除了明确“必须调用”，还要给出反例。例如：“如果用户只是问‘你们发什么快递’，不需要调用 query_order，直接回答即可。” 另外，把工具描述写得具体，比如“订单号格式为 ORD+8位数字”，能减少模型编造参数。

4.8 测试与评估

对智能体的测试不能只看最终回复，还要断言工具调用序列。可以用graph.get_state(config)取出所有消息，检查tool_calls的name和args是否符合预期。建议为每个工具写单元测试，再用集成测试覆盖多轮对话。评估指标包括：工具调用准确率、参数提取准确率、任务完成率、平均步数。

五、常见问题 FAQ

Q1：为什么模型不调用工具，直接回答？
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 做生产级可观测性。把这些组合起来，你就能手搓出一个真正可用的智能体系统。