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

上一篇的遗留问题

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

model_view = host.context_engine.build_model_view(...)
messages = model_view.messages

这个 build_model_view() 做的事情远不止“取出历史消息”。对于一个长时间运行的 Agent，对话历史会越来越长；一旦 token 超限，就会报错。更微妙的是：即使尚未超限，Agent 也必须决定——这一步究竟该把哪些内容交给模型查看。系统提示、历史消息、工具输出……把所有内容一股脑塞进去，并不一定比精心筛选更好。

本文将解剖这条管道：从“发现上下文快撑不住了”，到“决定让模型看到什么”。

---

结论先说

MyCodeAgent 将这个问题拆分为三个相互分离的关注点：

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

三层职责清晰，彼此不侵入。压缩不等于删除历史，而是换一种“读取方式”。

---

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

class HistoryManager:
    def __init__(self, ...):
        self._messages: List[Message] = []  
    
    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 篇）。

---

二、压缩决策：如何判断“快撑不住了”

class ContextBudgetPolicy:
    def should_compact(self, *, messages, pending_input, last_usage_tokens) -> CompactDecision:
        
        threshold = int(self.config.context_window * self.config.compression_threshold)

        
        estimated_from_messages = self.estimate_tokens(messages, pending_input)
        
        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

class ContextCompactor:
    def compact(self, messages: list[Message]) -> dict:
        
        rounds = self.round_segmenter.identify(source_messages)
        
        
        retain_start_round = len(rounds) - min_rounds
        retain_start_idx = rounds[retain_start_round].start_idx
        messages_to_compact = source_messages[:retain_start_idx]
        
        
        summary = self.summary_generator(messages_to_compact)
        
        
        checkpoint = self.compact_store.create_checkpoint(
            summary=summary,
            retain_start_idx=retain_start_idx,
            ...
        )

这里的关键词是 non-destructive（非破坏性）：压缩不会修改 HistoryManager 中的任何消息，而只是在 CompactStore 中记录一个 checkpoint，标明“摘要内容是什么”以及“从哪个 index 开始保留原文”。

RoundSegmenter 负责按照 user 消息的边界，将消息列表切分为“轮次”。这是因为压缩的粒度必须是一整轮对话——不能将一轮对话中的 user 消息和 assistant 消息拆开压缩，否则语义会断裂。

---

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

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")
        
        
        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 负责这一转换：

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)]  
        if msg.role == "tool":
            return [self._tool_message(msg)]        
        if msg.role == "summary":
            
            return [{"role": "system", "content": f"## Archived History Summary\n{msg.content}"}]

summary role 的消息会被转换为 system role 并发送给模型。这样模型会将其视为“背景信息”，而不是普通对话历史——这是一种语义层面的巧妙设计。

---

六、ModelView：本轮模型实际看到了什么

前述所有步骤最终汇聚到 ContextEngine.build_model_view()，并产出一个 ModelView：

def build_model_view(self, *, history_manager, pending_input, ...) -> ModelView:
    
    source_messages = history_manager.get_messages()
    projection = self.projection_builder.project(source_messages)
    history_messages = self.normalizer.normalize(projection.messages)

    
    system_messages = self.context_builder.get_system_messages()

    
    dynamic_messages = [{"role": "system", "content": rendered}] if self.session_memory else []

    
    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 与 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   
uv sync
uv run python main.py

上一篇的遗留问题

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

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 是事实日志，永不删改

class HistoryManager:
    def __init__(self, ...):
        self._messages: List[Message] = []  
    
    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 篇）。

---

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

class ContextBudgetPolicy:
    def should_compact(self, *, messages, pending_input, last_usage_tokens) -> CompactDecision:
        
        threshold = int(self.config.context_window * self.config.compression_threshold)

        
        estimated_from_messages = self.estimate_tokens(messages, pending_input)
        
        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

class ContextCompactor:
    def compact(self, messages: list[Message]) -> dict:
        
        rounds = self.round_segmenter.identify(source_messages)
        
        
        retain_start_round = len(rounds) - min_rounds
        retain_start_idx = rounds[retain_start_round].start_idx
        messages_to_compact = source_messages[:retain_start_idx]
        
        
        summary = self.summary_generator(messages_to_compact)
        
        
        checkpoint = self.compact_store.create_checkpoint(
            summary=summary,
            retain_start_idx=retain_start_idx,
            ...
        )

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

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

---

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

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")
        
        
        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 负责这个转换：

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)]  
        if msg.role == "tool":
            return [self._tool_message(msg)]        
        if msg.role == "summary":
            
            return [{"role": "system", "content": f"## Archived History Summary\n{msg.content}"}]

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

---

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

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

def build_model_view(self, *, history_manager, pending_input, ...) -> ModelView:
    
    source_messages = history_manager.get_messages()
    projection = self.projection_builder.project(source_messages)
    history_messages = self.normalizer.normalize(projection.messages)

    
    system_messages = self.context_builder.get_system_messages()

    
    dynamic_messages = [{"role": "system", "content": rendered}] if self.session_memory else []

    
    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   
uv sync
uv run python main.py

---

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

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