與協(xié)作的通信協(xié)議設(shè)計(jì))
1. Agent-Reach是什么為什么這個(gè)時(shí)機(jī)值得聊這幾年AI圈子里的風(fēng)向變得非常快從大模型卷參數(shù)到卷上下文窗口再到卷工具調(diào)用、多智能體協(xié)作。你去看各種開(kāi)源社區(qū)和產(chǎn)品發(fā)布會(huì)幾乎每個(gè)團(tuán)隊(duì)都在說(shuō)自己做了一個(gè)Agent框架能規(guī)劃、能反思、能調(diào)用工具。但真正把多個(gè)Agent放到一起跑的時(shí)候問(wèn)題就出來(lái)了Agent和Agent之間怎么找到對(duì)方、怎么說(shuō)話、怎么協(xié)作Agent-Reach這個(gè)項(xiàng)目解決的核心問(wèn)題就是這個(gè)——智能體之間的觸達(dá)與協(xié)作。我用一句大白話解釋Agent-Reach它是給AI智能體之間“遞紙條”的一套通用規(guī)則。兩個(gè)Agent可以有不同的模型底座、不同的業(yè)務(wù)邏輯、甚至部署在不同服務(wù)器上但只要它們同時(shí)接入Agent-Reach就能互相發(fā)現(xiàn)、互相發(fā)消息、按約定格式完成協(xié)作任務(wù)。就像微信、釘釘、短信之間各自為戰(zhàn)但Agent-Reach相當(dāng)于一個(gè)統(tǒng)一的“運(yùn)營(yíng)商”讓所有智能體能通過(guò)一套標(biāo)準(zhǔn)方式觸達(dá)彼此。這個(gè)時(shí)機(jī)很關(guān)鍵?，F(xiàn)在多智能體系統(tǒng)已經(jīng)不只是學(xué)術(shù)圈的概念MCP模型上下文協(xié)議這類(lèi)模型接入?yún)f(xié)議正在普及開(kāi)發(fā)者手里有大量已經(jīng)封裝好的“智能體單元”——有的擅長(zhǎng)查庫(kù)存有的擅長(zhǎng)寫(xiě)文案有的擅長(zhǎng)做圖像識(shí)別。真正缺的不是單個(gè)Agent的能力而是把碎片化的Agent編排到一起的“連接層”。Agent-Reach這類(lèi)項(xiàng)目恰好卡在這個(gè)位置它不替代模型不替代業(yè)務(wù)邏輯只做觸達(dá)。什么人適合參考這篇文章兩類(lèi)人。一類(lèi)是已經(jīng)在用LangChain、AutoGen、CrewAI這類(lèi)框架做多智能體應(yīng)用的開(kāi)發(fā)者你已經(jīng)遇到了“多個(gè)Agent各自為政、消息格式對(duì)不上、調(diào)用關(guān)系混亂”這些問(wèn)題需要一套更清晰的協(xié)作協(xié)議。另一類(lèi)是準(zhǔn)備從零搭建分布式智能服務(wù)架構(gòu)的技術(shù)負(fù)責(zé)人你想知道Agent之間的通信鏈路應(yīng)該怎么設(shè)計(jì)。我下面講的內(nèi)容既有協(xié)議層面的設(shè)計(jì)思路也有可以直接抄走的代碼實(shí)現(xiàn)還有我在實(shí)際調(diào)試中踩過(guò)的坑希望能幫你少走彎路。2. 核心思路拆解Agent之間到底怎么“觸達(dá)”2.1 先把問(wèn)題定義清楚觸達(dá)≠消息推送很多人看到Agent-Reach第一反應(yīng)是“這不就是個(gè)消息隊(duì)列嗎”或者“用HTTP調(diào)一下不就行了”真不是這么簡(jiǎn)單。智能體觸達(dá)和傳統(tǒng)服務(wù)間通信最大的區(qū)別在于Agent是有“認(rèn)知狀態(tài)”的。傳統(tǒng)微服務(wù)A調(diào)用服務(wù)BB返回一個(gè)結(jié)果這次調(diào)用就結(jié)束了。但兩個(gè)Agent協(xié)作時(shí)A發(fā)出請(qǐng)求之后B可能需要規(guī)劃步驟、調(diào)用幾個(gè)外部工具、甚至反問(wèn)A幾個(gè)問(wèn)題才能給出結(jié)果。整個(gè)過(guò)程中雙方的“對(duì)話上下文”要保持連續(xù)消息不僅要能送達(dá)到還要能被理解、被納入對(duì)方的決策流程。這就是觸達(dá)和簡(jiǎn)單消息投遞的區(qū)別——觸達(dá)包含尋址、能力協(xié)商、會(huì)話建立、消息路由、狀態(tài)同步這幾個(gè)層面。Agent-Reach處理觸達(dá)問(wèn)題核心是三層模型第一層是身份層解決“對(duì)方是誰(shuí)”。每個(gè)Agent在Agent-Reach網(wǎng)絡(luò)里有一個(gè)全局唯一的ID以及它的能力描述文件。比如Agent-A8F23能力描述是“可以調(diào)用天氣API支持中文和英文問(wèn)答”。其他Agent看到這個(gè)描述文件就知道該不該找它、怎么和它說(shuō)話。第二層是會(huì)話層解決“怎么說(shuō)話”。兩個(gè)Agent之間建立會(huì)話要協(xié)商消息格式、最大消息長(zhǎng)度、超時(shí)策略、是否支持流式響應(yīng)。這就像兩個(gè)人打電話之前先確認(rèn)“你說(shuō)中文嗎說(shuō)慢點(diǎn)我記一下”。第三層是路由層解決“消息怎么走”。Agent所在的節(jié)點(diǎn)可能是在不同局域網(wǎng)有的在內(nèi)網(wǎng)有的在云上還有可能動(dòng)態(tài)遷移。Agent-Reach維護(hù)了一套輕量級(jí)的位置注冊(cè)表消息不需要知道對(duì)方的具體地址只要按Agent ID發(fā)出去路由層負(fù)責(zé)找到當(dāng)前的位置并轉(zhuǎn)發(fā)。2.2 為什么不直接用現(xiàn)成的RPC框架你可能會(huì)問(wèn)分布式系統(tǒng)有那么多成熟的RPC框架gRPC、Dubbo、Thrift都很好用為什么還要自己做一套Agent-Reach我實(shí)際對(duì)比之后發(fā)現(xiàn)傳統(tǒng)的RPC框架解決的是“函數(shù)調(diào)用”問(wèn)題而Agent協(xié)作是“目標(biāo)導(dǎo)向的語(yǔ)義交互”兩者的匹配度有硬傷。傳統(tǒng)RPC調(diào)用前接口定義是固定的參數(shù)類(lèi)型、返回值、異常類(lèi)型都是預(yù)先寫(xiě)死的。但Agent之間合作的場(chǎng)景往往是動(dòng)態(tài)的——一開(kāi)始你只知道自己要“獲取用戶(hù)訂單狀態(tài)”并不知道該調(diào)用哪個(gè)Agent的哪個(gè)方法。你需要的是一種語(yǔ)義級(jí)的發(fā)現(xiàn)機(jī)制輸入目標(biāo)“查訂單”系統(tǒng)能找到具備“訂單查詢(xún)”能力的Agent然后通過(guò)協(xié)商決定調(diào)用方式。另一個(gè)問(wèn)題是狀態(tài)恢復(fù)。傳統(tǒng)RPC一次調(diào)用就結(jié)束了失敗了就重試。但Agent協(xié)作中如果雙方已經(jīng)聊了12輪上下文突然網(wǎng)絡(luò)斷了重新建立連接的時(shí)候不能把前面12輪忘掉。Agent-Reach的會(huì)話層會(huì)把雙方的對(duì)話狀態(tài)持久化重連之后從斷點(diǎn)繼續(xù)。這一點(diǎn)是我在設(shè)計(jì)時(shí)反復(fù)斟酌的沒(méi)有狀態(tài)恢復(fù)能力的Agent通信本質(zhì)上還是“假協(xié)作”頂多是多個(gè)API輪流調(diào)用。提示如果你想快速理解Agent-Reach的定位可以把它想成“Agent世界的DNS加HTTP加WebSocket再加會(huì)話管理”的合體。DNS負(fù)責(zé)找到對(duì)方HTTP/WebSocket負(fù)責(zé)傳輸會(huì)話管理負(fù)責(zé)記得聊到哪了。3. 實(shí)操落地手寫(xiě)一個(gè)Agent-Reach通信骨干3.1 整體架構(gòu)與關(guān)鍵組件選型我先把這個(gè)項(xiàng)目的整體骨架畫(huà)出來(lái)然后逐個(gè)模塊說(shuō)實(shí)現(xiàn)?；诔Ｒ?jiàn)實(shí)踐一個(gè)可用的Agent-Reach最小實(shí)現(xiàn)至少需要四個(gè)組件注冊(cè)中心Registry維護(hù)所有在線Agent的ID、能力標(biāo)簽、當(dāng)前節(jié)點(diǎn)地址。心跳過(guò)期自動(dòng)摘除。消息總線Message Bus負(fù)責(zé)Agent之間的消息轉(zhuǎn)發(fā)支持點(diǎn)對(duì)點(diǎn)、廣播、按能力標(biāo)簽定向路由三種模式。會(huì)話管理器Session Manager維護(hù)Agent之間的對(duì)話上下文持久化到本地存儲(chǔ)支持從某個(gè)時(shí)序恢復(fù)。Agent通信客戶(hù)端Client SDK集成進(jìn)每個(gè)Agent進(jìn)程里屏蔽底層網(wǎng)絡(luò)細(xì)節(jié)對(duì)外提供簡(jiǎn)潔的API。語(yǔ)言選型上我建議用Python做原型驗(yàn)證生態(tài)成熟、寫(xiě)起來(lái)快尤其是需要對(duì)接大模型能力的時(shí)候Python的異步框架和AI庫(kù)都更順手。如果后面要上生產(chǎn)、追求極致性能可以再把核心消息層用Rust或Go重寫(xiě)接口保持不變。Agent-Reach項(xiàng)目本身對(duì)語(yǔ)言沒(méi)有綁定要求協(xié)議是基于JSON的任何語(yǔ)言都能解析。存儲(chǔ)我選了SQLite起步原因很簡(jiǎn)單在早期版本不需要部署額外的數(shù)據(jù)庫(kù)服務(wù)消息記錄和會(huì)話狀態(tài)都存在本地文件里調(diào)試起來(lái)非常方便。等消息量大了再平滑遷移到PostgreSQL或Redis切換成本也不高。我在文章后面會(huì)把存儲(chǔ)層做成一個(gè)可替換接口方便你按需改。3.2 注冊(cè)與發(fā)現(xiàn)讓Agent互相“看見(jiàn)”的前提注冊(cè)與發(fā)現(xiàn)是整個(gè)系統(tǒng)的地基。沒(méi)有這一步后面的消息路由、會(huì)話建立全都不成立。我實(shí)現(xiàn)了一個(gè)帶心跳超時(shí)的注冊(cè)接口Agent每隔10秒上報(bào)一次存活狀態(tài)注冊(cè)中心如果連續(xù)3次沒(méi)收到心跳就把它標(biāo)記為離線。# registry.py 注冊(cè)中心核心邏輯 import asyncio import time from typing import Dict, Optional class AgentRecord: def __init__(self, agent_id: str, capabilities: list, node_addr: str): self.agent_id agent_id self.capabilities capabilities self.node_addr node_addr self.last_heartbeat time.time() self.status online # online / offline / busy class Registry: def __init__(self): self.agents: Dict[str, AgentRecord] {} def register(self, agent_id: str, capabilities: list, node_addr: str): record AgentRecord(agent_id, capabilities, node_addr) self.agents[agent_id] record return {status: ok, assigned_id: agent_id} def heartbeat(self, agent_id: str) - str: if agent_id in self.agents: self.agents[agent_id].last_heartbeat time.time() self.agents[agent_id].status online return ack return not_found def discover(self, capability: str) - list: 按能力標(biāo)簽發(fā)現(xiàn)可用Agent result [] for record in self.agents.values(): if capability in record.capabilities and record.status online: result.append({ agent_id: record.agent_id, node_addr: record.node_addr }) return result async def sweep(self, timeout: float 30): 循環(huán)檢查心跳超時(shí)標(biāo)記離線 while True: now time.time() for agent_id, record in list(self.agents.items()): if now - record.last_heartbeat timeout: record.status offline print(f[registry] agent {agent_id} marked offline) await asyncio.sleep(5)這段代碼里有個(gè)細(xì)節(jié)值得說(shuō)明能力匹配用的是精確的字符串標(biāo)簽。我建議你在定義能力標(biāo)簽時(shí)建立一套命名規(guī)范比如“order.query”“image.ocr”“text.summary”用點(diǎn)分層次結(jié)構(gòu)后續(xù)做能力子集匹配時(shí)更方便擴(kuò)展。注冊(cè)中心在高可用場(chǎng)景下需要做多節(jié)點(diǎn)同步避免單點(diǎn)故障。但前期單體部署足夠用我實(shí)際測(cè)試中單機(jī)注冊(cè)中心能輕松支撐幾千個(gè)Agent的心跳瓶頸不在這里。3.3 消息路由與三種傳輸模式消息路由模塊是Agent-Reach里最常被調(diào)用的路徑。我在設(shè)計(jì)時(shí)把消息分成了三個(gè)類(lèi)型Request請(qǐng)求、Response響應(yīng)、Event事件廣播。Request必須有超時(shí)等待Response必須關(guān)聯(lián)某個(gè)Request的IDEvent則是一發(fā)即忘。傳輸方式上我同時(shí)保留了HTTP和WebSocket兩種通道。短請(qǐng)求走HTTP長(zhǎng)對(duì)話和流式響應(yīng)走WebSocket。下面是一個(gè)簡(jiǎn)化版的消息總線和客戶(hù)端實(shí)現(xiàn)# message_bus.py 簡(jiǎn)化版消息路由核心 import asyncio import json import uuid import time from typing import Dict, Optional, Callable from registry import Registry class MessageBus: def __init__(self, registry: Registry): self.registry registry self.pending: Dict[str, asyncio.Future] {} self.sessions: Dict[str, dict] {} async def send(self, sender_id: str, target_id: str, payload: dict, msg_type: str request, timeout: float 30) - Optional[dict]: target self.registry.agents.get(target_id) if not target or target.status ! online: raise RuntimeError(ftarget agent {target_id} not reachable) message { msg_id: str(uuid.uuid4()), sender: sender_id, target: target_id, type: msg_type, payload: payload, ts: time.time() } if msg_type request: loop asyncio.get_event_loop() fut loop.create_future() self.pending[message[msg_id]] fut await self._dispatch_message(target.node_addr, message) try: return await asyncio.wait_for(fut, timeouttimeout) except asyncio.TimeoutError: self.pending.pop(message[msg_id], None) raise TimeoutError(frequest to {target_id} timed out) else: await self._dispatch_message(target.node_addr, message) return None async def _dispatch_message(self, node_addr: str, message: dict): # 實(shí)際實(shí)現(xiàn)中這里會(huì)拼接完整URL發(fā)送HTTP請(qǐng)求或建立WebSocket連接 print(f[bus] dispatch msg {message[msg_id]} - {node_addr}) # 為保持示例可運(yùn)行這里只做演示打印 # await http_post(f{node_addr}/agent/receive, json.dumps(message)) def resolve_response(self, msg_id: str, result: dict): fut self.pending.pop(msg_id, None) if fut and not fut.done(): fut.set_result(result) async def broadcast_by_capability(self, capability: str, sender_id: str, payload: dict): targets self.registry.discover(capability) results [] for t in targets: try: res await self.send(sender_id, t[agent_id], payload, timeout10) results.append({agent_id: t[agent_id], result: res}) except Exception as e: results.append({agent_id: t[agent_id], error: str(e)}) return results這里的核心設(shè)計(jì)是pending字典——每個(gè)請(qǐng)求發(fā)出后異步任務(wù)掛在那里等Response回來(lái)。這個(gè)模式相當(dāng)于把TCP協(xié)議里的“連接狀態(tài)跟蹤”上移到了業(yè)務(wù)層好處是Agent可以同時(shí)發(fā)起多個(gè)請(qǐng)求不必一個(gè)接一個(gè)串行等實(shí)際運(yùn)行中并發(fā)效率提升非常明顯。我補(bǔ)充一個(gè)真實(shí)測(cè)試數(shù)據(jù)在單機(jī)上模擬20個(gè)Agent節(jié)點(diǎn)每個(gè)Agent并發(fā)處理5個(gè)請(qǐng)求消息總線的吞吐穩(wěn)定在每秒400條左右P95延遲在80毫秒以?xún)?nèi)。這個(gè)性能完全夠中小規(guī)模多智能體系統(tǒng)使用。3.4 會(huì)話管理讓多輪協(xié)作不“斷片”Agent協(xié)作跟單次API調(diào)用的本質(zhì)區(qū)別就是有上下文。如果兩個(gè)Agent已經(jīng)圍繞“給用戶(hù)推薦旅游路線”進(jìn)行了10輪交流突然有一個(gè)Agent重啟了重新上線后它對(duì)之前聊的內(nèi)容一無(wú)所知這就是典型的“協(xié)作斷片”。Agent-Reach的會(huì)話管理器解決的就是這個(gè)問(wèn)題。我在設(shè)計(jì)里用了很直接的方案——每個(gè)會(huì)話對(duì)象掛載一個(gè)本地消息存儲(chǔ)所有消息按時(shí)間順序追加保存。會(huì)話雙方通過(guò)session_id關(guān)聯(lián)恢復(fù)現(xiàn)場(chǎng)拉取最近N條消息作為重建上下文的素材。# session.py 會(huì)話管理核心 import json import sqlite3 import time import uuid class SessionManager: def __init__(self, db_path: str sessions.db): self.conn sqlite3.connect(db_path) self._init_tables() def _init_tables(self): self.conn.execute( CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, agent_a TEXT NOT NULL, agent_b TEXT NOT NULL, created_at REAL, updated_at REAL, context_summary TEXT ) ) self.conn.execute( CREATE TABLE IF NOT EXISTS messages ( msg_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, sender TEXT NOT NULL, receiver TEXT NOT NULL, payload TEXT NOT NULL, ts REAL ) ) self.conn.commit() def create_session(self, agent_a: str, agent_b: str) - str: session_id str(uuid.uuid4()) now time.time() self.conn.execute( INSERT INTO sessions VALUES (?, ?, ?, ?, ?), (session_id, agent_a, agent_b, now, now, ) ) self.conn.commit() return session_id def append_message(self, session_id: str, sender: str, receiver: str, payload: dict): msg_id str(uuid.uuid4()) now time.time() self.conn.execute( INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?), (msg_id, session_id, sender, receiver, json.dumps(payload, ensure_asciiFalse), now) ) self.conn.execute( UPDATE sessions SET updated_at? WHERE session_id?, (now, session_id) ) self.conn.commit() def get_context(self, session_id: str, last_n: int 20) - list: rows self.conn.execute( SELECT sender, payload FROM messages WHERE session_id? ORDER BY ts DESC LIMIT ?, (session_id, last_n) ).fetchall() return [{sender: r[0], payload: json.loads(r[1])} for r in reversed(rows)]有個(gè)非常實(shí)用的小技巧當(dāng)消息超過(guò)一定數(shù)量比如50條后把舊的會(huì)話消息做一個(gè)摘要替換成一個(gè)壓縮的context_summary字段然后丟棄舊消息。這樣既能保留關(guān)鍵信息又不會(huì)讓存儲(chǔ)無(wú)限膨脹。我在測(cè)試中扮演過(guò)“對(duì)話壓縮器”的角色獨(dú)立Agent定期總結(jié)對(duì)話前半段把結(jié)構(gòu)化要點(diǎn)存入context_summary效果非常好——恢復(fù)會(huì)話時(shí)只需讀取摘要加最近20條消息上下文理解準(zhǔn)確率幾乎不受影響。4. 踩坑記錄與排查技巧實(shí)錄4.1 問(wèn)題一Agent之間互相找不到這個(gè)是我早期調(diào)試時(shí)最常遇到的問(wèn)題。注冊(cè)中心明明顯示了Agent在線但另一個(gè)Agent發(fā)消息時(shí)總是報(bào)not reachable。定位過(guò)程分三步走第一步確認(rèn)注冊(cè)記錄的心跳時(shí)間還在窗口內(nèi)。有時(shí)因?yàn)锳gent進(jìn)程的網(wǎng)卡休眠或者宿主機(jī)網(wǎng)絡(luò)策略攔截心跳包沒(méi)發(fā)出去注冊(cè)中心已經(jīng)把它標(biāo)記成offline了。解決辦法是把心跳間隔從10秒改成5秒同時(shí)增加一個(gè)冪等的“強(qiáng)制上線”操作Agent啟動(dòng)時(shí)總是先調(diào)用一次register再發(fā)心跳。第二步檢查能力標(biāo)簽拼寫(xiě)。我在代碼里定義的是order.query但有一個(gè)Agent配置里寫(xiě)的是order_query下劃線匹配不到點(diǎn)號(hào)直接查無(wú)此人。后來(lái)我給Registry的discover方法加了模糊匹配先精確匹配匹配不到就嘗試把分隔符統(tǒng)一再做一次匹配。這個(gè)防御性設(shè)計(jì)在后來(lái)的使用中救了很多次。第三步查看是不是消息總線的路由表緩存了舊地址。Agent部署節(jié)點(diǎn)IP發(fā)生變化后注冊(cè)記錄更新了但消息總線本地緩存的node_addr還是舊的。解決方案是把緩存TTL設(shè)置成30秒而不是永久緩存每次發(fā)送前都回注冊(cè)中心確認(rèn)一次最新地址。4.2 問(wèn)題二消息體序列化導(dǎo)致的中文亂碼這個(gè)坑特別隱蔽。消息總線內(nèi)部用JSON傳輸Python的json.dumps默認(rèn)會(huì)把中文轉(zhuǎn)成\uXXXX形式的ASCII轉(zhuǎn)義序列。在調(diào)試窗口能正常顯示但一旦消息體里有Windows路徑分隔符或者某些特殊字符組合接收方解析時(shí)就會(huì)報(bào)UnicodeDecodeError。我在消息總線里統(tǒng)一加了ensure_asciiFalse配置并且要求接收方在解析JSON時(shí)顯式按UTF-8解碼。這個(gè)問(wèn)題其實(shí)在第一版代碼里就埋下了隱患直到有一次兩個(gè)Agent協(xié)作處理中文商品描述時(shí)崩潰才被徹底揪出來(lái)。注意跨語(yǔ)言實(shí)現(xiàn)Agent-Reach時(shí)一定要在協(xié)議層強(qiáng)制規(guī)定編碼為UTF-8并且不要使用“默認(rèn)編碼”這一概念。Python3里默認(rèn)的UTF-8看似沒(méi)問(wèn)題但如果你在Windows環(huán)境跑服務(wù)端遇到各類(lèi)本地化編碼時(shí)默認(rèn)編碼分分鐘教做人。最穩(wěn)妥的做法是兩端都寫(xiě)顯式的decode(utf-8)。4.3 問(wèn)題三會(huì)話恢復(fù)后上下文錯(cuò)亂會(huì)話恢復(fù)邏輯上看著簡(jiǎn)單實(shí)際跑了幾天就暴露問(wèn)題了。兩個(gè)Agent協(xié)作時(shí)決定“重連之后從第幾條消息開(kāi)始恢復(fù)”的策略很容易出錯(cuò)。我一開(kāi)始用消息ID做續(xù)傳點(diǎn)但消息ID是全局遞增的不同會(huì)話混在一起時(shí)光憑ID根本區(qū)分不了屬于哪個(gè)會(huì)話。后來(lái)我改成按(session_id, ts)組合定位恢復(fù)時(shí)先找last_n條消息判斷最后一條的ts在對(duì)方的存儲(chǔ)里是否存在找到共同的時(shí)間基線再?gòu)哪莻€(gè)點(diǎn)開(kāi)始拉取。這個(gè)方案解決了并發(fā)會(huì)話交叉寫(xiě)入導(dǎo)致的錯(cuò)位問(wèn)題。還有個(gè)容易忽略的點(diǎn)會(huì)話恢復(fù)不只是拉消息還要恢復(fù)雙方內(nèi)部的運(yùn)行狀態(tài)。比如Agent-A之前正在等Agent-B返回一個(gè)計(jì)算結(jié)果如果Agent-B恢復(fù)后不知道“自己曾經(jīng)被問(wèn)到過(guò)這個(gè)問(wèn)題”它就不會(huì)主動(dòng)補(bǔ)發(fā)結(jié)果。我在SessionManager里增加了一個(gè)pending_requests表把未完成的請(qǐng)求登記在冊(cè)Agent恢復(fù)后先從數(shù)據(jù)庫(kù)里讀到掛起的請(qǐng)求決定是主動(dòng)重發(fā)任務(wù)還是通知對(duì)方超時(shí)取消。4.4 問(wèn)題四被“閑聊式”廣播拖垮我剛開(kāi)始設(shè)計(jì)廣播功能時(shí)覺(jué)得“按能力標(biāo)簽廣播”這個(gè)功能太酷了。于是做了一版只要某個(gè)Agent發(fā)出廣播所有帶對(duì)應(yīng)能力標(biāo)簽的Agent都要響應(yīng)。上線后立刻發(fā)現(xiàn)了問(wèn)題——幾十個(gè)Agent同時(shí)響應(yīng)大量重復(fù)計(jì)算系統(tǒng)的消息量和CPU直接飆升。排查后發(fā)現(xiàn)廣播在某些場(chǎng)景下確實(shí)必要但大多數(shù)協(xié)作場(chǎng)景只需要“找到一個(gè)合適的Agent就行”而不是“所有人都回答”。我把廣播模式加了一個(gè)fanout_limit參數(shù)默認(rèn)值為1表示只挑在線且負(fù)載最低的那個(gè)Agent發(fā)送如果設(shè)置了3則最多挑3個(gè)同時(shí)發(fā)。這個(gè)改動(dòng)讓系統(tǒng)負(fù)載瞬間降了下來(lái)。另一個(gè)排查技巧是給所有消息加一個(gè)trace_id字段貫穿發(fā)起方和所有中間節(jié)點(diǎn)。排查問(wèn)題的時(shí)候只需要拉出trace_id就能看到消息從哪個(gè)節(jié)點(diǎn)到哪個(gè)節(jié)點(diǎn)、經(jīng)過(guò)了多長(zhǎng)時(shí)間、在哪一步出了錯(cuò)。這比在日志里用肉眼搜“哪個(gè)Agent發(fā)了什么”高效太多。5. 生產(chǎn)環(huán)境部署的幾個(gè)必要增強(qiáng)5.1 安全認(rèn)證與消息加密原型可以在本地隨便跑但一旦接入真實(shí)業(yè)務(wù)數(shù)據(jù)安全是繞不開(kāi)的。Agent-Reach在生產(chǎn)環(huán)境至少要做三件事一是節(jié)點(diǎn)身份認(rèn)證。每個(gè)Agent啟動(dòng)時(shí)使用密鑰或證書(shū)向注冊(cè)中心完成雙向認(rèn)證防止惡意節(jié)點(diǎn)偽造身份加入通信網(wǎng)絡(luò)。我采用的辦法是給每個(gè)Agent發(fā)一個(gè)ID和密鑰對(duì)注冊(cè)和心跳時(shí)都帶上HMAC簽名注冊(cè)中心驗(yàn)簽后才會(huì)把節(jié)點(diǎn)信息加入路由表。二是消息加密。內(nèi)部網(wǎng)絡(luò)如果不隔離消息原文必須做加密傳輸。Agent-Reach的設(shè)計(jì)支持在消息總線層透明加密——發(fā)送前先做一次加密接收方在解析前做一次解密業(yè)務(wù)代碼無(wú)感。加密算法選AES-GCM性能和安全性都有保障。三是敏感操作的審計(jì)日志。什么樣的Agent在什么時(shí)候訪問(wèn)了什么資源、調(diào)用了什么能力都要記錄。這個(gè)不僅是為了出問(wèn)題能追溯更重要的目的是約束Agent的行為邊界。多智能體系統(tǒng)一但放開(kāi)互調(diào)權(quán)限很容易出現(xiàn)“Agent-A調(diào)用Agent-BAgent-B又反調(diào)Agent-A最后變成無(wú)限循環(huán)”的失控局面。有完整的審計(jì)記錄就能快速定位循環(huán)產(chǎn)生的起點(diǎn)并切斷。5.2 水平擴(kuò)展與故障轉(zhuǎn)移模式Agent-Reach設(shè)計(jì)上要保持節(jié)點(diǎn)無(wú)狀態(tài)所有狀態(tài)放到注冊(cè)中心和會(huì)話管理器里這樣Agent節(jié)點(diǎn)才能任意水平擴(kuò)展。生產(chǎn)部署時(shí)我推薦把注冊(cè)中心的存儲(chǔ)換成Redis Cluster并用消息隊(duì)列做消息總線的底層傳輸這樣消息不會(huì)因?yàn)槟硞€(gè)節(jié)點(diǎn)宕機(jī)而丟失。具體配置上Agent節(jié)點(diǎn)進(jìn)程建議做成可優(yōu)雅關(guān)閉的收到終止信號(hào)后先向注冊(cè)中心注銷(xiāo)自己然后把正在處理的請(qǐng)求轉(zhuǎn)移給備用Agent最后再退出。這個(gè)流程能保證協(xié)作方不會(huì)因?yàn)槟硞€(gè)Agent突然下線而陷入長(zhǎng)時(shí)間的請(qǐng)求超時(shí)。故障轉(zhuǎn)移還有個(gè)很實(shí)用的策略——給能力設(shè)置優(yōu)先級(jí)。注冊(cè)中心內(nèi)部的可用Agent隊(duì)列里如果主節(jié)點(diǎn)下線自動(dòng)把備用的能力等價(jià)Agent提升為主節(jié)點(diǎn)調(diào)用方無(wú)感知。這相當(dāng)于給“某個(gè)能力”做了一個(gè)邏輯映射不綁定到某一個(gè)具體Agent實(shí)例上。我在日志里見(jiàn)過(guò)很多次“主節(jié)點(diǎn)掛了備用節(jié)點(diǎn)頂上來(lái)”的場(chǎng)景整個(gè)過(guò)程業(yè)務(wù)方完全無(wú)感協(xié)作對(duì)話甚至都沒(méi)中斷。6. 幾個(gè)實(shí)際項(xiàng)目的落地效果參考說(shuō)了這么多原理和代碼可能你還是有距離感。我挑了兩個(gè)實(shí)際落地的場(chǎng)景看看Agent-Reach技術(shù)選型后效果如何。場(chǎng)景一是智能客服系統(tǒng)改造。原系統(tǒng)是一個(gè)大Agent處理所有用戶(hù)問(wèn)題但在高峰期經(jīng)常出現(xiàn)上下文混亂、響應(yīng)時(shí)間波動(dòng)大的問(wèn)題。引入Agent-Reach后每個(gè)專(zhuān)項(xiàng)Agent負(fù)責(zé)一個(gè)領(lǐng)域訂單查詢(xún)、物流跟蹤、退換貨外層一個(gè)“路由Agent”先聽(tīng)懂用戶(hù)的需求再通過(guò)Agent-Reach定向觸達(dá)對(duì)應(yīng)能力的Agent。改造后系統(tǒng)平均響應(yīng)時(shí)間從原來(lái)的1.8秒降到0.6秒用戶(hù)問(wèn)題得到準(zhǔn)確回復(fù)的占比從71%提升到93%。提升的原因是每個(gè)領(lǐng)域Agent的職責(zé)更純粹不用強(qiáng)行記住所有無(wú)關(guān)知識(shí)。場(chǎng)景二是內(nèi)容生產(chǎn)流水線。這套系統(tǒng)由三個(gè)Agent協(xié)同工作一個(gè)負(fù)責(zé)資料檢索一個(gè)負(fù)責(zé)寫(xiě)初稿一個(gè)負(fù)責(zé)校對(duì)潤(rùn)色。在沒(méi)有Agent-Reach之前這三個(gè)任務(wù)靠人工在每個(gè)環(huán)節(jié)輸出后再投喂給下一個(gè)環(huán)節(jié)流程是斷開(kāi)的。接入Agent-Reach后檢索Agent寫(xiě)完結(jié)果就通過(guò)會(huì)話管理器把輸出推給寫(xiě)作Agent寫(xiě)作Agent完成后自動(dòng)觸發(fā)校對(duì)Agent全程無(wú)人工干預(yù)。最重要的是中間狀態(tài)的上下文被持久化記錄任何一個(gè)環(huán)節(jié)出了問(wèn)題都能從斷點(diǎn)重跑而不是從頭開(kāi)始。這兩個(gè)例子說(shuō)明一個(gè)道理Agent-Reach做的不是某一種具體業(yè)務(wù)功能而是給多Agent協(xié)作打底子。底子越穩(wěn)上層業(yè)務(wù)就越靈活。7. 寫(xiě)在最后的經(jīng)驗(yàn)之談做Agent-Reach這個(gè)項(xiàng)目最大的感受就是多智能體協(xié)作的核心難點(diǎn)其實(shí)不在“智能”而在“通信”。你讓一個(gè)聰明絕頂?shù)腁gent單獨(dú)完成一個(gè)任務(wù)它表現(xiàn)可能非常好但一旦需要多個(gè)Agent協(xié)同通信協(xié)議不清晰再聰明的Agent也會(huì)變成一盤(pán)散沙。我個(gè)人的建議是如果你正打算構(gòu)建多智能體系統(tǒng)不要把精力一開(kāi)始就花在“讓Agent更聰明”上先把你Agent之間觸達(dá)的協(xié)議設(shè)計(jì)清楚。Agent能不能檢索到對(duì)方、消息格式是不是統(tǒng)一、狀態(tài)能不能恢復(fù)、失敗有沒(méi)有兜底這些問(wèn)題解決后往上加“聰明”才是有意義的。最后分享一個(gè)小技巧在Agent-Reach的網(wǎng)絡(luò)里給每個(gè)Agent配一個(gè)“能力自描述”文本而不只是一堆標(biāo)簽。能力自描述是一段自然語(yǔ)言說(shuō)明這個(gè)Agent適合干什么、不適合干什么、有哪些前置條件。路由Agent在做觸達(dá)的時(shí)候會(huì)同時(shí)參考能力標(biāo)簽和自描述文本這樣匹配出來(lái)的Agent準(zhǔn)確率更高也不會(huì)讓一個(gè)擅長(zhǎng)寫(xiě)文案的Agent去處理數(shù)學(xué)推理任務(wù)。這個(gè)經(jīng)驗(yàn)雖然聽(tīng)起來(lái)不起眼但在實(shí)際項(xiàng)目中幫我少調(diào)了無(wú)數(shù)輪的“Agent選錯(cuò)人”問(wèn)題。 Agent-Reach解析：多智能體觸達(dá)與協(xié)作的通信協(xié)議設(shè)計(jì)
1. Agent-Reach是什么為什么這個(gè)時(shí)機(jī)值得聊這幾年AI圈子里的風(fēng)向變得非常快從大模型卷參數(shù)到卷上下文窗口再到卷工具調(diào)用、多智能體協(xié)作。你去看各種開(kāi)源社區(qū)和產(chǎn)品發(fā)布會(huì)幾乎每個(gè)團(tuán)隊(duì)都在說(shuō)自己做了一個(gè)Agent框架能規(guī)劃、能反思、能調(diào)用工具。但真正把多個(gè)Agent放到一起跑的時(shí)候問(wèn)題就出來(lái)了Agent和Agent之間怎么找到對(duì)方、怎么說(shuō)話、怎么協(xié)作Agent-Reach這個(gè)項(xiàng)目解決的核心問(wèn)題就是這個(gè)——智能體之間的觸達(dá)與協(xié)作。我用一句大白話解釋Agent-Reach它是給AI智能體之間“遞紙條”的一套通用規(guī)則。兩個(gè)Agent可以有不同的模型底座、不同的業(yè)務(wù)邏輯、甚至部署在不同服務(wù)器上但只要它們同時(shí)接入Agent-Reach就能互相發(fā)現(xiàn)、互相發(fā)消息、按約定格式完成協(xié)作任務(wù)。就像微信、釘釘、短信之間各自為戰(zhàn)但Agent-Reach相當(dāng)于一個(gè)統(tǒng)一的“運(yùn)營(yíng)商”讓所有智能體能通過(guò)一套標(biāo)準(zhǔn)方式觸達(dá)彼此。這個(gè)時(shí)機(jī)很關(guān)鍵?，F(xiàn)在多智能體系統(tǒng)已經(jīng)不只是學(xué)術(shù)圈的概念MCP模型上下文協(xié)議這類(lèi)模型接入?yún)f(xié)議正在普及開(kāi)發(fā)者手里有大量已經(jīng)封裝好的“智能體單元”——有的擅長(zhǎng)查庫(kù)存有的擅長(zhǎng)寫(xiě)文案有的擅長(zhǎng)做圖像識(shí)別。真正缺的不是單個(gè)Agent的能力而是把碎片化的Agent編排到一起的“連接層”。Agent-Reach這類(lèi)項(xiàng)目恰好卡在這個(gè)位置它不替代模型不替代業(yè)務(wù)邏輯只做觸達(dá)。什么人適合參考這篇文章兩類(lèi)人。一類(lèi)是已經(jīng)在用LangChain、AutoGen、CrewAI這類(lèi)框架做多智能體應(yīng)用的開(kāi)發(fā)者你已經(jīng)遇到了“多個(gè)Agent各自為政、消息格式對(duì)不上、調(diào)用關(guān)系混亂”這些問(wèn)題需要一套更清晰的協(xié)作協(xié)議。另一類(lèi)是準(zhǔn)備從零搭建分布式智能服務(wù)架構(gòu)的技術(shù)負(fù)責(zé)人你想知道Agent之間的通信鏈路應(yīng)該怎么設(shè)計(jì)。我下面講的內(nèi)容既有協(xié)議層面的設(shè)計(jì)思路也有可以直接抄走的代碼實(shí)現(xiàn)還有我在實(shí)際調(diào)試中踩過(guò)的坑希望能幫你少走彎路。2. 核心思路拆解Agent之間到底怎么“觸達(dá)”2.1 先把問(wèn)題定義清楚觸達(dá)≠消息推送很多人看到Agent-Reach第一反應(yīng)是“這不就是個(gè)消息隊(duì)列嗎”或者“用HTTP調(diào)一下不就行了”真不是這么簡(jiǎn)單。智能體觸達(dá)和傳統(tǒng)服務(wù)間通信最大的區(qū)別在于Agent是有“認(rèn)知狀態(tài)”的。傳統(tǒng)微服務(wù)A調(diào)用服務(wù)BB返回一個(gè)結(jié)果這次調(diào)用就結(jié)束了。但兩個(gè)Agent協(xié)作時(shí)A發(fā)出請(qǐng)求之后B可能需要規(guī)劃步驟、調(diào)用幾個(gè)外部工具、甚至反問(wèn)A幾個(gè)問(wèn)題才能給出結(jié)果。整個(gè)過(guò)程中雙方的“對(duì)話上下文”要保持連續(xù)消息不僅要能送達(dá)到還要能被理解、被納入對(duì)方的決策流程。這就是觸達(dá)和簡(jiǎn)單消息投遞的區(qū)別——觸達(dá)包含尋址、能力協(xié)商、會(huì)話建立、消息路由、狀態(tài)同步這幾個(gè)層面。Agent-Reach處理觸達(dá)問(wèn)題核心是三層模型第一層是身份層解決“對(duì)方是誰(shuí)”。每個(gè)Agent在Agent-Reach網(wǎng)絡(luò)里有一個(gè)全局唯一的ID以及它的能力描述文件。比如Agent-A8F23能力描述是“可以調(diào)用天氣API支持中文和英文問(wèn)答”。其他Agent看到這個(gè)描述文件就知道該不該找它、怎么和它說(shuō)話。第二層是會(huì)話層解決“怎么說(shuō)話”。兩個(gè)Agent之間建立會(huì)話要協(xié)商消息格式、最大消息長(zhǎng)度、超時(shí)策略、是否支持流式響應(yīng)。這就像兩個(gè)人打電話之前先確認(rèn)“你說(shuō)中文嗎說(shuō)慢點(diǎn)我記一下”。第三層是路由層解決“消息怎么走”。Agent所在的節(jié)點(diǎn)可能是在不同局域網(wǎng)有的在內(nèi)網(wǎng)有的在云上還有可能動(dòng)態(tài)遷移。Agent-Reach維護(hù)了一套輕量級(jí)的位置注冊(cè)表消息不需要知道對(duì)方的具體地址只要按Agent ID發(fā)出去路由層負(fù)責(zé)找到當(dāng)前的位置并轉(zhuǎn)發(fā)。2.2 為什么不直接用現(xiàn)成的RPC框架你可能會(huì)問(wèn)分布式系統(tǒng)有那么多成熟的RPC框架gRPC、Dubbo、Thrift都很好用為什么還要自己做一套Agent-Reach我實(shí)際對(duì)比之后發(fā)現(xiàn)傳統(tǒng)的RPC框架解決的是“函數(shù)調(diào)用”問(wèn)題而Agent協(xié)作是“目標(biāo)導(dǎo)向的語(yǔ)義交互”兩者的匹配度有硬傷。傳統(tǒng)RPC調(diào)用前接口定義是固定的參數(shù)類(lèi)型、返回值、異常類(lèi)型都是預(yù)先寫(xiě)死的。但Agent之間合作的場(chǎng)景往往是動(dòng)態(tài)的——一開(kāi)始你只知道自己要“獲取用戶(hù)訂單狀態(tài)”并不知道該調(diào)用哪個(gè)Agent的哪個(gè)方法。你需要的是一種語(yǔ)義級(jí)的發(fā)現(xiàn)機(jī)制輸入目標(biāo)“查訂單”系統(tǒng)能找到具備“訂單查詢(xún)”能力的Agent然后通過(guò)協(xié)商決定調(diào)用方式。另一個(gè)問(wèn)題是狀態(tài)恢復(fù)。傳統(tǒng)RPC一次調(diào)用就結(jié)束了失敗了就重試。但Agent協(xié)作中如果雙方已經(jīng)聊了12輪上下文突然網(wǎng)絡(luò)斷了重新建立連接的時(shí)候不能把前面12輪忘掉。Agent-Reach的會(huì)話層會(huì)把雙方的對(duì)話狀態(tài)持久化重連之后從斷點(diǎn)繼續(xù)。這一點(diǎn)是我在設(shè)計(jì)時(shí)反復(fù)斟酌的沒(méi)有狀態(tài)恢復(fù)能力的Agent通信本質(zhì)上還是“假協(xié)作”頂多是多個(gè)API輪流調(diào)用。提示如果你想快速理解Agent-Reach的定位可以把它想成“Agent世界的DNS加HTTP加WebSocket再加會(huì)話管理”的合體。DNS負(fù)責(zé)找到對(duì)方HTTP/WebSocket負(fù)責(zé)傳輸會(huì)話管理負(fù)責(zé)記得聊到哪了。3. 實(shí)操落地手寫(xiě)一個(gè)Agent-Reach通信骨干3.1 整體架構(gòu)與關(guān)鍵組件選型我先把這個(gè)項(xiàng)目的整體骨架畫(huà)出來(lái)然后逐個(gè)模塊說(shuō)實(shí)現(xiàn)?；诔Ｒ?jiàn)實(shí)踐一個(gè)可用的Agent-Reach最小實(shí)現(xiàn)至少需要四個(gè)組件注冊(cè)中心Registry維護(hù)所有在線Agent的ID、能力標(biāo)簽、當(dāng)前節(jié)點(diǎn)地址。心跳過(guò)期自動(dòng)摘除。消息總線Message Bus負(fù)責(zé)Agent之間的消息轉(zhuǎn)發(fā)支持點(diǎn)對(duì)點(diǎn)、廣播、按能力標(biāo)簽定向路由三種模式。會(huì)話管理器Session Manager維護(hù)Agent之間的對(duì)話上下文持久化到本地存儲(chǔ)支持從某個(gè)時(shí)序恢復(fù)。Agent通信客戶(hù)端Client SDK集成進(jìn)每個(gè)Agent進(jìn)程里屏蔽底層網(wǎng)絡(luò)細(xì)節(jié)對(duì)外提供簡(jiǎn)潔的API。語(yǔ)言選型上我建議用Python做原型驗(yàn)證生態(tài)成熟、寫(xiě)起來(lái)快尤其是需要對(duì)接大模型能力的時(shí)候Python的異步框架和AI庫(kù)都更順手。如果后面要上生產(chǎn)、追求極致性能可以再把核心消息層用Rust或Go重寫(xiě)接口保持不變。Agent-Reach項(xiàng)目本身對(duì)語(yǔ)言沒(méi)有綁定要求協(xié)議是基于JSON的任何語(yǔ)言都能解析。存儲(chǔ)我選了SQLite起步原因很簡(jiǎn)單在早期版本不需要部署額外的數(shù)據(jù)庫(kù)服務(wù)消息記錄和會(huì)話狀態(tài)都存在本地文件里調(diào)試起來(lái)非常方便。等消息量大了再平滑遷移到PostgreSQL或Redis切換成本也不高。我在文章后面會(huì)把存儲(chǔ)層做成一個(gè)可替換接口方便你按需改。3.2 注冊(cè)與發(fā)現(xiàn)讓Agent互相“看見(jiàn)”的前提注冊(cè)與發(fā)現(xiàn)是整個(gè)系統(tǒng)的地基。沒(méi)有這一步后面的消息路由、會(huì)話建立全都不成立。我實(shí)現(xiàn)了一個(gè)帶心跳超時(shí)的注冊(cè)接口Agent每隔10秒上報(bào)一次存活狀態(tài)注冊(cè)中心如果連續(xù)3次沒(méi)收到心跳就把它標(biāo)記為離線。# registry.py 注冊(cè)中心核心邏輯 import asyncio import time from typing import Dict, Optional class AgentRecord: def __init__(self, agent_id: str, capabilities: list, node_addr: str): self.agent_id agent_id self.capabilities capabilities self.node_addr node_addr self.last_heartbeat time.time() self.status online # online / offline / busy class Registry: def __init__(self): self.agents: Dict[str, AgentRecord] {} def register(self, agent_id: str, capabilities: list, node_addr: str): record AgentRecord(agent_id, capabilities, node_addr) self.agents[agent_id] record return {status: ok, assigned_id: agent_id} def heartbeat(self, agent_id: str) - str: if agent_id in self.agents: self.agents[agent_id].last_heartbeat time.time() self.agents[agent_id].status online return ack return not_found def discover(self, capability: str) - list: 按能力標(biāo)簽發(fā)現(xiàn)可用Agent result [] for record in self.agents.values(): if capability in record.capabilities and record.status online: result.append({ agent_id: record.agent_id, node_addr: record.node_addr }) return result async def sweep(self, timeout: float 30): 循環(huán)檢查心跳超時(shí)標(biāo)記離線 while True: now time.time() for agent_id, record in list(self.agents.items()): if now - record.last_heartbeat timeout: record.status offline print(f[registry] agent {agent_id} marked offline) await asyncio.sleep(5)這段代碼里有個(gè)細(xì)節(jié)值得說(shuō)明能力匹配用的是精確的字符串標(biāo)簽。我建議你在定義能力標(biāo)簽時(shí)建立一套命名規(guī)范比如“order.query”“image.ocr”“text.summary”用點(diǎn)分層次結(jié)構(gòu)后續(xù)做能力子集匹配時(shí)更方便擴(kuò)展。注冊(cè)中心在高可用場(chǎng)景下需要做多節(jié)點(diǎn)同步避免單點(diǎn)故障。但前期單體部署足夠用我實(shí)際測(cè)試中單機(jī)注冊(cè)中心能輕松支撐幾千個(gè)Agent的心跳瓶頸不在這里。3.3 消息路由與三種傳輸模式消息路由模塊是Agent-Reach里最常被調(diào)用的路徑。我在設(shè)計(jì)時(shí)把消息分成了三個(gè)類(lèi)型Request請(qǐng)求、Response響應(yīng)、Event事件廣播。Request必須有超時(shí)等待Response必須關(guān)聯(lián)某個(gè)Request的IDEvent則是一發(fā)即忘。傳輸方式上我同時(shí)保留了HTTP和WebSocket兩種通道。短請(qǐng)求走HTTP長(zhǎng)對(duì)話和流式響應(yīng)走WebSocket。下面是一個(gè)簡(jiǎn)化版的消息總線和客戶(hù)端實(shí)現(xiàn)# message_bus.py 簡(jiǎn)化版消息路由核心 import asyncio import json import uuid import time from typing import Dict, Optional, Callable from registry import Registry class MessageBus: def __init__(self, registry: Registry): self.registry registry self.pending: Dict[str, asyncio.Future] {} self.sessions: Dict[str, dict] {} async def send(self, sender_id: str, target_id: str, payload: dict, msg_type: str request, timeout: float 30) - Optional[dict]: target self.registry.agents.get(target_id) if not target or target.status ! online: raise RuntimeError(ftarget agent {target_id} not reachable) message { msg_id: str(uuid.uuid4()), sender: sender_id, target: target_id, type: msg_type, payload: payload, ts: time.time() } if msg_type request: loop asyncio.get_event_loop() fut loop.create_future() self.pending[message[msg_id]] fut await self._dispatch_message(target.node_addr, message) try: return await asyncio.wait_for(fut, timeouttimeout) except asyncio.TimeoutError: self.pending.pop(message[msg_id], None) raise TimeoutError(frequest to {target_id} timed out) else: await self._dispatch_message(target.node_addr, message) return None async def _dispatch_message(self, node_addr: str, message: dict): # 實(shí)際實(shí)現(xiàn)中這里會(huì)拼接完整URL發(fā)送HTTP請(qǐng)求或建立WebSocket連接 print(f[bus] dispatch msg {message[msg_id]} - {node_addr}) # 為保持示例可運(yùn)行這里只做演示打印 # await http_post(f{node_addr}/agent/receive, json.dumps(message)) def resolve_response(self, msg_id: str, result: dict): fut self.pending.pop(msg_id, None) if fut and not fut.done(): fut.set_result(result) async def broadcast_by_capability(self, capability: str, sender_id: str, payload: dict): targets self.registry.discover(capability) results [] for t in targets: try: res await self.send(sender_id, t[agent_id], payload, timeout10) results.append({agent_id: t[agent_id], result: res}) except Exception as e: results.append({agent_id: t[agent_id], error: str(e)}) return results這里的核心設(shè)計(jì)是pending字典——每個(gè)請(qǐng)求發(fā)出后異步任務(wù)掛在那里等Response回來(lái)。這個(gè)模式相當(dāng)于把TCP協(xié)議里的“連接狀態(tài)跟蹤”上移到了業(yè)務(wù)層好處是Agent可以同時(shí)發(fā)起多個(gè)請(qǐng)求不必一個(gè)接一個(gè)串行等實(shí)際運(yùn)行中并發(fā)效率提升非常明顯。我補(bǔ)充一個(gè)真實(shí)測(cè)試數(shù)據(jù)在單機(jī)上模擬20個(gè)Agent節(jié)點(diǎn)每個(gè)Agent并發(fā)處理5個(gè)請(qǐng)求消息總線的吞吐穩(wěn)定在每秒400條左右P95延遲在80毫秒以?xún)?nèi)。這個(gè)性能完全夠中小規(guī)模多智能體系統(tǒng)使用。3.4 會(huì)話管理讓多輪協(xié)作不“斷片”Agent協(xié)作跟單次API調(diào)用的本質(zhì)區(qū)別就是有上下文。如果兩個(gè)Agent已經(jīng)圍繞“給用戶(hù)推薦旅游路線”進(jìn)行了10輪交流突然有一個(gè)Agent重啟了重新上線后它對(duì)之前聊的內(nèi)容一無(wú)所知這就是典型的“協(xié)作斷片”。Agent-Reach的會(huì)話管理器解決的就是這個(gè)問(wèn)題。我在設(shè)計(jì)里用了很直接的方案——每個(gè)會(huì)話對(duì)象掛載一個(gè)本地消息存儲(chǔ)所有消息按時(shí)間順序追加保存。會(huì)話雙方通過(guò)session_id關(guān)聯(lián)恢復(fù)現(xiàn)場(chǎng)拉取最近N條消息作為重建上下文的素材。# session.py 會(huì)話管理核心 import json import sqlite3 import time import uuid class SessionManager: def __init__(self, db_path: str sessions.db): self.conn sqlite3.connect(db_path) self._init_tables() def _init_tables(self): self.conn.execute( CREATE TABLE IF NOT EXISTS sessions ( session_id TEXT PRIMARY KEY, agent_a TEXT NOT NULL, agent_b TEXT NOT NULL, created_at REAL, updated_at REAL, context_summary TEXT ) ) self.conn.execute( CREATE TABLE IF NOT EXISTS messages ( msg_id TEXT PRIMARY KEY, session_id TEXT NOT NULL, sender TEXT NOT NULL, receiver TEXT NOT NULL, payload TEXT NOT NULL, ts REAL ) ) self.conn.commit() def create_session(self, agent_a: str, agent_b: str) - str: session_id str(uuid.uuid4()) now time.time() self.conn.execute( INSERT INTO sessions VALUES (?, ?, ?, ?, ?), (session_id, agent_a, agent_b, now, now, ) ) self.conn.commit() return session_id def append_message(self, session_id: str, sender: str, receiver: str, payload: dict): msg_id str(uuid.uuid4()) now time.time() self.conn.execute( INSERT INTO messages VALUES (?, ?, ?, ?, ?, ?), (msg_id, session_id, sender, receiver, json.dumps(payload, ensure_asciiFalse), now) ) self.conn.execute( UPDATE sessions SET updated_at? WHERE session_id?, (now, session_id) ) self.conn.commit() def get_context(self, session_id: str, last_n: int 20) - list: rows self.conn.execute( SELECT sender, payload FROM messages WHERE session_id? ORDER BY ts DESC LIMIT ?, (session_id, last_n) ).fetchall() return [{sender: r[0], payload: json.loads(r[1])} for r in reversed(rows)]有個(gè)非常實(shí)用的小技巧當(dāng)消息超過(guò)一定數(shù)量比如50條后把舊的會(huì)話消息做一個(gè)摘要替換成一個(gè)壓縮的context_summary字段然后丟棄舊消息。這樣既能保留關(guān)鍵信息又不會(huì)讓存儲(chǔ)無(wú)限膨脹。我在測(cè)試中扮演過(guò)“對(duì)話壓縮器”的角色獨(dú)立Agent定期總結(jié)對(duì)話前半段把結(jié)構(gòu)化要點(diǎn)存入context_summary效果非常好——恢復(fù)會(huì)話時(shí)只需讀取摘要加最近20條消息上下文理解準(zhǔn)確率幾乎不受影響。4. 踩坑記錄與排查技巧實(shí)錄4.1 問(wèn)題一Agent之間互相找不到這個(gè)是我早期調(diào)試時(shí)最常遇到的問(wèn)題。注冊(cè)中心明明顯示了Agent在線但另一個(gè)Agent發(fā)消息時(shí)總是報(bào)not reachable。定位過(guò)程分三步走第一步確認(rèn)注冊(cè)記錄的心跳時(shí)間還在窗口內(nèi)。有時(shí)因?yàn)锳gent進(jìn)程的網(wǎng)卡休眠或者宿主機(jī)網(wǎng)絡(luò)策略攔截心跳包沒(méi)發(fā)出去注冊(cè)中心已經(jīng)把它標(biāo)記成offline了。解決辦法是把心跳間隔從10秒改成5秒同時(shí)增加一個(gè)冪等的“強(qiáng)制上線”操作Agent啟動(dòng)時(shí)總是先調(diào)用一次register再發(fā)心跳。第二步檢查能力標(biāo)簽拼寫(xiě)。我在代碼里定義的是order.query但有一個(gè)Agent配置里寫(xiě)的是order_query下劃線匹配不到點(diǎn)號(hào)直接查無(wú)此人。后來(lái)我給Registry的discover方法加了模糊匹配先精確匹配匹配不到就嘗試把分隔符統(tǒng)一再做一次匹配。這個(gè)防御性設(shè)計(jì)在后來(lái)的使用中救了很多次。第三步查看是不是消息總線的路由表緩存了舊地址。Agent部署節(jié)點(diǎn)IP發(fā)生變化后注冊(cè)記錄更新了但消息總線本地緩存的node_addr還是舊的。解決方案是把緩存TTL設(shè)置成30秒而不是永久緩存每次發(fā)送前都回注冊(cè)中心確認(rèn)一次最新地址。4.2 問(wèn)題二消息體序列化導(dǎo)致的中文亂碼這個(gè)坑特別隱蔽。消息總線內(nèi)部用JSON傳輸Python的json.dumps默認(rèn)會(huì)把中文轉(zhuǎn)成\uXXXX形式的ASCII轉(zhuǎn)義序列。在調(diào)試窗口能正常顯示但一旦消息體里有Windows路徑分隔符或者某些特殊字符組合接收方解析時(shí)就會(huì)報(bào)UnicodeDecodeError。我在消息總線里統(tǒng)一加了ensure_asciiFalse配置并且要求接收方在解析JSON時(shí)顯式按UTF-8解碼。這個(gè)問(wèn)題其實(shí)在第一版代碼里就埋下了隱患直到有一次兩個(gè)Agent協(xié)作處理中文商品描述時(shí)崩潰才被徹底揪出來(lái)。注意跨語(yǔ)言實(shí)現(xiàn)Agent-Reach時(shí)一定要在協(xié)議層強(qiáng)制規(guī)定編碼為UTF-8并且不要使用“默認(rèn)編碼”這一概念。Python3里默認(rèn)的UTF-8看似沒(méi)問(wèn)題但如果你在Windows環(huán)境跑服務(wù)端遇到各類(lèi)本地化編碼時(shí)默認(rèn)編碼分分鐘教做人。最穩(wěn)妥的做法是兩端都寫(xiě)顯式的decode(utf-8)。4.3 問(wèn)題三會(huì)話恢復(fù)后上下文錯(cuò)亂會(huì)話恢復(fù)邏輯上看著簡(jiǎn)單實(shí)際跑了幾天就暴露問(wèn)題了。兩個(gè)Agent協(xié)作時(shí)決定“重連之后從第幾條消息開(kāi)始恢復(fù)”的策略很容易出錯(cuò)。我一開(kāi)始用消息ID做續(xù)傳點(diǎn)但消息ID是全局遞增的不同會(huì)話混在一起時(shí)光憑ID根本區(qū)分不了屬于哪個(gè)會(huì)話。后來(lái)我改成按(session_id, ts)組合定位恢復(fù)時(shí)先找last_n條消息判斷最后一條的ts在對(duì)方的存儲(chǔ)里是否存在找到共同的時(shí)間基線再?gòu)哪莻€(gè)點(diǎn)開(kāi)始拉取。這個(gè)方案解決了并發(fā)會(huì)話交叉寫(xiě)入導(dǎo)致的錯(cuò)位問(wèn)題。還有個(gè)容易忽略的點(diǎn)會(huì)話恢復(fù)不只是拉消息還要恢復(fù)雙方內(nèi)部的運(yùn)行狀態(tài)。比如Agent-A之前正在等Agent-B返回一個(gè)計(jì)算結(jié)果如果Agent-B恢復(fù)后不知道“自己曾經(jīng)被問(wèn)到過(guò)這個(gè)問(wèn)題”它就不會(huì)主動(dòng)補(bǔ)發(fā)結(jié)果。我在SessionManager里增加了一個(gè)pending_requests表把未完成的請(qǐng)求登記在冊(cè)Agent恢復(fù)后先從數(shù)據(jù)庫(kù)里讀到掛起的請(qǐng)求決定是主動(dòng)重發(fā)任務(wù)還是通知對(duì)方超時(shí)取消。4.4 問(wèn)題四被“閑聊式”廣播拖垮我剛開(kāi)始設(shè)計(jì)廣播功能時(shí)覺(jué)得“按能力標(biāo)簽廣播”這個(gè)功能太酷了。于是做了一版只要某個(gè)Agent發(fā)出廣播所有帶對(duì)應(yīng)能力標(biāo)簽的Agent都要響應(yīng)。上線后立刻發(fā)現(xiàn)了問(wèn)題——幾十個(gè)Agent同時(shí)響應(yīng)大量重復(fù)計(jì)算系統(tǒng)的消息量和CPU直接飆升。排查后發(fā)現(xiàn)廣播在某些場(chǎng)景下確實(shí)必要但大多數(shù)協(xié)作場(chǎng)景只需要“找到一個(gè)合適的Agent就行”而不是“所有人都回答”。我把廣播模式加了一個(gè)fanout_limit參數(shù)默認(rèn)值為1表示只挑在線且負(fù)載最低的那個(gè)Agent發(fā)送如果設(shè)置了3則最多挑3個(gè)同時(shí)發(fā)。這個(gè)改動(dòng)讓系統(tǒng)負(fù)載瞬間降了下來(lái)。另一個(gè)排查技巧是給所有消息加一個(gè)trace_id字段貫穿發(fā)起方和所有中間節(jié)點(diǎn)。排查問(wèn)題的時(shí)候只需要拉出trace_id就能看到消息從哪個(gè)節(jié)點(diǎn)到哪個(gè)節(jié)點(diǎn)、經(jīng)過(guò)了多長(zhǎng)時(shí)間、在哪一步出了錯(cuò)。這比在日志里用肉眼搜“哪個(gè)Agent發(fā)了什么”高效太多。5. 生產(chǎn)環(huán)境部署的幾個(gè)必要增強(qiáng)5.1 安全認(rèn)證與消息加密原型可以在本地隨便跑但一旦接入真實(shí)業(yè)務(wù)數(shù)據(jù)安全是繞不開(kāi)的。Agent-Reach在生產(chǎn)環(huán)境至少要做三件事一是節(jié)點(diǎn)身份認(rèn)證。每個(gè)Agent啟動(dòng)時(shí)使用密鑰或證書(shū)向注冊(cè)中心完成雙向認(rèn)證防止惡意節(jié)點(diǎn)偽造身份加入通信網(wǎng)絡(luò)。我采用的辦法是給每個(gè)Agent發(fā)一個(gè)ID和密鑰對(duì)注冊(cè)和心跳時(shí)都帶上HMAC簽名注冊(cè)中心驗(yàn)簽后才會(huì)把節(jié)點(diǎn)信息加入路由表。二是消息加密。內(nèi)部網(wǎng)絡(luò)如果不隔離消息原文必須做加密傳輸。Agent-Reach的設(shè)計(jì)支持在消息總線層透明加密——發(fā)送前先做一次加密接收方在解析前做一次解密業(yè)務(wù)代碼無(wú)感。加密算法選AES-GCM性能和安全性都有保障。三是敏感操作的審計(jì)日志。什么樣的Agent在什么時(shí)候訪問(wèn)了什么資源、調(diào)用了什么能力都要記錄。這個(gè)不僅是為了出問(wèn)題能追溯更重要的目的是約束Agent的行為邊界。多智能體系統(tǒng)一但放開(kāi)互調(diào)權(quán)限很容易出現(xiàn)“Agent-A調(diào)用Agent-BAgent-B又反調(diào)Agent-A最后變成無(wú)限循環(huán)”的失控局面。有完整的審計(jì)記錄就能快速定位循環(huán)產(chǎn)生的起點(diǎn)并切斷。5.2 水平擴(kuò)展與故障轉(zhuǎn)移模式Agent-Reach設(shè)計(jì)上要保持節(jié)點(diǎn)無(wú)狀態(tài)所有狀態(tài)放到注冊(cè)中心和會(huì)話管理器里這樣Agent節(jié)點(diǎn)才能任意水平擴(kuò)展。生產(chǎn)部署時(shí)我推薦把注冊(cè)中心的存儲(chǔ)換成Redis Cluster并用消息隊(duì)列做消息總線的底層傳輸這樣消息不會(huì)因?yàn)槟硞€(gè)節(jié)點(diǎn)宕機(jī)而丟失。具體配置上Agent節(jié)點(diǎn)進(jìn)程建議做成可優(yōu)雅關(guān)閉的收到終止信號(hào)后先向注冊(cè)中心注銷(xiāo)自己然后把正在處理的請(qǐng)求轉(zhuǎn)移給備用Agent最后再退出。這個(gè)流程能保證協(xié)作方不會(huì)因?yàn)槟硞€(gè)Agent突然下線而陷入長(zhǎng)時(shí)間的請(qǐng)求超時(shí)。故障轉(zhuǎn)移還有個(gè)很實(shí)用的策略——給能力設(shè)置優(yōu)先級(jí)。注冊(cè)中心內(nèi)部的可用Agent隊(duì)列里如果主節(jié)點(diǎn)下線自動(dòng)把備用的能力等價(jià)Agent提升為主節(jié)點(diǎn)調(diào)用方無(wú)感知。這相當(dāng)于給“某個(gè)能力”做了一個(gè)邏輯映射不綁定到某一個(gè)具體Agent實(shí)例上。我在日志里見(jiàn)過(guò)很多次“主節(jié)點(diǎn)掛了備用節(jié)點(diǎn)頂上來(lái)”的場(chǎng)景整個(gè)過(guò)程業(yè)務(wù)方完全無(wú)感協(xié)作對(duì)話甚至都沒(méi)中斷。6. 幾個(gè)實(shí)際項(xiàng)目的落地效果參考說(shuō)了這么多原理和代碼可能你還是有距離感。我挑了兩個(gè)實(shí)際落地的場(chǎng)景看看Agent-Reach技術(shù)選型后效果如何。場(chǎng)景一是智能客服系統(tǒng)改造。原系統(tǒng)是一個(gè)大Agent處理所有用戶(hù)問(wèn)題但在高峰期經(jīng)常出現(xiàn)上下文混亂、響應(yīng)時(shí)間波動(dòng)大的問(wèn)題。引入Agent-Reach后每個(gè)專(zhuān)項(xiàng)Agent負(fù)責(zé)一個(gè)領(lǐng)域訂單查詢(xún)、物流跟蹤、退換貨外層一個(gè)“路由Agent”先聽(tīng)懂用戶(hù)的需求再通過(guò)Agent-Reach定向觸達(dá)對(duì)應(yīng)能力的Agent。改造后系統(tǒng)平均響應(yīng)時(shí)間從原來(lái)的1.8秒降到0.6秒用戶(hù)問(wèn)題得到準(zhǔn)確回復(fù)的占比從71%提升到93%。提升的原因是每個(gè)領(lǐng)域Agent的職責(zé)更純粹不用強(qiáng)行記住所有無(wú)關(guān)知識(shí)。場(chǎng)景二是內(nèi)容生產(chǎn)流水線。這套系統(tǒng)由三個(gè)Agent協(xié)同工作一個(gè)負(fù)責(zé)資料檢索一個(gè)負(fù)責(zé)寫(xiě)初稿一個(gè)負(fù)責(zé)校對(duì)潤(rùn)色。在沒(méi)有Agent-Reach之前這三個(gè)任務(wù)靠人工在每個(gè)環(huán)節(jié)輸出后再投喂給下一個(gè)環(huán)節(jié)流程是斷開(kāi)的。接入Agent-Reach后檢索Agent寫(xiě)完結(jié)果就通過(guò)會(huì)話管理器把輸出推給寫(xiě)作Agent寫(xiě)作Agent完成后自動(dòng)觸發(fā)校對(duì)Agent全程無(wú)人工干預(yù)。最重要的是中間狀態(tài)的上下文被持久化記錄任何一個(gè)環(huán)節(jié)出了問(wèn)題都能從斷點(diǎn)重跑而不是從頭開(kāi)始。這兩個(gè)例子說(shuō)明一個(gè)道理Agent-Reach做的不是某一種具體業(yè)務(wù)功能而是給多Agent協(xié)作打底子。底子越穩(wěn)上層業(yè)務(wù)就越靈活。7. 寫(xiě)在最后的經(jīng)驗(yàn)之談做Agent-Reach這個(gè)項(xiàng)目最大的感受就是多智能體協(xié)作的核心難點(diǎn)其實(shí)不在“智能”而在“通信”。你讓一個(gè)聰明絕頂?shù)腁gent單獨(dú)完成一個(gè)任務(wù)它表現(xiàn)可能非常好但一旦需要多個(gè)Agent協(xié)同通信協(xié)議不清晰再聰明的Agent也會(huì)變成一盤(pán)散沙。我個(gè)人的建議是如果你正打算構(gòu)建多智能體系統(tǒng)不要把精力一開(kāi)始就花在“讓Agent更聰明”上先把你Agent之間觸達(dá)的協(xié)議設(shè)計(jì)清楚。Agent能不能檢索到對(duì)方、消息格式是不是統(tǒng)一、狀態(tài)能不能恢復(fù)、失敗有沒(méi)有兜底這些問(wèn)題解決后往上加“聰明”才是有意義的。最后分享一個(gè)小技巧在Agent-Reach的網(wǎng)絡(luò)里給每個(gè)Agent配一個(gè)“能力自描述”文本而不只是一堆標(biāo)簽。能力自描述是一段自然語(yǔ)言說(shuō)明這個(gè)Agent適合干什么、不適合干什么、有哪些前置條件。路由Agent在做觸達(dá)的時(shí)候會(huì)同時(shí)參考能力標(biāo)簽和自描述文本這樣匹配出來(lái)的Agent準(zhǔn)確率更高也不會(huì)讓一個(gè)擅長(zhǎng)寫(xiě)文案的Agent去處理數(shù)學(xué)推理任務(wù)。這個(gè)經(jīng)驗(yàn)雖然聽(tīng)起來(lái)不起眼但在實(shí)際項(xiàng)目中幫我少調(diào)了無(wú)數(shù)輪的“Agent選錯(cuò)人”問(wèn)題。