update

agents/Dockerfile.echo_agent

@@ -0,0 +1,16 @@
+# Read the doc: https://huggingface.co/docs/hub/spaces-sdks-docker
+# you will also find guides on how best to write your Dockerfile
+
+FROM python:3.12.11
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+COPY --chown=user . /app
+CMD ["uvicorn", "echo_agent:app", "--host", "0.0.0.0", "--port", "8000"]

agents/agent_manager_client.py

@@ -0,0 +1,73 @@
+from typing import Dict, List, Optional
+import httpx
+import os
+
+# 假设 agent_manager 的 models.py 可以在这里被导入，或者我们重新定义需要的模型
+# 为了避免循环依赖，这里直接定义需要的模型，或者从一个共享的包中导入
+from pydantic import BaseModel
+
+class AgentInfo(BaseModel):
+    id: str
+    agent_type: str
+    mcp_endpoint: str
+    status: str
+    created_at: str
+    last_heartbeat: str
+    metadata: Dict = {}
+
+class CreateAgentRequest(BaseModel):
+    agent_type: str
+    image_name: str
+    env_vars: Dict[str, str] = {}
+    resource_limits: Dict = {}
+    config: Dict = {}
+
+class AgentManagerClient:
+    def __init__(self, base_url: str):
+        self.base_url = base_url
+        self.client = httpx.AsyncClient()
+
+    async def create_agent(self, request: CreateAgentRequest) -> AgentInfo:
+        """
+        向 Agent Manager 请求创建一个新的 Agent 实例。
+        """
+        response = await self.client.post(f"{self.base_url}/agents", json=request.dict())
+        response.raise_for_status()
+        return AgentInfo(**response.json())
+
+    async def get_agent_info(self, agent_id: str) -> AgentInfo:
+        """
+        从 Agent Manager 获取指定 Agent 实例的详细信息。
+        """
+        response = await self.client.get(f"{self.base_url}/agents/{agent_id}")
+        response.raise_for_status()
+        return AgentInfo(**response.json())
+
+    async def list_agents(self, agent_type: Optional[str] = None) -> List[AgentInfo]:
+        """
+        从 Agent Manager 列出所有活跃的 Agent 实例。
+        """
+        params = {"agent_type": agent_type} if agent_type else {}
+        response = await self.client.get(f"{self.base_url}/agents", params=params)
+        response.raise_for_status()
+        return [AgentInfo(**agent_data) for agent_data in response.json()]
+
+    async def destroy_agent(self, agent_id: str) -> bool:
+        """
+        向 Agent Manager 请求销毁一个 Agent 实例。
+        """
+        response = await self.client.delete(f"{self.base_url}/agents/{agent_id}")
+        response.raise_for_status()
+        return response.status_code == 204 # No Content
+
+    async def stop_agent(self, agent_id: str) -> bool:
+        """
+        向 Agent Manager 请求停止一个 Agent 实例。
+        """
+        response = await self.client.post(f"{self.base_url}/agents/{agent_id}/stop")
+        response.raise_for_status()
+        return response.status_code == 200 # OK
+
+# 示例用法 (在实际 RootAgent 中使用)
+# AGENT_MANAGER_BASE_URL = os.getenv("AGENT_MANAGER_BASE_URL", "http://localhost:7860")
+# agent_manager_client = AgentManagerClient(base_url=AGENT_MANAGER_BASE_URL)

agents/echo_agent.py

