Context Engineering 整理

1. 上下文工程简介

上下文工程的概念可以参考 Anthropic 下面这段话：

Context engineering refers to the set of strategies for curating and maintaining the optimal set of tokens (information) during LLM inference, including all the other information that may land there outside of the prompts. —— Anthropic, Effective context engineering for AI agents

当上下文里的信号密度下降时，模型的行为会从“基于事实推理”滑向“基于上下文里最近出现的东西凑答案”。LangChain 把它归纳为 context poisoning / context distraction / context confusion；Anthropic 则称之为 context rot —— 窗口越长，模型准确召回关键信息的能力越差。因此，对上下文的管理就显得很必要了。

LangChain 将上下文管理拆成四个动作：Write / Select / Compress / Isolate，分别对应“把信息写到窗外”“按需选进来”“压缩已经在窗内的”“隔离到不同窗口”。下面按照这个结构，结合这段时间项目开发的经历对上下文管理的方法设计整理一下。

2. llm 摘要的问题所在

最直接的直觉是直接调用 llm 进行摘要。但是“只做摘要”有下面的问题：

• 摘要本身也是一次 LLM 调用，非确定、可能失败、可能丢关键细节；
• 摘要一旦出错，错误会被锁死，后续所有轮次都基于一个被污染的快照继续；
• 工具调用的协议完整性容易被破坏：OpenAI-compatible API 要求 role=tool 的消息必须紧跟在声明了对应 tool_call_id 的 assistant 消息之后，乱删一条就会触发协议错误；
• 摘要无法溯源：业务场景里“这个结论从哪条规范来的”往往比结论本身更重要，摘要会把来源行号、版本、hash 这些信息磨平。

因此，摘要比起上下文管理的主要方法、更适合作为一个可选的压缩层、放在更底层的确定性防线之后。

3. 上下文管理的三道防线

根据一些开源的 Agent Context 相关文档，可以把上下文管理拆分成三个层次：

层数	核心问题	对应 LangChain 策略	主要手段
第一层	能不能不塞进上下文？	Select	just-in-time 检索、Artifact 引用
第二层	必须塞进去的，怎么放才不会炸？	Compress	microcompact、协议感知压缩、evidence ledger、effective window
第三层	能不能把状态搬到窗外？	Write / Isolate	工作流状态机、加密版本化 Memory、CAS 切换

下面逐层展开。

$a.$ SELECT：将能放在外部的内容放在外部

有一些内容，比如打算提供给模型的外部知识，可以考虑不直接将所有内容塞给模型，而是给它一张目录或者一些检索工具，让它根据目录选择自己需要的章节去读/调用检索工具去获取内容。这就是 SELECT 层的核心思想。下面分别介绍一下这两种形态。

$i.$ just-in-time 检索

import re
from pathlib import Path

