A2A 代理编排系统
 
一个分布式多智能体系统，它利用A2A（智能体到智能体）协议、MCP（模型上下文协议）和LangGraph，在专门化的AI智能体之间协调执行复杂任务。
架构概述
User Query → Orchestrator Agent → [Math Agent | Weather Agent | ...] → Coordinated Response系统流程
- 用户输入协调器接收自然语言查询
- 规划协调器分析查询并创建执行计划
- 任务分配任务被分配给合适的专门代理
- 并行/顺序执行任务根据依赖关系执行
- 响应协调结果被合并并返回给用户
核心组件
常用工具：
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协议与其他代理进行通信：
沟通流程：
- 发现:
create_from_url()从（某处）获取代理卡/agent-card终端节点 - 连接建立具有600秒超时的持久HTTP客户端
- 发送消息：
send_message()发送A2A格式的请求 - 响应处理从结构化的A2A响应格式中提取文本
消息流
Orchestrator → HTTP POST /send-message → Agent Server
↓
A2A Protocol Message
↓
{
"id": "unique-id",
"params": {
"message": {
"role": "user",
"parts": [{"text": "user input"}]
}
}
}响应处理
代理返回结构化的响应，这些响应会经过多层处理：
- LangGraph 输出返回结构化格式（例如。，
MathResponseFormat) - 代理处理：
_process_response()提取相关内容 - A2A包装内容被封装在A2A消息格式中
- HTTP 响应通过HTTP发送的最终JSON响应
执行流水线
编排器工作流
- 规划阶段:
- 大型语言模型（LLM）分析查询并生成 ExecutionPlan - 根据能力将任务分配给合适的代理 - 计算依赖关系以确保正确排序
- 执行阶段:
- 根据任务关系构建的依赖图 - 已识别出可执行任务（无待处理依赖项） - 使用并行执行 asyncio.gather() - 收集结果并更新依赖项
- 协调阶段：
- 依赖任务的结果传递给后续任务 - 最终响应由所有任务输出整合而成 - 摘要及状态返回给用户
并行执行与顺序执行示例
- 案例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
环境设置
- 克隆项目并导航至项目目录：
git clone
cd A2A-Orchestrator- 设置环境变量：
创建一个 .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"
}
}发展
添加新代理
- 创建代理卡添加JSON配置到
agent_cards/ - 实现代理扩展
BaseAgent在agents/ - 创建服务器扩展
BaseAgentServer - 添加到管理器注册于
a2a_server_manager.py - 更新编排器代理将被自动发现
扩展功能
- 添加工具为新功能实现LangChain工具
- MCP集成通过模型上下文协议添加外部工具
- 自定义提示定义特定于代理的行为
prompts.py - 响应格式添加结构化输出模型
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客户端
故障排除
常见问题
- 端口冲突检查端口10003-10005是否可用
- 速率限制问题共享API密钥问题
- 问题所有代理都使用在中配置的相同OpenAI API密钥 settings.py - 影响高请求量可能会触发429“请求过多”错误
- 并行执行放大编曲器的并行任务执行可以发送多个同时请求。在复杂查询期间，会倍增速率限制压力。
- 不支持流媒体当前实现缺乏实时流功能
- 内存管理仅使用内存存储
调试
- 在代理中启用调试日志记录
- 检查单个代理的健康端点
- 使用
test_a2a_server.py用于集成测试
性能考量
- 并行执行独立任务同时运行
- 连接池HTTP客户端重用连接
- 内存管理代理使用记忆保存器来管理对话状态
- 超时处理对长时间运行的操作设置10分钟的超时限制
安全注意事项
- 本地开发当前配置仅限于本地主机
- API密钥安全存储，切勿提交至版本控制系统
- 网络访问考虑生产环境部署时的防火墙规则
未来改进方向
- 额外的专门人员（编码、研究等）
- 增强的依赖解析算法
- 监控和可观测性功能
- 生产就绪的部署配置
- 支持WebSocket进行实时通信
许可证
这个项目遵循Apache许可证，第2版进行授权 - 请参阅 许可证 详情见文件。
代码在（此处） common/ 目录来自 谷歌A2A项目 并且也根据Apache许可证，第2版进行授权。
此项目还利用了其他开源库（例如，LangGraph、MCP、FastMCP），这些库均受其各自许可证的约束。
致谢
- MCP可以翻译为“最小化成本生产”（在商业或经济学语境中，通常指以最低成本实现生产目标），但具体含义需根据上下文确定。在其他领域，MCP可能有不同含义，如“多通道处理”（Multi-Channel Processing）等 – 为系统设计提供了部分灵感的模型上下文协议。
- A2A（汽车制造商和装配商协会） – 用于构建编排器和代理的概念及架构模式。
- LangGraph – 用于实现可组合的代理工作流和结构化编排。