乐于分享
好东西不私藏

OpenClaw Agent循环机制深度解析:从’聊天’到’干活’的底层实现

OpenClaw Agent循环机制深度解析:从’聊天’到’干活’的底层实现

OpenClaw Agent循环机制深度解析:从”聊天”到”干活”的底层实现

本文面向有技术背景的读者,深入剖析 OpenClaw 的 Agent 循环机制,揭示其如何实现从”建议”到”执行”的跨越。

一、现象:为什么 OpenClaw 能”动手干活”?

2026年,OpenClaw 以破竹之势席卷科技圈,GitHub 收藏量突破 10 万,单周访问量超 200 万。与 ChatGPT、Claude 等”只会聊天”的 AI 不同,OpenClaw 最大的突破在于:它把 AI 从云端拉回本地,并赋予了它操作系统级别的权限

当你告诉 ChatGPT “帮我检查数据库”,它会给你一段 SQL 代码;而当你告诉 OpenClaw 同样的指令,它会自己连上数据库、执行检查 SQL、分析结果,发现问题时每天早上 9 点自动巡检并发微信提醒。

这种”从建议到执行”的跨越,核心在于 Agent 循环机制(Agent Loop)。本文将深入剖析这一机制的底层实现。


二、底层原理拆解:Agent 循环的架构设计

2.1 核心组件拓扑

OpenClaw 的架构由三个核心组件构成:

┌─────────────────────────────────────────────────────────┐
│                    Gateway(网关层)                      │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐      │
│  │ 渠道适配器   │  │ 模型路由    │  │ 技能调度    │      │
│  │ (Telegram/  │  │ (多模型聚合) │  │ (Skill系统) │      │
│  │  Discord/   │  │             │  │             │      │
│  │  WeChat)    │  │             │  │             │      │
│  └─────────────┘  └─────────────┘  └─────────────┘      │
└─────────────────────────────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────┐
│                    Agent(智能体层)                      │
│  ┌─────────────────────────────────────────────────┐    │
│  │              Agent Loop(核心循环)               │    │
│  │  1. 接收消息 → 2. 构建上下文 → 3. 调用模型       │    │
│  │  4. 解析响应 → 5. 执行工具 → 6. 循环/终止        │    │
│  └─────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────┘
                          │
                          ▼
┌─────────────────────────────────────────────────────────┐
│                   Skills(技能层)                       │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐   │
│  │文件操作   │ │浏览器控制│ │消息收发   │ │自定义技能│   │
│  └──────────┘ └──────────┘ └──────────┘ └──────────┘   │
└─────────────────────────────────────────────────────────┘

2.2 Agent Loop 的六大阶段

Agent 循环是一个迭代过程,每个循环包含六个阶段:

阶段 1:消息接收(Message Ingestion)

typescript// 简化的消息接收逻辑
interfaceInboundMessage {
channelId:string;      // 来源渠道 ID
userId:string;         // 用户 ID
content:string;        // 消息内容
attachments?:File[];   // 附件
timestamp:number;      // 时间戳
}

asyncfunctioningestMessage(raw:RawMessage): Promise<InboundMessage> {
// 1. 渠道适配器解析原始消息
constparsed=awaitchannelAdapter.parse(raw);

// 2. 会话管理器关联会话上下文
constsession=awaitsessionManager.getOrCreate(parsed.channelId, parsed.userId);

// 3. 存入消息队列,等待处理
awaitmessageQueue.enqueue({ ...parsed, sessionId:session.id });

returnparsed;
}

阶段 2:上下文构建(Context Assembly)

这是 Agent 循环中最关键的阶段之一。OpenClaw 需要从多个来源组装上下文:

typescriptinterfaceAgentContext {
systemPrompt:string;      // 系统提示词
conversationHistory:Message[];  // 对话历史
skillInstructions:string[];     // 技能指令
memoryContext:string;     // LCM 记忆上下文
workspaceContext:string;  // 工作区状态
toolDefinitions:Tool[];   // 可用工具定义
}