class DocumentReader:
  """一个通用的四级文档读取接口。

  核心约定：小文档 resolve 直接返回正文；大文档 resolve 只返回
  章节列表，迫使 agent 用 section 读取具体章节，避免大段正文
  一次性涌入上下文。
  """

  MAX_DIRECT_CHARS = 12000  # 超过这个阈值就不直接返回正文

  def __init__(self, docs_dir: Path):
    self.docs_dir = docs_dir

  def catalog(self) -> list[dict]:
    """L1：只返回文档目录，不读正文。"""
    return [
      {"id": path.stem, "title": self._title(path), "size": path.stat().st_size}
      for path in sorted(self.docs_dir.glob("*.md"))
    ]

  def resolve(self, doc_id: str) -> dict:
    """L2：返回元数据；小文档直接带正文，大文档只带章节列表。"""
    content = self._load(doc_id)
    sections = self._extract_headings(content)

    result = {
      "id": doc_id,
      "title": self._title(self.docs_dir / f"{doc_id}.md"),
      "sections": sections,
    }

    # 小文档直接返回全文，大文档要求 agent 用 section 读
    if len(content) <= self.MAX_DIRECT_CHARS:
      result["content"] = content
      result["requires_section"] = False
    else:
      result["requires_section"] = True
      result["hint"] = (
        f"Document is {len(content)} chars, larger than direct-read limit "
        f"{self.MAX_DIRECT_CHARS}. Use action='section' with one heading "
        f"or a hierarchical path like 'Parent > Child'."
      )

    return result

  def section(self, doc_id: str, selector: str) -> dict:
    """L3：只读一个章节。

    selector 支持：
    - 完整标题："错误码"
    - 层级路径："API 参考 > 错误码"
    - 唯一关键词："错误"（仅当只命中一个标题时）
    """
    content = self._load(doc_id)
    selected = self._extract_section(content, selector)
    if selected is None:
      return {
        "id": doc_id,
        "error": "SECTION_NOT_FOUND",
        "message": f"No section matches {selector!r}",
      }
    return {"id": doc_id, "section": selector, "content": selected}

  def search(self, query: str, limit: int = 5) -> list[dict]:
    """L4：全文检索，返回命中片段 + 来源位置。"""
    hits = []
    for path in sorted(self.docs_dir.glob("*.md")):
      content = path.read_text(encoding="utf-8")
      current_section = ""
      for line_no, line in enumerate(content.splitlines(), start=1):
        if line.startswith("#"):
          current_section = line.lstrip("#").strip()
        if query.lower() in line.lower():
          hits.append({
            "doc_id": path.stem,
            "section": current_section,
            "line": line_no,
            "excerpt": line.strip()[:200],
          })
    return hits[:limit]

  # ---------- helpers ----------
  def _load(self, doc_id: str) -> str:
    return (self.docs_dir / f"{doc_id}.md").read_text(encoding="utf-8")

  def _title(self, path: Path) -> str:
    first = path.read_text(encoding="utf-8").splitlines()[0]
    return first.lstrip("#").strip() if first.startswith("#") else path.stem

  def _extract_headings(self, content: str) -> list[dict]:
    """返回章节列表，保留层级，方便 agent 选择。"""
    return [
      {"level": len(match.group(1)), "title": match.group(2).strip()}
      for match in re.finditer(r"^(#{1,6})\s+(.+?)\s*$", content, re.MULTILINE)
    ][:80]  # 最多返回 80 个章节，防止章节列表本身过长

  def _extract_section(self, content: str, selector: str) -> str | None:
    wanted = [part.strip() for part in selector.split(">") if part.strip()]
    headings = [
      (len(match.group(1)), match.group(2).strip(), match.start())
      for match in re.finditer(r"^(#{1,6})\s+(.+?)\s*$", content, re.MULTILINE)
    ]

    # 先尝试严格匹配（层级路径或完整标题）
    stack: list[str] = []
    for index, (level, title, start) in enumerate(headings):
      stack = stack[: level - 1] + [title]
      if stack != wanted and title != selector.strip():
        continue
      end = next(
        (next_start for next_level, _, next_start in headings[index + 1 :] if next_level <= level),
        len(content),
      )
      return content[start:end].strip()

    # 回退：唯一的关键词包含匹配
    simple = selector.strip()
    if simple and ">" not in simple:
      matches = [item for item in headings if simple.casefold() in item[1].casefold()]
      if len(matches) == 1:
        level, _, start = matches[0]
        index = headings.index(matches[0])
        end = next(
          (next_start for next_level, _, next_start in headings[index + 1 :] if next_level <= level),
          len(content),
        )
        return content[start:end].strip()

    return None

然后把这组能力暴露成一个模型可调用的工具：

read_document = {
  "name": "read_document",
  "description": (
    "Read from the document library. "
    "Use catalog to discover documents, resolve to get metadata "
    "(small docs return full content, large docs return a section list), "
    "section to read a specific heading, and search to find snippets."
  ),
  "parameters": {
    "type": "object",
    "required": ["action"],
    "properties": {
      "action": {
        "enum": ["catalog", "resolve", "section", "search"],
        "description": "Which read mode to use."
      },
      "doc_id": {"type": "string"},
      "selector": {"type": "string"},
      "query": {"type": "string"},
      "limit": {"type": "integer", "default": 5},
    }
  }
}

下面是两个调用这个工具的简单例子：

user: 怎么刷新 token？
assistant: [调用 read_document: action=catalog]
tool: [{"id": "api-ref", "title": "API 参考手册", "size": 8192}, ...]
assistant: [调用 read_document: action=resolve, doc_id=api-ref]
tool: {"title": "API 参考手册", "requires_section": false,
       "content": "## 认证\n调用 /auth/refresh...\n## 错误码\n401..."}
assistant: 调用 /auth/refresh 接口，传入旧的 access token...

