【LLM面试专题】6.3 RAG与Agent：多Agent协作系统

1 为什么需要多Agent

一个Agent做所有事就像一个人完成整个大项目——容易出错、视野受限。多Agent把复杂任务分解给多个"专家"，每个专注自己的领域，通过协作达成更好结果。

更本质的原因：单个LLM调用是前馈过程（输入进、输出出），没有内在的反思或纠错机制。多Agent通过引入多轮交互、角色分工和外部反馈回路，模拟"写代码→审查→修改"的迭代改进过程，将LLM从"一次性生成器"转变为"协作推理网络"中的节点。

1.1 单Agent vs 多Agent

追问：多Agent一定比单Agent好吗？
不是。对于明确且步骤化的任务（如翻译），差异不大但成本多3-10倍。多Agent的优势主要体现在需要推理和判断的任务上——辩论和审查机制可以显著减少幻觉。

1.2 什么时候该用多Agent

优先用多Agent：

• 任务复杂需要多步处理（软件开发、研究分析）
• 需要多角度验证和高准确性（医学诊断、法律审查）
• 需要不同专业知识的协作
• 需要可追溯的推理过程

优先用单Agent：

• 任务简单且定义明确（摘要、翻译、分类）
• 对延迟有严格要求
• Token成本敏感
• 不需要多轮推理或验证

中间态方案：单Agent+工具使用（顺序调用多种工具模拟多Agent流程，但没有管理开销）。

2 四大架构模式

2.1 编排者-工作者模式（Orchestrator-Worker）

一个Orchestrator分解任务→分配给N个Worker→汇总结果。星型拓扑，Worker之间不直接通信。

Orchestrator: "分解任务" ├── Worker 1 (UX设计): UI设计稿 ├── Worker 2 (前端): 前端代码 ├── Worker 3 (后端): 后端API └── Worker 4 (数据库): 数据库Schema Orchestrator: "汇总输出最终方案"

适用：任务有明确层次结构，可自然分解为独立子任务。
代表：ChatDev、MetaGPT。

追问：Orchestrator上下文溢出怎么办？
分层Orchestrator——顶层只做粗粒度分解，每个子任务再分配子Orchestrator。或用外部存储持久化中间结果。

2.2 辩论模式（Debate）

多个Agent对同一问题各抒己见，通过辩论逼近真理。

Q: "这段代码有bug吗？" Round 1 (独立判断): Agent A: "第5行空指针异常" Agent B: "第5行已判空，A看错了" Agent C: "同意B，但第8行类型转换有隐患" Round 2 (辩论): Agent A: "重新检查确认判空了，同意C" → 共识: "第8行类型转换需显式转换"

辩论策略变体：

1. 独立-然后-辩论：先独立判断再讨论，减少锚定效应
2. 角色固定辩论：预分配"保守派"/“激进派”/“魔鬼代言人”
3. 分层辩论：先小组内辩论，小组代表再高层辩论

适用：需要高准确率的决策（医疗、法律、安全审计）、开放性问题。

2.3 流水线模式（Pipeline）

前一个Agent的输出是后一个Agent的输入，形成固定流程。

Agent A(需求分析) → Agent B(架构设计) → Agent C(编码) → Agent D(测试)

变体：严格流水线（瀑布）、部分反馈流水线（后步可回传前步）、多重流水线（并行模块最后合并）。

追问：流水线中一个Agent失败怎么办？
容错机制：重试（指数退避）、备用Agent、断点续传。可在阶段间插入"验证门"Agent检查输出质量。

2.4 黑板模式（Blackboard）

所有Agent共享一个公共存储，Agent异步读写信息，发现适合自己的任务就开始工作。

适用：任务依赖关系不确定、渐进式知识积累、Agent可随时加入退出。

2.5 四种模式对比

3 通信机制

3.1 消息传递 vs 共享内存

实践中常用混合架构：消息传递做实时协调+共享内存维护长期状态。

3.2 消息格式设计

{"message_id":"msg_20251201_001","from":"agent_planner","to":"agent_coder","type":"task_assignment","task_id":"task_001","content":{"description":"实现用户登录功能","requirements":["OAuth2","JWT","邮箱验证"],"depends_on":["task_000"]},"priority":"high","ttl":300}

消息类型：task_assignment / result / review_request / review_feedback / clarification / status_update / error_report。

4 Agent专业化策略

4.1 三种专业化维度

