多Agent系统架构设计：通信、协调与冲突解决

┌─────────────────────────────────────────────────┐
│          多Agent系统的三大核心挑战                │
├─────────────────────────────────────────────────┤
│                                                 │
│  1. 通信 (Communication)                        │
│     Agent A 如何告诉 Agent B 它需要什么？       │
│     - 消息格式                                  │
│     - 通信协议                                  │
│     - 异步 vs 同步                              │
│                                                 │
│  2. 协调 (Coordination)                         │
│     多个Agent如何协同完成一个任务？               │
│     - 任务分配                                  │
│     - 资源竞争                                  │
│     - 死锁避免                                  │
│                                                 │
│  3. 冲突解决 (Conflict Resolution)               │
│     当Agent意见不一致时怎么办？                   │
│     - 投票机制                                  │
│     - 仲裁机制                                  │
│     - 置信度加权                                │
│                                                 │
└─────────────────────────────────────────────────┘

from pydantic import BaseModel
from typing import Optional, Any
from enum import Enum
from datetime import datetime

class MessageType(str, Enum):
    REQUEST = "request"           # 请求帮助
    RESPONSE = "response"         # 响应请求
    NOTIFICATION = "notification"  # 通知（无需回复）
    PROPOSAL = "proposal"         # 提出方案
    VOTE = "vote"                 # 投票
    RESULT = "result"             # 最终结果

class AgentMessage(BaseModel):
    message_id: str
    from_agent: str
    to_agent: str | list[str]     # 可以广播
    message_type: MessageType
    timestamp: datetime
    
    # 核心内容
    content: str                   # 自然语言内容
    structured_data: Optional[dict] = None  # 结构化数据
    
    # 元数据
    correlation_id: Optional[str] = None  # 关联请求ID
    ttl: int = 3                  # 最大转发次数
    priority: int = 1              # 优先级 1-5
    requires_response: bool = False
    timeout_seconds: Optional[int] = None

import asyncio
from typing import Callable

class AgentCommunicator:
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.message_queue = asyncio.Queue()
        self.subscribers: dict[str, list[Callable]] = {}
    
    async def send(self, message: AgentMessage):
        """发送消息"""
        if isinstance(message.to_agent, str):
            # 单播
            await self._deliver(message.to_agent, message)
        else:
            # 多播
            for recipient in message.to_agent:
                await self._deliver(recipient, message)
    
    async def broadcast(self, message: AgentMessage):
        """广播消息（所有Agent都能收到）"""
        message.to_agent = "*"
        await self._publish(message)
    
    def subscribe(self, message_type: MessageType, handler: Callable):
        """订阅特定类型的消息"""
        if message_type not in self.subscribers:
            self.subscribers[message_type] = []
        self.subscribers[message_type].append(handler)
    
    async def receive(self) -> AgentMessage:
        """接收消息（阻塞）"""
        return await self.message_queue.get()
    
    async def _deliver(self, recipient: str, message: AgentMessage):
        """实际投递（简化版）"""
        # 生产环境应使用Redis/RabbitMQ等消息中间件
        recipient_queue = self._get_queue(recipient)
        await recipient_queue.put(message)

Manager Agent
    ├── 接收任务
    ├── 分解任务
    ├── 评估各Worker能力
    └── 分配子任务给Worker Agents

class ManagerAgent:
    def __init__(self, workers: list):
        self.workers = workers  # Worker Agent列表
        self.worker_profiles = {w.id: w.get_profile() for w in workers}
    
    async def assign_task(self, task: dict) -> dict:
        """分配任务给最合适的Worker"""
        # 评估每个Worker的适合度
        scores = []
        for worker in self.workers:
            score = await self._evaluate_fit(worker, task)
            scores.append((worker, score))
        
        # 选择最佳Worker
        best_worker, best_score = max(scores, key=lambda x: x[1])
        
        # 分配任务
        result = await best_worker.execute(task)
        return result
    
    async def _evaluate_fit(self, worker, task) -> float:
        """评估Worker适合度（0-1）"""
        profile = self.worker_profiles[worker.id]
        
        # 能力匹配
        capability_match = sum(
            1 for req in task.get("required_capabilities", [])
            if req in profile["capabilities"]
        ) / len(task.get("required_capabilities", [1]))
        
        # 负载情况
        load_factor = 1.0 - (worker.current_load / worker.max_load)
        
        # 历史成功率
        success_rate = profile["historical_success_rate"]
        
        # 综合评分
        return 0.4 * capability_match + 0.3 * load_factor + 0.3 * success_rate

