★ FEATURED ARTICLE OpenClaw 3.7 发布时最让我在意的不是又接入了多少新模型而是 ContextEngine 这个模块把 AI Agent 的上下文管理从主流程里彻底拆了出来。在 3.6 甚至更早的版本里上下文拼装逻辑普遍直接写在 agent 主循环里用户消息、系统提示词、工具返回、历史记录全部揉成一团用一堆字符串拼接和 if/else 分支硬塞进 prompt。结果就是换一个场景就要重写一段逻辑加一个数据源就要动主流程代码测试起来只能靠肉眼比对输出。ContextEngine 的出现相当于给上下文管理装了一套标准化的水管接口——每个数据源和每道处理工序都是独立的插件内核只负责把插件产出的东西按规则组装起来。这套机制解决的问题很具体Agent 的 prompt 不再是代码里写死的模板而是运行期由多个插件共同贡献的上下文块。整篇文章我会从接口设计、数据模型、注册加载机制、迁移步骤到 Token 治理一条线拆开讲代码以 OpenClaw 3.7 中常见实践为例最后附上我自己踩过的坑。适合正在维护 Agent 框架的工程团队也适合想摆脱“改 prompt 像改屎山”的个人开发者。1. ContextEngine 到底在解决什么问题1.1 硬编码上下文管理的典型困境先说个真实场景。之前我维护过一个客服类 Agent代码里有一段专门拼上下文的函数算上系统提示词、用户画像、渠道信息、历史对话、最近一次工具返回大概有七八个拼接段。最初写起来很爽每个来源一个字符串相加几行就完事。可一旦要加新的数据源比如接入新的客户标签系统就得在几十行的拼装函数里找个地方插入新逻辑同时还得判断当前对话是不是 VIP 用户、是不是售后渠道否则会多注入不该有的内容导致模型语气走偏。这类硬编码方案有三个很要命的问题上下文来源和组装顺序全部耦合在同一个函数里。没有任何抽象边界几个数据源之间互相 know 对方的存在。不同的接入场景无法复用。Web 端和客服工作台要的上下文不一样但硬编码只能靠参数和嵌套分支强行区分代码量翻倍。不可观测。最终 prompt 长什么样只有在线上出问题后通过日志重新拼一遍才看得到无法单独验证某个数据源的内容是否正确触发。这还只是功能层面的问题。工程层面更麻烦想对历史消息做不同的裁剪策略、想做 token 超限后的动态降级都得继续在同一个函数里加逻辑。到了某个临界点代码不是越写越清晰而是每个新需求都在削弱原来还算清晰的流程。ContextEngine 在这个背景下出现本质上就是给上下文管理重新定了边界谁提供内容谁处理内容以什么顺序组合三个维度彻底分开。1.2 ContextEngine 的核心定位和设计理念看名字就能猜到大方向ContextEngine 不是一个简单的 prompt 模板工具而是一个运行时的上下文编排引擎。它的核心模型是所有上下文内容都来自“插件”引擎本身只做收集、排序、过滤、组装和预算控制。类比一下更容易理解。早期的软件日志都是 print 到处飞后来有了 log4j/slog 之类的日志框架核心思想是把日志输出从业务代码里抽出去通过配置决定输出到文件还是控制台、格式是什么样。ContextEngine 对 Agent 上下文做的事情是同一个道理业务代码只声明“我需要哪些上下文源”至于这些上下文源怎么把数据变成文本块、按什么顺序排、超过预算时丢哪些块统统交给引擎和插件解决。这样主循环代码里不再出现任何直接拼 prompt 的字符串取而代之的是对 ContextEngine 的一次调用。从源码角度拆一下它的三个设计原则是以块Block为最小单元每个数据源输出一个或多个上下文块而不是一段大文字。所有处理动作都用插件表达包括过滤、去重、摘要、裁剪引擎不内置除基础排序之外的业务策略。每次请求上下文时都基于当前会话快照重新计算不用全局缓存保存拼好的 prompt。这三个原则让 ContextEngine 有了很强的扩展性。想新增一个企业知识库检索结果就写一个 ContextSource 插件想让提示词更简练就加一个 ContextProcessor 插件。改配置就能改变最终行为不用动 Agent 主流程。1.3 为什么说“告别硬编码”是实质性变化有人可能觉得不就是把字符串拼接拆成插件吗自己写个函数也能做到。但 ContextEngine 带来的不只是代码结构上的变化而是把上下文管理的整个生命周期都标准化了。先说扩展边界。在硬编码模式下新增一个上下文源你需要先看代码里哪里应该插、“插错了顺序会不会影响上下文”实际上是在改核心链路。在插件模式下新增一个源只需要实现接口、注册进配置。引擎负责在合适的时机调用你你只需要保证自己返回的块是正确的。可以把它理解为以前是主流程指挥数据源现在是主流程只做抽象调用数据源反过来向主流程贡献内容。然后是测试和线上排查。硬编码方案里一个 prompt 拼装问题通常要手工推演才能定位。ContextEngine 提供了快照和 dump 机制可以直接查看本次请求里到底有多少 ContextBlock、每个块来自哪个 source、token 消耗多少、最终排序结果如何。这意味着上下文问题从“玄学调参”变成了“看数据排查”对工程化非常关键。最后一个实际好处是灰度与演进。同一个 Agent 要在不同渠道上表现不同传统做法是开分支或者写一堆条件判断。有了 ContextEngine不同渠道可以对应不同的插件组和排序策略甚至可以在不改代码的情况下临时禁用某个数据源来对比效果。这种灵活度正是“告别硬编码”背后真正的工程价值。2. 源码级拆解接口契约与核心机制2.1 两个核心接口ContextSource 与 ContextProcessor在 OpenClaw 3.7 的源码里ContextEngine 对外暴露的接口并不复杂核心就是两个产出上下文的 ContextSource和加工上下文的 ContextProcessor。把这两个接口理解透整个引擎的工作机制就清楚了大半。先看 ContextSource。它的职责是“根据一个会话上下文产出一个或多个上下文块”。接口会被引擎在每个需要构建上下文的时机调用要求是异步的、支持增量返回from typing import AsyncIterable, Protocol, Any from dataclasses import dataclass class ContextSource(Protocol): # 插件在配置中的唯一标识比如 knowledge、history name: str # 返回一个权重值引擎用它来决定排序时是否提前或延后 def weight(self, session: SessionScope) - float: ... # 核心方法根据会话信息和预算产出上下文块 async def collect( self, session: SessionScope, budget: TokenBudget, ) - AsyncIterable[ContextBlock]: ...这个接口有几处值得注意的设计。第一个是weight()不是返回排序序号而是权重分。引擎在拿到所有块的 priority 之后会再叠加各个 source 的权重做调整。这样让插件可以表达“我这一类的信息一般比较重要但具体排序可以由引擎统一算”。第二个是collect()直接依赖预算对象源码里的要求是每个插件必须自行估算 token 数超出预算时只能返回部分内容。这个设计我们后文专门讲因为它直接影响 Token 治理效果。ContextProcessor 接口负责在引擎完成收集之后、最终组装之前对块列表做任意变换class ContextProcessor(Protocol): name: str def process( self, blocks: list[ContextBlock], session: SessionScope, ) - list[ContextBlock]: ...Processor 可以做的事情非常多去重、按长度裁剪、调用摘要模型压缩历史、剔除包含敏感信息的块、调整块与块之间的顺序。把策略从数据源中剥离出来是 ContextEngine 保持插件单一职责的关键。2.2 数据模型ContextBlock、SessionScope 与 TokenBudget接口只是骨架真正串起整个引擎的是几个数据类。先看最重要的 ContextBlock。每个数据源返回的内容都要包装成这个形态dataclass class ContextBlock: source: str kind: str content: str priority: int 50 token_count: int 0 metadata: dict[str, Any] field(default_factorydict) immutable: bool Falsesource 用来标记这个块来自哪个插件kind 是块的用途分类比如“system_prompt”“history window”“memory/todo”“tool_schema”。priority 表示基础优先级数字越大越靠前。metadata 里可以塞会话 ID、时间戳、原始文档路径等信息方便 Processor 做判断和排障。immutable 是一个很有意思的字段如果为 True那任何 Processor 都不能修改 content也不能丢弃该块主要用来保护系统级提示词不被后面的裁剪策略误伤。SessionScope 的含义比较直接但它不是简单地把用户 ID 和会话 ID 存起来而是包含当前请求的上下文环境dataclass class SessionScope: session_id: str user_id: str channel_type: str tenant_id: str | None None extra: dict[str, Any] field(default_factorydict)engine 在构建 prompt 前会把 SessionScope 填充好所有插件都基于这个只读对象来决策。这样保证了同一会话中多个插件看到的上下文信息是一致的也避免了硬编码时代那种“每个函数自己取参数”的混乱。TokenBudget 是实现可控性的关键工具dataclass class TokenBudget: total: int used: int 0 reserved: int 0 def remaining(self) - int: return self.total - self.used - self.reserved def can_afford(self, token_count: int) - bool: return self.remaining() token_count引擎会先把系统级、不可压缩的块 token 数标记为 reserved再把剩余额度按权重分给各个 ContextSource。每个插件在 collect 时要注意自己拿到了多少剩余额度从而决定返回多少内容。这个模型的好处是即使某个数据源返回了大量内容也不会因为它的实现质量问题把整个窗口撑爆。2.3 插件的发现与注册机制接口定义好了接下来看引擎如何在运行期加载插件。OpenClaw 3.7 的默认实现是同时支持两类声明一类是 Python entry points 自动发现另一类是 YAML 配置显式声明。源码里大致是这个流程# engine.yaml context_engine: sources: - name: system plugin: builtin.system_prompt priority: 100 - name: history plugin: builtin.sliding_history priority: 60 window: 20 - name: knowledge plugin: my_company.knowledge_source priority: 50 weight: 0.9 processors: - plugin: builtin.dedupe - plugin: builtin.trim max_blocks: 30 max_total_tokens: 3200引擎启动时会先加载 YAML根据 plugin 字段找到对应的类实例化后根据配置填充参数。YAML 里配置的 priority、weight 会作为插件默认值如果插件在 collect 阶段显式返回了 ContextBlock 里的 priority则以后者为准。自动发现则服务于那些不想写 YAML 的插件通过包管理器注册 entry point 后会被引擎统一收集。这个注册机制看着简单实际做好也不容易。它要求插件实例不能持有跨会话的可变状态否则高并发下会出现串数据之类的问题。源码里明确要求插件实现必须是“无状态或者会话内状态”所以我在写自定义插件时养成了一个习惯不在插件类里缓存任何用户级内容只在 metadata 里透传。2.4 生命周期与缓存状态ContextEngine 里的插件不是简单的函数而是有完整生命周期的组件。引擎对外暴露 init、start、stop 三个事件插件可以按需实现initialize插件实例被创建后调用适合做资源初始化比如建立数据库连接、加载模型配置。activate引擎准备开始使用该插件此时插件可以启动后台任务或预热缓存。deactivate引擎停止使用该插件插件需要释放临时资源。生命周期被设计为可重复触发目的是配合灰度发布。当你更新某数据源插件的权重或 prompt 内容时不需要重启主服务只要触发 deactivate/activate 循环重新加载配置即可。源码中有一个 version 字段用来标记插件版本能在 dump 输出中明确当前上下文是由哪一版插件生成的排查问题特别有用。缓存方面ContextEngine 几乎没有做全局缓存。它缓存的是插件数据而不是最终 prompt。比如 knowledge_source 插件的检索结果可以缓存十分钟但如果把最终拼好的 prompt 缓存复用一旦某个上游数据变了线上行为会非常难解释。所以源码里只在 SessionScope 层面提供了 request_id 和一个可选的本请求缓存字典数据源只要别在这个请求里重复计算重复内容就行。3. 把硬编码上下文迁移到 ContextEngine 的实操过程3.1 第一步盘点上下文来源划分插件边界动手迁移之前最重要的事情是盘点现状不要上来就拆。找出一段你现有的 agent 主流程把最终 prompt 里出现的每一个内容块都列出来标准只有一个这个块到底由谁产出、以什么条件出现。我以前迁移一个助手类 Agent 时整理出的上下文来源大致是系统人格固定文本任何时候都需要。当前用户信息登录后从用户服务拉取。对话历史走滑动窗口只保留最近 20 条。知识库检索结果当用户问题命中关键字时触发。待办事项列表仅在用户问到任务相关问题时注入。上一轮工具执行结果比如数据库查询返回。这份清单就是最原始的插件划分草案。接着要判断每个来源属于“数据源”还是“处理逻辑”。一般来源是数据策略是处理。比如“对话历史只保留最近 20 条”是处理逻辑它不应该写进数据源里而是由专门的 processor 去做裁剪。这个区分非常重要否则你又会把策略塞回插件里只是换了个位置继续硬编码。3.2 第二步实现一个自定义 ContextSource 插件具体怎么写一个插件我用一个待办事项来源举例。项目里有一个 todo_store存着用户当前未完成的待办项。旧代码是在拼 prompt 前查一次库再把结果用换行符拼进字符串。现在这段逻辑被完整搬进插件from typing import AsyncIterable from openclaw.context import ( ContextBlock, ContextSource, SessionScope, TokenBudget, ) class TodoListContextSource(ContextSource): name todo_list def __init__(self, todo_store, max_items: int 5): self._todo_store todo_store self._max_items max_items def weight(self, session: SessionScope) - float: # 优先处理任务类会话避免在闲聊场景里也注入待办 if session.channel_type task: return 0.95 if todo in session.extra.get(topics, []): return 0.8 return 0.3 async def collect( self, session: SessionScope, budget: TokenBudget, ) - AsyncIterable[ContextBlock]: if budget.remaining() 80: return items await self._todo_store.list_active( user_idsession.user_id, limitself._max_items, ) if not items: return lines [f- {item.title} (截止: {item.due_date}) for item in items] text 当前待办事项\n \n.join(lines) block ContextBlock( sourceself.name, kindmemory/todo, contenttext, priority80, token_countbudget.tokenizer.count(text), metadata{item_count: len(items)}, ) yield block这个实现里有几个地方是刻意为之的。空数据时直接 return不要产出空块避免在 prompt 中留下一段“当前待办事项”这种废文本。budget.remaining() 的判断是提前止损token 不够时宁可不要这段也不要挤掉更重要的内容。weight() 里做了场景区分闲聊时权重很低排序自然靠后甚至可能会被 trim processor 裁掉。这不是魔法而是把原来藏在 if/else 里的判断逻辑变成了可配置的插件权重。3.3 第三步用 Processor 统一处理后置逻辑数据源插件负责“吐内容”后置逻辑统一交给 Processor。在硬编码时代我们会在拼装完成后检查长度超了就把历史记录里的旧消息删掉还会担心同一个知识文档被检索到两次做了个去重函数。这些在 ContextEngine 里都可以拆成处理器。最简单的去重处理器from openclaw.context import ContextBlock, ContextProcessor, SessionScope class DedupeContextProcessor(ContextProcessor): name dedupe def process( self, blocks: list[ContextBlock], session: SessionScope, ) - list[ContextBlock]: seen_hashes: set[int] set() result: list[ContextBlock] [] for block in blocks: if block.immutable: result.append(block) continue h hash((block.source, block.content)) if h in seen_hashes: continue seen_hashes.add(h) result.append(block) return result再比如一个裁剪预算的 Processor按优先级整理块列表超 budget 时优先丢掉低优先级和高 token 的块class TrimByTokenBudgetProcessor(ContextProcessor): name trim_by_token def __init__(self, max_total_tokens: int): self._max_total_tokens max_total_tokens def process(self, blocks, session): used sum(b.token_count for b in blocks if b.immutable) movable [b for b in blocks if not b.immutable] movable.sort(keylambda b: (b.priority, -b.token_count), reverseTrue) keep [] for block in movable: if used block.token_count self._max_total_tokens: continue keep.append(block) used block.token_count return [b for b in blocks if b.immutable] keep写 Processor 时最需要注意的就是幂等性。同一个 Processor 不能因为执行两次就产生不同的结果。我在初版里写过一次裁剪处理器用的是“先排序再截断”但由于依赖的块列表本身是稳定顺序执行两遍没问题后来加了去重逻辑去重之后结果又变了。归根结底Processor 的输入输出都要保证与顺序无关否则很难以快照的方式做回归测试。3.4 第四步配置装配与调试验证代码写完之后实际迁移工作大部分时间花在“装配”上。把插件在 engine.yaml 里声明好告诉引擎哪些源应该启用、优先级如何、处理器顺序是什么。这里有一个非常容易踩的坑处理器的顺序会影响最终结果。比如先去重再裁剪与先裁剪再去重产生的块集合可以完全不同。建议的顺序是先去重再过滤敏感块再做裁剪最后做摘要。摘要这种操作通常比较慢放在最后可以减少被摘要的内容量。迁移后的验证也别用肉眼盯整个 prompt。ContextEngine 提供 dump 接口在一次请求结束后输出结构化的组装结果。我看的字段主要是共收集到多少个 ContextBlock分别来自哪个 source每个块的 priority 和 token_count经过所有处理器后哪些块被丢弃被哪个 processor 丢弃最终排序后的完整文本第一次跑通时我发现待办列表并没有出现在 prompt 里一查才发现是 TrimByTokenBudgetProcessor 里排序 key 写反了低优先级的块先被保留高优先级的待办被挤掉。没有 dump 机制这种问题真的很难定位。所以我的建议是任何涉及 ContextEngine 的开发环境都要保留 dump 日志这是最有价值的排障入口。4. 上下文组装顺序与 Token 治理机制4.1 排序策略priority、weight 与稳定顺序多数据源并存时组装顺序直接决定 Agent 的行为。把哪个块放在最前面、哪个块夹在中间模型理解的重点会完全不一样。ContextEngine 的排序规则是先按 ContextBlock.priority 降序排列当一个块有多个来源权重时再按插件 weight 降序处理。这里稍微展开一下。priority 的默认值是 50系统提示词一般给 100工具 schema 给 90历史记录给 60知识库给 40。这样排列的结果是模型先看到身份和工具使用说明再看长期记忆和知识最后是近期聊天记录。对话历史放最后是有道理的因为近期消息往往最贴近当前提问而且模型对越靠前的 token 注意力权重影响偏弱把历史放得太靠前反而会模糊重点。weight 的作用是二次排序。比如知识库插件和记忆插件 priority 相同但知识库在特定租户下更相关权重设成 0.9就会排在记忆前面优先把企业知识暴露给模型。这个设计比单纯用“优先级数字”灵活得多因为 priority 往往是静态的而 weight 可以根据 SessionScope 动态变化。除了排序本身引擎还保证了同优先级块的稳定顺序。实现方式是收集时记录每个块的自增 index排序 key 中把 index 作为最后一项。这一点看起来不起眼但对能否做快照测试很关键。没有稳定顺序同一请求跑两次结果都不一样Processor 的去重和裁剪逻辑就没法稳定复现。4.2 Token 预算分配预留、按需与降级策略Token 治理是 ContextEngine 里最值得学习的一块。它没有让每个插件自行设定长度上限而是先划定一个总预算再动态分配。流程可以简化成三步引擎把所有 immutableTrue 的块标记为 reserved它们占用的 token 数不可压缩。剩余额度按照插件的 weight 比例做“虚拟分配”。注意这里不是真的给每个插件锁死额度而是在 collect 时把 budget.remaining() 传给插件插件自己看着办。collect 结束后Processor 里的 trim 再次统一收敛保证最终不超过 total。举例来说如果总预算是 4000 token系统提示词占了 800 且 immutable那就剩 3200 分给所有数据源。如果会话场景是任务todo_list 的 weight 是 0.95历史记录是 0.8知识库是 0.5那它们拿到的虚拟占比高collect 时也能返回更多内容。闲聊场景下 todo_list 权重降到 0.3它自己就会收缩返回量。这套机制最大的价值是主动降级。在硬编码时代一旦某次知识库检索返回了异常庞大的一段文本大概率会直接挤爆 token 上限模型只能返回 400 错误。ContextEngine 里不会出现这种情况因为 knowledge 插件在 collect 前就看到 remaining 变少了会主动缩短返回片段即使没处理好trim processor 还会在上游统一拦截。4.3 会话级缓存与失效机制ContextEngine 的缓存设计很克制这与它要解决“上下文可解释”的目标一致。默认缓存只存在两个层面插件自带的内部缓存和引擎针对同一 session_id 做的瞬时缓存。瞬时缓存的有效期非常短只有几十毫秒本质上是避免在同一次请求中重复执行相同 source。它的 key 由 session_id、source name、插件版本和请求阶段四部分组成。这个所谓“插件版本”是个关键细节你改了插件行为和参数版本一变瞬时缓存自动失效不会出现“代码已更新但某个请求还在用旧逻辑”的诡异问题。对于跨请求的缓存引擎不做统一管理因为上下文管理中最怕脏数据。假如上一轮检索结果被缓存了但用户在两次请求之间新增了一条待办缓存里的待办列表就会过时模型就会给出错误回答。所以我实践中的做法是只有像“公司介绍”“固定 FAQ”这种变更频率极低的数据源在插件内部做十分钟左右的短期缓存。凡是涉及用户个性化数据的老老实实每次实时获取。4.4 常见问题排查表ContextEngine 上线后实际跑起来会遇到不少问题。我把常见的几类整理成了排查表对应的解决办法都是实测踩过坑之后总结出来的。现象可能原因排查方法与解决办法系统提示词在最终 prompt 里缺失trim processor 把 priority 低的系统块裁掉了或该块没有设置 immutable打开 dump 看丢弃原因给系统提示词块设置 priority100 和 immutableTrue知识库内容与历史消息重复检索命中已经出现在历史里的内容在 processor 链里加去重处理器按 content hash 判重超大 token 请求导致模型报错数据源插件没有遵守 budget 约束直接返回超长内容检查插件 collect 里的 remaining 判断增加兜底 trim processor同一次对话在 Web 端和任务端行为不一致同一 source 在不同场景权重差异太大或某个插件只在特定渠道注册对比两个渠道的 dump 输出确认插件组和 weight 配置是否符合预期插件更新后线上仍走旧逻辑引擎还在使用瞬时缓存插件版本号未更新每次修改插件实现后同步递增版本号手动触发 deactivate/activate多个待办数据源被重复注入底层数据被两个插件重复读取未做跨插件去重在 metadata 里带原始数据 IDprocessor 按 ID 去重排查这些问题的一个共同前提是必须打开 dump。如果在生产环境担心 dump 日志量太大可以只对指定 request_id 开启 full dump其余请求只记录块数量和总 token 数。这样既不影响性能又能保留足够的定位信息。5. 避坑心得与可落地的扩展方向这里聊一些迁移之后才真正体会到的经验。ContextEngine 本身不复杂真正复杂的是“如何在多个插件并存时保持可控”。我遇到的最大坑是插件职责越界。有人会把去重逻辑写进数据源插件里导致同一个数据源在不同响应中使用不同的裁剪规则最后很难解释一个问题“为什么这个待办列表这次只有三条下次却有五条”解决办法是在团队内部明确约定ContextSource 只产数据并估算 token所有过滤、去重、截断必须交给 Processor。这个约定比什么都重要写在 README 里都不为过。另一个值得分享的经验是快照测试。迁移期间我把上千条真实会话的输入输出保存下来作为黄金样本。每次修改插件或处理器后重新生成上下文快照与旧快照做 diff。ContextEngine 稳定排序和 dump 机制保证了这个流程完全可自动化。相比过去硬编码时代用人工跑几个 prompt 试效果现在的回归测试覆盖率高了一个数量级。如果你所在的团队还没有做这件事我强烈建议从迁移第一天就搭起来。往后扩展的方向也有不少。一个比较有意思的思路是把摘要模型作为 Processor 集成进来当历史记录超过预算时自动对早期轮次做语义压缩。另一个思路是把 ContextEngine 的插件抽象成跨语言协议这样不同语言实现的服务能通过 gRPC 贡献上下文块对多语言团队非常友好。我在自己的模拟项目里已经试过把 Python 插件与另一套 Java 服务打通调研下来的结论是可行但要注意序列化和版本同步成本。最后说一个小技巧。如果你刚接手一个项目想快速判断它有没有把上下文管理做好直接搜一下主流程代码里有没有大量字符串拼接和 if/else 判断。如果有你基本可以断定它还停留在硬编码时代如果没有而是看到一堆 context 相关的插件注册和 dump 调用那恭喜你这个项目的 Agent 上下文已经可以像积木一样自由拼装了。ContextEngine 不是银弹真正让上下文管理告别硬编码的是它引导出来的这套“提供者与处理器分离”的工程习惯。
阅读完成 · 觉得有帮助？