user: 401 错误一般怎么处理？
assistant: [调用 read_document: action=catalog]
tool: [{"id": "api-ref", "title": "API 参考手册", "size": 123456}, ...]
assistant: [调用 read_document: action=resolve, doc_id=api-ref]
tool: {"title": "API 参考手册", "requires_section": true,
       "sections": [
         {"level": 1, "title": "API 参考手册"},
         {"level": 2, "title": "认证"},
         {"level": 2, "title": "用户接口"},
         {"level": 3, "title": "列表用户"},
         {"level": 3, "title": "创建用户"},
         {"level": 2, "title": "错误码"}
       ],
       "hint": "Document is 123456 chars... Use action='section' with one heading or path like 'API 参考手册 > 错误码'."}
assistant: [调用 read_document: action=section, doc_id=api-ref, selector="错误码"]
tool: {"section": "错误码", "content": "## 错误码\n\n401 Unauthorized: token 过期或无效。请调用 /auth/refresh 刷新..."}
assistant: 401 通常表示 token 过期，需要调用 /auth/refresh 刷新...

这里的阈值设计（MAX_DIRECT_CHARS）是 SELECT 策略的关键细节：小文档直接返回，减少一次往返；大文档强制分章节，避免一次性把十几万字符砸进上下文。section 同时支持完整标题、层级路径和唯一关键词回退，让 agent 不必精确记住标题也能读到内容。

这就是 Anthropic 说的 just-in-time context retrieval：模型先拿到的是轻量级引用（目录、章节大纲、查询摘要），只有在真正需要时才通过工具把具体内容拉进上下文。Claude Code 用 glob/grep/head/tail 做类似的事；这里则是把“读文档”简单抽象成四级渐进接口。

$ii.$ Artifact 引用

SELECT 的另一种形态是：工作流中间产物不经过模型上下文，只把引用传给它。

以一个报告生成工作流为例：

1. extract_requirements：从用户输入里提取需求；
2. collect_data：按需求去数据库/文件里取数据；
3. compute_chart_data：把原始数据聚合成图表数据；
4. generate_report：生成最终报告。

如果每一步都把完整结果塞进 prompt 让下一步处理，上下文很快会被大段 JSON/CSV/中间表格撑爆。更关键的是，模型可能在搬运过程中改写中间产物，比如把计算好的数值四舍五入、把字段名改成它以为更好的名字、丢掉一些它觉得不重要的列。

这个问题的解决方式是：将中间产物全部持久化为 Artifact，模型只拿到一个轻量级引用。这首先是一种 SELECT 策略：避免大段中间产物涌入当前 prompt；而当这些 Artifact 需要跨轮存活、甚至会话重启后仍可恢复时，就进入了后面要讲的 WRITE / ISOLATE 范畴。

from dataclasses import dataclass, field
from datetime import datetime, timezone
import hashlib
import json
from typing import Any


@dataclass
class Artifact:
  artifact_id: str
  task_id: str
  step_id: str
  kind: str           # 例如 "requirements", "raw_data", "chart_data", "report"
  media_type: str
  schema_id: str      # 例如 "report.requirements.v1"
  payload: Any        # 实际数据，可能很大
  producer: str       # 哪个 step/skill 产生的
  created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

  @property
  def checksum(self) -> str:
    return hashlib.sha256(json.dumps(self.payload, sort_keys=True, ensure_ascii=False).encode()).hexdigest()[:16]

  @property
  def size(self) -> int:
    return len(json.dumps(self.payload, ensure_ascii=False).encode("utf-8"))


def artifact_reference(artifact: Artifact) -> dict[str, Any]:
  """返回一个安全的 Artifact 引用，适合放进模型上下文。"""
  return {
    "artifact_id": artifact.artifact_id,
    "task_id": artifact.task_id,
    "step_id": artifact.step_id,
    "kind": artifact.kind,
    "media_type": artifact.media_type,
    "schema_id": artifact.schema_id,
    "checksum": artifact.checksum,
    "size": artifact.size,
    "producer": artifact.producer,
    "created_at": artifact.created_at,
    # 注意：没有 payload
  }

这样模型看到的就只有下面的 JSON 了：

{
  "artifact_id": "art_01H...",
  "step_id": "collect_data",
  "kind": "raw_data",
  "schema_id": "report.raw_data.v1",
  "checksum": "a3f7e2d9...",
  "size": 45230,
  "producer": "data-collector"
}

