Agent 模式全景

复杂度 →
单Agent ──→ Router ──→ Fan-out/Fan-in ──→ Hierarchical ──→ Multi-Agent
  │            │              │                  │                │
  │  简单任务   │  按需分发     │  并行处理         │  管理者+执行者   │  自主协作
  │            │              │                  │                │
  └────────────┴──────────────┴──────────────────┴────────────────┘
模式适用场景复杂度延迟错误恢复
单 Agent简单问答、单一任务低低简单重试
Router多领域问答、任务分类中低+路由单分支重试
Fan-out/Fan-in并行研究、批量处理中高(并行)部分失败容忍
Hierarchical复杂项目、多步骤流程高高子任务级恢复
Multi-Agent开放式协作、辩论极高最高最复杂

单 Agent 架构

最基础的形态：一个 Agent + 一组工具。

from dataclasses import dataclass
from typing import Callable

@dataclass
class Tool:
    name: str
    description: str
    execute: Callable

class SingleAgent:
    def __init__(self, llm, tools: list[Tool], system_prompt: str):
        self.llm = llm
        self.tools = {t.name: t for t in tools}
        self.system_prompt = system_prompt

    async def run(self, task: str, max_steps=10):
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": task},
        ]

        for step in range(max_steps):
            response = await self.llm.complete(
                messages, tools=list(self.tools.values())
            )

            if response.finish_reason == "stop":
                return response.content

            # 执行工具调用
            for tool_call in response.tool_calls:
                tool = self.tools[tool_call.name]
                result = await tool.execute(**tool_call.arguments)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": str(result),
                })

        return "Max steps reached without completion"

适用场景：客服 Bot、代码助手、简单 RAG 问答。

局限：工具数量受限（>20 个工具时模型选择准确率下降）、单线程执行无法并行。

Router 模式

一个路由器 Agent 根据输入类型将任务分发给专门化的子 Agent。

@dataclass
class SpecialistAgent:
    name: str
    description: str       # 路由器用此描述做决策
    system_prompt: str
    tools: list[Tool]

class RouterAgent:
    def __init__(self, llm, specialists: list[SpecialistAgent]):
        self.llm = llm
        self.specialists = {s.name: s for s in specialists}

    async def run(self, task: str):
        # Step 1: 路由决策
        specialist = await self._route(task)

        # Step 2: 委托给专门 Agent
        agent = SingleAgent(
            self.llm,
            tools=specialist.tools,
            system_prompt=specialist.system_prompt,
        )
        return await agent.run(task)

    async def _route(self, task: str) -> SpecialistAgent:
        # 用小模型做快速路由
        routing_prompt = f"""根据用户输入选择最合适的专家：
{self._format_specialists()}
用户输入：{task}
只输出专家名称。"""

        name = await self.llm.complete(
            routing_prompt, model="gpt-4o-mini"
        )
        return self.specialists.get(
            name.strip(), self.specialists["general"]
        )

    def _format_specialists(self):
        return "\n".join(
            f"- {s.name}: {s.description}"
            for s in self.specialists.values()
        )

示例：

specialists = [
    SpecialistAgent(
        name="code_expert",
        description="编程、代码审查、技术架构问题",
        system_prompt="你是资深工程师...",
        tools=[code_search, run_code, git_ops],
    ),
    SpecialistAgent(
        name="data_analyst",
        description="数据分析、SQL查询、报表生成",
        system_prompt="你是数据分析师...",
        tools=[sql_query, chart_gen, data_export],
    ),
    SpecialistAgent(
        name="general",
        description="通用问答、闲聊",
        system_prompt="你是助手...",
        tools=[web_search],
    ),
]

Fan-out / Fan-in

将任务拆分成多个子任务并行执行，最后汇总结果。

import asyncio

