A2A 示例：Travel Planner OpenRouter

这是一个遵循 A2A（Agent2Agent）协议的 Python 实现。 基于 Google 官方 a2a-python SDK 和 OpenAI Python SDK 实现的旅行助手演示。 它是一个符合 OpenAI 模型规范的旅行助手，能够为您提供旅行规划服务。

源代码

A2A Travel Planner OpenRouter

项目架构

本项目演示了如何使用 A2A 协议构建可互操作的旅行规划代理，包括以下核心组件：

• Travel Planner Agent：基于 OpenAI 兼容接口的核心旅行助手逻辑
• Agent Executor：A2A 协议适配器，连接代理逻辑与 A2A 服务器
• A2A Server：符合 A2A 协议的服务器，提供标准化的代理间通信接口
• Loop Client：用于与 A2A 服务器交互的测试客户端

工作流程序列图

sequenceDiagram
    participant Client
    participant A2AServer
    participant RequestHandler
    participant Executor as TravelPlannerAgentExecutor
    participant Agent as TravelPlannerAgent
    participant LLM as OpenAI-Compatible LLM

    Client->>A2AServer: 请求代理卡片
    A2AServer->>Client: 返回代理卡片（技能、能力）

    Note over Client,A2AServer: 用户查询旅行规划

    Client->>A2AServer: message/sendStream（流式请求）
    A2AServer->>RequestHandler: 路由流式请求
    RequestHandler->>Executor: execute(context, event_queue)
    
    Executor->>Agent: stream(query)
    Agent->>LLM: chat.completions.create(stream=True)
    
    loop 流式响应处理
        LLM-->>Agent: 返回流式内容块
        Agent-->>Executor: yield {'content': chunk, 'done': False}
        Executor-->>RequestHandler: TaskArtifactUpdateEvent
        RequestHandler-->>A2AServer: 推送 SSE 事件
        A2AServer-->>Client: 流式内容更新
    end
    
    LLM-->>Agent: 最终响应完成
    Agent-->>Executor: yield {'content': '', 'done': True}
    Executor-->>RequestHandler: 最终 TaskArtifactUpdateEvent
    RequestHandler-->>A2AServer: 最终 SSE 事件
    A2AServer-->>Client: 流式响应完成

主要工作流程

1. 代理卡片检索：客户端首先从 A2A 服务器检索代理卡片，了解代理能力和技能
2. 流式请求处理：客户端发送包含用户查询的流式消息请求
3. 代理执行：代理执行器处理请求并调用旅行规划代理核心逻辑
4. LLM 交互：代理与 OpenAI 兼容的 LLM 进行流式对话
5. 实时响应：通过服务器发送事件（SSE）实时向客户端流式传输响应

快速开始

1. 配置环境变量：

复制示例文件并配置您的 API 凭据。

cp env.example .env

使用您的实际值编辑 .env 文件：

# 必需：您的 AI 模型服务 API 密钥
API_KEY=your_actual_api_key_here

# 可选：模型名称（默认：google/gemini-2.0-flash-001）
MODEL_NAME=google/gemini-2.0-flash-001

# 可选：API 服务的基础 URL
BASE_URL=https://openrouter.ai/api/v1

2. 安装依赖并启动服务器：

   uv venv
   source .venv/bin/activate
   
   uv sync
   uv run .
   

3. 在新终端中运行循环客户端：

   source .venv/bin/activate
   uv run loop_client.py
   

配置

应用程序使用环境变量进行配置：

• API_KEY（必需）：您的 AI 模型服务 API 密钥
• MODEL_NAME（可选）：要使用的模型名称（默认："google/gemini-2.0-flash-001"）
• BASE_URL（可选）：API 服务的基础 URL（默认："https://openrouter.ai/api/v1"）

技术特性

当前实现

• ✅ A2A 协议合规性：完全符合 Agent2Agent 协议规范
• ✅ 流式响应：支持实时流式内容生成
• ✅ OpenAI 兼容：支持任何 OpenAI 兼容的 API 接口
• ✅ 模块化设计：代理逻辑与协议适配的清晰分离
• ✅ 环境配置：灵活的环境变量配置

未来增强计划

任务状态管理增强

基于 Google A2A LangGraph 示例，计划添加以下功能：

• 🔄 任务生命周期管理：实现完整的任务状态跟踪（已提交 → 工作中 → 已完成/失败）
• 🔄 多轮对话支持：添加 input_required 状态以支持需要用户澄清的复杂旅行规划场景
• 🔄 任务持久化：为长时间运行的规划任务实现任务状态持久化
• 🔄 增强错误处理：更详细的错误状态和恢复机制
• 🔄 任务取消：支持取消正在进行的任务

状态管理示例

# 未来状态管理实现示例
class TravelPlannerTaskManager:
    async def handle_complex_query(self, query: str, context: RequestContext):
        # 检测是否需要更多信息
        if self.needs_clarification(query):
            return TaskStatus(
                state=TaskState.input_required,
                message="需要更多信息：请提供具体目的地、日期和预算范围"
            )
        
        # 执行复杂的多步骤规划
        task_id = await self.create_long_running_task(query)
        return TaskStatus(
            state=TaskState.working,
            taskId=task_id,
            message="正在创建详细的旅行计划..."
        )

计划功能添加

• 📋 结构化数据支持：添加 DataPart 支持，用于基于表单的旅行偏好收集
• 🖼️ 多媒体支持：支持 FilePart 用于生成和处理旅行图像、地图等
• 🔍 工具集成：集成外部 API（天气、航班、酒店等）作为工具调用
• 🌐 多语言支持：扩展多语言旅行规划能力
• 📊 分析指标：添加任务执行时间、成功率指标的收集

访问 A2A