如果你正在探索AI Agent開發(fā)可能已經(jīng)發(fā)現(xiàn)了一個關(guān)鍵瓶頸單個Agent能力有限但讓多個Agent協(xié)作起來卻異常困難。每個Agent都有自己的“大腦”模型和“技能”工具如何讓它們高效對話、傳遞信息、協(xié)同完成任務而不是各自為戰(zhàn)這正是Agent間通信Agent-to-Agent Communication簡稱A2A要解決的核心問題。很多人以為A2A只是簡單的消息傳遞就像在兩個程序間發(fā)條字符串。但實際上它遠不止于此。一個健壯的A2A機制需要處理狀態(tài)同步、意圖理解、任務分解、錯誤恢復和資源協(xié)調(diào)。沒有它你的多Agent系統(tǒng)很可能變成一群“聾啞”的智能體空有算力卻無法形成合力。本文將通過一個完整的、可運行的Demo帶你徹底理解A2A的實戰(zhàn)實現(xiàn)。我們將構(gòu)建一個簡單的“旅行規(guī)劃”場景一個行程規(guī)劃Agent負責生成旅行計劃一個預算評估Agent負責審核成本兩者通過明確的通信協(xié)議協(xié)作最終輸出一個合理且經(jīng)濟的方案。你將看到從環(huán)境搭建、Agent定義、通信通道建立到任務編排的完整鏈路并理解背后的設(shè)計思想與常見陷阱。讀完本文你將能清晰區(qū)分Agent間通信與普通API調(diào)用的本質(zhì)不同。使用流行的Agent框架本文以LangChain為例快速搭建一個可工作的A2A原型。掌握設(shè)計Agent通信協(xié)議的關(guān)鍵要素消息格式、會話管理和超時處理。規(guī)避在A2A實踐中最常見的幾個錯誤比如循環(huán)依賴和狀態(tài)混亂。1. A2A要解決的真實問題為什么不是簡單的函數(shù)調(diào)用在深入代碼之前我們必須先厘清一個根本問題既然Agent本質(zhì)上也是代碼模塊為什么不讓一個Agent直接調(diào)用另一個Agent的函數(shù)為什么需要專門的“通信”機制想象一個現(xiàn)實開發(fā)場景你需要構(gòu)建一個智能客服系統(tǒng)其中包含查詢理解Agent、知識檢索Agent和回復生成Agent。如果采用簡單的函數(shù)調(diào)用你會面臨以下挑戰(zhàn)強耦合理解Agent必須精確知道檢索Agent的函數(shù)簽名參數(shù)、返回值。一旦檢索Agent升級接口變化所有調(diào)用方都要修改。阻塞與超時理解Agent調(diào)用檢索Agent的函數(shù)時會同步等待其返回。如果檢索Agent因網(wǎng)絡或計算卡住整個調(diào)用鏈都會阻塞。狀態(tài)管理困難一次用戶對話可能涉及多個Agent的多次交互。函數(shù)調(diào)用模式難以維護跨Agent的會話上下文比如生成Agent如何知道當前回復是基于哪次檢索的結(jié)果。缺乏靈活性你很難動態(tài)地插入一個新的Agent例如一個情感分析Agent到工作流中因為調(diào)用關(guān)系是硬編碼的。A2A的核心思想是解耦與異步協(xié)作。它引入了幾個關(guān)鍵概念消息總線Message Bus或通信通道Agent不直接調(diào)用彼此而是向一個公共的“通道”發(fā)送消息或從其中接收消息。標準化消息格式所有通信都遵循預定義的格式通常包含發(fā)送者、接收者、消息類型、內(nèi)容、會話ID等就像網(wǎng)絡協(xié)議一樣。發(fā)布/訂閱Pub/Sub或點對點P2P模式Agent可以訂閱它關(guān)心的消息類型或者直接指定消息的接收者。這樣做的好處是顯而易見的系統(tǒng)更松耦合、易于擴展、支持異步處理、便于監(jiān)控和調(diào)試通信流。我們的Demo將具體體現(xiàn)這些優(yōu)勢。2. 環(huán)境準備與核心工具選型我們將使用LangChain和OpenAI API來構(gòu)建這個Demo。LangChain是一個強大的LLM應用開發(fā)框架它提供了構(gòu)建和編排Agent的高級抽象非常適合演示A2A概念。即使你之前沒有用過LangChain跟著步驟也能輕松上手。為什么選LangChain因為它內(nèi)置了Agent、Tool、Chain等概念并且對Agent間的結(jié)構(gòu)化通信有良好的支持通過AgentExecutor和自定義Tools模擬通信。這能讓我們聚焦于通信邏輯而非從零搭建Agent基礎(chǔ)設(shè)施。準備工作Python環(huán)境確保你已安裝Python 3.8或更高版本。安裝依賴創(chuàng)建一個新的虛擬環(huán)境然后安裝以下包。pip install langchain langchain-openaiOpenAI API密鑰你需要一個有效的OpenAI API密鑰。將其設(shè)置為環(huán)境變量。# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here安全提示永遠不要將API密鑰硬編碼在代碼中提交到版本庫。項目結(jié)構(gòu)預覽我們將創(chuàng)建以下文件travel_agent_demo/ ├── main.py # 主程序入口編排整個流程 ├── agents/ # Agent定義目錄 │ ├── __init__.py │ ├── planner_agent.py # 行程規(guī)劃Agent │ └── budget_agent.py # 預算評估Agent └── utils/ └── communication.py # 模擬通信通道的工具類3. 核心概念定義我們的Agent與通信協(xié)議在寫代碼前先定義清楚兩個Agent的角色和它們之間的“通信協(xié)議”。行程規(guī)劃Agent (PlannerAgent)職責根據(jù)用戶輸入如“我想去北京玩3天預算中等”生成一份詳細的行程草案包括日期、景點、活動、住宿建議等。輸出一個結(jié)構(gòu)化的行程文本。需要咨詢預算Agent嗎是的。它在生成草案后需要將草案發(fā)送給預算評估Agent進行審核。預算評估Agent (BudgetAgent)職責接收行程草案對其中的各項花費門票、住宿、餐飲、交通進行估算和評估判斷是否超出用戶預算并提供優(yōu)化建議如“某景點門票較貴可替換為...”。輸出一份評估報告包含總估算費用、是否超預算、優(yōu)化建議。行動將評估報告發(fā)回給行程規(guī)劃Agent。通信協(xié)議我們自定義的簡單格式 為了模擬A2A我們將設(shè)計一個簡單的Message類。在實際的LangChain多Agent系統(tǒng)中可能會使用更復雜的AgentAction、AgentFinish或自定義Tools來傳遞信息。# utils/communication.py from typing import Dict, Any, Optional from dataclasses import dataclass dataclass class AgentMessage: 定義Agent間通信的消息格式 sender: str # 發(fā)送者Agent名稱 receiver: str # 接收者Agent名稱 message_type: str # 消息類型如 “query”, “response”, “error” content: Dict[str, Any] # 消息內(nèi)容用字典存儲結(jié)構(gòu)化數(shù)據(jù) session_id: str # 會話ID用于關(guān)聯(lián)同一任務的多輪對話 timestamp: Optional[float] None # 時間戳 def to_dict(self): return { sender: self.sender, receiver: self.receiver, message_type: self.message_type, content: self.content, session_id: self.session_id, timestamp: self.timestamp }同時我們創(chuàng)建一個極簡的CommunicationChannel類來模擬消息總線# utils/communication.py class SimpleCommunicationChannel: 一個簡單的內(nèi)存消息通道用于演示目的。生產(chǎn)環(huán)境應使用消息隊列如RabbitMQ, Redis。 def __init__(self): self.messages [] # 存儲所有消息 def send_message(self, message: AgentMessage): print(f[Channel] {message.sender} - {message.receiver}: {message.message_type}) self.messages.append(message) # 在實際場景中這里會觸發(fā)接收者的消息處理回調(diào) def get_messages_for_agent(self, agent_name: str): 獲取發(fā)送給指定Agent的所有消息簡易過濾 return [msg for msg in self.messages if msg.receiver agent_name]這個通道非?；A(chǔ)僅用于演示消息的發(fā)送和存儲。在真實分布式系統(tǒng)中你會使用成熟的消息中間件。4. 構(gòu)建行程規(guī)劃Agent這個Agent的核心是使用LLMGPT來生成行程。我們將使用LangChain的create_react_agent模式并為其裝備一個關(guān)鍵的“工具”Tool——consult_budget_agent這個工具就是它與其他Agent通信的接口。# agents/planner_agent.py from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from utils.communication import AgentMessage, SimpleCommunicationChannel import uuid class PlannerAgent: def __init__(self, channel: SimpleCommunicationChannel, llmNone): self.name TravelPlanner self.channel channel self.llm llm or ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 定義Agent的“工具”。這里只有一個咨詢預算Agent。 tools [ Tool( nameConsultBudgetAgent, funcself._consult_budget_tool, # 工具的實際執(zhí)行函數(shù) description在生成初步行程草案后必須調(diào)用此工具將草案發(fā)送給預算評估AgentBudgetAgent進行審核。 輸入應該是包含 draft 和 user_budget 的JSON格式字符串。 例如{{draft: ...行程文本..., user_budget: 中等預算}} ) ] # 使用ReAct模式的Prompt模板 planner_prompt PromptTemplate.from_template( 你是一個專業(yè)的旅行行程規(guī)劃師。用戶的需求是{user_input} 你的任務是 1. 根據(jù)用戶需求生成一份詳細、合理、有趣的旅行行程草案。 2. 草案生成后你必須調(diào)用 ConsultBudgetAgent 工具將草案和用戶預算信息發(fā)送給預算評估Agent進行審核。 請開始你的工作。首先思考步驟然后生成草案最后調(diào)用工具。 ) # 創(chuàng)建ReAct Agent agent create_react_agent(llmself.llm, toolstools, promptplanner_prompt) self.agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) def _consult_budget_tool(self, query: str) - str: 這是PlannerAgent與BudgetAgent通信的工具函數(shù)。 它并不直接調(diào)用BudgetAgent而是向通信通道發(fā)送一條消息。 import json try: # 解析輸入來自LLM的調(diào)用 data json.loads(query) travel_draft data.get(draft, ) user_budget data.get(user_budget, ) # 構(gòu)建一條消息 message AgentMessage( senderself.name, receiverBudgetEvaluator, message_typebudget_review_request, content{ travel_draft: travel_draft, user_budget_constraint: user_budget }, session_idself.current_session_id # 需要從外部傳入 ) # 發(fā)送到通道 self.channel.send_message(message) return f消息已發(fā)送給BudgetEvaluator。請等待其評估結(jié)果。會話ID: {self.current_session_id} except Exception as e: return f調(diào)用咨詢工具時出錯{str(e)} def generate_plan(self, user_input: str, session_id: str) - str: 主執(zhí)行方法處理用戶輸入運行Agent并返回最終結(jié)果。 self.current_session_id session_id # 運行AgentExecutor。Agent會根據(jù)Prompt思考并在適當時機調(diào)用我們定義的工具。 result self.agent_executor.invoke({user_input: user_input}) return result.get(output, 行程規(guī)劃未完成。)關(guān)鍵點解析Tool作為通信橋梁ConsultBudgetAgent工具是PlannerAgent對外通信的“手”。當LLM決定需要預算審核時就會調(diào)用這個工具。異步通信模擬工具函數(shù)并不等待回復它只是發(fā)送一條消息到通道。這模擬了異步非阻塞的通信。實際的回復需要通過另一個機制如輪詢或回調(diào)來獲取我們在主流程中模擬這一點。會話IDsession_id至關(guān)重要它將同一任務的所有相關(guān)消息請求和響應關(guān)聯(lián)起來尤其是在并發(fā)處理多個用戶請求時。5. 構(gòu)建預算評估Agent這個Agent相對獨立它監(jiān)聽通道中發(fā)給自己的消息類型為budget_review_request處理消息內(nèi)容然后生成回復消息。# agents/budget_agent.py from langchain_openai import ChatOpenAI from langchain.prompts import PromptTemplate from utils.communication import AgentMessage, SimpleCommunicationChannel import json class BudgetAgent: def __init__(self, channel: SimpleCommunicationChannel, llmNone): self.name BudgetEvaluator self.channel channel self.llm llm or ChatOpenAI(modelgpt-3.5-turbo, temperature0.2) # 溫度低一些評估更穩(wěn)定 def process_incoming_message(self, message: AgentMessage) - AgentMessage: 處理收到的消息核心業(yè)務邏輯在這里。 if message.message_type ! budget_review_request: return None # 忽略其他類型的消息 travel_draft message.content.get(travel_draft) user_budget message.content.get(user_budget_constraint) # 使用LLM進行預算評估 evaluation_prompt PromptTemplate.from_template( 你是一個嚴格的旅行預算評估師。 請審核以下旅行行程草案 --- 行程草案開始 --- {draft} --- 行程草案結(jié)束 --- 用戶的預算要求是{budget}。 請執(zhí)行以下任務 1. 估算此行程的大致總費用按人民幣計算并分項住宿、交通、門票、餐飲簡要說明。 2. 判斷該估算是否明顯符合或超出用戶的預算要求。 3. 提供1-2條具體的優(yōu)化建議以控制成本如替換某個高價項目。 請以JSON格式輸出包含以下字段 - estimated_total_cost: 估算總費用字符串如“約3000元” - is_within_budget: 是否在預算內(nèi) (true/false) - cost_breakdown: 費用分項說明字符串 - optimization_suggestions: 優(yōu)化建議字符串列表 ) chain evaluation_prompt | self.llm evaluation_result_str chain.invoke({draft: travel_draft, budget: user_budget}).content try: evaluation_data json.loads(evaluation_result_str) except json.JSONDecodeError: evaluation_data {error: Failed to parse evaluation result, raw_output: evaluation_result_str} # 構(gòu)建回復消息 reply_message AgentMessage( senderself.name, receivermessage.sender, # 回復給發(fā)送者 message_typebudget_review_response, content{ original_session_id: message.session_id, evaluation: evaluation_data }, session_idmessage.session_id # 保持同一會話ID ) return reply_message def listen_and_process(self): 一個簡易的監(jiān)聽方法用于演示。在實際中這應該是一個持續(xù)運行的服務或事件驅(qū)動。 # 從通道獲取所有發(fā)給自己的消息 my_messages self.channel.get_messages_for_agent(self.name) replies [] for msg in my_messages: reply self.process_incoming_message(msg) if reply: self.channel.send_message(reply) # 將回復發(fā)送回通道 replies.append(reply) return replies關(guān)鍵點解析消息驅(qū)動BudgetAgent是被動觸發(fā)的它通過listen_and_process方法或類似的事件監(jiān)聽器從通道中獲取屬于自己的消息進行處理。結(jié)構(gòu)化輸出我們要求LLM以JSON格式輸出評估結(jié)果這便于程序化解析和后續(xù)處理。這是Agent間通信數(shù)據(jù)格式化的良好實踐。保持會話上下文回復消息中攜帶了original_session_id確保PlannerAgent能將其與最初的請求關(guān)聯(lián)起來。6. 主流程編排讓兩個Agent“對話”起來現(xiàn)在我們需要一個“導演”來協(xié)調(diào)這兩個Agent模擬完整的A2A工作流。這個導演負責初始化Agent、啟動任務、管理通信輪次。# main.py import uuid from utils.communication import SimpleCommunicationChannel from agents.planner_agent import PlannerAgent from agents.budget_agent import BudgetAgent def main(): # 1. 創(chuàng)建共享的通信通道 channel SimpleCommunicationChannel() # 2. 初始化兩個Agent并注入同一個通道實例 planner PlannerAgent(channelchannel) budget_evaluator BudgetAgent(channelchannel) # 3. 模擬用戶輸入 user_request 我想在周末去杭州玩兩天預算有限主要想看看西湖和靈隱寺體驗一下杭幫菜。 session_id str(uuid.uuid4())[:8] # 生成一個簡短的會話ID print(f 開始處理用戶請求 ) print(f用戶: {user_request}) print(f會話ID: {session_id}) print(- * 50) # 4. 啟動PlannerAgent print(f[主流程] 啟動PlannerAgent...) # 注意PlannerAgent在執(zhí)行中會通過工具發(fā)送消息到通道但不會阻塞等待回復。 plan_result planner.generate_plan(user_request, session_id) print(f[主流程] PlannerAgent初步計劃完成。) print(f初步行程草案:\n{plan_result}) print(- * 50) # 5. 模擬讓BudgetAgent處理通道中的請求消息并回復 print(f[主流程] 啟動BudgetAgent處理請求...) replies budget_evaluator.listen_and_process() if replies: print(f[主流程] BudgetAgent已處理 {len(replies)} 條請求并發(fā)出回復。) else: print(f[主流程] 通道中未找到待處理的請求。) print(- * 50) # 6. 模擬PlannerAgent獲取回復在實際中是事件驅(qū)動的 # 這里我們簡化處理直接從通道中查找回復消息。 print(f[主流程] 檢查BudgetAgent的回復...) feedback_messages channel.get_messages_for_agent(planner.name) budget_feedback None for msg in feedback_messages: if msg.message_type budget_review_response and msg.session_id session_id: budget_feedback msg.content.get(evaluation) break # 7. 整合最終結(jié)果 print(f\n 最終整合結(jié)果 ) print(f【用戶需求】{user_request}) print(f\n【生成的行程計劃】) print(plan_result) if budget_feedback: print(f\n【預算評估反饋】) print(f- 估算總費用: {budget_feedback.get(estimated_total_cost, N/A)}) print(f- 是否在預算內(nèi): {是 if budget_feedback.get(is_within_budget) else 否}) print(f- 費用分項: {budget_feedback.get(cost_breakdown, N/A)}) print(f- 優(yōu)化建議: {, .join(budget_feedback.get(optimization_suggestions, []))}) else: print(\n【預算評估反饋】未收到。) print(f\n 處理完成 ) if __name__ __main__: main()7. 運行與效果驗證現(xiàn)在運行這個程序看看兩個Agent是如何協(xié)作的。運行命令cd /path/to/travel_agent_demo python main.py預期輸出示例 開始處理用戶請求 用戶: 我想在周末去杭州玩兩天預算有限主要想看看西湖和靈隱寺體驗一下杭幫菜。 會話ID: a1b2c3d4 -------------------------------------------------- [主流程] 啟動PlannerAgent... [PlannerAgent思考過程...] 我需要先規(guī)劃一個兩天的杭州行程然后咨詢預算。 草案第一天西湖第二天靈隱寺餐飲推薦樓外樓... 現(xiàn)在調(diào)用ConsultBudgetAgent工具。 [Channel] TravelPlanner - BudgetEvaluator: budget_review_request [主流程] PlannerAgent初步計劃完成。 初步行程草案: 第一天游覽西湖... 第二天參觀靈隱寺... 餐飲建議樓外樓、知味觀... -------------------------------------------------- [主流程] 啟動BudgetAgent處理請求... [Channel] BudgetEvaluator - TravelPlanner: budget_review_response [主流程] BudgetAgent已處理 1 條請求并發(fā)出回復。 -------------------------------------------------- [主流程] 檢查BudgetAgent的回復... 最終整合結(jié)果 【用戶需求】我想在周末去杭州玩兩天預算有限主要想看看西湖和靈隱寺體驗一下杭幫菜。 【生成的行程計劃】 第一天游覽西湖... 第二天參觀靈隱寺... 餐飲建議樓外樓、知味觀... 【預算評估反饋】 - 估算總費用: 約1500元 - 是否在預算內(nèi): 是 - 費用分項: 住宿經(jīng)濟型400元交通公交/打車200元門票靈隱寺75元餐飲兩日800元其他50元。 - 優(yōu)化建議: 樓外樓人均較高可考慮新白鹿餐廳等性價比更高的杭幫菜館西湖游船可選擇公共交通船而非私人包船。 處理完成 如何驗證成功流程驗證觀察控制臺輸出確認出現(xiàn)了兩次[Channel] ...的日志這表示消息成功從PlannerAgent發(fā)送到BudgetAgent并且BudgetAgent成功回復。結(jié)果驗證最終輸出中包含了行程計劃和預算評估兩部分且評估內(nèi)容是針對行程草案的具體分析這表明兩個Agent的信息傳遞是有效的。會話隔離驗證你可以嘗試同時運行兩個不同的用戶請求需要擴展主程序觀察不同session_id的消息是否被正確路由和處理沒有混淆。8. 常見問題與排查思路在實現(xiàn)A2A時你可能會遇到以下典型問題問題現(xiàn)象可能原因排查方式解決方案Agent A發(fā)送了消息但Agent B沒反應1. 消息格式不符合接收者的預期類型。2. 通信通道未正確共享兩個Agent使用了不同的通道實例。3. 接收者的監(jiān)聽邏輯未啟動或存在Bug。1. 打印發(fā)送的消息內(nèi)容檢查sender,receiver,message_type是否正確。2. 檢查通道的send_message和get_messages_for_agent方法是否正常工作。3. 在接收者端添加日志確認其process_incoming_message方法被觸發(fā)。1. 統(tǒng)一并嚴格定義消息協(xié)議。2. 確保所有Agent注入的是同一個通道單例。3. 將監(jiān)聽邏輯改為事件驅(qū)動或定時輪詢并確保其持續(xù)運行。消息處理順序錯亂或丟失1. 使用簡易內(nèi)存通道在高并發(fā)下可能丟失消息。2. 沒有處理消息的確認和去重機制。1. 檢查通道實現(xiàn)是否線程安全。2. 為消息添加唯一ID并在處理邏輯中記錄已處理ID防止重復。1.生產(chǎn)環(huán)境務必使用專業(yè)的消息隊列如RabbitMQ、Kafka或Redis Pub/Sub它們提供持久化、確認和順序保證。2. 在消息體中增加message_id和sequence_num字段。Agent陷入循環(huán)對話Agent A等待B的回復B又向A發(fā)起新請求形成死循環(huán)。在消息內(nèi)容或會話上下文中添加對話輪次計數(shù)器。1. 設(shè)計清晰的對話流程和終止條件。2. 在Agent的決策邏輯中設(shè)置最大交互輪次限制。3. 使用AgentExecutor的max_iterations參數(shù)限制單個Agent的執(zhí)行步數(shù)。LLM在Tool調(diào)用時參數(shù)格式錯誤Tool的描述不夠清晰導致LLM生成的調(diào)用參數(shù)不符合func要求的格式。查看LangChain Agent執(zhí)行的詳細日志verboseTrue觀察LLM決定調(diào)用Tool時生成的中間動作Action。1. 優(yōu)化Tool的description用更清晰的例子說明輸入格式。2. 在Tool的func內(nèi)部添加更健壯的參數(shù)解析和錯誤處理并返回友好的錯誤信息給LLM讓其重試。會話Session混淆多個用戶請求同時處理時消息和回復匹配錯誤。檢查每個消息的session_id是否在請求和回復中保持一致。1. 在任務發(fā)起時生成全局唯一的session_id并貫穿整個工作流。2. 在通信通道中提供按session_id過濾消息的查詢方法。9. 最佳實踐與工程化建議Demo跑通只是第一步。要將A2A應用于實際項目你需要考慮更多工程化因素通信層抽象與中間件化不要自己造輪子Demo中的SimpleCommunicationChannel僅用于教學。真實項目應直接集成成熟的消息隊列。將通信層抽象為接口如MessageBus方便后續(xù)切換實現(xiàn)。使用異步處理利用asyncio或框架的異步支持如LangChain的異步API讓Agent在等待回復時不阻塞提高系統(tǒng)吞吐量。設(shè)計健壯的消息協(xié)議定義Schema使用Pydantic或dataclass嚴格定義消息體的結(jié)構(gòu)并進行驗證。包含元數(shù)據(jù)消息中除了業(yè)務數(shù)據(jù)content還應包含message_id、timestamp、version、priority等系統(tǒng)字段。設(shè)計錯誤消息類型定義標準的error消息類型用于在Agent間傳遞處理失敗的信息。Agent的職責與狀態(tài)管理單一職責每個Agent應專注于一個明確、有限的領(lǐng)域如規(guī)劃、評估、執(zhí)行。避免創(chuàng)建“全能”Agent。無狀態(tài)設(shè)計盡可能讓Agent無狀態(tài)其所需的所有上下文都來自傳入的消息。狀態(tài)應保存在外部存儲如數(shù)據(jù)庫、Redis中并通過session_id關(guān)聯(lián)。超時與重試為Agent間的請求-響應設(shè)置超時機制。對于可重試的錯誤實現(xiàn)指數(shù)退避的重試邏輯。可觀測性與調(diào)試全鏈路日志為每個session_id記錄完整的消息流便于追蹤問題?？梢暬ぞ呖紤]使用像LangSmith這樣的平臺來跟蹤和可視化復雜的多Agent工作流。監(jiān)控指標收集消息延遲、處理成功率、Agent調(diào)用次數(shù)等指標。安全與權(quán)限輸入驗證與清理每個Agent在處理來自其他Agent或外部的輸入時都應進行驗證和清理防止提示詞注入或非法操作。權(quán)限控制并非所有Agent都能互相通信?？梢栽O(shè)計一個簡單的權(quán)限層在消息發(fā)送前檢查發(fā)送者是否有權(quán)與接收者通信。通過這個小Demo你應該對Agent間通信A2A的核心價值——解耦、異步協(xié)作、靈活編排——有了直觀感受。它不是一個炫技的概念而是構(gòu)建復雜、可靠AI應用系統(tǒng)的必要基礎(chǔ)設(shè)施。下一步你可以嘗試擴展這個Demo增加第三個Agent如“酒店預訂Agent”實現(xiàn)真正的異步事件循環(huán)或者將內(nèi)存通道替換為Redis向生產(chǎn)環(huán)境邁出第一步。記住清晰的協(xié)議定義和健壯的通信層是讓多個AI智能體真正發(fā)揮協(xié)同智能的關(guān)鍵。