> **本文是「架构实战·第二季」第九篇**。上一篇我们讲了 ReAct 单 Agent 模式，本篇升级到"多 Agent 协作"。

---

## 引子：单 Agent 撑不住的复杂任务

去年我们做一个数据分析项目，让一个 Agent 自己跑完"读 SQL→抽取数据→生成图表→写报告→发邮件"全流程。

结果三天两头翻车：

- 拿到 50 万条数据，不知道该先看什么；
- 写到一半发现 SQL 写错了，要回到起点；
- 邮件发出去了，但格式乱了；
- 整体需要 2 小时，但单 Agent 跑 1.5 小时就开始"犯糊涂"。

后来我们把任务拆给 4 个 Agent 协作，全程 20 分钟搞定。

**多 Agent 的本质**：把大任务拆成小任务，让有"专长"的 Agent 各司其职。

---

## 一、为什么需要多 Agent

### 1.1 单 Agent 的瓶颈

| 瓶颈 | 表现 |
|------|------|
| 上下文窗口 | 任务复杂度上升 → 上下文爆炸 |
| 角色串扰 | 数据分析师 prompt 和邮件模板写在一起，互相污染 |
| 推理质量 | 任务多样，模型泛化能力下降 |
| 调试困难 | 不知道哪一步出问题 |
| 并发能力 | 单 Agent 串行，效率低 |

### 1.2 多 Agent 的核心优势

- **专精化**：每个 Agent 有独立的 system prompt、工具集、知识库；
- **分工**：任务拆分 → 各 Agent 负责自己擅长部分；
- **并发**：独立 Agent 可并行执行；
- **可调试**：每个 Agent 的输入输出独立可观测。

---

## 二、多 Agent 的四大模式

### 2.1 Supervisor 模式（监督者）

```
           ┌────────────┐
           │ Supervisor │
           │  派任务/总结 │
           └─────┬──────┘
        ┌────────┼────────┐
        ▼        ▼        ▼
    ┌──────┐ ┌──────┐ ┌──────┐
    │Agent1│ │Agent2│ │Agent3│
    │ 数据 │ │ 图表 │ │ 邮件 │
    └──────┘ └──────┘ └──────┘
```

**适用**：任务的"主线 + 多个分支"，需要协调。

**实现**：

```python
class SupervisorAgent:
    def __init__(self, sub_agents, llm):
        self.sub_agents = {a.name: a for a in sub_agents}
        self.llm = llm

    def run(self, task):
        # 1. 拆分任务
        subtasks = self.decompose(task)
        
        # 2. 分发
        results = {}
        for sub in subtasks:
            agent = self.sub_agents[sub["agent"]]
            results[sub["id"]] = agent.run(sub["input"])
        
        # 3. 汇总
        return self.summarize(task, results)
    
    def decompose(self, task):
        prompt = f"""把任务拆成多个子任务，每个子任务分配给一个专员：
        可用专员：{list(self.sub_agents.keys())}
        
        任务：{task}
        
        输出 JSON：
        {{"subtasks": [
          {{"id": "1", "agent": "data", "input": "..."}},
          ...
        ]}}"""
        return json.loads(self.llm.complete(prompt))["subtasks"]
```

### 2.2 Peer-to-Peer 模式（平等协作）

```
Agent1 ←→ Agent2 ←→ Agent3
  ↑ ↓       ↑ ↓       ↑ ↓
        消息总线
```

**适用**：需要多角色辩论、协同决策。

**实现**：

```python
class PeerAgent:
    def __init__(self, name, role, llm, peers, max_rounds=5):
        self.name = name
        self.role = role
        self.llm = llm
        self.peers = peers  # 其他 Agent 列表
        self.max_rounds = max_rounds
    
    def discuss(self, topic):
        history = []
        for round in range(self.max_rounds):
            my_view = self.my_view(topic, history)
            history.append({"agent": self.name, "view": my_view})
            
            # 让其他 Peer 也发表
            for peer in self.peers:
                peer_view = peer.my_view(topic, history)
                history.append({"agent": peer.name, "view": peer_view})
            
            # 检查是否收敛
            if self.is_consensus(history):
                break
        
        return history
```

### 2.3 Pipeline 模式（流水线）

```
Input → Agent1 → Agent2 → Agent3 → Output
```

**适用**：数据加工、文档处理等有明确顺序的任务。

**实现**：