class FanOutFanIn:
    def __init__(self, llm, worker_agent: SingleAgent):
        self.llm = llm
        self.worker = worker_agent

    async def run(self, task: str):
        # Step 1: 拆分任务
        subtasks = await self._decompose(task)

        # Step 2: 并行执行
        results = await asyncio.gather(*[
            self.worker.run(subtask) for subtask in subtasks
        ], return_exceptions=True)

        # Step 3: 处理失败
        successful = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                logger.warning(f"Subtask {i} failed: {result}")
            else:
                successful.append(result)

        # Step 4: 汇总
        synthesis = await self._synthesize(task, successful)
        return synthesis

    async def _decompose(self, task: str) -> list[str]:
        prompt = f"""将以下任务拆分为 3-5 个可并行的子任务。
任务：{task}
输出 JSON 数组：["子任务1", "子任务2", ...]"""
        result = await self.llm.complete(prompt)
        return json.loads(result)

    async def _synthesize(self, original_task, results):
        prompt = f"""原任务：{original_task}
以下是各子任务的执行结果：
{self._format_results(results)}
请综合所有结果，给出最终回答。"""
        return await self.llm.complete(prompt)

适用场景：研究报告生成（多个方向并行调研）、批量文档处理、多源数据聚合。

Hierarchical 模式

管理 Agent 负责任务规划和分配，执行 Agent 负责具体操作。

@dataclass
class TaskNode:
    id: str
    description: str
    status: str = "pending"  # pending/running/done/failed
    result: str = None
    children: list["TaskNode"] = None

class ManagerAgent:
    def __init__(self, llm, workers: dict[str, SingleAgent]):
        self.llm = llm
        self.workers = workers

    async def run(self, objective: str):
        # Step 1: 创建执行计划
        plan = await self._create_plan(objective)

        # Step 2: 按依赖顺序执行
        await self._execute_plan(plan)

        # Step 3: 汇总报告
        return await self._report(plan)

    async def _create_plan(self, objective) -> TaskNode:
        prompt = f"""目标：{objective}
可用执行者：{list(self.workers.keys())}
创建执行计划，输出 JSON：
{{"id": "root", "description": "...", "children": [...]}}"""
        result = await self.llm.complete(prompt)
        return self._parse_plan(json.loads(result))

    async def _execute_plan(self, node: TaskNode):
        if node.children:
            # 有子任务：递归执行
            for child in node.children:
                await self._execute_plan(child)
            # 子任务完成后综合
            child_results = {c.id: c.result for c in node.children}
            node.result = await self._synthesize(node.description, child_results)
        else:
            # 叶子任务：分配给 worker 执行
            node.status = "running"
            try:
                worker = await self._select_worker(node.description)
                node.result = await worker.run(node.description)
                node.status = "done"
            except Exception as e:
                node.status = "failed"
                node.result = str(e)

    async def _select_worker(self, task_desc):
        """根据任务描述选择最合适的 worker"""
        # 简单实现：用 LLM 选择
        return self.workers["general"]

Multi-Agent 协作

多个 Agent 自主协作，可以互相通信、辩论、审核。

@dataclass
class Message:
    sender: str
    receiver: str
    content: str
    round: int

class MultiAgentSystem:
    def __init__(self, agents: dict[str, SingleAgent], max_rounds=5):
        self.agents = agents
        self.max_rounds = max_rounds
        self.message_log = []

    async def discuss(self, topic: str) -> str:
        # 初始发言
        messages = [Message(
            sender="moderator",
            receiver="all",
            content=topic,
            round=0,
        )]

        for round_num in range(1, self.max_rounds + 1):
            round_messages = []

            for name, agent in self.agents.items():
                # 每个 Agent 看到所有历史消息后发言
                context = self._format_history(messages, exclude=name)
                response = await agent.run(
                    f"讨论记录：\n{context}\n\n请发表你的观点。"
                )
                round_messages.append(Message(
                    sender=name, receiver="all",
                    content=response, round=round_num,
                ))

            messages.extend(round_messages)

            # 检查是否达成共识
            if self._check_consensus(round_messages):
                break

        # Moderator 总结
        summary = await self._summarize(messages)
        return summary

    def _check_consensus(self, messages):
        """用 LLM 判断是否达成共识"""
        # 简化实现：检查是否有明显分歧关键词
        for msg in messages:
            if "反对" in msg.content or "不同意" in msg.content:
                return False
        return True

