多智能體觸達(dá)與協(xié)作框架解析)
1. 項(xiàng)目概述Agent-Reach 到底是什么先把結(jié)論放前面Agent-Reach 不是一個(gè)新的 AI 模型也不是某個(gè)大廠的閉源平臺(tái)而是一個(gè)面向多智能體Multi-Agent協(xié)作場(chǎng)景的觸達(dá)與協(xié)同框架。它的核心目標(biāo)非常樸素——讓不同的 AI Agent 之間能夠互相發(fā)現(xiàn)、互相調(diào)用、協(xié)同完成任務(wù)就像給每個(gè) Agent 配了一本通訊錄和一套協(xié)作協(xié)議。我做這個(gè)項(xiàng)目的直接起因是發(fā)現(xiàn)自己手頭的幾個(gè) Agent 都成了信息孤島。一個(gè)負(fù)責(zé)數(shù)據(jù)分析一個(gè)負(fù)責(zé)定時(shí)任務(wù)推送還有一個(gè)負(fù)責(zé)知識(shí)庫(kù)檢索但它們?cè)诟髯缘男∪ψ永锱艿迷贇g也沒(méi)法彼此配合。用戶問(wèn)一個(gè)問(wèn)題我需要手工把多個(gè) Agent 的結(jié)果拼起來(lái)再自己寫(xiě)膠水代碼做流轉(zhuǎn)。這套流程跑長(zhǎng)了人累系統(tǒng)也脆。Agent-Reach 要解決的就是這套Agent 之間如何觸達(dá)的問(wèn)題。你可以把它理解成一套面向 Agent 的輕量級(jí)通訊錄 路由器 調(diào)度器。Agent 啟動(dòng)時(shí)向 Reach 中心注冊(cè)自己的身份和能力其他 Agent 通過(guò)中心查詢、路由、調(diào)起目標(biāo) Agent 的能力整個(gè)過(guò)程通過(guò)標(biāo)準(zhǔn)化的消息格式完成。這個(gè)項(xiàng)目適合誰(shuí)適合那些手上有兩三個(gè)以上獨(dú)立 Agent想讓它們串聯(lián)起來(lái)干活的人。不管你是自己寫(xiě)的爬蟲(chóng) Agent、調(diào)用大模型 API 的對(duì)話 Agent還是基于開(kāi)源框架搭的 RAG 服務(wù)Agent-Reach 都能給你一套低成本的接入方案。它不綁定具體的 Agent 實(shí)現(xiàn)語(yǔ)言也不限制你用的是哪家大模型只要遵循簡(jiǎn)單的消息規(guī)范就能接入?yún)f(xié)作網(wǎng)絡(luò)。我在設(shè)計(jì)時(shí)給自己定了三個(gè)硬指標(biāo)第一接入成本必須低單 Agent 接入不超過(guò) 200 行代碼第二不依賴重量級(jí)中間件避免為了喝水挖口井第三要能容忍網(wǎng)絡(luò)中任何一個(gè) Agent 隨時(shí)離線不能一個(gè)節(jié)點(diǎn)掛掉就全盤(pán)崩潰。這些要求聽(tīng)起來(lái)簡(jiǎn)單真落地時(shí)牽扯到的設(shè)計(jì)決策比預(yù)期多得多。下面我把整個(gè)思路、實(shí)現(xiàn)和踩坑過(guò)程完整拆給你看。2. 整體設(shè)計(jì)與核心思路拆解2.1 為什么 Agent 之間需要觸達(dá)而不是硬編碼調(diào)用先說(shuō)一個(gè)我見(jiàn)過(guò)很多次的做法Agent A 要調(diào)用 Agent B就在 A 的代碼里硬編碼一個(gè) HTTP 請(qǐng)求指向 B 的地址。單機(jī)單服務(wù)的時(shí)候這沒(méi)什么問(wèn)題但一旦 Agent 數(shù)量上了 5 個(gè)、10 個(gè)這種方式立刻變成災(zāi)難——每個(gè) Agent 都要維護(hù)一份世界上所有其他 Agent的地址表誰(shuí)能調(diào)用誰(shuí)、誰(shuí)提供什么能力全都散落在各自的代碼里。改一個(gè) Agent 的地址所有依賴它的 Agent 都要跟著改。Agent-Reach 的思路是引入一個(gè)獨(dú)立的協(xié)調(diào)中心來(lái)解耦。這個(gè)中心不參與具體業(yè)務(wù)計(jì)算只做三件事登記Agent 上線時(shí)報(bào)備自己是誰(shuí)、能干什么、發(fā)現(xiàn)Agent 需要某個(gè)能力時(shí)從中心查詢誰(shuí)可以提供、轉(zhuǎn)達(dá)把調(diào)用請(qǐng)求通過(guò)消息的方式送達(dá)目標(biāo) Agent再把結(jié)果帶回來(lái)。用生活里的例子類比就是公司里一個(gè)大型項(xiàng)目幾十號(hào)人不會(huì)互相之間存著每個(gè)人的私人手機(jī)號(hào)而是統(tǒng)一通過(guò)行政部的通訊錄找人。你要找財(cái)務(wù)配合先查通訊錄找到財(cái)務(wù)的公開(kāi)聯(lián)系方式再通過(guò)官方渠道發(fā)協(xié)同請(qǐng)求。Agent-Reach 就是這套機(jī)制的數(shù)字化版本。2.2 核心架構(gòu)選型中心化協(xié)調(diào) 去中心化執(zhí)行在設(shè)計(jì)架構(gòu)時(shí)我糾結(jié)過(guò)兩個(gè)方向一是完全去中心化所有 Agent 通過(guò) P2P 方式互聯(lián)二是中心化協(xié)調(diào) 去中心化執(zhí)行的混合模式。最終我選了后者原因很現(xiàn)實(shí)P2P 方案在節(jié)點(diǎn)發(fā)現(xiàn)、消息路由、安全認(rèn)證上需要投入的工程量太大對(duì)于個(gè)人項(xiàng)目和中小團(tuán)隊(duì)來(lái)說(shuō)上手成本不成比例。Agent-Reach 的最終架構(gòu)是這樣的Reach Coordinator協(xié)調(diào)中心一個(gè)獨(dú)立部署的輕量級(jí)服務(wù)負(fù)責(zé)維護(hù) Agent 注冊(cè)表、能力路由表、消息隊(duì)列。它不解析消息內(nèi)容只做地址轉(zhuǎn)發(fā)和狀態(tài)監(jiān)控。Reach Agent SDK整合到各個(gè) Agent 中的客戶端庫(kù)負(fù)責(zé)注冊(cè)、心跳、消息收發(fā)。SDK 是接入 Agent-Reach 的唯一入口封裝了所有協(xié)議細(xì)節(jié)。Agent Node接入網(wǎng)絡(luò)的獨(dú)立 Agent 服務(wù)可以是用 FastAPI 寫(xiě)的 HTTP 服務(wù)也可以是一個(gè)消息隊(duì)列消費(fèi)者甚至是一個(gè)命令行腳本——只要它跑著 SDK就能參與協(xié)作。注冊(cè)與發(fā)現(xiàn)采用推模式Agent 啟動(dòng)時(shí)主動(dòng)向協(xié)調(diào)中心上報(bào)自己的能力清單之后每隔 30 秒發(fā)送一次心跳。協(xié)調(diào)中心如果連續(xù) 3 個(gè)心跳周期沒(méi)有收到某個(gè) Agent 的消息就將其標(biāo)記為離線并在路由時(shí)自動(dòng)過(guò)濾掉。這個(gè)機(jī)制保證了整個(gè)系統(tǒng)在 Agent 動(dòng)態(tài)上下線的場(chǎng)景下仍然穩(wěn)定。2.3 能力聲明與發(fā)現(xiàn)機(jī)制每個(gè) Agent 在注冊(cè)時(shí)需要提交一份能力聲明我這里叫它AgentCapability。它的結(jié)構(gòu)設(shè)計(jì)得盡量簡(jiǎn)單避免讓接入者寫(xiě)一大堆配置。實(shí)際使用的是這樣一份 JSON{ agent_id: data-query-agent, display_name: 數(shù)據(jù)查詢助手, capabilities: [ { name: query_metric, description: 查詢指定時(shí)間段的業(yè)務(wù)指標(biāo)數(shù)據(jù), params: { metric: string, start_time: string, end_time: string } }, { name: list_metrics, description: 列出所有可查詢的指標(biāo)名稱, params: {} } ], endpoint: http://localhost:8101/reach/callback }協(xié)調(diào)中心收到這份聲明后會(huì)將能力名作為索引寫(xiě)入路由表。以后任何 Agent 發(fā)出query_metric的調(diào)用請(qǐng)求協(xié)調(diào)中心就能立刻知道該把消息投遞給誰(shuí)。整個(gè)過(guò)程用的是能力名而不是Agent ID做尋址好處是調(diào)用方不需要關(guān)心目標(biāo)到底是誰(shuí)只需要關(guān)心自己需要什么能力——這是一種基于能力的服務(wù)發(fā)現(xiàn)模式比基于服務(wù)名的發(fā)現(xiàn)更符合 Agent 協(xié)作的語(yǔ)義。這里我還設(shè)計(jì)了一個(gè)細(xì)節(jié)能力名全局唯一。如果兩個(gè) Agent 注冊(cè)了同一個(gè)能力名協(xié)調(diào)中心會(huì)保留先注冊(cè)的那個(gè)后注冊(cè)的返回沖突錯(cuò)誤。后期的多實(shí)例負(fù)載均衡先不做保持機(jī)制簡(jiǎn)單可控。3. 核心細(xì)節(jié)解析與實(shí)操要點(diǎn)3.1 消息格式設(shè)計(jì)一份可校驗(yàn)的 JSON SchemaAgent 之間的通信消息是 Agent-Reach 的命脈。消息格式設(shè)計(jì)得既要通用、又要能強(qiáng)制校驗(yàn)否則協(xié)作網(wǎng)絡(luò)里跑著各種畸形消息排查問(wèn)題會(huì)非常痛苦。我最終定的消息頂層結(jié)構(gòu)是一個(gè)信封Envelope加一個(gè)載荷Payload{ message_id: uuid-v4, message_type: request | response | event | heartbeat, protocol_version: 1.0, timestamp: 2025-01-15T14:32:10Z, sender_id: user-query-agent, target_capability: query_metric, reply_to: user-query-agent, correlation_id: uuid-v4, payload: {} }幾個(gè)關(guān)鍵字段的作用message_id每條消息的唯一 ID用于至少一次投遞at-least-once delivery中的去重。message_type區(qū)分是請(qǐng)求、響應(yīng)、事件還是心跳。事件是廣播型消息請(qǐng)求/響應(yīng)是點(diǎn)對(duì)點(diǎn)消息。correlation_id關(guān)聯(lián) ID把請(qǐng)求和響應(yīng)配成一對(duì)。調(diào)用方發(fā)出請(qǐng)求時(shí)生成一個(gè) ID響應(yīng)方回傳時(shí)把同樣的 ID 帶上調(diào)用方靠它匹配結(jié)果。reply_to告訴接收方結(jié)果往哪回。為了強(qiáng)制校驗(yàn)我給每個(gè)消息類型都定義了一份 JSON Schema。在 SDK 里做收發(fā)兩端的校驗(yàn)不符合 Schema 的消息直接丟棄并記錄告警日志。這套設(shè)計(jì)在后期排查問(wèn)題的時(shí)候幫了大忙——至少有三次事故是靠消息不符合 Schema這一條快速定位到是哪個(gè) Agent 發(fā)的臟數(shù)據(jù)。3.2 觸達(dá)流程從請(qǐng)求到響應(yīng)的完整鏈路一次標(biāo)準(zhǔn)的觸達(dá)調(diào)用完整鏈路是這樣的調(diào)用方 Agent通過(guò)本地 SDK 的reach.call(capability_name, params)方法發(fā)出調(diào)用請(qǐng)求。SDK 自動(dòng)生成message_id和correlation_id并填充target_capability。協(xié)調(diào)中心收到消息后先做 Schema 校驗(yàn)再查路由表找到提供該能力且在線狀態(tài)的 Agent。如果找不到立即返回一個(gè)capability_not_found錯(cuò)誤響應(yīng)。協(xié)調(diào)中心把消息轉(zhuǎn)發(fā)給目標(biāo) Agent 的回調(diào)端點(diǎn)或投遞到其關(guān)聯(lián)的消息隊(duì)列中同時(shí)將調(diào)用狀態(tài)記為待響應(yīng)。目標(biāo) Agent收到消息處理完畢后把結(jié)果封裝成message_typeresponse的消息回傳給協(xié)調(diào)中心。協(xié)調(diào)中心根據(jù)correlation_id找到原始的等待者將響應(yīng)投遞給調(diào)用方 Agent 的reply_to端點(diǎn)。這套流程看起來(lái)跟傳統(tǒng)的 RPC 沒(méi)有本質(zhì)區(qū)別但有兩個(gè)關(guān)鍵差異一是路由依據(jù)是能力名而非服務(wù)地址這意味著目標(biāo) Agent 的物理位置對(duì)調(diào)用方完全透明二是消息是異步的調(diào)用方發(fā)出請(qǐng)求后可以繼續(xù)干自己的事等響應(yīng)回來(lái)再處理而不是像 HTTP RPC 那樣線程被阻塞。3.3 同步調(diào)用與異步回調(diào)的取舍雖然 Agent-Reach 本質(zhì)是異步消息傳輸?shù)芏鄻I(yè)務(wù)場(chǎng)景里調(diào)用方就是需要等一個(gè)結(jié)果回來(lái)再繼續(xù)往下走。為此我在 SDK 里封裝了一個(gè)同步等待器sync waiter調(diào)用方發(fā)出請(qǐng)求后SDK 內(nèi)部用correlation_id注冊(cè)一個(gè)掛起任務(wù)然后阻塞等待最多 N 秒可配置。如果超時(shí)未收到響應(yīng)SDK 會(huì)拋出一個(gè)超時(shí)異常提醒調(diào)用方目標(biāo) Agent 可能離線或處理時(shí)間過(guò)長(zhǎng)。這里的注意點(diǎn)阻塞等待只適合耗時(shí)短的調(diào)用比如配置查詢、指標(biāo)查詢、模型推理。但如果目標(biāo)能力是一個(gè)生成周報(bào)并發(fā)送郵件這種可能要跑 30 秒的耗時(shí)任務(wù)同步等待會(huì)浪費(fèi)調(diào)用方資源。我的建議是 SDK 提供兩種模式由調(diào)用方在call()時(shí)通過(guò)參數(shù)自行選擇——同步等待或者注冊(cè)回調(diào)函數(shù)異步處理。3.4 安全與權(quán)限控制Agent 之間互相觸達(dá)不能像局域網(wǎng)裸奔一樣誰(shuí)都能調(diào)誰(shuí)。我在第一版里加了最基礎(chǔ)的 Token 認(rèn)證機(jī)制每個(gè)注冊(cè)的 Agent 在啟動(dòng)時(shí)分配一個(gè) access_token所有消息帶上這個(gè) Token協(xié)調(diào)中心驗(yàn)證不通過(guò)直接拒收。在權(quán)限控制上我采用了能力白名單模式每個(gè) Agent 注冊(cè)時(shí)除了聲明自己的能力還要聲明我可以調(diào)用哪些其他人的能力。沒(méi)有聲明的調(diào)用請(qǐng)求協(xié)調(diào)中心會(huì)返回permission_denied錯(cuò)誤。比如數(shù)據(jù)分析 Agent 聲明自己可以查詢指標(biāo)也聲明可以調(diào)用周報(bào)生成 Agent 的生成報(bào)告能力——后者如果不聲明哪怕它知道周報(bào)生成 Agent 存在也無(wú)法觸達(dá)。這個(gè)設(shè)計(jì)對(duì)于多團(tuán)隊(duì)協(xié)作的場(chǎng)景尤其重要避免了一個(gè)信用較差的 Agent 獲得全網(wǎng)所有能力的調(diào)用權(quán)。4. 實(shí)操過(guò)程從零搭建一個(gè) Agent-Reach 協(xié)作環(huán)境4.1 環(huán)境準(zhǔn)備與技術(shù)選型我建議的環(huán)境配置如下Python 3.10我自己實(shí)測(cè) 3.9 也能跑但 3.10 以上的類型標(biāo)注更舒服FastAPI 作為協(xié)調(diào)中心和 Agent 的 HTTP 服務(wù)框架Redis 6.0僅作為消息隊(duì)列和注冊(cè)狀態(tài)存儲(chǔ)Docker Compose 用于一鍵拉起環(huán)境可選為什么選 Redis 而不是 Kafka因?yàn)?Kafka 對(duì)單機(jī)部署來(lái)說(shuō)還是偏重了。Agent-Reach 目前的規(guī)模根本用不到分區(qū)、副本和消費(fèi)者組這些重型特性Redis 的 Stream 數(shù)據(jù)結(jié)構(gòu)足夠支撐消息隊(duì)列的投遞需求還自帶 TTL 過(guò)期能力可以用來(lái)做注冊(cè)心跳過(guò)期判斷。真到了每天千萬(wàn)級(jí)消息的量級(jí)再把消息隊(duì)列替換成 Kafka 也不遲——協(xié)議層是跟具體隊(duì)列解耦的。4.2 搭建協(xié)調(diào)中心協(xié)調(diào)中心是 Agent-Reach 的大腦我先把它搭起來(lái)。以下是簡(jiǎn)化版的協(xié)調(diào)中心核心代碼用 FastAPI 實(shí)現(xiàn)# coordinator.py import asyncio import json import time import uuid from typing import Dict, Optional import redis.asyncio as redis from fastapi import FastAPI, HTTPException, Header from pydantic import BaseModel app FastAPI(titleAgent-Reach Coordinator) redis_client redis.from_url(redis://localhost:6379/0) AGENT_CAPABILITY_PREFIX agent:capability: AGENT_INFO_PREFIX agent:info: HEARTBEAT_TIMEOUT 90 # 3 * 30s class RegisterBody(BaseModel): agent_id: str display_name: str capabilities: list[dict] endpoint: str app.post(/agent/register) async def register(body: RegisterBody): agent_id body.agent_id token uuid.uuid4().hex key AGENT_INFO_PREFIX agent_id # 到期時(shí)間設(shè)為心跳超時(shí)的3倍 await redis_client.hset(key, mapping{ display_name: body.display_name, endpoint: body.endpoint, token: token, status: online, }) for cap in body.capabilities: cap_key AGENT_CAPABILITY_PREFIX cap[name] # 如果能力名已被其他Agent注冊(cè)保留先注冊(cè)者 if not await redis_client.exists(cap_key): await redis_client.hset(cap_key, mapping{ agent_id: agent_id, name: cap[name], endpoint: body.endpoint, }) await redis_client.expire(key, HEARTBEAT_TIMEOUT * 3) return {status: registered, agent_id: agent_id, token: token} app.post(/agent/heartbeat) async def heartbeat(body: dict): agent_id body.get(agent_id) key AGENT_INFO_PREFIX agent_id if not await redis_client.exists(key): raise HTTPException(status_code404, detailAgent not registered) await redis_client.expire(key, HEARTBEAT_TIMEOUT * 3) return {status: ok, agent_id: agent_id}這個(gè)版本的協(xié)調(diào)中心只做了注冊(cè)和心跳。實(shí)際完整版還包含調(diào)用轉(zhuǎn)發(fā)、結(jié)果回傳、消息去重等邏輯核心代碼量大約 600 行。之所以可以用這么少的代碼實(shí)現(xiàn)是因?yàn)榘押芏酄顟B(tài)管理交給 Redis 的 Key 過(guò)期機(jī)制自動(dòng)處理——Agent 離線時(shí)不需要顯式義下線消息只要心跳停止Key 過(guò)期后路由表自然查不到它。4.3 編寫(xiě)第一個(gè) Agent數(shù)據(jù)查詢 Agent我寫(xiě)數(shù)據(jù)查詢 Agent 作為第一個(gè)接入方。這個(gè) Agent 的能力是接收一個(gè)指標(biāo)名和時(shí)間范圍返回模擬的查詢結(jié)果。為了演示接入流程我用了一個(gè)最簡(jiǎn)單的 FastAPI 服務(wù)# agent_data_query.py import json from fastapi import FastAPI, Request, Response, BackgroundTasks app FastAPI() REGISTER_URL http://localhost:8100/agent/register CALLBACK_URL http://localhost:8101/reach/callback async def register_agent(): # 模擬注冊(cè)實(shí)際代碼中會(huì)保存token payload { agent_id: data-query-agent, display_name: 數(shù)據(jù)查詢助手, capabilities: [ { name: query_metric, description: 查詢指定時(shí)間段的業(yè)務(wù)指標(biāo)數(shù)據(jù), params: {metric: string, start_time: string, end_time: string} } ], endpoint: CALLBACK_URL, } async with httpx.AsyncClient() as client: resp await client.post(REGISTER_URL, jsonpayload) print(Register result:, resp.json()) app.post(/reach/callback) async def reach_callback(request: Request): data await request.json() # 收到調(diào)用請(qǐng)求 if data[message_type] request: metric data[payload][metric] result simulate_query(metric) # 回傳響應(yīng) response_msg { message_type: response, correlation_id: data[correlation_id], target_id: data[sender_id], payload: {result: result} } await send_to_coordinator(response_msg) return Response(status_code200) def simulate_query(metric: str): # 模擬返回?cái)?shù)據(jù)點(diǎn) return [{timestamp: 2025-01-15T10:00:00Z, value: 42}]注冊(cè)完成后數(shù)據(jù)查詢 Agent 就出現(xiàn)在協(xié)調(diào)中心的路由表里了。任何 Agent 發(fā)出query_metric調(diào)用請(qǐng)求協(xié)調(diào)中心都會(huì)把消息投遞到它的/reach/callback端點(diǎn)。4.4 編寫(xiě)第二個(gè) Agent決策 Agent調(diào)用方現(xiàn)在寫(xiě)一個(gè)決策 Agent它要做的業(yè)務(wù)是收到用戶問(wèn)題后判斷需要哪些指標(biāo)數(shù)據(jù)然后調(diào)用數(shù)據(jù)查詢 Agent 獲取數(shù)據(jù)再結(jié)合大模型生成回答。這個(gè) Agent 完全不知道數(shù)據(jù)查詢 Agent 的物理地址只知道我要調(diào)用query_metric這個(gè)能力。# agent_decision.py import asyncio from agent_reach_sdk import ReachClient client ReachClient( coordinator_urlhttp://localhost:8100, agent_iddecision-agent ) async def handle_user_question(question: str): # 第一步調(diào)用數(shù)據(jù)查詢Agent的能力 future await client.call_async( capability_namequery_metric, params{ metric: daily_active_users, start_time: 2025-01-14T00:00:00Z, end_time: 2025-01-15T00:00:00Z }, timeout10.0 ) result await future print(f拿到數(shù)據(jù): {result}) # 第二步結(jié)合大模型生成回答 answer generate_answer(question, result[data]) return answer我在這里使用了client.call_async而不是client.call目的就是展示兩種模式的區(qū)別。在這個(gè)場(chǎng)景里調(diào)用數(shù)據(jù)查詢是整體回答的中途步驟決策 Agent 必須等數(shù)據(jù)回來(lái)才能繼續(xù)所以用同步等待的效果其實(shí)更好。真正需要異步的往往是發(fā)完通知就不管了這類場(chǎng)景。4.5 測(cè)試全流程注冊(cè)、調(diào)用、響應(yīng)環(huán)境搭建完畢后我按照以下順序做了完整的端到端測(cè)試啟動(dòng)協(xié)調(diào)中心FastAPI Redis。啟動(dòng)數(shù)據(jù)查詢 Agent觀察注冊(cè)日志確認(rèn)注冊(cè)成功。啟動(dòng)決策 Agent確認(rèn)它也注冊(cè)成功。用決策 Agent 發(fā)一次調(diào)用觀察協(xié)調(diào)中心的路由日志。實(shí)測(cè)結(jié)果中從發(fā)出調(diào)用到收到結(jié)果完整鏈路耗時(shí)約 12 毫秒——這個(gè)延遲對(duì)一個(gè)跨進(jìn)程的 Agent 協(xié)作來(lái)說(shuō)非常理想。真正耗時(shí)的部分是目標(biāo) Agent 內(nèi)部的業(yè)務(wù)邏輯比如查數(shù)據(jù)庫(kù)、調(diào)大模型 APIAgent-Reach 本身不應(yīng)成為性能瓶頸。5. 常見(jiàn)問(wèn)題與排查技巧實(shí)錄5.1 Agent 注冊(cè)不成功排查思路我遇到過(guò)最典型的注冊(cè)失敗原因是能力名沖突。兩個(gè) Agent 同時(shí)注冊(cè)了send_notification這個(gè)能力名協(xié)調(diào)中心會(huì)保留先注冊(cè)的后注冊(cè)的直接報(bào)capability_conflict。排查時(shí)先看協(xié)調(diào)中心注冊(cè)接口的返回再查看 Redis 里的agent:capability:*鍵能快速定位是誰(shuí)占用了能力名。另一個(gè)常見(jiàn)坑是 Agent 的注冊(cè)請(qǐng)求到達(dá)協(xié)調(diào)中心卻立即斷開(kāi)連接。FastAPI 默認(rèn)的 Body 大小限制是 4KB如果能力聲明寫(xiě)得特別長(zhǎng)比如一個(gè)能力里塞了 20 個(gè)參數(shù)定義請(qǐng)求就會(huì)超過(guò)限制被拒收。解決方法是里多寫(xiě)一個(gè)參數(shù)把 limit 調(diào)大或者精簡(jiǎn)能力聲明的詳細(xì)程度。5.2 消息丟失Redis Stream 與至少一次投遞在初版實(shí)現(xiàn)里我用了 Redis List 做消息隊(duì)列結(jié)果是偶發(fā)性消息丟失。后來(lái)排查發(fā)現(xiàn)是目標(biāo) Agent 處理完消息后忘記 ACK消費(fèi)端崩潰導(dǎo)致消息丟失。換成 Redis Stream 后我把消費(fèi)邏輯改成了先處理、后 ACK并做了消息 ID 去重——這樣即使 Agent 在處理過(guò)程中崩潰重啟后還能從 Pending List 里撈回未確認(rèn)的消息重新處理。這里的關(guān)鍵心得是不要讓消費(fèi)者先 ACK 再處理。雖然這樣可以避免重復(fù)處理但代價(jià)是如果處理中途宕機(jī)這條消息就永久丟失了。寧可先處理、后 ACK配合消息 ID 去重來(lái)保證冪等也不能接受消息丟失。這是所有消息系統(tǒng)里我最堅(jiān)持的一條原則。5.3 調(diào)用超時(shí)與重試策略當(dāng)目標(biāo) Agent 處理耗時(shí)超過(guò)調(diào)用方的超時(shí)閾值調(diào)用方會(huì)收到 Timeout 異常。但要注意超時(shí)異常只代表調(diào)用方不想等不代表目標(biāo) Agent 沒(méi)在處理。這種情況下如果調(diào)用方立即重試可能導(dǎo)致同一個(gè)請(qǐng)求被處理兩次。我的解決方案是在消息里加了一個(gè)idempotency_key字段目標(biāo) Agent 收到請(qǐng)求后先去查 Redis 去重表——如果這個(gè) key 已經(jīng)處理過(guò)直接返回第一次的結(jié)果。超時(shí)后調(diào)用方如果要重試還是用同一個(gè)idempotency_key就能安全地做到至多一次副作用。5.4 循環(huán)調(diào)用死鎖Agent A 調(diào) BB 調(diào) A如果兩個(gè) Agent 互相提供能力而且每次都同步等待對(duì)方的結(jié)果就可能在極端場(chǎng)景下形成死鎖A 在等 B 回應(yīng)B 也在等 A 回應(yīng)。雖然邏輯上不太可能有人故意這樣設(shè)計(jì)但業(yè)務(wù)復(fù)雜后這種隱式循環(huán)是防不勝防的。我在 SDK 層加了一個(gè)簡(jiǎn)單防護(hù)每次消息轉(zhuǎn)發(fā)都帶上一個(gè)max_hop計(jì)數(shù)默認(rèn) 5。每經(jīng)過(guò)一個(gè) Agent 轉(zhuǎn)發(fā)計(jì)數(shù)減 1減到 0 就自動(dòng)返回max_hop_exceeded錯(cuò)誤。這樣即使出現(xiàn)循環(huán)調(diào)用也會(huì)在有限跳數(shù)內(nèi)被切斷而不是無(wú)限死鎖。6. 后續(xù)擴(kuò)展Agent-Reach 還能怎么玩這套框架設(shè)計(jì)時(shí)保留了足夠的擴(kuò)展空間我梳理了幾個(gè)高價(jià)值的后續(xù)方向供真正打算自己搭的人參考多語(yǔ)言 SDK 支持目前官方 SDK 只有 Python 版但協(xié)議層是 HTTP JSON理論上任何語(yǔ)言都能實(shí)現(xiàn)。我自己用 Node.js 寫(xiě)了一個(gè)簡(jiǎn)化版 SDK大約 150 行也能正常接入?yún)f(xié)調(diào)中心。能力負(fù)載均衡如果一個(gè)能力由多個(gè) Agent 提供比如多個(gè)數(shù)據(jù)源可以在路由表里給能力名掛多個(gè) Agent 地址按權(quán)重或輪詢策略分發(fā)請(qǐng)求。審計(jì)日志與追蹤鏈跨 Agent 協(xié)作的調(diào)試是最大的痛點(diǎn)之一。我后面給協(xié)調(diào)中心加了一個(gè)鏈路追蹤日志記錄每條消息從發(fā)出到響應(yīng)的完整時(shí)間線和跳轉(zhuǎn)路徑排查問(wèn)題效率提升了一個(gè)量級(jí)。語(yǔ)義路由目前的尋址是精確能力名匹配。下一步可以把能力描述向量化讓調(diào)用方用自然語(yǔ)言描述需求由協(xié)調(diào)中心通過(guò)語(yǔ)義匹配找到最合適的目標(biāo) Agent。這個(gè)方向有一定工程量但收益明顯尤其適合 Agent 數(shù)量變多后能力名不好管理的場(chǎng)景。在實(shí)際摸索 Agent-Reach 的過(guò)程中我最深的一個(gè)體會(huì)是做多 Agent 協(xié)作真正難的不是讓一堆大模型 API 互相調(diào)用而是把誰(shuí)能找到誰(shuí)、誰(shuí)能調(diào)誰(shuí)、消息怎么傳、失敗怎么辦這套基礎(chǔ)設(shè)施搭得足夠穩(wěn)。消息協(xié)議、注冊(cè)發(fā)現(xiàn)、心跳監(jiān)控、冪等去重——每一個(gè)環(huán)節(jié)單看都很簡(jiǎn)單組合在一起才構(gòu)成了可用的協(xié)作底座。這套東西做完之后我再接入新的 Agent 基本只需要半小時(shí)到現(xiàn)在已經(jīng)穩(wěn)定運(yùn)行了兩個(gè)多月沒(méi)有出過(guò)一次因?yàn)榭蚣鼙旧韺?dǎo)致的消息丟失或路由錯(cuò)亂。如果你手頭的 Agent 們也正各自為戰(zhàn)真的建議動(dòng)手試試這條路投資回報(bào)率比繼續(xù)堆模型能力高得多。 Agent-Reach：輕量級(jí)多智能體觸達(dá)與協(xié)作框架解析
1. 項(xiàng)目概述Agent-Reach 到底是什么先把結(jié)論放前面Agent-Reach 不是一個(gè)新的 AI 模型也不是某個(gè)大廠的閉源平臺(tái)而是一個(gè)面向多智能體Multi-Agent協(xié)作場(chǎng)景的觸達(dá)與協(xié)同框架。它的核心目標(biāo)非常樸素——讓不同的 AI Agent 之間能夠互相發(fā)現(xiàn)、互相調(diào)用、協(xié)同完成任務(wù)就像給每個(gè) Agent 配了一本通訊錄和一套協(xié)作協(xié)議。我做這個(gè)項(xiàng)目的直接起因是發(fā)現(xiàn)自己手頭的幾個(gè) Agent 都成了信息孤島。一個(gè)負(fù)責(zé)數(shù)據(jù)分析一個(gè)負(fù)責(zé)定時(shí)任務(wù)推送還有一個(gè)負(fù)責(zé)知識(shí)庫(kù)檢索但它們?cè)诟髯缘男∪ψ永锱艿迷贇g也沒(méi)法彼此配合。用戶問(wèn)一個(gè)問(wèn)題我需要手工把多個(gè) Agent 的結(jié)果拼起來(lái)再自己寫(xiě)膠水代碼做流轉(zhuǎn)。這套流程跑長(zhǎng)了人累系統(tǒng)也脆。Agent-Reach 要解決的就是這套Agent 之間如何觸達(dá)的問(wèn)題。你可以把它理解成一套面向 Agent 的輕量級(jí)通訊錄 路由器 調(diào)度器。Agent 啟動(dòng)時(shí)向 Reach 中心注冊(cè)自己的身份和能力其他 Agent 通過(guò)中心查詢、路由、調(diào)起目標(biāo) Agent 的能力整個(gè)過(guò)程通過(guò)標(biāo)準(zhǔn)化的消息格式完成。這個(gè)項(xiàng)目適合誰(shuí)適合那些手上有兩三個(gè)以上獨(dú)立 Agent想讓它們串聯(lián)起來(lái)干活的人。不管你是自己寫(xiě)的爬蟲(chóng) Agent、調(diào)用大模型 API 的對(duì)話 Agent還是基于開(kāi)源框架搭的 RAG 服務(wù)Agent-Reach 都能給你一套低成本的接入方案。它不綁定具體的 Agent 實(shí)現(xiàn)語(yǔ)言也不限制你用的是哪家大模型只要遵循簡(jiǎn)單的消息規(guī)范就能接入?yún)f(xié)作網(wǎng)絡(luò)。我在設(shè)計(jì)時(shí)給自己定了三個(gè)硬指標(biāo)第一接入成本必須低單 Agent 接入不超過(guò) 200 行代碼第二不依賴重量級(jí)中間件避免為了喝水挖口井第三要能容忍網(wǎng)絡(luò)中任何一個(gè) Agent 隨時(shí)離線不能一個(gè)節(jié)點(diǎn)掛掉就全盤(pán)崩潰。這些要求聽(tīng)起來(lái)簡(jiǎn)單真落地時(shí)牽扯到的設(shè)計(jì)決策比預(yù)期多得多。下面我把整個(gè)思路、實(shí)現(xiàn)和踩坑過(guò)程完整拆給你看。2. 整體設(shè)計(jì)與核心思路拆解2.1 為什么 Agent 之間需要觸達(dá)而不是硬編碼調(diào)用先說(shuō)一個(gè)我見(jiàn)過(guò)很多次的做法Agent A 要調(diào)用 Agent B就在 A 的代碼里硬編碼一個(gè) HTTP 請(qǐng)求指向 B 的地址。單機(jī)單服務(wù)的時(shí)候這沒(méi)什么問(wèn)題但一旦 Agent 數(shù)量上了 5 個(gè)、10 個(gè)這種方式立刻變成災(zāi)難——每個(gè) Agent 都要維護(hù)一份世界上所有其他 Agent的地址表誰(shuí)能調(diào)用誰(shuí)、誰(shuí)提供什么能力全都散落在各自的代碼里。改一個(gè) Agent 的地址所有依賴它的 Agent 都要跟著改。Agent-Reach 的思路是引入一個(gè)獨(dú)立的協(xié)調(diào)中心來(lái)解耦。這個(gè)中心不參與具體業(yè)務(wù)計(jì)算只做三件事登記Agent 上線時(shí)報(bào)備自己是誰(shuí)、能干什么、發(fā)現(xiàn)Agent 需要某個(gè)能力時(shí)從中心查詢誰(shuí)可以提供、轉(zhuǎn)達(dá)把調(diào)用請(qǐng)求通過(guò)消息的方式送達(dá)目標(biāo) Agent再把結(jié)果帶回來(lái)。用生活里的例子類比就是公司里一個(gè)大型項(xiàng)目幾十號(hào)人不會(huì)互相之間存著每個(gè)人的私人手機(jī)號(hào)而是統(tǒng)一通過(guò)行政部的通訊錄找人。你要找財(cái)務(wù)配合先查通訊錄找到財(cái)務(wù)的公開(kāi)聯(lián)系方式再通過(guò)官方渠道發(fā)協(xié)同請(qǐng)求。Agent-Reach 就是這套機(jī)制的數(shù)字化版本。2.2 核心架構(gòu)選型中心化協(xié)調(diào) 去中心化執(zhí)行在設(shè)計(jì)架構(gòu)時(shí)我糾結(jié)過(guò)兩個(gè)方向一是完全去中心化所有 Agent 通過(guò) P2P 方式互聯(lián)二是中心化協(xié)調(diào) 去中心化執(zhí)行的混合模式。最終我選了后者原因很現(xiàn)實(shí)P2P 方案在節(jié)點(diǎn)發(fā)現(xiàn)、消息路由、安全認(rèn)證上需要投入的工程量太大對(duì)于個(gè)人項(xiàng)目和中小團(tuán)隊(duì)來(lái)說(shuō)上手成本不成比例。Agent-Reach 的最終架構(gòu)是這樣的Reach Coordinator協(xié)調(diào)中心一個(gè)獨(dú)立部署的輕量級(jí)服務(wù)負(fù)責(zé)維護(hù) Agent 注冊(cè)表、能力路由表、消息隊(duì)列。它不解析消息內(nèi)容只做地址轉(zhuǎn)發(fā)和狀態(tài)監(jiān)控。Reach Agent SDK整合到各個(gè) Agent 中的客戶端庫(kù)負(fù)責(zé)注冊(cè)、心跳、消息收發(fā)。SDK 是接入 Agent-Reach 的唯一入口封裝了所有協(xié)議細(xì)節(jié)。Agent Node接入網(wǎng)絡(luò)的獨(dú)立 Agent 服務(wù)可以是用 FastAPI 寫(xiě)的 HTTP 服務(wù)也可以是一個(gè)消息隊(duì)列消費(fèi)者甚至是一個(gè)命令行腳本——只要它跑著 SDK就能參與協(xié)作。注冊(cè)與發(fā)現(xiàn)采用推模式Agent 啟動(dòng)時(shí)主動(dòng)向協(xié)調(diào)中心上報(bào)自己的能力清單之后每隔 30 秒發(fā)送一次心跳。協(xié)調(diào)中心如果連續(xù) 3 個(gè)心跳周期沒(méi)有收到某個(gè) Agent 的消息就將其標(biāo)記為離線并在路由時(shí)自動(dòng)過(guò)濾掉。這個(gè)機(jī)制保證了整個(gè)系統(tǒng)在 Agent 動(dòng)態(tài)上下線的場(chǎng)景下仍然穩(wěn)定。2.3 能力聲明與發(fā)現(xiàn)機(jī)制每個(gè) Agent 在注冊(cè)時(shí)需要提交一份能力聲明我這里叫它AgentCapability。它的結(jié)構(gòu)設(shè)計(jì)得盡量簡(jiǎn)單避免讓接入者寫(xiě)一大堆配置。實(shí)際使用的是這樣一份 JSON{ agent_id: data-query-agent, display_name: 數(shù)據(jù)查詢助手, capabilities: [ { name: query_metric, description: 查詢指定時(shí)間段的業(yè)務(wù)指標(biāo)數(shù)據(jù), params: { metric: string, start_time: string, end_time: string } }, { name: list_metrics, description: 列出所有可查詢的指標(biāo)名稱, params: {} } ], endpoint: http://localhost:8101/reach/callback }協(xié)調(diào)中心收到這份聲明后會(huì)將能力名作為索引寫(xiě)入路由表。以后任何 Agent 發(fā)出query_metric的調(diào)用請(qǐng)求協(xié)調(diào)中心就能立刻知道該把消息投遞給誰(shuí)。整個(gè)過(guò)程用的是能力名而不是Agent ID做尋址好處是調(diào)用方不需要關(guān)心目標(biāo)到底是誰(shuí)只需要關(guān)心自己需要什么能力——這是一種基于能力的服務(wù)發(fā)現(xiàn)模式比基于服務(wù)名的發(fā)現(xiàn)更符合 Agent 協(xié)作的語(yǔ)義。這里我還設(shè)計(jì)了一個(gè)細(xì)節(jié)能力名全局唯一。如果兩個(gè) Agent 注冊(cè)了同一個(gè)能力名協(xié)調(diào)中心會(huì)保留先注冊(cè)的那個(gè)后注冊(cè)的返回沖突錯(cuò)誤。后期的多實(shí)例負(fù)載均衡先不做保持機(jī)制簡(jiǎn)單可控。3. 核心細(xì)節(jié)解析與實(shí)操要點(diǎn)3.1 消息格式設(shè)計(jì)一份可校驗(yàn)的 JSON SchemaAgent 之間的通信消息是 Agent-Reach 的命脈。消息格式設(shè)計(jì)得既要通用、又要能強(qiáng)制校驗(yàn)否則協(xié)作網(wǎng)絡(luò)里跑著各種畸形消息排查問(wèn)題會(huì)非常痛苦。我最終定的消息頂層結(jié)構(gòu)是一個(gè)信封Envelope加一個(gè)載荷Payload{ message_id: uuid-v4, message_type: request | response | event | heartbeat, protocol_version: 1.0, timestamp: 2025-01-15T14:32:10Z, sender_id: user-query-agent, target_capability: query_metric, reply_to: user-query-agent, correlation_id: uuid-v4, payload: {} }幾個(gè)關(guān)鍵字段的作用message_id每條消息的唯一 ID用于至少一次投遞at-least-once delivery中的去重。message_type區(qū)分是請(qǐng)求、響應(yīng)、事件還是心跳。事件是廣播型消息請(qǐng)求/響應(yīng)是點(diǎn)對(duì)點(diǎn)消息。correlation_id關(guān)聯(lián) ID把請(qǐng)求和響應(yīng)配成一對(duì)。調(diào)用方發(fā)出請(qǐng)求時(shí)生成一個(gè) ID響應(yīng)方回傳時(shí)把同樣的 ID 帶上調(diào)用方靠它匹配結(jié)果。reply_to告訴接收方結(jié)果往哪回。為了強(qiáng)制校驗(yàn)我給每個(gè)消息類型都定義了一份 JSON Schema。在 SDK 里做收發(fā)兩端的校驗(yàn)不符合 Schema 的消息直接丟棄并記錄告警日志。這套設(shè)計(jì)在后期排查問(wèn)題的時(shí)候幫了大忙——至少有三次事故是靠消息不符合 Schema這一條快速定位到是哪個(gè) Agent 發(fā)的臟數(shù)據(jù)。3.2 觸達(dá)流程從請(qǐng)求到響應(yīng)的完整鏈路一次標(biāo)準(zhǔn)的觸達(dá)調(diào)用完整鏈路是這樣的調(diào)用方 Agent通過(guò)本地 SDK 的reach.call(capability_name, params)方法發(fā)出調(diào)用請(qǐng)求。SDK 自動(dòng)生成message_id和correlation_id并填充target_capability。協(xié)調(diào)中心收到消息后先做 Schema 校驗(yàn)再查路由表找到提供該能力且在線狀態(tài)的 Agent。如果找不到立即返回一個(gè)capability_not_found錯(cuò)誤響應(yīng)。協(xié)調(diào)中心把消息轉(zhuǎn)發(fā)給目標(biāo) Agent 的回調(diào)端點(diǎn)或投遞到其關(guān)聯(lián)的消息隊(duì)列中同時(shí)將調(diào)用狀態(tài)記為待響應(yīng)。目標(biāo) Agent收到消息處理完畢后把結(jié)果封裝成message_typeresponse的消息回傳給協(xié)調(diào)中心。協(xié)調(diào)中心根據(jù)correlation_id找到原始的等待者將響應(yīng)投遞給調(diào)用方 Agent 的reply_to端點(diǎn)。這套流程看起來(lái)跟傳統(tǒng)的 RPC 沒(méi)有本質(zhì)區(qū)別但有兩個(gè)關(guān)鍵差異一是路由依據(jù)是能力名而非服務(wù)地址這意味著目標(biāo) Agent 的物理位置對(duì)調(diào)用方完全透明二是消息是異步的調(diào)用方發(fā)出請(qǐng)求后可以繼續(xù)干自己的事等響應(yīng)回來(lái)再處理而不是像 HTTP RPC 那樣線程被阻塞。3.3 同步調(diào)用與異步回調(diào)的取舍雖然 Agent-Reach 本質(zhì)是異步消息傳輸?shù)芏鄻I(yè)務(wù)場(chǎng)景里調(diào)用方就是需要等一個(gè)結(jié)果回來(lái)再繼續(xù)往下走。為此我在 SDK 里封裝了一個(gè)同步等待器sync waiter調(diào)用方發(fā)出請(qǐng)求后SDK 內(nèi)部用correlation_id注冊(cè)一個(gè)掛起任務(wù)然后阻塞等待最多 N 秒可配置。如果超時(shí)未收到響應(yīng)SDK 會(huì)拋出一個(gè)超時(shí)異常提醒調(diào)用方目標(biāo) Agent 可能離線或處理時(shí)間過(guò)長(zhǎng)。這里的注意點(diǎn)阻塞等待只適合耗時(shí)短的調(diào)用比如配置查詢、指標(biāo)查詢、模型推理。但如果目標(biāo)能力是一個(gè)生成周報(bào)并發(fā)送郵件這種可能要跑 30 秒的耗時(shí)任務(wù)同步等待會(huì)浪費(fèi)調(diào)用方資源。我的建議是 SDK 提供兩種模式由調(diào)用方在call()時(shí)通過(guò)參數(shù)自行選擇——同步等待或者注冊(cè)回調(diào)函數(shù)異步處理。3.4 安全與權(quán)限控制Agent 之間互相觸達(dá)不能像局域網(wǎng)裸奔一樣誰(shuí)都能調(diào)誰(shuí)。我在第一版里加了最基礎(chǔ)的 Token 認(rèn)證機(jī)制每個(gè)注冊(cè)的 Agent 在啟動(dòng)時(shí)分配一個(gè) access_token所有消息帶上這個(gè) Token協(xié)調(diào)中心驗(yàn)證不通過(guò)直接拒收。在權(quán)限控制上我采用了能力白名單模式每個(gè) Agent 注冊(cè)時(shí)除了聲明自己的能力還要聲明我可以調(diào)用哪些其他人的能力。沒(méi)有聲明的調(diào)用請(qǐng)求協(xié)調(diào)中心會(huì)返回permission_denied錯(cuò)誤。比如數(shù)據(jù)分析 Agent 聲明自己可以查詢指標(biāo)也聲明可以調(diào)用周報(bào)生成 Agent 的生成報(bào)告能力——后者如果不聲明哪怕它知道周報(bào)生成 Agent 存在也無(wú)法觸達(dá)。這個(gè)設(shè)計(jì)對(duì)于多團(tuán)隊(duì)協(xié)作的場(chǎng)景尤其重要避免了一個(gè)信用較差的 Agent 獲得全網(wǎng)所有能力的調(diào)用權(quán)。4. 實(shí)操過(guò)程從零搭建一個(gè) Agent-Reach 協(xié)作環(huán)境4.1 環(huán)境準(zhǔn)備與技術(shù)選型我建議的環(huán)境配置如下Python 3.10我自己實(shí)測(cè) 3.9 也能跑但 3.10 以上的類型標(biāo)注更舒服FastAPI 作為協(xié)調(diào)中心和 Agent 的 HTTP 服務(wù)框架Redis 6.0僅作為消息隊(duì)列和注冊(cè)狀態(tài)存儲(chǔ)Docker Compose 用于一鍵拉起環(huán)境可選為什么選 Redis 而不是 Kafka因?yàn)?Kafka 對(duì)單機(jī)部署來(lái)說(shuō)還是偏重了。Agent-Reach 目前的規(guī)模根本用不到分區(qū)、副本和消費(fèi)者組這些重型特性Redis 的 Stream 數(shù)據(jù)結(jié)構(gòu)足夠支撐消息隊(duì)列的投遞需求還自帶 TTL 過(guò)期能力可以用來(lái)做注冊(cè)心跳過(guò)期判斷。真到了每天千萬(wàn)級(jí)消息的量級(jí)再把消息隊(duì)列替換成 Kafka 也不遲——協(xié)議層是跟具體隊(duì)列解耦的。4.2 搭建協(xié)調(diào)中心協(xié)調(diào)中心是 Agent-Reach 的大腦我先把它搭起來(lái)。以下是簡(jiǎn)化版的協(xié)調(diào)中心核心代碼用 FastAPI 實(shí)現(xiàn)# coordinator.py import asyncio import json import time import uuid from typing import Dict, Optional import redis.asyncio as redis from fastapi import FastAPI, HTTPException, Header from pydantic import BaseModel app FastAPI(titleAgent-Reach Coordinator) redis_client redis.from_url(redis://localhost:6379/0) AGENT_CAPABILITY_PREFIX agent:capability: AGENT_INFO_PREFIX agent:info: HEARTBEAT_TIMEOUT 90 # 3 * 30s class RegisterBody(BaseModel): agent_id: str display_name: str capabilities: list[dict] endpoint: str app.post(/agent/register) async def register(body: RegisterBody): agent_id body.agent_id token uuid.uuid4().hex key AGENT_INFO_PREFIX agent_id # 到期時(shí)間設(shè)為心跳超時(shí)的3倍 await redis_client.hset(key, mapping{ display_name: body.display_name, endpoint: body.endpoint, token: token, status: online, }) for cap in body.capabilities: cap_key AGENT_CAPABILITY_PREFIX cap[name] # 如果能力名已被其他Agent注冊(cè)保留先注冊(cè)者 if not await redis_client.exists(cap_key): await redis_client.hset(cap_key, mapping{ agent_id: agent_id, name: cap[name], endpoint: body.endpoint, }) await redis_client.expire(key, HEARTBEAT_TIMEOUT * 3) return {status: registered, agent_id: agent_id, token: token} app.post(/agent/heartbeat) async def heartbeat(body: dict): agent_id body.get(agent_id) key AGENT_INFO_PREFIX agent_id if not await redis_client.exists(key): raise HTTPException(status_code404, detailAgent not registered) await redis_client.expire(key, HEARTBEAT_TIMEOUT * 3) return {status: ok, agent_id: agent_id}這個(gè)版本的協(xié)調(diào)中心只做了注冊(cè)和心跳。實(shí)際完整版還包含調(diào)用轉(zhuǎn)發(fā)、結(jié)果回傳、消息去重等邏輯核心代碼量大約 600 行。之所以可以用這么少的代碼實(shí)現(xiàn)是因?yàn)榘押芏酄顟B(tài)管理交給 Redis 的 Key 過(guò)期機(jī)制自動(dòng)處理——Agent 離線時(shí)不需要顯式義下線消息只要心跳停止Key 過(guò)期后路由表自然查不到它。4.3 編寫(xiě)第一個(gè) Agent數(shù)據(jù)查詢 Agent我寫(xiě)數(shù)據(jù)查詢 Agent 作為第一個(gè)接入方。這個(gè) Agent 的能力是接收一個(gè)指標(biāo)名和時(shí)間范圍返回模擬的查詢結(jié)果。為了演示接入流程我用了一個(gè)最簡(jiǎn)單的 FastAPI 服務(wù)# agent_data_query.py import json from fastapi import FastAPI, Request, Response, BackgroundTasks app FastAPI() REGISTER_URL http://localhost:8100/agent/register CALLBACK_URL http://localhost:8101/reach/callback async def register_agent(): # 模擬注冊(cè)實(shí)際代碼中會(huì)保存token payload { agent_id: data-query-agent, display_name: 數(shù)據(jù)查詢助手, capabilities: [ { name: query_metric, description: 查詢指定時(shí)間段的業(yè)務(wù)指標(biāo)數(shù)據(jù), params: {metric: string, start_time: string, end_time: string} } ], endpoint: CALLBACK_URL, } async with httpx.AsyncClient() as client: resp await client.post(REGISTER_URL, jsonpayload) print(Register result:, resp.json()) app.post(/reach/callback) async def reach_callback(request: Request): data await request.json() # 收到調(diào)用請(qǐng)求 if data[message_type] request: metric data[payload][metric] result simulate_query(metric) # 回傳響應(yīng) response_msg { message_type: response, correlation_id: data[correlation_id], target_id: data[sender_id], payload: {result: result} } await send_to_coordinator(response_msg) return Response(status_code200) def simulate_query(metric: str): # 模擬返回?cái)?shù)據(jù)點(diǎn) return [{timestamp: 2025-01-15T10:00:00Z, value: 42}]注冊(cè)完成后數(shù)據(jù)查詢 Agent 就出現(xiàn)在協(xié)調(diào)中心的路由表里了。任何 Agent 發(fā)出query_metric調(diào)用請(qǐng)求協(xié)調(diào)中心都會(huì)把消息投遞到它的/reach/callback端點(diǎn)。4.4 編寫(xiě)第二個(gè) Agent決策 Agent調(diào)用方現(xiàn)在寫(xiě)一個(gè)決策 Agent它要做的業(yè)務(wù)是收到用戶問(wèn)題后判斷需要哪些指標(biāo)數(shù)據(jù)然后調(diào)用數(shù)據(jù)查詢 Agent 獲取數(shù)據(jù)再結(jié)合大模型生成回答。這個(gè) Agent 完全不知道數(shù)據(jù)查詢 Agent 的物理地址只知道我要調(diào)用query_metric這個(gè)能力。# agent_decision.py import asyncio from agent_reach_sdk import ReachClient client ReachClient( coordinator_urlhttp://localhost:8100, agent_iddecision-agent ) async def handle_user_question(question: str): # 第一步調(diào)用數(shù)據(jù)查詢Agent的能力 future await client.call_async( capability_namequery_metric, params{ metric: daily_active_users, start_time: 2025-01-14T00:00:00Z, end_time: 2025-01-15T00:00:00Z }, timeout10.0 ) result await future print(f拿到數(shù)據(jù): {result}) # 第二步結(jié)合大模型生成回答 answer generate_answer(question, result[data]) return answer我在這里使用了client.call_async而不是client.call目的就是展示兩種模式的區(qū)別。在這個(gè)場(chǎng)景里調(diào)用數(shù)據(jù)查詢是整體回答的中途步驟決策 Agent 必須等數(shù)據(jù)回來(lái)才能繼續(xù)所以用同步等待的效果其實(shí)更好。真正需要異步的往往是發(fā)完通知就不管了這類場(chǎng)景。4.5 測(cè)試全流程注冊(cè)、調(diào)用、響應(yīng)環(huán)境搭建完畢后我按照以下順序做了完整的端到端測(cè)試啟動(dòng)協(xié)調(diào)中心FastAPI Redis。啟動(dòng)數(shù)據(jù)查詢 Agent觀察注冊(cè)日志確認(rèn)注冊(cè)成功。啟動(dòng)決策 Agent確認(rèn)它也注冊(cè)成功。用決策 Agent 發(fā)一次調(diào)用觀察協(xié)調(diào)中心的路由日志。實(shí)測(cè)結(jié)果中從發(fā)出調(diào)用到收到結(jié)果完整鏈路耗時(shí)約 12 毫秒——這個(gè)延遲對(duì)一個(gè)跨進(jìn)程的 Agent 協(xié)作來(lái)說(shuō)非常理想。真正耗時(shí)的部分是目標(biāo) Agent 內(nèi)部的業(yè)務(wù)邏輯比如查數(shù)據(jù)庫(kù)、調(diào)大模型 APIAgent-Reach 本身不應(yīng)成為性能瓶頸。5. 常見(jiàn)問(wèn)題與排查技巧實(shí)錄5.1 Agent 注冊(cè)不成功排查思路我遇到過(guò)最典型的注冊(cè)失敗原因是能力名沖突。兩個(gè) Agent 同時(shí)注冊(cè)了send_notification這個(gè)能力名協(xié)調(diào)中心會(huì)保留先注冊(cè)的后注冊(cè)的直接報(bào)capability_conflict。排查時(shí)先看協(xié)調(diào)中心注冊(cè)接口的返回再查看 Redis 里的agent:capability:*鍵能快速定位是誰(shuí)占用了能力名。另一個(gè)常見(jiàn)坑是 Agent 的注冊(cè)請(qǐng)求到達(dá)協(xié)調(diào)中心卻立即斷開(kāi)連接。FastAPI 默認(rèn)的 Body 大小限制是 4KB如果能力聲明寫(xiě)得特別長(zhǎng)比如一個(gè)能力里塞了 20 個(gè)參數(shù)定義請(qǐng)求就會(huì)超過(guò)限制被拒收。解決方法是里多寫(xiě)一個(gè)參數(shù)把 limit 調(diào)大或者精簡(jiǎn)能力聲明的詳細(xì)程度。5.2 消息丟失Redis Stream 與至少一次投遞在初版實(shí)現(xiàn)里我用了 Redis List 做消息隊(duì)列結(jié)果是偶發(fā)性消息丟失。后來(lái)排查發(fā)現(xiàn)是目標(biāo) Agent 處理完消息后忘記 ACK消費(fèi)端崩潰導(dǎo)致消息丟失。換成 Redis Stream 后我把消費(fèi)邏輯改成了先處理、后 ACK并做了消息 ID 去重——這樣即使 Agent 在處理過(guò)程中崩潰重啟后還能從 Pending List 里撈回未確認(rèn)的消息重新處理。這里的關(guān)鍵心得是不要讓消費(fèi)者先 ACK 再處理。雖然這樣可以避免重復(fù)處理但代價(jià)是如果處理中途宕機(jī)這條消息就永久丟失了。寧可先處理、后 ACK配合消息 ID 去重來(lái)保證冪等也不能接受消息丟失。這是所有消息系統(tǒng)里我最堅(jiān)持的一條原則。5.3 調(diào)用超時(shí)與重試策略當(dāng)目標(biāo) Agent 處理耗時(shí)超過(guò)調(diào)用方的超時(shí)閾值調(diào)用方會(huì)收到 Timeout 異常。但要注意超時(shí)異常只代表調(diào)用方不想等不代表目標(biāo) Agent 沒(méi)在處理。這種情況下如果調(diào)用方立即重試可能導(dǎo)致同一個(gè)請(qǐng)求被處理兩次。我的解決方案是在消息里加了一個(gè)idempotency_key字段目標(biāo) Agent 收到請(qǐng)求后先去查 Redis 去重表——如果這個(gè) key 已經(jīng)處理過(guò)直接返回第一次的結(jié)果。超時(shí)后調(diào)用方如果要重試還是用同一個(gè)idempotency_key就能安全地做到至多一次副作用。5.4 循環(huán)調(diào)用死鎖Agent A 調(diào) BB 調(diào) A如果兩個(gè) Agent 互相提供能力而且每次都同步等待對(duì)方的結(jié)果就可能在極端場(chǎng)景下形成死鎖A 在等 B 回應(yīng)B 也在等 A 回應(yīng)。雖然邏輯上不太可能有人故意這樣設(shè)計(jì)但業(yè)務(wù)復(fù)雜后這種隱式循環(huán)是防不勝防的。我在 SDK 層加了一個(gè)簡(jiǎn)單防護(hù)每次消息轉(zhuǎn)發(fā)都帶上一個(gè)max_hop計(jì)數(shù)默認(rèn) 5。每經(jīng)過(guò)一個(gè) Agent 轉(zhuǎn)發(fā)計(jì)數(shù)減 1減到 0 就自動(dòng)返回max_hop_exceeded錯(cuò)誤。這樣即使出現(xiàn)循環(huán)調(diào)用也會(huì)在有限跳數(shù)內(nèi)被切斷而不是無(wú)限死鎖。6. 后續(xù)擴(kuò)展Agent-Reach 還能怎么玩這套框架設(shè)計(jì)時(shí)保留了足夠的擴(kuò)展空間我梳理了幾個(gè)高價(jià)值的后續(xù)方向供真正打算自己搭的人參考多語(yǔ)言 SDK 支持目前官方 SDK 只有 Python 版但協(xié)議層是 HTTP JSON理論上任何語(yǔ)言都能實(shí)現(xiàn)。我自己用 Node.js 寫(xiě)了一個(gè)簡(jiǎn)化版 SDK大約 150 行也能正常接入?yún)f(xié)調(diào)中心。能力負(fù)載均衡如果一個(gè)能力由多個(gè) Agent 提供比如多個(gè)數(shù)據(jù)源可以在路由表里給能力名掛多個(gè) Agent 地址按權(quán)重或輪詢策略分發(fā)請(qǐng)求。審計(jì)日志與追蹤鏈跨 Agent 協(xié)作的調(diào)試是最大的痛點(diǎn)之一。我后面給協(xié)調(diào)中心加了一個(gè)鏈路追蹤日志記錄每條消息從發(fā)出到響應(yīng)的完整時(shí)間線和跳轉(zhuǎn)路徑排查問(wèn)題效率提升了一個(gè)量級(jí)。語(yǔ)義路由目前的尋址是精確能力名匹配。下一步可以把能力描述向量化讓調(diào)用方用自然語(yǔ)言描述需求由協(xié)調(diào)中心通過(guò)語(yǔ)義匹配找到最合適的目標(biāo) Agent。這個(gè)方向有一定工程量但收益明顯尤其適合 Agent 數(shù)量變多后能力名不好管理的場(chǎng)景。在實(shí)際摸索 Agent-Reach 的過(guò)程中我最深的一個(gè)體會(huì)是做多 Agent 協(xié)作真正難的不是讓一堆大模型 API 互相調(diào)用而是把誰(shuí)能找到誰(shuí)、誰(shuí)能調(diào)誰(shuí)、消息怎么傳、失敗怎么辦這套基礎(chǔ)設(shè)施搭得足夠穩(wěn)。消息協(xié)議、注冊(cè)發(fā)現(xiàn)、心跳監(jiān)控、冪等去重——每一個(gè)環(huán)節(jié)單看都很簡(jiǎn)單組合在一起才構(gòu)成了可用的協(xié)作底座。這套東西做完之后我再接入新的 Agent 基本只需要半小時(shí)到現(xiàn)在已經(jīng)穩(wěn)定運(yùn)行了兩個(gè)多月沒(méi)有出過(guò)一次因?yàn)榭蚣鼙旧韺?dǎo)致的消息丟失或路由錯(cuò)亂。如果你手頭的 Agent 們也正各自為戰(zhàn)真的建議動(dòng)手試試這條路投資回報(bào)率比繼續(xù)堆模型能力高得多。