而不是 45KB 的原始数据本身。

对于可以完全由 Runtime 确定性计算出来的阶段（比如 collect_data → compute_chart_data），甚至不需要模型参与。我们只需要定义一个 projection / adapter：读取上一步 Artifact，用确定性代码把它投影成下一步 Artifact，然后把新引用写回状态。

class ArtifactProjection:
  """把一个 Artifact 投影成下一个 Artifact，不经过 LLM。"""

  def project_chart_data(self, raw_data_artifact: Artifact) -> Artifact:
    raw = raw_data_artifact.payload
    # 确定性聚合：分组、求和、生成图表数据
    chart_data = {
      "labels": [row["date"] for row in raw],
      "series": [row["value"] for row in raw],
    }
    return Artifact(
      artifact_id=new_id(),
      task_id=raw_data_artifact.task_id,
      step_id="compute_chart_data",
      kind="chart_data",
      media_type="application/json",
      schema_id="report.chart_data.v1",
      payload=chart_data,
      producer="chart-data-projection",
    )

  def project_report(self, chart_data_artifact: Artifact) -> Artifact:
    chart_data = chart_data_artifact.payload
    # 确定性渲染：把图表数据渲染成报告结构
    report = {
      "title": "月度数据报告",
      "sections": [{"type": "chart", "data": chart_data}],
    }
    return Artifact(
      artifact_id=new_id(),
      task_id=chart_data_artifact.task_id,
      step_id="generate_report",
      kind="report",
      media_type="application/json",
      schema_id="report.final.v1",
      payload=report,
      producer="report-projection",
    )

这些 projection 可以组装在 Artifact 处理管线中，模型完全不需要负责传入和读取，它只在需要语义理解或做决策时才会读取这些内容。

SELECT 与后面的 WRITE/ISOLATE 的边界听起来都是“把东西搬出上下文”，它们间的区别是：

• SELECT 关注的是“这一轮请求里，哪些内容本来就不该塞进 prompt”，比如外部知识库、当前用不到的中间产物；
• WRITE/ISOLATE 关注的是“跨轮甚至跨会话还需要保留的状态，能不能不放在窗口里”，比如工作流阶段产物、历史摘要。

前面的 Artifact 引用正好踩在这条边界上：只给引用不给正文，是 SELECT；把产物写到外部 store 并跨轮引用，是 WRITE。当它只在当前任务内按需读取时，就在 SELECT 的相关范围；当它会话中断后仍需恢复时，就进入了 WRITE/ISOLATE 的范畴。

$b.$ COMPACT：对上下文本身的处理

SELECT 把能搬出窗口的内容搬到了外面，但窗口里仍然必须保留系统提示、用户指令、工具调用历史和最近的 observation。这些内容需要还进一步压缩和治理。这整个过程通常会按这个顺序执行：

1. Cheap wins：先去掉重复/被取代的内容（microcompact）；
2. 结构压缩：以最小协议单元驱逐过期消息，被驱逐的内容收敛成 evidence ledger；
3. 语义压缩：必要时使用 LLM 做摘要，并设置熔断 fallback；
4. 前缀稳定：治理 system prompt 和消息排列，让稳定前缀享受 prompt caching。

如果走完这一步仍然超出模型的上下文窗口，就说明问题不在“窗内怎么放”，而在“能不能不放在窗内”——这正是第三层 WRITE/ISOLATE 要处理的事。

$i.$ microcompact

一个很常见的情况是模型反复读取同一个文件、反复 grep 同一个 pattern，对于这些操作，older result 除了证明“我曾经做过某些事情”之外已经没有价值。为了删掉这些冗余的消息同时不破坏消息的格式，可以按“逻辑调用键”把旧结果原地替换成占位符，不删除具体条目和消息顺序。比如 nonoka 框架的处理：

# nonoka/core/memory.py:393-435
def microcompact_superseded_tool_results(...):
  """Replace superseded duplicate tool results with a short placeholder.

  Repeating the same logical tool call (e.g. Read on the same file) makes
  every result but the newest stale. Older contents are replaced in place —
  entries are never removed — so assistant/tool protocol units stay intact.
  """
  ...
  result = list(entries)
  for index, key in enumerate(keys):
    if key is None or latest[key] == index:
      continue
    entry = result[index]
    placeholder = f"[superseded by newer {key[0]} result]"
    metadata = dict(entry.metadata)
    metadata["superseded"] = True
    result[index] = entry.model_copy(update={
      "content": placeholder,
      "tokens": count_tokens(placeholder),
      "metadata": metadata,
    })
  return result

