Agent编排：从简单到复杂的演进之路

2026年，Agent编排模式已经从早期的"提示词+循环"发展为成熟的架构模式体系。本文将系统梳理从最简单到最复杂的编排模式，帮助你在实际项目中做出正确的架构选择。

模式全景

复杂度 ──────────────────────────────────────────→

串行 ─→ 并行 ─→ 路由 ─→ 循环 ─→ 递归 ─→ 图式 ─→ 自治
 │       │       │       │       │       │       │
 │       │       │       │       │       │       └─ Agent自主决策
 │       │       │       │       │       └─ DAG + 条件边 + 子图
 │       │       │       │       └─ 分治递归
 │       │       │       └─ ReAct循环 / 反思循环
 │       │       └─ if-else路由 / 内容分发
 │       └─ 扇出-扇入 / Map-Reduce
 └─ Pipeline直线流程

模式1：串行编排（Pipeline）

最基础的编排模式，Agent按固定顺序执行。

适用场景

  • 步骤固定的E2E流程
  • 每步输出是下步输入
  • 不需要条件分支

代码示例

from typing import Any

class Pipeline:
    def __init__(self, steps: list):
        self.steps = steps
    
    async def run(self, input_data: Any) -> Any:
        data = input_data
        for i, step in enumerate(self.steps):
            data = await step(data)
            print(f"Step {i}: {step.__name__} completed")
        return data

# 定义步骤
async def extract(text): return {"raw_text": text, "chunks": split(text)}
async def embed(data): return {**data, "embeddings": [embed(c) for c in data["chunks"]]}
async def index(data): return {**data, "indexed": True, "count": len(data["embeddings"])}
async def respond(data): return f"已索引 {data['count']} 个文本块"

pipeline = Pipeline([extract, embed, index, respond])
result = await pipeline.run("一段长文本...")

优缺点

优点缺点
实现简单无法并行
易于调试任何步骤失败整个失败
行为可预测不支持条件分支

模式2：并行编排（Fan-out/Fan-in）

多个Agent同时执行不同子任务，结果聚合后输出。

适用场景

  • 子任务相互独立
  • 需要降低延迟
  • 多视角分析同一问题

代码示例

import asyncio

class FanOutFanIn:
    def __init__(self, workers: list, aggregator):
        self.workers = workers
        self.aggregator = aggregator
    
    async def run(self, input_data: Any) -> Any:
        # 扇出：并行执行所有worker
        results = await asyncio.gather(
            *[worker(input_data) for worker in self.workers]
        )
        # 扇入：聚合结果
        return await self.aggregator(results)

# 多视角分析
async def technical_analysis(data): return {"perspective": "技术", "score": 0.75}
async def fundamental_analysis(data): return {"perspective": "基本面", "score": 0.82}
async def sentiment_analysis(data): return {"perspective": "情绪", "score": 0.61}
async def macro_analysis(data): return {"perspective": "宏观", "score": 0.70}

async def aggregate(results):
    avg_score = sum(r["score"] for r in results) / len(results)
    return {
        "analyses": results,
        "consensus_score": round(avg_score, 2),
        "recommendation": "买入" if avg_score > 0.75 else "观望"
    }

analyzer = FanOutFanIn(
    workers=[technical_analysis, fundamental_analysis, sentiment_analysis, macro_analysis],
    aggregator=aggregate
)

模式3：路由编排（Router）

根据输入内容，动态选择执行路径。

适用场景

  • 客服系统（不同问题类型路由到不同处理Agent）
  • 多语言处理
  • 多领域知识问答

代码示例

from enum import Enum

class Route(Enum):
    TECHNICAL = "technical_agent"
    BUSINESS = "business_agent"
    LEGAL = "legal_agent"
    GENERAL = "general_agent"

class RouterAgent:
    def __init__(self, router_llm, agents: dict):
        self.router_llm = router_llm
        self.agents = agents
    
    async def route(self, query: str) -> Route:
        prompt = f"""判断以下问题属于哪个领域，只返回类别名：
        - TECHNICAL: 技术问题（编程、架构、部署）
        - BUSINESS: 业务问题（市场、财务、运营）  
        - LEGAL: 法律问题（合同、合规、知识产权）
        - GENERAL: 通用问题
        
        问题: {query}"""
        response = await self.router_llm.complete(prompt)
        return Route(response.text.strip().upper())
    
    async def run(self, query: str):
        route = await self.route(query)
        agent = self.agents[route.value]
        return await agent(query)

路由策略对比

策略准确率延迟成本
LLM路由92%+200ms中
规则路由78%0ms低
嵌入路由85%+20ms低
混合路由96%+150ms中

模式4：循环编排（ReAct / Reflection）

Agent在"思考-行动-观察"的循环中迭代，直到完成任务或达到终止条件。

适用场景

  • 需要多步推理的复杂任务
  • 需要自我纠错的场景
  • 开放式探索任务

代码示例

class ReActLoop:
    def __init__(self, llm, tools: dict, max_iterations=10):
        self.llm = llm
        self.tools = tools
        self.max_iterations = max_iterations
    
    async def run(self, task: str) -> str:
        messages = [{"role": "user", "content": task}]
        
        for i in range(self.max_iterations):
            # Thought: LLM思考下一步
            thought = await self.llm.complete(
                f"任务: {task}\n历史: {messages}\n"
                f"思考下一步行动，或输入'FINAL: 答案'来结束。"
            )
            
            # 判断是否结束
            if thought.text.startswith("FINAL:"):
                return thought.text[7:]
            
            # Action: 执行工具
            action, args = self.parse_action(thought.text)
            if action in self.tools:
                observation = await self.tools[action](**args)
            else:
                observation = f"工具 {action} 不存在"
            
            # Observation: 记录结果
            messages.append({
                "thought": thought.text,
                "action": action,
                "observation": observation
            })
        
        return "达到最大迭代次数，未完成任务"

