本文是三部曲系列的第一篇 关于构建模块化AI系统。
在这一部分中，我们将介绍 Google 的 A2A（Agent-to-Agent）协议，详细讲解其架构及通信流程，并通过代码示例展示 AI 代理如何在多代理生态系统中协作。
在第二篇中，我们将在此基础上探讨 A2A 如何与 MCP（模型上下文协议）集成，以构建不仅具备协作性，还具备工具和数据意识的 AI 系统——能够执行动态且真实的任务。
第三篇将揭示智能路由在优化多代理系统，实现性能提升和成本节约中的关键作用。
三周前，我盯着屏幕发呆，想着是怎么把事情搞乱七八糟的。
我有五个不同的AI服务需要相互交流——一个处理查询的OpenAI代理，一个检索文档的向量搜索系统，一个运行计算的代理，一个由Claude支持的摘要生成器，以及一个具有特定领域逻辑的自定义Python代理。
每个都有自己独特的API格式。每个对参数的期望不同。每个以自己独特的方式返回结果。我的代码看起来像一个混乱的翻译层——一个由API适配器、消息解析器和错误处理器组成的怪兽。
当我需要更换一个服务时？简直就是一场噩梦。每当 OpenAI 发布了一个新模型时，我不得不重写我的通信层的很大一部分。当我想要尝试 Anthropic 的新 Claude 模型时，情况也差不多。
我花费了80%的开发时间在管道工作上，而不是实际的人工智能逻辑。
这听起来熟悉吗？如果你最近尝试构建多代理AI系统，你很可能已经体会到了这种痛苦。而随着AI生态系统的迅速扩展和越来越多的专业化服务，这种状况只会越来越恶化。
为甚么代理间的通信成为了新的瓶颈人工智能领域正在分化成专门的服务——这确实是有道理的。不同的模型在各自的任务上表现出色。在特定应用场景中，微调后的专用模型通常能胜过通用模型。‘一款模型通吃所有’的做法正在被专门构建的人工智能服务取代。
这就带来了一个问题：这些专门的AI服务如何才能相互沟通？
没有统一标准，每个代理之间都需要定制代码。每个新加入的代理都会带来更多复杂性。这根本不可持续，拖慢了多代理AI系统的创新步伐。
我们需要的是AI领域的标准协议，例如HTTP，它能让任何代理与其它代理进行交流，而无需定制的翻译层。
看看Google的A2A协议这样，这正是谷歌的Agent-to-Agent (A2A) 协议要解决的问题。它提供了一种标准化的方式，让AI代理能够互相交流，不管其底层实现如何。
当我第一次发现A2A时，我有了那种“恍然大悟”的时刻。这里是我解决管道问题困扰的方案——一种通用语言，让我的代理人能够互相交流，而不需要我为每个代理人单独编写适配器。
该协议定义了以下内容：
- 文本消息、函数调用和响应的消息格式
- 对话结构和线程管理
- 代理能力的元数据
- 错误处理策略
但这里有个问题——虽然 Google 发布了该规范，但并没有一个可以直接用于生产的 Python 实现，这使得采用起来有些麻烦。
那就是我为什么要做 Python A2A：（一个项目）: https://github.com/themanojdesai/python-a2a。
Python A2A：节省时间的真实案例在我们详细研究Python A2A之前，让我通过一个实际的例子向你们展示代理之间标准化通信带来的变化。
如下是我连接OpenAI的代理到自定义工具时未标准化的代码：
的代码的样子如下：修改为
如下是我连接OpenAI的代理到自定义工具时未标准化的代码：
如下是我连接OpenAI的代理到自定义工具时未进行标准化的代码：
# 不标准化：OpenAI -> 自定义工具代理
def 获取天气数据(location):
# 调用自定义天气服务
响应 = requests.post(
"http://localhost:5001/weather",
json={"query": f"位置 {location} 的天气怎么样？"}
)
if 响应.status_code != 200:
return f"获取天气数据时出错：{响应.text}"
# 解析自定义 JSON 格式
天气数据 = 响应.json()
# 转换为 OpenAI 预期的格式
if "temperature" in 天气数据 and "conditions" in 天气数据:
return f"{location}: {天气数据['temperature']}°C, {天气数据['conditions']}"
else:
return f"天气数据格式错误：{天气数据}"
def 使用天气数据查询_openai(user_query, location):
# 首先获取 OpenAI 需要的天气数据
天气信息 = 获取天气数据(location)
# 现在使用重新格式化的天气数据调用 OpenAI
openai_response = openai.ChatCompletion.create(
model="gpt-4",
messages=[
{"role": "system", "content": "你是一个乐于助人的助手。"},
{"role": "user", "content": user_query},
{"role": "assistant", "content": f"让我查一下天气。{天气信息}"}
]
)
return openai_response.choices[0].message.content
# 我们还需要为每个代理组合编写各自的代码！现在我们用Python A2A来试试同样的：
从 python_a2a 导入 A2AClient, Message, TextContent, MessageRole
# 使用 A2A：任意代理 -> 任意其他代理
def plan_trip(location):
# 连接到专用代理 - 所有都使用相同的协议
weather_client = A2AClient("http://localhost:5001/a2a")
openai_client = A2AClient("http://localhost:5002/a2a")
# 询问天气代理
weather_message = Message(
content=TextContent(text=f"{location} 的天气预报是什么？"),
role=MessageRole.USER
)
weather_response = weather_client.send_message(weather_message)
# 向 OpenAI 代理询问，包含天气信息
planning_message = Message(
content=TextContent(
text=f"我计划去 {location} 旅行。天气预报：{weather_response.content.text}"
f"请提出活动建议。"
),
role=MessageRole.USER
)
planning_response = openai_client.send_message(planning_message)
return planning_response.content.text
# 这和任何 A2A 兼容代理都能一起用，不需要额外的适配器哦！ 看出来了差别吗？使用A2A，我不需要为每种代理组合定制适配器。它们都用同一种语言交流。我可以替换、添加代理，或者重新配置整个系统，而不需要改动通信层。
这使我在最近的项目上节省了一半的开发时间。
Python A2A：让A2A协议更易访问Python A2A 是一个全面实现 Google 的 A2A 协议，旨在：
- 直观 — 简单的 API 让 Python 开发者感觉很自然。
- 生产就绪 — 专为真实世界的应用程序打造，具备强大的错误处理能力。
- 框架无关 — 可与 Flask、FastAPI、Django 或任何其他框架无缝集成。
- 依赖最少 — 核心功能仅需
requests库。
开始很容易，只需……
在终端中运行以下命令来安装python-a2a包。
pip install python-a2a如果你想用它和特定的LLM提供商。
# 集成OpenAI
pip install "python-a2a[openai]"
# 集成Anthropic Claude
pip install "python-a2a[anthropic]"
# 安装所有可选的依赖项
pip install "python-a2a[all]"我们从一个简单的例子开始：一个回应消息的代理。
from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server
class EchoAgent(A2AServer):
"""一个简单的代理，它会把消息加上个前缀再回应回来。"""
def handle_message(self, message):
if message.content.type == "text":
return Message(
content=TextContent(text=f"回应: {message.content.text}"),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
# 启动服务
if __name__ == "__main__":
agent = EchoAgent()
run_server(agent, host="0.0.0.0", port=5000)将其保存为 echo_agent.py，运行后，你就可以在 [http://localhost:5000/a2a](http://localhost:5000/a2a) 上运行一个兼容 A2A 的代理了。
我们现在来跟它聊聊
from python_a2a import A2AClient, Message, TextContent, MessageRole
# 创建一个客户端与我们的代理进行通信
client = A2AClient("http://localhost:5000/a2a") # A2A客户端
# 发送一条消息
message = Message(
content=TextContent(text="你好，有人在吗？") # 文本内容
role=MessageRole.USER # 消息角色
)
response = client.send_message(message)
# 打印响应
print(f"代理: {response.content.text}")简单吧？但真正的力量在于构建更复杂的智能代理时展现出来。
代理间的函数调用：我特别喜欢A2A的一个功能就是标准化的代理间函数调用。这对提供特定功能的专门代理来说非常重要。
这里有一个提供了一些数学函数的计算器程序。
import math
from python_a2a import (
A2AServer, Message, TextContent, FunctionCallContent,
FunctionResponseContent, FunctionParameter, MessageRole, run_server
)
class CalculatorAgent(A2AServer):
"""我是一个能帮你做数学计算的小助手。"""
def handle_message(self, message):
if message.content.type == "text":
return Message(
content=TextContent(
text="我是一个计算器代理。你可以调用我的函数：\n"
"- calculate: 基本的加减乘除运算（操作，a，b）\n"
"- sqrt: 求平方根（值）"
),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
elif message.content.type == "function_call":
function_name = message.content.name
params = {p.name: p.value for p in message.content.parameters}
try:
if function_name == "calculate":
operation = params.get("operation", "add")
a = float(params.get("a", 0))
b = float(params.get("b", 0))
if operation == "add":
result = a + b
# 其他运算...
return Message(
content=FunctionResponseContent(
name="calculate",
response={"结果": result}
),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
# 其他功能...
except Exception as e:
return Message(
content=FunctionResponseContent(
name=function_name,
response={"出错了": str(e)}
),
role=MessageRole.AGENT,
parent_message_id=message.message_id,
conversation_id=message.conversation_id
)
if __name__ == "__main__":
agent = CalculatorAgent()
run_server(agent, host="0.0.0.0", port=5001)以下是我们如何调用它的功能。
from python_a2a import (
A2AClient, Message, FunctionCallContent,
FunctionParameter, MessageRole
)
client = A2AClient("http://localhost:5001/a2a")
# 创建一个函数调用的消息
function_call = Message(
content=FunctionCallContent(
name="calculate",
parameters=[
FunctionParameter(name="operation", value="add"),
FunctionParameter(name="a", value=5),
FunctionParameter(name="b", value=3)
]
),
role=MessageRole.USER
)
response = client.send_message(function_call)
if response.content.type == "function_response":
result = response.content.response.get("result")
# 获取函数调用的结果
if result is not None:
print(f"结果为: {result}") # 输出结果为: 8最棒的是任何兼容A2A的代理都可以调用这些函数。这意味着你的OpenAI代理程序可以调用你自定义的计算器代理，接着可以调用天气代理等等——所有这些都使用同样的标准协议。
轻松创建基于大型语言模型的代理Python A2A 包含了与流行 LLM 供应商的即用型集成。这里有一个 OpenAI 支持的代理。
import os
from python_a2a import OpenAIA2AServer, run_server
# 创建一个由OpenAI驱动的代理
agent = OpenAIA2AServer(
api_key=os.environ['OPENAI_API_KEY'],
model="gpt-4",
system_prompt="系统提示="你是一个乐于助人的AI助手。""
)
# 启动服务器
if __name__ == "__main__":
run_server(agent, host="0.0.0.0", port=5002)只需三行代码（加上必要的导入），你就可以拥有一个与OpenAI兼容的代理程序。你也可以对Anthropic的Claude做相同的事，或者为其他服务商构建自定义集成。
一个实际案例：研究助手的工作流程比如说，让我给你展示一个更复杂的现实世界示例：一个研究助手，它协调多个代理人来回答研究中的问题。
这受到一个项目的启发，我为一个客户构建的项目，该客户需要分析科学论文的需求。工作流程如下：具体步骤如下。
- 一个根据研究问题生成搜索查询的LLM代理
- 一个查找相关信息的搜索代理
- 一个将信息整合成连贯答案的总结代理
用Python A2A看起来是这样的：
from python_a2a import (
A2AClient, Message, TextContent, MessageRole, Conversation
)
def research_workflow(query):
# 连接到专用代理程序
llm_client = A2AClient("http://localhost:5002/a2a") # LLM代理
search_client = A2AClient("http://localhost:5003/a2a") # 搜索代理
summarize_client = A2AClient("http://localhost:5004/a2a") # 摘要代理
# 跟踪整个工作流程对话
conversation = Conversation()
conversation.create_text_message(
text=f"研究问题：{query}",
role=MessageRole.USER
)
# 步骤1：生成搜索查询词
print("生成搜索查询词...")
search_request = Message(
content=TextContent(
text=f"根据这个问题：'{query}'，生成3个特定的搜索查询词，以帮助找到相关信息。"
),
role=MessageRole.USER
)
search_queries_response = llm_client.send_message(search_request)
conversation.add_message(search_queries_response)
# 步骤2：检索信息
print("检索信息...")
search_message = Message(
content=TextContent(
text=f"搜索以回答该问题：{query}\n\n"
f"使用这些查询词：\n{search_queries_response.content.text}"
),
role=MessageRole.USER
)
search_results = search_client.send_message(search_message)
conversation.add_message(search_results)
# 步骤3：综合信息内容
print("综合信息内容...")
summarize_message = Message(
content=TextContent(
text=f"综合这些信息内容以回答问题：'{query}'\n\n"
f"以下信息：\n{search_results.content.text}"
),
role=MessageRole.USER
)
summary_response = summarize_client.send_message(summarize_message)
conversation.add_message(summary_response)
# 添加最终答案到对话
conversation.create_text_message(
text=f"对你的研究问题的回答：\n\n{summary_response.content.text}",
role=MessageRole.AGENT
)
return conversation
# 示例用法
if __name__ == "__main__":
query = input("你的研究问题是？")
result = research_workflow(query)
print("\n研究完成！")
print("=" * 50)
print(result.messages[-1].content.text)在 A2A 之前，这需要编写数百行自定义适配器代码来处理每个服务的不同 API 格式。使用 Python A2A，代码变得干净、易于维护且可扩展。
而最好的地方是？我可以将这些代理中的任何一个换成不同的版本而无需修改工作流代码。如果有更好的搜索代理出现，我只需指向它的A2A端点就行了。
构建Python A2A的心得体会我在构建Python A2A的过程中学到了很多关于代理通信的知识。以下是一些关键的见解。
1 消息结构真的很重要A2A协议定义了一个清晰的消息结构，其中包括内容类型、角色和消息ID。这种结构化的做法使得对话过程比无结构的JSON更容易追踪和调试。
2. 函数调用这功能真是太厉害了代理间的标准化的函数调用改变了游戏规则。它使得专业代理能够以一种统一的方式展现其能力，使得其他任何代理都能使用。
3. 对话需要分段处理该协议通过对对话分组的支持（通过父消息ID和对话ID来标识），使得在复杂的多方交互中保持上下文变得可能。
4. 错误处理需要统一在A2A之前，每个服务都有自己独特的错误格式。有了A2A，错误处理变得更加一致，调试变得更加简单。
多代理AI的未来我们正处在多代理人工智能革命的开端。随着模型越来越专业化，公司对特定领域定制代理的需求将会增加，标准化沟通的需求也会随之增加。
想象一下这样的未来
- 可以轻松连接来自不同供应商的各种专用AI服务
- 特定领域的代理可以跨项目共享和重用
- 复杂的代理编排变得像使用API一样简单
- AI系统可以像乐高积木一样拼接在一起
这就是A2A和Python A2A正在共同努力打造的未来。
使用Python入门如果你已经准备好简化代理间的通信并开始构建更加模块化的AI系统，这里是如何开始的一些方法：
Python A2A 的优点之一在于你可以从小处开始——只需将一个代理转换为 A2A，然后根据所获得的好处逐步在整个系统中推广它。
一个为什么标准化很重要所以我创建了Python A2A，因为我厌倦了把大部分开发时间都花在处理基础设施这些琐事上，而不是实际的AI逻辑。A2A协议的标准化特性为我节省了无数小时，让我的多代理系统更易于维护和扩展。
如果你正在开发多代理AI系统，你可能也在面对和我一样的挑战。AI生态系统正变得越来越碎片化，如果没有标准化，复杂性将变得无法控制。
Python A2A 为你提供了一个简单、开箱即用的方法来实现 Google 的 A2A 协议，从而开始构建更模块化且互操作性更强的代理系统。
试试看，然后跟我说它怎样改变了你的开发流程。
资源- Python A2A GitHub 代码库
- Python A2A 的 PyPI 页面
- Python A2A 官方文档
-
如果你觉得这篇文章对你有帮助的话，我很乐意通过LinkedIn 或Twitter 与你联系。我对其他人正在构建的多代理系统（multi-agent systems）一直很感兴趣。
共同學習，寫下你的評論
評論加載中...
作者其他優(yōu)質(zhì)文章