asyncfunctionassembleContext(session:Session): Promise<AgentContext> {
// 1. 加载系统提示词(包含 SOUL.md、IDENTITY.md 等)
constsystemPrompt=awaitloadSystemPrompt(session);

// 2. 加载对话历史(可能有 LCM 压缩)
consthistory=awaitloadConversationHistory(session);

// 3. 加载技能指令(根据当前会话绑定的技能)
constskills=awaitskillRegistry.getInstructions(session.skillIds);

// 4. LCM 记忆检索(语义搜索相关记忆)
constmemory=awaitlcm.search(session.lastQuery);

// 5. 工作区上下文(文件状态、Git 状态等)
constworkspace=awaitworkspaceManager.getStatus();

return { systemPrompt, conversationHistory:history, skillInstructions:skills, memoryContext:memory, workspaceContext:workspace, toolDefinitions:awaitgetToolDefinitions(skills) };
}

阶段 3:模型调用(Model Invocation)

OpenClaw 支持多模型路由,可以根据任务类型、成本、性能等因素动态选择模型:

typescriptinterfaceModelRouter {
selectModel(task:Task): Promise<ModelConfig>;
execute(model:ModelConfig, context:AgentContext): Promise<ModelResponse>;
}

// 模型选择策略
constROUTING_RULES= {
// 简单问答 → 快速模型
'simple_qa': { model:'deepseek-v4-flash', maxTokens:1000 },

// 代码生成 → 代码模型
'code_gen': { model:'claude-3.7-sonnet', maxTokens:4000 },

// 复杂推理 → 高级模型
'complex_reasoning': { model:'gpt-4-turbo', maxTokens:8000 },

// 工具调用 → 工具优化模型
'tool_call': { model:'claude-3.7-sonnet', maxTokens:4000 }
};

asyncfunctioninvokeModel(context:AgentContext): Promise<ModelResponse> {
// 1. 分析任务类型
consttaskType=classifyTask(context.conversationHistory);

// 2. 选择模型
constmodelConfig=ROUTING_RULES[taskType];

// 3. 构建请求(支持思考模式)
constrequest= {
model:modelConfig.model,
messages:buildMessages(context),
tools:context.toolDefinitions,
thinking:session.thinkingMode?'enabled':'disabled'
  };

// 4. 调用模型 API(通过 Gateway 中转)
returnawaitgateway.request(request);
}

阶段 4:响应解析(Response Parsing)

模型返回的响应可能包含多种类型:文本、工具调用、思考内容等:

typescriptinterfaceParsedResponse {
textContent?:string;       // 文本内容
toolCalls?:ToolCall[];     // 工具调用请求
thinkingContent?:string;   // 思考过程(可选)
finishReason:'stop'|'tool_use'|'max_tokens';
}

functionparseResponse(raw:ModelResponse):ParsedResponse {
constresult:ParsedResponse= { finishReason:raw.stop_reason };

// 1. 提取思考内容(如果有)
if (raw.thinking) {
result.thinkingContent=raw.thinking;
  }

// 2. 提取文本内容
for (constblockofraw.content) {
if (block.type==='text') {
result.textContent=block.text;
    } elseif (block.type==='tool_use') {
result.toolCalls=result.toolCalls|| [];
result.toolCalls.push({
id:block.id,
name:block.name,
arguments:block.input
      });
    }
  }

returnresult;
}

阶段 5:工具执行(Tool Execution)

这是 Agent “动手干活”的核心。OpenClaw 的工具系统基于 Skills 架构:

typescriptinterfaceToolExecutor {
execute(call:ToolCall): Promise<ToolResult>;
}

asyncfunctionexecuteTools(calls:ToolCall[]): Promise<ToolResult[]> {
constresults:ToolResult[] = [];

for (constcallofcalls) {
// 1. 安全检查(沙箱隔离)
constpermission=awaitsandbox.checkPermission(call);
if (!permission.allowed) {
results.push({ error:`Permission denied: ${permission.reason}` });
continue;
    }

// 2. 查找技能处理器
constskill=skillRegistry.find(call.name);
if (!skill) {
results.push({ error:`Unknown tool: ${call.name}` });
continue;
    }

// 3. 执行工具(在沙箱环境中)
try {
constresult=awaitsandbox.execute(skill, call.arguments);
results.push({ output:result });
    } catch (error) {
results.push({ error:error.message });
    }
  }

returnresults;
}

