先用一分钟理解项目
大模型上下文达到上限时,宿主 Agent 会把旧对话压缩成摘要。这个摘要通常是自动生成的,问题在于:模型可能保留“最后选了 PostgreSQL”,却忘记“为什么放弃 SQLite”;也可能留下大量 grep、npm install 和临时错误,把真正的项目约束挤掉。
提取候选记忆
先别被类型名吓到:它们其实是在传递不同形态的同一件事
下面先把几个名字翻译成人话。你可以把整个系统想成一个“整理会议记录的人”:先把原始对话抄清楚,再圈出可能重要的句子,决定哪些直接留下,哪些直接删掉,哪些需要问项目负责人,最后给摘要器一份整理说明。
| 代码里的名字 | 人话 | 它解决什么问题 | 一个直观例子 |
|---|---|---|---|
ConversationMessage | 一条“整理过格式”的对话记录 | Pi、Harness 的原始消息长得不一样,Core 先把它们翻译成同一种格式 | 角色是 user,内容是“不能修改 public API”,并带有来源 ID |
MemoryCandidate | 从对话里圈出来的一条“可能值得记住的事实” | 把整段长对话拆成一条条可以单独判断的记忆 | “SQLite 因并发问题被放弃” |
InspectionResult | 这次检查的总成绩单 | 不仅列出所有候选,还告诉宿主:哪些自动 Keep、哪些自动 Drop、哪些要问人 | PostgreSQL → auto Keep;grep 输出 → auto Drop;SQLite → Review |
ReviewDecision | 人对某一条候选做出的选择 | 把 UI 上的“Keep / Drop”准确对应回某个候选,而不是靠数组顺序猜 | {candidate_id: "sqlite-1", action: "keep"} |
CompactionGuidance | 给最终摘要器的“整理要求” | 告诉原生摘要器哪些内容不能忘、哪些可以忽略、哪些还没有完全确定 | Must preserve:PostgreSQL;Can discard:grep;Unresolved:SQLite 原因 |
完整例子:一条“SQLite 被放弃”的信息到底去了哪里?
假设用户和 Agent 之前有下面几条对话。真实项目会有很多消息,这里只截取最能说明问题的 5 条:
# 宿主 Agent 的原始上下文
[user] 目标:实现 OAuth,但不能修改 public API。
[user] 约束:必须保持现有 API 兼容。
[assistant] 我们最终选择 PostgreSQL。
[user] SQLite 因为并发写入锁问题被放弃;不要再次尝试这条路线。
[user] TODO:auth.py 还没有完成 OAuth callback。
[tool] grep -R OAuth src/ → 一些临时搜索输出
第 1 步:宿主准备压缩
上下文变长后,Pi 或 Harness 决定执行 compaction。宿主把“准备被压缩的旧消息”交给适配器。此时还没有删除消息,只是进入审查阶段。
第 2 步:Adapter 把消息翻译成 Core 看得懂的格式
Pi 里的角色可能叫 toolResult,Harness 里的工具消息还包含 block。Adapter 只做“格式翻译”,不在这里决定记不记。
# 归一化后的 ConversationMessage(示意)
[
{"role": "user", "content": "目标:实现 OAuth,但不能修改 public API。", "id": "msg-001"},
{"role": "user", "content": "约束:必须保持现有 API 兼容。", "id": "msg-002"},
{"role": "assistant", "content": "我们最终选择 PostgreSQL。", "id": "msg-003"},
{"role": "user", "content": "SQLite 因并发写入锁问题被放弃。", "id": "msg-004"},
{"role": "tool", "content": "grep -R OAuth src/", "id": "msg-006"}
]
这就是 ConversationMessage 的作用:它不是新消息,也不是摘要,只是给不同宿主消息套上同一件“外套”。
第 3 步:Python Core 把长对话拆成候选记忆
Inspector 会逐条读取内容,必要时按句子拆开,并给每条候选加上类别、重要性、置信度和来源。规则模式尽量保留原文,不凭空创造用户没有说过的结论。
| 候选内容 | 类别 | 重要性 / 置信度 | 来源 |
|---|---|---|---|
| 目标是实现 OAuth,不能修改 public API | goal / constraint | 高 / 高 | msg-001 |
| 必须保持现有 API 兼容 | constraint | 高 / 高 | msg-002 |
| PostgreSQL 是最终选择 | decision | 高 / 高 | msg-003 |
| SQLite 因并发写入锁问题被放弃 | failed_attempt | 中 / 中 | msg-004 |
| auth.py 的 OAuth callback 还未完成 | todo | 高 / 高 | msg-005 |
| grep 的临时搜索输出 | tool_output | 低 / 高 | msg-006 |
这些一条条的候选就是 MemoryCandidate。它存在的意义是:系统可以逐条讨论,而不是对整段对话做一个无法解释的“保留/删除”决定。
第 4 步:Policy 先自动处理确定的,再把边界情况交给人
Policy 看两件事:这条信息有多重要,以及判断有多确定。类别还会提供方向:目标、约束、决定、TODO、失败尝试偏向保留;工具日志和临时输出偏向丢弃。
| 候选 | Policy 结果 | 为什么 |
|---|---|---|
| OAuth 目标、API 约束、PostgreSQL 决定、auth.py TODO | auto Keep | 重要性和置信度都高,且属于关键类别 |
| grep 临时输出 | auto Drop | 工具输出,重要性低且判断很确定 |
| SQLite 失败原因 | Review | 它很可能重要,但当前策略不让系统擅自替用户决定 |
在这个例子里,UI 只需要让用户回答 SQLite 这一条:
# ReviewDecision:用户在宿主 UI 中点击 Keep
{ "candidate_id": "candidate-sqlite", "action": "keep" }
第 5 步:Python Core 生成 Guidance
Core 把自动结果和人工结果合并,生成三段很明确的“给摘要器的要求”:
Must preserve:
- 目标是实现 OAuth,不能修改 public API。
- 必须保持现有 API 兼容。
- PostgreSQL 是最终数据库选择。
- auth.py 的 OAuth callback 还未完成。
- SQLite 因并发写入锁问题被放弃。
Can discard:
- grep 的临时搜索输出。
Unresolved / review carefully:
- (none)
这里的 CompactionGuidance 可以理解成“摘要器的检查清单”。它不是最终摘要,也没有替宿主决定最终段落怎么写。
第 6 步:Bridge 把结果送回宿主,宿主继续原生压缩
如果使用 Provider 模式,Python 发现自己需要模型帮助时,会通过 JSONL 发出 provider_request;宿主用当前模型返回结构化结果。最终 Python 返回 InspectionResult 或 Guidance,Adapter 把 Guidance 加到原生摘要请求的额外指令里。
# 最终结果由宿主自己的 summarizer 写成,例如:
## Project intent
Implement OAuth without changing the public API.
## Decisions and constraints
PostgreSQL is final. SQLite was rejected because concurrent writes caused locking problems.
## Pending work
Finish the OAuth callback in auth.py.
一、整体架构:核心稳定,适配器贴近宿主
项目刻意分成两类代码。Python Core 不知道 Pi 或 DeepSeek Harness 的内部 API;适配器只负责把宿主的消息、模型和交互能力转换成 Core 能理解的接口。
| 层 | 负责什么 | 不负责什么 |
|---|---|---|
| Python Core | 候选记忆抽取、评分、Keep/Drop/Review 策略、Guidance 生成 | 不管理 Agent loop、token window、session 持久化 |
| JSONL Bridge | 把 Python 子进程和宿主 TypeScript 进程连接起来 | 不持有 API Key,不自己选择宿主模型 |
| Pi Adapter | 接 Pi hook,询问 Pi UI,调用 Pi 原生 compact(...) | 不重写 Pi 的摘要格式和 session 事务 |
| DeepSeek Adapter | 装饰 BasicCompactionEngine.summarize(),接 Harness UI 和 ctx.llm | 不重写 Harness 的 compaction range、token accounting、持久化 |
二、Python Core:把模糊的“记住什么”结构化
Core 的公开入口很小,方便任何未来的宿主接入:
guardian = ContextGuardian(provider=None)
result = guardian.inspect(messages)
decisions = [
{"candidate_id": result.review[0].id, "action": "keep"}
]
guidance = guardian.build_guidance(result.candidates, decisions)
2.1 数据模型
source_message_id + category + content 做 SHA-1 截断值,输入不变时 ID 稳定。2.2 Inspector:先保守提取,再交给策略
规则模式会从用户消息、助手消息、工具结果中提取原文候选;它不凭空改写语义。比如出现 must、decision、TODO、abandoned 等信号时,分到对应类别;工具输出通常标成低价值。
Provider 模式则让宿主当前模型输出 CandidateBatch,再由 Pydantic 校验和同一套 Policy 重新归类。模型可以帮助发现原子化候选,但最终是否 Keep/Drop 仍由 Policy 和人来控制。
for message in messages:
for sentence in split_into_atomic_lines(message.content):
candidate = classify_without_inventing_facts(sentence, message.role)
candidate.suggested_action = policy.decide(candidate)
candidates.append(candidate)
2.3 Policy:三种动作和类别优先级
| 条件 | 动作 | 直觉 |
|---|---|---|
| 重要性 ≥ 0.80 且置信度 ≥ 0.80,且属于目标/约束/决定/TODO/失败尝试等保留类 | Keep | 高价值且证据足,自动保留 |
| 属于 tool_output / temporary,且置信度 ≥ 0.80 | Drop | 确定是噪声,自动丢弃 |
| 重要性 ≤ 0.25 且置信度 ≥ 0.80 | Drop | 低价值且判断稳定 |
| 其他情况 | Review | 不确定,不替人做决定 |
类别优先级是覆盖层:goal、constraint、decision、failed_attempt、todo、working_state 等偏向保留;tool_output、temporary 偏向丢弃。阈值本身仍是可解释的数值策略。
2.4 Guidance:不是摘要,而是摘要的“护栏”
Context Guardian review guidance. This is guidance for the native compaction summary, not a replacement summary.
Must preserve:
- PostgreSQL is the final database choice.
- auth.py is still incomplete and needs the OAuth callback implementation.
Can discard:
- grep output from a transient diagnostic command.
Unresolved / review carefully:
- SQLite was abandoned because concurrent writes caused locking problems.
Guidance 的职责是告诉原生 summarizer “哪些事实不可丢、哪些噪声可以忽略、哪些事项仍需谨慎处理”。最终摘要仍由 Pi 或 DeepSeek Harness 自己生成,因此可以保留宿主已有的摘要格式、文件追踪和边界语义。
三、JSONL Bridge:为什么要双向协议
Python Core 需要模型 Provider,但“当前模型”和认证信息属于宿主。于是 TypeScript 进程启动 Python 子进程,通过 stdin/stdout 传递 JSONL。Python 需要模型时,不能自己请求 API,而是发回一个 provider_request,由宿主执行后回传。
协议帧的核心约束
- 版本化:
protocol_version: 1,未来可以演进。 - 可关联:每个请求和 Provider 子请求都有
request_id。 - 可验证:Python 端用 Pydantic 验证模型返回结构。
- 可终止:超时、AbortSignal、子进程退出都会结束等待。
- 有边界:单帧最大 2 MB,避免异常 payload 拖垮进程。
- 可诊断:stdout 不混入普通日志,stderr 单独记录。
# Python HostModelProvider 的伪代码
send({ "type": "provider_request", "request_id": id, "schema": schema })
while true:
frame = read_json_line()
if frame.request_id != id:
raise ProviderError("unmatched response")
if not frame.ok:
raise ProviderError(frame.error)
return schema.validate(frame.data)
modelRegistry 使用;DeepSeek Harness 由 ctx.llm 使用。Key 不通过 JSONL 传递。四、两个适配器:共享意图,但不能共享宿主代码
适配器需要针对各自宿主的生命周期、消息结构、认证方式和 UI 交互方式实现。这不是重复造轮子,而是把宿主差异隔离起来:将来增加 DeepSeek Harness、Claude Code 或其他 harness 时,复用 Core,新增宿主适配器和宿主专属测试。
4.1 Pi Adapter 的路线
- 监听 Pi 的
session_before_compact,读取event.preparation.messagesToSummarize和已有previousSummary。 - 把 Pi 的
toolResult等消息转换成 Core 的ConversationMessage。 - 通过 JSONL 请求 Python Core 检查。Provider 请求回到 Pi 当前模型,通过
completeSimple得到结构化 JSON。 - 自动应用 Keep/Drop;Review 候选逐项调用
ctx.ui.confirm。没有 UI 时按保守 Keep。 - 生成 Guidance,读取当前模型认证,调用 Pi 官方
compact(...),把 Guidance 合并到customInstructions。 - 任意 Guardian 错误都返回
undefined,让 Pi 继续自己的原生 compaction。
const nativeResult = await compact(
event.preparation,
ctx.model,
auth.apiKey,
auth.headers,
event.customInstructions + "\n\n" + guidance.text,
event.signal,
ctx.thinkingLevel,
)
return { compaction: nativeResult }
4.2 DeepSeek Harness Adapter 的路线
- 通过
cordis.patch.yml禁用官方compaction-basicrow,插入 Context Guardian 的ContextGuardianCompactionEngine。 - 继承
BasicCompactionEngine,只覆盖受保护的summarize(input, agent, signal)seam。 - 调用
ctx.llm.stream,读取 Agent 当前 request header 的 provider/model。认证仍由 Harness 内部处理。 - Review 候选通过
ctx.userQuestions.ask进入 Harness 的 Web question composer;没有 answerer 时保守 Keep。 - 把 Guidance 作为额外的 plugin user message 加到原生 summarizer 输入中,再调用
super.summarize(...)。 - Harness 继续负责 compact range、token meter、compaction markers、summary framing 和 session persistence。
protected override async summarize(input, agent, signal) {
try {
const inspection = await bridge.inspect(this.ctx, agent, normalizeInput(input), signal)
const decisions = autoDecisions(inspection) + await review(inspection.review)
const guidance = await bridge.guidance(..., decisions, signal)
return super.summarize(appendGuidance(input, guidance.text), agent, signal)
} catch (error) {
logWarning(error)
return super.summarize(input, agent, signal)
}
}
4.3 为什么不做一个“完全通用的适配器测试”
| 共享测试 | 宿主专属测试 |
|---|---|
| 候选模型校验、规则分类、阈值、稳定 ID、Guidance、JSONL 请求关联 | Pi hook 数据转换、Pi confirm、Pi compact helper、Pi auth registry |
| Provider 合法/非法输出、超时、错误和规则降级 | Harness 的 summarize seam、Cordis patch、ctx.llm、userQuestions UI |
| 不依赖某个宿主的核心回归 | 真实宿主 UI 和原生 compaction 事务的端到端验证 |
五、失败策略:为什么是 fail-open
Context Guardian 是增强层,不应因为自己的 bug 让 Agent 无法继续工作。核心原则是:Guardian 失败,宿主原生 compaction 继续。
Python 未安装、Provider 返回非法 JSON、模型请求失败、超时、桥接进程退出、用户取消、宿主 API 不兼容。
无 UI 时不假装“人已经审过”,而是对未决候选 Keep,然后把结果交给原生摘要器。
try:
inspection = guardian.inspect(...)
decisions = apply_auto_actions(inspection) + human_review(inspection.review)
guidance = guardian.build_guidance(...)
return native_compact(guidance)
except Exception:
log("Context Guardian unavailable; continue native compaction")
return native_compact_without_guidance()
这也解释了为什么测试不能只检查“Guardian 抛错了”:真正的验收标准是,增强层失败后,宿主仍然可以完成原生压缩。
六、如何验证:从快到完整
项目提供多层验证。它们不是互相替代,而是分别定位 Core、协议、宿主装载和人工交互问题。
| 层级 | 入口 | 验证什么 | 是否需要 API Key |
|---|---|---|---|
| Python Core | .venv313/bin/python -m pytest -q | 模型、Inspector、Policy、Guidance、CLI、Provider、Verification | 否 |
| 确定性 fixture | context-guardian verify examples/conversation.json | 关键记忆 retention、噪声 removal、ID 稳定、Guidance 三段 | 否 |
| Pi bridge / adapter | npm run pi-smoke | Pi extension 装载、JSONL bridge、规则降级 | 通常否 |
| Pi 交互 fixture | npm run pi-fixture-smoke | 预置长对话、真实 Pi UI、Keep/Drop、原生 compaction | 使用 Pi 当前登录模型时需要 |
| DeepSeek Harness bridge | npm run test:dsh | 消息归一化、host provider、local guidance | 否,测试使用 replay |
| DeepSeek Harness 交互 fixture | npm run dsh-fixture-smoke | 临时 profile、预置长 session、Harness Web、Review UI、native compaction | 否,使用 replay;需要能启动 DSH Web |
推荐验证顺序
- 先跑确定性 Core 验证,排除规则和 Guidance 问题。
- 再跑对应 adapter 的 bridge/typecheck,排除消息和协议问题。
- 最后运行目标宿主的交互 fixture。Pi 和 DeepSeek Harness 要分别验证,因为它们的 hook/UI contract 不同。
- 人工验收时至少确认:项目目标、API 约束、PostgreSQL、SQLite 放弃原因、未完成的
auth.py被保留;grep、npm install 和已解决临时错误被标为低价值。
七、发布和使用边界
当前可展示的项目亮点
- 不是单纯 Prompt,而是有数据模型、策略、协议、宿主 adapter 和测试矩阵。
- 明确复用宿主模型和认证,不把密钥传给 Python 子进程。
- 保留宿主原生 compaction,避免与 session persistence、token accounting 冲突。
- 有 Human-in-the-loop,也有 headless 下的可解释降级行为。
- 可以用同一 Python Core 扩展新的 harness adapter。
当前需要诚实说明的限制
- 模型输出不是可信输入,必须做 schema 校验和边界控制。
- 人工 Review 会增加一次等待,候选越多,交互成本越高。
- 适配器依赖宿主 API;DeepSeek Harness 当前仍是 developer preview。
- 没有 UI answerer 时,Review 不能伪装成已确认,只能保守 Keep。
- 发布到 PyPI 前要先处理 distribution name 冲突:
context-guardian已被占用,不能直接覆盖,应选择新的包名并同步 README/安装命令。
八、面试官拷打:问题与参考回答
下面的问题刻意偏尖锐,适合用来检查自己是否真的理解了项目,而不是只会复述 README。
1. 你这个项目和“给摘要 Prompt 加几句话”有什么本质区别?
2. 为什么不直接让模型生成最终摘要?这样不是更简单吗?
3. 为什么模型返回了候选后还要再过一遍 Policy?
4. 你为什么不把 API Key 传给 Python,让 Python 自己调用模型?
ctx.llm;如果把 Key 复制给 Python,就会增加泄露面、产生双重配置,也可能绕过宿主的认证刷新和 provider 选择。现在 Python 只发结构化 Provider 请求,真正的模型调用在 TypeScript 宿主侧完成。5. JSONL 为什么比 HTTP 更合适?
6. fail-open 会不会导致 Guardian 失效却没人知道?
7. 为什么 Pi 和 DeepSeek Harness 必须有两套端到端测试?
session_before_compact,UI 是 ctx.ui.confirm,原生调用是 compact(...);Harness 的入口是继承 engine 的 summarize(),UI 是 Cordis user-question answerer,原生调用是 super.summarize()。只有各自宿主的 fixture 才能验证消息转换、认证复用、UI 和 compaction 事务真的接上了。8. 如果用户没有 UI,Review 候选怎么办?
9. 这个项目最大的性能成本是什么?
10. 这个项目现在能不能马上说“已经发布到 PyPI,别人 pip install 就能用”?
context-guardian 这个 distribution name 已被其他项目占用,正式发布前必须换一个未占用的包名,并同步入口文档和安装命令。适配器也要分别按 Pi/Harness 的包管理和版本范围发布,不能把“仓库可运行”直接等同于“两个公共 registry 都已发布”。11. 你会如何证明它真的减少了“错误遗忘”?
12. 如果宿主升级 API,适配器坏了怎么办?
最后用一张图复述
当面试官问“Context Guardian 到底做了什么”时,可以按下面这条链路回答:
重点不是“我写了一个摘要器”,而是“我把摘要前最容易出错的记忆取舍决策,做成了可解释、可交互、可复用且不破坏宿主原生能力的控制层”。