Python A2A: Google 的智能体到智能体协议的全面指南

靖奇發表於 2025-4-16 14:33:00

Python A2A: Google 的智能体到智能体协议的全面指南

简介
Python A2A 是 Google 的智能体到智能体 (Agent-to-Agent, A2A) 协议的实现，旨在标准化 AI 智能体之间的通信。该协议解决了 AI 生态系统中的一个主要挑战：使不同的 AI 服务能够无需自定义转换层即可无缝通信。
随着 AI 领域分化为各种专业服务，每种服务都有自己的 API 格式和参数，开发人员花费过多时间构建通信基础设施，而不是专注于 AI 逻辑。Python A2A 通过提供一种标准化的方式让 AI 智能体互相交流，解决了这个问题，无论它们的底层实现如何。
<h2 id="1" data-heading="true">Python A2A 入门</h2>
<h3 id="2" data-heading="true">安装</h3>
<pre class="hljs code__pre"><code class="language-bash"># 基本安装 pip install python-a2a # 用于 OpenAI 集成 pip install "python-a2a" # 用于 Anthropic Claude 集成 pip install "python-a2a" # 安装所有可选依赖 pip install "python-a2a"</code></pre>
<h2 id="3" data-heading="true">核心概念</h2>
Python A2A 实现了 A2A 协议的几个关键概念：
<ol>
<li>1. 消息结构：定义了文本、函数调用和响应的格式</li>
<li>2. 对话线程：支持在多次交互中维护上下文</li>
<li>3. 函数调用：智能体暴露和调用函数的标准化方式</li>
<li>4. 错误处理：一致的错误格式</li>
</ol>
<h2 id="4" data-heading="true">构建您的第一个 A2A 智能体</h2>
让我们从一个简单的回声智能体开始，它会响应消息：
<pre class="hljs code__pre"><code class="language-python">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"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)</code></pre>
现在，让我们创建一个客户端来与我们的智能体交流：
<pre class="hljs code__pre"><code class="language-python">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 ) response = client.send_message(message) # 打印响应 print(f"Agent says: {response.content.text}")</code></pre>
<h2 id="5" data-heading="true">智能体之间的函数调用</h2>
A2A 的强大功能之一是标准化的函数调用。这是一个提供数学函数的计算器智能体：
<pre class="hljs code__pre"><code class="language-python">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: 基础算术 (operation, a, b)\n"                          "- sqrt: 平方根 (value)"                 ),                 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                     elif operation == "subtract":                         result = a - b                     elif operation == "multiply":                         result = a * b                     elif operation == "divide":                         if b == 0:                             raise ValueError("不能除以零")                         result = a / b                     else:                         raise ValueError(f"未知操作: {operation}")                     return Message(                         content=FunctionResponseContent(                             name="calculate",                             response={"result": result}                         ),                         role=MessageRole.AGENT,                         parent_message_id=message.message_id,                         conversation_id=message.conversation_id                     )                 elif function_name == "sqrt":                     value = float(params.get("value", 0))                     if value < 0:                         raise ValueError("不能计算负数的平方根")                     result = math.sqrt(value)                     return Message(                         content=FunctionResponseContent(                             name="sqrt",                             response={"result": 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={"error": 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)</code></pre>
以下是如何调用计算器智能体的函数：
<pre class="hljs code__pre"><code class="language-python">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: {result}")  # 输出: Result: 8</code></pre>
<h2 id="6" data-heading="true">基于 LLM 的智能体</h2>
Python A2A 包含了与流行 LLM 提供商的即用型集成。这是一个由 OpenAI 驱动的智能体：
<pre class="hljs code__pre"><code class="language-python">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)</code></pre>
同样，您可以创建一个由 Anthropic Claude 驱动的智能体：
<pre class="hljs code__pre"><code class="language-python">import os from python_a2a import ClaudeA2AServer, run_server # 创建一个由 Anthropic Claude 驱动的智能体 agent = ClaudeA2AServer(     api_key=os.environ["ANTHROPIC_API_KEY"],     model="claude-3-opus-20240229",     system_prompt="你是一个有帮助的 AI 助手。" ) # 运行服务器 if __name__ == "__main__":     run_server(agent, host="0.0.0.0", port=5003)</code></pre>
<h2 id="7" data-heading="true">多智能体工作流</h2>
A2A 的真正威力在于连接多个智能体。让我们构建一个研究助手工作流：
<pre class="hljs code__pre"><code class="language-python">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}', "                  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("综合信息...")     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)</code></pre>
<h2 id="8" data-heading="true">高级示例：使用 A2A 进行天气和行程规划</h2>
这是一个示例，展示了 A2A 如何简化智能体之间的通信：
<pre class="hljs code__pre"><code class="language-python">from python_a2a import 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 的智能体 - 无需自定义适配器！</code></pre>
<h2 id="9" data-heading="true">管理对话</h2>
Python A2A 提供了一个 <code>Conversation</code> 类来管理多轮交互：
<pre class="hljs code__pre"><code class="language-python">from python_a2a import Conversation, MessageRole, A2AClient # 创建一个新对话 conversation = Conversation() # 添加用户消息 conversation.create_text_message(     text="今天天气怎么样？",     role=MessageRole.USER ) # 连接到一个智能体 weather_agent = A2AClient("http://localhost:5001/a2a") # 发送对话中的最后一条消息 last_message = conversation.messages[-1] response = weather_agent.send_message(last_message) # 将响应添加到对话中 conversation.add_message(response) # 继续对话 conversation.create_text_message(     text="谢谢！我需要带伞吗？",     role=MessageRole.USER ) # 发送新消息，包括对话历史 response = weather_agent.send_message(     conversation.messages[-1],     conversation_id=conversation.conversation_id ) # 也添加这个响应 conversation.add_message(response) # 打印整个对话 for msg in conversation.messages:     role = "用户" if msg.role == MessageRole.USER else "智能体"     print(f"{role}: {msg.content.text}")</code></pre>
<h2 id="10" data-heading="true">错误处理</h2>
Python A2A 提供标准化的错误处理：
<pre class="hljs code__pre"><code class="language-python">from python_a2a import A2AClient, Message, TextContent, MessageRole client = A2AClient("http://localhost:5001/a2a") try:     message = Message(         content=TextContent(text="你好"),         role=MessageRole.USER     )     response = client.send_message(message) except ConnectionError as e:     print(f"连接错误: {e}") except TimeoutError as e:     print(f"超时错误: {e}") except Exception as e:     print(f"意外错误: {e}")</code></pre>
<h2 id="11" data-heading="true">创建自定义 A2A 智能体</h2>
以下是创建您自己的 A2A 兼容智能体的模板：
<pre class="hljs code__pre"><code class="language-python">from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server class MyCustomAgent(A2AServer):     def __init__(self, **kwargs):         super().__init__(**kwargs)         # 在这里初始化您的智能体特定组件     def handle_message(self, message):         """处理收到的消息并返回响应。         这是处理智能体消息的主要入口点。         """         if message.content.type == "text":             # 处理文本消息             user_text = message.content.text             # 生成您的响应（替换为您的逻辑）             response_text = f"你说: {user_text}"             return Message(                 content=TextContent(text=response_text),                 role=MessageRole.AGENT,                 parent_message_id=message.message_id,                 conversation_id=message.conversation_id             )         elif message.content.type == "function_call":             # 处理函数调用（如果您的智能体支持）             # ...             pass         # 如果没有其他条件匹配，返回默认响应         return Message(             content=TextContent(text="我不知道如何处理这种消息类型。"),             role=MessageRole.AGENT,             parent_message_id=message.message_id,             conversation_id=message.conversation_id         ) # 运行您的智能体 if __name__ == "__main__":     my_agent = MyCustomAgent()     run_server(my_agent, host="0.0.0.0", port=5005)</code></pre>
<h2 id="12" data-heading="true">结论</h2>
Python A2A 提供了一种标准化的方式让 AI
智能体进行通信，使构建多智能体系统变得更加容易。通过采用这个协议，您可以：
<ol>
<li>1. 减少在通信基础设施上花费的开发时间</li>
<li>2. 轻松切换和升级智能体实现</li>
<li>3. 构建模块化、可扩展的 AI 系统</li>
<li>4. 利用专业智能体完成不同任务</li>
</ol>
A2A 协议及其 Python 实现正在帮助构建一个未来，在这个未来中，AI 系统可以像乐高积木一样组合，让专业智能体无缝合作。
欲了解更多信息，请查看 Python A2A 的 GitHub 仓库、PyPI 页面和文档。
Python A2A: Google 的智能体到智能体协议的全面指南
  
来源：https://www.cnblogs.com/sing1ee/p/18828654

頁: [1]

琼殿技术论坛's Archiver

Python A2A: Google 的智能体到智能体协议的全面指南