```python
class Pipeline:
    def __init__(self, stages):
        self.stages = stages  # [agent1, agent2, agent3]
    
    async def run(self, input_data):
        result = input_data
        for stage in self.stages:
            result = await stage.process(result)
            # 可选：每步加审计
            audit_log(f"Pipeline stage {stage.name} completed", result)
        return result
```

### 2.4 Debate 模式（对抗辩论）

```
Agent（正方）  ←→  Agent（反方）
       ↓
     裁判 Agent
```

**适用**：决策审查、风险评估、内容质量审查。

**实现**：

```python
class DebateAgent:
    def __init__(self, llm_pro, llm_con, judge):
        self.pro = llm_pro
        self.con = llm_con
        self.judge = judge
    
    def debate(self, decision, rounds=3):
        history = []
        for i in range(rounds):
            arg_pro = self.pro.argue(decision, history, side="正方")
            history.append({"side": "pro", "arg": arg_pro})
            
            arg_con = self.con.argue(decision, history, side="反方")
            history.append({"side": "con", "arg": arg_con})
        
        verdict = self.judge.decide(decision, history)
        return {
            "decision": verdict,
            "history": history
        }
```

### 2.5 模式对比

| 模式 | 复杂度 | 适用 |
|------|--------|------|
| Supervisor | 中 | 任务分层清晰 |
| Peer-to-Peer | 高 | 协同决策 |
| Pipeline | 低 | 顺序处理 |
| Debate | 高 | 审查/挑战 |

---

## 三、Agent 间通信协议

### 3.1 通信方式对比

| 方式 | 实现 | 适用 |
|------|------|------|
| 共享内存 | Redis 共享 state | 同步、快 |
| 消息队列 | Kafka/RabbitMQ | 异步、解耦 |
| HTTP API | RESTful | 跨语言、跨系统 |
| 工具调用 | 把其他 Agent 当工具 | 紧耦合 |
| 文件/数据库 | S3 + 文件 | 异步、审计 |

### 3.2 共享内存实现

```python
class SharedBlackboard:
    """多 Agent 共享的"黑板"模式"""
    
    def __init__(self):
        self.board = {}
        self.lock = asyncio.Lock()
    
    async def write(self, key, value, author):
        async with self.lock:
            self.board[key] = {
                "value": value,
                "author": author,
                "ts": now()
            }
    
    def read(self, key):
        return self.board.get(key, {}).get("value")
    
    def read_all(self):
        return self.board

class DataAgent:
    def __init__(self, board):
        self.board = board
    
    async def run(self, query):
        data = await self.query_db(query)
        await self.board.write("raw_data", data, author="data")
        return data

class ChartAgent:
    async def run(self, _):
        data = self.board.read("raw_data")
        if not data:
            return {"error": "no data"}
        chart = await self.generate_chart(data)
        await self.board.write("chart", chart, author="chart")
        return chart
```

### 3.3 消息队列实现

适合长时间运行或分布式：

```python
class MessageBus:
    """基于 Kafka 的 Agent 消息总线"""
    
    def __init__(self, kafka_bootstrap):
        self.producer = KafkaProducer(bootstrap_servers=kafka_bootstrap)
    
    async def publish(self, topic, msg):
        self.producer.send(topic, json.dumps(msg).encode())
    
    def subscribe(self, topic, group_id):
        return KafkaConsumer(topic, group_id=group_id, ...)
```

---

## 四、冲突解决机制

多 Agent 协作最容易出问题的地方：两个 Agent 都想主导、结论冲突。

### 4.1 常见冲突场景

| 场景 | 表现 |
|------|------|
| 结论冲突 | A 说"产品做"，B 说"产品别做" |
| 资源冲突 | 两个 Agent 同时改一份文档 |
| 责任推诿 | 出错时都说"不是我的问题" |
| 状态不一致 | Agent A 用的旧版本数据 |

### 4.2 冲突解决策略

**策略 1：优先级仲裁**

```python
class PriorityArbitrator:
    def resolve(self, conflict):
        # 按角色优先级
        PRIORITY = {"supervisor": 100, "expert": 80, "junior": 50}
        
        parties = conflict["parties"]
        winner = max(parties, key=lambda p: PRIORITY[p["role"]])
        return winner["view"]
```

**策略 2：投票制**

```python
def vote_resolution(agents_results):
    """多数决"""
    counter = Counter(agents_results)
    if counter.most_common(1)[0][1] > len(agents_results) / 2:
        return counter.most_common(1)[0][0]
    return None  # 平票，转人工
```

**策略 3：置信度权重**

