多Agent系统设计 - 从单体到协作架构¶

面试趋势：2028年，Multi-Agent系统设计已成为字节、阿里、腾讯AI岗位面试的标配题型 前置知识：AI Agent开发实战/ 系列

1. 为什么面试要考Multi-Agent¶

单Agent系统在复杂任务上遇到瓶颈： - 单一LLM的上下文窗口有限，难以同时处理多种子任务 - 工具调用链过长时，单Agent容易"迷路"或累积错误 - 不同子任务需要不同的专业知识/模型/工具

Multi-Agent通过"分工协作"解决这些问题——这与真实的软件工程团队类似。

2. 核心架构模式¶

2.1 顺序管道模式（Pipeline）¶

Text Only

任务 → Agent A(规划) → Agent B(执行) → Agent C(审核) → 结果

示例: 代码生成系统
  用户需求 → 需求分析Agent → 代码生成Agent → 代码审查Agent → 测试Agent → 输出

适用场景：流程明确、步骤固定的任务优点：简单可靠，每步可独立监控缺点：无法处理需要反复迭代的任务

2.2 层级管理模式（Hierarchical）¶

Text Only

             ┌─────────────┐
             │ 管理者Agent  │  ← 负责任务分解、分配、汇总
             │ (Orchestrator)│
             └──────┬──────┘
          ┌─────────┼─────────┐
          ↓         ↓         ↓
     ┌────────┐ ┌────────┐ ┌────────┐
     │专家A   │ │专家B   │ │专家C   │
     │(搜索)  │ │(分析)  │ │(写作)  │
     └────────┘ └────────┘ └────────┘

适用场景：需要多种专业技能协作的复杂任务关键设计：管理者Agent的Prompt设计决定系统上限

2.3 辩论/讨论模式（Debate）¶

Text Only

问题 → Agent A(正方观点) ←──对话──→ Agent B(反方观点)
                    ↓ 多轮辩论
              裁判Agent(综合判断) → 最终结论

适用场景：需要多角度分析的决策任务（代码审查、方案评估、风险分析）优点：通过对抗性讨论减少单一视角偏见

2.4 去中心化协作模式（Decentralized）¶

Text Only

     Agent A ←──────→ Agent B
       ↑ ↘              ↗ ↑
       │   ↘          ↗   │
       │    Agent D ←→    │
       │   ↗          ↘   │
       ↓ ↗              ↘ ↓
     Agent C ←──────→ Agent E

     每个Agent有自己的状态，通过消息传递协作
     没有中心管理者

适用场景：模拟社会行为/市场交易/复杂生态系统

3. 生产级Multi-Agent系统关键设计¶

3.1 通信机制¶

Text Only

方案对比:
┌──────────────┬────────────────┬───────────────┬──────────────┐
│ 方案         │ 实现           │ 优点          │ 缺点         │
├──────────────┼────────────────┼───────────────┼──────────────┤
│ 直接调用     │ 函数调用/API   │ 简单、低延迟   │ 紧耦合       │
├──────────────┼────────────────┼───────────────┼──────────────┤
│ 消息队列     │ Redis/Kafka    │ 解耦、可追溯   │ 延迟增加     │
├──────────────┼────────────────┼───────────────┼──────────────┤
│ 共享状态     │ 黑板模式       │ 灵活、可扩展   │ 状态一致性   │
│              │ (Blackboard)   │              │ 管理复杂      │
├──────────────┼────────────────┼───────────────┼──────────────┤
│ MCP/A2A协议  │ 标准化协议     │ 互操作、生态   │ 协议开销     │
│              │                │ 兼容性好       │              │
└──────────────┴────────────────┴───────────────┴──────────────┘

3.2 状态管理¶

Text Only

Multi-Agent状态管理层次:

1. Agent内部状态
   - 当前任务上下文(对话历史/中间结果)
   - 工具调用记录
   - 存储: 内存 or Redis(持久化)

2. 共享工作区(Workspace)
   - 所有Agent可读写的公共区域
   - 存储: 文件系统/数据库/向量库
   - 示例: 代码文件、文档草稿、检索结果

3. 全局任务状态
   - 任务进度追踪(哪些子任务完成、哪些进行中)
   - 依赖关系图
   - 存储: 状态机(Redis) or 工作流引擎(Temporal)

3.3 错误处理与恢复¶

生产环境中Agent会失败——这是面试的关键追问点：

Text Only

错误类型与处理策略:

1. LLM输出格式错误
   → 重试(最多3次) + 更严格的Prompt + 结构化输出(JSON Schema)

2. 工具调用失败
   → 重试 + 降级方案(换工具/手动处理) + 超时控制

3. Agent死循环
   → 设置最大轮次(max_iterations) + 循环检测(相似输出检测)
   → 超限后向管理者Agent汇报，由管理者决策

4. Agent间通信超时
   → 超时设置 + 异步化 + 死信队列 + 人工兜底

5. 全局任务失败
   → 检查点(Checkpoint)机制: 定期保存状态
   → 从最近检查点恢复，无需从头开始

3.4 可观测性（面试加分项）¶

Text Only

Multi-Agent监控体系:

Trace层(端到端链路追踪):
  │
  ├── 请求ID贯穿所有Agent调用
  ├── 每个Agent的输入/输出/耗时/Token消耗
  ├── 工具调用详情和返回值
  └── 使用OpenTelemetry + Langfuse/LangSmith

Metrics层(聚合指标):
  │
  ├── 任务成功率(按类型/复杂度分)
  ├── 平均完成时间
  ├── 每个Agent的平均调用次数
  ├── Token消耗和成本
  └── 使用Prometheus + Grafana

Log层(调试信息):
  │
  ├── Agent间消息传递记录
  ├── 决策推理过程(Thought/Action/Observation)
  └── 结构化日志(JSON) + ELK

