A2 A Orchestrator

A2A 代理编排系统

!

一个分布式多智能体系统，它利用A2A（智能体到智能体）协议、MCP（模型上下文协议）和LangGraph，在专门化的AI智能体之间协调执行复杂任务。

架构概述

User Query → Orchestrator Agent → [Math Agent | Weather Agent | ...] → Coordinated Response

系统流程

1. 用户输入协调器接收自然语言查询
2. 规划协调器分析查询并创建执行计划
3. 任务分配任务被分配给合适的专门代理
4. 并行/顺序执行任务根据依赖关系执行
5. 响应协调结果被合并并返回给用户

核心组件

常用工具：

1. 基础代理（BaseAgent）：

• 特点：

- 大型语言模型（LLM）初始化支持OpenAI（GPT）和Google（Gemini）两种模型 - 内存管理使用LangGraph的 MemorySaver 为了保持对话的连贯性 - 异步生命周期处理异步初始化 _ensure_initialized() - 工具集成工具和提示定义的抽象方法 - 响应处理标准化的响应处理流程

2. BaseAgentExecutor（基类代理执行器）：

• 特点：

- 上下文管理进程 RequestContext 带有用户输入和会话数据 - 事件流用途 EventQueue 将响应流式传输回客户端 - 错误处理将异常转换为适当的A2A错误格式 - 生命周期管理确保在执行前代理被正确初始化

3. BaseAgentServer：（可翻译为）基础代理服务器

• 特点：

- 代理卡加载从JSON文件动态加载配置 - 服务器生命周期管理uvicorn服务器的启动/关闭 - A2A（Any-to-Any，任意到任意）集成创建 A2AStarletteApplication 配备合适的处理人员 - 请求路由用途 DefaultRequestHandler 符合A2A协议规范

4. 代理卡系统：

• 代理卡定义为：

- 能力代理所能执行的操作（流式处理、多模态处理等） - 技能带有示例和标签的特定功能 - 终点（或结局指标）网址和通信偏好 - 元数据版本、描述、支持的模式

代理商；经纪人

1. 交响乐协调器代理（端口10003）

• 目的中央协调器，负责规划和执行复杂的多智能体工作流程
• 能力:

- 查询分析与任务分解 - 智能代理路由 - 并行和顺序任务执行 - 依赖管理

• 模型GPT-4.1
• 技能任务规划，代理路由

2. 数学代理（端口10004）

• 目的专业数学计算代理
• 能力算术运算和功率计算
• 工具加、减、乘、除、平方、立方、幂运算
• 模型GPT-4.1
• 响应格式带有分步解题过程的结构化数学输出

3. 天气代理（端口10005）

• 目的使用MCP（模型上下文协议）获取天气信息
• 能力当前天气及预报
• 工具MCP天气服务器集成
• 模型GPT-4.1（注：GPT-4.1并非官方发布的版本名称，这里假设其为一个假设或特定语境下的版本标识进行翻译）
• 技能查询任何地点的天气

代理间通信

远程代理连接

协调器通过HTTP使用A2A协议与其他代理进行通信：

沟通流程：

1. 发现: create_from_url() 从（某处）获取代理卡 /agent-card 终端节点
2. 连接建立具有600秒超时的持久HTTP客户端
3. 发送消息： send_message() 发送A2A格式的请求
4. 响应处理从结构化的A2A响应格式中提取文本

消息流

Orchestrator → HTTP POST /send-message → Agent Server
                ↓
            A2A Protocol Message
                ↓
        {
          "id": "unique-id",
          "params": {
            "message": {
              "role": "user",
              "parts": [{"text": "user input"}]
            }
          }
        }

响应处理

代理返回结构化的响应，这些响应会经过多层处理：

1. LangGraph 输出返回结构化格式（例如。， MathResponseFormat)
2. 代理处理： _process_response() 提取相关内容
3. A2A包装内容被封装在A2A消息格式中
4. HTTP 响应通过HTTP发送的最终JSON响应

执行流水线

编排器工作流

1. 规划阶段:

- 大型语言模型（LLM）分析查询并生成 ExecutionPlan - 根据能力将任务分配给合适的代理 - 计算依赖关系以确保正确排序

1. 执行阶段:

- 根据任务关系构建的依赖图 - 已识别出可执行任务（无待处理依赖项） - 使用并行执行 asyncio.gather() - 收集结果并更新依赖项

1. 协调阶段：

- 依赖任务的结果传递给后续任务 - 最终响应由所有任务输出整合而成 - 摘要及状态返回给用户

并行执行与顺序执行示例

• 案例1：测试单一代理

- 输入：

  What is 5 + 7?

