Skip to content

Compress a long context

Your backend calls one model and the history keeps growing. Add a CompressionPolicy to the call: the oldest or longest parts shrink before the model sees them. No agent framework is needed. How it works: Context compression.

flowchart LR
  H["your history"] --> R["LLMRequest(compression=policy)"]
  R --> P["your model port"]
  P -->|"over 50% of the window?<br/>shrink to 40%"| C["compressed messages"]
  C --> M[("your model")]

1. A script or a FastAPI service

Your model, wrapped once (Bring your own model):

def my_llm(request: LLMRequest) -> str:
    """Your own model call: an HTTP API, a local model... Here it says what it was sent."""
    chars = sum(len(m.content) for m in request.messages)
    return f"{len(request.messages)} messages, {chars} characters"


model = CallableModel(my_llm, provider="local", model="my-model")

One call with compression on:

policy = CompressionPolicy(
    model_max_tokens=4_000,  # your model's context window, in tokens
    start_when_full_percent=50,  # compress once the request fills half of it...
    shrink_to_percent=40,  # ...down to 40%
)


async def answer(history: list[Message]) -> str:
    request = LLMRequest(messages=tuple(history), compression=policy)
    reply = await model.generate(request)  # the oldest turns are dropped first
    return reply.text

The same in a FastAPI route. Never trust a client's Messages: take only user and assistant text within size limits, drop pinned and raw, and add your own system prompt:

class ChatTurn(BaseModel):
    """A turn as a client may send it: no system role; `pinned`, `raw` and the rest dropped."""

    role: Literal["user", "assistant"]
    content: str = Field(max_length=20_000)  # characters per turn


class ChatBody(BaseModel):
    messages: list[ChatTurn] = Field(min_length=1, max_length=200)


def make_app() -> Any:
    from fastapi import FastAPI  # imported only when the app is built

    app = FastAPI()

    @app.post("/chat")
    async def chat(body: ChatBody) -> dict[str, str]:
        turns = [Message(role=turn.role, content=turn.content) for turn in body.messages]
        return {"reply": await answer([SYSTEM, *turns])}  # your system prompt, not theirs

    return app

With the example's 14-message history, the model gets 10:

14 messages, 8339 characters
10 messages, 5575 characters

The first run logs one WARNING: tokens are estimated as characters / 4. Pass count_tokens= (your tokenizer, list[Message] -> int) for exact counts.

2. Pick what shrinks

Each policy below is used for one call:

async def ask(policy: CompressionPolicy) -> str:
    reply = await model.generate(LLMRequest(messages=history(), compression=policy))
    return reply.text
Policy Free? What the model gets
Trim only yes the system prompt and the newest turns
Cut only yes every message, long ones cut with a marker
Summarise only one call to your summary model a summary in place of the older turns
Combined cheap steps first, the summary only if still needed the smallest result
Most relevant one call to your reranker the turns that best match the question

Trim only: drop the oldest turns, keep the last two.

trim_only = CompressionPolicy(**LIMITS, steps=[RemoveOldMessages(keep_first=0, keep_last=2)])

Cut only: keep every turn, cut each long one to 60 tokens.

cut_only = CompressionPolicy(**LIMITS, steps=[CutLongMessages(max_tokens_per_message=60)])

Summarise only, with your own (cheaper) model. A cache= on the policy (or on the step) means the same turns are never summarised twice:

summarise_only = CompressionPolicy(
    **LIMITS,
    steps=[Summarize(model=summariser, keep_recent_messages=2)],
    cache=MemoryCache(),  # the same turns are never summarised (or paid for) twice
)

Combined, with a stronger fallback used once if the provider still says "too long":

combined = CompressionPolicy(
    **LIMITS,
    steps=[
        ShrinkToolResults(),  # free: old, big tool results become a short note
        CutLongMessages(max_tokens_per_message=100),  # free: long messages are cut
        Summarize(model=summariser, keep_recent_messages=4),  # one call to your summariser
    ],
    if_still_too_long=CompressionPolicy(**LIMITS, steps=[RemoveOldMessages(keep_last=2)]),
)

Keep only the most relevant turns, scored by your reranker. Return a RerankResult with usage, and each call is a priced stage="compress" ledger entry:

def my_reranker(query: str, texts: list[str]) -> RerankResult:
    """Your reranker (an API or a local cross-encoder); scores in 0..1, one per text."""
    scores = [1.0 if "3" in text else 0.1 for text in texts]
    return RerankResult(scores=scores, usage=Usage(units=1))  # usage is what gets billed


ledger = Ledger()
reranker = CallableReranker(my_reranker, provider="my-rerank", model="v1", ledger=ledger)
relevant_only = CompressionPolicy(
    **LIMITS, steps=[KeepMostRelevant(scorer=RerankerScorer(reranker), keep_count=2)]
)

What the example prints for its 10-message history:

before: 10 messages, 5493 characters
trim: 4 messages, 1407 characters
cut: 10 messages, 1833 characters
summarise: 4 messages, 914 characters
combined: 6 messages, 1385 characters
relevant: 4 messages, 1407 characters

3. Big system prompt?

The system message is never removed. By default no step shortens it either.

Option 1 (recommended): split it. Keep only your rules in the system message. Send long reference text (documents, product data) as its own message, which steps may cut or summarise like any other:

split = (
    Message(role="system", content="You answer help requests. Cite the article."),  # rules only
    Message(role="user", content=f"Reference material:\n{REFERENCE}"),  # may be cut or summarised
    Message(role="user", content="So what can I claim?"),
)

Option 2: mark the reference part. Put it in <reference>...</reference>. With include_system_prompt=True, SummarizeLongMessage summarises only those blocks and checks every other line stays word for word. Without a block it leaves the prompt alone:

rules_and_reference = (
    "You answer help requests. Cite the article.\n"  # rules: always kept word for word
    f"<reference>\n{REFERENCE}\n</reference>"  # only this block may be summarised
)
big_system = CompressionPolicy(
    **LIMITS,
    steps=[
        SummarizeLongMessage(
            model=summariser,
            bigger_than_tokens=500,
            include_system_prompt=True,  # may shorten the system text; never removes it
            instructions="Keep article numbers.",
        ),
    ],
)

CutLongMessages and MyOwnStep also take include_system_prompt=True, but shorten the text blindly, rules included: use them only for text you trust.

4. Your own agent loop

The adapter can be the agent framework itself: run_tool_loop with your model is the loop, and it compresses before every model call.

registry = Registry()


@registry.tool
def exchange_rate(currency: str) -> str:
    """Today's EUR rate for a currency."""
    return f"{currency}: 0.92"


async def agent(history: list[Message]) -> str:
    """The adapter as the agent loop: your model, your tools, compression before each call."""
    result = await run_tool_loop(model, registry, history, compression=policy)
    return result.final.text


every_call = CompressingModel(model, policy)  # or compress every call of one model

Checklist

  • A model outside the packaged price table needs model_max_tokens or a context_window cap.
  • Pass count_tokens= for exact counts.
  • Add if_still_too_long= so a provider overflow is retried once with a stronger policy.
  • Pin what must never change: Message(role="user", content=..., pinned=True).
  • Watch context.compressed events and the compress stage in the ledger.

Using a framework? See Compression by framework.