1. 先搞清楚 A2A 通信到底要解决什么问题如果你正在接触 AI Agent 开发或者看过一些关于多智能体协作的讨论可能会觉得“Agent 之间如何对话”是个很酷但有点抽象的概念。Agent2AgentA2A通信说白了就是让两个或多个独立的 AI 智能体程序能够像人和人之间发消息、打电话一样互相传递信息、请求和结果。这个需求在什么场景下会变得非常实际举个例子你有一个专门负责分析数据的 Agent和一个专门负责生成图表的 Agent。当用户需要一份数据报告时你肯定不希望手动把数据从分析 Agent 那里复制出来再粘贴到图表 Agent 里。你希望它们能自己“聊”起来分析 Agent 算完数据后直接告诉图表 Agent“嘿这是结果画个折线图。” 图表 Agent 收到后生成图片再回复“图好了存这儿了。” 这个过程就是 A2A 通信。所以这个 Demo 的核心价值不是展示某个高深的理论而是把一个抽象的多智能体协作概念变成一个你能跑起来、能看见数据流动的、可复现的工程实践。它适合两类人一是刚接触 Agent 开发想理解“协作”到底怎么落地的开发者二是已经写过单个 Agent现在想把多个 Agent 串联起来完成复杂任务的人。最关键的一点是A2A 通信的实现底层上和你熟悉的微服务间调用、消息队列通信没有本质区别都是程序间的数据交换。但它的特殊性在于通信的“内容”往往是自然语言指令或结构化的任务描述而通信的“触发”和“路由”逻辑则需要由 Agent 自身的“大脑”通常是 LLM来决定。这带来了新的挑战如何定义清晰的通信协议如何确保消息不丢失、不被误解如何管理对话状态下面我们就抛开复杂的框架用一个最小化的 Demo把 A2A 通信的骨架搭出来让你能直观地看到两个 Agent 是如何“对话”并完成一个简单任务的。2. 动手前的准备环境与核心思路在开始写代码之前我们需要明确两件事运行环境和本次 Demo 的设计思路。这能帮你避开一开始就陷入复杂框架选择的困境。环境准备这个 Demo 为了足够轻量和聚焦我们选择 Python 作为实现语言。你需要准备Python 环境建议使用 Python 3.8 或以上版本。确保你的终端或 IDE 可以正常运行 Python 脚本。必要的库我们主要会用到requests来模拟 HTTP 通信这是 Agent 间通信最直观的方式之一以及json来处理数据。通常这些是内置或极易安装的。# 如果需要可以安装 requests pip install requests一个可用的 LLM APIAgent 的“大脑”需要大语言模型。为了 Demo 的通用性我们使用 OpenAI 兼容的 API例如 OpenAI 官方 API、或国内一些提供兼容接口的服务。你需要准备一个有效的 API Key 和 Base URL。重要将你的 API 密钥和地址保存在环境变量或一个安全的配置文件中不要硬编码在代码里。这里我们假设你已将其设置为环境变量OPENAI_API_KEY和OPENAI_BASE_URL。Demo 设计思路我们不引入 LangChain、AutoGen 等重型框架而是用最朴素的代码结构来揭示原理。我们将创建两个 Agent任务规划 Agent它的职责是理解用户的原始请求并将其拆解成具体的、可执行的步骤。例如用户说“帮我分析一下销售数据并总结”规划 Agent 会输出“第一步调用数据分析 Agent。第二步将分析结果传给报告生成 Agent。”任务执行 Agent它负责执行具体的子任务。为了简化我们这个 Demo 里只实现一个“算术计算 Agent”它专门处理数学计算请求。通信流程如下用户向“任务规划 Agent”提出请求。“任务规划 Agent”思考后发现请求中包含计算任务于是生成一条消息发送给“算术计算 Agent”。“算术计算 Agent”收到消息执行计算并将结果返回给“任务规划 Agent”。“任务规划 Agent”汇总结果最终回复给用户。这个流程清晰地展示了“请求-路由-执行-回复”的 A2A 通信闭环。我们将用 HTTP 客户端-服务器模式来模拟这个过程一个 Agent 充当临时服务器另一个充当客户端。在生产环境中这可能会被替换为更健壮的消息队列如 RabbitMQ、Kafka或 RPC 框架。3. 从零搭建实现两个能对话的 Agent现在我们开始编写代码。我们会创建三个主要的 Python 文件来组织代码保持结构清晰。3.1 第一步构建 Agent 的通用“大脑”模块首先我们创建一个llm_client.py文件。这个模块封装了与 LLM API 的交互是所有 Agent 的思考核心。# llm_client.py import os import requests import json class LLMClient: def __init__(self): # 从环境变量获取 API 配置 self.api_key os.getenv(“OPENAI_API_KEY”) self.base_url os.getenv(“OPENAI_BASE_URL”, “https://api.openai.com/v1”) # 提供默认值 self.headers { “Content-Type”: “application/json”, “Authorization”: f“Bearer {self.api_key}” } def chat_completion(self, messages, model“gpt-3.5-turbo”): “”“发送聊天请求到 LLM API并返回回复内容。”“” if not self.api_key: raise ValueError(“OPENAI_API_KEY 环境变量未设置”) url f“{self.base_url}/chat/completions” data { “model”: model, “messages”: messages, “temperature”: 0.1, # 降低随机性使输出更稳定 “max_tokens”: 500 } try: response requests.post(url, headersself.headers, jsondata, timeout30) response.raise_for_status() # 检查 HTTP 错误 result response.json() return result[“choices”][0][“message”][“content”].strip() except requests.exceptions.RequestException as e: print(f“LLM API 请求失败: {e}”) if response: print(f“响应内容: {response.text}”) return None except KeyError as e: print(f“解析 LLM 响应失败: {e}原始响应: {result}”) return None # 创建一个全局客户端实例方便导入 llm_client LLMClient()关键点解析封装与复用我们将 LLM 调用封装成一个类这样每个 Agent 都可以导入并使用同一个客户端避免重复配置。错误处理网络请求和 API 调用可能失败。我们使用try-except捕获异常并打印有意义的错误信息这对于调试至关重要。参数设置temperature设为较低值0.1是为了让 Agent 的决策更确定、可复现这在多步协作中很重要。3.2 第二步创建算术计算 Agent执行者接下来创建math_agent.py。这个 Agent 相对简单它监听一个 HTTP 端点等待来自其他 Agent 的计算请求。# math_agent.py from flask import Flask, request, jsonify import json import re from llm_client import llm_client app Flask(__name__) def calculate_expression(expression: str) - str: “”“ 安全地评估一个算术表达式。 注意在生产环境中直接使用 eval() 是极度危险的 这里为了 Demo 简化使用 LLM 来解析和计算更安全。 “”“ # 构建一个让 LLM 进行计算的提示词 prompt f“”” 你是一个算术计算器。请计算以下表达式并只返回最终的数字结果不要任何解释。 表达式{expression} “”” messages [{“role”: “user”, “content”: prompt}] result llm_client.chat_completion(messages) return result if result else “计算失败” app.route(‘/calculate’, methods[‘POST’]) def handle_calculation(): “”“处理来自其他 Agent 的计算请求。”“” data request.get_json() if not data or ‘expression’ not in data: return jsonify({“error”: “请求格式错误需要 ‘expression’ 字段”}), 400 expression data[‘expression’] print(f“[算术计算 Agent] 收到计算请求: {expression}”) # 执行计算 result calculate_expression(expression) print(f“[算术计算 Agent] 计算结果: {result}”) # 返回结果 return jsonify({ “from”: “MathAgent”, “result”: result, “original_expression”: expression }) if __name__ ‘__main__’: # 启动一个简单的 Flask 服务器监听 5001 端口 print(“算术计算 Agent 启动监听 http://localhost:5001”) app.run(host‘0.0.0.0’, port5001, debugFalse) # 生产环境应设置 debugFalse关键点解析HTTP 服务我们使用轻量级的 Flask 框架让这个 Agent 具备接收 HTTP POST 请求的能力。它暴露了一个/calculate端点。通信协议我们定义了一个简单的 JSON 协议请求体必须包含expression字段。响应体包含发送者标识 (from)、结果 (result) 和原始表达式。安全性切记直接使用 Python 的eval()函数执行用户输入的字符串是严重的安全漏洞。这里我们“偷懒”但更安全地让 LLM 来计算避免了代码注入风险。在实际项目中你可能需要实现或引入一个安全的数学表达式解析器。日志输出在控制台打印日志 (print)能让你清晰地看到通信的发生和结果这是调试分布式 Agent 系统的生命线。3.3 第三步创建任务规划 Agent协调者最后创建planner_agent.py。这个 Agent 更复杂它需要理解用户目标决定何时以及如何调用其他 Agent。# planner_agent.py import requests import json from llm_client import llm_client # 定义已知的其他 Agent 的服务地址 AGENT_REGISTRY { “MathAgent”: “http://localhost:5001/calculate”, # 未来可以扩展 “DataAnalysisAgent”: “http://localhost:5002/analyze”, # “ReportAgent”: “http://localhost:5003/generate” } class PlannerAgent: def __init__(self): self.conversation_history [] # 可选用于维护更复杂的对话状态 def plan_and_execute(self, user_query: str) - str: “”“ 核心方法规划并执行任务。 1. 分析用户查询判断是否需要调用其他 Agent。 2. 如果需要生成调用指令并发送请求。 3. 整合结果返回给用户。 “”“ print(f“[任务规划 Agent] 收到用户请求: {user_query}”) # 步骤1让 LLM 分析是否需要调用其他 Agent以及调用谁 planning_prompt f“”” 你是一个任务规划助手。用户说“{user_query}” 请判断完成这个请求是否需要调用专门的工具或助手例如计算器、数据分析员等。 如果需要请严格按以下 JSON 格式回复且只回复这个 JSON {{ “need_agent”: true, “agent_name”: “Agent的名称”, // 例如 “MathAgent” “task_description”: “给该 Agent 的清晰指令例如要计算的表达式”, “reason”: “简要说明为什么需要调用它” }} 如果不需要请回复 {{ “need_agent”: false, “response”: “你直接给用户的回答” }} “”” planning_messages [{“role”: “user”, “content”: planning_prompt}] planning_result llm_client.chat_completion(planning_messages) # 解析 LLM 的规划结果 try: plan json.loads(planning_result) except json.JSONDecodeError: print(f“[任务规划 Agent] 解析规划结果失败: {planning_result}”) return “抱歉我在规划任务时出现了问题。” if not plan.get(‘need_agent’, False): # 不需要调用其他 Agent直接回复 return plan.get(‘response’, ‘我无法处理这个请求。’) # 步骤2需要调用其他 Agent agent_name plan.get(‘agent_name’) task_desc plan.get(‘task_description’) if agent_name not in AGENT_REGISTRY: return f“抱歉我无法找到名为 ‘{agent_name}’ 的助手。” agent_url AGENT_REGISTRY[agent_name] print(f“[任务规划 Agent] 决定调用 {agent_name}, 指令: {task_desc}”) # 步骤3构建请求并发送给目标 Agent # 这里需要根据不同的 Agent 调整请求格式。我们假设 MathAgent 需要 expression request_payload {“expression”: task_desc} # 这是一个简化映射实际中可能需要更复杂的逻辑 try: response requests.post(agent_url, jsonrequest_payload, timeout10) response.raise_for_status() agent_response response.json() print(f“[任务规划 Agent] 收到 {agent_name} 的回复: {agent_response}”) # 步骤4整合结果生成最终回复给用户 final_prompt f“”” 你刚刚协调了一个任务。 用户的原始问题是“{user_query}” 你调用了 {agent_name} 来处理给它指令是“{task_desc}”。 {agent_name} 返回的结果是{agent_response[‘result’]}。 请根据这个结果生成一个对用户友好、完整的最终答复。 “”” final_messages [{“role”: “user”, “content”: final_prompt}] final_response llm_client.chat_completion(final_messages) return final_response if final_response else “任务执行完成但生成最终回复时出错。” except requests.exceptions.RequestException as e: print(f“[任务规划 Agent] 调用 {agent_name} 失败: {e}”) return f“抱歉在与 ‘{agent_name}’ 通信时出现了问题。” def main(): agent PlannerAgent() print(“任务规划 Agent 就绪。请输入您的问题输入 ‘quit’ 退出:”) while True: user_input input(“ “).strip() if user_input.lower() ‘quit’: break if user_input: answer agent.plan_and_execute(user_input) print(f“\n[最终答复] {answer}\n”) if __name__ ‘__main__’: main()关键点解析Agent 注册表AGENT_REGISTRY是一个简单的字典维护了已知 Agent 的名称和访问地址。这是实现服务发现和路由的雏形。规划阶段第一个 LLM 调用 (planning_prompt) 是关键。它让 Agent 自己判断是否需要协作、找谁协作。这体现了 Agent 的自主性。协议适配在request_payload {“expression”: task_desc}这一行我们做了一个强假设所有计算请求都映射到expression字段。在实际系统中你需要一个更智能的“适配层”可能根据agent_name来构造不同的请求体。结果整合收到执行 Agent 的回复后规划 Agent 并没有直接转发结果而是进行了第二次 LLM 调用 (final_prompt)将原始问题、调用过程和原始结果整合成一个通顺的自然语言回复。这提升了用户体验。错误处理对网络请求 (requests.post) 进行了异常捕获防止因为一个 Agent 挂掉导致整个系统无响应。4. 运行与验证看两个 Agent 如何协作现在让我们把整个系统跑起来观察通信过程。4.1 启动系统你需要打开两个终端窗口。终端 1启动算术计算 Agentpython math_agent.py你应该看到输出算术计算 Agent 启动监听 http://localhost:5001终端 2启动任务规划 Agentpython planner_agent.py你应该看到输出任务规划 Agent 就绪。请输入您的问题输入 ‘quit’ 退出:4.2 测试通信在终端 2 的任务规划 Agent 提示符 () 后输入用户请求。让我们设计几个测试用例测试用例 1简单的计算 请帮我计算一下 125 乘以 88 等于多少观察控制台输出理想情况终端 2规划 Agent[任务规划 Agent] 收到用户请求: 请帮我计算一下 125 乘以 88 等于多少终端 2规划 Agent[任务规划 Agent] 决定调用 MathAgent, 指令: 125 * 88终端 1计算 Agent[算术计算 Agent] 收到计算请求: 125 * 88终端 1计算 Agent[算术计算 Agent] 计算结果: 11000LLM 计算得出终端 2规划 Agent[任务规划 Agent] 收到 MathAgent 的回复: {‘from’: ‘MathAgent’, ‘result’: ‘11000’, …}终端 2规划 Agent 最终输出[最终答复] 125 乘以 88 的计算结果是 11000。测试用例 2需要推理的复杂请求 我的项目预算还剩 5000 元团队有 4 个人如果每人每天餐补 50 元还能支撑多少天观察控制台输出规划 Agent 应该能解析出核心计算是5000 / (4 * 50)。它会生成类似5000 / (4 * 50)或5000 / 200的指令发送给 MathAgent。MathAgent 计算得到25。规划 Agent 整合后回复“根据计算您的预算还能支撑 25 天。”测试用例 3无需调用其他 Agent 的请求 你好今天天气怎么样规划 Agent 的 LLM 在规划阶段应判断need_agent为false并尝试直接生成一个回复例如“我是一个任务助手无法获取实时天气信息建议您查看天气预报应用或网站。”4.3 验证成功的关键点通过这个 Demo你应该能验证以下几个 A2A 通信的核心环节请求路由规划 Agent 能正确识别出需要计算并找到对应的 MathAgent。协议通信数据以 JSON 格式通过 HTTP 在 Agent 间传递。任务执行MathAgent 能接收请求、执行计算、返回结构化结果。结果整合规划 Agent 能将原始问题、执行过程和原始结果整合成用户友好的最终答复。松耦合两个 Agent 独立运行只通过定义的 API 接口进行交互。你可以修改或替换其中一个只要接口不变另一个就不受影响。5. 从 Demo 到生产关键考量与扩展方向这个 Demo 跑通只是理解了 A2A 通信的“骨骼”。要应用到真实项目你必须考虑更多“肌肉”和“神经”。5.1 当前 Demo 的局限性脆弱的协议映射规划 Agent 里写死了{“expression”: task_desc}。如果新增一个“翻译 Agent”其请求格式是{“text”: “xxx”, “target_lang”: “en”}当前代码就无法处理。需要一个更通用的“动作-参数”映射机制。同步阻塞调用requests.post是同步的。如果 MathAgent 计算耗时很长规划 Agent 会一直阻塞等待。在生产中你可能需要异步通信如消息队列或至少使用异步 HTTP 客户端。无状态管理PlannerAgent中的conversation_history没有被有效利用。复杂的多轮对话需要维护完整的对话历史以便 Agent 理解上下文。错误处理与重试网络调用失败后只是简单返回错误信息。应有重试机制、熔断策略和更优雅的降级处理。缺乏服务发现Agent 地址硬编码在AGENT_REGISTRY中。当 Agent 动态扩缩容或更换端口时需要像 Consul、Etcd 或简单的服务注册中心来管理。安全性完全没有认证和授权。任何知道端口的人都可以向你的 Agent 发送请求。5.2 扩展方向与实战建议基于以上局限你可以从以下几个方向深化1. 定义清晰的通信协议与动作空间不要用自然语言字符串作为指令直接传递。应该为每个 Agent 定义一套结构化的“动作”Actions。// 规划 Agent 发出的指令 { “action”: “calculate_math_expression”, “parameters”: { “expression”: “125 * 88” } }// 规划 Agent 发出的另一个指令 { “action”: “translate_text”, “parameters”: { “text”: “Hello, world!”, “source_lang”: “en”, “target_lang”: “zh” } }每个 Agent 都注册自己支持的动作列表。规划 Agent 的 LLM 负责生成符合这个结构的指令。2. 引入消息队列进行异步解耦将同步 HTTP 调用改为向消息队列如 Redis Streams, RabbitMQ发布消息。执行 Agent 作为消费者从队列拉取任务处理完后再将结果发布到另一个队列或通过回调通知规划 Agent。这带来了更好的可靠性、削峰填谷和能力扩展。3. 实现基础的 Agent 框架与管理器你可以抽象出一个BaseAgent类包含注册、心跳、任务拉取、结果上报等方法。然后实现一个AgentManager负责接收规划 Agent 的任务。查找有能力支持对应 Action且空闲的 Agent。派发任务并监控超时。收集结果并返回。4. 为通信增加安全层认证在 HTTP 请求头中添加 API Key 或 JWT Token 进行验证。授权检查调用方是否有权执行某个动作。加密对敏感数据考虑使用 HTTPS 和 payload 加密。5. 完善可观测性除了print语句需要集成日志系统如 structlog, loguru记录每个任务的唯一 ID、流转路径、耗时、状态。接入监控如 Prometheus metrics跟踪 Agent 的健康状态、队列长度、错误率。踩坑经验不要过度设计起步像这个 Demo 一样先用最简单的 HTTP 同步调用把核心流程跑通。过早引入消息队列和复杂框架会增加理解成本。日志是你的眼睛在开发阶段在每个关键步骤收到请求、调用 LLM、发送请求、收到回复都打印详细的、结构化的日志。这是定位问题最快的方法。从单轮对话到多轮对话维护对话历史看似简单但要注意上下文长度限制。需要设计摘要或滑动窗口机制防止历史过长导致 LLM 性能下降或遗忘关键信息。测试边界情况专门测试网络超时、目标 Agent 无响应、LLM 返回非预期格式、用户输入恶意指令等情况确保你的系统有基本的鲁棒性。这个 Demo 就像一张地图标出了 A2A 通信的主要地标规划、路由、调用、整合。拿着这张地图你就能根据自己的业务地形去修筑更坚固的道路协议、架设更高效的桥梁异步通信、并设立路标和哨站可观测性与安全最终构建出真正能投入生产的智能体协作系统。