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 = [m for m in self.history if m.get("role") != "system"]
    if not non_system_messages:
        return

    estimated_tokens = self._estimate_tokens_rough(self.history)
    if (
        len(non_system_messages) < self.short_term_compact_threshold
        and estimated_tokens < self.short_term_compact_token_threshold
    ):
        return

    recent_keep = min(self.short_term_recent_keep, self.short_term_max_messages)
    if len(non_system_messages) <= recent_keep:
        return

    head = 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 = []
for msg in self.history:
    if msg.get("role") != "system":
        continue
    content = msg.get("content")
    if isinstance(content, str) and content.startswith(self.HISTORY_SUMMARY_PREFIX):
        continue
    base_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] = []
    for msg in messages:
        role = str(msg.get("role", "unknown"))
        content = msg.get("content")
        tool_calls = msg.get("tool_calls")

        if tool_calls:
            names = []
            for tc in tool_calls:
                fn = tc.get("function") or {}
                name = fn.get("name") or "unknown_tool"
                names.append(name)
            lines.append(f"- assistant requested tools: {', '.join(names)}")

        if isinstance(content, str) and content.strip():
            collapsed = " ".join(content.strip().split())
            lines.append(f"- {role}: {collapsed[:220]}")

summary = "\n".join(lines)
if len(summary) > self.short_term_summary_max_chars:
    return summary[: self.short_term_summary_max_chars] + "\n...[summary_truncated]"
return summary

当前版本采用“规则压缩”：保留角色信息、工具调用信息和内容摘要，并加上摘要长度上限。
后续升级为 LLM 摘要时，可以在不改调用方逻辑的前提下平滑替换。

---

9. 和当前系列的衔接关系

• 第 1~4 篇：把 Agent 跑起来（能调用工具）
• 第 5 篇：让 Agent 在长对话下还能稳定工作
• 第 6 篇：把关键事实写入长期记忆，跨会话复用

---

10. 下一篇衔接

第 6 篇进入长期记忆：MEMORY.md 持久化 + 召回注入。

---

11. 看到这里，不妨支持一下

如果这篇对你有帮助，欢迎点赞、收藏、关注。
后面第 6~8 篇会进入记忆与检索的核心区域（也是最实用的部分）。
有余力的话欢迎打赏支持，我会把实战代码和踩坑案例更新得更快。