阶段 6:循环决策(Loop Decision)

根据执行结果决定是继续循环还是终止:

typescriptasyncfunctiondecideNextStep(
response:ParsedResponse,
toolResults:ToolResult[]
): Promise<'continue'|'stop'> {
// 1. 如果没有工具调用,直接终止
if (!response.toolCalls?.length) {
return'stop';
  }

// 2. 检查是否需要继续(工具返回了需要处理的结果)
constneedsContinue=toolResults.some(r => r.needsFollowUp);
if (needsContinue) {
return'continue';
  }

// 3. 检查循环次数限制(防止无限循环)
if (session.loopCount>=MAX_LOOPS) {
return'stop';
  }

// 4. 默认继续(让模型决定下一步)
return'continue';
}

2.3 完整的 Agent 循环伪代码

typescriptasyncfunctionagentLoop(session:Session): Promise<void> {
letshouldContinue=true;
letloopCount=0;

while (shouldContinue&&loopCount<MAX_LOOPS) {
// 阶段 1: 消息接收(已在主循环外处理)

// 阶段 2: 上下文构建
constcontext=awaitassembleContext(session);

// 阶段 3: 模型调用
constresponse=awaitinvokeModel(context);

// 阶段 4: 响应解析
constparsed=parseResponse(response);

// 发送文本回复(如果有)
if (parsed.textContent) {
awaitsendReply(session.channelId, parsed.textContent);
    }

// 阶段 5: 工具执行
lettoolResults:ToolResult[] = [];
if (parsed.toolCalls?.length) {
toolResults=awaitexecuteTools(parsed.toolCalls);

// 将工具结果添加到对话历史
session.history.push({
role:'assistant',
content:parsed.toolCalls,
toolResults
      });
    }

// 阶段 6: 循环决策
shouldContinue=awaitdecideNextStep(parsed, toolResults);
loopCount++;
  }
}

三、代码级实战:自定义 Skill 开发

理解了 Agent 循环机制后,我们可以开发自定义 Skill 来扩展 OpenClaw 的能力。

3.1 Skill 文件结构

my-skill/
├── SKILL.md           # 技能定义文件(必须)
├── scripts/           # 脚本目录
│   └── main.py        # 主脚本
└── tools/             # 工具定义(可选)
    └── my_tool.json

3.2 SKILL.md 定义示例

markdown---
name: oracle-db-monitor
description: Oracle 数据库监控与巡检技能
metadata:
  openclaw:
    emoji: "🗄️"
---

# Oracle 数据库监控技能

## 功能

-自动巡检 Oracle 数据库健康状况
-检测慢查询、锁等待、空间不足等问题
-发送微信/邮件告警

## 工具定义

### oracle_check_health

检查数据库健康状况。

参数:
-host: 数据库主机地址
-port: 端口号(默认 1521)
-service: 服务名
-user: 用户名
-password: 密码(建议使用环境变量)

3.3 脚本实现

python#!/usr/bin/env python3
import oracledb
import json
import sys

def check_health(host: str, port: int, service: str, user: str, password: str) -> dict:
"""检查 Oracle 数据库健康状况"""
    dsn = f"{host}:{port}/{service}"
    
    results = {
        "status": "healthy",
        "issues": [],
        "metrics": {}
    }
    
    try:
        with oracledb.connect(user=user, password=password, dsn=dsn) as conn:
            with conn.cursor() as cursor:
                # 检查表空间使用率
                cursor.execute("""
                    SELECT tablespace_name, 
                           ROUND(used_space / total_space * 100, 2) as usage_pct
                    FROM (
                        SELECT tablespace_name,
                               SUM(bytes) as total_space,
                               SUM(bytes) - SUM(NVL(free_space, 0)) as used_space
                        FROM dba_data_files 
                        LEFT JOIN dba_free_space USING (tablespace_name)
                        GROUP BY tablespace_name
                    )
                    WHERE used_space / total_space * 100 > 80
                """)
                
                for row in cursor:
                    results["issues"].append({
                        "type": "tablespace_usage",
                        "tablespace": row[0],
                        "usage_pct": row[1]
                    })
                
                # 检查慢查询
                cursor.execute("""
                    SELECT sql_id, elapsed_time / 1000000 as elapsed_sec
                    FROM v$sql
                    WHERE elapsed_time > 10000000
                    ORDER BY elapsed_time DESC
                    FETCH FIRST 5 ROWS ONLY
                """)
                
                for row in cursor:
                    results["metrics"]["slow_queries"] = results["metrics"].get("slow_queries", [])
                    results["metrics"]["slow_queries"].append({
                        "sql_id": row[0],
                        "elapsed_sec": row[1]
                    })
                
                if results["issues"]:
                    results["status"] = "warning"
                    
    except Exception as e:
        results["status"] = "error"
        results["error"] = str(e)
    
    return results