```python
def weighted_resolution(results):
    """每个 Agent 输出置信度，按置信度加权"""
    weights = [r["confidence"] for r in results]
    views = [r["answer"] for r in results]
    return weighted_majority(views, weights)
```

**策略 4：辩论收敛**

```python
class DebateResolver:
    def resolve(self, decision, rounds=3):
        """反对方不断 challenge，最终收敛"""
        history = []
        for i in range(rounds):
            challenge = self.opponent.challenge(decision, history)
            if self.is_valid_challenge(challenge):
                decision = self.refine(decision, challenge)
                history.append({"round": i, "challenge": challenge})
            else:
                break  # 没有新反对意见了
        return decision
```

### 4.3 共享资源的并发控制

```python
class ResourceLock:
    """Agent 写共享资源的锁"""
    
    def __init__(self):
        self.locks = {}
    
    async def acquire(self, resource, agent, timeout=30):
        # 基于 Redis 的分布式锁
        if await redis.set(f"lock:{resource}", agent, nx=True, ex=timeout):
            return True
        return False
    
    def release(self, resource):
        redis.delete(f"lock:{resource}")

# 使用：Agent 改文档前必须 acquire
class DocumentAgent:
    async def update_doc(self, doc_id, content):
        if not await lock.acquire(f"doc:{doc_id}", "doc-agent"):
            return {"error": "another agent is updating"}
        
        try:
            doc = await doc_db.update(doc_id, content)
            audit_log("doc.updated", {"agent": "doc-agent", "doc_id": doc_id})
            return doc
        finally:
            lock.release(f"doc:{doc_id}")
```

---

## 五、实战：搭建"产品+开发+测试"三 Agent 团队

### 5.1 场景

> 老板说："我要一个待办功能。"

需要：
1. 产品 Agent：拆解需求，写 PRD；
2. 开发 Agent：写代码、提 PR；
3. 测试 Agent：自动写测试、验证。

### 5.2 三 Agent 实现

```python
PRODUCT_AGENT_PROMPT = """你是产品经理 Agent。
你的职责：
1. 接收用户需求
2. 拆解成可执行的功能点
3. 写出 PRD（包括用户故事、验收标准、边界条件）
输出格式：标准 PRD 模板"""

DEV_AGENT_PROMPT = """你是开发 Agent。
你的职责：
1. 根据 PRD 生成代码
2. 写完代码后通知测试 Agent
3. 如果测试失败，修复并重提
限制：代码必须通过 linter。"""

QA_AGENT_PROMPT = """你是测试 Agent。
你的职责：
1. 根据 PRD 写测试用例
2. 跑测试，给出测试报告
3. 如果失败，明确指出失败原因给开发 Agent"""

class Team:
    def __init__(self, llm):
        self.product = Agent(llm, prompt=PRODUCT_AGENT_PROMPT, tools=[search_past_prd])
        self.dev = Agent(llm, prompt=DEV_AGENT_PROMPT, tools=[read_file, edit_file, run_command])
        self.qa = Agent(llm, prompt=QA_AGENT_PROMPT, tools=[run_test])
        self.bus = MessageBus("kafka:9092")
    
    def run(self, user_request):
        # Step 1: 产品 PRD
        prd = self.product.run(user_request)
        self.bus.publish("prd.ready", prd)
        
        # Step 2: 开发
        code = self.dev.run(prd)
        self.bus.publish("code.ready", code)
        
        # Step 3: 测试
        test_report = self.qa.run({"prd": prd, "code": code})
        
        # Step 4: 失败则回到 Step 2
        while not test_report["all_passed"]:
            fixed_code = self.dev.run({"error": test_report["failures"]})
            test_report = self.qa.run({"prd": prd, "code": fixed_code})
        
        return {"prd": prd, "code": code, "test_report": test_report}
```

### 5.3 进阶：Supervisor 调度

任务复杂时用 Supervisor 调度：

```python
class TeamSupervisor:
    def __init__(self, team, llm):
        self.team = team
        self.llm = llm
    
    def plan(self, request):
        return json.loads(self.llm.complete(f"""你是技术总监。
        团队成员：product / dev / qa。
        任务：{request}
        
        给出执行计划（JSON）：
        {{"steps": [
          {{"agent": "product", "task": "..."}},
          {{"agent": "dev", "task": "...", "depends_on": [0]}}
        ]}}"""))
    
    def run(self, request):
        plan = self.plan(request)
        results = {}
        for step in plan["steps"]:
            deps = step.get("depends_on", [])
            inputs = {d: results[d] for d in deps}
            inputs["task"] = step["task"]
            
            agent = getattr(self.team, step["agent"])
            results[step["id"]] = agent.run(inputs)
        return results
```

