Code Agent 解剖（12）：Harness 设计之二——上下文工程

上一篇的遗留问题

上一篇（第 11 篇）解剖了控制流：单一主循环、不可变状态机、完成门。其中有一行代码被轻轻带过：

# loop.py _prepare_step_context()
model_view = host.context_engine.build_model_view(...)
messages = model_view.messages

这个 build_model_view() 做的事情远不止"取出历史消息"。一个长时间运行的 agent，对话历史会越来越长，token 超限了就报错。更微妙的是：不超限的时候 agent 也要决定——这一步到底把什么送给模型看。系统提示、历史消息、工具输出……什么都塞进去并不比精心筛选更好。

这一篇解剖这条管道：从"发现快撑不住了"到"决定给模型看什么"。

结论先说

MyCodeAgent 把这个问题拆成三个分离的关注点：

关注点	模块	一句话
History 是什么	`HistoryManager`	append-only 事实日志，从不删消息
模型看到什么	`ModelView` + `ProjectionBuilder`	读时投影，压缩后折叠旧消息
什么时候压缩	`ContextBudgetPolicy`	双来源估算 + 阈值决策

三层职责清晰，互不侵入。压缩不等于删除历史，而是换一种"读法"。

一、History 是事实日志，永不删改

# runtime/history.py
class HistoryManager:
    def __init__(self, ...):
        self._messages: List[Message] = []  # append-only，只增不删
    
    def append_user(self, content: str, ...) -> Message: ...
    def append_assistant(self, content: str, ...) -> Message: ...
    def append_tool_result(self, ...) -> Message: ...
    
    def get_messages(self) -> List[Message]:
        return list(self._messages)  # 返回副本，外部无法修改

HistoryManager 是一个纯 append-only 的事实列表。它不知道"压缩"这件事，也不会因为压缩而少一条消息。

这个设计很重要：它意味着 agent 崩溃时，可以从 transcript 完整重建历史，不会因为"压缩把消息删掉了"而丢失上下文（见第 10 篇）。

二、压缩决策：怎么判断"快撑不住了"

# runtime/context/budget.py
class ContextBudgetPolicy:
    def should_compact(self, *, messages, pending_input, last_usage_tokens) -> CompactDecision:
        # 默认：128000 × 0.8 = 102400 tokens
        threshold = int(self.config.context_window * self.config.compression_threshold)
 
        # 来源 1：从消息内容估算（字符数 // 3）
        estimated_from_messages = self.estimate_tokens(messages, pending_input)
        # 来源 2：上一步 LLM 实际返回的 usage token 数 + 新输入
        estimated_from_usage = int(last_usage_tokens or 0) + len(pending_input or "") // 3
        
        # 取最悲观的那个
        estimated = max(estimated_from_messages, estimated_from_usage)
 
        if message_count < 3:
            return CompactDecision(False, "messages_not_enough", ...)
        if estimated < threshold:
            return CompactDecision(False, "below_threshold", ...)
        return CompactDecision(True, "threshold_exceeded", ...)

有两个设计细节值得注意：

1. 为什么不精确计 token？

精确 token 计数需要调用 tokenizer，而 tokenizer 是模型相关的——不同模型规则不同，而且调用有开销。字符数 // 3 是一个保守近似（中英文混排平均值），宁可提前压缩，不能等报错了才反应。

2. 为什么用两个来源取最大值？

单纯从消息列表估算会低估——因为 tool_calls JSON、tool_name 字段等 metadata 也占 token，但不在 content 里。而 last_usage_tokens 是 LLM 上一步实际消耗的数字，更准，但只知道过去，不含当前新输入。取 max 是"悲观估计"的工程习惯：宁可多压，不能不压。

三、压缩怎么做：非破坏性 checkpoint

# runtime/context/compact.py
class ContextCompactor:
    def compact(self, messages: list[Message]) -> dict:
        # 1. 按 user 消息切割轮次
        rounds = self.round_segmenter.identify(source_messages)
        
        # 2. 保留最近 min_retain_rounds 轮（默认 10 轮）原文
        retain_start_round = len(rounds) - min_rounds
        retain_start_idx = rounds[retain_start_round].start_idx
        messages_to_compact = source_messages[:retain_start_idx]
        
        # 3. 对"旧历史"调 LLM 生成摘要
        summary = self.summary_generator(messages_to_compact)
        
        # 4. 把摘要 + 分割点存入 CompactStore
        checkpoint = self.compact_store.create_checkpoint(
            summary=summary,
            retain_start_idx=retain_start_idx,
            ...
        )
        # 注意：source_messages 没有被修改

这里的关键词是 non-destructive：压缩不修改 HistoryManager 里的任何消息，只是在 CompactStore 里记录了一个 checkpoint，写明"摘要内容是什么"、"从哪个 index 开始是原文"。

RoundSegmenter 负责把消息列表按 user 消息边界切成"轮次"，这是因为压缩的粒度是完整的对话轮——不能把一轮对话的 user 和 assistant 分开压缩，那样语义会断掉。

四、读时投影：模型看到的不等于存的

# runtime/context/projection.py
class ProjectionBuilder:
    def project(self, source_messages: list[Message]) -> ProjectionResult:
        checkpoint = self.compact_store.active_checkpoint
        
        if not checkpoint:
            # 还没压缩：投影 = 全量历史
            return ProjectionResult(messages=source, projection_mode="full_history")
        
        # 有 checkpoint：投影 = [摘要消息] + source[retain_start_idx:]
        summary = Message(content=checkpoint.summary, role="summary", ...)
        return ProjectionResult(
            messages=[summary] + source[retain_start_idx:],
            projection_mode="compact_checkpoint",
        )

