DataMate技术笔记(2)——项目技术栈和通信流程

🎯 项目概述

• 人机交互: ADK Web UI (给人用)

• 机器间通信: A2A协议 (给其他Agent调用)

🏗️ 技术栈架构

核心组件

┌─────────────────────────────────────────────────────────────┐
│                Provider Agent 项目                          │
├─────────────────────────────────────────────────────────────┤
│  ADK层 (Agent Development Kit)                             │
│  ├── Agent: 业务逻辑 + 工具调用                             │
│  ├── Tools: scan_data, match_request, provide_access      │
│  └── Model: LiteLLM (避免Vertex AI收费)                    │
├─────────────────────────────────────────────────────────────┤
│  A2A层 (Agent-to-Agent Protocol)                          │
│  ├── Agent Card: 能力声明                                 │
│  ├── Request Handler: 协议转换                            │
│  └── Task Store: 状态管理                                 │
├─────────────────────────────────────────────────────────────┤
│  服务层                                                    │
│  ├── ADK Web UI: http://localhost:8000 (人用)             │
│  └── A2A API: http://localhost:9000/a2a_api (Agent用)     │
└─────────────────────────────────────────────────────────────┘

技术选型

• ADK: Google Agent Development Kit v1.0 (Agent框架)

• A2A: Agent-to-Agent Protocol v0.2 (Agent间通信标准)

• LiteLLM: 多模型支持 (避免收费，支持100+模型)

• Starlette: Web框架 (A2A服务基础)

• Uvicorn: ASGI服务器

📡 通信流程详解

1. Agent发现流程 (A2A协议核心)

sequenceDiagram
    participant C as Consumer Agent
    participant P as Provider Agent
    participant AC as Agent Card

    Note over C,P: 阶段1: 服务发现
    C->>P: GET /.well-known/agent.json
    P->>AC: 读取Agent Card
    AC->>P: 返回能力信息
    P->>C: Agent Card (JSON)

    Note over C: 解析Provider能力
    Note over C: - 支持的技能
    Note over C: - 输入输出格式
    Note over C: - API端点信息

{
  "name": "Data Sharing Provider Agent",
  "url": "http://localhost:9000/a2a_api",
  "skills": [
    {
      "id": "scan_local_data",
      "name": "Scan Local Data",
      "examples": ["What datasets do you have?"]
    },
    {
      "id": "match_data_request",
      "name": "Match Data Request",
      "examples": ["I need sales data for Q2"]
    }
  ],
  "capabilities": {"streaming": true}
}

2. 任务执行流程 (A2A + ADK协作)

sequenceDiagram
    participant CA as Consumer Agent
    participant A2A as A2A Handler
    participant EXE as Agent Executor
    participant ADK as ADK Agent
    participant TOOL as Tool Implementation

    Note over CA,TOOL: 阶段2: 任务执行

    CA->>A2A: POST /a2a_api/<br/>{"message": "What data do you have?"}

    A2A->>EXE: DefaultRequestHandler.execute()
    EXE->>ADK: ProviderAgentExecutor → ADK Agent

    Note over ADK: LLM分析请求<br/>决定调用scan_local_data工具

    ADK->>TOOL: scan_local_data("all datasets")
    TOOL->>TOOL: 扫描本地文件系统
    TOOL->>ADK: {"datasets": ["sales.csv", "users.json"]}

    Note over ADK: LLM生成友好回复

    ADK->>EXE: "I found 2 datasets: sales.csv and users.json"
    EXE->>A2A: Agent响应
    A2A->>CA: {"result": {"text": "I found 2 datasets..."}}

3. 流式响应流程 (长任务处理)

sequenceDiagram
    participant CA as Consumer Agent
    participant A2A as A2A Streaming
    participant QUEUE as Event Queue
    participant ADK as ADK Agent

    CA->>A2A: POST /message/stream
    A2A->>QUEUE: 创建事件队列
    A2A->>ADK: 异步执行任务

    loop 处理进度
        ADK->>QUEUE: enqueue_event("正在扫描...")
        QUEUE->>A2A: 获取事件
        A2A->>CA: {"content": "正在扫描..."}

        ADK->>QUEUE: enqueue_event("找到3个文件...")
        QUEUE->>A2A: 获取事件
        A2A->>CA: {"content": "找到3个文件..."}
    end

    ADK->>QUEUE: enqueue_event("扫描完成", final=true)
    QUEUE->>A2A: 获取最终事件
    A2A->>CA: {"content": "扫描完成", "final": true}

