AI Agent Workflow Design Patterns: 从 Chain 到 Orchestrator 的 8 大生产级模式 🎯🔄

发布日期：2026-07-08 · 小玉米技术博客

🚀 引言

2026 年，AI Agent 已从"单次问答"进化到"多步编排"。无论是 AutoGPT 的 Plan-then-Execute、LangGraph 的状态机，还是 LlamaIndex 的 Workflow 引擎，核心都围绕 工作流设计模式 (Workflow Design Patterns) 展开。本文系统梳理 8 大生产级 Agent 工作流模式，每种模式配套完整的 Python 实现代码、适用场景、性能基准与 Anti-Pattern 警示。

🏗️ 8 大模式总览

1. 🔗 Chain Pattern (链式)

最基础的模式：前一步的输出是下一步的输入。适合确定性流水线。

from dataclasses import dataclass, field
from typing import Any, Callable, Awaitable
import asyncio

@dataclass
class WorkflowStep:
    name: str
    handler: Callable[[Any], Awaitable[Any]]
    retry_count: int = 3

class ChainWorkflow:
    """串行链式工作流"""
    def __init__(self, steps: list[WorkflowStep]):
        self.steps = steps

    async def execute(self, initial_input: Any) -> dict:
        context = {"input": initial_input, "outputs": [], "errors": []}
        current_input = initial_input
        for step in self.steps:
            for attempt in range(step.retry_count):
                try:
                    output = await step.handler(current_input)
                    context["outputs"].append({step.name: output})
                    current_input = output
                    break
                except Exception as e:
                    context["errors"].append(f"{step.name} attempt {attempt+1}: {e}")
                    if attempt == step.retry_count - 1:
                        raise
                    await asyncio.sleep(2 ** attempt)
        return context

性能基准 (10 轮测试):

⚠️ Anti-Pattern: 超过 8 步的链会累积延迟和错误。应拆分为子链或用 Orchestrator 替代。

2. 🌊 Parallel Pattern (扇出)

同时执行多个独立任务，然后聚合结果。适合信息检索和多源分析。

@dataclass
class ParallelWorkflow:
    tasks: dict[str, Callable[[], Awaitable[Any]]]
    max_concurrency: int = 5
    timeout: float = 30.0

    async def execute(self) -> dict:
        semaphore = asyncio.Semaphore(self.max_concurrency)
        async def run_task(name: str) -> tuple[str, Any | None]:
            async with semaphore:
                try:
                    result = await asyncio.wait_for(self.tasks[name](), timeout=self.timeout)
                    return name, result
                except asyncio.TimeoutError:
                    return name, {"error": f"Timeout after {self.timeout}s"}
                except Exception as e:
                    return name, {"error": str(e)}
        results = await asyncio.gather(*[run_task(n) for n in self.tasks])
        return dict(results)

性能基准 (并行度 vs 延迟):

3. 🚦 Router Pattern (路由)

根据输入内容或意图，将任务分发到不同的处理路径。适合多意图 Agent 系统。

@dataclass
class RouterRule:
    name: str
    condition: Callable[[Any], bool]
    handler: Callable[[Any], Awaitable[Any]]

class RouterWorkflow:
    def __init__(self, rules: list[RouterRule], default_handler=None):
        self.rules = rules
        self.default_handler = default_handler

    async def route(self, input_data: Any) -> dict:
        for rule in self.rules:
            if rule.condition(input_data):
                result = await rule.handler(input_data)
                return {"route": rule.name, "result": result}
        if self.default_handler:
            result = await self.default_handler(input_data)
            return {"route": "default", "result": result}
        return {"route": "unmatched", "error": "No matching route"}

4. 🎭 Orchestrator Pattern (编排器)

最强大的模式：一个中央 Orchestrator 动态分解任务、分发给子 Agent 执行，并聚合结果。适用于复杂的多步骤业务流程。

@dataclass
class Task:
    id: str
    description: str
    dependencies: list[str] = field(default_factory=list)
    result: Any = None
    status: str = "pending"

@dataclass
class SubAgent:
    name: str
    capability: str
    execute_fn: Callable[[Task], Awaitable[Any]]