反思循环变体

class ReflectionLoop:
    """在ReAct基础上增加自我反思"""
    
    async def run_with_reflection(self, task: str) -> str:
        for iteration in range(self.max_iterations):
            # 执行任务
            result = await self.execute(task)
            
            # 自我评估
            critique = await self.llm.complete(
                f"任务: {task}\n结果: {result}\n"
                f"评估结果质量，指出问题。如果合格回答'PASS'。"
            )
            
            if critique.text.strip() == "PASS":
                return result
            
            # 根据反思改进
            task = f"原任务: {task}\n上次结果: {result}\n改进建议: {critique.text}\n请重新完成。"
        
        return result

模式5：图式编排（Graph）

最强大的编排模式，支持DAG、条件边、循环、子图。

适用场景

  • 复杂企业流程
  • 需要可审计性的场景
  • 多阶段审批流程

代码示例

from langgraph import StateGraph
from typing import TypedDict

class OrderState(TypedDict):
    order_id: str
    items: list
    total: float
    risk_score: float
    approved: bool
    tracking_number: str

# 构建订单处理图
graph = StateGraph(OrderState)

# 添加节点
graph.add_node("validate", validate_order)
graph.add_node("risk_check", check_risk)
graph.add_node("auto_approve", auto_approve_low_risk)
graph.add_node("manual_review", send_to_human)
graph.add_node("process_payment", charge_payment)
graph.add_node("ship", create_shipment)
graph.add_node("notify", send_notification)
graph.add_node("reject", send_rejection)

# 定义边
graph.set_entry_point("validate")
graph.add_edge("validate", "risk_check")

# 条件路由
graph.add_conditional_edges(
    "risk_check",
    lambda s: "auto_approve" if s["risk_score"] < 0.3 else "manual_review"
)

# 人工审核后的路由
graph.add_conditional_edges(
    "manual_review",
    lambda s: "process_payment" if s["approved"] else "reject"
)

graph.add_edge("auto_approve", "process_payment")
graph.add_edge("process_payment", "ship")
graph.add_edge("ship", "notify")
graph.add_edge("reject", "notify")
graph.set_finish_point("notify")

app = graph.compile()

模式6：递归分治编排

将大任务分解为子任务，递归处理。

适用场景

  • 大型文档处理
  • 代码库级别的代码审查
  • 多层级数据分析
class DivideConquer:
    def __init__(self, llm, max_depth=3):
        self.llm = llm
        self.max_depth = max_depth
    
    async def run(self, task: str, depth=0) -> str:
        if depth >= self.max_depth:
            # 基础情况：直接处理
            return await self.llm.complete(task)
        
        # 分解任务
        subtasks = await self.decompose(task)
        
        # 递归处理子任务
        results = await asyncio.gather(
            *[self.run(subtask, depth+1) for subtask in subtasks]
        )
        
        # 合并结果
        return await self.merge(results, task)

选型决策树

你的任务是否固定步骤？
├─ 是 → 步骤间是否有依赖？
│       ├─ 是 → 串行Pipeline
│       └─ 否 → 并行Fan-out/Fan-in
└─ 否 → 是否需要条件分支？
        ├─ 是 → 分支是否复杂？
        │       ├─ 简单(2-3路) → Router模式
        │       └─ 复杂(多条件多路径) → 图式Graph
        └─ 否 → 是否需要自我纠错？
                ├─ 是 → ReAct/Reflection循环
                └─ 否 → 是否需要分解大任务？
                        ├─ 是 → 递归分治
                        └─ 否 → 自治Agent

反模式警示

反模式1：过度编排

问题：简单任务使用图式编排，增加了不必要的复杂性。

# ❌ 不好的做法：简单任务用复杂图
graph = StateGraph(State)
graph.add_node("step1", simple_step_1)
graph.add_node("step2", simple_step_2)
graph.add_edge("step1", "step2")
app = graph.compile()

# ✅ 好的做法：简单串行
result = await step2(await step1(input_data))

反模式2：无限循环

问题：Reflection循环没有设置最大迭代次数或终止条件。

# ❌ 危险做法
while True:
    result = await agent.run(task)
    if await is_good_enough(result):
        break

# ✅ 安全做法
for i in range(max_iterations):
    result = await agent.run(task)
    if await is_good_enough(result):
        break
    task = f"改进上次结果: {result}"

反模式3：状态泄漏

问题：Agent间的状态共享没有隔离，导致意外的副作用。

# ❌ 全局共享状态
shared_state = {}

async def agent_a():
    shared_state["key"] = "value"  # 可能被agent_b意外修改

# ✅ 显式状态传递
async def agent_a(state):
    new_state = {**state, "key": "value"}
    return new_state

结论

Agent编排没有银弹。选择编排模式时，遵循以下原则：

  1. 从简单开始：能串行解决的不用并行，能路由解决的不用图式
  2. 匹配复杂度：编排复杂度应该匹配业务复杂度
  3. 设置护栏：所有循环都需要最大迭代限制和超时
  4. 可观测性优先：选择可以追踪和调试的编排模式
  5. 预留扩展空间：设计时考虑未来可能需要增加的分支和节点

编排模式是Agent系统的骨架——选对了，系统稳固可扩展；选错了，系统脆弱难维护。花时间理解每种模式的适用场景，是架构设计最重要的投资。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。