Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 13 additions & 6 deletions agents/s06_context_compact.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,9 @@
|
v
[Layer 1: micro_compact] (silent, every turn)
Replace tool_result content older than last 3
Replace non-read_file tool_result content older than last 3
with "[Previous: used {tool_name}]"
Keep read_file results intact as reference material
|
v
[Check: tokens > 50000?]
Expand Down Expand Up @@ -56,6 +57,7 @@
THRESHOLD = 50000
TRANSCRIPT_DIR = WORKDIR / ".transcripts"
KEEP_RECENT = 3
PRESERVE_RESULT_TOOLS = {"read_file"}


def estimate_tokens(messages: list) -> int:
Expand Down Expand Up @@ -83,13 +85,18 @@ def micro_compact(messages: list) -> list:
for block in content:
if hasattr(block, "type") and block.type == "tool_use":
tool_name_map[block.id] = block.name
# Clear old results (keep last KEEP_RECENT)
# Clear old results (keep last KEEP_RECENT). Preserve read_file outputs because
# they are reference material; compacting them into placeholders can make the
# agent reread the same files and loop.
to_clear = tool_results[:-KEEP_RECENT]
for _, _, result in to_clear:
if isinstance(result.get("content"), str) and len(result["content"]) > 100:
tool_id = result.get("tool_use_id", "")
tool_name = tool_name_map.get(tool_id, "unknown")
result["content"] = f"[Previous: used {tool_name}]"
if not isinstance(result.get("content"), str) or len(result["content"]) <= 100:
continue
tool_id = result.get("tool_use_id", "")
tool_name = tool_name_map.get(tool_id, "unknown")
if tool_name in PRESERVE_RESULT_TOOLS:
continue
result["content"] = f"[Previous: used {tool_name}]"
return messages


Expand Down
18 changes: 9 additions & 9 deletions docs/en/s06-context-compact.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,9 @@ Every turn:
|
v
[Layer 1: micro_compact] (silent, every turn)
Replace tool_result > 3 turns old
Replace older non-read_file tool results
with "[Previous: used {tool_name}]"
Keep read_file results intact as reference material
|
v
[Check: tokens > 50000?]
Expand All @@ -42,19 +43,18 @@ continue [Layer 2: auto_compact]

## How It Works

1. **Layer 1 -- micro_compact**: Before each LLM call, replace old tool results with placeholders.
1. **Layer 1 -- micro_compact**: Before each LLM call, replace old non-`read_file` tool results with placeholders while preserving prior `read_file` output as reference material.

```python
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))
# ... collect tool_results and build tool_name_map ...
if len(tool_results) <= KEEP_RECENT:
return messages
for _, _, part in tool_results[:-KEEP_RECENT]:
tool_name = tool_name_map.get(part.get("tool_use_id", ""), "unknown")
if tool_name == "read_file":
continue
if len(part.get("content", "")) > 100:
part["content"] = f"[Previous: used {tool_name}]"
return messages
Expand Down Expand Up @@ -107,7 +107,7 @@ Transcripts preserve full history on disk. Nothing is truly lost -- just moved o
|----------------|------------------|----------------------------|
| Tools | 5 | 5 (base + compact) |
| Context mgmt | None | Three-layer compression |
| Micro-compact | None | Old results -> placeholders|
| Micro-compact | None | Old non-read results -> placeholders; keep file context |
| Auto-compact | None | Token threshold trigger |
| Transcripts | None | Saved to .transcripts/ |

Expand All @@ -118,6 +118,6 @@ cd learn-claude-code
python agents/s06_context_compact.py
```

1. `Read every Python file in the agents/ directory one by one` (watch micro-compact replace old results)
1. `Read every Python file in the agents/ directory one by one` (watch micro-compact keep older `read_file` context while shrinking older command output)
2. `Keep reading files until compression triggers automatically`
3. `Use the compact tool to manually compress the conversation`
18 changes: 9 additions & 9 deletions docs/ja/s06-context-compact.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,9 @@ Every turn:
|
v
[Layer 1: micro_compact] (silent, every turn)
Replace tool_result > 3 turns old
Replace older non-read_file tool results
with "[Previous: used {tool_name}]"
Keep read_file results intact as reference material
|
v
[Check: tokens > 50000?]
Expand All @@ -42,19 +43,18 @@ continue [Layer 2: auto_compact]

## 仕組み

1. **第1層 -- micro_compact**: 各LLM呼び出しの前に、古いツール結果をプレースホルダーに置換する
1. **第1層 -- micro_compact**: 各LLM呼び出しの前に、古い非 `read_file` ツール結果をプレースホルダーに置換しつつ、以前の `read_file` 出力は参照材料として保持する

```python
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))
# ... tool_results を集め、tool_name_map を構築する ...
if len(tool_results) <= KEEP_RECENT:
return messages
for _, _, part in tool_results[:-KEEP_RECENT]:
tool_name = tool_name_map.get(part.get("tool_use_id", ""), "unknown")
if tool_name == "read_file":
continue
if len(part.get("content", "")) > 100:
part["content"] = f"[Previous: used {tool_name}]"
return messages
Expand Down Expand Up @@ -107,7 +107,7 @@ def agent_loop(messages: list):
|----------------|------------------|----------------------------|
| Tools | 5 | 5 (base + compact) |
| Context mgmt | None | Three-layer compression |
| Micro-compact | None | Old results -> placeholders|
| Micro-compact | None | 古い非 read_file 結果 -> プレースホルダー; ファイル文脈は保持 |
| Auto-compact | None | Token threshold trigger |
| Transcripts | None | Saved to .transcripts/ |

Expand All @@ -118,6 +118,6 @@ cd learn-claude-code
python agents/s06_context_compact.py
```