实践中三种策略混合使用：角色Prompt定义视角+工具集赋予执行能力+模型级联优化成本。

4.2 角色设计最佳实践

# 专业化提示设计prompt_architect=""" 你是一名软件架构师。职责： 1. 根据需求设计系统架构 2. 选择技术栈 3. 确保可扩展性和可维护性 输出格式：架构决策记录（ADR） """

关键发现（ChatDev实验）：角色越具体（“前端Node.js开发者"vs泛泛的"开发者”），输出质量越好。

5 共识机制

5.1 四种共识方式

置信度获取三种方法：

1. 直接法：LLM输出时附带置信度分数（log probability）
2. 间接法：多次采样统计一致性
3. 自洽性：同一LLM多次回答同一问题，通过一致性估计置信度

辩论收敛检测：

• 观点稳定：连续两轮所有Agent观点不再变化
• 熵下降：观点分布的熵低于阈值
• 最大差异：所有Agent间最大差异小于阈值

追问：共识机制本身失败怎么办？
降级策略：自动切换到Human-in-the-Loop，或回退到单个最强Agent的答案。

6 错误传播与缓解

6.1 五类错误

6.2 缓解策略

追问：怎么衡量错误传播程度？
错误传播率=受影响后续步骤数/总步骤数。也可以用敏感性分析：引入已知错误，观察系统响应。

7 群体迷思（Groupthink）

7.1 问题本质

多Agent互相影响→多样性下降→趋向一致但错误的结论。在单模型多Agent系统中更严重——所有Agent底层是同一个LLM，多样性天然不足。

7.2 缓解策略

追问：辩论轮数怎么确定？
自适应策略：连续两轮所有Agent观点不再变化时终止。或固定轮数后强制终止取最佳答案。

8 真实案例

8.1 ChatDev（代码生成）

架构：编排-工作者变体，模拟软件公司组织架构。
角色：CEO→CTO→程序员→审查员→测试员。
通信：结构化消息传递（Chat Chain格式）。
结果：比单Agent完成率高29.4%，bug更少。
关键发现：角色越具体，代码质量越好。

8.2 SWE-Agent + SWE-Bench

多Agent方法（Agentless+多轮调试）比单Agent多解决约40%的问题。
最佳实践：发现问题→生成补丁→验证→迭代的流水线模式。

8.3 Generative Agents（Stanford）

架构：分布式黑板模式，每个Agent有独立记忆流。
关键发现：Agent表现出涌现行为——未明确编程的社交行为（组织派对、分享信息、形成观点）。
启示：多Agent不仅用于任务求解，也可用于社会模拟。

8.4 这些系统在什么时候会失败？

1. 角色定义冲突→Agent"角色混淆"
2. 缺乏人类监督时→生成不符合约束的内容
3. 高度创造性任务→多Agent有时反而不如单Agent（太多观点延迟决策）

9 评估框架

9.1 评估维度

9.2 评估方法

自动评估：

1. 端到端任务评估：在HumanEval/GSM8K/HotpotQA上比较多Agent vs 单Agent
2. 过程评估：检查中间产出质量（子任务分解合理性、通信相关性）
3. 消融实验：移除单个Agent观察性能变化→衡量边际贡献

人工评估：

1. 最终输出评分（李克特量表1-5分）
2. 过程质量评估（交互流畅性、合理性）
3. 人机盲测对比

10 面试高频问答

Q: 多Agent如何避免群体迷思？
角色多样性+独立预思考+Devil’s Advocate+结构化辩论+外部验证。单模型多Agent更严重（共享相同盲点）。

Q: 四种架构模式怎么选？
编排-工作者：层次化任务。辩论：需要验证的决策。流水线：有序多阶段。黑板：依赖不确定。

Q: Agent的评测体系怎么设计？
分层评测：工具调用准确性→推理质量→任务完成度→安全性→端到端体验。

Q: 什么情况下不该用多Agent？
简单任务（成本3-15倍但收益不大）、高实时性（延迟太高）、工具不可靠（频繁重试体验差）。

Q: 怎么设计支持1000+工具的FC系统？
不能全塞prompt（Token爆炸）。方案：离线索引所有工具描述→在线用query检索Top-K→只注入K个工具Schema→LLM从中选择。

本章要点清单