🔄 内部数据流转

ADK Agent内部流程

用户请求 → LLM理解 → 选择工具 → 执行工具 → LLM整合 → 返回结果
    ↓           ↓          ↓         ↓         ↓         ↓
"找数据"   → 分析意图 → scan_data → 扫描文件 → 格式化 → "找到3个数据集"

A2A协议处理流程

HTTP请求 → JSON解析 → A2A验证 → 转发给Executor → ADK处理 → 格式化响应 → HTTP返回
    ↓         ↓         ↓            ↓          ↓          ↓         ↓
POST /   → message → 协议检查 → Agent调用 → 工具执行 → A2A格式 → JSON响应

💎 A2A协议的核心价值

1. 标准化通信

• 问题: 每个Agent都有自己的API格式，无法互通

• 解决: A2A提供统一的消息格式和发现机制

• 价值: Agent间可以像微服务一样组合

2. 服务发现机制

# 不用A2A - 硬编码集成
weather_response = requests.post("http://weather-service/api/v1/get",
                                json={"city": "Munich"})

# 使用A2A - 自动发现
agent_card = discover_agent("http://weather-agent/.well-known/agent.json")
if "weather_query" in agent_card.skills:
    response = a2a_client.send_message("What's weather in Munich?")

3. 协议层抽象

• 底层: 各种Agent用不同技术栈 (ADK, LangChain, CrewAI...)

• A2A层: 统一的消息协议

• 价值: 技术栈无关，可以混合使用

4. 实际应用场景

企业数据流水线

数据请求 → 路由Agent → 发现Provider → 数据获取 → 处理Agent → 结果返回

多Agent协作示例

用户: "帮我分析慕尼黑的房价趋势"

主Agent → 发现房价数据Agent (Provider) → 获取数据
       → 发现分析Agent → 数据分析
       → 发现可视化Agent → 生成图表
       → 整合结果返回用户

5. 与传统API的对比

🛠️ 项目实现细节

关键文件结构

provider-agent/
├── __main__.py          # 主服务启动 (A2A + 路由)
├── agent.py             # ADK标准Agent (给adk web用)
├── provider_agent/
│   ├── provider_agent.py     # ADK Agent配置 (LiteLLM)
│   ├── provider_agent_executor.py  # A2A执行器
│   └── provider_tools.py     # 业务工具实现
└── .env                 # 环境配置 (API Keys)

启动顺序

1. 设置环境变量 (.env文件)

2. 启动A2A服务: python __main__.py (端口9000)

3. 启动ADK UI: adk web (端口8000)

4. 测试两个接口:
• 人工测试: http://localhost:8000
• Agent测试: http://localhost:9000/a2a_api

重要概念理解

ADK Agent vs A2A Executor

• ADK Agent: 真正的AI逻辑，处理LLM交互和工具调用

• A2A Executor: 协议适配器，将A2A消息转换为ADK调用

Agent Card vs Agent实例

• Agent Card: 静态能力声明，给其他Agent看的"简历"

• Agent实例: 运行时的实际AI，处理具体请求

🚀 扩展方向

水平扩展 (更多Agent)

Provider Agent ←→ Consumer Agent ←→ Analysis Agent ←→ Visualization Agent

垂直扩展 (更多工具)

# 在ADK Agent中添加更多工具
tools=[
    scan_local_data,
    match_data_request,
    provide_data_access,
    # 新增工具
    validate_data_quality,
    encrypt_sensitive_data,
    log_access_audit
]

协议升级

• A2A v0.3: 支持更复杂的认证

• 状态同步: 多Agent间共享上下文

• 编排模式: 工作流定义

🎯 关键要点总结

1. A2A = Agent间的HTTP协议标准

2. ADK = 单个Agent的开发框架

3. 两者结合 = 可组合的Agent生态系统

4. 核心价值 = 标准化 + 自动发现 + 协作能力

5. 实际应用 = 企业级Agent网络，像微服务但更智能