class OrchestratorWorkflow:
    def __init__(self, agents: list[SubAgent]):
        self.agents = agents
        self.task_graph: dict[str, Task] = {}
        self.execution_log: list = []

    def decompose(self, goal: str) -> list[Task]:
        """分解目标为 DAG 任务图"""
        task_defs = [
            Task(id="plan", description="制定执行计划"),
            Task(id="research", description="收集信息", dependencies=["plan"]),
            Task(id="analyze", description="分析数据", dependencies=["research"]),
            Task(id="generate", description="生成输出", dependencies=["analyze"]),
            Task(id="review", description="质量检查", dependencies=["generate"]),
        ]
        for t in task_defs:
            self.task_graph[t.id] = t
        return task_defs

    def _get_ready_tasks(self) -> list[Task]:
        """拓扑排序：获取可执行的 ready 任务"""
        ready = []
        for task in self.task_graph.values():
            if task.status != "pending":
                continue
            deps_met = all(
                self.task_graph[d].status == "completed"
                for d in task.dependencies
            )
            if deps_met:
                ready.append(task)
        return ready

    async def execute(self, goal: str) -> dict:
        self.decompose(goal)
        completed_count = 0
        total = len(self.task_graph)
        while completed_count < total:
            ready = self._get_ready_tasks()
            if not ready:
                waiting = [t.id for t in self.task_graph.values() if t.status == "pending"]
                raise RuntimeError(f"Deadlock. Waiting: {waiting}")
            async def run_task(task: Task):
                task.status = "running"
                agent = self.assign_agent(task)
                task.result = await agent.execute_fn(task)
                task.status = "completed"
                return task
            await asyncio.gather(*[run_task(t) for t in ready])
            completed_count = sum(1 for t in self.task_graph.values() if t.status == "completed")
        return {
            "goal": goal,
            "results": {t.id: t.result for t in self.task_graph.values()},
            "total_tasks": total,
            "completed": completed_count,
        }

性能基准 (5 任务 DAG):

5. 🔄 Evaluator-Optimizer Pattern (评估-优化)

生成器 + 评估器组成的迭代循环。生成初版 → 评估质量 → 反馈优化。适合内容生成、代码翻译、文档撰写。

class EvaluatorOptimizer:
    def __init__(self, generator, evaluator,
                 max_iterations: int = 5,
                 quality_threshold: float = 0.85):
        self.generator = generator
        self.evaluator = evaluator
        self.max_iterations = max_iterations
        self.quality_threshold = quality_threshold

    async def generate(self, prompt: str) -> dict:
        history = []
        current_prompt = prompt
        for i in range(self.max_iterations):
            output = await self.generator(current_prompt)
            eval_result = await self.evaluator(prompt, output)
            history.append({"iteration": i+1, "score": eval_result.score, "feedback": eval_result.feedback})
            if eval_result.passed:
                return {"final_output": output, "iterations": i+1, "history": history, "status": "passed"}
            current_prompt = f"""{prompt}\nPrev score: {eval_result.score:.2f}\nFeedback: {eval_result.feedback}\nImprove."""
        return {"final_output": history[-1]["output"], "iterations": self.max_iterations, "history": history, "status": "max_reached"}

⚠️ Anti-Pattern: 超过 5 轮迭代后边际效益递减。应设置硬上限并记录失败案例供离线优化。

6. 👑 Supervisor Pattern (监管者)

一个"元 Agent"（Supervisor）负责委派任务给多个子 Agent 并监督执行。适合高度不确定、需要动态决策的场景。

class SupervisorWorkflow:
    def __init__(self, agents: dict[str, SubAgent], supervisor_llm: Callable):
        self.agents = agents
        self.supervisor_llm = supervisor_llm
        self.context = {"history": [], "artifacts": {}}

    async def run(self, objective: str) -> dict:
        for round_num in range(10):
            decision = await self.supervisor_llm(objective, self.context)
            if decision.action == "complete":
                return {"objective": objective, "rounds": round_num+1,
                        "final_output": self.context.get("final_output")}
            elif decision.action == "delegate":
                agent = self.agents.get(decision.target_agent)
                result = await agent.execute_fn(Task(id=f"r{round_num}", description=decision.task_description))
                self.context["artifacts"][f"r{round_num}"] = result
            elif decision.action == "escalate":
                return {"objective": objective, "status": "escalated", "reasoning": decision.reasoning}

