在人工智能迅猛发展的今天，多智能体系统（Multi-Agent Systems）正在成为热门方向。但一个奇怪的现象出现了：随着AI模型越来越聪明，多智能体之间的“沟通”反而成了新的瓶颈。

为什么？因为人工智能世界正在飞速分化成各种专业服务。

过去，我们总想着做一个“万能AI”，啥都会。但实践证明，不同任务需要不同专长的模型。比如，一个擅长写作的语言模型，和一个专门做金融分析的模型，它们的内核、推理方式乃至交互习惯，可能天差地别。

经过特定领域微调的模型，往往能在专业用例中远超通用大模型的表现。因此，“一刀切”的通用AI思路，正在加速让位于专业化、定制化的AI服务。

这就带来了一个麻烦：如果每一个AI代理都是特定训练出来的专家，我们怎么让他们互相沟通协作？

想象一下，如果每个国家说自己的语言，又没有翻译，那国际交流会变成灾难。同样的，在没有统一协议的情况下，AI代理之间每建立一次连接，就需要写一堆定制化的对接代码。加一个代理，复杂度飙升，加十个，系统几乎不可维护。这不仅拉高了开发成本，还严重阻碍了多智能体系统的创新速度。

换句话说，多智能体AI系统，急需一个“AI版的HTTP”——一种标准化协议，让任何代理可以和任何其他代理无障碍交流，不需要每次都重新发明轮子。

这时，Google带着它的“代理到代理（Agent-to-Agent，简称A2A）”协议登场了。

A2A协议是什么？简单讲，它就是为了打破代理之间孤岛效应而诞生的一套通用通信标准。不管你的底层代理是大模型、小模型、微服务，还是硬件控制器，只要遵循A2A协议，就可以像HTTP一样，轻松打通彼此之间的消息流动。

当我第一次了解到A2A协议时，脑海里那句“终于有人解决了这个世纪难题”的感叹，几乎是脱口而出。

想象一下，以前我们要让两个不同厂家的AI代理对话，需要单独写一堆API接口，还要处理数据格式、函数调用规范、错误处理规则等。而A2A直接一刀切，统一了整个流程：

• 定义了文本、函数调用和响应的标准消息格式

• 规定了完整的对话结构与上下文线索管理

• 要求代理公开自己的功能元数据，方便动态发现与调用

• 提出了标准化的错误处理机制，避免每次都踩坑

说白了，A2A就是帮AI之间立下了一套“社交礼仪”和“共同语言”。

不过美中不足的是，谷歌虽然发布了规范，但并没有给出一个开箱即用的、面向生产环境的Python实现。

这就催生了我去动手构建一个开源版本的Python A2A库——既能让代理快速上手互联互通，又不需要自己从零造轮子。

为什么一定要有Python版本？因为Python依然是AI开发的主力军，特别是在多智能体系统中，绝大多数代理和调度器都是用Python编写的。如果想让A2A在开发者中迅速落地，一个轻量好用的Python SDK几乎是刚需。

现在使用 Python A2A 实现同样的效果：

发现区别了吗？有了 A2A，我不需要为每个代理组合定制适配器。它们都使用同一种语言。我可以更换代理、添加新代理，或者重新配置整个系统，而无需触及通信层。

Python A2A是 Google A2A 协议的全面实现，旨在：

• 直观——简单的 API，让 Python 开发人员感觉很自然
• 生产就绪——专为实际应用构建，具有强大的错误处理功能
• 与框架无关——可与 Flask、FastAPI、Django 或任何其他框架配合使用
• 最小依赖性——核心功能只需要requests库

入门很简单：

pip 安装 python-a2a

如果您想与特定的 LLM 提供商一起使用它：

# 对于 OpenAI 集成
pip install "python-a2a[openai]" 
# 对于 Anthropic Claude 集成
pip install "python-a2a[anthropic]" 
# 对于所有可选依赖项
pip install "python-a2a[all]"

构建您的第一个 A2A 代理

让我们从一些简单的事情开始：响应消息的回声代理。

从python_a2a导入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"Echo: {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，运行它，您就会得到一个在上运行的 A2A 兼容代理http://localhost:5000/a2a。

现在我们来谈谈它：

from python_a2a import A2AClient, Message, TextContent, MessageRole 

