1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
|
"""Show the full summary content after compaction"""
import json
from anthropic import Anthropic
from anthropic.lib.tools import beta_tool
@beta_tool
def search(query: str) -> str:
"""Search for information."""
return json.dumps(
{
"results": [
{"title": f"Result for {query}", "content": "Lorem ipsum " * 100},
{"title": f"More on {query}", "content": "Detailed info " * 100},
]
}
)
@beta_tool
def done(summary: str) -> str: # noqa: ARG001
"""Call when finished."""
return "Complete"
client = Anthropic()
runner = client.beta.messages.tool_runner(
model="claude-sonnet-4-20250514",
max_tokens=4096,
tools=[search, done],
messages=[
{
"role": "user",
"content": "You MUST search for EACH of these animals ONE BY ONE: dogs, cats, birds, fish, horses, elephants, lions, tigers, bears, wolves. After searching for ALL of them, call done.",
}
],
compaction_control={
"enabled": True,
"context_token_threshold": 3000, # Even lower threshold
},
)
prev_msg_count = 0
for i, message in enumerate(runner):
curr_msg_count = len(list(runner._params["messages"]))
print(f"Turn {i + 1}: {message.usage.input_tokens} input tokens, {curr_msg_count} messages")
if curr_msg_count < prev_msg_count:
print("=" * 70)
print("š COMPACTION OCCURRED!")
print("=" * 70)
print(f"Messages went from {prev_msg_count} ā {curr_msg_count}")
print(f"Input tokens: {message.usage.input_tokens}")
print("\nNEW MESSAGES LIST:")
print("-" * 70)
for msg in runner._params["messages"]:
role = msg.get("role", "?")
content = msg.get("content", "")
if isinstance(content, list):
for block in content:
if isinstance(block, dict) and block.get("type") == "text":
print(f"\n[{role}] TEXT BLOCK:")
print(block.get("text", ""))
elif isinstance(content, str):
print(f"\n[{role}]:")
print(content)
print("-" * 70)
prev_msg_count = curr_msg_count
print("\nā
Done!")
|
