一个用于管理和协调多个AI Agent的Python框架，能够自动发现Agent功能、生成执行计划、分解复杂任务，并提供完整的任务追踪。

A Python framework for managing and coordinating multiple AI Agents, capable of automatically discovering Agent capabilities, generating execution plans, decomposing complex tasks, and providing complete task tracing.

• 自动发现Agent：通过RESTful API自动发现和理解其他Agent的功能

• 智能任务规划：根据用户需求和可用Agent生成最优执行计划

• 任务自动分解：当上下文超过阈值时，自动将复杂任务切割为子任务

• 完整追踪系统：为每个任务和子任务生成唯一的traceID，便于追踪和调试

• 并行执行：支持并行执行多个子任务，提高处理效率

• 结果聚合：自动聚合多个子任务的执行结果

• Automatic Agent Discovery: Automatically discover and understand other Agents' capabilities via RESTful API

• Intelligent Task Planning: Generate optimal execution plans based on user requirements and available Agents

• Automatic Task Decomposition: Automatically split complex tasks into sub-tasks when context exceeds threshold

• Complete Tracing System: Generate unique traceID for each task and sub-task for easy tracking and debugging

• Parallel Execution: Support parallel execution of multiple sub-tasks for improved processing efficiency

• Result Aggregation: Automatically aggregate execution results from multiple sub-tasks

• Python 3.7+
• pip

pip install -r requirements.txt

from agent_orchestrator import AgentCoordinator

# 初始化协调器，指定Agent注册表URL和上下文阈值
# Initialize coordinator, specifying Agent registry URL and context threshold
coordinator = AgentCoordinator(
    registry_url="http://agent-registry:8000/agents",
    context_threshold=1000  # 当任务描述超过1000字符时进行分解 / Split task when description exceeds 1000 characters
)

# 处理简单任务 / Process simple task
result = coordinator.process_task("分析这句话的情感：我今天很高兴！")

# 处理复杂任务（会自动分解） / Process complex task (will be automatically decomposed)
complex_task = """
请对以下文本进行详细分析：

人工智能是计算机科学的一个分支，它致力于创造能够模拟人类智能的机器。
人工智能的应用领域非常广泛，包括自然语言处理、计算机视觉、机器人技术、自动推理等。
近年来，随着深度学习技术的突破，人工智能在各个领域都取得了显著的进展。
...
"""
result = coordinator.process_task(complex_task)

# 获取任务的traceID / Get task traceID
trace_id = result.get("trace_id")
print(f"任务已提交，traceID: {trace_id}")

# 获取任务执行状态 / Get task execution status
status = coordinator.get_task_status(trace_id)
print(f"任务状态: {status.get('summary', {}).get('status')}")
print(f"执行时间: {status.get('summary', {}).get('execution_time'):.2f}秒")

# 获取所有可用的Agent / Get all available Agents
agents = coordinator.get_available_agents()
for agent_url, agent_info in agents.items():
    print(f"- {agent_info.get('name')}: {agent_url}")

python app.py

服务将在 http://localhost:5000 上运行。 The service will run on http://localhost:5000.

• POST /api/task：提交任务 / Submit task
• GET /api/trace/<trace_id>：获取任务执行追踪信息 / Get task execution trace information
• GET /api/agents：获取可用Agent列表 / Get available Agent list
• POST /api/agents/register：手动注册Agent / Manually register Agent
• POST /api/agents/update：更新Agent列表 / Update Agent list
• GET /api/health：健康检查 / Health check

python examples/example_usage.py

这个脚本演示了如何使用Agent协调器处理简单任务和复杂任务，以及如何查询任务状态和可用Agent。 This script demonstrates how to use the Agent Coordinator to process simple and complex tasks, as well as how to query task status and available Agents.

每个Agent应提供标准化的元数据，描述其功能和API： Each Agent should provide standardized metadata describing its functionality and API:

{
  "name": "TextAnalyzerAgent",
  "version": "1.0.0",
  "description": "分析文本内容，提取关键信息",
  "url": "http://text-analyzer-agent:8000",
  "endpoints": [
    {
      "name": "analyze",
      "method": "POST",
      "url": "/api/analyze",
      "description": "分析文本内容",
      "parameters": [
        {
          "name": "text",
          "type": "string",
          "required": true,
          "description": "要分析的文本"
        },
        {
          "name": "options",
          "type": "object",
          "required": false,
          "description": "分析选项"
        }
      ]
    }
  ]
}

• AGENT_REGISTRY_URL：Agent注册表URL / Agent registry URL
• CONTEXT_THRESHOLD：上下文大小阈值 / Context size threshold
• PORT：API服务端口（默认5000） / API service port (default: 5000)
• AGENT_CONFIG_FILE：JSON配置文件路径（默认：config/agents.json） / JSON configuration file path (default: config/agents.json)

Agent协调器支持通过JSON文件预先注册Agent，这样可以在启动时直接加载Agent列表，无需通过注册表发现。

The Agent Coordinator supports pre-registering Agents through a JSON file, which allows loading the Agent list directly at startup without discovering through the registry.

{
  "registered_agents": [
    {
      "url": "http://mock-sentiment-agent:8000",
      "name": "情感分析Agent",
      "version": "1.0.0",
      "description": "分析文本的情感倾向，支持中文和英文",
      "endpoints": [
        {
          "name": "analyze",
          "method": "POST",
          "url": "/api/analyze",
          "description": "分析文本情感",
          "parameters": [
            {
              "name": "text",
              "type": "string",
              "required": true,
              "description": "要分析的文本"
            }
          ]
        }
      ]
    },
    {
      "url": "http://mock-text-agent:8000",
      "name": "文本分析Agent",
      "version": "1.0.0",
      "description": "分析文本内容，提取关键信息、主题和实体"
    }
  ]
}

在初始化Agent协调器时，可以通过config_file参数指定配置文件路径：

You can specify the configuration file path through the config_file parameter when initializing the Agent Coordinator:

from agent_orchestrator import AgentCoordinator

# 初始化协调器，使用配置文件注册Agent
# Initialize coordinator with config file
coordinator = AgentCoordinator(
    registry_url=None,  # 不使用注册表 / Do not use registry
    config_file="config/agents.json"  # 配置文件路径 / Configuration file path
)

也可以通过环境变量指定配置文件路径：

You can also specify the configuration file path through environment variables:

export AGENT_CONFIG_FILE=config/agents.json
python app.py

如果不指定配置文件路径，Agent协调器会尝试从以下位置加载配置文件：

If no configuration file path is specified, the Agent Coordinator will try to load the configuration file from the following locations:

1. 当前工作目录下的 config/agents.json / config/agents.json in the current working directory
2. 项目根目录下的 config/agents.json / config/agents.json in the project root directory

MIT License