A2A Protocol

基于A2A和ADK的内容规划代理 (Content Planner Agent)

deutschEnglishfrançaisहिंदी日本語한국어中文
MILO•
Share
ADK
Content Planner Agent Based on A2A and ADK

项目概述

Content Planner Agent 是一个基于 Google Agent Development Kit (ADK) 和 Python A2A SDK 构建的智能内容规划代理。该代理能够根据高层次的内容描述，创建详细的内容大纲。

什么是A2A Protocol

A2A Protocol（Agent2Agent 协议）是一种专为 AI 智能体设计的开放标准协议。它的核心目标是实现不同平台、不同技术栈下的智能体之间的互操作性，让它们能够像“同事”一样协作完成任务，无论背后用的是什么技术。

技术栈

  • Python: 3.10+
  • UV: Python 包管理器
  • Google ADK: Google Agent Development Kit
  • A2A SDK: Agent-to-Agent 通信协议
  • Gemini 2.5 Flash: 大语言模型
  • Google Search: 搜索工具
  • Uvicorn: ASGI 服务器

前置要求

1. 环境准备

确保您的系统已安装以下软件：

# 检查 Python 版本 (需要 3.10+)
python --version

# 安装 UV 包管理器 (如果尚未安装)
curl -LsSf https://astral.sh/uv/install.sh | sh
# 或使用 pip
pip install uv

2. API 密钥

您需要获取 Google API 密钥以使用 Gemini 模型和 Google Search 功能：

  1. 访问 Google AI Studio
  2. 创建新的 API 密钥
  3. 保存密钥以备后用

项目结构

samples/python/agents/content_planner/
├── __init__.py                 # 包初始化文件
├── __main__.py                # 主入口文件
├── agent_executor.py          # 代理执行器
├── content_planner_agent.py   # 内容规划代理定义
├── pyproject.toml            # 项目配置文件
├── requirements.txt          # 依赖列表
├── .env.example             # 环境变量示例
└── README.md               # 项目说明

快速开始

步骤 1: 克隆项目并导航到目录

# 假设您已经有 a2a-samples 项目
git clone https://github.com/a2aproject/a2a-samples.git
cd a2a-samples/samples/python/agents/content_planner

步骤 2: 配置环境变量

# 复制环境变量示例文件
cp .env.example .env

# 编辑 .env 文件，添加您的 Google API 密钥
echo "GOOGLE_API_KEY=your_actual_api_key_here" > .env

步骤 3: 安装依赖并运行代理

# 使用 UV 安装依赖并运行项目
uv run .

# 注意：
"gradio>=5.30.0" 在当前 agent 中，是不需要的。

默认情况下，代理将在 http://localhost:10001 启动。

步骤 4: 测试代理 (新终端窗口)

# 导航到 CLI 客户端目录
cd samples/python/hosts/cli

# 连接到代理并发送消息
uv run . --agent http://localhost:10001

步骤 5: 与代理交互

在 CLI 客户端中，您可以发送如下消息：

Create an outline for a short, upbeat, and encouraging X post about learning Java

代码详解

1. 主入口文件 (__main__.py)

@click.command()
@click.option("--host", default="localhost")
@click.option("--port", default=10001)
def main(host, port):
    # Agent card (metadata)
    agent_card = AgentCard(
        name='Content Planner Agent',
        description=content_planner_agent.description,
        url=f'http://{host}:{port}',
        version="1.0.0",
        defaultInputModes=["text", "text/plain"],
        defaultOutputModes=["text", "text/plain"],
        capabilities=AgentCapabilities(streaming=True),
        skills=[
            AgentSkill(
                id="content_planner",
                name="Creates outlines for content",
                description="Creates outlines for content given a high-level description of the content",
                tags=["plan", "outline"],
                examples=[
                    "Create an outline for a short, upbeat, and encouraging X post about learning Java",
                ],
            )
        ],
    )

关键组件说明：

  • AgentCard: 代理的元数据卡片，包含名称、描述、URL、版本等信息
  • AgentSkill: 定义代理的技能，包括 ID、名称、描述、标签和示例
  • AgentCapabilities: 代理的能力配置，如是否支持流式输出

2. 代理定义 (content_planner_agent.py)

from google.adk.agents import Agent
from google.adk.tools import google_search

root_agent = Agent(
    name="content_planner_agent",
    model="gemini-2.5-flash",
    description=("Planning agent that creates a detailed and logical outline for a piece of content,"
                 "given a high-level description."),
    instruction=("You are an expert content planner. Your task is to create a detailed and logical outline for a piece"
                 "of content, given a high-level description."),
    tools=[google_search],
)

关键特性：

  • 模型: 使用 Gemini 2.5 Flash 作为底层 LLM
  • 工具: 集成 Google Search 工具，可以搜索相关信息
  • 指令: 明确定义代理的角色和任务

3. 代理执行器 (agent_executor.py)

