：DeepAgents編排、MCP工具接入與A2A通信)
1. 從單體到集群為什么我們需要超級多智能體1.1 一個真實的需求場景去年下半年我接手了一個企業(yè)內部知識助手的項目需求聽起來不復雜幫員工查制度文檔、走審批流程、生成周報。一開始我用的是單體 Agent 方案一個模型加一堆工具函數(shù)跑起來確實快。但上線兩周后問題就來了——查文檔的 Agent 經(jīng)常被審批流程的上下文污染生成周報時又因為要調用太多工具導致響應時間飆到十幾秒。最要命的是我想給文檔檢索單獨換一個更便宜的模型結果發(fā)現(xiàn)整個系統(tǒng)耦合在一起改一處動全身。這個經(jīng)歷讓我徹底轉向了多智能體架構。所謂超級多智能體不是簡單地把幾個 Agent 拼在一起而是要解決三個核心問題可編排誰先誰后、誰調用誰、可互通Agent 之間怎么傳消息、怎么共享狀態(tài)、可擴展新增一個 Agent 要不要改老代碼。這三個問題對應到技術選型上就是 DeepAgents 負責編排、MCP 負責工具接入、A2A 負責 Agent 間通信、Skills 負責能力封裝。如果你正在做 Agent 項目或者被單體 Agent 的耦合問題折磨過這套組合拳值得花時間研究。它不依賴某個特定廠商核心思路可以遷移到任何 Agent 框架上。1.2 四個關鍵詞到底在說什么先把概念理清楚不然后面容易繞暈。DeepAgents在這里指的是一種深度編排的 Agent 架構模式核心思想是把復雜任務拆成有向圖每個節(jié)點是一個 Agent 或一個工具調用邊代表數(shù)據(jù)流和控制流。它和簡單的鏈式調用Chain最大的區(qū)別在于支持條件分支、循環(huán)、并行和人工介入節(jié)點。MCPModel Context Protocol是一個開放協(xié)議解決的是Agent 怎么標準化地調用外部工具和數(shù)據(jù)源的問題。你可以把它理解成 Agent 世界的 USB 接口——不管你是數(shù)據(jù)庫、文件系統(tǒng)還是第三方 API只要實現(xiàn) MCP 協(xié)議任何支持 MCP 的 Agent 都能直接接入。熱搜里有人問mcp 是軟件協(xié)議還是硬件協(xié)議答案是軟件協(xié)議它定義的是通信格式和調用規(guī)范跟硬件層面的協(xié)議比如 USB、I2C完全是兩回事。A2AAgent to Agent解決的是 Agent 之間的通信問題。MCP 管的是 Agent 調工具A2A 管的是 Agent 調 Agent。它定義了 Agent 之間怎么發(fā)現(xiàn)對方、怎么發(fā)消息、怎么處理異步響應。熱搜里出現(xiàn)的a2a spring說明已經(jīng)有人在 Spring 生態(tài)里做 A2A 的集成了。Skills是能力的封裝單元。一個 Skill 可以是一個提示詞模板、一段代碼邏輯、一組工具調用的組合。它的價值在于復用——你寫好一個生成周報的 Skill可以在多個 Agent 里直接引用不用重復實現(xiàn)。熱搜里claude agent skills: a first principles deep dive和codex skills都指向同一個趨勢Skills 正在成為 Agent 能力復用的標準方式。1.3 這套架構適合誰不是所有項目都需要上多智能體。如果你的需求是單輪問答、簡單工具調用單體 Agent 完全夠用硬上多智能體只會增加復雜度。但如果你遇到以下情況這套架構就值得考慮任務流程超過 5 個步驟且步驟之間有依賴關系需要多個專業(yè)領域的 Agent 協(xié)作比如一個查數(shù)據(jù)、一個做分析、一個寫報告工具數(shù)量超過 10 個單體 Agent 的提示詞已經(jīng)塞不下需要給不同 Agent 配不同模型來控制成本系統(tǒng)需要頻繁新增能力不想每次改核心代碼下面我會按照實際搭建的順序從架構設計到落地實現(xiàn)把每個環(huán)節(jié)講透。2. 架構設計編排層、通信層、能力層怎么分2.1 三層架構的職責邊界我在實際項目中把系統(tǒng)分成三層每層職責清晰層與層之間通過標準協(xié)議通信。編排層DeepAgents負責決策任務怎么拆、下一步走哪個節(jié)點、什么時候結束。這一層不關心具體工具怎么調只關心流程控制。我通常用一個有向無環(huán)圖DAG來描述流程節(jié)點類型包括Agent 節(jié)點、工具節(jié)點、條件判斷節(jié)點、人工審核節(jié)點、并行分支節(jié)點。通信層A2A MCP負責傳輸Agent 之間怎么傳消息用 A2AAgent 調工具用 MCP。這一層的關鍵是協(xié)議標準化只要協(xié)議統(tǒng)一底層換實現(xiàn)不影響上層。能力層Skills負責執(zhí)行每個 Skill 是一個獨立的能力單元包含提示詞、工具綁定、輸入輸出定義。Skill 可以注冊到多個 Agent 上實現(xiàn)復用。這樣分層的最大好處是新增一個 Agent 只需要在編排層加節(jié)點、在能力層加 Skill通信層完全不用動。我試過在一個已有 8 個 Agent 的系統(tǒng)里新增一個合同審查Agent從寫 Skill 到接入流程只花了半天。2.2 為什么選 DAG 而不是狀態(tài)機編排層可以用狀態(tài)機也可以用 DAG。我最終選 DAG 的原因是狀態(tài)機適合線性流程但多智能體場景下經(jīng)常需要并行和條件分支。比如同時查三個數(shù)據(jù)源然后匯總這種需求用狀態(tài)機寫起來很別扭DAG 天然支持并行節(jié)點。DAG 的另一個好處是可觀測。每個節(jié)點的輸入輸出都能記錄下來出問題時可以精確定位是哪個節(jié)點出了錯。我在生產(chǎn)環(huán)境里給每個節(jié)點都加了日志埋點排查問題時直接看節(jié)點級別的 trace比看一整條鏈路的日志高效得多。2.3 通信協(xié)議選型MCP 和 A2A 的分工很多人搞不清 MCP 和 A2A 的區(qū)別我用一個類比說明MCP 像是人用工具A2A 像是人跟人說話。你用手電筒工具不需要跟手電筒商量直接按開關就行這是 MCP。但你要跟同事協(xié)作就得說話、等回復、確認理解這是 A2A。具體到實現(xiàn)上MCP 的調用是同步或異步的請求-響應模式Agent 發(fā)一個調用請求工具返回結果。A2A 則更復雜支持流式響應、多輪對話、任務委派。我在項目里用 MCP 接入了文件系統(tǒng)、數(shù)據(jù)庫、搜索引擎三類工具用 A2A 實現(xiàn)了研究 Agent和寫作 Agent之間的協(xié)作。注意MCP 和 A2A 不是互斥的。一個 Agent 可以同時是 MCP 的客戶端調工具和 A2A 的服務端被其他 Agent 調用。這種雙重身份在多智能體系統(tǒng)里很常見。2.4 Skills 的粒度怎么把握Skills 的粒度設計是個經(jīng)驗活。太粗復用性差太細組合起來麻煩。我的原則是一個 Skill 對應一個完整的業(yè)務動作。比如查詢員工考勤是一個 Skill它內部可能調了三個 MCP 工具查數(shù)據(jù)庫、格式化數(shù)據(jù)、計算統(tǒng)計但對外只暴露一個接口。這樣其他 Agent 引用時不用關心內部實現(xiàn)。反過來調用數(shù)據(jù)庫這種太底層的操作不適合做成 Skill它應該是 MCP 工具。Skill 的抽象層級要高于工具低于完整的 Agent。3. 核心細節(jié)MCP 工具接入與 A2A 通信實現(xiàn)3.1 MCP 工具接入的完整流程MCP 的核心是標準化。不管你的工具是 Python 函數(shù)、HTTP API 還是數(shù)據(jù)庫查詢都要包裝成 MCP 協(xié)議規(guī)定的格式。我以接入一個文檔檢索工具為例走一遍完整流程。首先定義工具的元數(shù)據(jù)包括名稱、描述、輸入?yún)?shù) schema。這部分用 JSON Schema 描述目的是讓 Agent 知道這個工具能干什么、需要什么參數(shù)。{ name: search_documents, description: 根據(jù)關鍵詞檢索企業(yè)制度文檔, inputSchema: { type: object, properties: { query: {type: string, description: 檢索關鍵詞}, top_k: {type: integer, default: 5, description: 返回結果數(shù)量} }, required: [query] } }然后實現(xiàn)工具的執(zhí)行邏輯。這部分可以用任何語言寫只要最終暴露成 MCP 服務端即可。我用 Python 寫了一個簡單的實現(xiàn)def search_documents(query: str, top_k: int 5): # 實際檢索邏輯 results vector_store.search(query, top_k) return [{title: r.title, content: r.content} for r in results]最后把工具注冊到 MCP 服務端Agent 通過 MCP 客戶端連接后就能自動發(fā)現(xiàn)并調用這個工具。這里有個坑要注意工具描述的質量直接決定 Agent 會不會正確調用。我一開始寫的描述很簡略結果 Agent 經(jīng)常在不需要檢索的時候也去調這個工具。后來我把描述改得更具體加上僅在用戶詢問制度、流程、規(guī)定時使用這樣的約束誤調用率明顯下降。3.2 A2A 通信的消息格式設計A2A 通信的關鍵是消息格式要統(tǒng)一。我定義了一個基礎消息結構包含發(fā)送方、接收方、消息類型、負載和上下文 ID。{ from: research_agent, to: writing_agent, type: task_delegation, context_id: task_20240115_001, payload: { task: 根據(jù)以下研究結果撰寫報告, data: {...}, requirements: 字數(shù) 800 字包含數(shù)據(jù)引用 } }context_id很重要它讓多個 Agent 之間的多輪交互能關聯(lián)起來。比如研究 Agent 發(fā)了任務后寫作 Agent 可能中途回來問問題這時候靠 context_id 就能找到原始任務上下文。A2A 的響應支持流式和非流式兩種。流式適合長任務寫作 Agent 可以邊寫邊把內容推給調用方非流式適合短任務一次性返回結果。我在項目里默認用流式因為用戶體驗更好而且能提前發(fā)現(xiàn)生成方向跑偏的問題。3.3 Skills 的注冊與發(fā)現(xiàn)機制Skills 要能被復用就需要一個注冊中心。我實現(xiàn)了一個簡單的 Skill Registry每個 Skill 注冊時提供名稱、描述、輸入輸出 schema 和執(zhí)行入口。class SkillRegistry: def __init__(self): self.skills {} def register(self, name, description, schema, handler): self.skills[name] { description: description, schema: schema, handler: handler } def get(self, name): return self.skills.get(name) def list_all(self): return list(self.skills.keys())Agent 在初始化時從 Registry 拉取可用的 Skill 列表根據(jù)任務需要動態(tài)綁定。這樣新增 Skill 不用改 Agent 代碼只要注冊進去就行。熱搜里提到的find skills和skills 推薦反映了大家對這個機制的關注。我的經(jīng)驗是Skill 的發(fā)現(xiàn)不能只靠名稱匹配還要有語義檢索。我后來加了一個基于向量相似度的 Skill 檢索Agent 用自然語言描述需求就能找到合適的 Skill比硬編碼名稱靈活得多。3.4 狀態(tài)管理與上下文傳遞多智能體系統(tǒng)里狀態(tài)管理是最容易出問題的地方。我的做法是全局狀態(tài)用共享存儲局部狀態(tài)用消息傳遞。全局狀態(tài)包括任務 ID、用戶信息、會話歷史存在 Redis 里所有 Agent 都能讀寫。局部狀態(tài)包括當前節(jié)點的中間結果通過 A2A 消息在 Agent 之間傳遞不落盤。這樣設計的好處是避免了狀態(tài)不一致。我踩過的坑是早期把所有狀態(tài)都放在消息里傳結果一個 Agent 改了狀態(tài)但沒傳給下一個導致數(shù)據(jù)丟失。后來把關鍵狀態(tài)抽到共享存儲問題就解決了。提示共享存儲要加版本號或樂觀鎖防止多個 Agent 并發(fā)寫同一份數(shù)據(jù)時互相覆蓋。4. 實操落地從零搭建一個多智能體系統(tǒng)4.1 環(huán)境準備與依賴安裝我以 Python 技術棧為例列出核心依賴。如果你用其他語言思路是一樣的只是庫不同。pip install fastapi uvicorn redis pydantic pip install mcp-client mcp-server pip install a2a-sdk pip install langgraph # 用于 DAG 編排MCP 和 A2A 的 SDK 目前還在快速迭代建議鎖定版本號避免升級導致接口不兼容。我在項目里用 requirements.txt 固定了版本每次升級前先在測試環(huán)境驗證。Redis 用于共享狀態(tài)存儲FastAPI 用于暴露 HTTP 接口Pydantic 用于數(shù)據(jù)校驗。LangGraph 是我常用的 DAG 編排庫它原生支持條件分支和循環(huán)比手寫狀態(tài)機省事。4.2 定義第一個 Agent 和 Skill先定義一個最簡單的文檔檢索 Agent它綁定一個檢索文檔的 Skill。from pydantic import BaseModel class SearchInput(BaseModel): query: str top_k: int 5 class SearchOutput(BaseModel): results: list total: int def search_handler(input: SearchInput) - SearchOutput: results vector_store.search(input.query, input.top_k) return SearchOutput(resultsresults, totallen(results)) # 注冊 Skill registry.register( namesearch_documents, description檢索企業(yè)制度文檔適用于查詢規(guī)定、流程、政策, schema{input: SearchInput, output: SearchOutput}, handlersearch_handler )然后定義 Agent它從 Registry 獲取 Skill通過 MCP 調用工具通過 A2A 接收任務。class DocumentAgent: def __init__(self, registry, mcp_client, a2a_client): self.registry registry self.mcp mcp_client self.a2a a2a_client self.skill registry.get(search_documents) async def handle(self, message): input_data SearchInput(**message.payload) result self.skill[handler](input_data) return {status: success, data: result.dict()}這個 Agent 的結構很清晰接收消息、解析輸入、執(zhí)行 Skill、返回結果。新增能力只需要注冊新 Skill 并綁定到 Agent 上。4.3 編排多個 Agent 的 DAG有了單個 Agent 后用 LangGraph 把它們編排成 DAG。我以一個員工咨詢場景為例用戶提問后先判斷問題類型如果是制度類問題走文檔 Agent如果是數(shù)據(jù)類問題走數(shù)據(jù) Agent最后統(tǒng)一由匯總 Agent 生成回答。from langgraph.graph import StateGraph, END def route_question(state): if 制度 in state[question] or 流程 in state[question]: return document_agent elif 數(shù)據(jù) in state[question] or 統(tǒng)計 in state[question]: return data_agent else: return general_agent graph StateGraph(AgentState) graph.add_node(router, route_question) graph.add_node(document_agent, document_node) graph.add_node(data_agent, data_node) graph.add_node(general_agent, general_node) graph.add_node(summarizer, summarize_node) graph.set_entry_point(router) graph.add_conditional_edges(router, route_question, { document_agent: document_agent, data_agent: data_agent, general_agent: general_agent }) graph.add_edge(document_agent, summarizer) graph.add_edge(data_agent, summarizer) graph.add_edge(general_agent, summarizer) graph.add_edge(summarizer, END) app graph.compile()這個 DAG 的關鍵點是路由節(jié)點。路由可以用規(guī)則實現(xiàn)也可以用一個小模型做意圖分類。我在項目里用的是規(guī)則加模型兜底常見問題走規(guī)則規(guī)則匹配不到時調模型判斷兼顧速度和準確率。4.4 參數(shù)計算與性能調優(yōu)多智能體系統(tǒng)的性能瓶頸通常在兩個地方Agent 間的通信延遲和模型調用次數(shù)。我做過一組實測數(shù)據(jù)供參考。場景Agent 數(shù)量平均響應時間模型調用次數(shù)單體 Agent12.1s1-2雙 Agent 串行23.8s2-3三 Agent 并行34.2s3-4五 Agent 混合56.5s5-7從數(shù)據(jù)看Agent 數(shù)量增加會帶來明顯的延遲增長。優(yōu)化方向有三個一是并行化能并行的節(jié)點不要串行二是緩存相同輸入的結果緩存起來三是模型分級簡單任務用便宜的小模型。我在項目里給每個 Agent 配了不同的模型路由用 7B 小模型文檔檢索用 13B 模型匯總生成用 70B 大模型。這樣整體成本降了約 60%響應時間只增加了 15%。4.5 部署與監(jiān)控部署上我推薦容器化每個 Agent 一個容器通過消息隊列或 HTTP 通信。這樣單個 Agent 出問題不影響整體也方便水平擴展。監(jiān)控要覆蓋三個層面節(jié)點級別的執(zhí)行時間、Agent 級別的調用成功率、系統(tǒng)級別的吞吐量。我用 Prometheus 采集指標Grafana 做可視化。關鍵指標包括每個節(jié)點的 P99 延遲、A2A 消息的丟失率、MCP 工具調用的錯誤率。注意多智能體系統(tǒng)的調試比單體復雜得多。建議在開發(fā)階段開啟全鏈路日志記錄每個節(jié)點的輸入輸出和 Agent 間的消息。生產(chǎn)環(huán)境可以采樣記錄避免日志量過大。5. 常見問題與排查技巧實錄5.1 Agent 之間消息丟失或亂序這是 A2A 通信最常見的問題。表現(xiàn)是下游 Agent 收不到消息或者收到的消息順序不對。排查思路先看消息隊列的積壓情況如果積壓嚴重說明消費能力不足再看消息的 context_id 是否一致不一致說明發(fā)送方生成邏輯有問題最后看網(wǎng)絡層是否有丟包。我的解決方案是給消息加序號和確認機制。發(fā)送方給每條消息編號接收方收到后回 ACK超時未收到 ACK 就重發(fā)。這樣能保證消息不丟但要注意去重避免重發(fā)導致重復處理。5.2 MCP 工具調用超時MCP 工具調用超時通常有三個原因工具本身執(zhí)行慢、網(wǎng)絡延遲、Agent 等待策略不合理。我遇到過一次數(shù)據(jù)庫查詢工具超時排查發(fā)現(xiàn)是查詢沒加索引全表掃描導致。加了索引后從 8 秒降到 200 毫秒。所以工具本身的性能要先保證。網(wǎng)絡延遲方面如果 MCP 服務端和 Agent 不在同一臺機器建議加連接池和重試機制。Agent 側的等待策略要設置合理的超時時間我一般設 30 秒超過就返回降級結果。5.3 Skill 沖突與優(yōu)先級當多個 Skill 功能重疊時Agent 可能選錯。比如同時有查詢文檔和搜索知識庫兩個 SkillAgent 不知道該用哪個。解決辦法是給 Skill 加優(yōu)先級和適用場景標簽。Agent 選擇時先按場景過濾再按優(yōu)先級排序。另外Skill 的描述要寫清楚邊界避免功能重疊。我在項目里維護了一個 Skill 沖突檢查清單新增 Skill 時先檢查是否與已有 Skill 重疊重疊的要么合并要么明確分工。5.4 常見問題速查表問題現(xiàn)象可能原因排查方法解決方案Agent 不響應消息未送達檢查消息隊列和 ACK加重試和確認機制工具調用失敗參數(shù)格式錯誤查看 MCP 調用日志校驗輸入 schema響應時間過長串行節(jié)點過多分析 DAG 執(zhí)行路徑并行化可并行節(jié)點結果不一致狀態(tài)競爭檢查共享存儲讀寫加鎖或版本號Skill 選錯描述不清晰查看 Agent 決策日志優(yōu)化 Skill 描述和優(yōu)先級模型成本過高大模型調用過多統(tǒng)計各 Agent 模型用量分級用模型加緩存5.5 幾個踩過的坑第一個坑是過度設計。我一開始給每個 Agent 都配了完整的 MCP 和 A2A 能力結果系統(tǒng)復雜度飆升調試困難。后來發(fā)現(xiàn)很多 Agent 只需要 MCP 調工具不需要 A2A砍掉后系統(tǒng)清爽很多。第二個坑是忽略冪等性。A2A 消息重發(fā)時如果接收方?jīng)]有冪等處理會導致重復執(zhí)行。比如重復扣款、重復發(fā)郵件。我的做法是給每個任務加唯一 ID接收方處理前先檢查是否已處理過。第三個坑是日志太多。全鏈路日志在調試時很有用但生產(chǎn)環(huán)境日志量太大影響性能。后來改成采樣記錄只對錯誤和慢請求記錄完整日志。第四個坑是Skill 版本管理。Skill 更新后正在運行的 Agent 可能還在用舊版本。我的解決方案是給 Skill 加版本號Agent 綁定具體版本更新時灰度切換。5.6 性能優(yōu)化的幾個實用技巧緩存是最有效的優(yōu)化手段。我在 MCP 工具層加了結果緩存相同參數(shù)的調用直接返回緩存結果。對于文檔檢索這類讀多寫少的場景緩存命中率能到 40% 以上。批處理也很重要。如果多個 Agent 需要調用同一個工具可以合并成一次批量調用。比如三個 Agent 都要查用戶信息合并成一次查詢比三次單獨查詢快得多。異步化是另一個方向。A2A 通信默認是異步的但很多實現(xiàn)里 Agent 處理消息是同步的。改成異步后Agent 可以在等待下游響應時處理其他任務吞吐量能提升 2-3 倍。最后是模型分級。不是所有任務都需要大模型。路由、分類、簡單抽取用 7B 模型就夠只有生成和復雜推理才需要大模型。我在項目里做了模型分級后成本降了 60%效果幾乎沒影響。這套架構我用了大半年從最初的 3 個 Agent 擴展到現(xiàn)在的 12 個新增能力基本不用改核心代碼。如果你也在做類似的項目建議先從兩三個 Agent 的小系統(tǒng)開始跑通了再逐步擴展。一上來就搞十幾個 Agent調試成本會讓你懷疑人生。 超級多智能體架構實戰(zhàn)：DeepAgents編排、MCP工具接入與A2A通信
1. 從單體到集群為什么我們需要超級多智能體1.1 一個真實的需求場景去年下半年我接手了一個企業(yè)內部知識助手的項目需求聽起來不復雜幫員工查制度文檔、走審批流程、生成周報。一開始我用的是單體 Agent 方案一個模型加一堆工具函數(shù)跑起來確實快。但上線兩周后問題就來了——查文檔的 Agent 經(jīng)常被審批流程的上下文污染生成周報時又因為要調用太多工具導致響應時間飆到十幾秒。最要命的是我想給文檔檢索單獨換一個更便宜的模型結果發(fā)現(xiàn)整個系統(tǒng)耦合在一起改一處動全身。這個經(jīng)歷讓我徹底轉向了多智能體架構。所謂超級多智能體不是簡單地把幾個 Agent 拼在一起而是要解決三個核心問題可編排誰先誰后、誰調用誰、可互通Agent 之間怎么傳消息、怎么共享狀態(tài)、可擴展新增一個 Agent 要不要改老代碼。這三個問題對應到技術選型上就是 DeepAgents 負責編排、MCP 負責工具接入、A2A 負責 Agent 間通信、Skills 負責能力封裝。如果你正在做 Agent 項目或者被單體 Agent 的耦合問題折磨過這套組合拳值得花時間研究。它不依賴某個特定廠商核心思路可以遷移到任何 Agent 框架上。1.2 四個關鍵詞到底在說什么先把概念理清楚不然后面容易繞暈。DeepAgents在這里指的是一種深度編排的 Agent 架構模式核心思想是把復雜任務拆成有向圖每個節(jié)點是一個 Agent 或一個工具調用邊代表數(shù)據(jù)流和控制流。它和簡單的鏈式調用Chain最大的區(qū)別在于支持條件分支、循環(huán)、并行和人工介入節(jié)點。MCPModel Context Protocol是一個開放協(xié)議解決的是Agent 怎么標準化地調用外部工具和數(shù)據(jù)源的問題。你可以把它理解成 Agent 世界的 USB 接口——不管你是數(shù)據(jù)庫、文件系統(tǒng)還是第三方 API只要實現(xiàn) MCP 協(xié)議任何支持 MCP 的 Agent 都能直接接入。熱搜里有人問mcp 是軟件協(xié)議還是硬件協(xié)議答案是軟件協(xié)議它定義的是通信格式和調用規(guī)范跟硬件層面的協(xié)議比如 USB、I2C完全是兩回事。A2AAgent to Agent解決的是 Agent 之間的通信問題。MCP 管的是 Agent 調工具A2A 管的是 Agent 調 Agent。它定義了 Agent 之間怎么發(fā)現(xiàn)對方、怎么發(fā)消息、怎么處理異步響應。熱搜里出現(xiàn)的a2a spring說明已經(jīng)有人在 Spring 生態(tài)里做 A2A 的集成了。Skills是能力的封裝單元。一個 Skill 可以是一個提示詞模板、一段代碼邏輯、一組工具調用的組合。它的價值在于復用——你寫好一個生成周報的 Skill可以在多個 Agent 里直接引用不用重復實現(xiàn)。熱搜里claude agent skills: a first principles deep dive和codex skills都指向同一個趨勢Skills 正在成為 Agent 能力復用的標準方式。1.3 這套架構適合誰不是所有項目都需要上多智能體。如果你的需求是單輪問答、簡單工具調用單體 Agent 完全夠用硬上多智能體只會增加復雜度。但如果你遇到以下情況這套架構就值得考慮任務流程超過 5 個步驟且步驟之間有依賴關系需要多個專業(yè)領域的 Agent 協(xié)作比如一個查數(shù)據(jù)、一個做分析、一個寫報告工具數(shù)量超過 10 個單體 Agent 的提示詞已經(jīng)塞不下需要給不同 Agent 配不同模型來控制成本系統(tǒng)需要頻繁新增能力不想每次改核心代碼下面我會按照實際搭建的順序從架構設計到落地實現(xiàn)把每個環(huán)節(jié)講透。2. 架構設計編排層、通信層、能力層怎么分2.1 三層架構的職責邊界我在實際項目中把系統(tǒng)分成三層每層職責清晰層與層之間通過標準協(xié)議通信。編排層DeepAgents負責決策任務怎么拆、下一步走哪個節(jié)點、什么時候結束。這一層不關心具體工具怎么調只關心流程控制。我通常用一個有向無環(huán)圖DAG來描述流程節(jié)點類型包括Agent 節(jié)點、工具節(jié)點、條件判斷節(jié)點、人工審核節(jié)點、并行分支節(jié)點。通信層A2A MCP負責傳輸Agent 之間怎么傳消息用 A2AAgent 調工具用 MCP。這一層的關鍵是協(xié)議標準化只要協(xié)議統(tǒng)一底層換實現(xiàn)不影響上層。能力層Skills負責執(zhí)行每個 Skill 是一個獨立的能力單元包含提示詞、工具綁定、輸入輸出定義。Skill 可以注冊到多個 Agent 上實現(xiàn)復用。這樣分層的最大好處是新增一個 Agent 只需要在編排層加節(jié)點、在能力層加 Skill通信層完全不用動。我試過在一個已有 8 個 Agent 的系統(tǒng)里新增一個合同審查Agent從寫 Skill 到接入流程只花了半天。2.2 為什么選 DAG 而不是狀態(tài)機編排層可以用狀態(tài)機也可以用 DAG。我最終選 DAG 的原因是狀態(tài)機適合線性流程但多智能體場景下經(jīng)常需要并行和條件分支。比如同時查三個數(shù)據(jù)源然后匯總這種需求用狀態(tài)機寫起來很別扭DAG 天然支持并行節(jié)點。DAG 的另一個好處是可觀測。每個節(jié)點的輸入輸出都能記錄下來出問題時可以精確定位是哪個節(jié)點出了錯。我在生產(chǎn)環(huán)境里給每個節(jié)點都加了日志埋點排查問題時直接看節(jié)點級別的 trace比看一整條鏈路的日志高效得多。2.3 通信協(xié)議選型MCP 和 A2A 的分工很多人搞不清 MCP 和 A2A 的區(qū)別我用一個類比說明MCP 像是人用工具A2A 像是人跟人說話。你用手電筒工具不需要跟手電筒商量直接按開關就行這是 MCP。但你要跟同事協(xié)作就得說話、等回復、確認理解這是 A2A。具體到實現(xiàn)上MCP 的調用是同步或異步的請求-響應模式Agent 發(fā)一個調用請求工具返回結果。A2A 則更復雜支持流式響應、多輪對話、任務委派。我在項目里用 MCP 接入了文件系統(tǒng)、數(shù)據(jù)庫、搜索引擎三類工具用 A2A 實現(xiàn)了研究 Agent和寫作 Agent之間的協(xié)作。注意MCP 和 A2A 不是互斥的。一個 Agent 可以同時是 MCP 的客戶端調工具和 A2A 的服務端被其他 Agent 調用。這種雙重身份在多智能體系統(tǒng)里很常見。2.4 Skills 的粒度怎么把握Skills 的粒度設計是個經(jīng)驗活。太粗復用性差太細組合起來麻煩。我的原則是一個 Skill 對應一個完整的業(yè)務動作。比如查詢員工考勤是一個 Skill它內部可能調了三個 MCP 工具查數(shù)據(jù)庫、格式化數(shù)據(jù)、計算統(tǒng)計但對外只暴露一個接口。這樣其他 Agent 引用時不用關心內部實現(xiàn)。反過來調用數(shù)據(jù)庫這種太底層的操作不適合做成 Skill它應該是 MCP 工具。Skill 的抽象層級要高于工具低于完整的 Agent。3. 核心細節(jié)MCP 工具接入與 A2A 通信實現(xiàn)3.1 MCP 工具接入的完整流程MCP 的核心是標準化。不管你的工具是 Python 函數(shù)、HTTP API 還是數(shù)據(jù)庫查詢都要包裝成 MCP 協(xié)議規(guī)定的格式。我以接入一個文檔檢索工具為例走一遍完整流程。首先定義工具的元數(shù)據(jù)包括名稱、描述、輸入?yún)?shù) schema。這部分用 JSON Schema 描述目的是讓 Agent 知道這個工具能干什么、需要什么參數(shù)。{ name: search_documents, description: 根據(jù)關鍵詞檢索企業(yè)制度文檔, inputSchema: { type: object, properties: { query: {type: string, description: 檢索關鍵詞}, top_k: {type: integer, default: 5, description: 返回結果數(shù)量} }, required: [query] } }然后實現(xiàn)工具的執(zhí)行邏輯。這部分可以用任何語言寫只要最終暴露成 MCP 服務端即可。我用 Python 寫了一個簡單的實現(xiàn)def search_documents(query: str, top_k: int 5): # 實際檢索邏輯 results vector_store.search(query, top_k) return [{title: r.title, content: r.content} for r in results]最后把工具注冊到 MCP 服務端Agent 通過 MCP 客戶端連接后就能自動發(fā)現(xiàn)并調用這個工具。這里有個坑要注意工具描述的質量直接決定 Agent 會不會正確調用。我一開始寫的描述很簡略結果 Agent 經(jīng)常在不需要檢索的時候也去調這個工具。后來我把描述改得更具體加上僅在用戶詢問制度、流程、規(guī)定時使用這樣的約束誤調用率明顯下降。3.2 A2A 通信的消息格式設計A2A 通信的關鍵是消息格式要統(tǒng)一。我定義了一個基礎消息結構包含發(fā)送方、接收方、消息類型、負載和上下文 ID。{ from: research_agent, to: writing_agent, type: task_delegation, context_id: task_20240115_001, payload: { task: 根據(jù)以下研究結果撰寫報告, data: {...}, requirements: 字數(shù) 800 字包含數(shù)據(jù)引用 } }context_id很重要它讓多個 Agent 之間的多輪交互能關聯(lián)起來。比如研究 Agent 發(fā)了任務后寫作 Agent 可能中途回來問問題這時候靠 context_id 就能找到原始任務上下文。A2A 的響應支持流式和非流式兩種。流式適合長任務寫作 Agent 可以邊寫邊把內容推給調用方非流式適合短任務一次性返回結果。我在項目里默認用流式因為用戶體驗更好而且能提前發(fā)現(xiàn)生成方向跑偏的問題。3.3 Skills 的注冊與發(fā)現(xiàn)機制Skills 要能被復用就需要一個注冊中心。我實現(xiàn)了一個簡單的 Skill Registry每個 Skill 注冊時提供名稱、描述、輸入輸出 schema 和執(zhí)行入口。class SkillRegistry: def __init__(self): self.skills {} def register(self, name, description, schema, handler): self.skills[name] { description: description, schema: schema, handler: handler } def get(self, name): return self.skills.get(name) def list_all(self): return list(self.skills.keys())Agent 在初始化時從 Registry 拉取可用的 Skill 列表根據(jù)任務需要動態(tài)綁定。這樣新增 Skill 不用改 Agent 代碼只要注冊進去就行。熱搜里提到的find skills和skills 推薦反映了大家對這個機制的關注。我的經(jīng)驗是Skill 的發(fā)現(xiàn)不能只靠名稱匹配還要有語義檢索。我后來加了一個基于向量相似度的 Skill 檢索Agent 用自然語言描述需求就能找到合適的 Skill比硬編碼名稱靈活得多。3.4 狀態(tài)管理與上下文傳遞多智能體系統(tǒng)里狀態(tài)管理是最容易出問題的地方。我的做法是全局狀態(tài)用共享存儲局部狀態(tài)用消息傳遞。全局狀態(tài)包括任務 ID、用戶信息、會話歷史存在 Redis 里所有 Agent 都能讀寫。局部狀態(tài)包括當前節(jié)點的中間結果通過 A2A 消息在 Agent 之間傳遞不落盤。這樣設計的好處是避免了狀態(tài)不一致。我踩過的坑是早期把所有狀態(tài)都放在消息里傳結果一個 Agent 改了狀態(tài)但沒傳給下一個導致數(shù)據(jù)丟失。后來把關鍵狀態(tài)抽到共享存儲問題就解決了。提示共享存儲要加版本號或樂觀鎖防止多個 Agent 并發(fā)寫同一份數(shù)據(jù)時互相覆蓋。4. 實操落地從零搭建一個多智能體系統(tǒng)4.1 環(huán)境準備與依賴安裝我以 Python 技術棧為例列出核心依賴。如果你用其他語言思路是一樣的只是庫不同。pip install fastapi uvicorn redis pydantic pip install mcp-client mcp-server pip install a2a-sdk pip install langgraph # 用于 DAG 編排MCP 和 A2A 的 SDK 目前還在快速迭代建議鎖定版本號避免升級導致接口不兼容。我在項目里用 requirements.txt 固定了版本每次升級前先在測試環(huán)境驗證。Redis 用于共享狀態(tài)存儲FastAPI 用于暴露 HTTP 接口Pydantic 用于數(shù)據(jù)校驗。LangGraph 是我常用的 DAG 編排庫它原生支持條件分支和循環(huán)比手寫狀態(tài)機省事。4.2 定義第一個 Agent 和 Skill先定義一個最簡單的文檔檢索 Agent它綁定一個檢索文檔的 Skill。from pydantic import BaseModel class SearchInput(BaseModel): query: str top_k: int 5 class SearchOutput(BaseModel): results: list total: int def search_handler(input: SearchInput) - SearchOutput: results vector_store.search(input.query, input.top_k) return SearchOutput(resultsresults, totallen(results)) # 注冊 Skill registry.register( namesearch_documents, description檢索企業(yè)制度文檔適用于查詢規(guī)定、流程、政策, schema{input: SearchInput, output: SearchOutput}, handlersearch_handler )然后定義 Agent它從 Registry 獲取 Skill通過 MCP 調用工具通過 A2A 接收任務。class DocumentAgent: def __init__(self, registry, mcp_client, a2a_client): self.registry registry self.mcp mcp_client self.a2a a2a_client self.skill registry.get(search_documents) async def handle(self, message): input_data SearchInput(**message.payload) result self.skill[handler](input_data) return {status: success, data: result.dict()}這個 Agent 的結構很清晰接收消息、解析輸入、執(zhí)行 Skill、返回結果。新增能力只需要注冊新 Skill 并綁定到 Agent 上。4.3 編排多個 Agent 的 DAG有了單個 Agent 后用 LangGraph 把它們編排成 DAG。我以一個員工咨詢場景為例用戶提問后先判斷問題類型如果是制度類問題走文檔 Agent如果是數(shù)據(jù)類問題走數(shù)據(jù) Agent最后統(tǒng)一由匯總 Agent 生成回答。from langgraph.graph import StateGraph, END def route_question(state): if 制度 in state[question] or 流程 in state[question]: return document_agent elif 數(shù)據(jù) in state[question] or 統(tǒng)計 in state[question]: return data_agent else: return general_agent graph StateGraph(AgentState) graph.add_node(router, route_question) graph.add_node(document_agent, document_node) graph.add_node(data_agent, data_node) graph.add_node(general_agent, general_node) graph.add_node(summarizer, summarize_node) graph.set_entry_point(router) graph.add_conditional_edges(router, route_question, { document_agent: document_agent, data_agent: data_agent, general_agent: general_agent }) graph.add_edge(document_agent, summarizer) graph.add_edge(data_agent, summarizer) graph.add_edge(general_agent, summarizer) graph.add_edge(summarizer, END) app graph.compile()這個 DAG 的關鍵點是路由節(jié)點。路由可以用規(guī)則實現(xiàn)也可以用一個小模型做意圖分類。我在項目里用的是規(guī)則加模型兜底常見問題走規(guī)則規(guī)則匹配不到時調模型判斷兼顧速度和準確率。4.4 參數(shù)計算與性能調優(yōu)多智能體系統(tǒng)的性能瓶頸通常在兩個地方Agent 間的通信延遲和模型調用次數(shù)。我做過一組實測數(shù)據(jù)供參考。場景Agent 數(shù)量平均響應時間模型調用次數(shù)單體 Agent12.1s1-2雙 Agent 串行23.8s2-3三 Agent 并行34.2s3-4五 Agent 混合56.5s5-7從數(shù)據(jù)看Agent 數(shù)量增加會帶來明顯的延遲增長。優(yōu)化方向有三個一是并行化能并行的節(jié)點不要串行二是緩存相同輸入的結果緩存起來三是模型分級簡單任務用便宜的小模型。我在項目里給每個 Agent 配了不同的模型路由用 7B 小模型文檔檢索用 13B 模型匯總生成用 70B 大模型。這樣整體成本降了約 60%響應時間只增加了 15%。4.5 部署與監(jiān)控部署上我推薦容器化每個 Agent 一個容器通過消息隊列或 HTTP 通信。這樣單個 Agent 出問題不影響整體也方便水平擴展。監(jiān)控要覆蓋三個層面節(jié)點級別的執(zhí)行時間、Agent 級別的調用成功率、系統(tǒng)級別的吞吐量。我用 Prometheus 采集指標Grafana 做可視化。關鍵指標包括每個節(jié)點的 P99 延遲、A2A 消息的丟失率、MCP 工具調用的錯誤率。注意多智能體系統(tǒng)的調試比單體復雜得多。建議在開發(fā)階段開啟全鏈路日志記錄每個節(jié)點的輸入輸出和 Agent 間的消息。生產(chǎn)環(huán)境可以采樣記錄避免日志量過大。5. 常見問題與排查技巧實錄5.1 Agent 之間消息丟失或亂序這是 A2A 通信最常見的問題。表現(xiàn)是下游 Agent 收不到消息或者收到的消息順序不對。排查思路先看消息隊列的積壓情況如果積壓嚴重說明消費能力不足再看消息的 context_id 是否一致不一致說明發(fā)送方生成邏輯有問題最后看網(wǎng)絡層是否有丟包。我的解決方案是給消息加序號和確認機制。發(fā)送方給每條消息編號接收方收到后回 ACK超時未收到 ACK 就重發(fā)。這樣能保證消息不丟但要注意去重避免重發(fā)導致重復處理。5.2 MCP 工具調用超時MCP 工具調用超時通常有三個原因工具本身執(zhí)行慢、網(wǎng)絡延遲、Agent 等待策略不合理。我遇到過一次數(shù)據(jù)庫查詢工具超時排查發(fā)現(xiàn)是查詢沒加索引全表掃描導致。加了索引后從 8 秒降到 200 毫秒。所以工具本身的性能要先保證。網(wǎng)絡延遲方面如果 MCP 服務端和 Agent 不在同一臺機器建議加連接池和重試機制。Agent 側的等待策略要設置合理的超時時間我一般設 30 秒超過就返回降級結果。5.3 Skill 沖突與優(yōu)先級當多個 Skill 功能重疊時Agent 可能選錯。比如同時有查詢文檔和搜索知識庫兩個 SkillAgent 不知道該用哪個。解決辦法是給 Skill 加優(yōu)先級和適用場景標簽。Agent 選擇時先按場景過濾再按優(yōu)先級排序。另外Skill 的描述要寫清楚邊界避免功能重疊。我在項目里維護了一個 Skill 沖突檢查清單新增 Skill 時先檢查是否與已有 Skill 重疊重疊的要么合并要么明確分工。5.4 常見問題速查表問題現(xiàn)象可能原因排查方法解決方案Agent 不響應消息未送達檢查消息隊列和 ACK加重試和確認機制工具調用失敗參數(shù)格式錯誤查看 MCP 調用日志校驗輸入 schema響應時間過長串行節(jié)點過多分析 DAG 執(zhí)行路徑并行化可并行節(jié)點結果不一致狀態(tài)競爭檢查共享存儲讀寫加鎖或版本號Skill 選錯描述不清晰查看 Agent 決策日志優(yōu)化 Skill 描述和優(yōu)先級模型成本過高大模型調用過多統(tǒng)計各 Agent 模型用量分級用模型加緩存5.5 幾個踩過的坑第一個坑是過度設計。我一開始給每個 Agent 都配了完整的 MCP 和 A2A 能力結果系統(tǒng)復雜度飆升調試困難。后來發(fā)現(xiàn)很多 Agent 只需要 MCP 調工具不需要 A2A砍掉后系統(tǒng)清爽很多。第二個坑是忽略冪等性。A2A 消息重發(fā)時如果接收方?jīng)]有冪等處理會導致重復執(zhí)行。比如重復扣款、重復發(fā)郵件。我的做法是給每個任務加唯一 ID接收方處理前先檢查是否已處理過。第三個坑是日志太多。全鏈路日志在調試時很有用但生產(chǎn)環(huán)境日志量太大影響性能。后來改成采樣記錄只對錯誤和慢請求記錄完整日志。第四個坑是Skill 版本管理。Skill 更新后正在運行的 Agent 可能還在用舊版本。我的解決方案是給 Skill 加版本號Agent 綁定具體版本更新時灰度切換。5.6 性能優(yōu)化的幾個實用技巧緩存是最有效的優(yōu)化手段。我在 MCP 工具層加了結果緩存相同參數(shù)的調用直接返回緩存結果。對于文檔檢索這類讀多寫少的場景緩存命中率能到 40% 以上。批處理也很重要。如果多個 Agent 需要調用同一個工具可以合并成一次批量調用。比如三個 Agent 都要查用戶信息合并成一次查詢比三次單獨查詢快得多。異步化是另一個方向。A2A 通信默認是異步的但很多實現(xiàn)里 Agent 處理消息是同步的。改成異步后Agent 可以在等待下游響應時處理其他任務吞吐量能提升 2-3 倍。最后是模型分級。不是所有任務都需要大模型。路由、分類、簡單抽取用 7B 模型就夠只有生成和復雜推理才需要大模型。我在項目里做了模型分級后成本降了 60%效果幾乎沒影響。這套架構我用了大半年從最初的 3 個 Agent 擴展到現(xiàn)在的 12 個新增能力基本不用改核心代碼。如果你也在做類似的項目建議先從兩三個 Agent 的小系統(tǒng)開始跑通了再逐步擴展。一上來就搞十幾個 Agent調試成本會讓你懷疑人生。