# 创建一个客户端与我们的代理对话
client = A2AClient( "http://localhost:5000/a2a" ) 
# 发送消息
message = Message( 
    content=TextContent(text= "Hello, is this thing on?" , 
    role=MessageRole.USER 
) 
respond = client.send_message(message) 
# 打印响应
print ( f"Agent says: {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.类型== “function_call”：
            function_name = message.content.name 
            params = {p.name：p.value for p in message.content.parameters}

尝试：
如果function_name == “calculate”：
                    operation = params.get（“operation”，“add”）
                    a = float（params.get（“a”，0））
                    b = float（params.get（“b”，0））

如果operation == “add”：
                        result = a + b 
#其他操作...

返回消息（
                        content = FunctionResponseContent（
                            name = “calculate”，
                            response = { “result”：result} 
                        ），
                        role = MessageRole.AGENT，
                        parent_message_id = message.message_id，
                        conversation_id = message.conversation_id 
                    ）

#其他功能...

除外异常为e：
返回消息（
                    内容=FunctionResponseContent（
                        name=function_name，
                        response={ “error” : str (e)} 
                    ), 
                    role=MessageRole.AGENT, 
                    parent_message_id=message.message_id, 
                    conversation_id=message.conversation_id 
                )
如果__name__ == “__main__” : 
    agent = CalculatorAgent() 
    run_server(agent, host= “0.0.0.0” , port= 5001 )

以下是我们调用其函数的方式：

从python_a2a导入（
    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）
如果response.content.type == “function_response”：
    result = respond.content.response.get（“result”）
如果result不是None：print（f“Result：{result} ”）   #输出：Result：8

妙就妙在，任何兼容 A2A 的代理都可以调用这些函数。这意味着你的 OpenAI 代理可以调用你的自定义计算器代理，后者又可以调用天气代理等等——所有这些都使用相同的标准化协议。

LLM 助力代理，轻松上手

Python A2A 包含与主流 LLM 提供商的即用型集成。以下是一个由 OpenAI 驱动的代理：

导入操作系统
从python_a2a导入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 )

只需三行代码（加上导入），您就能拥有一个完全兼容 A2A 的 OpenAI 代理。您可以对 Anthropic Claude 执行同样的操作，也可以为其他提供商构建自定义集成。

真实案例：研究助理工作流程

让我向您展示一个更复杂的现实世界示例：协调多个代理来回答研究问题的研究助理。

这个灵感来自于我为一位需要分析科学论文的客户构建的一个项目。其工作流程包括：

1. 根据研究问题生成搜索查询的 LLM 代理
2. 检索相关信息的搜索代理
3. 将信息综合成连贯答案的总结代理

使用 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" )   # 搜索代理
    summary_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} ', " 
f" 生成 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 ( "正在合成信息..." ) 
    summary_message = Message( 
        content=TextContent( 
            text= f"合成此信息以回答问题：' {query} '\n\n" 
f"信息:\n {search_results.content.text} "
         ), 
        role=MessageRole.USER 
    ) 
    summary_response = summary_client.send_message(summarize_message)
    对话。添加消息（摘要响应）

# 将最终答案添加到对话
    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 服务
• 特定领域的代理可以在项目之间共享和重用
• 复杂的代理编排变得像连接 API 一样简单
• 人工智能系统可以像乐高积木一样组装

这就是 A2A 和 Python A2A 正在帮助构建的未来。

Python A2A 入门

如果您准备简化代理通信并开始构建更加模块化的 AI 系统，请按照以下步骤开始：

1. 安装 Python A2A：pip install python-a2a
2. 查看GitHub 存储库中的示例
3. 阅读文档以获取详细的 API 参考

Python A2A 的优点在于您可以从小处着手 – 仅将一个代理转换为 A2A，然后随着您看到的好处逐渐在整个系统中采用它。

标准化为何重要

我构建了 Python A2A，因为我厌倦了将大部分开发时间花在管道设计上，而不是实际的 AI 逻辑上。A2A 协议提供的标准化为我节省了无数时间，并使我的多智能体系统更易于维护和扩展。

如果你正在构建多智能体人工智能系统，你可能面临着和我一样的挑战。人工智能生态系统只会变得越来越碎片化，如果没有标准化，其复杂性将变得难以管理。

Python A2A 为您提供了一种简单、可用于生产的方法来实现 Google 的 A2A 协议，并立即开始构建更加模块化、可互操作的代理系统。

尝试一下，然后让我知道它如何改变您的开发过程。

资源

• Python A2A GitHub 存储库
• PyPI 上的 Python A2A
• Python A2A 文档
• Google 的 A2A 协议规范
  
  参考：https://medium.com/@the_manoj_desai/meet-google-a2a-the-protocol-that-will-revolutionize-multi-agent-ai-systems-80d55a4583ed

（文：AI技术研习社）