内容速览章节核心内容一、编排到底是什么Workflow vs Agent、编排在架构中的位置二、七种编排模式Prompt Chaining / Routing / Parallelization / ReAct / Plan-and-Execute / Orchestrator-workers / Evaluator-Optimizer三、生产环境绕不开的问题会话管理、并发控制、错误分级、Agent特有错误、成本控制四、几条编排经验工具设计、意图消歧、可观测性从哪里开始四步上手指南总结两条核心转变选框架的焦虑想做Agent编排打开搜索框LangGraph、CrewAI、AutoGen、OpenAI Agents SDK、Google ADK、Dify……每个都说自己是最佳选择。看看PyPI的下载数据2026年7月框架月下载量GitHub StarsLangGraph6680万37.7KOpenAI Agents SDK3156万28KGoogle ADK1561万20.8KCrewAI1087万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, toolstools) 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 ChainingAnthropic步骤串联前一步输出是后一步输入流程确定的任务RoutingAnthropic分类输入导向不同处理逻辑多意图混合场景ParallelizationAnthropic多个LLM同时处理子任务结果聚合独立子任务可并行ReActYao et al. 2022思考→行动→观察循环往复流程不确定需灵活调工具Plan-and-ExecuteLangGraph先规划再执行失败时重新规划复杂多步骤任务Orchestrator-workersAnthropic中央LLM动态拆解任务分配给worker子任务不预定义Evaluator-OptimizerAnthropic一个生成另一个评估循环优化对输出质量要求高七种编排模式这七种不是七选一。大多数生产系统是混合模式主流程用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(modeldeepseek-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 answerAnthropic特别提到可以在步骤之间加”门控”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.03629ICLR 2023。LLM交替生成推理痕迹Thought和执行动作Action根据环境反馈Observation调整后续行为。用户帮我查一下山东省2024年产业园的申报情况 思考需要先搜索产业园列表 行动search_park(region山东, year2024) 观察找到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, toolstools) 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_asciiFalse) }) # 防止上下文溢出 if count_tokens(messages) MAX_CONTEXT_TOKENS * 0.7: messages truncate_messages(messages, keep_recent6) 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(), timeout30) 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_retries3): 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次LLMtoken消耗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(modelmodels.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(namesearch_policy, description搜索政策文档, ...), Tool(namequery_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和评测集来发现。每个人的情况不同以上是根据实际项目总结的一些理解不是标准答案。关键是找到适合自己团队的方式。