7. 🤝 Human-in-the-Loop Pattern (人机协同)

在关键节点插入人工审批或输入。适合金融交易、代码部署、敏感数据处理。

@dataclass
class ApprovalRequest:
    task_id: str
    action_description: str
    risk_level: str  # low | medium | high
    approved: bool | None = None
    human_feedback: str | None = None

class HumanInTheLoopWorkflow:
    def __init__(self, approval_callback):
        self.approval_callback = approval_callback

    async def check_and_approve(self, request: ApprovalRequest) -> ApprovalRequest:
        if request.risk_level in ("high", "medium"):
            return await self.approval_callback(request)
        request.approved = True
        return request

    async def execute_with_safeguards(self, task: Task) -> dict:
        risk_level = "high" if "deploy" in task.description.lower() else "medium"
        approval = await self.check_and_approve(ApprovalRequest(
            task_id=task.id, action_description=task.description,
            risk_level=risk_level, suggested_action=f"Execute: {task.description}",
        ))
        if not approval.approved:
            return {"status": "rejected", "human_feedback": approval.human_feedback}
        return {"status": "approved", "result": await task.execute()}

8. 🔄 Agentic Loop Pattern (自主循环)

观察 (Observe) → 思考 (Think) → 行动 (Act) 的连续循环。Agent 在执行过程中不断评估进度并调整策略。

class AgenticLoopWorkflow:
    def __init__(self, llm: Callable, tools: dict[str, Callable]):
        self.llm = llm
        self.tools = tools

    async def run(self, objective: str, max_steps: int = 25) -> dict:
        state = {"objective": objective, "steps": [], "completed": False}
        for step_num in range(max_steps):
            # Think
            context = f"Objective: {objective}\nSteps: {len(state['steps'])}\n"
            plan = await self.llm(context + "What's next?")
            # Act
            result = await self._execute_action(plan)
            state["steps"].append({"step": step_num+1, "action": plan, "result": result})
            # Observe
            assessment = await self.llm(f"Result: {result[:200]}\nObjective met? YES/NO")
            if "YES" in assessment.upper():
                return {"objective": objective, "steps": step_num+1, "completed": True, "history": state["steps"]}
        return {"objective": objective, "steps": max_steps, "completed": False, "history": state["steps"]}

📊 模式选择决策树

输入任务
  ├─ 确定性步骤? ──→ Chain Pattern
  ├─ 多源并行? ───→ Parallel Pattern
  ├─ 条件分支? ───→ Router Pattern
  ├─ 复杂多步骤? ──→ Orchestrator Pattern
  │   └─ 需迭代优化? ──→ Evaluator-Optimizer
  ├─ 高度不确定性? ─→ Supervisor Pattern
  ├─ 关键决策需人? ─→ Human-in-the-Loop
  └─ 自主探索执行? ─→ Agentic Loop Pattern

🔮 组合模式与未来趋势

混合模式架构 (Hybrid Pattern Architecture)

生产级 Agent 系统通常组合多种模式：

class HybridAgentSystem:
    async def process(self, request: Request) -> Response:
        router = RouterWorkflow([...])
        route = await router.route(request)
        if route["route"] == "complex":
            orchestrator = OrchestratorWorkflow([...])
            result = await orchestrator.execute(request.goal)
        else:
            loop = AgenticLoopWorkflow(llm, tools)
            result = await loop.run(request.objective)
        hitl = HumanInTheLoopWorkflow(approval_callback)
        return await hitl.execute_with_safeguards(result)

2026-2027 趋势

• 自适应模式切换: Agent 运行时动态在 Chain/Orchestrator/Supervisor 之间切换
• Workflow-as-Code: 用 Python DSL 声明工作流 → 自动编译为 DAG 执行图
• 可观测 Workflow: 每个模式都嵌入 OpenTelemetry 追踪
• 多模态 Workflow: 同时编排文本、图片、音频、代码的执行流水线
• 遗传 Workflow 优化: 使用进化算法自动搜索最优模式组合

🎯 总结

选择建议: 从最简模式开始（Chain/Router），逐渐引入 Orchestrator。只有高度不确定的任务才使用 Supervisor 或纯 Agentic Loop。

小玉米技术博客 · 2026-07-08 · 本文配套代码见 GitHub