压缩而不丢失历史记录#
在保留原始已存储消息的同时总结较早的轮次。最近的轮次仍完整可用，因此长对话可以在你配置的上下文预算内继续进行。
运行示例#
在仓库根目录下：
uv run python docs/_examples/compaction.py
输出包含：
Raw messages: 6; checkpoint boundary: 3; raw tail: 3
这个程序使用确定性的离线模型，并在需要时使用临时存储。它不需要 API 密钥，由文档测试套件执行。
在应用中添加压缩#
from zett_agent.client import create_agent
from zett_agent.extensions.compaction import CompactionExtension
from zett_agent.extensions.sqlite import SQLiteSessionExtension
from zett_agent.extensions.tool_guidelines import ToolGuidelinesExtension
history = SQLiteSessionExtension("sessions.sqlite")
client = await create_agent(
model,
extensions=[
history,
ToolGuidelinesExtension(),
CompactionExtension(
summarizing_model,
max_tokens=40_000,
keep_recent_tokens=8_000,
),
],
)
model 是你的主适配器；summarizing_model 是一个能够把上下文总结为纯文本、且不进行工具调用的适配器。它们可以是同一个实例。如果省略传给 CompactionExtension 的模型，它会使用请求所用的模型。在应用关闭时关闭存储以及你自己拥有的模型提供商；参见集成到应用。
上面的数字只是示例，不是检测到的模型上限。请根据你实际使用的模型和预期工具输出来设置它们。当估计的上下文超过 max_tokens 时，自动压缩会在模型步骤之前运行；该估计包含系统指令和工具定义。之前某个模型步骤报告的输入用量可以改进这个估计。
keep_recent_tokens 是一个下限，而不是硬性上限。包含它的用户轮次会连同其助手/工具消息一起完整保留。单个过大的当前轮次无法通过拆分它来解决；请减少它的输入/工具结果，或选择更大的上下文窗口。内置的文本分词器只是一个估计，对图像尤其如此，它也无法探知你端点实际的上下文容量。
需要配置的内容#
选择一个用于总结的模型、压缩的阈值，以及最近消息的预算。示例故意设置了很小的限制，好让两次请求就触发一次摘要：
async def main() -> None:
with TemporaryDirectory(prefix="zett-compaction-") as directory:
storage = SQLiteSessionExtension(Path(directory) / "sessions.sqlite")
try:
client = await create_agent(
AnswerModel(),
config=AgentRunConfig(session_id="compaction-demo"),
extensions=[
storage,
CompactionExtension(
SummaryModel(), max_tokens=100, keep_recent_tokens=10, count_tokens=demo_tokens
),
],
event_dispatcher=CompactionEvents(),
)
original = "I prefer Python. " * 100
await client.run(original)
await client.run("Continue with Python and keep all the current request words in the recent context.")
raw = await storage.list_raw_messages("compaction-demo")
view = await storage.storage.load("compaction-demo")
assert len(raw) == 6 and raw[1].message.content == original
assert view.snapshot is not None
assert view.snapshot.compacted_through_sequence == 3
assert len(view.raw_tail) == 3
print("Raw messages: 6; checkpoint boundary: 3; raw tail: 3")
finally:
await storage.close()
原始的六条消息仍保留在 SQLite 中；下一次请求可以恢复摘要及其边界之后的三条消息。摘要是有损的：请确认它保留了你工作流需要的事实。
重要
基于字词的 token 计数器只是一个确定性的测试夹具。请使用适合你模型的 token 估算，为响应和工具结果留出余地，并把限制设置在模型提供商的上下文窗口以下。
主动请求压缩#
CompactionExtension 也支持按需运行：AgentClient.compact 会恢复会话，通过 on_compact 请每个扩展执行压缩，并返回它存储的检查点。不会追加任何输入消息，也不会发起主模型调用，因此 UI 可以在读者提出要求的瞬间对对话进行总结:
stored = await client.compact()
print(stored.summary if stored is not None else "nothing to compact")
这一轮的事件会像流式运行那样到达客户端的 event_dispatcher，上面那些 Compaction started 行就是这样产生的。
显式的 compact() 过程不会等待自动阈值。它仍然需要较早的完整对话来总结，并且会跳过没有变小的替换结果。None 可能表示没有符合条件的窗口，或者没有有用的压缩；它并不意味着请求失败。带有 applied=False 的 COMPACTION_COMPLETED 表示生成了一个摘要但被丢弃了。
压缩是一次模型请求，可能产生费用，也可能失败。缺失的、空的或进行工具调用的摘要都是协议错误，而不是成功的检查点。摘要是有损的：对于历史措辞必须精确的工作流，请编写测试并审查摘要内容。保留原始日志以得到完整记录。
完整源码#
1"""Compact old turns and verify Raw Log preservation in a temporary database."""
2
3import asyncio
4from collections.abc import AsyncIterator, Sequence
5from pathlib import Path
6from tempfile import TemporaryDirectory
7
8from zett_agent.agent import AgentRunConfig
9from zett_agent.client import create_agent
10from zett_agent.dispatcher import AgentEventDispatcher
11from zett_agent.events import AgentEvent
12from zett_agent.extensions.compaction import CompactionExtension
13from zett_agent.extensions.sqlite import SQLiteSessionExtension
14from zett_agent.messages import (
15 AnyMessage,
16 AssistantMessage,
17)
18from zett_agent.model import (
19 ModelEvent,
20 ModelRequest,
21 ModelResponse,
22 RetryOptions,
23)
24
25
26def demo_tokens(messages: Sequence[AnyMessage]) -> int:
27 """A reproducible word-token estimate for this ASCII-only teaching example."""
28 return sum(len(str(message.content).split()) for message in messages)
29
30
31class AnswerModel:
32 retry = RetryOptions(max_retries=0)
33
34 async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
35 yield ModelEvent.completed(ModelResponse(AssistantMessage(content="Noted.")))
36
37
38class SummaryModel:
39 retry = RetryOptions(max_retries=0)
40
41 async def stream(self, request: ModelRequest) -> AsyncIterator[ModelEvent]:
42 yield ModelEvent.reasoning("Keep the durable preference.")
43 yield ModelEvent.text("The user prefers Python.")
44 yield ModelEvent.completed(ModelResponse(AssistantMessage(content="The user prefers Python.")))
45
46
47class CompactionEvents(AgentEventDispatcher):
48 async def on_compaction_started_event(self, event: AgentEvent) -> None:
49 print("Compaction started")
50
51 async def on_compaction_completed_event(self, event: AgentEvent) -> None:
52 print(f"Compaction applied: {event.applied}")
53
54
55async def main() -> None:
56 with TemporaryDirectory(prefix="zett-compaction-") as directory:
57 storage = SQLiteSessionExtension(Path(directory) / "sessions.sqlite")
58 try:
59 client = await create_agent(
60 AnswerModel(),
61 config=AgentRunConfig(session_id="compaction-demo"),
62 extensions=[
63 storage,
64 CompactionExtension(
65 SummaryModel(), max_tokens=100, keep_recent_tokens=10, count_tokens=demo_tokens
66 ),
67 ],
68 event_dispatcher=CompactionEvents(),
69 )
70 original = "I prefer Python. " * 100
71 await client.run(original)
72 await client.run("Continue with Python and keep all the current request words in the recent context.")
73 raw = await storage.list_raw_messages("compaction-demo")
74 view = await storage.storage.load("compaction-demo")
75 assert len(raw) == 6 and raw[1].message.content == original
76 assert view.snapshot is not None
77 assert view.snapshot.compacted_through_sequence == 3
78 assert len(view.raw_tail) == 3
79 print("Raw messages: 6; checkpoint boundary: 3; raw tail: 3")
80 finally:
81 await storage.close()
82
83
84if __name__ == "__main__":
85 asyncio.run(main())