这一步跑在真正的压缩之前，很多时候只凭它就能把上下文压回预算内，避免触发更重的摘要或滑动窗口。

$ii.$ 协议感知压缩与 evidence ledger

OpenAI-compatible API 要求 role=tool 消息必须跟在声明了对应 tool_call_id 的 assistant 消息之后。如果滑动窗口只删了 assistant 消息而留下 tool 结果，下一轮请求会直接报错。常见的处理方法是把“一次 assistant tool_calls + 它声明的所有 tool 响应”当成最小驱逐单元：

@classmethod
def _is_complete_unit(cls, entries: list[MemoryEntry], start: int) -> bool:
  entry = entries[start]
  calls = cls._tool_calls(entry)
  if entry.role != MemoryRole.ASSISTANT or not calls:
    return True
  expected = {str(call.get("id") or call.get("tool_call_id")) for call in calls if call.get("id") or call.get("tool_call_id")}
  actual: set[str] = set()
  index = start + 1
  while index < len(entries) and entries[index].role == MemoryRole.TOOL:
    tool_call_id = entries[index].metadata.get("tool_call_id")
    if tool_call_id:
      actual.add(str(tool_call_id))
    index += 1
  return not expected or expected.issubset(actual)

_evict 的时候不是随便删，而是整组删除：

def _pop_protocol_unit(entries: list[MemoryEntry], start: int = 0) -> list[MemoryEntry]:
  """Remove one chat exchange without orphaning tool results.

  A tool result can only be sent to a provider after the assistant message
  that declared its ID, so an assistant tool-call message and its contiguous
  results must be trimmed together.
  """
  result = list(entries)
  if start >= len(result):
    return result

  entry = result[start]
  tool_calls = getattr(entry, "tool_calls", None) or []
  if entry.role != MemoryRole.ASSISTANT or not tool_calls:
    # 普通消息：单独删除
    return result[:start] + result[start + 1:]

  # assistant + tool_calls：整组删除，直到不属于该 tool_calls 的 tool 消息为止
  ids_to_remove = {
    str(c.get("id") or c.get("tool_call_id"))
    for c in tool_calls
    if c.get("id") or c.get("tool_call_id")
  }
  end = start + 1
  while end < len(result) and result[end].role == MemoryRole.TOOL:
    call_id = str(result[end].metadata.get("tool_call_id") or "")
    if call_id in ids_to_remove:
      end += 1
    else:
      break
  return result[:start] + result[end:]

直接删除旧工具结果会丢失“我做过什么、得到什么”的审计线索。nonoka 的折中办法是：把被驱逐的内容收敛成一条有界的 evidence ledger，而不是直接丢掉。假设 agent 正在处理一个任务，当前上下文是这样的：

system: 你是一个 coding assistant
user: 帮我看看这个项目的配置文件
assistant: [调用 read_file: config.yaml]
tool: <5000 tokens 的完整文件内容>
assistant: [调用 grep: pattern="DATABASE"]
tool: <800 tokens 的匹配结果>
user: 数据库端口是多少？

如果此时上下文快满了，最粗暴的做法是直接把最早的 tool 结果删掉。但删掉之后，当用户问“数据库端口是多少”时，agent 会失去“我读过 config.yaml”这条线索，可能再读一次文件，或者更糟——瞎编一个答案。而把被驱逐的 tool 调用收敛成一条 evidence ledger、仍然以 system 消息的形式留在上下文里的结果如下：

system: 你是一个 coding assistant
system: [Compacted evidence ledger]
        [
          {"tool": "read_file", "arguments": {"path": "config.yaml"},
           "result": "server:\n  host: 0.0.0.0\n...[omitted]...\n  port: 5432"},
          {"tool": "grep", "arguments": {"pattern": "DATABASE"},
           "result": "DATABASE_URL=postgresql://...\n...[omitted]..."}
        ]
user: 数据库端口是多少？

这样 agent 仍然知道：前面读过 config.yaml，文件里数据库端口是 5432，前面还做过一次 grep DATABASE。它不需要重新读文件，就能直接回答用户的问题。