• 多Agent = 多个专家协作 > 一个通才单干，但成本3-15倍
• 四大架构：编排-工作者（星型）、辩论（全连接）、流水线（链式）、黑板（广播）
• 通信两种：消息传递（实时）+共享内存（持久），实践常用混合
• 专业化三维度：角色（Prompt定义）+工具（能力赋予）+模型（成本优化）
• 共识四种：投票/置信度加权/辩论/仲裁，看任务类型选择
• 错误传播是核心挑战：级联/幻觉传播/确认偏误/信息遗漏/任务漂移
• 群体迷思缓解：独立预思考+角色多样性+外部验证
• 评估：端到端（任务完成率）+过程（通信效率）+消融（边际贡献）

附录A：Multi-Agent通信协议设计

A.1 消息总线实现

classMessageBus:"""Agent间消息总线，支持发布-订阅模式"""def__init__(self):self.queues={}self.subscribers={}self.message_log=[]defpublish(self,topic,message):"""发布消息到指定主题"""iftopicinself.subscribers:foragent_idinself.subscribers[topic]:ifagent_idnotinself.queues:self.queues[agent_id]=[]self.queues[agent_id].append(message)self.message_log.append(message)defsubscribe(self,topic,agent_id):"""订阅主题"""iftopicnotinself.subscribers:self.subscribers[topic]=[]self.subscribers[topic].append(agent_id)defconsume(self,agent_id):"""消费消息"""ifagent_idinself.queuesandself.queues[agent_id]:returnself.queues[agent_id].pop(0)returnNone

A.2 通信可靠性保证

工程实践：大多数Agent系统使用"至少一次"+幂等处理。消息去重用唯一message_id，重复消息检查后丢弃。

A.3 消息超时与重试

classReliableMessageSender:def__init__(self,bus,timeout=30,max_retries=3):self.bus=bus self.timeout=timeout self.max_retries=max_retries self.pending={}asyncdefsend_and_wait(self,message):"""发送消息并等待确认"""forattemptinrange(self.max_retries):self.bus.publish(message.to_topic,message)try:ack=awaitasyncio.wait_for(self.bus.wait_for_ack(message.id),timeout=self.timeout)returnackexceptasyncio.TimeoutError:ifattempt<self.max_retries-1:continueraiseTimeoutError(f"消息{message.id}未收到确认")

附录B：Multi-Agent状态管理

B.1 全局状态 vs 局部状态

B.2 状态同步策略

追问：多Agent同时修改同一状态怎么办？
方案一：分布式锁（一次只允许一个Agent修改）。方案二：乐观锁+版本号（修改时检查版本，冲突时重试）。方案三：最后写入胜出+合并策略（适合非关键状态）。

B.3 检查点与恢复

classCheckpointManager:"""多Agent系统的检查点管理"""def__init__(self,storage):self.storage=storage self.checkpoints=[]defsave_checkpoint(self,global_state,agent_states,step):checkpoint={"step":step,"global_state":global_state,"agent_states":agent_states,"timestamp":datetime.now().isoformat()}self.storage.save(f"checkpoint_{step}",checkpoint)self.checkpoints.append(step)defrestore(self,step=None):"""恢复到指定检查点"""ifstepisNone:step=self.checkpoints[-1]ifself.checkpointselseNoneifstepisNone:raiseValueError("无可恢复的检查点")returnself.storage.load(f"checkpoint_{step}")

附录C：Multi-Agent系统设计模式详解

C.1 主从模式的深度设计

┌─────────────┐ │ Orchestrator │ │ (GPT-4级别) │ └──────┬──────┘ │ ┌──────────────┼──────────────┐ ↓ ↓ ↓ ┌──────────────┐┌──────────────┐┌──────────────┐ │ Research Agent││ Coding Agent ││ Review Agent │ │ (搜索+分析) ││ (代码生成) ││ (质量检查) │ │ GPT-3.5 ││ GPT-4 ││ Claude │ └──────────────┘└──────────────┘└──────────────┘

Orchestrator的职责清单：

1. 任务分解：将复杂任务分解为可独立执行的子任务
2. 能力匹配：根据子任务需求选择合适的Worker
3. 依赖排序：确定子任务的执行顺序
4. 结果聚合：合并各Worker的输出，解决冲突
5. 质量控制：检查最终结果的完整性和一致性

C.2 辩论模式的实现细节

