1. 从单 Agent 到多 Agent：为什么需要编排模式
单个 Agent 能写代码、能查文档、能做决策，但一旦业务链路变长，问题就暴露了：上下文越堆越长、推理成本指数上升、一个环节出错整条链路卡死。我试过用一个 Agent 硬扛「需求拆解→代码生成→测试→部署」全流程，结果单次调用 token 量轻松破十万，成本高不说，出错后它自己根本发现不了。
这时候就需要把任务拆开，交给多个专职 Agent 协作。而协作方式的不同，就形成了四种主流编排模式：Sequential 串行、Parallel 并行、Hierarchical 层级、Orchestration 中心调度。它们不是互斥的，而是像积木一样可以组合——大项目里往往是 Orchestration 顶层调度，下面挂 Hierarchical 子树，子树里再嵌 Sequential 链和 Parallel 组。
这篇文章聚焦这四种模式的差异与选型依据，每种都给出可运行的配置骨架和验证动作。为了让调用验证更省事，我会用 TaoToken 统一 Key 和 API 通道接入，这样切换模型、管理密钥都不用改代码结构。适合正在搭多 Agent 系统、纠结该用哪种编排方式的开发者。
2. TaoToken 前置准备：统一 Key 与 API 通道
多 Agent 系统最烦的一件事是：不同 Agent 可能要用不同模型，每个模型一套 Key、一套 Base URL，管理起来头大。TaoToken 的思路是提供一个统一的 API 通道，你只需要一个 Key，就能在多个模型之间切换。
2.1 获取 API Key
打开 TaoToken 控制台，进入 API Keys 页面创建一个新 Key。建议按项目或按环境（开发/测试/生产）分别建 Key，方便后续做用量隔离和权限回收。
创建后把 Key 存到环境变量里，别硬编码进代码：
export TAOTOKEN_API_KEY="sk-你的key"2.2 确认 API 地址
TaoToken 的 API 入口是https://taotoken.net/api，兼容 OpenAI 的接口格式。也就是说，你原来用openaiSDK 写的代码，只需要改base_url和api_key两个参数就能接过来。
2.3 用 curl 做一次最小验证
在写复杂编排之前，先用一条最简单的请求确认通道是通的：
curl https://taotoken.net/api/v1/chat/completions \ -H "Content-Type: application/json" \ -H "Authorization: Bearer $TAOTOKEN_API_KEY" \ -d '{ "model": "gpt-4o-mini", "messages": [{"role": "user", "content": "回复两个字：通了"}] }'如果返回里能看到正常的choices结构，说明 Key 和通道都没问题。这一步别跳过，后面所有编排示例都依赖这个基础通道。
3. Sequential 串行编排：最基础的流水线
3.1 核心逻辑
Sequential 就是一条线：Agent1 的输出喂给 Agent2，Agent2 的输出喂给 Agent3，依次往下。适合步骤明确、前后有依赖的任务，比如「翻译→润色→排版」这种。
它的数学模型很直白：总耗时是所有 Agent 耗时之和，总成功率是所有 Agent 成功率的乘积。这意味着链路越长，整体成功率越低——所以串行链一般控制在 3 到 5 个环节，每个环节加上重试机制。
3.2 配置骨架
用 LangChain 的SequentialChain可以快速搭起来。先装依赖：
pip install langchain langchain-openai然后定义三个环节的 Prompt 和 Chain：
import os from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.chains import LLMChain, SequentialChain llm = ChatOpenAI( model="gpt-4o-mini", base_url="https://taotoken.net/api/v1", api_key=os.environ["TAOTOKEN_API_KEY"], temperature=0.3, ) translate_prompt = ChatPromptTemplate.from_template( "把下面的中文翻译成英文，保持技术术语准确：\n{source_text}" ) translate_chain = LLMChain( llm=llm, prompt=translate_prompt, output_key="translated" ) polish_prompt = ChatPromptTemplate.from_template( "把下面的英文润色成学术论文风格：\n{translated}" ) polish_chain = LLMChain( llm=llm, prompt=polish_prompt, output_key="polished" ) format_prompt = ChatPromptTemplate.from_template( "把下面的英文排版成 Markdown 格式，标题用 ##：\n{polished}" ) format_chain = LLMChain( llm=llm, prompt=format_prompt, output_key="final" ) pipeline = SequentialChain( chains=[translate_chain, polish_chain, format_chain], input_variables=["source_text"], output_variables=["final"], verbose=True, ) result = pipeline({"source_text": "多 Agent 编排是复杂 LLM 应用的核心骨架。"}) print(result["final"])3.3 验证动作
跑通后观察verbose=True打印的中间输出，确认每个环节的output_key都被正确传递。如果某个环节输出为空，多半是 Prompt 里的变量名和上一个环节的output_key对不上。
4. Parallel 并行编排：把独立子任务同时跑
4.1 核心逻辑
Parallel 适合子任务之间没有依赖的场景。比如你要同时分析一篇文档的「技术点」「风险点」「商业价值」，这三个分析互不干扰，完全可以并发跑，最后汇总。
总耗时约等于最慢那个子任务的时间，而不是所有子任务之和。但要注意：并发会带来瞬时 token 消耗峰值，如果用的是按量计费的通道，得留意速率限制。
4.2 配置骨架
用asyncio配合 LangChain 的异步接口：
import asyncio import os from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.chains import LLMChain llm = ChatOpenAI( model="gpt-4o-mini", base_url="https://taotoken.net/api/v1", api_key=os.environ["TAOTOKEN_API_KEY"], ) def build_chain(instruction): prompt = ChatPromptTemplate.from_template( instruction + "\n\n文档内容：\n{doc}" ) return LLMChain(llm=llm, prompt=prompt) tech_chain = build_chain("提取这篇文档的技术要点，列 3 条：") risk_chain = build_chain("提取这篇文档的风险点，列 3 条：") biz_chain = build_chain("提取这篇文档的商业价值，列 3 条：") async def run_parallel(doc): tasks = [ tech_chain.ainvoke({"doc": doc}), risk_chain.ainvoke({"doc": doc}), biz_chain.ainvoke({"doc": doc}), ] results = await asyncio.gather(*tasks) return { "tech": results[0]["text"], "risk": results[1]["text"], "biz": results[2]["text"], } doc = "某公司发布新一代推理芯片，算力提升 3 倍，但功耗和散热是主要挑战。" out = asyncio.run(run_parallel(doc)) for k, v in out.items(): print(f"=== {k} ===\n{v}\n")4.3 验证动作
在asyncio.gather前后各打一个时间戳，对比串行执行的总耗时。正常情况下并行版本应该接近最慢那个子任务的耗时。如果发现耗时和串行差不多，检查是不是某个环节用了同步的invoke而不是ainvoke。
5. Hierarchical 层级编排：复杂任务的分层决策
5.1 核心逻辑
Hierarchical 模仿公司组织架构：顶层有个「经理 Agent」负责拆解目标、分配任务，底层有多个「员工 Agent」负责执行。经理不干具体活，只做决策和协调。
它适合任务复杂、需要多层级判断的场景。比如一个「技术方案评审」系统：经理 Agent 先判断方案属于前端还是后端，再分给对应的专家 Agent 评审，最后汇总意见。
5.2 配置骨架
用 CrewAI 来演示，它对层级协作的支持比较直观：
pip install crewaiimport os from crewai import Agent, Task, Crew, Process from langchain_openai import ChatOpenAI llm = ChatOpenAI( model="gpt-4o-mini", base_url="https://taotoken.net/api/v1", api_key=os.environ["TAOTOKEN_API_KEY"], ) manager = Agent( role="技术评审经理", goal="判断方案类型并分配给合适的专家", backstory="你负责统筹技术方案评审，只做分配和汇总。", llm=llm, allow_delegation=True, ) frontend_expert = Agent( role="前端专家", goal="评审前端相关方案", backstory="你精通前端架构与性能优化。", llm=llm, ) backend_expert = Agent( role="后端专家", goal="评审后端相关方案", backstory="你精通后端架构与数据库设计。", llm=llm, ) review_task = Task( description="评审以下技术方案，给出可行性意见：{proposal}", expected_output="一份包含风险点和改进建议的评审报告", agent=manager, ) crew = Crew( agents=[manager, frontend_expert, backend_expert], tasks=[review_task], process=Process.hierarchical, manager_llm=llm, ) result = crew.kickoff(inputs={"proposal": "把用户中心从单体拆成微服务，前端改用 SSR。"}) print(result)5.3 验证动作
观察manager是否真的把任务分给了frontend_expert和backend_expert。如果所有活都是 manager 自己干的，检查allow_delegation=True是否生效，以及Process.hierarchical是否配置正确。
6. Orchestration 中心调度：动态任务的中枢
6.1 核心逻辑
Orchestration 是最灵活的模式：有一个中央调度 Agent，它不执行具体任务，只负责监听状态、分配任务、处理异常、汇总结果。它适合任务流程动态变化、需要实时调整的场景，比如客服机器人要根据用户意图随时切换处理路径。
6.2 配置骨架
用 AutoGen 演示中心调度的思路：
pip install pyautogenimport os import autogen config_list = [{ "model": "gpt-4o-mini", "base_url": "https://taotoken.net/api/v1", "api_key": os.environ["TAOTOKEN_API_KEY"], }] llm_config = {"config_list": config_list, "temperature": 0.2} orchestrator = autogen.AssistantAgent( name="Orchestrator", system_message="你是调度中心，负责判断用户意图并分配给对应专家，不直接回答业务问题。", llm_config=llm_config, ) billing_agent = autogen.AssistantAgent( name="BillingAgent", system_message="你只处理账单相关问题。", llm_config=llm_config, ) tech_agent = autogen.AssistantAgent( name="TechAgent", system_message="你只处理技术故障问题。", llm_config=llm_config, ) user = autogen.UserProxyAgent( name="User", human_input_mode="NEVER", max_consecutive_auto_reply=3, code_execution_config=False, ) groupchat = autogen.GroupChat( agents=[orchestrator, billing_agent, tech_agent, user], messages=[], max_round=6, ) manager = autogen.GroupChatManager(groupchat=groupchat, llm_config=llm_config) user.initiate_chat(manager, message="我的账单这个月多扣了 50 块，帮我查一下。")6.3 验证动作
看调度中心是否把「账单问题」路由给了BillingAgent而不是TechAgent。如果路由错了，调整orchestrator的system_message，把意图分类规则写得更明确。
7. 四种模式对比与选型依据
| 模式 | 适用场景 | 耗时特征 | 容错性 | 典型组合位置 |
|---|---|---|---|---|
| Sequential | 步骤明确、前后依赖 | 各环节耗时之和 | 低，需逐环节重试 | 底层子链 |
| Parallel | 子任务独立、追求效率 | 最慢子任务耗时 | 中，单点失败可降级 | 底层子组 |
| Hierarchical | 任务复杂、需分层决策 | 取决于决策深度 | 中高，经理可重分配 | 中间子树 |
| Orchestration | 流程动态、需实时调整 | 取决于调度开销 | 高，可动态重规划 | 顶层调度 |
选型时先问三个问题：子任务之间有没有依赖？流程是固定的还是动态的？出错后能不能自动恢复？依赖强、流程固定就用 Sequential；无依赖、要提速就用 Parallel；需要分层判断就用 Hierarchical；流程会变、要容错就用 Orchestration。
实际项目里很少只用一种。比如一个 AI 开发助手，顶层用 Orchestration 判断 issue 类型，中间用 Hierarchical 拆解任务，底层用 Sequential 串起「定位→修复→测试」，其中「后端修复/前端修复」又可以并行跑。
8. 本篇常见错排查
报错一：AuthenticationError: Invalid API key
先确认环境变量TAOTOKEN_API_KEY有没有正确导出，再确认base_url是不是https://taotoken.net/api/v1。注意末尾的/v1别漏，很多兼容接口的路径都带这一层。
报错二：RateLimitError在 Parallel 模式下频繁出现
并发请求会瞬间打高 QPS。解决办法是给asyncio.gather加信号量限流：
sem = asyncio.Semaphore(3) async def limited(chain, payload): async with sem: return await chain.ainvoke(payload)报错三：Sequential 链中间输出为空
九成是output_key和下一个 Prompt 的变量名不一致。把verbose=True打开，逐个环节核对输入输出变量名。
报错四：Hierarchical 模式下 manager 不分配任务
检查allow_delegation是否为True，以及Process.hierarchical是否真的生效。有些版本需要显式传manager_llm。
报错五：Orchestration 路由错误
调度中心的system_message太模糊。把意图分类规则写成明确的 if-else 式描述，比如「涉及金额、扣费、退款的一律归 BillingAgent」。
9. 接入验证与后续动作
四种模式的配置骨架都跑通后，建议做一次端到端验证：用同一个业务场景（比如「分析一篇技术文档」），分别用四种模式实现，对比耗时、成本和输出质量。这样你对选型的直觉会建立得很快。
如果你还在搭基础通道，先去 TaoToken 的 API Keys 页面把 Key 建好，再对照接入文档确认base_url和请求格式。想先验证模型输出质量，可以直接在模型对话里试几条 Prompt。如果是长期做编码类 Agent、需要稳定跑大量请求，可以看看 Coding Plan，用量和成本会更可控。
编排模式本身不复杂，难的是根据业务特征选对组合方式。先把 Sequential 和 Parallel 这两个基础款用熟，再往上叠 Hierarchical 和 Orchestration，踩坑会少很多。
