05 Agent编排：别再纠结选LangChain还是LangGraph了
目录
- Prompt Chaining / Routing / Parallelization / ReAct / Plan-and-Execute / Orchestrator-workers / Evaluator-Optimizer
三、生产环境绕不开的问题
- 工具设计、意图消歧、可观测性
选框架的焦虑
想做Agent编排，打开搜索框，LangGraph、CrewAI、AutoGen、OpenAI Agents SDK、Google ADK、Dify……每个都说自己是最佳选择。
看看PyPI的下载数据（2026年7月）：
| 框架 | 月下载量 | GitHub Stars |
|---|---|---|
| LangGraph | 6680万 | 37.7K |
| OpenAI Agents SDK | 3156万 | 28K |
| Google ADK | 1561万 | 20.8K |
| CrewAI | 1087万 | 55.9K |
数据摆在这，很多人会想”选LangGraph就对了，下载量最高”。
但Anthropic在《Building Effective Agents》里说了一句被很多人忽略的话：“最成功的实现不用复杂框架，而是用简单、可组合的模式。”
根据经验，这句话是对的。我最终选了LangGraph，但选的过程没那么纠结，因为不管用哪个框架，你要解决的问题是一样的：流程怎么组织、工具怎么设计、异常怎么处理。框架只是帮你省了写胶水代码的功夫，Agent好不好用，取决于你对编排的理解。
框架是工具，不是答案。 决定Agent编排质量的是三件事：用什么模式组织流程、怎么设计工具让模型选对、上线后怎么处理会话并发和错误。这三个问题和你选哪个框架无关，但和你的Agent能不能跑进生产环境有关。
这篇文章讲的就是这三件事。
一、编排到底是什么
一个类比
编排像菜谱，框架像厨具。
菜谱决定：先切什么、再炒什么、火候多大、什么时候放盐。厨具决定：用什么刀、什么锅、什么灶。大多数人纠结用什么厨具，但真正决定菜好不好吃的是菜谱。
类比到Agent：编排决定先识别意图、再选工具、再执行、再生成回答。框架决定用LangGraph还是自己写循环。
Workflow vs Agent
这是编排中最核心的一对概念。但它们不是非此即彼，而是一个光谱的两端。
Workflow：流程确定，每一步预定义，LLM在固定节点做生成或判断。Anthropic的定义是”LLMs and tools are orchestrated through predefined code paths”。
Agent：流程不确定，LLM自己决定下一步做什么。Anthropic的定义是”LLMs dynamically direct their own processes and tool usage”。
# 光谱的左端：Workflow，流程完全确定
def workflow(user_input):
intent = classify_intent(user_input)
if intent == "query":
result = query_database(user_input)
else:
result = search_knowledge(user_input)
return format_response(result)
# 光谱的右端：Agent，流程完全不确定
def agent(user_input, tools):
messages = [{"role": "user", "content": user_input}]
while True:
response = llm.chat(messages, tools=tools)
if response.tool_calls:
result = execute_tool(response.tool_calls[0])
messages.append({"role": "tool", "content": result})
else:
return response.content
大多数生产系统在光谱中间：Workflow定义主流程，关键节点用Agent处理不确定性。先判断你的任务需要多大的灵活性，再决定用哪种。
二、七种编排模式
业界总结了几种核心编排模式。其中五种来自Anthropic的《Building Effective Agents》（2024年12月），另外两种来自学术界和工程实践。
全景
| 模式 | 来源 | 一句话 | 适合场景 |
|---|---|---|---|
| Prompt Chaining | Anthropic | 步骤串联，前一步输出是后一步输入 | 流程确定的任务 |
| Routing | Anthropic | 分类输入，导向不同处理逻辑 | 多意图混合场景 |
| Parallelization | Anthropic | 多个LLM同时处理子任务，结果聚合 | 独立子任务可并行 |
| ReAct | Yao et al. 2022 | 思考→行动→观察，循环往复 | 流程不确定，需灵活调工具 |
| Plan-and-Execute | LangGraph | 先规划再执行，失败时重新规划 | 复杂多步骤任务 |
| Orchestrator-workers | Anthropic | 中央LLM动态拆解任务分配给worker | 子任务不预定义 |
| Evaluator-Optimizer | Anthropic | 一个生成，另一个评估，循环优化 | 对输出质量要求高 |
七种编排模式
这七种不是七选一。大多数生产系统是混合模式：主流程用Prompt Chaining或Routing，关键节点用ReAct，独立子任务用Parallelization并行，质量要求高的地方加Evaluator-Optimizer。
Anthropic的建议：从最简单的模式开始。 能用Prompt Chaining解决就用它，需要灵活调用工具再加ReAct。先用LLM API直接写代码理解原理，再考虑用框架。
下面展开讲三种最常用的模式。另外四种简要说明：
- Parallelization（并行化）：多个子任务同时执行，结果聚合。适合独立子任务可并行的场景，比如同时搜索多个数据源再合并结果。
- Plan-and-Execute（规划执行）：先让LLM生成完整计划，再逐步执行，失败时重新规划。适合复杂多步骤任务，比如”帮我做一份季度分析报告”。
- Orchestrator-workers（编排者-工人）：中央LLM动态拆解任务分配给worker，适合子任务不预定义的场景，比如”帮我调研这5个竞品”。
- Evaluator-Optimizer（评估优化）：一个LLM生成，另一个评估，循环优化直到满意。适合对输出质量要求高的场景，比如生成技术文档后自动审阅修订。
这四种模式各有适用场景，但在实际项目中，大多数系统用前三种（Chaining/Routing/ReAct）就能覆盖80%的需求。后四种是进阶选项，等你遇到具体问题再学不迟。
Prompt Chaining：步骤串联
最简单的编排方式。任务分解为序列步骤，每步处理前一步的输出。
举个例子，政策解读：
用户输入："2024年产业园申报有什么要求？"
→ 步骤1：提取关键词（"2024""产业园""申报""要求"）
→ 步骤2：根据关键词搜索政策文档
→ 步骤3：根据文档内容生成解读
→ 步骤4：格式化输出
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
llm = ChatOpenAI(model="deepseek-v4-flash")
extract_prompt = ChatPromptTemplate.from_template(
"从以下问题中提取搜索关键词，返回JSON数组：\n{question}"
)
answer_prompt = ChatPromptTemplate.from_template(
"根据以下政策文档，回答用户问题。引用文档时标注来源。\n"
"文档：{docs}\n问题：{question}"
)
def chain(question: str) -> str:
keywords = (extract_prompt | llm).invoke({"question": question})
docs = search_docs(parse_keywords(keywords))
answer = (answer_prompt | llm).invoke({"docs": docs, "question": question})
return answer
Anthropic特别提到，可以在步骤之间加”门控”（programmatic checks），比如检查中间结果是否符合预期格式，不符合就重试。这比让LLM自己判断”上一步对不对”可靠得多。
Routing：路由分发
分类输入，导向专门的后续处理。Anthropic的原话是”Without this workflow, optimizing for one kind of input can hurt performance on other inputs”，不分类，优化一种输入就会伤害另一种。
VALID_ROUTES = {"knowledge", "data", "chat", "report"}
def route(user_input: str) -> str:
prompt = f"""判断用户意图，返回一个词：
- knowledge：知识问答（问政策、问规定）
- data：数据查询（问数字、问统计）
- report：报告生成（要求生成分析）
- chat：闲聊
用户输入：{user_input}
只返回一个词，不要解释。"""
intent = llm.invoke(prompt).content.strip()
return intent if intent in VALID_ROUTES else "chat"
def handle(user_input: str) -> str:
target = route(user_input)
handlers = {
"knowledge": knowledge_graph,
"data": data_graph,
"report": report_graph,
"chat": chat_graph
}
return handlers.get(target, chat_graph).invoke(user_input)
关键设计：分类器要轻量（用小模型、简单Prompt），分类结果要记录（方便分析路由准确率），要有默认路由（分类失败时的兜底）。
ReAct：工具循环
来源：Yao et al.的论文《ReAct: Synergizing Reasoning and Acting in Language Models》（arXiv:2210.03629，ICLR 2023）。
LLM交替生成推理痕迹（Thought）和执行动作（Action），根据环境反馈（Observation）调整后续行为。
用户："帮我查一下山东省2024年产业园的申报情况"
思考：需要先搜索产业园列表
行动：search_park(region="山东", year=2024)
观察：找到12个产业园
思考：用户可能想看汇总
行动：analyze_park_status(parks=[...])
观察：8个已申报，3个待申报，1个未开始
生成回答
def react(user_input: str, tools: list, max_iterations: int = 10) -> str:
messages = [
{"role": "system", "content": "你是一个数据查询助手。"},
{"role": "user", "content": user_input}
]
for i in range(max_iterations):
response = llm.chat(messages, tools=tools)
if not response.tool_calls:
return response.content
for call in response.tool_calls:
result = execute_tool(call.name, call.arguments)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": json.dumps(result, ensure_ascii=False)
})
# 防止上下文溢出
if count_tokens(messages) > MAX_CONTEXT_TOKENS * 0.7:
messages = truncate_messages(messages, keep_recent=6)
return "抱歉，处理过程超过限制，请简化您的问题。"
ReAct的局限性：每步都需要完整的LLM调用，延迟和成本随步骤数线性增长；推理链越长，后续步骤越容易放大前面的错误。研究还发现（arXiv:2604.02155），更多推理不等于更好的推理。在小模型上，简短推理链比长推理链准确率高45%。
三、生产环境绕不开的问题
Demo阶段不会遇到的问题，上线后一个接一个。
3.1 会话与并发
Demo阶段会话存在内存里，刷新就没了。生产环境会话要持久化，支持恢复。同时还要处理并发。用户发了一个问题，Agent正在处理，用户又发了一个新问题，两个请求同时修改会话状态，结果互相覆盖。
import asyncio
# 会话快照：每个请求结束后保存状态
@app.post("/chat")
async def chat(request: ChatRequest):
session = session_manager.get_or_create(request.session_id)
messages = session.get_messages()
# 获取会话级锁，防止并发
lock = session_locks.get_lock(request.session_id)
if lock.locked():
return sse_event("error", "上一个问题还在处理中，请稍等")
try:
acquired = await asyncio.wait_for(lock.acquire(), timeout=30)
if not acquired:
return sse_event("error", "获取锁超时，请重试")
try:
result = await run_agent(request.message, messages)
session.save_turn({
"user_message": request.message,
"tool_calls": result.tool_calls,
"assistant_message": result.content,
"timestamp": now()
})
return sse_stream(result)
finally:
lock.release()
except asyncio.TimeoutError:
return sse_event("error", "获取锁超时，请重试")
不只是保存消息，还要保存工具调用记录、PlanTrace、用户反馈。这些数据是后续排查问题和优化的基础。
3.2 取消机制：用户等不及了点取消
Agent调了3个工具，用户等了10秒不耐烦了，点了取消。取消是协作式的：设置取消标志，当前节点主动检查。
class CancelFlag:
def __init__(self):
self._flags: dict[str, bool] = {}
def set(self, request_id: str):
self._flags[request_id] = True
def is_cancelled(self, request_id: str) -> bool:
return self._flags.get(request_id, False)
# 每个节点检查取消标志
async def think_node(state: AgentState):
if cancel_flag.is_cancelled(state["request_id"]):
return {"status": "cancelled", "message": "用户取消了请求"}
response = await llm.achat(state["messages"])
return {"messages": [response]}
取消后会话状态保留，用户可以继续新的对话，不会丢失之前的上下文。
3.3 错误处理：不能所有错误同一种处理
Demo阶段try-catch一切，出错就报”系统错误”。生产环境需要两类错误分别处理。
基础设施错误：超时、限流、权限，四级分类：
async def execute_with_retry(fn, *args, max_retries=3):
for attempt in range(max_retries):
try:
return await fn(*args)
except TimeoutError:
if attempt < max_retries - 1:
await asyncio.sleep(2 ** attempt)
continue
raise
except RateLimitError:
return await fn.with_model("deepseek-v4-flash")(*args)
except PermissionError:
return "抱歉，您没有权限执行此操作"
except Exception as e:
if "token" in str(e).lower() and "limit" in str(e).lower():
return await fn.with_truncated_history()(*args)
raise
return "处理过程遇到问题，请简化您的问题后重试"
| 错误类型 | 处理策略 | 例子 |
|---|---|---|
| 可重试 | 指数退避重试 | LLM超时、网络抖动 |
| 可降级 | 换策略 | LLM拒绝→改写Prompt、Token超限→截断历史 |
| 不可重试 | 直接告知用户 | 权限不够、数据不存在 |
| 强制停止 | 最大迭代限制 | 死循环、Token爆炸 |
Agent特有错误，比基础设施错误更难排查，因为它们不会抛异常，而是”静默失败”：
Agent错误分类
| 错误类型 | 表现 | 检测方法 |
|---|---|---|
| 工具幻觉 | 模型编造了不存在的工具名或参数 | 对比tools_selected和实际工具列表 |
| 工具误选 | 选了错的工具但没报错 | PlanTrace里看工具选择和用户意图是否匹配 |
| 推理错误 | 逻辑链断了，后续步骤全部错 | 检查中间结果是否符合预期格式/范围 |
| 错误累积 | 前一步的错在后续步骤中放大 | 对比每步的输出质量，发现退化趋势 |
一个真实场景：用户问”帮我查山东产业园的数据”，Agent选了search_policy工具而不是query_data。没有报错，返回了一堆政策文档，用户以为数据查完了。这种错误只有看PlanTrace才能发现。
研究表明（arXiv:2510.07248），工具幻觉的一个主要来源是”schema misalignment”，模型在预训练时记住了某些命名习惯，和你定义的工具名冲突。解决方案之一是让工具名更符合模型的预训练分布（比如用search_documents而不是doc_qry）。
研究还发现（arXiv:2604.02155），推理错误和推理长度是非单调关系。更多推理不等于更好的推理。在小模型上，简短推理链比长推理链准确率高45%。过长的推理链中，28%是选错了工具，18%是编造了工具。
这些Agent特有错误不会出现在try-catch里，只能通过PlanTrace和评测集来发现。
3.4 成本控制：一个简单问题调了8次LLM
用户问”销售额怎么样”，Agent不知道”怎么样”什么意思，于是自己决定查趋势、查同比、查分区域、查异常。一个简单问题，调了8次LLM，token消耗5000+。
几个控制策略：
# 1. 简单问题用Workflow，不用Agent Loop
def handle_simple_query(message: str) -> str:
intent = classify_intent(message)
if intent in ["greeting", "simple_query"]:
return direct_answer(message)
return agent_loop(message)
# 2. 分级模型：简单任务用小模型
def get_model_for_task(complexity: str):
models = {
"simple": "deepseek-v4-flash", # 便宜
"medium": "deepseek-v4-pro", # 中等
"complex": "deepseek-v4-reasoning" # 贵但强
}
return ChatOpenAI(model=models.get(complexity, models["simple"]))
# 3. 对话历史压缩
def compress_history(messages: list, max_tokens: int = 4000):
if count_tokens(messages) <= max_tokens:
return messages
system_msgs = [m for m in messages if m["role"] == "system"]
recent = messages[-6:] # 最近3轮
middle = messages[len(system_msgs):-6]
if middle:
summary = llm.invoke(f"总结以下对话的要点：{middle}").content
return system_msgs + [{"role": "system", "content": f"历史摘要：{summary}"}] + recent
return system_msgs + recent
成本控制要从设计开始，不是上线后发现账单太高再改。
四、几条编排经验
工具设计决定Agent效果
Anthropic的经验：工具说明书的质量直接影响Agent的效果。很多时候Agent效果不好，不是模型不行，是工具定义有问题。
工具说明书四段式：
search_policy:
name: search_policy
description: 搜索政策文档
parameters:
query:
type: string
required: true
description: 搜索关键词
region:
type: string
required: false
description: 地区筛选
returns: |
{"results": [{"title": "标题", "content": "内容", "source": "来源"}]}
设计原则：
- 一件事一个工具：不要一个工具做太多事，Agent会选错
- 参数明确：类型、必填、校验规则都要定义，参数说明不清Agent会传错
- 返回结构化：返回JSON，不要返回自然语言
- 幂等：同一请求重放多次，副作用只生效一次
Demo阶段用硬编码规则选工具：
# 硬编码规则：扩展难、回归难
if "政策" in query or "文件" in query:
use search_policy()
elif "数据" in query or "统计" in query:
use query_data()
生产环境用工具说明书，让LLM自己选：
# 新增工具只需加说明书
tools = [
Tool(name="search_policy", description="搜索政策文档", ...),
Tool(name="query_data", description="查询统计数据", ...),
]
好处是新增工具不需要改规则，LLM根据说明书自己判断。选错了可以查PlanTrace，反过来优化说明书。
意图消歧常被低估
用户问”销售额怎么样”，至少有5种理解：要趋势？要同比？要分区域？要异常检测？要排名？
换一个更好的模型不一定能解决这个问题。研究表明，工具选择失败（本质上是意图理解失败）占Agent全部失败的30%以上（arXiv:2604.02155），而通过优化工具命名和描述（不换模型）就能减少80%的工具幻觉（arXiv:2510.07248）。
四种策略：
| 策略 | 做法 | 适合场景 |
|---|---|---|
| 不追问 | 用最常见的理解直接回答 | 简单问题、容错率高 |
| 给选项 | 给2-3个选项让用户选 | 关键决策 |
| 先粗后细 | 先给粗略回答，再追问细化 | 探索性问题 |
| 历史推断 | 根据对话历史推断意图 | 多轮对话 |
追问太多用户烦，追问太少答非所问。平衡点是：简单问题不追问，关键决策给选项，探索性问题先粗后细。
一个实操建议：把意图消歧的策略写进System Prompt。比如”当用户问题有多种理解时，先给最常见的一种回答，然后列出其他可能的理解让用户选择”。这比换一个更大的模型便宜得多。
可观测是迭代的基础
Agent出了问题，不知道哪里出的、为什么出的，就没法优化。
最小可观测方案：记录每一问的完整PlanTrace。
plan_trace = {
"request_id": "req-2026-07-21-001",
"user_query": "上个月销售额多少",
"intent": "data_query",
"tools_selected": ["query_sales"],
"tool_calls": [
{
"tool": "query_sales",
"args": {"period": "last_month"},
"result": {"total": 1234567},
"latency_ms": 230,
"error": None
}
],
"final_answer": "上个月销售额为123.46万元",
"token_usage": {"input": 1200, "output": 350},
"total_latency_ms": 1850
}
有了PlanTrace，排查问题就容易了：
| 现象 | 看PlanTrace哪个字段 |
|---|---|
| 选错工具 | tools_selected 和用户意图是否匹配 |
| 工具返回空 | tool_calls.result |
| 回答不对 | final_answer 和 tool_calls 的结果对比 |
| 响应太慢 | tool_calls[].latency_ms 找到慢的那步 |
| 成本太高 | token_usage 看哪步消耗最多 |
PlanTrace不只是调试工具，更是迭代的基础。收集线上PlanTrace，按意图分类，找到高频失败模式，针对性优化工具说明书或Prompt。这是Agent持续改进的闭环。
从哪里开始
看完这些，第一步做什么？
选一个核心业务场景，用Prompt Chaining写最小版本。 不要一上来就搞多Agent、Evaluator-Optimizer这些复杂模式。找一个你最熟悉的业务场景（比如政策查询、数据统计），用最简单的串联流程跑通。
加上PlanTrace。 从第一版就记录每一问的工具调用、输入输出、耗时。这不只是调试工具，更是后续优化的基础。没有PlanTrace，你永远不知道Agent为什么出错。
用评测集验证。 准备10-20个典型问题和期望答案，每次改Prompt或工具后跑一遍。不需要复杂的评测框架，一个脚本够了。核心是建立”改了什么→效果变好还是变差”的反馈循环。
遇到具体问题再加模式。 意图分不清加Routing，子任务可并行加Parallelization，输出质量不稳定加Evaluator-Optimizer。每加一个模式，都要有明确的问题驱动，不要为了”架构好看”而加。
总结
从”选框架”到”理解模式”。 框架会变，但编排模式不会变。七种核心模式不依赖任何特定框架。Anthropic说得对：最成功的实现不用复杂框架，而是用简单、可组合的模式。
从”Demo能跑”到”生产能用”。 会话管理、并发控制、错误分级、成本控制，这些Demo阶段不会遇到的问题，才是Agent能不能上线的关键。Agent特有的错误（工具幻觉、推理错误、错误累积）比基础设施错误更难排查，需要通过PlanTrace和评测集来发现。
每个人的情况不同，以上是根据实际项目总结的一些理解，不是标准答案。关键是找到适合自己团队的方式。欢迎评论区交流~
相关关键词：AI Agent编排、LangGraph、LangChain、Agent框架对比、生产级Agent、ReAct模式、Agent开发实战
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
5
0- 0
已为社区贡献3条内容
所有评论(0)