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