辩论模式示例

# 一个辩论系统：生成者 vs 审核者
debaters = {
    "proposer": SingleAgent(
        llm, tools=[web_search],
        system_prompt="你是方案提出者，提出解决方案并论证其优势"
    ),
    "critic": SingleAgent(
        llm, tools=[fact_check],
        system_prompt="你是审核者，找出方案中的问题和风险"
    ),
    "judge": SingleAgent(
        llm, tools=[],
        system_prompt="你是裁判，综合双方观点给出最终结论"
    ),
}

system = MultiAgentSystem(debaters, max_rounds=3)
result = await system.discuss("是否应该用微服务架构重写单体应用？")

状态管理

from enum import Enum
from dataclasses import dataclass, field
from typing import Any

class AgentState(Enum):
    IDLE = "idle"
    THINKING = "thinking"
    ACTING = "acting"
    WAITING = "waiting"
    DONE = "done"
    ERROR = "error"

@dataclass
class AgentContext:
    task_id: str
    state: AgentState = AgentState.IDLE
    history: list[dict] = field(default_factory=list)
    working_memory: dict[str, Any] = field(default_factory=dict)
    checkpoints: list[dict] = field(default_factory=list)

    def checkpoint(self):
        """保存当前状态用于回滚"""
        import copy
        self.checkpoints.append({
            "state": self.state,
            "history": copy.deepcopy(self.history),
            "memory": copy.deepcopy(self.working_memory),
        })

    def rollback(self):
        """回滚到上一个检查点"""
        if self.checkpoints:
            cp = self.checkpoints.pop()
            self.state = cp["state"]
            self.history = cp["history"]
            self.working_memory = cp["memory"]
            return True
        return False

错误恢复

class ErrorRecovery:
    def __init__(self, agent, max_retries=3, fallback_strategy=None):
        self.agent = agent
        self.max_retries = max_retries
        self.fallback = fallback_strategy

    async def run_with_recovery(self, task: str):
        for attempt in range(self.max_retries):
            try:
                return await self.agent.run(task)
            except ToolExecutionError as e:
                # 工具失败：换工具或跳过
                logger.warning(f"Tool error on attempt {attempt}: {e}")
                self.agent.tools.pop(e.tool_name, None)
            except LLMError as e:
                # LLM 失败：换模型重试
                logger.warning(f"LLM error, switching model: {e}")
                self.agent.llm = self._get_backup_llm()
            except TimeoutError:
                # 超时：减少 context 重试
                self.agent.context.history = \
                    self.agent.context.history[-5:]

        # 所有重试失败，执行降级策略
        if self.fallback:
            return await self.fallback(task)
        raise AgentFailedError(f"All {self.max_retries} attempts failed")

模式选择决策树

任务需要几个步骤？
├─ 1-3 步 → 单 Agent
└─ 4+ 步
    ├─ 步骤可并行？
    │   ├─ 是 → Fan-out/Fan-in
    │   └─ 否
    │       ├─ 有明确的层级关系？ → Hierarchical
    │       └─ 需要多角度思考？ → Multi-Agent
    └─ 需要不同领域专家？
        ├─ 是 → Router 模式
        └─ 否 → 单 Agent + 更多工具

总结

Agent 模式的选择取决于任务复杂度：简单任务用单 Agent 足矣，不要过度设计；Router 模式适合多领域分发；Fan-out/Fan-in 适合并行处理；Hierarchical 适合复杂多步骤项目；Multi-Agent 适合需要辩论/审核的场景。生产环境中最关键的是状态管理（可回滚）和错误恢复（可降级）。从单 Agent 开始，按需升级——不要一开始就搞多 Agent 系统，复杂度会吃掉你。

加入讨论

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

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