Compact without losing history#

Summarize older turns while keeping the original stored messages. Recent turns remain available in full, so a long conversation can continue within your configured context budget.

Run it#

From the repository root:

uv run python docs/_examples/compaction.py

The output includes:

Raw messages: 6; checkpoint boundary: 3; raw tail: 3

This program uses deterministic offline models and temporary storage where needed. It requires no API key and is executed by the documentation test suite.

Add compaction to an application#

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 is your primary adapter; summarizing_model is an adapter capable of summarizing context as plain text without tool calls. They can be the same instance. If you omit the model passed to CompactionExtension, it uses the request’s model. Close the storage and any providers you own at application shutdown; see Integrate an application.

The numbers above are illustrative, not a detected model limit. Set them for the model and expected tool outputs you actually use. Automatic compaction runs before a model step when its estimated context crosses max_tokens; the estimate includes system instructions and tool definitions. Reported input usage from a previous model step can improve that estimate.

keep_recent_tokens is a minimum, not a hard upper bound. The containing user turn is kept whole with its assistant/tool messages. One oversized current turn cannot be solved by splitting it; reduce its input/tool results or choose a larger context window. The built-in text tokenizer is an estimate, especially for images, and does not discover your endpoint’s actual context capacity.

What to configure#

Choose a summarizing model, a threshold for compaction, and a budget for recent messages. The example sets deliberately small limits so two requests trigger a summary:

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()

The original six messages remain in SQLite; the next request can restore the summary plus the three messages after its boundary. A summary is lossy: verify that it preserves the facts your workflow needs.

Important

The word-based token counter is only a deterministic test fixture. Use a token estimate appropriate to your model, leave room for the response and tool results, and set limits below the provider’s context window.

Ask for compaction#

CompactionExtension also runs on demand: AgentClient.compact restores the session, asks every extension to compact through on_compact, and returns the checkpoint it stored. No input message is appended and no primary model call is made, so a UI can summarize a conversation the moment a reader asks for it:

stored = await client.compact()
print(stored.summary if stored is not None else "nothing to compact")

The events of the pass reach the client’s event_dispatcher exactly as a streamed run does, which is how the Compaction started lines above are produced.

An explicit compact() pass does not wait for the automatic threshold. It still needs older complete dialogue to summarize, and skips a replacement that is not smaller. None can mean there is no eligible window or no useful reduction; it does not mean the request failed. COMPACTION_COMPLETED with applied=False reports a generated summary that was discarded.

Compaction is a model request and can incur charges or fail. A missing, empty, or tool-calling summary is a protocol error, not a successful checkpoint. Summaries are lossy: use tests and review their content for workflows where exact historical wording is important. Keep the raw log for a full transcript.

Complete source#

Download compaction.py.

 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())

Next: Keep and restore conversations for durable history, or What the model sees for choosing context limits.