为什么需要 Agent 编排
单个 Agent 能力有限。复杂任务需要多个 Agent 协作：一个负责检索、一个负责分析、一个负责生成报告。如何编排这些 Agent 是生产环境的核心问题。
编排模式分类
| 模式 | 结构 | 适用场景 | 复杂度 |
|---|---|---|---|
| 串行 | A -> B -> C | 流水线任务 | 低 |
| 并行 | A,B,C -> 合并 | 独立子任务 | 低 |
| 循环 | A -> B -> 判断 -> (A 或 结束) | 迭代优化 | 中 |
| 路由 | Input -> Router -> [A/B/C] | 分类分发 | 中 |
| 监督者 | Supervisor -> [A,B,C] | 中心调度 | 中 |
| 分层 | Top Supervisor -> Sub-Supervisors -> Workers | 大规模团队 | 高 |
| 图式 | DAG/状态机 | 复杂工作流 | 高 |
1. Router Pattern
一个路由 Agent 根据输入决定调用哪个专家 Agent：
from typing import Literal
from pydantic import BaseModel
class RouteDecision(BaseModel):
next_agent: Literal["coder", "researcher", "writer"]
reason: str
async def router_agent(query: str) -> RouteDecision:
prompt = f"""分析用户请求，决定由哪个 Agent 处理:
- coder: 编程/代码/技术实现
- researcher: 调研/搜索/信息收集
- writer: 写作/总结/翻译
用户请求: {query}"""
resp = await llm.beta.chat.completions.parse(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
response_format=RouteDecision
)
return resp.choices[0].message.parsed
async def routed_pipeline(query: str):
decision = await router_agent(query)
agents = {"coder": coder_agent, "researcher": researcher_agent, "writer": writer_agent}
agent = agents[decision.next_agent]
return await agent(query)
2. Supervisor Pattern
Supervisor 管理多个 Worker Agent，决定调用顺序、判断是否完成：
class Supervisor:
def __init__(self, workers: dict, llm, max_steps=10):
self.workers = workers
self.llm = llm
self.max_steps = max_steps
async def run(self, task: str) -> str:
context = {"task": task, "history": [], "results": {}}
for step in range(self.max_steps):
decision = await self._plan(context)
if decision["action"] == "FINISH":
return decision.get("final_answer", "")
if decision["action"] == "CALL_WORKER":
worker = self.workers[decision["worker_name"]]
result = await worker(decision["task_for_worker"])
context["history"].append({
"worker": decision["worker_name"],
"input": decision["task_for_worker"],
"output": result
})
context["results"][decision["worker_name"]] = result
return "Max steps reached"
async def _plan(self, context) -> dict:
prompt = f"""你是任务监督者。根据当前状态决定下一步。
任务: {context['task']}
已执行步骤: {context['history']}
可选动作: CALL_WORKER 或 FINISH
输出 JSON: {{"action": "...", "worker_name": "...", "task_for_worker": "...", "final_answer": "..."}}"""
resp = await self.llm.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"}
)
return json.loads(resp.choices[0].message.content)
3. 并行编排
多个 Agent 同时工作，结果合并：
import asyncio
async def parallel_orchestration(query: str):
tasks = [
search_web(query),
search_knowledge_base(query),
analyze_query_intent(query)
]
web_results, kb_results, intent = await asyncio.gather(*tasks)
final_answer = await merge_agent(
query=query, web=web_results, kb=kb_results, intent=intent
)
return final_answer
4. 循环迭代模式
Agent 输出经过审查，不合格则重新执行：
class IterativeRefiner:
async def run(self, task: str, max_iter: int = 3):
result = await generator_agent(task)
for i in range(max_iter):
review = await reviewer_agent(task, result)
if review.score >= 0.8:
return result
result = await generator_agent(
f"{task}\n\n审查意见: {review.feedback}\n请改进。"
)
return result
5. 层次化编排
大规模 Agent 系统需要分层管理：
Top Supervisor
/ \
Sub-Supervisor A Sub-Supervisor B
/ \ / \
Agent1 Agent2 Agent3 Agent4
class HierarchicalOrchestrator:
def __init__(self):
self.sub_a = Supervisor("research-team")
self.sub_b = Supervisor("writing-team")
self.sub_a.workers = {"searcher": searcher_agent, "analyzer": analyzer_agent}
self.sub_b.workers = {"drafter": drafter_agent, "editor": editor_agent}
self.top = Supervisor("top")
self.top.workers = {
"research-team": self.sub_a.run,
"writing-team": self.sub_b.run
}
async def execute(self, task: str):
return await self.top.run(task)
6. 图式编排（LangGraph）
用状态机定义复杂工作流：
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator
class AgentState(TypedDict):
query: str
docs: list
answer: str
quality_score: float
messages: Annotated[list, operator.add]
def retrieve_node(state: AgentState) -> AgentState:
docs = vector_search(state["query"])
return {"docs": docs, "messages": ["retrieved"]}
def generate_node(state: AgentState) -> AgentState:
answer = llm_generate(state["query"], state["docs"])
return {"answer": answer, "messages": ["generated"]}
def evaluate_node(state: AgentState) -> AgentState:
score = evaluate_answer(state["answer"], state["query"])
return {"quality_score": score, "messages": ["evaluated"]}
def should_retry(state: AgentState) -> str:
if state["quality_score"] < 0.7:
return "retrieve"
return END
graph = StateGraph(AgentState)
graph.add_node("retrieve", retrieve_node)
graph.add_node("generate", generate_node)
graph.add_node("evaluate", evaluate_node)
graph.set_entry_point("retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", "evaluate")
graph.add_conditional_edges("evaluate", should_retry)
app = graph.compile()
result = app.invoke({"query": "什么是 RAG？", "docs": [], "answer": "",
"quality_score": 0, "messages": []})
选型建议
| 场景 | 推荐模式 | 理由 |
|---|---|---|
| 固定流水线 | 串行 | 简单可靠 |
| 独立子任务 | 并行 | 性能最优 |
| 分类分发 | Router | 清晰解耦 |
| 中心调度 | Supervisor | 灵活可控 |
| 大规模系统 | 层次化 | 可扩展 |
| 复杂工作流 | 图式(LangGraph) | 表达力强 |
实战建议
- 从简单开始：先用串行模式跑通，再根据瓶颈优化
- 控制 Agent 数量：超过 5 个 Agent 时协调成本急剧上升，考虑分层
- 超时与降级：每个 Agent 设置超时，超时后用降级策略（返回缓存或默认值）
- 可观测性：编排系统必须有完整 trace，否则出问题完全无法定位
- 成本控制：Supervisor 用小模型（GPT-4o-mini），Worker 按任务复杂度选模型—
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞，认知差异创造无限可能。