Claude code底层实现原理（内存管理与并发）

s06: Context Compact (上下文压缩)

s01 > s02 > s03 > s04 > s05 > [ s06 ] | s07 > s08 > s09 > s10 > s11 > s12

“上下文总会满, 要有办法腾地方”– 三层压缩策略, 换来无限会话。

问题

上下文窗口是有限的。读一个 1000 行的文件就吃掉 ~4000 token; 读 30 个文件、跑 20 条命令, 轻松突破 100k token。不压缩, 智能体根本没法在大项目里干活。

解决方案

三层压缩, 激进程度递增:

Every turn: +------------------+ | Tool call result | +------------------+ | v [Layer 1: micro_compact] (silent, every turn) Replace tool_result > 3 turns old with "[Previous: used {tool_name}]" | v [Check: tokens > 50000?] | | no yes | | v v continue [Layer 2: auto_compact] Save transcript to .transcripts/ LLM summarizes conversation. Replace all messages with [summary]. | v [Layer 3: compact tool] Model calls compact explicitly. Same summarization as auto_compact.

工作原理

1. 第一层 – micro_compact: 每次 LLM 调用前, 将旧的 tool result 替换为占位符。

def micro_compact(messages: list) -> list: tool_results = [] for i, msg in enumerate(messages): if msg["role"] == "user" and isinstance(msg.get("content"), list): for j, part in enumerate(msg["content"]): if isinstance(part, dict) and part.get("type") == "tool_result": tool_results.append((i, j, part)) if len(tool_results) <= KEEP_RECENT: return messages for _, _, part in tool_results[:-KEEP_RECENT]: if len(part.get("content", "")) > 100: part["content"] = f"[Previous: used {tool_name}]" return messages

1. 第二层 – auto_compact: token 超过阈值时, 保存完整对话到磁盘, 让 LLM 做摘要。

def auto_compact(messages: list) -> list: # Save transcript for recovery transcript_path = TRANSCRIPT_DIR / f"transcript_{int(time.time())}.jsonl" with open(transcript_path, "w") as f: for msg in messages: f.write(json.dumps(msg, default=str) + "\n") # LLM summarizes response = client.messages.create( model=MODEL, messages=[{"role": "user", "content": "Summarize this conversation for continuity..." + json.dumps(messages, default=str)[:80000]}], max_tokens=2000, ) return [ {"role": "user", "content": f"[Compressed]\n\n{response.content[0].text}"}, {"role": "assistant", "content": "Understood. Continuing."}, ]

1. 第三层 – manual compact:compact工具按需触发同样的摘要机制。

2. 循环整合三层:

def agent_loop(messages: list): while True: micro_compact(messages) # Layer 1 if estimate_tokens(messages) > THRESHOLD: messages[:] = auto_compact(messages) # Layer 2 response = client.messages.create(...) # ... tool execution ... if manual_compact: messages[:] = auto_compact(messages) # Layer 3

完整历史通过 transcript 保存在磁盘上。信息没有真正丢失, 只是移出了活跃上下文。

相对 s05 的变更

试一试

cd learn-claude-code python agents/s06_context_compact.py

试试这些 prompt (英文 prompt 对 LLM 效果更好, 也可以用中文):

1. Read every Python file in the agents/ directory one by one(观察 micro-compact 替换旧结果)
2. Keep reading files until compression triggers automatically
3. Use the compact tool to manually compress the conversation

s08: Background Tasks (后台任务)

s01 > s02 > s03 > s04 > s05 > s06 | s07 > [ s08 ] s09 > s10 > s11 > s12

“慢操作丢后台, agent 继续想下一步”– 后台线程跑命令, 完成后注入通知。

问题

有些命令要跑好几分钟:npm install、pytest、docker build。阻塞式循环下模型只能干等。用户说 “装依赖, 顺便建个配置文件”, 智能体却只能一个一个来。

解决方案

Main thread Background thread +-----------------+ +-----------------+ | agent loop | | subprocess runs | | ... | | ... | | [LLM call] <---+------- | enqueue(result) | | ^drain queue | +-----------------+ +-----------------+ Timeline: Agent --[spawn A]--[spawn B]--[other work]---- | | v v [A runs] [B runs] (parallel) | | +-- results injected before next LLM call --+

工作原理

1. BackgroundManager 用线程安全的通知队列追踪任务。

class BackgroundManager: def __init__(self): self.tasks = {} self._notification_queue = [] self._lock = threading.Lock()

1. run()启动守护线程, 立即返回。

def run(self, command: str) -> str: task_id = str(uuid.uuid4())[:8] self.tasks[task_id] = {"status": "running", "command": command} thread = threading.Thread( target=self._execute, args=(task_id, command), daemon=True) thread.start() return f"Background task {task_id} started"

1. 子进程完成后, 结果进入通知队列。

def _execute(self, task_id, command): try: r = subprocess.run(command, shell=True, cwd=WORKDIR, capture_output=True, text=True, timeout=300) output = (r.stdout + r.stderr).strip()[:50000] except subprocess.TimeoutExpired: output = "Error: Timeout (300s)" with self._lock: self._notification_queue.append({ "task_id": task_id, "result": output[:500]})

1. 每次 LLM 调用前排空通知队列。

def agent_loop(messages: list): while True: notifs = BG.drain_notifications() if notifs: notif_text = "\n".join( f"[bg:{n['task_id']}] {n['result']}" for n in notifs) messages.append({"role": "user", "content": f"<background-results>\n{notif_text}\n" f"</background-results>"}) messages.append({"role": "assistant", "content": "Noted background results."}) response = client.messages.create(...)

循环保持单线程。只有子进程 I/O 被并行化。

相对 s07 的变更

试一试

cd learn-claude-code python agents/s08_background_tasks.py

试试这些 prompt (英文 prompt 对 LLM 效果更好, 也可以用中文):

1. Run "sleep 5 && echo done" in the background, then create a file while it runs
2. Start 3 background tasks: "sleep 2", "sleep 4", "sleep 6". Check their status.
3. Run pytest in the background and keep working on other things