多Agentåä½œç³»ç»Ÿï¼šæ™ºèƒ½ä½“é€šä¿¡ä¸Žä»»åŠ¡åˆ†é åœ¨å•ä½“ AI Agent çš„èƒ½åŠ›è¾¹ç•Œé€æ¸æ¸ æ™°çš„ä»Šå¤©ï¼Œå¤šæ™ºèƒ½ä½“ï¼ˆMulti-Agentï¼‰åä½œç³»ç»Ÿæ­£åœ¨æˆä¸ºæž„å»ºå¤æ‚æ™ºèƒ½åº”ç”¨çš„ä¸»æµèŒƒå¼ã€‚é€šè¿‡å°†å¤šä¸ªå ·å¤‡ä¸“é•¿çš„æ™ºèƒ½ä½“ç»„ç»‡èµ·æ¥ï¼Œæˆ‘ä»¬å¯ä»¥å®Œæˆå•ä¸€ Agent æ— æ³•èƒœä»»çš„å¤æ‚ä»»åŠ¡ã€‚æœ¬æ–‡å°†ä»Žæž¶æž„è®¾è®¡ã€é€šä¿¡æœºåˆ¶ã€è§’è‰²åˆ†å·¥åˆ°åä½œæµç¨‹ï¼Œç³»ç»Ÿæ€§åœ°ä»‹ç»å¤š Agent ç³»ç»Ÿçš„æ ¸å¿ƒåŽŸç†ä¸Žå®žè·µæ–¹æ³•ã€‚ä¸€ã€ä¸ºä»€ä¹ˆéœ€è¦å¤š Agent 系统单体 Agent é¢ä¸´ä¸‰ä¸ªæ ¸å¿ƒç“¶é¢ˆï¼šèƒ½åŠ›è¿‡è½½ï¼šè¦æ±‚ä¸€ä¸ª Agent åŒæ—¶ç²¾é€šæ•°æ®åˆ†æžã€ä»£ç ç”Ÿæˆã€æ–‡æ¡ˆæ’°å†™ç­‰å¤šä¸ªé¢†åŸŸï¼Œå¾€å¾€å¯¼è‡´æ¯ä¸ªé¢†åŸŸéƒ½åªèƒ½æµ å°è¾„æ­¢ã€‚ä¸Šä¸‹æ–‡é™åˆ¶ï¼šå¤æ‚ä»»åŠ¡çš„ä¸­é—´çŠ¶æ€ä¿¡æ¯è¿‡å¤šï¼Œå• Agent çš„ä¸Šä¸‹æ–‡çª—å£å®¹æ˜“è¢«å æ»¡ã€‚ä»»åŠ¡è€¦åˆï¼šä¸åŒå­ä»»åŠ¡ä¹‹é—´çš„ä¾èµ–å ³ç³»åœ¨å• Agent ä¸­éš¾ä»¥æ¸ æ™°è¡¨è¾¾ï¼Œå¯¼è‡´æ‰§è¡Œæµç¨‹æ··ä¹±ã€‚å¤š Agent 系统通过分而治之的思路，让每个 Agent ä¸“æ³¨äºŽç‰¹å®šé¢†åŸŸï¼Œå†é€šè¿‡åä½œæœºåˆ¶ä¸²è”èµ·æ¥ï¼Œä»Žæ ¹æœ¬ä¸Šè§£å†³ä¸Šè¿°é—®é¢˜ã€‚äºŒã€å¤š Agent 系统架构2.1 å ¸åž‹æž¶æž„æ¨¡å¼å¸¸è§çš„å¤š Agent 架构有三种模式：| 模式 | 结构 | 适用场景 | |------|------|----------| |层级式| 一个主控 Agent 调度多个子 Agent | 任务可分解为明确的子步骤 | |对等式| 多个 Agent 平等协商 | 需要多视角讨论的场景 | |混合式| 层级与对等结合 | 复杂企业级应用 |2.2 æ ¸å¿ƒç»„ä»¶ä¸€ä¸ªå®Œæ•´çš„å¤š Agent ç³»ç»Ÿé€šå¸¸åŒ å«ä»¥ä¸‹ç»„ä»¶ï¼šä»»åŠ¡åˆ†è§£å™¨ï¼šå°†ç”¨æˆ·è¯·æ±‚æ‹†åˆ†ä¸ºå¯æ‰§è¡Œçš„å­ä»»åŠ¡è°ƒåº¦å™¨ï¼šè´Ÿè´£ä»»åŠ¡åˆ†é ä¸Žæ‰§è¡Œé¡ºåºæŽ§åˆ¶Agent 注册中心：管理所有可用 Agent çš„èƒ½åŠ›æè¿°å ±äº«è®°å¿†ï¼šå­˜å‚¨ä¸­é—´ç»“æžœå’Œå ¨å±€çŠ¶æ€é€šä¿¡æ€»çº¿ï¼šAgent ä¹‹é—´çš„æ¶ˆæ¯ä¼ é€’é€šé“ä¸‰ã€Agent é€šä¿¡ä¸Žæ¶ˆæ¯ä¼ é€’3.1 消息协议设计Agent ä¹‹é—´çš„é€šä¿¡éœ€è¦æ ‡å‡†åŒ–çš„æ¶ˆæ¯æ ¼å¼ã€‚ä»¥ä¸‹æ˜¯ä¸€ä¸ªå ¸åž‹çš„æ¶ˆæ¯ç»“æž„ï¼šfrom dataclasses import dataclass from enum import Enum from typing import Optional, Dict, Any class MessageType(Enum): TASK_ASSIGN task_assign # ä»»åŠ¡åˆ†é  TASK_RESULT task_result # 任务结果 QUERY query # 查询请求 BROADCAST broadcast # 广播消息 COORDINATION coordination # 协调消息 dataclass class AgentMessage: msg_id: str # æ¶ˆæ¯å”¯ä¸€æ ‡è¯† sender: str # å‘é€è€ ID receiver: str # æŽ¥æ”¶è€ ID（broadcast时为空） msg_type: MessageType # 消息类型 payload: Dict[str, Any] # æ¶ˆæ¯å† å®¹ timestamp: float # 时间戳 reply_to: Optional[str] None # 回复的消息ID def to_dict(self) - dict: return { msg_id: self.msg_id, sender: self.sender, receiver: self.receiver, msg_type: self.msg_type.value, payload: self.payload, timestamp: self.timestamp, reply_to: self.reply_to }3.2 通信总线实现通信总线是多 Agent ç³»ç»Ÿçš„ç¥žç»ç³»ç»Ÿï¼Œè´Ÿè´£æ¶ˆæ¯çš„å¯é æŠ•é€’ï¼šimport asyncio import uuid from typing import Callable, Dict, List from collections import defaultdict class MessageBus: def __init__(self): self.subscribers: Dict[str, List[Callable]] defaultdict(list) self.message_history: List[AgentMessage] [] def subscribe(self, agent_id: str, handler: Callable): Agentè®¢é˜ ç‰¹å®šç±»åž‹çš„æ¶ˆæ¯ self.subscribers[agent_id].append(handler)