if __name__ == "__main__":
    args = json.loads(sys.stdin.read())
    result = check_health(**args)
    print(json.dumps(result))

3.4 配置定时巡检

使用 OpenClaw 的 Cron 功能,可以设置每日自动巡检:

bash# 添加定时任务:每天 9:00 巡检
openclawcronadd--schedule"0 9 * * *"\
--task"检查 Oracle 数据库健康状况,如果有问题发微信提醒"

四、独到观点:Agent 循环的本质是什么?

4.1 从”对话”到”程序”的范式转换

传统 AI 助手的本质是 对话系统:用户问 → AI 答。而 OpenClaw 的 Agent 循环机制,本质上是一个 程序执行系统

  • 输入:用户的自然语言指令
  • 处理:AI 解析意图、规划步骤、调用工具
  • 输出:实际的执行结果(文件操作、系统命令、消息发送等)

这种范式转换的核心在于 工具调用 机制。工具调用不是简单的”插件”,而是 AI 与现实世界的 接口。通过这个接口,AI 从”虚拟世界”跨越到了”物理世界”。

4.2 为什么”本地优先”至关重要?

OpenClaw 强调”本地优先”设计,这不仅是隐私问题,更是 能力问题

  1. 1操作系统权限:只有本地运行,AI 才能访问文件系统、执行命令、控制浏览器。云端 AI 永远无法获得这些权限。
  1. 1实时性:本地执行不需要网络往返,响应速度更快。
  1. 1可靠性:不依赖外部服务,断网也能工作。
  1. 1成本:用自己的硬件和模型 API,没有订阅费。

4.3 与竞品的架构对比

特性 OpenClaw Cursor ChatGPT Claude Desktop
运行位置 本地 本地 云端 本地
操作系统权限 ✅ 完全 ✅ IDE 内 ❌ 无 ✅ 受限
多模型支持 ✅ 自由选择 ❌ 锁定 ❌ 锁定 ❌ 锁定
开源
技能扩展 ✅ Skill 系统 ❌ 插件受限 ✅ GPTs
24/7 运行 ✅ 后台服务
定时任务 ✅ Cron
持久记忆 ✅ LCM ✅ 受限

4.4 未来展望:Agent 的”操作系统”

OpenClaw 代表了 AI Agent 发展的一个重要方向:Agent 正在成为新的操作系统

  • 传统操作系统:应用程序 → 系统调用 → 内核 → 硬件
  • Agent 操作系统:自然语言 → Agent 循环 → Skills → 操作系统

在这个新范式中:

  • 自然语言 是新的编程语言
  • Skills 是新的应用程序
  • Agent 循环 是新的内核

五、总结

OpenClaw 的 Agent 循环机制,通过六个阶段的迭代(消息接收 → 上下文构建 → 模型调用 → 响应解析 → 工具执行 → 循环决策),实现了从”聊天”到”干活”的跨越。

核心要点:

  1. 1上下文构建 是智能的基础,LCM 记忆系统让 Agent “记得住”
  2. 2工具执行 是行动的核心,Skills 架构让 Agent “有能力”
  3. 3循环决策 是自主的关键,让 Agent 能够”自己想办法”

对于开发者,理解 Agent 循环机制不仅能更好地使用 OpenClaw,也为开发下一代 AI 应用提供了架构参考。未来,Agent 操作系统可能成为人机交互的新范式。