OpenClaw 架构深度解析：Agent Loop 执行循环

┌─────────────────────────────────────────┐
│              Agent Loop                  │
│                                          │
│   ┌──────┐    ┌──────┐    ┌──────┐      │
│   │ 感知  │───▶│ 决策  │───▶│ 行动  │     │
│   │Perceive│   │Decide│   │ Act  │      │
│   └──┬───┘    └──────┘    └──┬───┘      │
│      │                        │          │
│      │    ┌──────┐            │          │
│      └────│ 记忆  │◀───────────┘          │
│           │Memory│                        │
│           └──────┘                        │
└─────────────────────────────────────────┘

interface Perception {
  source: "user" | "cron" | "system" | "agent";
  channel: string;          // "telegram", "webchat", "cron" ...
  rawContent: string;       // 原始内容
  metadata: {
    timestamp: number;
    priority: "low" | "normal" | "high" | "urgent";
    context?: Context;      // 上下文继承
  };
}

# OpenClaw 决策引擎伪代码
def decide(perception, memory):
    # 1. 加载上下文
    context = memory.retrieve(perception)
    
    # 2. 意图分类
    intent = llm.classify(
        perception.raw_content,
        categories=["task", "question", "command", "chat", "emergency"]
    )
    
    # 3. 工具选择（ReAct 模式）
    if intent == "task":
        available_tools = skills.get_available_tools()
        plan = llm.plan(
            perception=perception,
            context=context,
            tools=available_tools
        )
        # plan = { steps: ["search", "summarize", "send"], tools: ["web_search", "tts"] }
    
    # 4. 安全检查
    safety_check(plan)  # 拦截危险操作
    
    return plan

Thought: 用户想知道今天A股涨幅最大的板块，我需要查询实时行情
Action: call stock_api(market="A", query="top_sector_by_gain")
Observation: { sector: "半导体", gain: "3.2%", top_stock: "中芯国际" }
Thought: 拿到数据了，现在需要整理成用户友好的格式
Action: format_response(data, style="table")
Observation: "今日A股涨幅最大板块：半导体 +3.2%，领涨股：中芯国际"
Thought: 回复完成
Action: respond("今日A股涨幅最大板块：半导体 +3.2%...")

# 路由策略
routing:
  - condition: "task.type == 'coding'"
    model: "claude-3.5-sonnet"
    reason: "代码任务 Claude 更强"
    
  - condition: "task.lang == 'zh' and task.complexity == 'low'"
    model: "glm-4-flash"
    reason: "中文简单任务用 GLM 降成本"
    
  - condition: "task.type == 'vision'"
    model: "gpt-4o"
    reason: "多模态任务"
    
  - condition: "budget.remaining < 1.0"
    model: "qwen-turbo"
    reason: "预算不足时降级"

interface ToolCall {
  tool: string;          // 工具名称，如 "web_search"
  parameters: object;    // 调用参数
  timeout: number;       // 超时时间（ms）
  retry: {
    max: number;         // 最大重试次数
    backoff: "linear" | "exponential";
  };
  sandbox: boolean;      // 是否在沙箱中执行
  requireApproval: boolean; // 是否需要用户审批
}

class ToolExecutor:
    def execute(self, tool_call: ToolCall) -> ToolResult:
        # 1. 权限检查
        if tool_call.requireApproval:
            approved = self.ask_user(tool_call)
            if not approved:
                return ToolResult(success=False, error="用户拒绝")
        
        # 2. 沙箱隔离
        if tool_call.sandbox:
            with Sandbox() as sb:
                result = sb.run(tool_call.tool, tool_call.parameters)
        else:
            result = self.tools[tool_call.tool](**tool_call.parameters)
        
        # 3. 结果验证
        if not result.success and tool_call.retry:
            result = self.retry_with_backoff(tool_call)
        
        # 4. 审计日志
        self.audit_log(tool_call, result)
        
        return result

// 用户："帮我查北京和上海明天的天气"
const plan = {
  steps: [
    { tool: "weather_api", params: { city: "北京" } },
    { tool: "weather_api", params: { city: "上海" } }
  ],
  parallel: true  // 两个查询同时执行
};

const results = await executor.executeParallel(plan.steps);
// 耗时 = max(北京查询, 上海查询) 而非两者之和

┌──────────────────┐
│   上下文窗口      │  ← 当前对话（短期记忆）
│   (Context Window)│
├──────────────────┤
│   会话缓存        │  ← 最近 N 轮对话摘要
│   (Session Cache) │
├──────────────────┤
│   本地记忆        │  ← MEMORY.md + daily notes
│   (Local Memory)  │
├──────────────────┤
│   向量索引        │  ← 语义检索（可选）
│   (Vector Store)  │
└──────────────────┘

class AgentLoop:
    def run(self, perception):
        steps = 0
        while steps < self.max_steps:
            context = self.memory.retrieve(perception)
            plan = self.decide(perception, context)
            
            if plan.is_complete():
                return plan.result
            
            result = self.act(plan)
            self.memory.update(result)
            
            perception = self.perceive_next(result)
            steps += 1
            
            if self.circuit_breaker.tripped():
                break
        
        return Result(status="incomplete", reason="达到步数上限")

L1: 聊天机器人     → "你说我答"
L2: 任务执行者     → "你定目标，我执行"  ← OpenClaw 当前
L3: 自我管理者     → "你定方向，我规划+执行"  ← 2026 路线图
L4: 自主演化者     → "我发现问题，我解决"  ← 未来