---

## 六、可观测与成本

### 6.1 关键指标

```promql
agent_subtask_total{agent, status}            # 子任务总数
agent_subtask_duration_seconds{agent}         # 单 Agent 执行时长
agent_token_used{agent, stage}                # 各 Agent token 用量
agent_conflict_total{resolution}              # 冲突发生次数
agent_consensus_rounds                        # 收敛轮数
agent_step_failure_rate{agent, stage}         # 失败率
```

### 6.2 成本控制

```python
class CostAwareTeam:
    def __init__(self, team, budget):
        self.team = team
        self.budget = budget  # 美元
    
    def run(self, task):
        estimated = self.estimate_cost(task)
        if estimated > self.budget:
            return {"error": "超出预算", "estimated": estimated}
        
        actual = 0
        for agent_name, agent_task in self.decompose(task):
            cost = agent.run(agent_task)
            actual += cost
            if actual > self.budget:
                return {"error": "执行中超出预算", "spent": actual}
        
        return {"task": task, "cost": actual}
```

---

## 七、踩坑提示

### 坑 1：Agent 间循环依赖

**症状**：Agent A 等 B 的输出，B 等 A 的输出——死锁。

**解决**：

```python
# 拓扑排序 + DAG 检查
def validate_dag(steps):
    for step in steps:
        deps = step.get("depends_on", [])
        if step["id"] in transitive_closure(deps):
            return False
    return True
```

### 坑 2：Token 用量爆炸

**症状**：多 Agent 跑一个任务，token 是单 Agent 的 5-10 倍。

**解决**：
- 上小模型（如 7B）做局部决策；
- 关键节点用大模型；
- Agent 间通信压缩（不传全量上下文，只传摘要）。

### 坑 3：状态在不同 Agent 间不一致

**症状**：Agent A 写了状态，Agent B 看不到。

**解决**：状态写入必须等 confirm，Agent 启动时强制刷新。

### 坑 4：Agent 互相"甩锅"

**症状**：出问题时都说"不是我的事"。

**解决**：用集中审计 + 责任标记：

```python
audit_log.info({
    "stage": "chart_generation",
    "executor": "chart-agent",
    "input": "...",  
    "output": "...",
    "consumed_input": ["raw_data"],
    "produced_output": ["chart_url"]
})
```

---

## 八、本篇要点回顾

| 模式 | 适用 |
|------|------|
| Supervisor | 任务拆分清晰 |
| Peer-to-Peer | 协同决策 |
| Pipeline | 顺序处理 |
| Debate | 审查挑战 |

**Agent 间协作的三个核心原则**：

1. **任务必须可拆分**——不能拆分就别用多 Agent；
2. **通信协议必须清晰**——否则就是一团乱麻；
3. **冲突解决必须预设**——临时讨论效率极低。

---

## 九、下一篇预告

下一篇 【架构实战】**AI对话系统：从客服机器人到智能助手**，我们聚焦 AI Agent 的"高频落地形态"——对话系统：

- 对话系统的四大模块（NLU、DST、Policy、NLG）与 LLM 重新定义；
- 短期记忆 vs 长期记忆工程；
- 多轮对话状态管理；
- 实战：从零搭建一个企业级客服对话系统。

---

> **作者**：架构实战 第二季  
> **版本**：v1.0  
> **最后更新**：2026-10  
> **全文约 5000 字**

本文是「架构实战·第二季」第九篇。上一篇我们讲了 ReAct 单 Agent 模式，本篇升级到"多 Agent 协作"。

【架构实战】多Agent协作：角色分工与冲突解决机制

研发部经理

架构设计

人工智能

后端

多Agent协作模式通过任务拆解、专精分工与并行处理解决单Agent上下文爆炸、角色污染与推理退化问题，覆盖Supervisor、Pipeline、Peer-to-Peer、Debate四大架构及通信协议、冲突仲裁机制，实现复杂任务高效落地，配以成本控制与可观测性实践。

2026上云采购 | AI焕新·智启新局

cscbot

lexiang

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【架构实战】多Agent协作：角色分工与冲突解决机制-腾讯云开发者社区-腾讯云

【架构实战】多Agent协作：角色分工与冲突解决机制

【架构实战】多Agent协作：角色分工与冲突解决机制

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