上下文压缩算法与策略

export API_KEY="你的 DeepSeek API Key"
python main.py

coding-agent-compact/
├── compact.py            # ← 新增：压缩全流程（压缩 prompt、水位线、重建历史）
├── session.py            # ← 改动：新增 archive_session，重写会话文件前存档完整对话
├── main.py               # ← 改动：命令参数分发；发请求前检查上下文水位
├── agent/
│   └── file_state.py     # ← 改动：新增 paths()，压缩后恢复文件时要用
└── ui/
    └── commands.py       # ← 改动：新增 /compact 命令，/status 显示上下文占用

async def run_compact(state, custom_instructions: str = ""):
    prompt = COMPACT_PROMPT
    extra = custom_instructions.strip()
    if extra:
        prompt += "\n\n补充要求：\n" + extra

    # 一次 LLM 调用：带着全部历史跑压缩 prompt，产出摘要
    result = await summarizer.run(prompt, message_history=state.history)
    summary = extract_summary(result.output)

    # 重写会话文件之前先存档完整对话记录
    transcript_path = session.archive_session(state.session_id)

    # 刷新文件读写的 readFileState
    old_state = state.read_file_state
    state.read_file_state = ReadFileState()
    # 重建历史：一条摘要消息 + 恢复最近读过的文件
    restored = restore_file_messages(old_state, state.read_file_state)
    state.history = [build_summary_message(summary, transcript_path)] + restored
    # 重写会话文件，仅保留压缩后的内容
    session.rewrite_messages(state.session_id, state.history)

    # 旧检查点的 history_index 指向已被压缩掉的消息，整体丢弃
    fh = state.file_history
    if fh is not None and fh.checkpoints:
        fh.drop_from(fh.checkpoints[0])

cmd_name, _, args = user_input[1:].partition(" ")
command = COMMANDS.get(cmd_name)
# ...
if command.takes_args:
    result = command.handler(state, args.strip())
else:
    result = command.handler(state)

# 压缩专用 agent：不注册任何工具，模型想调也调不了
summarizer = Agent(model)

COMPACT_PROMPT = (
    "重要：只输出纯文本，不要调用任何工具。\n\n"
    # ...省略几条禁止调用工具的具体规则...
    "你的任务是为到目前为止的对话写一份详细摘要，重点关注用户的明确请求和你已经做过的事情。摘要要保留足够的技术细节，让后续工作能基于它无缝继续。\n"
    "这条压缩指令本身不属于要总结的对话——任何段落都不要提到或引用它，不要把它算作用户消息，也不要把写摘要说成当前工作。\n\n"
    "写正式摘要之前，先在 <analysis> 标签里打草稿整理思路：按顺序过一遍对话，核对用户的请求、你的处理方式、关键决策、文件名和代码片段、报错和修复过程。\n\n"
    "然后在 <summary> 标签里输出正式摘要，包含以下几段：\n\n"
    "1. 主要请求和意图：详细记录用户的所有明确请求和意图\n"
    "2. 关键技术概念：列出涉及的重要技术概念、框架和方案\n"
    "3. 文件和代码：列出查看过、修改过、新建过的具体文件和代码段，附上关键代码片段\n"
    "4. 错误与修复：列出遇到过的报错和修复方法，特别注意用户要求你换种做法的反馈\n"
    "5. 全部用户消息：一字不改地列出所有用户消息（工具结果和这条压缩指令都不算），不要转述——它们是理解用户反馈和意图变化的关键\n"
    "6. 当前工作与下一步：精确描述写摘要前正在做的事情和紧接着的下一步，逐字引用最近几条对话原文，确保接续时不跑偏\n\n"
    "再次提醒：不要调用任何工具。只输出纯文本——先 <analysis> 块，后 <summary> 块。"
)

def extract_summary(text: str) -> str:
    text = re.sub(r"<analysis>.*?</analysis>", "", text, flags=re.DOTALL)
    match = re.search(r"<summary>(.*?)</summary>", text, re.DOTALL)
    text = match.group(1) if match else text
    # 正文里残留的标签字样一并清掉
    return re.sub(r"</?(analysis|summary)>", "", text).strip()

def restore_file_messages(old_state, new_state) -> list:
    picked, used = [], 0
    for path in reversed(old_state.paths()):
        try:
            size = os.path.getsize(path)
        except OSError:
            continue
        if len(picked) >= RESTORE_MAX_FILES or used + size > RESTORE_MAX_CHARS:
            break
        picked.append(path)
        used += size
    # 恢复顺序保持原来的读取顺序
    picked.reverse()
    # 和 @ 引用同一套机制：伪装成 read_file 调用塞进历史，顺带登记进新 readFileState
    return build_mention_messages(picked, new_state)

def context_tokens(history) -> int:
    # 从后往前找
    # 寻找最近的一条带有 usage 的请求
    for msg in reversed(history):
        if msg.kind == "response" and msg.usage.input_tokens:
            # input + output 是总的上下文 token 数
            return msg.usage.input_tokens + msg.usage.output_tokens
    return 0

# 模型的上下文窗口上限（需要根据不同模型进行调整）
CONTEXT_WINDOW = 131_072

# 给 summarizer agent 调用预留的空间
# 我们之前写的 COMPACT_PROMPT，以及摘要输出
# 都要占用上下文，所以必须额外预留一些上下文
COMPACT_OUTPUT_RESERVE = 20_000

# 再留一些安全余量
AUTO_COMPACT_BUFFER = 10_000

def compact_threshold() -> int:
    return CONTEXT_WINDOW - COMPACT_OUTPUT_RESERVE - AUTO_COMPACT_BUFFER

async def on_submit(user_input):
    # 前面处理完 / 开头的命令...

    # 发请求前检查上下文水位，越过阈值就先自动压缩再继续
    await compact.auto_compact_if_needed(state)

    # 之后才是创建检查点、@ 引用注入、Agent 循环
    state.file_history.make_checkpoint(len(state.history), user_input)
    ...

async def auto_compact_if_needed(state):
    if state.compact_failures >= MAX_COMPACT_FAILURES:
        return
    used = context_tokens(state.history)
    threshold = compact_threshold()
    if used < threshold:
        return
    console.print(f"上下文接近上限（{used:,} / {threshold:,} tokens），自动压缩中\n")
    try:
        await run_compact(state)
        state.compact_failures = 0
    except Exception:
        state.compact_failures += 1
        # 失败只提示一句，不阻断本轮对话
        console.warn(f"自动压缩失败\n")

def archive_session(session_id: str) -> Path:
    archive_dir = project_dir() / "compact-history"
    archive_dir.mkdir(parents=True, exist_ok=True)
    # 历史对话记录文件名带有时间戳，防止覆盖
    path = archive_dir / f"{session_id}-{datetime.now():%Y%m%d-%H%M%S}.jsonl"
    shutil.copy2(session_file(session_id), path)
    return path

SUMMARY_WRAPPER = (
    "本会话由一段因上下文写满而被压缩的对话延续而来，以下是之前对话的摘要：\n\n"
    "{summary}\n\n"
    "如果摘要里缺少你需要的细节（具体代码片段、报错原文等），"
    "可以用 read_file 读取压缩前的完整对话记录：{transcript}\n"
    "请基于这份摘要继续工作。不要向用户复述摘要内容，直接从对话中断的地方接着干。"
)

def build_summary_message(summary: str, transcript_path) -> ModelRequest:
    content = SUMMARY_WRAPPER.format(summary=summary, transcript=transcript_path)
    return ModelRequest(parts=[UserPromptPart(content=content)])