class ADKAgentExecutor(AgentExecutor):
    def __init__(self, agent, status_message="Processing request...", artifact_name="response"):
        self.agent = agent
        self.status_message = status_message
        self.artifact_name = artifact_name
        self.runner = Runner(
            app_name=agent.name,
            agent=agent,
            artifact_service=InMemoryArtifactService(),
            session_service=InMemorySessionService(),
            memory_service=InMemoryMemoryService(),
        )

核心功能：

  • Runner: ADK 运行器，管理代理的执行
  • 服务组件:
    • ArtifactService: 管理生成的工件
    • SessionService: 管理会话状态
    • MemoryService: 管理对话记忆

系统架构流程图

graph TB
    A[用户请求] --> B[A2A CLI 客户端]
    B --> C[HTTP 请求]
    C --> D[A2A Starlette 应用]
    D --> E[DefaultRequestHandler]
    E --> F[ADKAgentExecutor]
    F --> G[ADK Runner]
    G --> H[Content Planner Agent]
    H --> I[Gemini 2.5 Flash 模型]
    H --> J[Google Search 工具]
    I --> K[生成内容大纲]
    J --> K
    K --> L[响应工件]
    L --> M[任务完成]
    M --> N[返回结果给用户]
    
    subgraph "ADK 组件"
        O[InMemoryArtifactService]
        P[InMemorySessionService]
        Q[InMemoryMemoryService]
    end
    
    G --> O
    G --> P
    G --> Q

详细执行流程

1. 初始化阶段

sequenceDiagram
    participant Main as __main__.py
    participant Agent as content_planner_agent
    participant Executor as ADKAgentExecutor
    participant Server as A2AStarletteApplication
    
    Main->>Agent: 加载代理配置
    Main->>Executor: 创建执行器实例
    Main->>Server: 创建 A2A 服务器
    Server->>Server: 启动 Uvicorn 服务器

2. 请求处理阶段

sequenceDiagram
    participant Client as CLI 客户端
    participant Server as A2A 服务器
    participant Handler as DefaultRequestHandler
    participant Executor as ADKAgentExecutor
    participant Runner as ADK Runner
    participant Model as Gemini 2.5 Flash
    participant Search as Google Search
    
    Client->>Server: 发送内容规划请求
    Server->>Handler: 路由请求
    Handler->>Executor: 执行代理任务
    Executor->>Runner: 启动 ADK 运行器
    Runner->>Model: 调用 Gemini 模型
    Runner->>Search: 执行 Google 搜索
    Model->>Runner: 返回生成内容
    Search->>Runner: 返回搜索结果
    Runner->>Executor: 合并结果
    Executor->>Handler: 返回工件
    Handler->>Server: 完成任务
    Server->>Client: 返回内容大纲

高级配置

自定义端口

# 在指定端口启动代理
uv run . --port=8080

自定义主机

# 在指定主机和端口启动
uv run . --host=0.0.0.0 --port=8080

环境变量配置

在 .env 文件中可以配置更多选项：

GOOGLE_API_KEY=your_api_key_here
# 可以添加其他配置项
LOG_LEVEL=INFO

故障排除

常见问题

  1. API 密钥错误

    错误: Invalid API key
解决: 检查 .env 文件中的 GOOGLE_API_KEY 是否正确

  2. 端口占用

    错误: Port 10001 is already in use
解决: 使用 --port 参数指定其他端口

  3. 依赖安装失败

    错误: Failed to install dependencies
解决: 确保 UV 已正确安装，尝试 uv sync

扩展和定制

添加新工具

在 content_planner_agent.py 中添加新工具：

from google.adk.tools import google_search, web_search

root_agent = Agent(
    # ... 其他配置
    tools=[google_search, web_search],  # 添加更多工具
)

修改模型

root_agent = Agent(
    name="content_planner_agent",
    model="gemini-1.5-pro",  # 使用不同的模型
    # ... 其他配置
)

自定义指令

root_agent = Agent(
    # ... 其他配置
    instruction=(
        "You are a specialized content planner for technical documentation. "
        "Create detailed outlines that include code examples and best practices."
    ),
)

最佳实践

  1. 安全性:

    • 始终将 API 密钥存储在环境变量中
    • 不要将 .env 文件提交到版本控制

  2. 性能优化:

    • 使用适当的模型大小
    • 合理配置内存和会话服务

  3. 错误处理:

    • 实现适当的错误处理和日志记录
    • 提供有意义的错误消息

  4. 测试:

    • 编写单元测试和集成测试
    • 使用不同的输入测试代理响应

总结

Content Planner Agent 展示了如何使用 Google ADK 和 A2A 协议构建智能代理。通过本指南，您应该能够：

  • 理解项目的整体架构
  • 成功运行和测试代理
  • 根据需要进行定制和扩展
  • 解决常见问题

这个代理可以作为构建更复杂多代理系统的基础，例如完整的内容创作工作流。

更多A2A Protocol 示例