@@ -0,0 +1,134 @@
+from typing import Dict, Any
+from datetime import datetime
+import asyncio
+import json
+import os
+import redis
+import uvicorn
+from fastapi import FastAPI, BackgroundTasks
+
+from .mcp_agent_interface import MCPAgent
+
+# Redis 配置
+REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
+REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
+REDIS_DB = int(os.getenv("REDIS_DB", 0))
+AGENT_CHANNEL = "agent_discovery_channel"
+AGENT_KEY_PREFIX = "agent:"
+HEARTBEAT_INTERVAL = 5 # seconds
+
+class EchoAgent(MCPAgent):
+    def __init__(self, agent_id: str, agent_type: str, config: Dict[str, Any]):
+        super().__init__(agent_id, agent_type, config)
+        self.redis_client: Optional[redis.Redis] = None
+        self.heartbeat_task: Optional[asyncio.Task] = None
+        self.mcp_endpoint = os.getenv("MCP_ENDPOINT", f"http://localhost:8000") # Agent 自身的 MCP 服务地址
+
+    async def initialize(self):
+        """
+        Agent 启动时的初始化逻辑，连接 Redis 并注册。
+        """
+        self.redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
+        try:
+            self.redis_client.ping()
+            print(f"Agent {self.id} connected to Redis successfully!")
+        except redis.exceptions.ConnectionError as e:
+            print(f"Agent {self.id} could not connect to Redis: {e}")
+            # 可以在这里选择退出或以降级模式运行
+
+        await self._register_agent()
+        self.heartbeat_task = asyncio.create_task(self._send_heartbeat())
+        print(f"EchoAgent {self.id} initialized and registered.")
+
+    async def shutdown(self):
+        """
+        Agent 关闭时的清理逻辑，发送下线消息并停止心跳。
+        """
+        if self.heartbeat_task:
+            self.heartbeat_task.cancel()
+            try:
+                await self.heartbeat_task
+            except asyncio.CancelledError:
+                pass
+        await self._unregister_agent()
+        if self.redis_client:
+            self.redis_client.close()
+        print(f"EchoAgent {self.id} shutdown.")
+
+    async def _register_agent(self):
+        """
+        向 Redis 注册 Agent 信息。
+        """
+        agent_info = {
+            "id": self.id,
+            "agent_type": self.agent_type,
+            "mcp_endpoint": self.mcp_endpoint,
+            "status": "running",
+            "created_at": datetime.now().isoformat(),
+            "last_heartbeat": datetime.now().isoformat(),
+            "metadata": json.dumps(self.config) # 将配置作为元数据存储
+        }
+        self.redis_client.hmset(f"{AGENT_KEY_PREFIX}{self.id}", agent_info)
+        self.redis_client.publish(AGENT_CHANNEL, json.dumps({"id": self.id, "event_type": "ONLINE"}))
+        print(f"Agent {self.id} registered to Redis.")
+
+    async def _unregister_agent(self):
+        """
+        向 Redis 发送下线消息。
+        """
+        self.redis_client.publish(AGENT_CHANNEL, json.dumps({"id": self.id, "event_type": "OFFLINE"}))
+        # Agent Manager 会负责从 Redis 中删除键
+        print(f"Agent {self.id} sent OFFLINE message to Redis.")
+
+    async def _send_heartbeat(self):
+        """
+        定期向 Redis 发送心跳。
+        """
+        while True:
+            await asyncio.sleep(HEARTBEAT_INTERVAL)
+            try:
+                self.redis_client.hset(f"{AGENT_KEY_PREFIX}{self.id}", "last_heartbeat", datetime.now().isoformat())
+                self.redis_client.publish(AGENT_CHANNEL, json.dumps({"id": self.id, "event_type": "HEARTBEAT"}))
+                # print(f"Agent {self.id} sent HEARTBEAT.")
+            except Exception as e:
+                print(f"Error sending heartbeat for Agent {self.id}: {e}")
+
+    async def process_subtask(self, subtask_data: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        EchoAgent 的核心逻辑：接收消息并返回。
+        """
+        print(f"EchoAgent {self.id} received subtask: {subtask_data}")
+        message = subtask_data.get("message", "No message provided.")
+        return {"status": "completed", "output": f"Echo: {message}", "agent_id": self.id}
+
+# FastAPI 应用实例
+app = FastAPI()
+echo_agent: Optional[EchoAgent] = None
+
+@app.on_event("startup")
+async def startup_event():
+    global echo_agent
+    agent_id = os.getenv("AGENT_ID", str(uuid.uuid4()))
+    agent_type = os.getenv("AGENT_TYPE", "EchoAgent")
+    config = json.loads(os.getenv("AGENT_CONFIG", "{}"))
+    
+    echo_agent = EchoAgent(agent_id=agent_id, agent_type=agent_type, config=config)
+    await echo_agent.initialize()
+
+@app.on_event("shutdown")
+async def shutdown_event():
+    if echo_agent:
+        await echo_agent.shutdown()
+
+@app.post("/process_subtask")
+async def process_subtask_endpoint(subtask_data: Dict[str, Any]):
+    if not echo_agent:
+        raise HTTPException(status_code=500, detail="Agent not initialized.")
+    return await echo_agent.process_subtask(subtask_data)
+
+@app.get("/health")
+async def health_check():
+    return {"status": "ok", "agent_id": echo_agent.id if echo_agent else "N/A"}
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=8000)

agents/mcp_agent_interface.py

@@ -1,32 +1,39 @@
 from abc import ABC, abstractmethod
-from typing import Any, Dict
+from typing import Dict, Any
 
-class MCPAgentService(ABC):
+class MCPAgent(ABC):
     """
-    MCP Agent 服务的抽象基类。
-    所有具体的 MCP Agent 服务都应继承此接口并实现其抽象方法。
+    抽象基类，定义 MCP Agent 的通用接口。
+    所有具体的 Agent 都应继承此接口并实现其方法。
     """
+    def __init__(self, agent_id: str, agent_type: str, config: Dict[str, Any]):
+        self.id = agent_id
+        self.agent_type = agent_type
+        self.config = config
 
     @abstractmethod
-    async def execute(self, subtask: Dict[str, Any]) -> Dict[str, Any]:
+    async def process_subtask(self, subtask_data: Dict[str, Any]) -> Dict[str, Any]:
         """
-        执行子任务的方法。
+        处理一个子任务。
         
         Args:
-            subtask: 包含子任务描述和所需参数的字典。
+            subtask_data: 包含子任务详细信息的字典。
             
         Returns:
-            包含子任务执行结果的字典。
+            包含子任务处理结果的字典。
         """
         pass
 
     @abstractmethod
-    def get_capabilities(self) -> Dict[str, Any]:
+    async def initialize(self):
         """
-        返回 Agent 服务的能力描述。
-        这将帮助 RootAgent 智能地调度任务。
-        
-        Returns:
-            包含 Agent 能力描述的字典。
+        Agent 启动时的初始化逻辑，例如加载长期记忆、连接外部服务等。
+        """
+        pass
+
+    @abstractmethod
+    async def shutdown(self):
+        """
+        Agent 关闭时的清理逻辑，例如保存短期记忆、断开连接等。
         """
         pass

agents/requirements.txt

@@ -0,0 +1,3 @@
+fastapi
+uvicorn[standard]
+redis

agents/root_agent.py

@@ -1,241 +1,181 @@
-# codes/agents/root_agent.py
+from typing import Dict, Any, List, Optional
 import asyncio
-import json # 导入 json 模块
-import os # 导入 os 模块
-from qwen_agent.agents import Assistant # 导入 Qwen-Agent 的 Assistant 类
-from qwen_agent.tools.base import BaseTool # 如果需要自定义工具，可以导入
+import os
+import json
+import httpx
+import redis
+
+from .agent_manager_client import AgentManagerClient, AgentInfo, CreateAgentRequest
+from .mcp_agent_interface import MCPAgent # 导入 MCP Agent 接口
+
+# Redis 配置
+REDIS_HOST = os.getenv("REDIS_HOST", "localhost")
+REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
+REDIS_DB = int(os.getenv("REDIS_DB", 0))
+AGENT_CHANNEL = "agent_discovery_channel"
+AGENT_KEY_PREFIX = "agent:"
 
 class RootAgent:
     def __init__(self):
+        AGENT_MANAGER_BASE_URL = os.getenv("AGENT_MANAGER_BASE_URL", "http://localhost:7860")
+        self.agent_manager_client = AgentManagerClient(base_url=AGENT_MANAGER_BASE_URL)
+        self.active_agents: Dict[str, AgentInfo] = {} # 内存缓存活跃 Agent
+        self.redis_client: Optional[redis.Redis] = None
+        self.redis_pubsub: Optional[redis.client.PubSub] = None
+        self.discovery_task: Optional[asyncio.Task] = None
+
+    async def initialize(self):
         """
-        RootAgent 的初始化方法。
-        负责设置 Agent 的基本配置，如记忆模块、规划模块等。
+        RootAgent 启动时的初始化逻辑，连接 Redis 并启动 Agent 发现。
         """
-        print("RootAgent 初始化...")
-        # TODO: 初始化记忆模块
-        # TODO: 初始化反思模块
-        # TODO: 初始化 Agent 注册表
-        
-        # 初始化 Qwen-Agent 的 Assistant 作为规划 Agent
-        # 配置 LLM 模型为与 OpenAI 兼容的 Gemini 模型
-        # 从 .env 文件加载环境变量
-        from dotenv import load_dotenv
-        load_dotenv(dotenv_path='.env')
-
-        # 从环境变量中获取 OpenAI 兼容模型的 API 密钥和基础 URL
-        openai_api_key = os.getenv("OPENAI_API_KEY")
-        openai_base_url = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1") # 默认使用 OpenAI 官方 API 地址
-        model = os.getenv("MODEL", "")
-
-        if not openai_api_key:
-            print("警告: 未设置 OPENAI_API_KEY 环境变量。OpenAI 兼容模型可能无法正常工作。")
-
-        self.planner_agent = Assistant(llm={
-            'model': model, # OpenAI 兼容模型
-            'api_key': openai_api_key,
-            'base_url': openai_base_url,
-            'model_type': 'oai' # 更正为 'oai' 以兼容 qwen-agent
-        })
-        print("Qwen-Agent Planner Assistant (OpenAI 兼容) 初始化完成。")
-
-    async def process_task(self, task_description: str) -> dict:
+        self.redis_client = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB)
+        try:
+            self.redis_client.ping()
+            print("RootAgent connected to Redis successfully!")
+        except redis.exceptions.ConnectionError as e:
+            print(f"RootAgent could not connect to Redis: {e}")
+            # 可以在这里选择退出或以降级模式运行
+
+        self.redis_pubsub = self.redis_client.pubsub()
+        self.redis_pubsub.subscribe(AGENT_CHANNEL)
+        self.discovery_task = asyncio.create_task(self._agent_discovery_listener())
+
+        # 从 Redis 加载所有已知的 Agent
+        await self._load_existing_agents()
+        print("RootAgent initialized.")
+
+    async def shutdown(self):
         """
-        处理高层级任务的入口方法。
-        
-        Args:
-            task_description: 任务的描述字符串。
-            
-        Returns:
-            包含任务处理结果的字典。
+        RootAgent 关闭时的清理逻辑。
         """
-        print(f"RootAgent 接收到任务: {task_description}")
-        
-        # 1. 任务接收与拆分 (规划)
-        subtasks = await self._plan_task(task_description)
-        print(f"任务已拆分为子任务: {subtasks}")
-        
-        # 2. 任务调度与编排
-        # 简单的依赖管理：按顺序执行，假设依赖关系已在 _plan_task 中处理为线性顺序
-        # 实际中需要更复杂的 DAG 调度器
-        task_results = {}
-        for subtask in subtasks:
-            # 检查并等待依赖任务完成
-            for dep_id in subtask.get("dependencies", []):
-                if dep_id not in task_results or task_results[dep_id]["status"] != "completed":
-                    # 实际中这里需要更复杂的等待机制或错误处理
-                    print(f"等待子任务 {dep_id} 完成才能执行 {subtask['id']}")
-                    # 暂时跳过，实际应阻塞或重新调度
-                    task_results[subtask["id"]] = {"subtask": subtask, "status": "skipped", "reason": f"Dependency {dep_id} not completed"}
-                    break
-            else:
-                print(f"开始调度子任务: {subtask['name']} (ID: {subtask['id']})")
-                agent_service = await self._dispatch_subtask(subtask)
-                if agent_service:
-                    # 假设 agent_service.execute 返回一个包含结果和状态的字典
-                    subtask_result = await agent_service.execute(subtask) 
-                    task_results[subtask["id"]] = subtask_result
-                else:
-                    task_results[subtask["id"]] = {"subtask": subtask, "status": "failed", "reason": "No suitable agent found"}
-            
-            print(f"子任务 {subtask['id']} 结果: {task_results.get(subtask['id'], 'N/A')}")
-
-        # 3. 结果聚合
-        results_list = list(task_results.values())
-        final_result = await self._aggregate_results(results_list)
-        
-        # 4. 反思与学习 (可选)
-        await self._reflect_on_execution(task_description, subtasks, results_list, final_result)
-        
-        return final_result
-
-    async def _plan_task(self, task_description: str) -> list[dict]:
+        if self.discovery_task:
+            self.discovery_task.cancel()
+            try:
+                await self.discovery_task
+            except asyncio.CancelledError:
+                pass
+        if self.redis_pubsub:
+            self.redis_pubsub.unsubscribe(AGENT_CHANNEL)
+        if self.redis_client:
+            self.redis_client.close()
+        print("RootAgent shutdown.")
+
+    async def _agent_discovery_listener(self):
         """
-        根据任务描述进行规划，拆分为子任务，并建立依赖关系。
-        
-        Args:
-            task_description: 任务的描述字符串。
-            
-        Returns:
-            一个包含子任务及其依赖关系的列表。每个子任务是一个字典，
-            包含 'id', 'name', 'description', 'dependencies' 等字段。
+        监听 Redis 频道，实时更新活跃 Agent 列表。
         """
-        print(f"RootAgent 正在使用 Qwen-Agent 规划任务: {task_description}")
-        
-        # 构造 Qwen-Agent 的输入提示
-        # Qwen-Agent 的 run 方法通常期望一个消息列表作为输入
-        messages = [
-            {
-                'role': 'user',
-                'content': f"""请将以下高层级任务分解为一系列可执行的子任务，并以 JSON 数组的格式返回。
-每个子任务对象应包含以下字段：
-- 'id': 唯一的子任务ID (字符串)
-- 'name': 子任务的名称 (字符串)
-- 'description': 子任务的详细描述 (字符串)
-- 'dependencies': 该子任务依赖的其他子任务ID列表 (字符串数组，如果无依赖则为空数组)
-- 'required_agent_type': 执行该子任务所需的 Agent 类型 (字符串，例如：DataCollectorAgent, DataAnalyzerAgent, ReportGeneratorAgent)
-
-任务描述：{task_description}
-
-请确保输出是有效的 JSON 格式，并且只包含 JSON 数组，不包含任何额外的文本或解释。
-"""
-            }
-        ]
-        
-        # 调用 Qwen-Agent 进行规划
-        # Qwen-Agent 的 run 方法通常返回一个包含 Agent 思考过程和最终结果的字典
-        # 我们需要从结果中提取规划的子任务
-        # 封装同步生成器迭代，并使用 asyncio.to_thread 包装
-        def run_sync_planner():
-            last_item = None
-            # 迭代生成器，获取最终结果
-            # 将 messages 传递给 run 方法
-            for item in self.planner_agent.run(messages):
-                last_item = item
-            return last_item
-
-        response = await asyncio.to_thread(run_sync_planner)
-        
-        subtasks = []
-        try:
-            # 检查 response 是否为列表，并且包含至少一个字典
-            if isinstance(response, list) and len(response) > 0 and isinstance(response[0], dict):
-                # 假设实际的 JSON 字符串在第一个字典的 'content' 键中
-                content_str = response[0].get('content', '')
-                
-                # 提取 Markdown 代码块中的 JSON 字符串
-                # 查找 '```json\n' 和 '\n```' 之间的内容
-                json_start = content_str.find('```json\n')
-                json_end = content_str.rfind('\n```')
-                
-                if json_start != -1 and json_end != -1 and json_start < json_end:
-                    json_str = content_str[json_start + len('```json\n'):json_end]
-                    parsed_output = json.loads(json_str)
-                    if isinstance(parsed_output, list):
-                        subtasks = parsed_output
-                    else:
-                        print(f"Qwen-Agent 提取的 JSON 不是列表格式: {parsed_output}")
-                else:
-                    print(f"Qwen-Agent 返回的 content 字段不包含预期的 JSON 代码块: {content_str}")
-            else:
-                print(f"Qwen-Agent 返回非预期格式的响应: {response}")
-        except json.JSONDecodeError as e:
-            print(f"解析 Qwen-Agent 规划结果时发生 JSON ��误: {e}")
-            print(f"Qwen-Agent 原始响应: {response}")
-        except Exception as e:
-            print(f"解析 Qwen-Agent 规划结果时发生未知错误: {e}")
-            print(f"Qwen-Agent 原始响应: {response}")
-            
-        # 如果 Qwen-Agent 未能生成有效规划，返回一个默认的模拟规划
-        if not subtasks:
-            print("Qwen-Agent 未能生成有效规划，返回模拟子任务列表。")
-            subtasks = [
-                {
-                    "id": "task_001",
-                    "name": "市场数据收集",
-                    "description": f"收集关于 {task_description} 的最新市场数据",
-                    "dependencies": [],
-                    "required_agent_type": "DataCollectorAgent"
-                },
-                {
-                    "id": "task_002",
-                    "name": "数据分析",
-                    "description": "对收集到的市场数据进行分析，识别关键趋势",
-                    "dependencies": ["task_001"],
-                    "required_agent_type": "DataAnalyzerAgent"
-                },
-                {
-                    "id": "task_003",
-                    "name": "报告生成",
-                    "description": "根据数据分析结果生成一份简要报告",
-                    "dependencies": ["task_002"],
-                    "required_agent_type": "ReportGeneratorAgent"
-                }
-            ]
-        
-        return subtasks
-
-    async def _dispatch_subtask(self, subtask: dict):
+        if not self.redis_pubsub:
+            return
+
+        while True:
+            message = self.redis_pubsub.get_message(ignore_subscribe_messages=True)
+            if message:
+                try:
+                    data = json.loads(message['data'].decode('utf-8'))
+                    agent_id = data.get("id")
+                    event_type = data.get("event_type")
+
+                    if agent_id:
+                        if event_type == "HEARTBEAT" or event_type == "ONLINE":
+                            # 从 Redis 获取最新的 Agent 信息
+                            agent_data = self.redis_client.hgetall(f"{AGENT_KEY_PREFIX}{agent_id}")
+                            if agent_data:
+                                agent_info = AgentInfo(**{k.decode('utf-8'): v.decode('utf-8') for k, v in agent_data.items()})
+                                self.active_agents[agent_id] = agent_info
+                                # print(f"RootAgent discovered/updated agent: {agent_id} ({agent_info.agent_type})")
+                            else:
+                                print(f"RootAgent received {event_type} for {agent_id}, but no data in Redis.")
+                        elif event_type == "OFFLINE":
+                            if agent_id in self.active_agents:
+                                del self.active_agents[agent_id]
+                                print(f"RootAgent removed offline agent: {agent_id}")
+                except json.JSONDecodeError:
+                    print(f"RootAgent received invalid JSON message from Redis: {message['data']}")
+                except Exception as e:
+                    print(f"RootAgent error processing Redis message: {e}")
+            await asyncio.sleep(0.1)
+
+    async def _load_existing_agents(self):
         """
-        调度子任务到合适的 MCP Agent 服务。
+        从 Redis 加载所有已知的 Agent 到内存缓存。
         """
-        # 这是一个占位符实现，未来将从注册表选择 Agent 或请求 Agent Manager 创建
-        print(f"调度子任务: {subtask['name']}")
-        # TODO: 实现 Agent 发现和调度逻辑
-        # TODO: 如果没有合适的 Agent，请求 Agent Manager 创建
-        
-        # 模拟一个 Agent 服务调用
-        class MockAgentService:
-            async def execute(self, subtask_data: dict) -> dict:
-                print(f"MockAgentService 正在执行 {subtask_data['name']}")
-                await asyncio.sleep(1) # 模拟耗时操作
-                return {"subtask": subtask_data, "status": "completed", "output": f"已完成 {subtask_data['name']}"}
-        
-        # 暂时返回一个 Mock Agent 服务实例
-        return MockAgentService()
-
-    async def _aggregate_results(self, results: list) -> dict:
+        for key in self.redis_client.keys(f"{AGENT_KEY_PREFIX}*"):
+            agent_data = self.redis_client.hgetall(key)
+            if agent_data:
+                agent_info = AgentInfo(**{k.decode('utf-8'): v.decode('utf-8') for k, v in agent_data.items()})
+                self.active_agents[agent_info.id] = agent_info
+                print(f"RootAgent loaded existing agent: {agent_info.id} ({agent_info.agent_type})")
+
+    async def _get_or_create_agent(self, agent_type: str, image_name: str = "echo-agent:latest") -> AgentInfo:
         """
-        聚合所有子任务的结果。
+        尝试从缓存中获取一个可用 Agent，如果不存在则请求 Agent Manager 创建。
         """
-        print(f"聚合子任务结果: {results}")
-        # 这是一个占位符实现
-        return {"status": "completed", "final_output": "所有子任务已处理", "details": results}
+        # 1. 检查内存缓存中是否有可用 Agent
+        for agent_id, agent_info in self.active_agents.items():
+            if agent_info.agent_type == agent_type and agent_info.status == "running":
+                print(f"Found existing agent {agent_id} of type {agent_type}.")
+                return agent_info
+
+        # 2. 如果没有，请求 Agent Manager 创建新 Agent
+        print(f"No active agent of type {agent_type} found. Requesting Agent Manager to create one.")
+        create_request = CreateAgentRequest(
+            agent_type=agent_type,
+            image_name=image_name,
+            env_vars={"REDIS_HOST": REDIS_HOST, "REDIS_PORT": str(REDIS_PORT), "REDIS_DB": str(REDIS_DB)},
+            config={"some_initial_config": "value"}
+        )
+        try:
+            new_agent_info = await self.agent_manager_client.create_agent(create_request)
+            self.active_agents[new_agent_info.id] = new_agent_info # 添加到缓存
+            print(f"Agent Manager created new agent: {new_agent_info.id} ({new_agent_info.agent_type})")
+            return new_agent_info
+        except httpx.HTTPStatusError as e:
+            print(f"Failed to create agent via Agent Manager: {e.response.text}")
+            raise RuntimeError(f"Failed to create agent {agent_type}: {e}")
+        except Exception as e:
+            print(f"An unexpected error occurred while creating agent: {e}")
+            raise RuntimeError(f"Failed to create agent {agent_type}: {e}")
 
-    async def _reflect_on_execution(self, original_task: str, subtasks: list, results: list, final_result: dict):
+    async def process_task(self, task_data: Dict[str, Any]) -> Dict[str, Any]:
         """
-        对任务执行过程进行反思和学习。
+        RootAgent 的核心任务处理逻辑：任务拆分、调度、结果聚合。
         """
-        print("RootAgent 正在反思执行过程...")
-        # TODO: 实现反思逻辑，例如评估成功率、识别瓶颈、更新长期记忆
-        pass
-
-# 示例用法 (仅用于测试 RootAgent 模块本身，实际由 codes/app.py 调用)
-if __name__ == "__main__":
-    import asyncio
-
-    async def main():
-        root_agent = RootAgent()
-        result = await root_agent.process_task("分析市场数据并生成报告")
-        print(f"\n最终任务结果: {result}")
-
-    asyncio.run(main())
+        print(f"RootAgent received task: {task_data}")
+
+        # 示例：简单任务拆分和调度
+        # 假设 task_data 包含 "subtasks" 列表，每个子任务有 "agent_type" 和 "payload"
+        subtasks_definitions = task_data.get("subtasks", [])
+        if not subtasks_definitions:
+            return {"status": "failed", "error": "No subtasks defined in the task."}
+
+        results = []
+        for subtask_def in subtasks_definitions:
+            agent_type = subtask_def.get("agent_type")
+            payload = subtask_def.get("payload")
+
+            if not agent_type or not payload:
+                results.append({"status": "failed", "error": "Invalid subtask definition."})
+                continue
+
+            try:
+                # 获取或创建 Agent
+                agent_info = await self._get_or_create_agent(agent_type)
+                
+                # 调用 Agent 的 MCP 服务
+                async with httpx.AsyncClient() as client:
+                    response = await client.post(
+                        f"{agent_info.mcp_endpoint}/process_subtask",
+                        json=payload,
+                        timeout=30.0 # 设定超时
+                    )
+                    response.raise_for_status()
+                    subtask_result = response.json()
+                    results.append({"agent_type": agent_type, "result": subtask_result})
+            except Exception as e:
+                results.append({"agent_type": agent_type, "status": "failed", "error": str(e)})
+                print(f"Error processing subtask with agent {agent_type}: {e}")
+
+        # 聚合结果
+        final_output = {"status": "completed", "aggregated_results": results}
+        print(f"RootAgent finished processing task. Results: {final_output}")
+        return final_output

app.py

@@ -1,28 +1,38 @@
-import sys
+from fastapi import FastAPI, HTTPException, status
+from typing import Dict, Any
+import asyncio
 import os
 
-# 将当前文件所在的目录（即 root_agent 目录）添加到 Python 路径
-sys.path.append(os.path.dirname(__file__))
-
-from fastapi import FastAPI
-from agents.root_agent import RootAgent
+from root_agent.agents.root_agent import RootAgent
 
 app = FastAPI()
-root_agent = RootAgent()
+root_agent_instance: Optional[RootAgent] = None
 
-@app.get("/")
-async def read_root():
-    return {"Hello": "World!"}
+@app.on_event("startup")
+async def startup_event():
+    global root_agent_instance
+    root_agent_instance = RootAgent()
+    await root_agent_instance.initialize()
 
-from pydantic import BaseModel
+@app.on_event("shutdown")
+async def shutdown_event():
+    if root_agent_instance:
+        await root_agent_instance.shutdown()
 
-class TaskRequest(BaseModel):
-    task_description: str
+@app.get("/")
+async def read_root():
+    return {"message": "Worker service with RootAgent is running!"}
 
 @app.post("/process_task")
-async def process_task_endpoint(request: TaskRequest):
+async def process_task_endpoint(task_data: Dict[str, Any]):
     """
-    接收并处理高层级任务的 API 端点。
+    接收外部任务请求，并交由 RootAgent 处理。
     """
-    result = await root_agent.process_task(request.task_description)
-    return result
+    if not root_agent_instance:
+        raise HTTPException(status_code=500, detail="RootAgent not initialized.")
+    
+    try:
+        result = await root_agent_instance.process_task(task_data)
+        return result
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error processing task: {e}")