OpenClaw 智能体深度解析：构建多智能体协作系统的完整实现

from pydantic import BaseModel
from typing import List, Dict
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] = []  # 依赖的子任务 ID
    params: Dict

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

class Planner:
    def __init__(self, llm_model="gpt-4o-mini"):
        self.llm = ChatOpenAI(model=llm_model, temperature=0.2)
    
    def plan(self, goal: str) -> Plan:
        prompt = f"""
        将以下目标拆解为子任务，每个子任务指定执行类型（code/browser/db/search）和依赖关系。
        输出 JSON 数组，每个元素含 id, description, agent_type, depends_on, params。
        目标：{goal}
        """
        response = self.llm.invoke(prompt)
        data = json.loads(response.content)
        # 验证依赖无环（拓扑排序）
        tasks = [SubTask(**item) for item in data]
        self._validate_dag(tasks)
        return Plan(tasks=tasks)
    
    def _validate_dag(self, tasks):
        # 简单环检测
        pass

from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate

class CodeExecutorAgent:
    def __init__(self, llm):
        tools = [
            Tool(name="python_shell", func=self._run_python, description="执行 Python 代码"),
            Tool(name="file_io", func=self._read_write_file, description="读写文件")
        ]
        prompt = PromptTemplate.from_template("你是一个 Python 专家，仅输出可执行代码。任务：{input}")
        self.agent = create_react_agent(llm, tools, prompt)
        self.executor = AgentExecutor(agent=self.agent, tools=tools, verbose=False)
    
    def _run_python(self, code: str) -> str:
        # 沙箱执行（隔离环境）
        try:
            exec_globals = {}
            exec(code, exec_globals)
            return str(exec_globals.get("result", "执行成功"))
        except Exception as e:
            return f"错误: {str(e)}"
    
    def _read_write_file(self, command: str):
        # 简易文件操作
        pass
    
    def execute(self, task: SubTask) -> str:
        return self.executor.invoke({"input": task.description + f" 参数: {task.params}"})["output"]

import chromadb
from sentence_transformers import SentenceTransformer

class SharedMemory:
    def __init__(self, path="./claw_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):
        emb = self.encoder.encode(result).tolist()
        self.collection.add(
            documents=[result],
            embeddings=[emb],
            ids=[task_id],
            metadatas=[{"task_id": task_id}]
        )
    
    def retrieve(self, query: str, top_k=3) -> List[str]:
        q_emb = self.encoder.encode(query).tolist()
        results = self.collection.query(query_embeddings=[q_emb], n_results=top_k)
        return results['documents'][0]

class Arbiter:
    def __init__(self, llm):
        self.llm = llm
    
    def arbitrate(self, results: Dict[str, str], goal: str) -> str:
        # 若只有一个结果，直接返回
        if len(results) == 1:
            return list(results.values())[0]
        # 对于数值型结果，取加权平均（权重由 Agent 置信度决定）
        # 对于文本型，交由 LLM 融合
        prompt = f"""
        目标：{goal}
        以下是从不同 Agent 获得的结果，请整合成一份完整、一致的答案：
        {json.dumps(results, indent=2)}
        输出最终答案：
        """
        return self.llm.invoke(prompt).content

from concurrent.futures import ThreadPoolExecutor, as_completed

class OpenClawController:
    def __init__(self):
        self.planner = Planner()
        self.memory = SharedMemory()
        self.agents = {
            "code": CodeExecutorAgent(llm),
            "browser": BrowserAgent(llm),
            # 其他
        }
        self.arbiter = Arbiter(llm)
        self.executor_pool = ThreadPoolExecutor(max_workers=4)
    
    def run(self, goal: str) -> str:
        plan = self.planner.plan(goal)
        # 先检索记忆中是否有相似完成方案
        cached = self.memory.retrieve(goal, top_k=1)
        if cached:
            return f"[缓存结果] {cached[0]}"
        
        # 按拓扑顺序执行（此处简化，实际需处理依赖）
        task_map = {t.id: t for t in plan.tasks}
        results = {}
        futures = []
        # 并行执行无依赖任务
        for task in plan.tasks:
            if not task.depends_on:
                future = self.executor_pool.submit(self._execute_task, task)
                futures.append((task.id, future))
        for task_id, future in as_completed(futures):
            results[task_id] = future.result()
            self.memory.store(task_id, results[task_id])
        # 汇总所有结果
        final = self.arbiter.arbitrate(results, goal)
        self.memory.store("final_" + goal, final)
        return final
    
    def _execute_task(self, task: SubTask) -> str:
        agent = self.agents.get(task.agent_type)
        if not agent:
            return f"错误：无 {task.agent_type} 类型 Agent"
        return agent.execute(task)