资讯动态

OpenClaw四层架构源码拆解:从事件循环到智能体编排的完整实践

发布时间:2026/10/17 6:11:55 来源:尧图企业网站定制
1. 为什么 OpenClaw 值得拆开看一个不算新的新架构如果你混开源圈子或者做智能体应用开发最近两年大概率反复听到过OpenClaw这个名字。坦白说我第一次看到它的时候心里想的是又来了一个套壳框架——毕竟这几年的套壳项目太多了README 写得天花乱坠拉下来一看全是胶水代码。但实际用了一段时间、把源码从头到尾啃过一遍之后我改变了看法OpenClaw 的设计思路确实有它独到的地方尤其是它的四层架构不是那种为了方便画架构图而硬拆出来的四层而是每一层都承担了非常明确、几乎不可互相替代的职责。这篇文章想做的事情很简单从源码拆解的角度把 OpenClaw 的四层架构讲清楚——每一层到底在干什么、它依赖什么、边界在哪、为什么这么分然后落地到我自己动手搭建一个基于 OpenClaw 的最小可用系统的完整过程。适合谁看两类人一是像我一样搞智能体应用开发、被各种框架选择困难症折磨的开发者二是刚入门、想搞懂智能体项目到底是怎么组织代码的进阶学习者。如果你只是想要一份 Quickstart 照着敲完就跑那官方文档已经够了但如果你想知道为什么跑起来之后是这个样子、出了问题该去哪层排查这篇文章就是给你写的。先说结论OpenClaw 的四层架构从底层往上依次是基础设施层、核心引擎层、能力服务层、应用编排层。名字听起来很常规但它真正让我觉得有价值的地方在于每一层都严格遵循了向下依赖、向上屏蔽的规则——下层不关心上层怎么用上层不知道下层的实现细节层与层之间靠一套定义得很克制的接口协议通信。这种设计直接带来的好处是可替换性极强。比如你不想用它的默认记忆存储把核心引擎层里那一个存储适配器换掉就行其他三层代码一行都不用改。我在拆源码之前先画了一个问题清单这个项目的入口在哪里/ 配置是怎么流转到每一层的/ 它凭什么声称自己可扩展拆完之后这些问题都有了答案。接下来我按自己的阅读路径从底层开始往上一层一层讲。2. 基础设施层不显眼却决定能不能跑起来的部分2.1 这层到底放了什么东西很多人读源码习惯先从入口或者核心算法读起但我建议先看基础设施层因为这一层决定了你部署时的心态。OpenClaw 的基础设施层包括日志系统、配置管理、常量定义、通用工具函数、运行时上下文context的创建与管理。没啥花哨的全都是支撑性代码。举个例子它的配置管理模块。OpenClaw 支持从环境变量、YAML 文件、命令行参数三处来源加载配置并且有一个明确的优先级顺序命令行参数 环境变量 配置文件 内置默认值。这个设计本身不稀奇很多项目都是这么做的但让我比较欣赏的一点是它对配置来源可追踪的处理——当配置加载完成之后系统会生成一份最终生效配置快照包含每个配置项的来源标注在调试模式下可以直接打印出来。也就是说你不会遇到我明明在 YAML 里写了这个参数程序为什么不生效这种经典困惑。再看日志系统。OpenClaw 没有自己造轮子而是在底层封装了结构化日志每个日志条目都带上 request_id 和 layer 字段。这个设计在开发的时候你可能觉得多余但一旦进入多智能体协作或者异步任务排错阶段没有 request_id 的日志基本上等于不可用。我在搭建的时候同时跑了 5 个智能体做不同任务日志全部混在一起靠的就是按 request_id 过滤来定位到具体某个智能体的完整调用链。2.2 基础设施层的设计边界不提供业务能力这层唯一的职责是提供所有其他层都会用到的东西但它有一个非常严格的边界绝对不会出现任何业务逻辑。比如日志模块里不会有当用户意图是订机票时额外记录一条日志这种代码配置模块里也看不到任何跟某个具体能力插件相关的默认配置项。这个边界的意义在于基础设施层可以被完整地抽出去做单元测试不需要任何依赖。我自己实测把这层单独拎出来跑一遍测试用例耗时基本在毫秒级没有网络调用、没有外部服务依赖。这反过来也保证了上层所有模块的测试都能在一个干净的环境里启动——你做集成测试的时候是不会被日志系统本身出 bug这种问题干扰的。从实战角度我要特别提一下 OpenClaw 的常量定义文件。它把所有魔法数字和魔法字符串都收敛到了几个常量类里比如超时时间、重试次数、消息类型名、事件名。我后来在二次开发时增加了一个新的消息类型全程只动了常量定义和对应 handler 注册没有去满项目搜字符串。这个好习惯大家在自研项目里真的应该学起来。3. 核心引擎层源码里最值得反复读的部分3.1 事件循环与消息总线一切的节拍器核心引擎层是整个 OpenClaw 的心脏它不直接实现任何业务能力而是提供了一个让各种能力可以协同工作的运行框架。如果你打开源码第一眼就会被一个东西吸引一个全局的事件循环event loop以及围绕它的消息总线message bus。这个事件循环的工作模型不复杂维护一个消息队列不断从队列里取出消息根据消息类型分发到对应的 handler处理完之后把新的结果消息再放回队列。整个流程是单线程异步的。很多人听到单线程会下意识觉得性能不行但在智能体这种 IO 密集型大量调用外部 API、数据库、模型接口场景下异步单线程反而是最合理的选择——省掉了大量的锁竞争和上下文切换开销。为了让你感受一下这个中心的节拍是什么样的我简化了事件循环的核心代码基于 OpenClaw 源码结构还原class EventLoop: def __init__(self): self.queue asyncio.Queue() self.handlers {} def register_handler(self, event_type: str, handler): self.handlers[event_type] handler async def post(self, event: Event): await self.queue.put(event) async def run_forever(self): while True: event await self.queue.get() handler self.handlers.get(event.type) if handler: await handler.handle(event) else: logger.warning(fno handler for event: {event.type})这里有个细节值得注意register_handler的注册机制。OpenClaw 在启动时会扫描所有已安装的能力插件插件通过一个装饰器把我关心哪类消息告诉引擎。这个机制直接决定了插件的热插拔能力——你新增一个插件只要它在启动时被扫描到并注册了事件处理函数就能立刻参与整个系统的工作循环。3.2 Agent Runtime状态机与生命周期管理事件循环之上核心引擎层还有一块非常重要的东西Agent Runtime也就是智能体的运行时生命周期管理。每个智能体实例都会被封装成一个状态机状态包括 idle、running、waiting_input、error、terminated 等。整个生命周期事件循环驱动收到任务消息 - running需要外部输入 - waiting_input出现不可恢复的异常 - error。为什么状态机这种老派设计在这里这么重要因为多智能体场景下同时存在大量不同状态的 Agent 实例如果不上状态机状态流转会彻底失控。我之前的项目里就吃过这个亏用 Python 的 threading 全局变量来管理智能体状态结果稍微复杂一点就出现两个线程互相踩状态的情况。OpenClaw 用状态机强制了流转路径比如 running 是绝对不能直接跳到 terminated 的必须先经过一个取消cancelling状态。这个约束看着死板但恰恰是它能支撑复杂场景的底气。如果你要深入阅读这部分的源码我建议关注三个文件agent.pyAgent 基类、runtime.py运行时管理、state.py状态定义与流转校验。其中state.py里有一个字典专门定义合法流转路径这个设计非常直观也方便做代码审查——你一眼就能看到一个状态能不能到达另一个状态。3.3 插件管理机制的核心契约先行插件系统的设计直接反映了一个智能体框架的成熟度。OpenClaw 的插件管理放在核心引擎层但它并没有把具体插件实现塞进来——只定义了插件需要遵循的契约接口与协议真正的实现在更上层。一个插件被加载需要声明三样东西元信息名称、版本、作者、能力列表它能执行哪些操作、依赖声明它需要其他插件或服务提供什么能力。元信息和依赖声明通过一个plugin.toml文件来描述能力列表则在代码里通过注册函数暴露。这套机制带来的最大价值是启动时的依赖完整性检查。OpenClaw 在加载插件时会做一次拓扑排序如果 A 插件声明依赖 B 插件的能力而 B 没被安装或加载启动直接报错而不是等运行到某个功能时才炸给你看。这个体验有多重要做过工程项目的人应该都懂——前置依赖检查永远比运行期报错好排查一百倍。4. 能力服务层可替换的工具箱4.1 记忆存储从内存到数据库的平滑切换现在进入第三层能力服务层。这一层把智能体可能会用到的基础能力以独立服务的方式提供出来。最典型的能力是记忆存储、模型调用、工具执行沙箱、向量检索。先说记忆存储。OpenClaw 默认提供了一个基于 SQLite 的本地记忆存储实现而源码把它定义成了一个MemoryBackend接口。接口只包含几个方法save_conversation、get_recent_conversations、search_memories、clear_expired。这意味着你完全可以把默认的 SQLite 实现换成一个基于 PostgreSQL 或者 Redis 的实现而不碰任何其他层的代码。我在做一个小 demo 的时候就把记忆后端换成了 Redis原因是本地测试时发现自己同时跑多个智能体实例SQLite 的写入锁频繁冲突。整个过程只花了一个多小时实现接口、写一个 factory 注册、改配置。这个体验让我彻底明白了面向接口编程在架构设计里的真实分量。4.2 模型接入层统一抽象的关键接口模型接入部分也是同样的思路。OpenClaw 没有把模型调用直接写在某个 Agent 类里面而是抽出了一个LLMBackend接口支持不同的模型提供商OpenAI、Anthropic、本地部署模型等。接口很克制就三个方法complete补全、chat多轮对话、stream_chat流式对话。这里有个特别重要的设计它与上层之间传输的不是模型响应原文而是一个统一的ModelResponse数据结构。这个结构里除了文本内容还带上了 token 消耗、延迟、模型名、raw_response 等元信息。这个元信息保留的做法对实际运维价值巨大——你可以基于这些数据做成本统计、延迟监控、模型效果对比而不用再去改上层代码。4.3 工具沙箱智能体动手的边界模型调用和记忆存储好理解工具沙箱可能很多人一上来不太清楚是干什么的。简单说当智能体需要执行一段代码、操作文件系统或者调用某个命令行工具时这些动作都必须在沙箱里跑。我刚开始觉得这个设计过度防御了——我的智能体只是调个 API 而已为什么要沙箱直到有一次我给智能体接了一个读取 Excel 文件并分析的工具这个工具内部调了一个第三方库第三方库在特定情况下会往用户目录写临时文件。如果整个进程共享同一个工作目录那这个临时文件就会污染主进程的目录。有了沙箱所有这些副作用都被限制在一个可摧毁的临时环境里用完即焚。5. 应用编排层真正面向场景的最后一公里5.1 编排层要解决的核心问题目标怎么拆、顺序怎么定前面三层解决了能力怎么实现、系统怎么运转但用户真正面对的是我让它帮我完成一个复杂任务它怎么拆分和安排——这是应用编排层的工作。OpenClaw 的编排层支持两种模式基于 Plan-Execute 的显式编排和基于动态流程的隐式编排。前者适合任务边界清晰的场景比如读取文件 - 清洗数据 - 生成图表 - 发送报告后者适合目标模糊、需要模型动态决策的场景比如帮我研究一下这个方向的论文并整理成综述。在源码层面显式编排对应一个Workflow类它维护一个节点图DAG每个节点绑定一个具体的 Agent 或者能力调用。隐式编排则是一个Orchestrator类它接收一个任务目标然后循环执行决策下一步 - 执行 - 观察结果 - 判断是否完成。我把两种模式的简化代码对比一下# 显式编排定义一个固定的两步工作流 workflow Workflow() workflow.add_step(fetch_data, fetch_agent) workflow.add_step(analyze, analyze_agent) result await workflow.run(给我分析这个网址的文章)# 隐式编排动态决定下一步 orchestrator Orchestrator(llm_backend) result await orchestrator.run(你需要下载这个网页内容提取核心观点再写一份摘要保存到当前目录)实际用下来我的体感是任务越明确显式编排越可控任务越模糊隐式编排越灵活。如果任务的步骤数超过 4 步我强烈建议至少把主流程用显式编排搭好把真正动态的部分留给隐式决策——全靠大模型临场发挥几步之内它就很容易在细节上反复横跳。5.2 Agent 之间的协作呼叫、响应与仲裁应用编排层还有一个重要职责多 Agent 协作。OpenClaw 的协作模型是呼叫-响应-仲裁三段式一个 Agent 发现自己搞不定当前任务或者需要别的能力就向编排层发出一个呼叫请求编排层根据请求内容选择合适的 Agent 或能力分配任务当多个 Agent 都给出了候选结果时编排层需要做仲裁。仲裁逻辑是可以定制的OpenClaw 默认提供了三种先到先得、投票制、评分制。我在一个模拟场景里让多个 Agent 分别解读一份财报并给出结论对比了这三种仲裁方式实测下来评分制最稳投票制最容易从众。原因不复杂投票制天然会被表达能力强但信息掌握不全的 Agent 带偏而评分制要求每个 Agent 给出置信度再按置信度加权结果要理性得多。6. 从零搭建一个最小 OpenClaw 系统我的操作记录6.1 环境准备与初始化容易踩的坑理论讲完上实操。我是在一台 8GB 内存的普通开发机上完成的搭建操作系统是 Ubuntu 22.04Python 版本 3.11。整个流程走下来最大的一个坑是版本兼容性OpenClaw 对 Python 版本有硬性要求3.10 以下和 3.12 以上都会在安装依赖阶段报错所以一开始就老老实实建一个 3.11 的虚拟环境最省事。初始化流程如下# 1. 创建虚拟环境务必指定 3.11 python3.11 -m venv .venv source .venv/bin/activate # 2. 安装 OpenClaw 主包 pip install openclaw # 3. 初始化项目骨架 openclaw init my_first_project cd my_first_project # 4. 启动开发模式内置热加载 openclaw devopenclaw init生成的项目骨架结构值得看一眼因为它就是四层架构在项目形态上的具体体现my_first_project/ ├── config/ # 配置目录基础设施层 │ ├── settings.yaml │ └── logging.yaml ├── core/ # 核心引擎层通常不修改 │ ├── event_loop.py │ ├── agent.py │ └── plugin_manager.py ├── capabilities/ # 能力服务层可替换实现 │ ├── memory_backend.py │ ├── llm_backend.py │ └── sandbox.py ├── workflows/ # 应用编排层你的主战场 │ ├── my_workflow.py │ └── orchestrator.py └── plugins/ # 插件目录 └── my_first_plugin/ ├── plugin.toml └── main.pycore 通常不修改这句话实际执行的意思是你想扩展某个核心行为优先通过继承或者插件注册机制来做而不是直接去改事件循环的实现。直接改 core 里的源码短期内确实能爽但后续每次升级 OpenClaw 版本都要处理冲突——这个我替你们踩过。6.2 写一个自定义插件并注册到系统现在来干一件正事写一个自定义插件让它能被系统扫描到并参与事件循环。这个插件的作用很简单——收到一个包含 weather 的事件时返回一行模拟的天气数据。第一步创建插件目录和plugin.toml[plugin] name weather_demo version 0.1.0 description A simple weather demo plugin [capabilities] provides [weather_query] [config] api_url https://api.example.com/weather第二步写主逻辑# plugins/weather_demo/main.py from openclaw.decorators import on_event on_event(weather_query) async def handle_weather_query(event, context): city event.payload.get(city, beijing) # 这里为了演示只返回一条模拟数据 data { city: city, temperature: 28, humidity: 45, source: local } await context.reply_to_sender( event.message_id, payloaddata, statussuccess )注意这里用了on_event装饰器这正是前面核心引擎层提到的插件注册机制的入口。在openclaw dev模式下修改这个文件保存后系统会自动重载插件不需要手动重启。这里有个我实际碰到的问题第一次写插件时我把on_event的装饰器当作普通函数调用——写成了on_event(weather_query)却忘记它在源码里要求传入一个event_type字符串其实是对的。真正让我折腾了半小时的问题是插件扫描时要求本目录所有模块必须能被 import不能有语法错误否则整个插件目录会被跳过并且只会在启动日志里给一行 warning。你不会在控制台看到红色报错所以必须习惯去翻日志不然你完全感受不到插件没加载这件事。6.3 编排一个双 Agent 协作流程插件跑通之后我编排了一个双 Agent 协作流程Agent A 负责抓取数据Agent B 负责生成分析结论。目的是测试多 Agent 之间的呼叫响应与仲裁机制。首先写一个编排配置文件放在workflows/下workflow: id: dual_agent_demo steps: - name: fetch_step agent: fetch_agent on_success: analyze_step - name: analyze_step agent: analyze_agent requires: fetch_step然后在代码里定义两个 Agent并在编排层把它们串起来# workflows/dual_agent_demo.py from openclaw import Workflow, WorkflowRunner async def main(): workflow Workflow(dual_agent_demo) workflow.agent(fetch_agent) async def fetch_agent(task): # 模拟获取数据 return {raw_data: [1, 2, 3, 4, 5]} workflow.agent(analyze_agent) async def analyze_agent(task): raw_data task.get_input(fetch_step).result[raw_data] # 模拟分析 return {conclusion: fsum{sum(raw_data)}} runner WorkflowRunner(workflow) result await runner.run( input_data{url: https://example.com/data} ) print(workflow result:, result)这段代码真正跑通之后我的感受是编排层的抽象确实降低了多 Agent 代码的耦合度——两个 Agent 之间没有直接引用而是通过任务输入/输出传递数据。如果你要替换 Agent B 的实现不会影响 Agent A反过来也一样。这就是应用编排层存在的意义它管的是流程和结果不管每个 Agent 内部怎么做。6.4 把默认记忆后端切到 Redis最后为了让记忆存储不那么玩具化我把默认的 SQLite 记忆后端切到了 Redis。Step 很简单实现MemoryBackend接口通过 factory 注册改配置我写的简化版 Redis 记忆后端如下# capabilities/redis_memory.py import redis.asyncio as aioredis from openclaw.base import MemoryBackend class RedisMemoryBackend(MemoryBackend): def __init__(self, redis_url: str): self.client aioredis.from_url(redis_url) async def save_conversation(self, sender_id, message): key fconv:{sender_id}:{message.timestamp} await self.client.set(key, message.text) async def get_recent_conversations(self, sender_id, limit20): keys await self.client.keys(fconv:{sender_id}:*) keys sorted(keys, reverseTrue)[:limit] return [await self.client.get(k) for k in keys] async def search_memories(self, sender_id, keyword): # 简化版直接扫全量实际应接向量检索 all_keys await self.client.keys(fconv:{sender_id}:*) matches [] for k in all_keys: text await self.client.get(k) if keyword in text: matches.append({key: k, text: text}) return matches然后在配置文件的 capabilities 段里切换memory: backend: redis redis_url: redis://localhost:6379/0这里我要提醒的一点是*不要在生产环境用上面的keys方式做搜索——Redis 的 keys 命令在大数据量下会阻塞单线程。我这个 demo 只是为了验证架构的可替换性如果你的数据量真的很大请务必接向量数据库或者 RedisSearch。同理save_conversation里的timestamp字段一定要是稳定递增的否则后面按时间排序取最近记录会乱。7. 拆解过程中最让我意外的三个点7.1 官方文档没说但源码里写了的隐藏约定读完源码后你会发现 OpenClaw 有一种重约定、轻配置的味道。但有些约定只存在于源码注释和命名规范里官方文档并没有展开讲。举两个例子插件目录名必须是合法 Python 包名。看起来像是废话但如果你和别人协作很可能出现my-plugin这种带中划线的目录名然后整个插件加载失败。事件类型的命名全局唯一。OpenClaw 在注册 handler 时不会检查事件名冲突如果两个插件注册了同一个事件类型后加载的会覆盖先加载的而且不会给你任何提示。这个行为如果你不知道会出现为什么我的插件生效了但结果不对这种灵异事件。7.2 到底哪一层才是性能瓶颈的根源为了找到整个系统跑复杂任务时的性能瓶颈我做了一个简单的压测同一个任务分别测只有基础设施引擎和加上能力服务层编排层两种情况记录平均响应时间。结果很符合预期但也很有参考价值场景平均耗时主要耗时点纯引擎事件循环无外部调用5ms消息队列读写引擎记忆后端SQLite45msSQLite 写入引擎记忆后端Redis18ms网络往返引擎LLM 调用900ms模型响应从这个小小的对照表能看出来瓶颈从来不在框架本身而在你接的外部依赖。如果你的系统变慢了先去看外部调用的耗时而不是急着优化事件循环的队列实现。这句话说起来简单但我见过太多团队把时间花在调框架内部参数上最后发现第三方服务超时才是罪魁祸首。7.3 四层架构在排查问题时的实际价值最后说一个更深一点的体感。我在搭建过程中故意制造了一个故障让RedisMemoryBackend在启动时抛异常。你猜系统会怎么样答案是核心引擎照常启动其余三层完全正常只是记忆相关的功能不可用日志里清晰地标注了是 capabilities 层某组件初始化失败。这个故障实验其实验证了架构设计里很重要的一点分层越清晰故障定位越精准。你把日志按 layer 字段过滤之后几乎不用猜问题出在哪——它直接告诉你memory backend got an error。这种故障可排查性不是靠写日志堆出来的而是由代码结构天然决定的。上层不会吞掉下层的异常下层也不会假装自己是上层。8. 如果我是你我会怎么学这套架构把 OpenClaw 源码拆完一遍我最想对读者说的一句话是不要把这个项目当成一个框架文档去读而要把它当成一个架构范本去拆。你可以完全不用 OpenClaw但它的分层思路——基础设施收敛全局杂物、引擎层定义运行规则、能力层做到可替换、编排层专注业务编排——这套组织方式对你的任何智能体项目都有直接借鉴意义。如果你打算自己动手拆一遍源码我建议按这个顺序来先读state.py状态定义理解 Agent 生命周期再读event_loop.py事件循环理解系统怎么动起来然后抽一个你最熟悉的能力比如记忆存储顺着接口找到默认实现再换一个实现跑通最后去workflows/目录里看编排层提供的 demo修改一两个节点看看数据流变化。这个顺序的好处是你每一步读到的知识都会被下一步立刻验证而不是一个文件一个文件地云阅读。我始终觉得架构这种东西只有亲手改过一点什么才会真正长在自己身上。我在动手做这套搭建的过程中最明显的一个感受是以前写智能体应用脑子里全是模型怎么调、提示词怎么优化很少去想整个系统的骨架该怎么立。现在再回头看自己早年的代码问题几乎都出在架构上——逻辑互相纠缠、模块边界模糊、换个配置要牵扯半天。OpenClaw 的四层架构不是银弹但它让我看到了一套被实践验证过的组织方式也让我在下一个自研项目里有了可以借鉴的骨架。希望你读完这篇拆解之后也能动手打开源码找到属于你自己的那个为什么这样设计的答案。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