当前位置：首页 > news >正文

龙虾-OpenClaw一文详细了解-手搓OpenClaw-5 短期历史窗口与压缩

news 2026/7/6 10:23:58

0. 为什么要手搓 OpenClaw

OpenClaw 很强，但完整工程体量也很大。对于大多数开发者来说，直接阅读全量代码会有三个痛点：

模块多：Gateway、Agent、Tools、Sessions、Channels 互相耦合
路径长：一条消息从输入到回复，跨越多个子系统
调试难：没有自己的“最小版本”，很难定位问题

所以这个系列采用一个更实用的学习路径：
先做最小闭环，再逐步补齐能力。

代码地址:https://gitee.com/wisdomfriend/openclaw_py

1. 目标

用 Python 从 0 到 1 复现 OpenClaw 的核心能力：

Agent Loop（工具调用 + 多轮推理）
Session 与并发隔离
记忆系统（短期 + 长期）
Skills 系统（分层加载）
Web/Telegram 等渠道接入

第一篇的阶段目标是：

跑起 FastAPI 服务
打通一个最小/v1/chat对话接口
具备会话隔离与并发控制（每会话锁 + 全局信号量）

2. 目标架构

3. 本篇目标

实现短期历史管理能力：

历史窗口裁剪（保留最近关键消息）
触发式压缩（达到阈值再做摘要）
压缩后继续多轮对话，不丢核心目标

4. 为什么第 5 篇先做这个

到第 4 篇为止，Agent 已经有 Provider + Tool Runtime。
下一步如果不先控制上下文，就会遇到：

对话越长，token 消耗越高
模型开始丢上下文，回复变飘
工具结果越来越多，prompt 膨胀明显

所以第 5 篇先解决“短期窗口 + 压缩”，第 6 篇再做长期记忆。

5. 本篇范围

做什么

定义滑动窗口策略（最近 N 条）
设计压缩触发条件（按消息数或 token 估算）
保留不可压缩信息（system 指令、最新工具结果）
给出压缩前后对比与验证方式

不做什么

不做长期记忆落盘（第 6 篇）
不做 RAG 检索融合（第 7 篇）

6. 设计草图：窗口 + 压缩

7. 项目关键代码

7.1 配置层：窗口与压缩阈值

文件：openclaw_py/app/config.py

short_term_max_messages:int=Field(default=40,validation_alias="SHORT_TERM_MAX_MESSAGES")short_term_compact_threshold:int=Field(default=24,validation_alias="SHORT_TERM_COMPACT_THRESHOLD",)short_term_recent_keep:int=Field(default=10,validation_alias="SHORT_TERM_RECENT_KEEP")short_term_compact_token_threshold:int=Field(default=3000,validation_alias="SHORT_TERM_COMPACT_TOKEN_THRESHOLD",)short_term_summary_max_chars:int=Field(default=1800,validation_alias="SHORT_TERM_SUMMARY_MAX_CHARS",)

这组参数把“按消息数触发”和“按 token 粗估触发”结合起来，避免仅按条数裁剪导致上下文体积失控。

7.2 Agent 压缩触发与保留窗口

文件：openclaw_py/app/core/agent.py

def_maybe_compact_history(self)->None:non_system_messages=[mforminself.historyifm.get("role")!="system"]ifnotnon_system_messages:returnestimated_tokens=self._estimate_tokens_rough(self.history)if(len(non_system_messages)<self.short_term_compact_thresholdandestimated_tokens<self.short_term_compact_token_threshold):returnrecent_keep=min(self.short_term_recent_keep,self.short_term_max_messages)iflen(non_system_messages)<=recent_keep:returnhead=non_system_messages[:-recent_keep]tail=non_system_messages[-recent_keep:]summary=self._compact_messages(head)

这里的recent_keep = min(...)用于分离“策略值”和“系统硬上限”：
可调的保留策略不会突破系统可承受的最大窗口。

7.3 历史摘要注入与去重

文件：openclaw_py/app/core/agent.py

base_system_messages=[]formsginself.history:ifmsg.get("role")!="system":continuecontent=msg.get("content")ifisinstance(content,str)andcontent.startswith(self.HISTORY_SUMMARY_PREFIX):continuebase_system_messages.append(msg)summary_message={"role":"system","content":f"{self.HISTORY_SUMMARY_PREFIX}\n{summary}",}self.history=[*base_system_messages,summary_message,*tail]self.compaction_count+=1

会主动移除旧的[history_summary]，保证每轮压缩后只保留最新摘要，避免摘要层层嵌套。

8. 规则压缩实现细节

文件：openclaw_py/app/core/agent.py

def_compact_messages(self,messages:list[dict[str,Any]])->str:lines:list[str]=[]formsginmessages:role=str(msg.get("role","unknown"))content=msg.get("content")tool_calls=msg.get("tool_calls")iftool_calls:names=[]fortcintool_calls:fn=tc.get("function")or{}name=fn.get("name")or"unknown_tool"names.append(name)lines.append(f"- assistant requested tools:{', '.join(names)}")ifisinstance(content,str)andcontent.strip():collapsed=" ".join(content.strip().split())lines.append(f"-{role}:{collapsed[:220]}")

summary="\n".join(lines)iflen(summary)>self.short_term_summary_max_chars:returnsummary[:self.short_term_summary_max_chars]+"\n...[summary_truncated]"returnsummary