# python_a2a_travel

**Repository Path**: enzoism/python_a2a_travel

## Basic Information

- **Project Name**: python_a2a_travel
- **Description**: 使用a2a开发旅行agent
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-12-23
- **Last Updated**: 2025-12-23

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Google的A2A智能体群聊

> 针对Google的A2A智能体群聊，进行一个基础的Demo演示

----

# 1-核心知识点

- 1-开发一个最简单的A2A大模型demo示例
- 2-熟悉A2A代码框架和基础使用

---

# 2-参考网址

- A2A代码示例仓库：https://gitee.com/enzoism/python_a2a_travel

-----

# 3-动手实操

## 1-UV环境搭建

```shell
# 1-uv环境搭建
uv python pin 3.11.4
uv init python_a2a && cd python_a2a
uv venv && source .venv/bin/activate
uv add python-dotenv pydantic

# 2-uv安装requirements.txt
uv python pin 3.11.4
uv venv && source .venv/bin/activate
uv pip install -r requirements.txt
uv add -r requirements.txt
```

----

## 2-A2A项目

![](https://i-blog.csdnimg.cn/direct/a9176db301f148da987efbc8ac1ffb7b.png)

----

## 3-启动A2A

```shell
adk web
```

- 打印日志

```shell
(python_a2a) (base) MacBook-Pro:python_a2a rong$ adk web
INFO:     Started server process [31551]
INFO:     Waiting for application startup.

+-----------------------------------------------------------------------------+
| ADK Web Server started                                                      |
|                                                                             |
| For local testing, access at http://127.0.0.1:8000.                         |
+-----------------------------------------------------------------------------+

INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)


```

![](https://i-blog.csdnimg.cn/direct/3e00b7528c8c4f748d76e3205b515ec2.png)

## 4-访问网址

>  拷贝控制台的地址：http://127.0.0.1:8000

![](https://i-blog.csdnimg.cn/direct/a9176db301f148da987efbc8ac1ffb7b.png)

----

# 4-代码说明

## 目录

- [项目概述](#项目概述)
- [技术架构](#技术架构)
- [项目结构](#项目结构)
- [环境配置](#环境配置)
- [核心代码解析](#核心代码解析)
- [运行方式](#运行方式)
- [扩展开发](#扩展开发)

---

## 项目概述

本项目是一个基于 Google ADK (Agent Development Kit) 的 **Agent-to-Agent (A2A)** 通信框架示例。通过多代理协作模式，实现了一个智能旅行规划系统，展示了主代理如何与多个子代理协同工作以完成复杂任务。

### 核心特性

- **多代理协作**：主代理 (Coordinator) 与子代理 (Flight Agent、Hotel Agent) 协同工作
- **自然语言处理**：通过 LLM 理解用户意图并提取结构化信息
- **分布式任务执行**：不同代理专注于特定领域的任务处理
- **统一状态管理**：自动管理会话状态和代理间通信

### 应用场景

```
用户输入 → 协调员代理 → [航班代理 + 酒店代理] → 综合结果
```

---

## 技术架构

### 架构分层

```
┌─────────────────────────────────────────────────────────┐
│                    用户交互层                            │
│              (自然语言输入/输出)                         │
└────────────────────┬────────────────────────────────────┘
                     │
┌────────────────────▼────────────────────────────────────┐
│              协调员代理 (Coordinator)                     │
│     • 信息收集与验证                                      │
│     • 任务分发与协调                                      │
│     • 结果聚合与呈现                                      │
└─────┬──────────────────────────┬────────────────────────┘
      │                          │
┌─────▼────────┐         ┌──────▼─────────┐
│  航班代理     │         │   酒店代理      │
│ (Flight Agent)│         │ (Hotel Agent)  │
│              │         │                │
│ • 航班查询   │         │ • 酒店查询     │
│ • 价格计算   │         │ • 房间推荐     │
│ • 时刻匹配   │         │ • 设施展示     │
└──────────────┘         └────────────────┘
```

### 技术栈

| 组件 | 技术 | 版本要求 | 说明 |
|------|------|----------|------|
| Agent 框架 | `google-adk` | ≥1.21.0 | Google Agent Development Kit |
| LLM 集成 | `lite-llm` | - | 统一的 LLM 接口层 |
| 数据验证 | `pydantic` | ≥2.12.5 | 数据模型与验证 |
| 环境管理 | `python-dotenv` | ≥1.2.1 | 环境变量加载 |
| 代码执行 | `rawdog-ai` | ≥0.1.6 | 动态代码执行 (可选) |
| Python | - | ≥3.11.4 | 运行环境 |

---

## 项目结构

### 目录树

```
python_a2a/
├── .env                        # 环境变量配置 (API密钥等)
├── .gitignore                  # Git忽略规则
├── pyproject.toml              # 项目元数据与依赖声明
├── requirements.txt            # 简化的依赖列表
├── settings.py                 # 全局配置模块
├── uv.lock                     # 依赖锁定文件 (UV包管理器)
│
├── travel_planner_agent/       # 主应用包
│   ├── agent.py                # 【核心】主协调员代理定义
│   ├── subagents/              # 子代理模块
│   │   ├── flight_agent.py     # 航班查询代理
│   │   └── hotel_agent.py      # 酒店推荐代理
│   └── .adk/
│       └── session.db          # 会话持久化存储
│
└── .venv/                      # Python虚拟环境
```

### 关键文件说明

| 文件 | 作用 | 关键点 |
|------|------|--------|
| `agent.py` | 定义协调员代理，整合子代理 | 导出 `root_agent` 作为入口 |
| `subagents/flight_agent.py` | 航班查询子代理 | 导出 `flight_agent` 实例 |
| `subagents/hotel_agent.py` | 酒店推荐子代理 | 导出 `hotel_agent` 实例 |
| `settings.py` | 全局配置，环境变量管理 | 统一加载 `.env` |
| `.env` | 敏感信息存储 | **不提交到版本控制** |

---

## 环境配置

### 1. 安装依赖

#### 使用 UV (推荐，快速)

```bash
# 安装 UV (如果尚未安装)
curl -LsSf https://astral.sh/uv/install.sh | sh

# 同步依赖
uv sync

# 激活虚拟环境
source .venv/bin/activate  # Linux/Mac
# 或
.venv\Scripts\activate     # Windows
```

#### 使用传统 pip

```bash
# 创建虚拟环境
python3.11 -m venv .venv

# 激活虚拟环境
source .venv/bin/activate  # Linux/Mac

# 安装依赖
pip install -r requirements.txt
```

### 2. 配置环境变量

创建 `.env` 文件（已包含在项目中）：

```bash
# OpenAI 兼容 API 配置
OPENAI_API_KEY=your_api_key_here
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=openai/gpt-4

# 或使用其他兼容服务 (例如 DeepSeek)
OPENAI_API_KEY=sk-xxxxxxxxxxxx
OPENAI_BASE_URL=https://api.deepseek.com/v1
OPENAI_MODEL=openai/deepseek-chat
```

> **注意**：`.env` 文件已在 `.gitignore` 中，不会提交到代码仓库

### 3. 验证安装

```python
# test_setup.py
from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from settings import OPENAI_MODEL

print(f"✓ ADK 已安装")
print(f"✓ 模型配置: {OPENAI_MODEL}")

# 测试模型连接
model = LiteLlm(model=OPENAI_MODEL)
print("✓ 环境配置完成")
```

---

## 核心代码解析

### 1. 配置管理 (`settings.py`)

```python
import os
from pathlib import Path
from dotenv import load_dotenv

# 显式指定 .env 路径，避免路径问题
load_dotenv(Path(__file__).resolve().parent / ".env")

# 导出环境变量
OPENAI_BASE_URL = os.getenv("OPENAI_BASE_URL")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_MODEL = os.getenv("OPENAI_MODEL")
```

**设计原则**：
- **单一职责**：配置管理独立于业务逻辑
- **路径安全**：使用 `Path` 对象确保跨平台兼容
- **集中管理**：所有环境变量统一加载

---

### 2. 子代理实现

#### 航班代理 (`flight_agent.py`)

```python
from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from settings import OPENAI_MODEL

# 初始化模型
openai_model = LiteLlm(model=OPENAI_MODEL)

# 定义航班代理
flight_agent = LlmAgent(
    name="flight_agent",
    model=openai_model,
    description="根据出发地、目的地、旅行日期和预算提供建议的航班。",

    instruction="""
    你是一个航班预订代理。协调员会给你以下信息：
    - 出发地
    - 目的地
    - 起始日期
    - 结束日期
    - 预算金额
    - 预算货币

    返回1-2个模拟航班选项，包括：
    - 航空公司名称
    - 出发和返回日期/时间
    - 价格（以指定货币表示）
    - 班级（经济舱/商务舱）

    确保总价格在预算范围内。
    """
)
```

**关键设计**：

| 参数 | 作用 | 示例 |
|------|------|------|
| `name` | 代理的唯一标识 | `"flight_agent"` |
| `model` | LLM 模型实例 | `LiteLlm(model="...")` |
| `description` | 功能描述，用于代理发现 | "提供建议的航班" |
| `instruction` | 详细的行为指令 | Markdown格式的提示词 |

---

#### 酒店代理 (`hotel_agent.py`)

```python
from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from settings import OPENAI_MODEL

openai_model = LiteLlm(model=OPENAI_MODEL)

hotel_agent = LlmAgent(
    name="hotel_agent",
    model=openai_model,
    description="在给定的旅行日期内找到符合目的地和预算的酒店。",

    instruction="""
    你是一个酒店预订代理。协调员会给你以下信息：
    - 目的地
    - 起始日期
    - 结束日期
    - 预算金额
    - 预算货币

    返回1-2个模拟酒店建议，包括：
    - 酒店名称
    - 每晚价格和总成本（以给定货币表示）
    - 主要设施

    确保总价格符合预算。
    """
)
```

---

### 3. 主协调员代理 (`agent.py`)

```python
from datetime import datetime
from dotenv import load_dotenv
from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from settings import OPENAI_MODEL
from travel_planner_agent.subagents.flight_agent import flight_agent
from travel_planner_agent.subagents.hotel_agent import hotel_agent

load_dotenv()

openai_model = LiteLlm(model=OPENAI_MODEL)
today = datetime.today().date()

# 协调员代理
coordinator_agent = LlmAgent(
    name="TravelCoordinator",
    model=openai_model,
    description="主要的协调员代理，收集旅行偏好并查询子代理。",

    instruction=f"""
    你是一个旅行规划协调员。
    你的任务是从用户那里收集旅行偏好，并与子代理协调，
    提供航班、酒店建议和日程安排。

    步骤1：提取信息
    - 出发地、目的地
    - 起始日期、结束日期 (格式: YYYY-MM-DD)
    - 预算金额、预算货币

    步骤2：处理缺失信息
    - 起始日期缺失：建议使用 {today}
    - 结束日期缺失：根据天数计算
    - 货币缺失：默认为 "USD"

    步骤3：确认信息
    向用户确认旅行偏好

    步骤4：调用子代理
    - `flight_agent` 用于航班建议
    - `hotel_agent` 用于酒店建议

    步骤5：呈现结果
    - 旅行摘要
    - 航班建议
    - 酒店建议
    - 总估算成本
    - 一日行程计划示例
    """,

    sub_agents=[flight_agent, hotel_agent]  # 🔑 关键：子代理列表
)

# 导出根代理 (ADK 框架约定)
root_agent = coordinator_agent
```

**核心机制**：

1. **`sub_agents` 参数**：声明子代理列表，框架自动处理代理间通信
2. **模板字符串**：使用 `f"{today}"` 动态插入当前日期
3. **导出约定**：必须导出 `root_agent` 变量作为入口点

---

## 运行方式

### 方式 1：使用 ADK CLI (推荐)

```bash
# 安装 ADK CLI (如果需要)
pip install google-adk

# 启动交互式会话
adk start travel_planner_agent

# 或者指定配置文件
adk start --config travel_planner_agent/.adk/config.json
```

### 方式 2：程序化调用

```python
# run_agent.py
import asyncio
from google.adk.agents import LlmAgent
from travel_planner_agent.agent import root_agent

async def main():
    # 用户输入
    user_message = """
    我想从北京到上海旅行，从 2025-01-15 到 2025-01-20，
    预算是 5000 元人民币
    """

    # 调用代理
    response = await root_agent.arun(user_message)

    # 输出结果
    print(response.content)

if __name__ == "__main__":
    asyncio.run(main())
```

### 方式 3：作为服务运行

```python
# server.py
from google.adk.server import AgentServer
from travel_planner_agent.agent import root_agent

# 启动 FastAPI 服务器
server = AgentServer(agent=root_agent)
server.run(host="0.0.0.0", port=8000)
```

然后通过 HTTP 调用：

```bash
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "帮我计划从纽约到洛杉矶的旅行"}'
```

---

## 扩展开发

### 添加新的子代理

假设要添加一个"租车代理"：

#### 1. 创建新文件 `travel_planner_agent/subagents/car_rental_agent.py`

```python
from google.adk.agents import LlmAgent
from google.adk.models.lite_llm import LiteLlm
from settings import OPENAI_MODEL

openai_model = LiteLlm(model=OPENAI_MODEL)

car_rental_agent = LlmAgent(
    name="car_rental_agent",
    model=openai_model,
    description="提供目的地租车服务建议",

    instruction="""
    你是一个租车代理。你会收到：
    - 目的地
    - 起始日期
    - 结束日期
    - 预算

    返回 1-2 个租车选项，包括：
    - 租车公司
    - 车型
    - 每日租金和总价
    - 保险选项
    """
)
```

#### 2. 修改 `agent.py`，集成新代理

```python
from travel_planner_agent.subagents.car_rental_agent import car_rental_agent

coordinator_agent = LlmAgent(
    # ... 其他参数不变 ...

    instruction="""
    ... 现有指令 ...

    步骤4：调用子代理
    - `flight_agent` 用于航班建议
    - `hotel_agent` 用于酒店建议
    - `car_rental_agent` 用于租车建议  # 新增

    步骤5：呈现结果
    ... 现有内容 ...
    - 租车建议  # 新增
    """,

    sub_agents=[flight_agent, hotel_agent, car_rental_agent]  # 添加新代理
)
```

---

### 集成外部 API

使用 **工具 (Tools)** 扩展代理能力：

```python
from google.adk.tools import tool

@tool
def search_flights(origin: str, destination: str, date: str) -> dict:
    """调用真实航班 API"""
    # 示例：使用 requests 调用外部 API
    import requests
    response = requests.get(
        "https://api.example.com/flights",
        params={
            "origin": origin,
            "destination": destination,
            "date": date
        }
    )
    return response.json()

# 将工具绑定到代理
flight_agent = LlmAgent(
    name="flight_agent",
    model=openai_model,
    tools=[search_flights],  # 添加工具列表
    instruction="..."
)
```

---

### 数据持久化

ADK 自动将会话状态存储在 `.adk/session.db` (SQLite)。

自定义持久化配置：

```python
from google.adk.storage import SQLiteStorage

# 自定义存储路径
storage = SQLiteStorage(db_path="/path/to/custom.db")

coordinator_agent = LlmAgent(
    name="TravelCoordinator",
    model=openai_model,
    storage=storage,  # 注入自定义存储
    # ...
)
```

---

## 最佳实践

### 1. 提示词工程

```python
# ✅ 推荐：结构化、清晰的指令
instruction = """
你是航班代理。请按以下步骤操作：

步骤1：解析输入
- 提取出发地、目的地、日期
- 验证日期格式 (YYYY-MM-DD)

步骤2：生成建议
- 返回 2-3 个选项
- 包含价格、时间、航空公司

步骤3：验证预算
- 确保总价不超过预算
"""

# ❌ 避免：模糊、冗长的指令
instruction = "你是一个很棒的航班代理，请帮助用户找到最好的航班..."
```

### 2. 代理职责分离

```python
# ✅ 推荐：单一职责
flight_agent = LlmAgent(name="flight_agent", ...)
hotel_agent = LlmAgent(name="hotel_agent", ...)
car_agent = LlmAgent(name="car_agent", ...)

# ❌ 避免：全能代理
super_agent = LlmAgent(name="super_agent", ...)  # 处理所有任务
```

### 3. 错误处理

```python
from google.adk.agents import LlmAgent
from google.adk.exceptions import AgentError

try:
    response = await root_agent.arun(user_input)
except AgentError as e:
    print(f"代理执行失败: {e}")
    # 实现回退逻辑或用户通知
```

---

## 依赖项详解

### 核心依赖

```toml
# pyproject.toml
[project]
name = "python-a2a"
version = "0.1.0"
requires-python = ">=3.11.4"

dependencies = [
    "google-adk>=1.21.0",   # Agent 框架核心
    "pydantic>=2.12.5",     # 数据验证
    "python-dotenv>=1.2.1", # 环境变量
    "rawdog-ai>=0.1.6",     # 代码执行能力
]
```

### 依赖关系图

```
python-a2a
├── google-adk (Agent Development Kit)
│   ├── llm-models (LiteLLM 集成)
│   ├── storage (SQLite 持久化)
│   └── communication (Agent 间消息传递)
│
├── pydantic (数据验证)
│   ├── 类型检查
│   └── 序列化/反序列化
│
├── python-dotenv
│   └── .env 文件解析
│
└── rawdog-ai (可选)
    └── 动态代码执行沙箱
```

---

## 常见问题

### Q1: 如何切换 LLM 提供商?

修改 `.env` 文件：

```bash
# OpenAI
OPENAI_MODEL=openai/gpt-4

# Anthropic Claude
OPENAI_MODEL=anthropic/claude-3-sonnet
OPENAI_BASE_URL=https://api.anthropic.com

# 本地模型 (Ollama)
OPENAI_MODEL=ollama/llama2
OPENAI_BASE_URL=http://localhost:11434
```

### Q2: 代理间如何传递数据?

ADK 框架自动处理。在 `instruction` 中使用 **变量引用**：

```python
instruction="""
将以下信息传递给 flight_agent:
- {origin}      # 自动从上下文提取
- {destination}
- {budget}
"""
```

### Q3: 如何调试代理行为?

启用详细日志：

```python
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("google.adk")
logger.setLevel(logging.DEBUG)
```

---

## 进阶话题

### 1. 流式响应

```python
async for chunk in root_agent.astream("帮我规划旅行"):
    print(chunk.content, end="", flush=True)
```

### 2. 并行代理调用

```python
from google.adk.agents import ParallelAgent

# 并行组
parallel_agent = ParallelAgent(
    name="parallel_search",
    sub_agents=[flight_agent, hotel_agent],
    mode="all"  # 同时调用所有子代理
)
```

### 3. 条件路由

```python
from google.adk.agents import ConditionalAgent

router = ConditionalAgent(
    name="task_router",
    rules={
        "flight": flight_agent,
        "hotel": hotel_agent
    },
    default=coordinator_agent
)
```

---

## 资源链接

- **Google ADK 文档**: https://github.com/google/adk
- **LiteLLM 文档**: https://docs.litellm.ai/
- **Pydantic 文档**: https://docs.pydantic.dev/

---

## 总结

A2A 框架通过 **代理协作模式** 简化了复杂 AI 应用的开发：

| 优势 | 说明 |
|------|------|
| **模块化** | 每个代理专注于单一任务 |
| **可扩展** | 轻松添加新的子代理 |
| **可维护** | 代码结构清晰，职责分离 |
| **生产就绪** | 内置持久化、错误处理 |

通过本指南，你应该能够：
- ✅ 理解 A2A 架构设计
- ✅ 创建自己的多代理系统
- ✅ 集成外部 API 和工具
- ✅ 部署到生产环境

**开始构建你的第一个 A2A 应用吧！** 🚀