1. Manager广播任务提案（Call for Proposals）
2. Worker评估自己能否完成，提交投标（Bid）
3. Manager评估投标，选择最优Worker
4. 与选中的Worker签订合约（Contract）
5. Worker执行任务，提交结果
6. Manager确认结果，支付（虚拟）奖励

class ContractNetManager:
    async def announce_task(self, task: dict):
        """广播任务"""
        cfp = {
            "task_id": task["id"],
            "description": task["description"],
            "deadline": task["deadline"],
            "reward": task["reward"]
        }
        await self.broadcast("cfp", cfp)
    
    async def evaluate_bids(self, bids: list) -> dict:
        """评估投标，选择最优"""
        scored_bids = []
        for bid in bids:
            # 综合评估：能力 + 报价 + 工期
            score = (
                0.4 * bid["capability_score"] +
                0.3 * (1.0 / bid["price"]) +  # 价格越低越好
                0.3 * (1.0 / bid["estimated_time"])  # 时间越短越好
            )
            scored_bids.append((bid, score))
        
        return max(scored_bids, key=lambda x: x[1])[0]

class CapabilityRegistry:
    """Agent能力注册中心"""
    
    def __init__(self):
        self.registry = {}  # {capability: [agent_ids]}
    
    def register(self, agent_id: str, capabilities: list):
        for cap in capabilities:
            if cap not in self.registry:
                self.registry[cap] = []
            self.registry[cap].append(agent_id)
    
    def find_agents_with_capability(self, capability: str) -> list:
        return self.registry.get(capability, [])

class ResourceManager:
    def __init__(self):
        self.allocated = {}  # {resource_id: agent_id}
        self.waiting = []    # [(agent_id, resource_id)]
    
    def detect_deadlock(self) -> list:
        """使用资源分配图检测死锁"""
        # 构建等待图
        wait_graph = {}
        for agent, resource in self.waiting:
            holder = self.allocated.get(resource)
            if holder:
                if agent not in wait_graph:
                    wait_graph[agent] = []
                wait_graph[agent].append(holder)
        
        # 检测环
        visited = set()
        path = []
        
        def dfs(node):
            if node in path:
                return path[path.index(node):]  # 发现环
            if node in visited:
                return None
            visited.add(node)
            path.append(node)
            
            for neighbor in wait_graph.get(node, []):
                result = dfs(neighbor)
                if result:
                    return result
            
            path.pop()
            return None
        
        for node in list(wait_graph.keys()):
            result = dfs(node)
            if result:
                return result  # 返回死锁环
        
        return None  # 无死锁

class HierarchicalResourceAllocator:
    """按层次分配资源，避免循环等待"""
    
    def __init__(self, resources: list):
        # 为资源分配全局顺序
        self.resource_order = {r: i for i, r in enumerate(resources)}
    
    async def request_resources(self, agent_id: str, resources: list) -> bool:
        """按资源编号升序请求，避免死锁"""
        sorted_resources = sorted(resources, key=lambda r: self.resource_order[r])
        
        acquired = []
        try:
            for resource in sorted_resources:
                await self._acquire(resource, agent_id)
                acquired.append(resource)
            return True
        except TimeoutError:
            # 获取失败，释放已获取的
            for resource in acquired:
                await self._release(resource, agent_id)
            return False

class ConfidenceWeightedVoting:
    async def resolve(self, proposals: list[dict]) -> dict:
        """
        proposals: [{
            "agent_id": "agent_1",
            "proposal": "买入",
            "confidence": 0.85,
            "reasoning": "..."
        }, ...]
        """
        # 按提案内容分组
        groups = {}
        for p in proposals:
            key = p["proposal"]
            if key not in groups:
                groups[key] = []
            groups[key].append(p)
        
        # 计算加权得分
        scores = {}
        for proposal, supporters in groups.items():
            # 加权置信度之和
            weighted_score = sum(
                s["confidence"] * self._get_agent_weight(s["agent_id"])
                for s in supporters
            )
            scores[proposal] = weighted_score
        
        # 返回最高分提案
        return max(scores.items(), key=lambda x: x[1])[0]
    
    def _get_agent_weight(self, agent_id: str) -> float:
        """根据Agent历史表现给权重"""
        # 可以从数据库查询
        weights = {
            "domain_expert": 1.5,
            "data_analyst": 1.2,
            "generalist": 1.0
        }
        return weights.get(agent_id, 1.0)