这是整个上下文工程最核心的一步：读时投影。

source_messages 有 200 条，压缩 checkpoint 的 retain_start_idx 是 120，那么投影结果就是：

[summary（前 120 条的摘要）] + [source[120:] 即最近 80 条原文]

模型看到的是 81 条（1 条摘要 + 80 条原文），但 HistoryManager 里依然有 200 条完整历史。

五、MessageNormalizer：格式转换

History 里存的是 Message 对象（runtime 内部格式），LLM API 要的是 OpenAI 风格的 dict 列表。MessageNormalizer 负责这个转换：

# runtime/context/normalizer.py
class MessageNormalizer:
    def _normalize_one(self, msg: Message) -> list[dict]:
        if msg.role == "user":
            return [{"role": "user", "content": msg.content}]
        if msg.role == "assistant":
            return [self._assistant_message(msg)]  # 含 tool_calls 还原
        if msg.role == "tool":
            return [self._tool_message(msg)]        # 含 tool_call_id 关联
        if msg.role == "summary":
            # 摘要变成 system 消息，放在 history 最前
            return [{"role": "system", "content": f"## Archived History Summary\n{msg.content}"}]

summary role 的消息会被转成 system role 发给模型，这样模型就把它当作"背景信息"而不是对话历史来处理——这是一个语义上的巧妙设计。

六、ModelView：这轮模型看到了什么

所有前面的步骤汇聚到 ContextEngine.build_model_view()，产出一个 ModelView：

# runtime/context/engine.py（build_model_view 核心逻辑）
def build_model_view(self, *, history_manager, pending_input, ...) -> ModelView:
    # 1. History 有界投影（projection + normalize）
    source_messages = history_manager.get_messages()
    projection = self.projection_builder.project(source_messages)
    history_messages = self.normalizer.normalize(projection.messages)
 
    # 2. System 层（agent 人格 + 工具说明 + 项目规则）
    system_messages = self.context_builder.get_system_messages()
 
    # 3. 动态 system：Session Memory（跨 run 摘要）
    dynamic_messages = [{"role": "system", "content": rendered}] if self.session_memory else []
 
    # 4. 拼接：system 在前，history 在后
    messages = list(system_messages) + dynamic_messages + list(history_messages)
 
    return ModelView(
        messages=messages,
        system_message_count=len(system_messages),
        history_message_count=len(history_messages),
        source_message_count=projection.source_message_count,
        projection_mode=projection.projection_mode,
        ...
    )

最终送给模型的消息顺序：

[Constitution][Tool Contracts][Code Law]  ← system 层（第 04 篇）
[Session Memory?]                          ← 跨 run 摘要（第 10 篇）
[summary? + 最近 N 轮原文]                 ← history 投影

ModelView 不只是消息列表，还带着元数据——projection_mode（是否压缩过）、source_message_count vs history_message_count（压缩前后条数对比）——这些是给 trace 用的，让你能观察到"这步模型实际看了多少东西"。

七、两条触发路径

loop 里有两个地方会触发压缩：

每步开始（主动）          模型调用时（被动）
       ↓                         ↓
compact_if_needed()       分类异常 → PROMPT_TOO_LONG
  → budget 估算超阈值       → reactive_compact()
  → compactor.compact()    → compactor.compact()
  → 记录 checkpoint         → 重建 model view，内层 continue 重试

主动压缩（compact_if_needed）在每步开始、构建 model view 之前运行。估算发现快撑不住了，先压再组装，下一步模型就看压缩后的版本。

被动压缩（reactive_compact）是兜底。万一估算没触发但模型真的报了 PROMPT_TOO_LONG，立即强制压缩，重建 model view，continue 重试当前步骤。有重试次数上限，耗尽后走 MODEL_RECOVERY_FAILED 终止路径。

设计亮点

1. History 与 ModelView 分离

HistoryManager 是不可变的事实日志，ModelView 是每步临时构造的"模型视图"。两者分离意味着压缩不会丢历史，崩溃可以恢复，调试时也能对比"存了什么"和"模型看到了什么"。

2. 读时投影而非写时删除

压缩结果存成 checkpoint（summary + 分割点），每次读 history 时动态折叠。这比"压缩时删消息"更安全，可以随时回滚（清除 checkpoint 就回到全量历史），也不会因为压缩 bug 导致历史损坏。

3. 双路保险

主动估算 + 被动捕获异常，两条路都能触发压缩。即便估算算法有误差，模型报错后也能自愈而不是直接崩溃。

小结

设计选择	方案	工程价值
History 存储	append-only 事实日志	压缩不丢历史，崩溃可完整恢复
压缩方式	读时投影 + checkpoint	可回滚，不因压缩 bug 损坏历史
token 估算	双来源取最大值	悲观估计，宁早触发不晚触发
触发路径	主动估算 + 被动捕获异常	估算失效时有兜底，自愈而不崩溃
ModelView	消息列表 + 可观测元数据	调试时能对比"存了什么"和"模型看到了什么"

关于本系列的源码

本系列所有分析均基于开源项目 MyCodeAgent。

源码里已经按照本系列文章的讲解顺序，在关键位置加入了配套注释——读文章时可以对照代码，也可以直接克隆下来自己跑、改、扩展，基于它开发你自己的 agent。

git clone https://github.com/chendongqi/MyCodeAgent
cd MyCodeAgent
cp .env.example .env   # 填入你的 LLM API key
uv sync
uv run python main.py

欢迎访问 PrimeSkills —— 一个精心策划的 AI Agent 与技能市场，所有内容均经过真实企业级工作流验证。没有噱头，只有真正有效的东西。

更多实用知识和有趣产品，欢迎访问我的个人主页