引言：为什么 Agent 需要状态机
一个没有状态管理的 Agent 是"随机游走"的 Agent。它可能陷入无限循环、无法从失败中恢复、产生前后矛盾的行动，或者在复杂任务中"失忆"。状态机（State Machine）是解决这些问题的经典方法：通过明确定义 Agent 的所有可能状态、状态间的转换条件和转换动作，让 Agent 的行为从"不可预测"变为"可理解、可调试、可控制"。LangGraph 在 2026 年已成为构建有状态 Agent 的主流框架，但其核心理念——图结构状态机——才是真正需要理解的工程思维。—## 一、状态机的核心概念### 1.1 Agent 状态机的四要素状态（State）：Agent 当前的"快照" - 已收集的信息 - 已完成的步骤 - 当前的决策上下文 转换（Transition）：从一个状态到另一个状态的路径 - 触发条件（如"工具调用完成"） - 转换动作（如"更新收集到的信息"） 节点（Node）：执行特定操作的函数 - LLM 推理节点 - 工具执行节点 - 条件判断节点 边（Edge）：节点间的连接 - 普通边：无条件连接 - 条件边：根据状态决定下一步### 1.2 为什么图结构优于链式结构链式结构（Chain）： Step1 → Step2 → Step3 → Step4 缺点：无法回溯、无法条件分支、失败后只能重头图结构（Graph）： Step1 → 判断 → Step2（成功路径） ↓ Step3（失败路径）→ Step4（重试）→ 判断... 优点：支持循环、分支、回滚、并行—## 二、LangGraph 状态机工程实践### 2.1 定义 Agent 状态pythonfrom typing import TypedDict, Annotated, List, Optionalfrom langgraph.graph import StateGraph, ENDimport operatorclass AgentState(TypedDict): """Agent 的完整状态定义""" # 用户原始输入 user_query: str # 任务分解结果 subtasks: List[str] # 工具调用结果（使用 operator.add 支持追加） tool_results: Annotated[List[dict], operator.add] # 当前思考过程 thoughts: str # 最终答案 final_answer: Optional[str] # 错误信息（用于错误恢复） error: Optional[str] # 迭代计数器（防止无限循环） iteration_count: int # 当前阶段标记 phase: str # "planning" | "executing" | "reviewing" | "done"### 2.2 构建节点函数pythonfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, SystemMessageimport jsonllm = ChatOpenAI(model="gpt-4o", temperature=0)def planning_node(state: AgentState) -> dict: """规划节点：将复杂任务分解为子任务""" response = llm.invoke([ SystemMessage(content="""将用户任务分解为3-5个具体可执行的子任务。 输出 JSON：{"subtasks": ["子任务1", "子任务2", ...], "strategy": "执行策略说明"}"""), HumanMessage(content=state["user_query"]) ]) try: result = json.loads(response.content) return { "subtasks": result["subtasks"], "thoughts": result["strategy"], "phase": "executing", "iteration_count": 0 } except json.JSONDecodeError: return { "error": "规划阶段解析失败", "phase": "error" }def execution_node(state: AgentState) -> dict: """执行节点：执行当前子任务""" current_task_idx = state["iteration_count"] if current_task_idx >= len(state["subtasks"]): return {"phase": "reviewing"} current_task = state["subtasks"][current_task_idx] # 调用工具执行子任务（简化示例） result = execute_tool(current_task) return { "tool_results": [{"task": current_task, "result": result}], "iteration_count": state["iteration_count"] + 1, "phase": "executing" }def review_node(state: AgentState) -> dict: """审查节点：评估执行结果并生成最终答案""" context = "\n".join([ f"任务：{r['task']}\n结果：{r['result']}" for r in state["tool_results"] ]) response = llm.invoke([ SystemMessage(content="基于以下任务执行结果，生成完整的最终回答。"), HumanMessage(content=f"原始问题：{state['user_query']}\n\n执行结果：\n{context}") ]) return { "final_answer": response.content, "phase": "done" }def error_recovery_node(state: AgentState) -> dict: """错误恢复节点：处理失败并决定是否重试""" print(f"处理错误: {state['error']}") # 最多重试 3 次 if state["iteration_count"] < 3: return { "error": None, "iteration_count": state["iteration_count"] + 1, "phase": "planning" # 回到规划阶段重试 } return { "final_answer": f"任务执行失败（已重试{state['iteration_count']}次）：{state['error']}", "phase": "done" }### 2.3 构建图结构与条件边pythonfrom langgraph.graph import StateGraph, ENDdef route_by_phase(state: AgentState) -> str: """条件路由：根据当前 phase 决定下一步""" phase = state.get("phase", "planning") if phase == "error": return "error_recovery" elif phase == "executing": # 检查是否还有未执行的子任务 if state["iteration_count"] < len(state["subtasks"]): return "execution" else: return "review" elif phase == "reviewing": return "review" elif phase == "done": return END else: return "planning"# 构建图workflow = StateGraph(AgentState)# 添加节点workflow.add_node("planning", planning_node)workflow.add_node("execution", execution_node)workflow.add_node("review", review_node)workflow.add_node("error_recovery", error_recovery_node)# 设置入口点workflow.set_entry_point("planning")# 添加条件边workflow.add_conditional_edges( "planning", route_by_phase, { "execution": "execution", "error_recovery": "error_recovery", END: END })workflow.add_conditional_edges( "execution", route_by_phase, { "execution": "execution", # 继续执行下一个子任务 "review": "review", # 所有子任务完成，进入审查 })workflow.add_conditional_edges( "review", route_by_phase, {END: END})workflow.add_conditional_edges( "error_recovery", route_by_phase, { "planning": "planning", # 重试 END: END # 放弃 })# 编译图app = workflow.compile()—## 三、状态持久化与断点续传### 3.1 使用 Checkpointer 持久化状态pythonfrom langgraph.checkpoint.sqlite import SqliteSaverimport sqlite3# 使用 SQLite 持久化（生产环境推荐 PostgreSQL）conn = sqlite3.connect("agent_checkpoints.db", check_same_thread=False)memory = SqliteSaver(conn)# 编译时注入 checkpointerapp = workflow.compile(checkpointer=memory)# 执行时指定 thread_id 实现会话级持久化config = {"configurable": {"thread_id": "user_session_001"}}result = await app.ainvoke( {"user_query": "分析竞品并生成报告"}, config=config)# 中断后恢复：传入同一 thread_id 即可从断点继续# 后续请求会自动加载上次状态result2 = await app.ainvoke( {"user_query": ""}, # 空输入，从上次断点继续 config=config)### 3.2 人在回路（Human-in-the-Loop）pythonfrom langgraph.graph import interruptdef high_risk_action_node(state: AgentState) -> dict: """高风险操作前请求人工确认""" action = state["pending_action"] # 暂停执行，等待人工确认 confirmation = interrupt({ "message": f"即将执行高风险操作：{action}", "action": action, "requires_approval": True }) if confirmation.get("approved"): result = execute_risky_action(action) return {"action_result": result, "phase": "executing"} else: return {"phase": "done", "final_answer": "用户取消了操作"}—## 四、调试与可视化### 4.1 状态追踪与调试python# 流式执行，实时查看每个节点的状态变化async for event in app.astream( {"user_query": "帮我分析最新的 AI 趋势"}, config={"configurable": {"thread_id": "debug_001"}}): for node_name, state_update in event.items(): print(f"\n{'='*40}") print(f"节点: {node_name}") print(f"状态变更: {json.dumps(state_update, ensure_ascii=False, indent=2)}")### 4.2 状态机可视化python# 生成 Mermaid 图（在支持的环境中渲染）print(app.get_graph().draw_mermaid())# 输出：# graph TD# __start__ --> planning# planning -->|executing| execution# planning -->|error| error_recovery# execution -->|next_task| execution# execution -->|all_done| review# review --> __end__# error_recovery -->|retry| planning# error_recovery --> __end__—## 五、常见状态机模式### 5.1 重试模式pythonMAX_RETRIES = 3def retry_router(state: AgentState) -> str: if state.get("error") and state["retry_count"] < MAX_RETRIES: return "retry" elif state.get("error"): return "failed" return "success"### 5.2 并行执行模式pythonfrom langgraph.graph import Senddef dispatch_parallel_tasks(state: AgentState) -> List[Send]: """将多个子任务并行分发给相同节点""" return [ Send("execute_subtask", {"task": task, "context": state["context"]}) for task in state["subtasks"] ]workflow.add_conditional_edges( "planning", dispatch_parallel_tasks, ["execute_subtask"])—## 结语状态机思维是构建可靠 Agent 系统的核心工程方法。明确的状态定义、条件路由、错误恢复路径和断点续传，让 Agent 从"黑盒自动化"变成"可理解的工作流"。当 Agent 在生产环境出问题时，有状态机设计的系统能精确定位"哪个状态出了问题"；而没有状态机的系统，只能全程重试或人工介入。这就是状态机工程的核心价值。