- 结果：

  Result: context_id=None extensions=None kind='message' message_id='6a628346-4caa-4f2e-be2b-ac75dfc7f01b' metadata=None parts=[Part(root=TextPart(kind='text', metadata=None, text='Execution Summary: A single math calculation task to compute the sum of 5 and 7.\n\nTask 1 (Math Agent): 5 + 7 = 12\n'))] reference_task_ids=None role= task_id=None

• 案例2：测试具有并发任务的多个代理

- 输入：

  Calculate 3 * 4 and tell me the weather in New York

- 结果：

  context_id=None extensions=None kind='message' message_id='d59c464a-b0e7-4ef2-9a2b-394a518c7bec' metadata=None parts=[Part(root=TextPart(kind='text', metadata=None, text='Execution Summary: First, calculate 3 * 4 using the Math Agent. Second, get the current weather in New York using the Weather Agent. Both tasks are independent and can be executed in parallel.\n\nTask 1 (Math Agent): 3 * 4 = 12\nTask 2 (Weather Agent): It seems there was an issue retrieving the weather for New York. Could you please try again later?\n'))] reference_task_ids=None role= task_id=None

• 案例3：测试具有顺序（依赖）任务的多个代理

- 输入：

  First calculate 3 × 4. Then, using that result as the day number of this month, tell me the weather in Cairo on that day.

- 结果：

  Result: context_id=None extensions=None kind='message' message_id='406de694-0deb-45be-a360-6f22345e0219' metadata=None parts=[Part(root=TextPart(kind='text', metadata=None, text='Execution Summary: First, calculate 3 × 4 to get 12. Then, get the weather in Cairo on the 12th day of this month.\n\nTask 1 (Math Agent): 3 × 4 = 12\nTask 2 (Weather Agent): The weather forecast for Cairo on the 12th day of this month is currently unavailable. Please try again later or provide additional details for assistance.\n'))] reference_task_ids=None role= task_id=None

目录结构

A2A-Orchestrator/
├── a2a_server/                               # Core package
│   ├── agent_cards/                          # Agent configuration
│   │   ├── math_agent_card.json
│   │   ├── orchestrator_agent_card.json
│   │   └── weather_agent_card.json
│   ├── agents/                               # Agent implementations
│   │   ├── math_agent_server/
│   │   ├── orchestrator_agent_server/
│   │   └── weather_agent_server/
│   ├── common/                               # Shared utilities
│   │   ├── agent_card_loader.py
│   │   ├── base_agent.py
│   │   ├── base_agent_executor.py
│   │   ├── base_agent_server.py
│   │   ├── models.py
│   │   ├── prompts.py
│   │   └── remote_agent_connection.py
│   └── mcp/                                  # Model Context Protocol
│       ├── servers/
│       │   └── weather.py
│       └── servers.json
├── a2a_server_manager.py                     # Main server manager
├── test_a2a_server.py                        # Integration tests
├── logger.py                                 # Debugging code
├── requirements.txt
├── pyproject.toml
├── README.md
└── settings.py

主要特点

智能编排

• 动态规划自动将复杂查询分解为可执行任务
• 依赖管理处理顺序和并行任务执行
• 代理发现自动发现并利用可用的专用代理

并行执行

• 独立任务同时运行以获得最佳性能
• 依赖解析（或依赖关系解析）当任务依赖于先前结果时，按顺序执行
• 混合执行根据需要结合并行和顺序模式

可扩展架构

• 插件系统易于添加新的专业代理
• MCP集成支持模型上下文协议以实现外部工具集成
• 代理卡基于JSON的代理能力描述

安装与设置

先决条件

• Python 3.10+
• UV 包管理器（推荐）或 pip

环境设置

1. 克隆项目并导航至项目目录：

   git clone 
   cd A2A-Orchestrator

1. 设置环境变量：

创建一个 .env 文件并设置环境变量：

   # settings.py
   OPENAI_API_KEY = "your-openai-key"
   OPENAI_BASE_URL = "https://api.openai.com/v1"  # Optional
   GOOGLE_API_KEY = "your-google-key"  # For Gemini models

安装方法

选项1：使用紫外线（推荐）

# Install UV if not already installed
curl -LsSf https://astral.sh/uv/install.sh | sh

# Install dependencies
uv sync

# Activate virtual environment
source .venv/bin/activate  # On Windows: .venv\Scripts\activate

选项2：使用Python/Pip

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install -r requirements.txt

运行系统

启动所有服务器

# Using UV
uv run python a2a_server_manager.py

# Using Python
python a2a_server_manager.py

这将同时启动所有代理服务器：

• 编排器代理：localhost:10003
• 数学代理：localhost:10004
• 天气代理：localhost:10005

启动独立代理（替代方案）

# Math Agent only
uv run python -m a2a_server.agents.math_agent_server

# Weather Agent only
uv run python -m a2a_server.agents.weather_agent_server

# Orchestrator only
uv run python -m a2a_server.agents.orchestrator_agent_server

测试

运行集成测试

