OpenClaw多智能体协作框架内核深度解析：从任务分解到仲裁闭环

from pydantic import BaseModel
from typing import List, Dict, Optional
import json
from langchain_openai import ChatOpenAI

class SubTask(BaseModel):
    id: str
    description: str
    agent_type: str  # "code" | "browser" | "db" | "search"
    depends_on: List[str] = []
    params: Dict[str, str] = {}

class TaskPlan(BaseModel):
    tasks: List[SubTask]

class Planner:
    def __init__(self, model="gpt-4o-mini"):
        self.llm = ChatOpenAI(model=model, temperature=0.2)
        self.example = """
目标: "分析某公司财报并生成摘要"
输出:
[
  {"id":"t1","description":"下载财报PDF","agent_type":"browser","depends_on":[],"params":{"url":"..."}},
  {"id":"t2","description":"提取关键财务指标","agent_type":"code","depends_on":["t1"],"params":{}},
  {"id":"t3","description":"生成摘要文本","agent_type":"llm","depends_on":["t2"],"params":{"style":"简洁"}}
]
"""
    def plan(self, goal: str) -> TaskPlan:
        prompt = f"按此格式将目标拆解为子任务（JSON数组）：\n示例：{self.example}\n目标：{goal}"
        resp = self.llm.invoke(prompt)
        raw = resp.content.strip().replace("```json","").replace("```","")
        tasks = json.loads(raw)
        # 校验依赖闭环
        self._validate_acyclic(tasks)
        return TaskPlan(tasks=[SubTask(**t) for t in tasks])
    
    def _validate_acyclic(self, tasks):
        # 拓扑排序检测环
        pass

import subprocess
import tempfile
import os

class CodeAgent:
    def __init__(self):
        self.tools = {
            "python": self._exec_python,
            "file_read": self._read_file,
            "file_write": self._write_file
        }
    
    def _exec_python(self, code: str) -> str:
        # 使用临时文件执行，防止污染
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
            f.write(code)
            path = f.name
        try:
            result = subprocess.run(
                ["python3", path],
                capture_output=True,
                timeout=10,
                text=True
            )
            return result.stdout or result.stderr
        except subprocess.TimeoutExpired:
            return "Error: execution timeout"
        finally:
            os.unlink(path)
    
    def execute(self, task: SubTask) -> str:
        # task.params中指示调用哪个tool及参数
        tool_name = task.params.get("tool", "python")
        tool_input = task.params.get("input", "")
        if tool_name in self.tools:
            return self.tools[tool_name](tool_input)
        return f"Unsupported tool: {tool_name}"

import chromadb
from sentence_transformers import SentenceTransformer

class SharedMemory:
    def __init__(self, path="./openclaw_memory"):
        self.client = chromadb.PersistentClient(path=path)
        self.collection = self.client.get_or_create_collection("task_results")
        self.encoder = SentenceTransformer('all-MiniLM-L6-v2')
    
    def store(self, task_id: str, result: str, metadata: dict = None):
        emb = self.encoder.encode(result).tolist()
        self.collection.add(
            documents=[result],
            embeddings=[emb],
            ids=[task_id],
            metadatas=[metadata or {"task_id": task_id}]
        )
    
    def retrieve(self, query: str, top_k=3) -> List[Dict]:
        q_emb = self.encoder.encode(query).tolist()
        results = self.collection.query(query_embeddings=[q_emb], n_results=top_k)
        return [{"id": ids[i], "content": docs[i]} for i, ids in enumerate(results['ids'][0])]

from concurrent.futures import ThreadPoolExecutor, as_completed

class Orchestrator:
    def __init__(self):
        self.agents = {
            "code": CodeAgent(),
            "browser": BrowserAgent(),  # 省略实现
            "db": DBAgent(),
            "search": SearchAgent()
        }
        self.memory = SharedMemory()
        self.executor = ThreadPoolExecutor(max_workers=4)
    
    def run(self, goal: str) -> str:
        plan = Planner().plan(goal)
        # 先检索记忆缓存
        cached = self.memory.retrieve(goal, top_k=1)
        if cached:
            return f"[Cached] {cached[0]['content']}"
        
        # 构建依赖映射
        task_map = {t.id: t for t in plan.tasks}
        results = {}
        futures = {}
        # 初始：所有无依赖任务提交
        for t in plan.tasks:
            if not t.depends_on:
                futures[self.executor.submit(self._execute_task, t)] = t.id
        
        # 动态调度：完成一个任务后，检查后续可执行任务
        while futures:
            done, _ = wait(futures.keys(), return_when=ALL_COMPLETED)  # 简化处理
            for future in done:
                tid = futures.pop(future)
                results[tid] = future.result()
                self.memory.store(tid, results[tid])
                # 检查是否有任务的所有依赖均已满足
                for t in plan.tasks:
                    if t.id not in results and set(t.depends_on).issubset(results.keys()):
                        futures[self.executor.submit(self._execute_task, t)] = t.id
        
        # 仲裁最终结果（多任务融合）
        return self._arbitrate(results, goal)
    
    def _execute_task(self, task: SubTask):
        agent = self.agents.get(task.agent_type)
        if not agent:
            return f"Error: No agent for type {task.agent_type}"
        return agent.execute(task)
    
    def _arbitrate(self, results: Dict[str, str], goal: str) -> str:
        # 简单拼接，复杂场景可调用LLM综合
        if len(results) == 1:
            return list(results.values())[0]
        # 调用仲裁器（可基于投票或LLM生成）
        # 此处返回合并后的摘要
        combined = "\n".join([f"{tid}: {res}" for tid, res in results.items()])
        prompt = f"任务目标：{goal}\n各子任务结果：\n{combined}\n请输出最终答案："
        return self.llm.invoke(prompt).content

import concurrent.futures

def _execute_task_with_timeout(self, task, timeout=30):
    with concurrent.futures.ThreadPoolExecutor() as pool:
        future = pool.submit(self._execute_task, task)
        try:
            return future.result(timeout=timeout)
        except concurrent.futures.TimeoutError:
            return f"Timeout: task {task.id} exceeded {timeout}s"