具体的抽取方法可以根据具体任务来设计，这里采用简单的截取前后部分内容实现：遍历被删除的条目，用 tool_call_id 把 assistant 的调用和 tool 的响应配对，然后提取工具名、参数、结果预览、退出码、工作区变更和 artifact 引用。内容超过 800 token 的会用首尾截断，避免一条大结果把 ledger 撑爆：

@classmethod
def _ledger_items(cls, removed: list[MemoryEntry]) -> list[dict[str, Any]]:
  items: list[dict[str, Any]] = []
  call_args: dict[str, dict[str, Any]] = {}  # call_id -> {tool, arguments}
  for entry in removed:
    if entry.role == MemoryRole.ASSISTANT:
      # 记录 assistant 想调用什么工具、传了什么参数
      for call in cls._tool_calls(entry):
        call_id = str(call.get("id") or call.get("tool_call_id") or "")
        function = call.get("function") if isinstance(call.get("function"), dict) else {}
        args = function.get("arguments")
        if isinstance(args, str):
          try:
            args = json.loads(args)
          except json.JSONDecodeError:
            args = {"raw": args[:500]}
        call_args[call_id] = {
          "tool": function.get("name"),
          "arguments": args if isinstance(args, dict) else {},
        }
    elif entry.role == MemoryRole.TOOL:
      # 用 call_id 把调用和响应配对，生成一条审计项
      call_id = str(entry.metadata.get("tool_call_id") or "")
      item = dict(call_args.get(call_id, {}))
      item.update({
        "tool_call_id": call_id or None,
        "artifact_ref": entry.metadata.get("artifact_ref"),
        "exit_code": entry.metadata.get("exit_code"),
        "workspace_changes": entry.metadata.get("workspace_changes"),
        "result": cls._preview(entry.content),
      })
      items.append(item)
    elif entry.role in {MemoryRole.USER, MemoryRole.ASSISTANT} and entry.content:
      # 普通对话只保留角色 + 内容预览
      items.append({"role": entry.role.value, "content": cls._preview(entry.content)})
  return items

这样生成的一条 ledger 项如下：

{
  "tool": "read_file",
  "arguments": { "path": "config.yaml" },
  "tool_call_id": "call_01xxx",
  "exit_code": 0,
  "artifact_ref": null,
  "workspace_changes": null,
  "result": "server:\n  host: 0.0.0.0\n...[omitted]...\n  port: 5432"
}

如果 ledger 拼好后超过 12 KB，就从最旧项开始丢弃；如果连清空 ledger 都救不回预算，就直接不要 ledger。但注意：ledger 只是做了审计摘要、没有破坏 llm message 的结构，协议完整性在前面 _pop_protocol_unit 那一步已经保证——即使 ledger 被清空，assistant/tool 的配对也不会乱。

$iii.$ 上下文窗口预算管理

上下文的预算管理不应该是简单的“到达 max_tokens 再砍”，而是应该把窗口分成三块来管理：已经占用的上下文、提前触发的缓冲带、留给模型输出的空间。

举个例子，假设配置是：

• max_tokens = 8192：模型上下文窗口总上限；
• reserve_output_tokens = 4096：每次调用模型时，必须预留出 4096 token 给它生成回复；
• compaction_buffer_tokens = 2048：在接近上限前 2048 token 就开始压缩，不然如果等撑满上下文窗口时才压缩的话，模型可能正在生成长回复，中途发现窗口不够，导致输出被截断或报错；

那么对应的三个区域如下：

0                      4096                    6144               8192 tokens
 ┌───────────────────────┬───────────────────────┬─────────────────────┐
 │  Target Context Size  │            Reserved Output Space            │
 │     (0 ~ 4096)        │                 (4096 ~ 8192)               │
 └───────────────────────┴───────────────────────┼─────────────────────┤
                                                 │ Trigger Buffer Zone │
                                                 │    (6144 ~ 8192)    │
                                                 └─────────────────────┘
                             ▲                           ▲
 ────────────────────────────┼───────────────────────────┼──────────────
                             │                           │
                   [Target Post-Compress]             [Trigger Line]
                   target = max - reserve         trigger = max - buffer
                          = 8192 - 4096                  = 8192 - 2048
                          = 4096 tokens                  = 6144 tokens

实际运行中：