4. 面试实战题：设计一个AI自动化数据分析系统¶

面试官："设计一个面向业务分析师的AI系统，用户上传数据集后，系统自动生成分析报告（包含数据清洗、统计分析、可视化、洞察总结）。"

需求澄清¶

数据规模：CSV/Excel，单文件不超过100MB，百列千万行以内
输出：包含图表的PDF/HTML报告
用户：非技术人员，用自然语言描述分析需求
安全：数据不出企业内网

架构设计：层级Multi-Agent¶

Text Only

用户: "分析这份销售数据，找出增长最快的品类和地区"

管理者Agent(Orchestrator)
  │
  ├─ 1. 数据理解Agent
  │     输入: 原始数据集 + 用户描述
  │     工具: pandas profiling, schema推断
  │     输出: 数据摘要(列含义、类型、分布、质量问题)
  │
  ├─ 2. 数据清洗Agent
  │     输入: 原始数据 + 数据摘要 + 清洗建议
  │     工具: pandas(代码沙箱执行)
  │     输出: 清洗后数据 + 清洗日志
  │
  ├─ 3. 分析规划Agent
  │     输入: 用户需求 + 数据摘要
  │     输出: 分析计划(要做哪些统计/对比/趋势分析)
  │
  ├─ 4. 统计分析Agent
  │     输入: 清洗后数据 + 分析计划
  │     工具: pandas, scipy, statsmodels(代码沙箱)
  │     输出: 统计结果(表格/数值)
  │
  ├─ 5. 可视化Agent
  │     输入: 统计结果 + 可视化建议
  │     工具: matplotlib, plotly(代码沙箱)
  │     输出: 图表文件(.png/.html)
  │
  └─ 6. 报告生成Agent
        输入: 所有中间结果 + 用户原始需求
        输出: 结构化报告(Markdown → PDF)

审查Agent(独立):
  审查最终报告的数据准确性和结论合理性

关键设计决策¶

为什么用Multi-Agent而不是单Agent?

维度	单Agent方案	Multi-Agent方案
Prompt长度	一个Prompt塞所有指令，容易遗漏	每个Agent职责清晰，Prompt简短精准
调试	出错后难以定位是哪个环节	每个Agent独立可测试/可替换
可靠性	某步失败全部重来	检查点机制，从失败点恢复
并行化	串行执行	无依赖的Agent可并行(统计+可视化)
成本	每步都用最大模型	简单Agent用小模型(7B)，关键Agent用大模型

代码执行安全：

Text Only

数据分析Agent需要执行代码 → 安全隔离至关重要

方案: Docker沙箱
  - 每次分析创建独立Docker容器
  - 只挂载数据文件(只读)和输出目录(可写)
  - 无网络访问
  - CPU/内存/执行时间限制
  - 执行完成后销毁容器

成本控制：

Text Only

Agent模型选择策略:
  数据理解Agent   → 7B本地模型(理解数据结构，不需要强推理)
  清洗Agent       → 7B本地模型(生成pandas代码)
  分析规划Agent   → 70B/API(需要理解业务需求，制定策略)
  统计分析Agent   → 7B本地模型(生成统计代码)
  可视化Agent     → 7B本地模型(生成绑定代码)
  报告生成Agent   → 70B/API(需要生成流畅的分析叙述)
  审查Agent       → 70B/API(需要推理能力验证结论)

  预估成本: 单次分析约 ¥0.5-2 (混合模型) vs ¥5-15 (全部用API大模型)

5. 面试高频追问¶

Q: 如何防止Agent间的"传话损失"？ - 使用结构化的消息格式(JSON Schema)而非自然语言传递 - 关键数据直接通过共享存储传递(文件/数据库)，而非嵌入消息 - 管理者Agent做中间结果验证

Q: 如何处理Agent之间的分歧？ - 层级模式：管理者Agent最终裁决 - 辩论模式：多轮讨论后投票或由裁判Agent决策 - 置信度机制：每个Agent输出置信度，管理者加权决策

Q: Multi-Agent系统如何做评估？

Text Only

三层评估:
  1. 单Agent评估: 每个Agent单独测试(输入→输出质量)
  2. 集成评估: 端到端任务完成率、质量、耗时
  3. 对比评估: Multi-Agent vs 单Agent vs 人工 的成本和质量对比

Q: 没有框架的话，你会怎么实现一个最小Multi-Agent系统？

Python

# 最简Multi-Agent实现骨架
from dataclasses import dataclass

@dataclass
class Message:
    sender: str
    content: str
    metadata: dict

class Agent:
    def __init__(self, name, system_prompt, model, tools=None):
        self.name = name
        self.system_prompt = system_prompt
        self.model = model
        self.tools = tools or []

    def run(self, messages: list[Message]) -> Message:
        # 调用LLM + 工具执行
        response = self.model.generate(
            system=self.system_prompt,
            messages=[m.content for m in messages],
            tools=self.tools
        )
        return Message(sender=self.name, content=response, metadata={})

class Orchestrator:
    def __init__(self, agents: dict[str, Agent], workflow: list):
        self.agents = agents
        self.workflow = workflow  # 执行顺序/依赖图

    def save_checkpoint(self, agent_name: str, result: Message):
        """保存检查点（用于故障恢复）"""
        print(f"[Checkpoint] Agent={agent_name}, Result={result.content[:50]}...")

    def run(self, task: str) -> str:
        shared_context = [Message(sender="user", content=task, metadata={})]
        for step in self.workflow:
            agent = self.agents[step["agent"]]
            result = agent.run(shared_context)
            shared_context.append(result)
            if step.get("checkpoint"):
                self.save_checkpoint(step["agent"], result)
        return shared_context[-1].content

最后更新：2026年2月