路由、扩散、反应、评估与批准：五种LangGraph模式
了解 LangGraph 中五种代理工作流模式，从路由器和 ReAct 循环到评估器门控及人工审批，以及每种模式在实际应用中所需的约束机制。
更长的提示语很少能解决人工智能功能不可靠的问题。当系统需要浏览、编写代码、执行合规性检查或编辑面向用户的文本时，单次非确定性模型调用显得过于脆弱。结构化设计能起到帮助作用：模型负责在需要判断时进行推理，而代码则控制流程走向、循环及终止条件。下文介绍了五种此类模式，每种都配有可运行的 Python LangGraph 示例，以及投入生产前需解决的注意事项。
为何图结构适合智能体工作流
传统程序是按线性顺序运行的。而智能体需要循环、条件分支以及持久化状态：如果生成的代码测试失败，系统必须捕获错误并重新尝试。
LangGraph 将其建模为有向图：
- 节点 是执行单一任务的 Python 函数，例如 SQL 查询或模型调用。
如需更深入地了解这些基本要素，请参阅LangGraph实践：状态、节点与边以及五种智能体模式。
模式1：路由器
路由器是位于入口处的分类器。它不会将所有请求都发送给庞大且成本高昂的模型，而是通过轻量级步骤将每个请求转发到专门的模型、子图或本地工具。
┌───> [Specialized Coding Agent] ───> [END]
[START] ──> [Router]
└───> [General Knowledge Agent] ───> [END]
利用它可降低延迟与成本，或将意图匹配到相应的专用工具。
在温度为0时的小型模型会将查询标记为coding或general，而add_conditional_edges则将该标记映射到对应的处理节点。route_decision在模型返回意外结果时会回退到general。
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
# 1. Define the shared state
class RouterState(TypedDict):
query: str
route: str
response: str
# Use a fast, cost-effective model for classification
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 2. Define the Nodes
def classify_query(state: RouterState):
prompt = f"""Classify the following user query into one of two categories: 'coding' or 'general'.
Respond with exactly one word, either 'coding' or 'general'.
Query: {state['query']}"""
response = model.invoke([HumanMessage(content=prompt)])
classification = response.content.strip().lower()
return {"route": classification}
def handle_coding(state: RouterState):
return {"response": "Executing advanced syntax processing and code compilation logic..."}
def handle_general(state: RouterState):
return {"response": "Processing casual conversation or general knowledge search..."}
# 3. Define Conditional Routing Logic
def route_decision(state: RouterState) -> Literal["coding", "general"]:
return state["route"] if state["route"] in ["coding", "general"] else "general"
# 4. Construct the Graph
workflow = StateGraph(RouterState)
workflow.add_node("classifier", classify_query)
workflow.add_node("coding_agent", handle_coding)
workflow.add_node("general_agent", handle_general)
workflow.add_edge(START, "classifier")
workflow.add_conditional_edges("classifier", route_decision, {
"coding": "coding_agent",
"general": "general_agent"
})
workflow.add_edge("coding_agent", END)
workflow.add_edge("general_agent", END)
# Compile and Run
app = workflow.compile()
result = app.invoke({"query": "How do I implement a binary search tree in Python?"})
print(f"Route Taken: {result['route']}\nResponse: {result['response']}")
示例编写时使用的模型名称为当时最新的版本；请替换为您所使用服务提供商的当前版本。结构化输出比解析单个单词更为可靠。
模式2：协调器与工作节点
对于那些单次提示无法处理的复杂任务，协调器会将目标拆分为多个独立的子任务，由工作节点分别完成，最后再由合成器整合这些结果。
┌───> [Worker A: Section 1] ───┐
[START] ──> [Orchestrator] ├───> [Worker B: Section 2] ───┼───> [Synthesizer] ───> [END]
└───> [Worker C: Section 3] ───┘
这种模式适用于报告等多篇幅内容以及多来源的研究工作。
协调器会请求一个包含两个子主题的 JSON 列表，workers 节点分别为每个子主题编写一段内容，随后合成器将它们合并起来。
import json
from typing import TypedDict, List
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class OrchestratorState(TypedDict):
topic: str
tasks: List[str]
worker_outputs: List[str]
final_report: str
model = ChatOpenAI(model="gpt-4o", temperature=0.2)
def orchestrator_plan(state: OrchestratorState):
prompt = f"Create a JSON list of exactly two sub-topics needed to write a comprehensive guide about: {state['topic']}. Return ONLY a valid JSON list of strings."
response = model.invoke([HumanMessage(content=prompt)])
tasks = json.loads(response.content.strip())
return {"tasks": tasks, "worker_outputs": []}
def worker_execute(state: OrchestratorState):
outputs = []
for task in state["tasks"]:
prompt = f"Write a brief, highly technical paragraph explaining: {task}"
response = model.invoke([HumanMessage(content=prompt)])
outputs.append(response.content)
return {"worker_outputs": outputs}
def synthesize_report(state: OrchestratorState):
combined_context = "\n\n".join(state["worker_outputs"])
prompt = f"Combine the following sections into a cohesive newsletter update regarding {state['topic']}:\n\n{combined_context}"
response = model.invoke([HumanMessage(content=prompt)])
return {"final_report": response.content}
# Graph Construction
orchestrator_flow = StateGraph(OrchestratorState)
orchestrator_flow.add_node("orchestrator", orchestrator_plan)
orchestrator_flow.add_node("workers", worker_execute)
orchestrator_flow.add_node("synthesizer", synthesize_report)
orchestrator_flow.add_edge(START, "orchestrator")
orchestrator_flow.add_edge("orchestrator", "workers")
orchestrator_flow.add_edge("workers", "synthesizer")
orchestrator_flow.add_edge("synthesizer", END)
app = orchestrator_flow.compile()
output = app.invoke({"topic": "Quantum Computing Security Implications"})
print(output["final_report"])
有两个需要注意的地方：该工作节点是按顺序处理任务的，因此无法实现并行运行；LangGraph 的 Send API 可以为每个任务分配一个工作节点，并通过状态还原器收集结果。此外，模型有时会将 JSON 数据包裹在 Markdown 代码块中，因此应使用结构化输出来验证计划，而不要仅依赖 json.loads 对原始文本进行解析。
模式 3：ReAct——循环中的推理与行动
ReAct 是在推理和行动之间交替进行：模型先评估当前情况，调用搜索或数据库查询等工具，观察结果，一旦能够给出答案就会停止。
┌────────────────────────┐
▼ │
[START] ──> [Reasoner (Thought)] ───> (Should Call Tool?) ───> [Tool Executor (Act)]
│
└─ (Has Final Answer) ──> [END]
这种模式适用于需要处理无法预先预测的数据的研究、支持及调试类智能体。
推理器会请求执行 ACTION: call_stock_api 或 FINAL: ...，并附上最新的观测结果。工具会返回模拟的报价数据，随后回到 reasoner 以完成循环。should_continue 参数会将循环次数限制在三次以内。
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class ReActState(TypedDict):
user_input: str
agent_thought: str
tool_output: str
final_answer: str
loop_count: int
model = ChatOpenAI(model="gpt-4o", temperature=0)
def reason(state: ReActState):
loop_count = state.get("loop_count", 0) + 1
tool_context = f"\nTool Observation: {state.get('tool_output', '')}" if loop_count > 1 else ""
prompt = f"""You are a ReAct agent. Your goal is to find the current stock price of AAPL.
Current Loop: {loop_count} {tool_context}
Decide your next step. You must respond in one of two ways:
1. If you need data, say: 'ACTION: call_stock_api'
2. If you have the data, provide the answer starting with: 'FINAL: [your answer]'
User Request: {state['user_input']}"""
response = model.invoke([HumanMessage(content=prompt)]).content.strip()
if "FINAL:" in response:
return {"final_answer": response.replace("FINAL:", "").strip(), "loop_count": loop_count, "agent_thought": "done"}
else:
return {"agent_thought": "call_tool", "loop_count": loop_count}
def call_tool(state: ReActState):
print("-> System: Executing external stock database API call...")
mock_api_result = "$185.40 USD (Up 1.2% today)"
return {"tool_output": mock_api_result}
def should_continue(state: ReActState) -> Literal["call_tool", "end"]:
# Hard loop-break guardrail to prevent infinite execution loops
if state["agent_thought"] == "call_tool" and state["loop_count"] < 3:
return "call_tool"
return "end"
react_flow = StateGraph(ReActState)
react_flow.add_node("reasoner", reason)
react_flow.add_node("tool_executor", call_tool)
react_flow.add_edge(START, "reasoner")
react_flow.add_conditional_edges("reasoner", should_continue, {
"call_tool": "tool_executor",
"end": END
})
react_flow.add_edge("tool_executor", "reasoner")
app = react_flow.compile()
result = app.invoke({"user_input": "What is the market status of Apple right now?", "loop_count": 0})
print(f"\nFinal Agent Resolution:\n{result['final_answer']}")
如果在得到 FINAL: 结果之前就达到次数上限，final_answer 将不会被设置，且最后的 print 操作会引发 KeyError，因此需要处理这种情况。实际系统通常使用原生工具调用而非字符串标记来实现类似功能。关于 TypeScript 版本的实现，可参阅 用于 LLM 工具开发的受限智能体循环模式。
模式 4：评估器与优化器
能够自我评估的模型往往会机械地通过自己的输出，因此需要由优化器负责生成和修改内容，同时由独立的、更严格的评估器进行审核。
┌───> [Optimizer (Generate/Refine)] ───> [Evaluator (Critique)]
│ │
└──────────────── (If Rejected) ────────────────┼───> [Approved] ───> [END
这种方法适用于文案撰写、代码生成、翻译以及那些有严格质量或监管要求的场景。温度值为0.7的较便宜模型会先写出初步内容，随后通过多次迭代整合反馈。温度值为0的评估器会检查内容中是否包含future或smart这类关键词，并以固定的ACCEPTED/FEEDBACK格式给出反馈；routing_gate则负责将被拒的内容发回。
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
class EvaluationState(TypedDict):
task: str
draft: str
feedback: str
accepted: bool
iterations: int
generator_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
evaluator_llm = ChatOpenAI(model="gpt-4o", temperature=0)
def generate_draft(state: EvaluationState):
iterations = state.get("iterations", 0) + 1
feedback_context = f"\nPrevious Feedback to incorporate: {state.get('feedback', '')}" if iterations > 1 else ""
prompt = f"""Write a catchy, 3-sentence marketing slogan for: '{state['task']}'.
{feedback_context}
Provide ONLY the slogan."""
response = generator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
return {"draft": response, "iterations": iterations}
def evaluate_draft(state: EvaluationState):
prompt = f"""Review the following marketing slogan for the product '{state['task']}':
Slogan: "{state['draft']}"
CRITERIA: The slogan must include the exact word 'future' or 'smart'.
Respond in EXACTLY the following format:
ACCEPTED: True or False
FEEDBACK: [If rejected, explain what needs fixing. If accepted, leave blank.]"""
response = evaluator_llm.invoke([HumanMessage(content=prompt)]).content.strip()
accepted = "ACCEPTED: True" in response
feedback = response.split("FEEDBACK:")[-1].strip() if not accepted else ""
return {"accepted": accepted, "feedback": feedback}
def routing_gate(state: EvaluationState) -> Literal["refine", "approve"]:
if state["accepted"] or state["iterations"] >= 3:
return "approve"
return "refine"
eval_flow = StateGraph(EvaluationState)
eval_flow.add_node("generator", generate_draft)
eval_flow.add_node("evaluator", evaluate_draft)
eval_flow.add_edge(START, "generator")
eval_flow.add_edge("generator", "evaluator")
eval_flow.add_conditional_edges("evaluator", routing_gate, {
"refine": "generator",
"approve": END
})
app = eval_flow.compile()
result = app.invoke({"task": "Eco-friendly Electric Skateboards", "iterations": 0})
print(f"Final Slogan: {result['draft']}\nTotal Iterations: {result['iterations']}")
无论怎样，该机制在三次迭代后也会直接批准内容，因此输出结果可能并未真正通过审核；应将accepted标志与结果一同保存。在代码中实现此类关键词规则更为经济且可靠。
模式5：人工介入
对于创建表、支出资金或向客户发送邮件等高风险操作，检查点功能可以让流程在敏感节点之前停止，保存当前状态并等待审批。
[START] ──> [Stager] ──> ⛔ (State Saved to DB / Graph Pauses)
│
[DevOps Manager Clicks "Approve"]
│
▼
[Executor (Run Production Deploy)] ──> [END]
可将其用于数据迁移、部署、支付或批量邮件发送操作。
stager负责准备命令，而executor则负责执行该命令。通过加入检查点功能及参数interrupt_before=["executor"]，可在准备阶段后停止流程。每次执行都会通过thread_id进行标识，因此get_state可以显示已保存的值以及待处理的('executor',)步骤；update_state用于记录审批结果，之后可通过invoke(None, config)继续执行流程。
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
class DeploymentState(TypedDict):
command: str
approved: bool
execution_log: str
# 1. Initialize thread checkpoint memory saver
memory = MemorySaver()
def stage_deployment(state: DeploymentState):
print("-> System: Staging server deployment commands...")
return {"command": "sudo systemctl restart production_api"}
def execute_deployment(state: DeploymentState):
print("-> System: Execution approved. Running command on production servers...")
return {"execution_log": f"Successfully executed: {state['command']}"}
hitl_flow = StateGraph(DeploymentState)
hitl_flow.add_node("stager", stage_deployment)
hitl_flow.add_node("executor", execute_deployment)
hitl_flow.add_edge(START, "stager")
hitl_flow.add_edge("stager", "executor")
hitl_flow.add_edge("executor", END)
# CRITICAL: Define the interrupt checkpoint before the executor node runs
app = hitl_flow.compile(checkpointer=memory, interrupt_before=["executor"])
# --- SIMULATING THE ACTIVE DEPLOYMENT WORKFLOW ---
config = {"configurable": {"thread_id": "prod_deploy_001"}}
# 1. Kick off the graph execution
initial_state = app.invoke({"command": "", "approved": False}, config)
# Verify the graph successfully halted its progress
print(f"\n[Current Graph State]: {app.get_state(config).values}")
print(f"[Next Pending Steps]: {app.get_state(config).next}") # Next step will say: ('executor',)
print("\n--- Halting Execution. Waiting for DevOps Manager Review... ---\n")
# 2. Simulate Human Reviewing the State and Updating with Approval
app.update_state(config, {"approved": True}, as_node="stager")
# 3. Resume execution thread seamlessly from the exact checkpoint
final_output = app.invoke(None, config)
print(f"[Final System Output]: {final_output['execution_log']}")
有三点需要注意。MemorySaver仅存在于内存中；那些在重启后仍需持续的暂停操作需要基于数据库的检查点机制。executor从不检查approved状态，因此应添加相应检查或条件逻辑，在检测到拒绝时终止运行。较新版本的LangGraph还提供了interrupt()函数，建议查阅最新文档以了解推荐方案。
选择架构模式
可靠性取决于结构是否与问题匹配，而非模型规模大小或提示词长度：
- Router：处理多种具有不同成本或技能要求的请求类型。
- Orchestrator和workers：将大型任务拆分为多个独立部分进行处理。
- ReAct：所需信息仅在运行时才能确定。
各种模式可以组合使用：路由器可以将任务分配给ReAct智能体，其最终操作需等待批准。在任何情况下都要遵循三条准则：限制每个循环的次数，验证代码所依赖的模型输出，同时记录结果是否获得批准或已尝试次数是否用尽。这样一来，模型无需一次就做到完美，因为工作流允许它进行路由、测试、重试，并最终交由人工处理。
相关阅读
- 在 LangGraph 中构建 ReAct 研究智能体：大脑、手与路由器 — 了解如何将 ReAct 的推理-行动-观察循环实现为 LangGraph 的子图，包括强制反思、迭代预算以及并行散收集研究机制。
- 从零开始构建 AI 智能体：模式、ReAct 与 LangGraph — 学习 AI 智能体的核心概念——规划、工具使用、反思以及 ReAct 模式——并了解 LangChain 和 LangGraph 如何帮助手动构建此类智能体。