1. 当前上下文占用 4096 token 以下：不用管，空间充足；
2. 当前上下文占用 4096 ~ 6144 token：仍然不用压缩，因为还在缓冲带里；
3. 当前上下文占用 超过 6144 token：触发压缩，把历史压回 4096 token；
4. 压缩后调用模型：4096 token 上下文 + 4096 token 预留输出 = 正好 8192，不会爆窗。

下面是这个上下文管理组件的简单实现：

class WorkingMemory:
  def __init__(
    self,
    max_tokens: int = 8192,
    reserve_output_tokens: int = 4096,
    compaction_buffer_tokens: int = 2048,
    ...
  ):
    ...

  def _compaction_target(self) -> int:
    """压缩目标：max_tokens - reserve_output_tokens"""
    target = self.max_tokens - self.reserve_output_tokens
    return target if target > 0 else self.max_tokens

  async def _enforce_budget(self) -> None:
    self.entries = microcompact_superseded_tool_results(self.entries, self._count_tokens)
    total = sum(e.tokens for e in self.entries)
    target = self._compaction_target()  # 4096
    # 触发线：max - buffer（6144），但不能低于 target
    if total <= max(self.max_tokens - self.compaction_buffer_tokens, target):
      return
    # 超过触发线，开始压缩到 target
    ...

$iv.$ 语义压缩以及熔断

在压缩时可以接入一个 summary_llm 做语义压缩，一旦 summarizer 连续失败 3 次，就永久回退到确定性压缩：

# nonoka/core/memory.py:482-485, 591-606
self._summary_failures = 0
self._summary_disabled = False

try:
    response = await self.summary_llm.chat(...)
except Exception:
    # A broken summariser must not break the session
    self._summary_failures += 1
    if self._summary_failures >= 3:
        self._summary_disabled = True
    result = await self.context_compactor.compact(...)
    self.entries = result.entries
    return

智能摘要用于提升体验，可靠性由确定性机制兜底。

$v.$ prompt caching：让稳定前缀真正省钱

prompt caching（也叫 prefix caching）不是把上下文变短，而是让相同的 token 更便宜。它的前提是：连续多轮请求的前缀必须字节级稳定。前缀一旦变化，provider 就要重新计算 KV，前面的 token 既要多花钱，又要多耗时。

所以 prompt caching 和 context engineering 是绑在一起的：往 prompt 里放什么、按什么顺序放都可以直接决定缓存命中率。

下面几条是在设计上下文结构时的关键原则，这里简单提一下：

1. 动态内容往后放，稳定内容往前放：prefix cache 通常从消息列表的最开头匹配。因此 system prompt 里只放 identity、工具定义、全局规则这类几乎不变的内容；用户当前问题、最新 tool 结果、memory 快照等动态内容，放在后面的 user/assistant/tool 消息里。

2. 不要把每轮都变的字段塞进 system prompt：这些都会让 system prompt 的字节发生变化，导致整段前缀缓存失效。如果必须加，要么放到 system prompt 的末尾（只让末尾之后失效），要么放到 user 消息里。

压缩会改变消息列表，因此我们选择尽量只影响中间不稳定的区域：

• 不要每轮都重建 system prompt；
• 优先压缩/替换中间的历史消息，保留最近几条消息作为后缀断点；
• microcompact 或 evidence ledger 替换后的消息尽量保持长度相近，避免产生大段字节抖动。

当 COMPACT 完整执行完整个流程后上下文仍然超限时，下一步就不是继续压榨窗口，而是把状态搬出窗口，也就是下一节要讲述的。

$c.$ WRITE / ISOLATE：状态能不能不放在窗口里

前面两层解决的是“窗口里的内容怎么管理”。第三层问的是一个更前置的问题：这件事必须放在窗口里吗？

LangChain 的 WRITE / ISOLATE 策略就是解决这个问题的：把状态写到窗外，窗口里只保留引用或最新摘要：

• WRITE：一些中间状态不进窗口，而是直接持久化到外部存储；
• ISOLATE：跨轮需要保留的上下文摘要，也持久化到外部，窗口里只放一条引用或最新快照或者提供 SELECT 可使用的索引 / 按需查询工具。

$i.$ WRITE：把中间产物写到窗外

