1. 先搞清楚 A2A 协议到底解决了什么问题如果你正在接触 AI Agent 开发或者想把手头的 AI 能力从一个地方“搬”到另一个地方那么 A2A 协议是你绕不开的一个概念。它不是什么高深莫测的学术理论而是一个解决实际工程问题的思路如何让不同的 AI 应用或服务能像人一样稳定、可靠地互相调用和协作。很多人一听到“协议”就觉得复杂其实它的核心目标很简单标准化 AI 之间的对话。想象一下你有一个擅长写代码的 AI 和一个擅长画图的 AI你想让它们合作完成一个“生成一个登录页面”的任务。如果没有一个约定好的沟通方式它们可能互相听不懂对方在说什么或者任务传着传着就丢了。A2A 协议要解决的就是这个“沟通”和“协作”的标准化问题。所以这篇文章不是讲协议的理论细节而是从实战出发带你走一遍在一个具体的开发场景里怎么理解 A2A 的思想并把它落地成一个能跑通的、可验证的协作流程。我们重点关注的是从单点能力到协同工作的关键步骤以及在这个过程中最容易踩的坑。2. 动手前的准备环境、工具与核心思想在开始写代码之前先把环境和思路理清楚。A2A 的实现方式有很多可以用现成的框架也可以基于消息队列或 RPC 自己设计。为了聚焦于协议本身的理解我们这里选择一个轻量级、易于理解的实现路径使用 FastAPI 构建 Web 服务作为 Agent并通过 HTTP 请求模拟 A2A 的交互。这种方式虽然简单但足以清晰地展示任务分解、请求路由、结果聚合等核心环节。你需要准备的环境Python 3.8这是大多数 AI 相关库的基础。FastAPI 与 Uvicorn用于快速构建 API 服务。pip install fastapi uvicornRequests用于服务间调用。pip install requests一个代码编辑器或 IDE如 VSCode、PyCharm。OpenAI API Key 或本地 LLM 服务作为 AI 的“大脑”。我们将使用 OpenAI 的 API 作为示例但思路完全适用于 Claude、文心一言等任何提供 API 的模型。核心思想拆解A2A 实战不是让两个 AI 模型直接“对话”而是让两个封装了 AI 能力的服务进行对话。每个服务Agent都有明确的身份Role它是干什么的例如代码生成器、文案写手、数据分析师。能力Capability它能接收什么输入产出什么输出例如输入“功能描述”输出“Python 代码”。接口Endpoint别人怎么调用它一个固定的 HTTP API 地址和参数格式。我们的目标就是创建两个这样的 Agent 服务并让它们根据一个总任务进行协作。3. 实战第一步构建两个独立的 Agent 服务我们先构建两个最简单的 Agent一个代码生成器CoderAgent一个代码审查员ReviewerAgent。它们的协作流程是用户提出一个编程需求 - CoderAgent 生成代码 - ReviewerAgent 审查代码并给出反馈。3.1 创建 CoderAgent接收需求生成代码这个 Agent 提供一个 API接收用户的需求描述调用大模型生成对应的代码片段。首先创建项目目录a2a_demo并在其中创建coder_agent.py# coder_agent.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import openai import os # 1. 初始化 FastAPI 应用和 OpenAI 客户端 app FastAPI(titleCoderAgent) openai.api_key os.getenv(OPENAI_API_KEY) # 从环境变量读取 Key # 2. 定义输入输出的数据模型这就是“协议”的一部分 class CodeRequest(BaseModel): requirement: str # 用户需求描述 language: str python # 编程语言默认 Python class CodeResponse(BaseModel): code: str agent: str CoderAgent # 3. 核心端点/generate_code app.post(/generate_code, response_modelCodeResponse) async def generate_code(request: CodeRequest): Agent 能力根据需求生成代码。 输入需求描述、编程语言。 输出生成的代码片段。 if not openai.api_key: raise HTTPException(status_code500, detailOpenAI API key not configured) # 构建给大模型的提示词Prompt prompt f 你是一个资深的{request.language}程序员。请根据以下需求生成简洁、高效、可运行的代码。 只输出代码不要包含任何解释性文字。 需求{request.requirement} try: # 调用 OpenAI API response openai.ChatCompletion.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.7, max_tokens500 ) generated_code response.choices[0].message.content.strip() # 返回标准化响应 return CodeResponse(codegenerated_code) except Exception as e: raise HTTPException(status_code500, detailfFailed to generate code: {str(e)}) # 4. 启动服务独立运行 if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8001) # CoderAgent 运行在 8001 端口关键点解析明确的接口POST /generate_code接收 JSON 格式的CodeRequest返回 JSON 格式的CodeResponse。这个请求和响应的结构就是 CoderAgent 对外公布的“协议”。能力边界这个 Agent 只做一件事——生成代码。它不关心代码好坏也不负责运行。错误处理对 API Key 缺失和 OpenAI 调用异常做了基本处理避免服务崩溃。在终端 A 启动它export OPENAI_API_KEY你的OpenAI API Key python coder_agent.py服务启动后你可以用curl或 Postman 测试POST http://localhost:8001/generate_code Body:{requirement: 写一个函数计算斐波那契数列的第n项, language: python}。3.2 创建 ReviewerAgent审查代码提供反馈接下来创建第二个 Agent它接收一段代码并给出审查意见。在项目目录下创建reviewer_agent.py# reviewer_agent.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import openai import os app FastAPI(titleReviewerAgent) openai.api_key os.getenv(OPENAI_API_KEY) # 定义协议ReviewerAgent 的输入输出格式 class ReviewRequest(BaseModel): code: str # 需要审查的代码 language: str python class ReviewResponse(BaseModel): review: str # 审查意见 agent: str ReviewerAgent app.post(/review_code, response_modelReviewResponse) async def review_code(request: ReviewRequest): Agent 能力审查代码指出潜在问题并提供改进建议。 输入代码片段、编程语言。 输出审查意见。 prompt f 你是一个严格的代码审查员。请审查以下{request.language}代码从以下角度给出意见 1. 代码风格和可读性。 2. 潜在的逻辑错误或边界条件处理。 3. 性能优化建议。 4. 安全性问题如果存在。 请用清晰、有条理的列表形式输出。 代码 {request.language} {request.code} try: response openai.ChatCompletion.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.3, # 温度调低让输出更稳定、聚焦 max_tokens600 ) review_text response.choices[0].message.content.strip() return ReviewResponse(reviewreview_text) except Exception as e: raise HTTPException(status_code500, detailfFailed to review code: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8002) # ReviewerAgent 运行在 8002 端口在终端 B 启动它export OPENAI_API_KEY你的OpenAI API Key python reviewer_agent.py现在我们有了两个完全独立、自治的 Agent 服务。它们各自监听不同的端口拥有明确的职责和 API 接口。这就是 A2A 的基石每个 Agent 都是一个可独立部署和调用的微服务。4. 实现 A2A 协作构建一个协调者Orchestrator两个 Agent 已经就位但它们还不知道对方的存在也不会自动协作。我们需要一个协调者Orchestrator。它的职责是接收用户的原始任务理解任务需要哪些 Agent 参与按顺序调用它们并管理整个流程和数据流。在项目目录下创建orchestrator.py# orchestrator.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel import requests import asyncio app FastAPI(titleA2A-Orchestrator) # 定义 Orchestrator 对外的接口协议 class UserTask(BaseModel): description: str # 用户原始任务描述 class TaskResponse(BaseModel): final_code: str code_review: str status: str # 定义我们已知的 Agent 网络地址服务发现的基础 AGENT_REGISTRY { coder: http://localhost:8001/generate_code, reviewer: http://localhost:8002/review_code, } app.post(/execute_task, response_modelTaskResponse) async def execute_task(task: UserTask): 协调者核心逻辑 1. 解析任务决定工作流本例是固定流程先写代码再审查。 2. 调用 CoderAgent。 3. 将 CoderAgent 的结果传给 ReviewerAgent。 4. 汇总结果返回给用户。 print(f[Orchestrator] 开始处理任务: {task.description}) # 步骤 1: 调用 CoderAgent coder_payload { requirement: task.description, language: python } try: coder_response requests.post(AGENT_REGISTRY[coder], jsoncoder_payload, timeout30) coder_response.raise_for_status() # 检查 HTTP 错误 generated_code coder_response.json()[code] print(f[Orchestrator] CoderAgent 返回代码长度: {len(generated_code)}) except requests.exceptions.RequestException as e: raise HTTPException(status_code502, detailfCoderAgent 调用失败: {str(e)}) # 步骤 2: 调用 ReviewerAgent reviewer_payload { code: generated_code, language: python } try: reviewer_response requests.post(AGENT_REGISTRY[reviewer], jsonreviewer_payload, timeout30) reviewer_response.raise_for_status() review_comments reviewer_response.json()[review] print(f[Orchestrator] ReviewerAgent 返回审查意见) except requests.exceptions.RequestException as e: # 即使审查失败也返回生成的代码但标记审查失败 review_comments f代码审查服务暂时不可用: {str(e)} print(f[Orchestrator] ReviewerAgent 调用失败使用降级方案) # 步骤 3: 汇总并返回最终结果 return TaskResponse( final_codegenerated_code, code_reviewreview_comments, statuscompleted ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000) # Orchestrator 运行在 8000 端口在终端 C 启动协调者python orchestrator.py现在A2A 协作链路已经打通用户向Orchestrator (8000端口)发送一个任务。Orchestrator将任务需求发送给CoderAgent (8001端口)。CoderAgent生成代码返回给Orchestrator。Orchestrator将代码发送给ReviewerAgent (8002端口)。ReviewerAgent返回审查意见。Orchestrator将代码和审查意见打包返回给用户。你可以用以下命令测试整个流程curl -X POST http://localhost:8000/execute_task \ -H Content-Type: application/json \ -d {description: 写一个Python函数它接收一个URL返回该网页的标题}如果一切正常你会得到一个包含生成代码和审查意见的 JSON 响应。这就是一个最基础的、可运行的 A2A 协议实战案例。5. 从“能跑通”到“能实用”的关键优化上面的例子演示了核心流程但离生产环境还差得远。以下是几个必须考虑的优化点这也是 A2A 协议实战中真正的价值所在。5.1 服务发现与注册让 Agent 动态上线我们硬编码了 Agent 的地址localhost:8001这非常脆弱。在实际系统中Agent 可能动态启动、停止或更换地址。需要一个服务注册与发现机制。简单方案使用一个共享的数据库或配置文件Agent 启动时向其中注册自己的地址和能力。成熟方案集成 Consul、Etcd 或 Nacos或者使用 Kubernetes Service。Orchestrator 修改不再硬编码AGENT_REGISTRY而是从一个中心化的注册表查询可用的CoderAgent和ReviewerAgent。5.2 异步、超时与重试提升鲁棒性我们的协调者使用requests.post(...)是同步阻塞调用。如果某个 Agent 响应慢会拖慢整个流程。异步调用使用httpx或aiohttp库进行异步 HTTP 调用让多个 Agent 可以并行工作如果任务允许。设置超时每个调用都必须设置超时时间如timeout30防止因某个 Agent 挂起导致整个请求被卡死。重试机制对于网络抖动等临时性失败应该设计重试逻辑例如最多重试3次每次间隔递增。熔断与降级如果某个 Agent 连续失败可以暂时将其标记为不可用熔断并执行降级策略例如跳过审查步骤直接返回代码。5.3 任务编排与工作流引擎我们的协调者逻辑是硬编码的先A后B。复杂的任务可能需要动态的任务流。工作流定义可以使用 YAML 或 DSL 来定义任务流程例如“coder - reviewer - tester”。引入工作流引擎对于复杂场景可以考虑集成像Prefect、Airflow或Kubernetes Jobs来管理任务依赖和状态。状态持久化长任务需要将中间状态如生成的代码、审查意见保存到数据库支持断点续跑。5.4 统一的通信协议与数据格式我们定义了CodeRequest/CodeResponse但这只是两个 Agent 之间的约定。一个大的 A2A 系统需要更统一的协议。标准化信封Envelope每个 Agent 间的消息可以包裹在一个标准信封里包含task_id,source_agent,destination_agent,payload实际数据,timestamp等。使用消息队列用 RabbitMQ、Kafka 或 Redis Stream 代替直接的 HTTP 调用。Agent 订阅主题发布消息实现解耦和异步通信。这才是更接近“Agent 间自主对话”的形态。采用现有标准关注像OpenAI 的 Function Calling、LangChain 的 Agent 协议或AutoGen 的群聊模式它们都在尝试定义 Agent 间的交互标准。5.5 监控、日志与可观测性当多个 Agent 协作时问题排查变得困难。分布式追踪为每个用户请求生成一个唯一的trace_id并在这个请求流经 Orchestrator、CoderAgent、ReviewerAgent 时都将这个trace_id记录在日志中。这样可以在日志系统中轻松串联起一次完整调用的所有步骤。结构化日志不要只打印print使用structlog或json-logger输出结构化的 JSON 日志便于后续收集和分析例如发送到 ELK 或 Loki。关键指标监控每个 Agent 的调用延迟、成功率、以及 Orchestrator 的任务吞吐量。6. 常见问题与排查清单当你按照上面的步骤实践时可能会遇到以下问题。按照这个顺序排查能节省大量时间。问题1服务启动失败端口被占用。排查netstat -an | grep 8000(Linux/macOS) 或Get-NetTCPConnection -LocalPort 8000(Windows PowerShell) 查看端口占用。解决杀掉占用进程或修改代码中的端口号确保 Orchestrator 和 Agent 的端口不冲突。问题2调用 Orchestrator 返回502 Bad Gateway或连接错误。排查1确认所有服务Orchestrator, CoderAgent, ReviewerAgent都已成功启动。检查各自的终端是否有报错日志。排查2检查orchestrator.py中的AGENT_REGISTRY地址是否正确。如果 Agent 运行在容器或不同机器上localhost需要替换为实际 IP 或服务名。排查3检查环境变量OPENAI_API_KEY是否在所有终端都已正确设置。问题3Agent 服务返回了代码但内容为空或不符合预期。排查1直接单独测试 Agent 的 API。用curl调用http://localhost:8001/generate_code看是否正常返回。这能隔离出是 Agent 的问题还是 Orchestrator 调用的问题。排查2查看 Agent 服务的日志。大模型 API 调用可能因为额度不足、网络问题或提示词Prompt不佳而失败。排查3优化提示词Prompt。我们的示例 Prompt 比较简单对于复杂需求可能需要更详细的指令和示例Few-shot。问题4任务执行速度很慢。排查1确认是哪个环节慢。在 Orchestrator 的代码中加入时间戳日志记录调用每个 Agent 的开始和结束时间。排查2大模型 API 调用通常是瓶颈。考虑是否可以使用更快的模型如gpt-3.5-turbo比gpt-4快或调整max_tokens限制输出长度。排查3如果多个步骤没有依赖关系将其改为异步并行调用。问题5如何扩展到更多 Agent方法在AGENT_REGISTRY中添加新的 Agent 地址和能力描述。修改 Orchestrator 的工作流逻辑决定在什么条件下调用哪个 Agent。这其实就是路由Routing逻辑可以根据任务内容、负载均衡策略或 Agent 的专业领域来动态选择。7. 总结A2A 协议实战的核心收获通过这个从零搭建的实战案例你应该能清晰地感受到A2A 协议开发的核心不是追求理论的完备性而是解决一系列工程问题定义清晰的接口每个 Agent 对外提供什么服务Endpoint输入输出是什么Schema这就是最基础的协议。实现服务自治每个 Agent 应该可以独立开发、测试、部署和扩展。设计协调逻辑需要一个大脑Orchestrator或一套规则工作流引擎来指挥 Agent 们协作。这个协调者本身也可以是一个 Agent。处理分布式问题网络调用必然伴随超时、重试、熔断、降级必须为这些情况设计预案。保障可观测性没有完善的日志、监控和追踪多 Agent 系统出了问题就是噩梦。不要被“协议”这个词吓到。你可以从最简单的 HTTP API 调用开始就像我们刚才做的那样。先让两个服务能对话再逐步引入消息队列、服务发现、工作流引擎。每一步的升级都是为了解决当前架构遇到的具体痛点如耦合太紧、扩展不便、排查困难。最终一个健壮的 A2A 系统会让你的 AI 能力像乐高积木一样可以灵活地组合、复用去应对更复杂的任务。而这一切的起点就是先动手让第一个 Agent 服务跑起来并成功调用第二个。