class DebateResolver:
    """让Agent通过辩论解决分歧"""
    
    async def resolve(self, proposals: list, max_rounds: int = 3) -> dict:
        if len(proposals) == 1:
            return proposals[0]["proposal"]
        
        # 多轮辩论
        for round_num in range(max_rounds):
            print(f"--- 辩论第 {round_num + 1} 轮 ---")
            
            # 每个Agent看到其他Agent的论证
            for proposal in proposals:
                agent_id = proposal["agent_id"]
                argument = await self._generate_argument(
                    agent_id, proposals, round_num
                )
                proposal["arguments"].append(argument)
            
            # 评估是否达成共识
            if self._check_consensus(proposals):
                break
        
        # 最终投票
        return await self._final_vote(proposals)
    
    async def _generate_argument(self, agent_id: str, proposals: list, round_num: int) -> str:
        """生成辩论论据"""
        prompt = f"""你是 {agent_id}。
        当前各Agent的提案和理由：
        {self._format_proposals(proposals)}
        
        请在第{round_num + 1}轮辩论中：
        1. 重申你的提案的合理性
        2. 指出其他提案的问题
        3. 提出新的考虑因素（如果有的话）
        """
        return await self.llm.complete(prompt)

class ArbitratorAgent:
    """专门的仲裁Agent，在冲突时做最终决策"""
    
    async def arbitrate(self, conflict: dict) -> dict:
        """
        conflict: {
            "task": "...",
            "proposals": [...],
            "context": {...}
        }
        """
        # 分析每个提案
        analyses = []
        for proposal in conflict["proposals"]:
            analysis = await self._analyze_proposal(proposal, conflict["context"])
            analyses.append(analysis)
        
        # 综合决策
        decision = await self._make_decision(analyses)
        
        return {
            "winning_proposal": decision["winner"],
            "reasoning": decision["reasoning"],
            "confidence": decision["confidence"]
        }
    
    async def _analyze_proposal(self, proposal: dict, context: dict) -> dict:
        prompt = f"""分析以下提案：
        提案：{proposal['proposal']}
        理由：{proposal.get('reasoning', '')}
        上下文：{context}
        
        请从以下维度评估（0-10分）：
        1. 可行性
        2. 风险
        3. 收益
        4. 与上下文的匹配度
        """
        return await self.llm.complete(prompt)

┌─────────────────────────────────────────────────┐
│              用户消息                           │
└────────────────┬────────────────────────────────┘
                 ↓
┌─────────────────────────────────────────────────┐
│          路由Agent (意图识别)                    │
│  判断：咨询/投诉/退换货/技术问题                │
└────────────────┬────────────────────────────────┘
                 ↓
        ┌────────┴────────┐
        ↓                 ↓
┌──────────────┐   ┌──────────────┐
│ 咨询Agent群  │   │ 投诉Agent群  │
│  - 产品专家  │   │  - 情绪安抚  │
│  - 库存查询  │   │  - 补偿方案  │
│  - 订单查询  │   │  - 升级处理  │
└──────┬───────┘   └──────┬───────┘
       ↓                   ↓
┌─────────────────────────────────────────────────┐
│              决策融合Agent                       │
│  综合多个Agent的意见，生成最终回复               │
└────────────────┬────────────────────────────────┘
                 ↓
┌─────────────────────────────────────────────────┐
│              质量检查Agent                       │
│  检查回复：礼貌性、准确性、合规性               │
│  不通过则打回决策融合Agent                      │
└────────────────┬────────────────────────────────┘
                 ↓
┌─────────────────────────────────────────────────┐
│              发送回复给用户                      │
└─────────────────────────────────────────────────┘

用户: "我要退货"
  ↓
路由Agent → 判断为"退换货" → 激活退换货Agent群
  ↓
库存查询Agent: "查询订单号XXX的购买记录..."
  ↓ (异步)
政策查询Agent: "根据退货政策，该商品符合退货条件..."
  ↓ (异步)
补偿方案Agent: "建议全额退款 + 10元优惠券..."
  ↓
决策融合Agent: 综合三方意见 → "可以退货，将全额退款..."
  ↓
质量检查Agent: 检查通过
  ↓
回复用户