SELECT 里已经说过，模型应该搬运 Artifact 引用而不是正文。WRITE 就是把那个引用背后的产物真正持久化到外部存储，让它跨步骤、跨轮次、跨会话都能被找回。只要某个产物后面还要被用到，就应该写进 ArtifactStore，而不是让 LLM 从上下文里回忆。

一个极简的存储接口如下：

class ArtifactStore:
  def put(self, task_id: str, artifact: Artifact) -> None:
    """持久化产物，带 checksum 和 schema_id。"""
    ...

  def get(self, task_id: str, step_id: str) -> Artifact | None:
    """按任务+步骤读取产物；失败返回 None，不凭空生成。"""
    ...

  def list_refs(self, task_id: str) -> list[dict]:
    """返回该任务下所有产物的引用列表，供模型选择读取。"""
    ...

这里面的 Artifact 就是 SELECT 里同一种结构：它带 schema_id、checksum、producer，模型拿到的是轻量引用，真正的 payload 留在外部 store。

• 工作流引擎从 store 读取上一步产物，用确定性代码计算下一步产物，再写回 store；
• 会话中断、进程重启后，只要 task_id 还在，就能从 store 重建状态；
• 模型上下文里只保留当前需要的引用，不需要“回忆”之前做过什么。

$ii.$ ISOLATE：跨轮上下文搬出窗口

ISOLATE 通常就是 Agent Memory 系统的入口。换句话说，ISOLATE 本质上是 context engineering 视角下的 Memory：它回答的是“哪些跨轮状态需要被记住，但不该放在窗口里”。

context engineering 只负责判断“什么该搬出去”，至于这些状态怎么存、怎么查、怎么做版本链和加密，属于 Memory 相关内容，会在另一篇文章里展开。

COMPACT 能把历史消息压成摘要，但如果这条摘要也要在窗口里待很多轮，它本身会继续膨胀。ISOLATE 的做法是：把摘要写到窗外（SQLite、向量库、外部 Memory 服务等），窗口里只保留一个指向外部 checkpoint 的引用，或者最新快照的一句话预览。

窗口里只需要类似这样：

[当前几轮消息...]

[外部摘要引用] active_checkpoint: chk_abc123
preview: "用户想排查登录失败，已确认密码策略无异常，待确认网络连通性。"

而不是把几十轮历史全部塞进去。

更完整的 Memory 设计见 Memory 相关博客。这里只写一下 context engineering 相关设计：跨轮状态可以并且应该被隔离出窗口。

即使压缩和隔离都失败，系统也可以通过对 prompt 副本做临时裁剪，至少保留摘要 + 最近 2 条消息：

// src/core/context.ts:125-138
while (promptTokens() > this.options.maxContextTokens && result.length > 2) {
  result.splice(ctx.summary ? 1 : 0, 1);
}

注意：临时裁剪只动 prompt 副本，不动持久化 history。这样本轮请求能继续，等后续条件好转再真正压缩。

4. 三层架构间的协调

将这三层合起来，一个 Agent Loop 的上下文治理流程如下：

User Request / Tool Execution Output
                  │
                  ▼
┌────────────────────────────────────────────────────────┐
│ Layer 1: SELECT (Information Filtering)                │
│ • Current-turn reference over copy; fetch sections     │
│   over full book; resolve deterministically            │
└────────────────────────────────────────────────────────┘
                  │
                  ▼ [Payload destined for context window]
┌────────────────────────────────────────────────────────┐
│ Layer 2: COMPACT (In-Context Compression)              │
│ • Micro-compact  → Protocol-aware compression          │
│ • Evidence ledger → Maximizes effective window         │
│ • Optional summarization with circuit-breaker fallback │
│ • Stable prefix   → Prompt caching for cost saving     │
└────────────────────────────────────────────────────────┘
                  │
                  ▼ [State preserved across turns]
┌────────────────────────────────────────────────────────┐
│ Layer 3: WRITE / ISOLATE (Externalization & Offloading)│
│ • Phase outputs   → Persisted Artifacts / Workflow     │
│ • Multi-turn ctx  → External Memory / checkpoint       │
│ • Prompt-only trimming as final safety buffer          │
└────────────────────────────────────────────────────────┘

这三层不是并列的三种实现，而是同一个上下文管理的三个递进层次：

• 能 SELECT 的，就不 COMPRESS；
• 能 COMPRESS 的，就不 WRITE；
• 必须跨轮保留的，才 ISOLATE 到窗外。