当前位置：首页 > news >正文

Agent 上线就崩？LangGraph 把 Demo 变成生产系统的最后一公里

news 2026/8/9 15:41:51

《LangGraph并不难，难的是知道什么时候不该用》看起来是个大话题，但真落到项目里，常常就是几个具体选择。下面我尽量按实际开发时会遇到的问题来讲。

摘要

前阵子帮一个团队做 Code Review，他们的 Agent 在本地跑得好好的，一问模型也是最新的，结果上线第一天就炸了。

不是模型调用失败，而是状态乱了。用户问了一个边缘问题，Agent 卡在一个循环里，日志里全是Max iterations exceeded，但没人知道它到底卡在哪一步。更糟糕的是，当用户追问时，上下文已经被污染了，整个对话逻辑彻底偏离。

这时候我才意识到，LangGraph 的真正价值，不在于“让 Agent 能动起来”，而在于“让 Agent 在失控前能被叫停”。

很多开发者把 LangGraph 当成 LangChain 的替代品，其实这是个误区。LangChain 擅长单点任务的编排，比如“调用工具 -> 解析结果 -> 返回”。但一旦任务涉及多步决策、条件分支、甚至人工介入，那种线性思维就会失效。

今天不聊怎么搭个 Hello World，聊聊怎么把 LangGraph 写成能扛住生产流量的系统。

为什么脚本式 Agent 撑不住生产环境

我们早期的 Agent 写法，通常是这样的：

def agent_step(state): thought = llm.invoke(state["messages"]) action = parse_action(thought) if action == "search": result = search_tool(action.input) return state + result elif action == "confirm": return wait_for_human()

这种写法在 Demo 阶段没问题，但它有三个致命缺陷：

1. 状态不可追溯。一旦出错，你只能看最后的输出，不知道中间经历了多少次循环、调用了什么工具。
2. 没有边界。如果模型幻觉严重，可能陷入无限递归，直到超时。
3. 无法中断。用户中途退出，或者需要人工确认时，系统不知道停在哪一步。

LangGraph 的核心思想是显式定义状态和转换。它强制你把 Agent 看作一个图（Graph），每个节点（Node）是一个函数，每条边（Edge）是一个条件判断。

这种显式化，是工程化的第一步。

State：别让数据在函数间裸奔

在 LangGraph 中，State是所有节点共享的唯一数据源。我见过太多人用dict当 State，结果不同节点之间字段命名混乱，A 节点写user_id，B 节点读uid，最后排查时一脸懵。

建议的做法是定义一个 TypedDict 或 Pydantic BaseModel，明确每个字段的类型和默认值。

from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # 消息列表，自动追加 user_id: str current_step: str tool_results: dict is_approved: bool # 人工审批标志

注意operator.add这个注解。它告诉 LangGraph，当新消息写入时，不要覆盖旧消息，而是追加。这对于保持对话历史至关重要。

实战建议：State 设计要遵循“最小必要原则”。不要把所有中间变量都塞进去，只放节点间需要传递的数据。临时变量就在节点内部处理，保持 State 干净。

Edge：条件分支是稳定性的关键

图中的边决定了流程走向。LangGraph 支持两种边：

1. 普通边：固定从节点 A 到节点 B。
2. 条件边：根据 State 的值，动态决定走向。

最常见的场景是“工具调用后，判断是否需要继续循环”。

from langgraph.graph import StateGraph, END def should_continue(state: AgentState) -> str: last_message = state["messages"][-1] if "tool_call" in last_message.content: return "tools" # 继续调用工具 return "finalize" # 结束，生成最终答案 workflow = StateGraph(AgentState) workflow.add_node("agent", agent_node) workflow.add_node("tools", tool_node) workflow.add_conditional_edges( "agent", should_continue, { "tools": "tools", "finalize": END } )

这里有一个容易被忽视的细节：条件函数必须是确定性的。如果你依赖模型输出的随机性来做分支判断，调试时会非常痛苦。最好是在节点内部把模型输出解析成结构化数据，再根据结构化数据做分支。

人工审批节点：生产环境的必要妥协

这是很多 Demo 里缺失的一环，也是生产环境最容易被忽视的稳定性保障。

当 Agent 执行高风险操作（比如删除数据、发送邮件、调用支付接口）时，必须插入一个人工审批节点。

def human_approval_node(state: AgentState) -> AgentState: # 这里可以集成消息队列或 UI 回调 # 比如发送 Slack 消息，等待 webhook 返回 approval = wait_for_human_action(state["pending_action"]) state["is_approved"] = approval return state workflow.add_node("human_approval", human_approval_node) workflow.add_edge("tools", "human_approval") workflow.add_conditional_edges( "human_approval", lambda s: "finalize" if s["is_approved"] else "retry", {"finalize": END, "retry": "agent"} )

踩坑经验：人工审批节点必须是异步的。不要让 HTTP 请求阻塞线程。在我的项目里，我们用了 Redis 作为状态存储，审批节点写入 Redis，等待回调更新。这样即使服务重启，状态也不会丢失。

工程化落地：日志、监控与回滚

写完图只是第一步。在生产环境中，你需要回答三个问题：

1. 现在走到哪一步了？
2. 如果出错，怎么回滚？
3. 怎么监控性能瓶颈？

1. 可观测性：检查点（Checkpointer）

LangGraph 内置了 Checkpointer 机制。每次状态更新，都会保存到后端（内存、SQL、Redis 等）。这意味着你可以随时“回溯”到之前的状态。

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() graph = workflow.compile(checkpointer=memory) # 执行时传入 thread_id thread_config = {"config": {"configurable": {"thread_id": "user_123"}}} result = graph.invoke(input, thread_config)

通过thread_id，你可以查询某个用户的所有历史交互，甚至恢复到上一步状态重新执行。这对于调试和合规审计至关重要。

2. 异常兜底：超时与重试

不要依赖模型的稳定性，要依赖工程的容错性。

最大迭代次数：在StateGraph编译时设置recursion_limit。
节点超时：每个节点函数内部设置超时，超时后抛出异常，由全局异常处理器捕获。
降级策略：当工具调用失败时，返回一个默认的兜底回答，而不是让 Agent 卡死。

import asyncio async def tool_node(state: AgentState) -> AgentState: try: result = await asyncio.wait_for( call_external_api(state["query"]), timeout=5.0 ) state["tool_results"] = result except asyncio.TimeoutError: state["tool_results"] = {"error": "timeout"} # 可以选择标记状态，让 agent 节点生成一个“查询超时”的回复 return state