# Make sure all servers are running first
python test_a2a_server.py

手动测试

# Test individual agents via HTTP API
curl -X POST http://localhost:10004/send-message \
  -H "Content-Type: application/json" \
  -d '{"message": "Calculate 5 + 7"}'

配置

代理卡

每个代理都有一张JSON配置卡，定义了：

• 能力和技能
• 支持的输入/输出模式
• 工具描述及示例
• API终端点

示例结构：

{
  "name": "Math Agent",
  "description": "Mathematical computation specialist",
  "url": "http://localhost:10004/",
  "skills": [
    {
      "id": "add",
      "name": "Addition",
      "description": "Add two numbers",
      "examples": ["5 + 7", "add 10 and 20"]
    }
  ]
}

MCP服务器配置

天气代理使用MCP进行外部工具集成：

{
  "Weather": {
    "command": "python",
    "args": ["-m", "a2a_server.mcp.servers.weather"],
    "transport": "stdio"
  },
  "Weather (UV)": {
    "command": "uv",
    "args": ["run", "python", "-m", "a2a_server.mcp.servers.weather"],
    "transport": "stdio"
  }
}

发展

添加新代理

1. 创建代理卡添加JSON配置到 agent_cards/
2. 实现代理扩展 BaseAgent 在 agents/
3. 创建服务器扩展 BaseAgentServer
4. 添加到管理器注册于 a2a_server_manager.py
5. 更新编排器代理将被自动发现

扩展功能

1. 添加工具为新功能实现LangChain工具
2. MCP集成通过模型上下文协议添加外部工具
3. 自定义提示定义特定于代理的行为 prompts.py
4. 响应格式添加结构化输出模型 models.py

依赖项

核心堆栈

• a2a-sdk（注：这通常指的是某个特定的软件开发工具包（SDK），在没有具体上下文的情况下，直接翻译为“a2a软件开发工具包”或保持原样“a2a-sdk”都是可以的，具体取决于需要传达的精确含义和上下文。）代理间通信协议
• langgraph基于图的代理编排
• LangChain（语言链）大型语言模型（LLM）框架与工具集成
• fastmcp模型上下文协议实现
• pydantic（注：这是一个Python库的名称，直接翻译可能无法准确传达其含义，但在此处可保持原样或解释为“一个用于数据验证的Python库”）数据验证和序列化

大型语言模型（LLM）提供商

• OpenAI（开放人工智能研究所）GPT模型（主要）
• 谷歌双子座模型（可选）

网络框架

• FastAPI/UvicornHTTP服务器基础设施
• httpx（注：httpx通常是一个用于HTTP请求的Python库，但在此处仅作为词汇翻译，不涉及具体功能或用途的解释）用于代理间通信的异步HTTP客户端

故障排除

常见问题

1. 端口冲突检查端口10003-10005是否可用
2. 速率限制问题共享API密钥问题

- 问题所有代理都使用在中配置的相同OpenAI API密钥 settings.py - 影响高请求量可能会触发429“请求过多”错误

1. 并行执行放大编曲器的并行任务执行可以发送多个同时请求。在复杂查询期间，会倍增速率限制压力。
2. 不支持流媒体当前实现缺乏实时流功能
3. 内存管理仅使用内存存储

调试

• 在代理中启用调试日志记录
• 检查单个代理的健康端点
• 使用 test_a2a_server.py 用于集成测试

性能考量

• 并行执行独立任务同时运行
• 连接池HTTP客户端重用连接
• 内存管理代理使用记忆保存器来管理对话状态
• 超时处理对长时间运行的操作设置10分钟的超时限制

安全注意事项

• 本地开发当前配置仅限于本地主机
• API密钥安全存储，切勿提交至版本控制系统
• 网络访问考虑生产环境部署时的防火墙规则

未来改进方向

• 额外的专门人员（编码、研究等）
• 增强的依赖解析算法
• 监控和可观测性功能
• 生产就绪的部署配置
• 支持WebSocket进行实时通信

许可证

这个项目遵循Apache许可证，第2版进行授权 - 请参阅 许可证 详情见文件。

代码在（此处） common/ 目录来自 谷歌A2A项目 并且也根据Apache许可证，第2版进行授权。

此项目还利用了其他开源库（例如，LangGraph、MCP、FastMCP），这些库均受其各自许可证的约束。

致谢

• MCP可以翻译为“最小化成本生产”（在商业或经济学语境中，通常指以最低成本实现生产目标），但具体含义需根据上下文确定。在其他领域，MCP可能有不同含义，如“多通道处理”（Multi-Channel Processing）等 – 为系统设计提供了部分灵感的模型上下文协议。
• A2A（汽车制造商和装配商协会） – 用于构建编排器和代理的概念及架构模式。
• LangGraph – 用于实现可组合的代理工作流和结构化编排。