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