跳转到内容

Agent2Agent (A2A) 协议

Agent2Agent (A2A) 协议是谷歌推出的一项开放标准，旨在实现 AI 智能体之间的通信和互操作性，无论它们是基于何种框架或由哪个供应商构建。

在 Pydantic，我们构建了 FastA2A 库，以便在 Python 中更轻松地实现 A2A 协议。

我们还构建了一个便捷方法，可以将 Pydantic AI 智能体暴露为 A2A 服务器——让我们快速了解一下如何使用它。

agent_to_a2a.py
from pydantic_ai import Agent

agent = Agent('openai:gpt-4.1', instructions='Be fun!')
app = agent.to_a2a()

你可以通过 uvicorn agent_to_a2a:app --host 0.0.0.0 --port 8000 运行该示例。

这将把智能体暴露为一个 A2A 服务器，你可以开始向它发送请求。

查看更多关于将 Pydantic AI 智能体暴露为 A2A 服务器的信息。

FastA2A

FastA2A 是 A2A 协议在 Python 中的一种与智能体框架无关的实现。该库旨在与任何智能体框架配合使用，并不局限于 Pydantic AI。

设计

FastA2A 基于 Starlette 构建，这意味着它完全兼容任何 ASGI 服务器。

考虑到 A2A 协议的特性，在使用之前理解其设计非常重要，因为作为开发者，你需要提供一些组件。

  • Storage：用于保存和加载任务，以及存储对话的上下文。
  • Broker：用于调度任务。
  • Worker：用于执行任务。

让我们来看看这些组件是如何协同工作的。

flowchart TB
    Server["HTTP Server"] <--> |Sends Requests/<br>Receives Results| TM

    subgraph CC[Core Components]
        direction RL
        TM["TaskManager<br>(coordinates)"] --> |Schedules Tasks| Broker
        TM <--> Storage
        Broker["Broker<br>(queues & schedules)"] <--> Storage["Storage<br>(persistence)"]
        Broker --> |Delegates Execution| Worker
    end

    Worker["Worker<br>(implementation)"]

FastA2A 允许你自带 Storage、Broker 和 Worker。

理解任务与上下文

在 A2A 协议中

  • 任务 (Task)：代表智能体的一次完整执行。当客户端向智能体发送消息时，会创建一个新任务。智能体运行直至完成（或失败），这整个执行过程被视为一个任务。最终的输出作为任务产物 (artifact) 存储。

  • 上下文 (Context)：代表一个可以跨越多个任务的对话线程。A2A 协议使用 context_id 来维持对话的连续性。

  • 当发送一条不带 context_id 的新消息时，服务器会生成一个新的 ID。
  • 后续的消息可以包含相同的 context_id 以继续对话。
  • 所有共享相同 context_id 的任务都可以访问完整的消息历史记录。

存储架构

Storage 组件有两个用途：

  1. 任务存储 (Task Storage)：以 A2A 协议格式存储任务，包括其状态、产物和消息历史记录。
  2. 上下文存储 (Context Storage)：以针对特定智能体实现优化的格式存储对话上下文。

这种设计允许智能体存储丰富的内部状态（例如，工具调用、推理轨迹），同时也能存储符合 A2A 格式的特定于任务的消息和产物。

例如，一个 Pydantic AI 智能体可能会将其完整的内部消息格式（包括工具调用和响应）存储在上下文存储中，而只将符合 A2A 规范的消息存储在任务历史记录中。

安装

FastA2A 已在 PyPI 上以 fasta2a 的名称发布，因此安装非常简单：

pip install fasta2a

uv add fasta2a

唯一的依赖项是：

你可以通过安装 Pydantic AI 的 a2a 附加项来包含 FastA2A。

pip install 'pydantic-ai-slim[a2a]'

uv add 'pydantic-ai-slim[a2a]'

Pydantic AI 智能体转 A2A 服务器

要将 Pydantic AI 智能体暴露为 A2A 服务器，你可以使用 to_a2a 方法。

agent_to_a2a.py
from pydantic_ai import Agent

agent = Agent('openai:gpt-4.1', instructions='Be fun!')
app = agent.to_a2a()

由于 app 是一个 ASGI 应用程序，它可以与任何 ASGI 服务器一起使用。

uvicorn agent_to_a2a:app --host 0.0.0.0 --port 8000

因为 to_a2a 的目标是成为一个便捷方法，所以它接受与 FastA2A 构造函数相同的参数。

使用 to_a2a() 时，Pydantic AI 会自动：

  • 将完整的对话历史记录（包括工具调用和响应）存储在上下文存储中。
  • 确保后续使用相同 context_id 的消息能够访问完整的对话历史记录。
  • 将智能体的结果持久化为 A2A 产物 (artifacts)。
  • 字符串结果会变成 TextPart 产物，并同时出现在消息历史记录中。
  • 结构化数据（Pydantic 模型、数据类、元组等）会变成 DataPart 产物，数据被包装为 {"result": <your_data>} 的形式。
  • 产物中包含带有类型信息和（如果可用）JSON schema 的元数据。