1. `Read every Python file in the agents/ directory one by one` (micro-compactが古い結果を置換するのを観察する)
1. `Read every Python file in the agents/ directory one by one` (micro-compact が古い `read_file` 文脈を保持しつつ、より古いコマンド出力を縮める様子を観察する)
2. `Keep reading files until compression triggers automatically`
3. `Use the compact tool to manually compress the conversation`
18 changes: 9 additions & 9 deletions docs/zh/s06-context-compact.md
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,9 @@ Every turn:
|
v
[Layer 1: micro_compact] (silent, every turn)
Replace tool_result > 3 turns old
Replace older non-read_file tool results
with "[Previous: used {tool_name}]"
Keep read_file results intact as reference material
|
v
[Check: tokens > 50000?]
Expand All @@ -42,19 +43,18 @@ continue [Layer 2: auto_compact]

## 工作原理

1. **第一层 -- micro_compact**: 每次 LLM 调用前, 将旧的 tool result 替换为占位符。
1. **第一层 -- micro_compact**: 每次 LLM 调用前, 将旧的非 `read_file` tool result 替换为占位符, 同时保留先前的 `read_file` 输出作为参考材料

```python
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))
# ... 收集 tool_results 并构建 tool_name_map ...
if len(tool_results) <= KEEP_RECENT:
return messages
for _, _, part in tool_results[:-KEEP_RECENT]:
tool_name = tool_name_map.get(part.get("tool_use_id", ""), "unknown")
if tool_name == "read_file":
continue
if len(part.get("content", "")) > 100:
part["content"] = f"[Previous: used {tool_name}]"
return messages
Expand Down Expand Up @@ -107,7 +107,7 @@ def agent_loop(messages: list):
|----------------|------------------|--------------------------------|
| Tools | 5 | 5 (基础 + compact) |
| 上下文管理 | 无 | 三层压缩 |
| Micro-compact | 无 | 旧结果 -> 占位符 |
| Micro-compact | 无 | 旧的非 read_file 结果 -> 占位符;保留文件上下文 |
| Auto-compact | 无 | token 阈值触发 |
| Transcripts | 无 | 保存到 .transcripts/ |

Expand All @@ -120,6 +120,6 @@ python agents/s06_context_compact.py

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

1. `Read every Python file in the agents/ directory one by one` (观察 micro-compact 替换旧结果)
1. `Read every Python file in the agents/ directory one by one` (观察 micro-compact 保留旧的 `read_file` 上下文, 同时压缩更旧的命令输出)
2. `Keep reading files until compression triggers automatically`
3. `Use the compact tool to manually compress the conversation`
77 changes: 77 additions & 0 deletions tests/test_s06_micro_compact.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
import ast
import types
import unittest
from pathlib import Path


def load_micro_compact():
source = Path("agents/s06_context_compact.py").read_text()
tree = ast.parse(source)
selected = []
for node in tree.body:
if isinstance(node, ast.Assign):
for target in node.targets:
if isinstance(target, ast.Name) and target.id in {"KEEP_RECENT", "PRESERVE_RESULT_TOOLS"}:
selected.append(node)
break
elif isinstance(node, ast.FunctionDef) and node.name == "micro_compact":
selected.append(node)
module = ast.Module(body=selected, type_ignores=[])
ast.fix_missing_locations(module)
namespace = {}
exec(compile(module, "micro_compact_subset", "exec"), namespace)
return namespace["micro_compact"], namespace["KEEP_RECENT"]


def make_tool_use(tool_use_id: str, name: str):
return types.SimpleNamespace(type="tool_use", id=tool_use_id, name=name)


class MicroCompactTests(unittest.TestCase):
def setUp(self):
self.micro_compact, self.keep_recent = load_micro_compact()

def test_preserves_old_read_file_results_but_compacts_old_command_output(self):
long_read = "A" * 200
long_bash = "B" * 200
recent_1 = "C" * 200
recent_2 = "D" * 200
recent_3 = "E" * 200
messages = [
{"role": "assistant", "content": [make_tool_use("u1", "read_file")]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "u1", "content": long_read}]},
{"role": "assistant", "content": [make_tool_use("u2", "bash")]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "u2", "content": long_bash}]},
{"role": "assistant", "content": [make_tool_use("u3", "bash")]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "u3", "content": recent_1}]},
{"role": "assistant", "content": [make_tool_use("u4", "read_file")]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "u4", "content": recent_2}]},
{"role": "assistant", "content": [make_tool_use("u5", "edit_file")]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "u5", "content": recent_3}]},
]

self.micro_compact(messages)

self.assertEqual(messages[1]["content"][0]["content"], long_read)
self.assertEqual(messages[3]["content"][0]["content"], "[Previous: used bash]")
self.assertEqual(messages[5]["content"][0]["content"], recent_1)
self.assertEqual(messages[7]["content"][0]["content"], recent_2)
self.assertEqual(messages[9]["content"][0]["content"], recent_3)

def test_does_nothing_when_tool_results_do_not_exceed_keep_recent(self):
messages = [
{"role": "assistant", "content": [make_tool_use("u1", "read_file")]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "u1", "content": "A" * 200}]},
{"role": "assistant", "content": [make_tool_use("u2", "bash")]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "u2", "content": "B" * 200}]},
{"role": "assistant", "content": [make_tool_use("u3", "edit_file")]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "u3", "content": "C" * 200}]},
]
before = repr(messages)
self.micro_compact(messages)
self.assertEqual(repr(messages), before)
self.assertEqual(self.keep_recent, 3)


if __name__ == "__main__":
unittest.main()
Loading