classDebateSession:"""多Agent辩论会话管理"""def__init__(self,agents,topic,max_rounds=5):self.agents=agents# 每个Agent有独立角色和promptself.topic=topic self.max_rounds=max_rounds self.rounds=[]defrun_round(self,round_num):responses=[]foragentinself.agents:# 每个Agent看到其他Agent上一轮的回答context=self._build_context(agent,round_num)response=agent.generate(context)responses.append({"agent":agent.name,"response":response})self.rounds.append(responses)# 收敛检测ifself._check_convergence(round_num):returnTrue# 辩论结束returnFalsedef_check_convergence(self,round_num):ifround_num<2:returnFalseprev=self.rounds[-2]curr=self.rounds[-1]# 检查观点是否趋于一致returnall(self._semantic_similarity(p["response"],c["response"])>0.9forp,cinzip(prev,curr))defget_final_answer(self):"""获取最终答案（投票/仲裁）"""last_round=self.rounds[-1]# 置信度加权投票scores=[]forrinlast_round:confidence=r["agent"].self_evaluate_confidence(r["response"])scores.append((r["response"],confidence))returnmax(scores,key=lambdax:x[1])[0]

C.3 流水线模式的错误处理

classPipelineWithGates:"""带验证门的流水线"""def__init__(self,stages):self.stages=stages# [(agent, validator), ...]asyncdefexecute(self,input_data):current=input_datafori,(agent,validator)inenumerate(self.stages):result=awaitagent.process(current)# 验证门：检查输出质量quality=validator.evaluate(result)ifquality<0.7:# 质量不达标# 重试当前阶段（最多3次）forretryinrange(3):feedback=validator.get_feedback(result)result=awaitagent.process(current,feedback=feedback)quality=validator.evaluate(result)ifquality>=0.7:breakifquality<0.7:raisePipelineError(f"阶段{i}质量不达标:{quality}")current=resultreturncurrent

附录D：Multi-Agent安全与权限

D.1 权限隔离

D.2 安全防护层级

安全防护 ├── 输入层：Prompt注入检测、用户身份验证 ├── 决策层：工具调用权限检查、操作风险评估 ├── 执行层：沙箱隔离、资源配额、超时控制 ├── 输出层：敏感信息过滤、合规性检查 └── 审计层：操作日志、异常告警、事后分析

D.3 Agent身份与信任

classAgentIdentity:"""Agent身份管理"""def__init__(self,agent_id,role,permissions):self.agent_id=agent_id self.role=role self.permissions=set(permissions)self.trust_level=1.0# 初始信任度defcan_execute(self,action):returnaction.required_permissioninself.permissionsdefupdate_trust(self,success_rate):"""根据历史成功率动态调整信任度"""self.trust_level=0.8*self.trust_level+0.2*success_rate

附录E：Multi-Agent常见面试追问链

追问链1：架构选择
Q1: 多Agent有哪几种架构模式？→ 编排-工作者/辩论/流水线/黑板
Q2: 编排-工作者和流水线的核心区别？→ 星型vs链式，中心化调度vs顺序执行
Q3: 辩论模式的最大风险？→ 群体迷思（共享相同预训练数据→相同盲点）
Q4: 黑板模式的调试困难在哪？→ Agent交互非确定性，难以复现
Q5: 四种模式能混合使用吗？→ 可以，如编排-工作者内部子任务用流水线

追问链2：错误传播
Q1: 多Agent系统有哪几类错误？→ 级联/幻觉传播/确认偏误/信息遗漏/任务漂移
Q2: 级联错误怎么缓解？→ 检查点验证+信息溯源+回滚机制
Q3: 幻觉传播和单Agent幻觉有什么区别？→ 多Agent中幻觉会被其他Agent"确认"放大
Q4: 怎么检测任务漂移？→ 每N步检查当前子任务与原始目标的偏离度
Q5: N版本编程是什么？→ 同一任务多Agent独立执行+比较结果，类似冗余备份

追问链3：工程实践
Q1: 多Agent的Token成本怎么优化？→ 模型级联（简单任务用小模型）+缓存+限制辩论轮数
Q2: 怎么保证通信可靠性？→ 至少一次投递+幂等处理+确认-重传
Q3: 状态同步用什么方案？→ 小系统集中式，大系统事件驱动+版本号
Q4: Agent的权限怎么管理？→ 最小权限原则+分层权限+审计日志
Q5: 多Agent系统的评估指标？→ 任务完成率+协作效率+鲁棒性+可扩展性