08 AI 助手为什么会＂跑着跑着就死了＂?

                    —— 拆解 AI 编程助手的"上下文压缩"四层管线

一、一个所有 AI 用户都遇到过的场景

python3 demo_context_compact.py

# demo_context_compact.py — 四层上下文压缩管线最小可运行版# 纯标准库，无需安装任何依赖，无需 API Key。# （L1-L3 为 s08 教学版原文，L4 用假摘要代替真实 LLM 调用）from pathlib import PathTOOL_RESULTS_DIR = Path(”.task_outputs”) / ”tool-results”# ═══════════ 伪造一段”跑了很多轮”的对话 ═══════════def fake_messages():    msgs = [{”role”: ”user”, ”content”: ”帮我写一个命令行工具 hello.py”}]    for i in range(60):# 60 轮工具调用（每条结果约 600 字符）        msgs.append({”role”: ”assistant”, ”content”: [            {”type”: ”tool_use”, ”id”: f”t{i}”, ”name”: ”read_file”,             ”input”: {”path”: f”src/module{i}.py”}}]})        msgs.append({”role”: ”user”, ”content”: [            {”type”: ”tool_result”, ”tool_use_id”: f”t{i}”,             ”content”: (”def func_%d():\n    return %d\n” % (i, i)) * 25}]})# 再补一条约 540KB 的大输出（模拟 cat 一个大文件），用于触发 L3    msgs.append({”role”: ”assistant”, ”content”: [        {”type”: ”tool_use”, ”id”: ”tbig”, ”name”: ”bash”,         ”input”: {”command”: ”cat data/dump.json”}}]})    msgs.append({”role”: ”user”, ”content”: [        {”type”: ”tool_result”, ”tool_use_id”: ”tbig”,         ”content”: '{”rows”: [' + ','.join(f'[1,”row-{i}”,{i}]' for i in range(30000)) + ']}'}]})    return msgs# ═══════════ L1: 裁掉中间的老对话 ═══════════def _has_tool_use(m):    return m.get(”role”) == ”assistant” and any(        b.get(”type”) == ”tool_use” for b in m.get(”content”, []) if isinstance(b, dict))def _is_tool_result(m):    return m.get(”role”) == ”user” and any(        b.get(”type”) == ”tool_result” for b in m.get(”content”, []) if isinstance(b, dict))def snip_compact(messages, max_messages=50):    if len(messages) <= max_messages:        return messages    head_end, tail_start = 3, len(messages) - (max_messages - 3)# 切口保护: 不能把 tool_use 和它的 tool_result 拆开    if _has_tool_use(messages[head_end - 1]):        while head_end < len(messages) and _is_tool_result(messages[head_end]):            head_end += 1    if _is_tool_result(messages[tail_start]) and _has_tool_use(messages[tail_start - 1]):        tail_start -= 1    return (messages[:head_end]            + [{”role”: ”user”, ”content”: f”[snipped {tail_start - head_end} messages]”}]            + messages[tail_start:])# ═══════════ L2: 旧工具结果占位 ═══════════def micro_compact(messages):    results = [(m, b) for m in messages if m.get(”role”) == ”user”               and isinstance(m.get(”content”), list)               for b in m[”content”] if isinstance(b, dict) and b.get(”type”) == ”tool_result”]    if len(results) <= 3:# 只保留最近 3 条完整内容        return messages    for _, b in results[:-3]:        if len(b.get(”content”, ””)) > 120:            b[”content”] = ”[Earlier tool result compacted. Re-run if needed.]”    return messages# ═══════════ L3: 大结果落盘（上下文只留标记 + 预览） ═══════════def tool_result_budget(messages, max_bytes=200_000):    last = messages[-1]    if not last or last.get(”role”) != ”user” or not isinstance(last.get(”content”), list):        return messages    blocks = [(i, b) for i, b in enumerate(last[”content”])              if isinstance(b, dict) and b.get(”type”) == ”tool_result”]    total = sum(len(str(b.get(”content”, ””))) for _, b in blocks)    if total <= max_bytes:        return messages    for _, b in sorted(blocks, key=lambda p: len(str(p[1].get(”content”, ””))), reverse=True):        if total <= max_bytes:            break        TOOL_RESULTS_DIR.mkdir(parents=True, exist_ok=True)        path = TOOL_RESULTS_DIR / f”{b.get('tool_use_id', 'x')}.txt”        if not path.exists():            path.write_text(str(b[”content”]))        b[”content”] = (f”\nFull output: {path}\n”                        f”Preview:\n{str(b['content'])[:2000]}\n”)        total = sum(len(str(x.get(”content”, ””))) for _, x in blocks)    return messages# ═══════════ L4: 全量摘要（演示用假摘要，真实实现是调 LLM） ═══════════def summarize_history(messages):    return (”当前目标：写命令行工具 hello.py。已读取 60 个模块，接口已掌握。”            ”剩余工作：编写 CLI 入口、接入 argparse、补充单元测试。”            ”用户约束：Python 3 语法、无第三方依赖。”)def compact_history(messages):    return [{”role”: ”user”, ”content”: ”[Compacted]\n\n” + summarize_history(messages)}]# ═══════════ 合起来跑（顺序不能换: budget → snip → micro → 摘要） ═══════════def estimate_size(msgs):    return len(str(msgs))if __name__ == ”__main__”:    msgs = fake_messages()    orig = estimate_size(msgs)    print(f”伪造对话: {len(msgs)} 条消息, {orig/1000:.0f}KB”)    msgs = tool_result_budget(msgs);  print(f”L3 落盘后: {estimate_size(msgs)/1000:.0f}KB”)    msgs = snip_compact(msgs);        print(f”L1 裁剪后: {estimate_size(msgs)/1000:.0f}KB / {len(msgs)} 条”)    msgs = micro_compact(msgs);       print(f”L2 占位后: {estimate_size(msgs)/1000:.0f}KB”)    msgs = compact_history(msgs);     print(f”L4 摘要后: {estimate_size(msgs)/1000:.2f}KB”)    print(f”总压缩率: {100 * (1 - estimate_size(msgs) / orig):.1f}%”)

伪造对话: 123 条消息, 697KBL3 落盘后: 62KBL1 裁剪后: 26KB / 52 条L2 占位后: 10KBL4 摘要后: 0.14KB总压缩率: 100.0%