Context compression

Compression policies, steps, scorers, CompressingModel and the message-format converters. Concepts: Context compression.

agent_adapter.context

Context compression: shrink the messages sent to a model, under fixed safety invariants.

SkipReason = Literal['', 'under_threshold', 'under_target', 'nothing_to_compress'] module-attribute

TokenCounter = Callable[[list[Message]], int] module-attribute

BaseCompressionStep

Bases: ABC

Base class for steps: a fixed name, apply, and config() for cache keys and logs.

apply(messages, context) abstractmethod async

The compressed list; see CompressionStep.

config()

The step's settings as JSON-safe values; objects and functions by their identity.

CallableReranker

Wraps fn(query, texts) as a RerankerPort that writes a ledger entry and events.

A sync fn runs in a worker thread; timeout_s bounds each call (a timed-out thread is abandoned, not stopped). The entry is priced as provider:model units in the price table.

Raises:

Type	Description
ConfigError	a bad provider, model, stage or timeout_s.
ToolUsageError	fn is not callable.

rerank(query, texts, *, ctx=None) async

One score per text, in order; no call (and no entry) for an empty list.

Raises:

Type	Description
ToolUsageError	query is not a str or texts not a sequence of str.
ModelTimeout	the call took longer than timeout_s.
ModelOutputInvalid	fn returned the wrong count, a non-finite score or bad usage.
ModelError	fn raised a non-adapter error (it is the __cause__).

CompressingModel

An LLMPort that compresses every request's messages with policy before model.

A request's own compression wins over policy. when(request, ctx) (sync or async) can skip compression for some calls. On ModelContextOverflow the policy's if_still_too_long runs once and the call is retried once. Nested layers (a wrapper in a wrapper, or run_tool_loop(compression=) around one) compress and retry once in all: an inner layer passes a request the outer one compressed straight through.

Raises:

Type	Description
ConfigError	policy is not a CompressionPolicy, when is not callable, or no context window is known for model (see the policy's model_max_tokens).

cache_key_fields()

The wrapped port's key material plus the policy's settings.

generate(request) async

The wrapped port's reply to the compressed request.

Raises:

Type	Description
ConfigError	the policy or a step's output is invalid.
ModelError	the wrapped port's error (a second overflow included).

stream(request) async

The wrapped port's stream of the compressed request.

Raises:

Type	Description
ConfigError	as generate.
ModelError	as generate; an overflow after the first event is not retried.

CompressionLimits dataclass

The two token counts one compression works with.

CompressionPolicy dataclass

When and how far to compress, and with which steps; give percents or token counts.

Raises:

Type	Description
ConfigError	a setting is out of range, or percents and token counts are mixed.

limits(model_max_tokens=None)

The token limits; percents apply to the policy's model_max_tokens, else this one.

Raises:

Type	Description
ConfigError	percents are used and no model size is known, or it is not positive.

resolved_steps()

The steps to run, in order: the policy's, else the defaults.

CompressionReport dataclass

What one compression did; counts only, never message text.

reached_target property

True when the result is at or under the target.

CompressionStep

Bases: Protocol

One stage of a compression pipeline.

apply returns the new message list. It must stop as soon as the conversation is under context.target_tokens, and should leave protected messages and tool rounds alone; the invariant guard reverts any change that breaks a rule anyway.

ContextSummary

Bases: BaseModel

The summariser's structured output.

CutLongMessages dataclass

Bases: BaseCompressionStep

Cut the text of messages over max_tokens_per_message, oldest first, keeping one end.

Only text is cut, never tool calls; a marker says how many characters went. With include_system_prompt=True, an unpinned system message is cut too, after the others.

apply(messages, context) async

messages with long texts cut until under the target.

EmbeddingScorer

Cosine similarity of your embedder's vectors; negative similarity counts as 0.

Raises:

Type	Description
ConfigError	embedder is not an EmbeddingPort.

score(query, texts) async

The cosine similarity of each text to query, clamped to 0..1.

Raises:

Type	Description
ModelOutputInvalid	the embedder returned vectors of different sizes.
ModelError	the embedder failed.

KeepMostRelevant dataclass

Bases: BaseCompressionStep

Score every unprotected unit against compare_to; keep the keep_count best ones.

The rest are dropped least relevant first (older first on a tie), only as many as needed. compare_to is the latest user message's text by default, or any text you give.

apply(messages, context) async

messages without the least relevant units until under the target.

Raises:

Type	Description
ConfigError	the scorer returned bad scores.

MyOwnStep dataclass

Bases: BaseCompressionStep

Your function as a step: list[Message] -> list[Message], sync or async.

The invariant guard still applies to what it returns; with include_system_prompt=True it may also shorten the system message's text (never remove it or change anything else).

apply(messages, context) async

What the function returns for a copy of messages.

Raises:

Type	Description
ConfigError	the function returned something other than a list of Message.

MyScorer

Your function fn(query, texts) -> list[float] in 0..1; a sync one runs in a thread.

Raises:

Type	Description
ConfigError	fn is not callable.

score(query, texts) async

What fn returns for query and a copy of texts; its errors propagate.

Protection

Which messages no step may change or remove, fixed on the input of one compression.

Protected: system messages, pinned messages, rounds carrying approval state, the latest user message and the last unit (the newest turn, including a pending tool round). A unit with one protected message is protected whole.

is_protected(message)

True when no step may change or remove message.

RemoveDuplicates dataclass

Bases: BaseCompressionStep

Drop an older copy of a later message or tool round, oldest first, only as many as needed.

match="exact" compares role, text and tool calls (not call ids); "similar" asks scorer and treats texts scoring above same_if_score_above as the same.

apply(messages, context) async

messages without older duplicates until under the target.

RemoveOldMessages dataclass

Bases: BaseCompressionStep

Drop the oldest messages between the first and last ones kept, only as many as needed.

keep_first and keep_last count non-system messages; a tool round on the edge is kept.

apply(messages, context) async

messages with the oldest middle units removed until under the target.

RerankResult

Bases: BaseModel

Scores in input order, plus what the call used, for billing.

RerankerPort

Bases: Protocol

Scores how relevant each text is to query; one float per text, in order.

RerankerScorer

Your reranker's scores, which must be in 0..1 (apply a sigmoid to raw logits first).

Raises:

Type	Description
ConfigError	reranker is not a RerankerPort.

score(query, texts) async

The reranker's score of each text for query.

Raises:

Type	Description
ModelError	the reranker failed.

Scorer

Bases: Protocol

Scores each text against query; one float in 0..1 per text, in order.

ShrinkToolResults dataclass

Bases: BaseCompressionStep

Shorten tool results over bigger_than_tokens older than older_than_turns, oldest first.

A turn is one assistant message; the newest older_than_turns turns are left alone. With save_to, the full text is saved there for save_for_seconds (None: no expiry), under the run's thread or run id, and the note gives an id for recall_tool.

apply(messages, context) async

messages with old big tool results shortened until under the target.

StepContext dataclass

What a step may read during one compress call; the same object for every step.

add_summary_cost(usd)

Record what a model call made by a step cost.

Raises:

Type	Description
ConfigError	usd is negative or not a finite number.

is_protected(message)

True when the step must not change or remove message.

message_tokens(message)

Tokens of one message (cached); steps keep a running total with these.

tokens(messages)

Tokens of the whole list, with the policy's counter.

under_target(tokens)

True once tokens is at or under the target: the step must stop.

units(messages)

messages as atomic units, each flagged protected or not.

StepFailure dataclass

A step that raised and was skipped (on_step_error="skip"); no exception text.

Summarize dataclass

Bases: BaseCompressionStep

Replace the oldest messages with summaries written by your model.

The newest keep_recent_messages non-system messages stay as they are (a tool round on the edge is kept whole), and so does every protected message. Each unbroken run of unprotected units between protected ones gets its own summary in its own place, so the order around pinned and approval messages never changes. Runs go oldest first, and only as many units as the target needs.

apply(messages, context) async

messages with old runs replaced by summaries, when that saves tokens.

Raises:

Type	Description
ConfigError	a summary_style function returned something other than a str.
ModelError	the model call failed (an invalid reply only skips the step).

SummarizeLongMessage dataclass

Bases: BaseCompressionStep

Replace the text of each message over bigger_than_tokens with your model's summary.

Oldest first, only as many as needed; tool calls and ids stay. With include_system_prompt=True, an unpinned system prompt is shortened too, last, but only inside its <reference>...</reference> blocks: every other line stays word for word.

apply(messages, context) async

messages with long texts summarised until under the target.

Raises:

Type	Description
ModelError	the model call failed (an invalid reply only skips that message).

TokenMeter

Counts with one counter and caches single-message counts while the messages live.

of(message)

Tokens of one message, cached by identity.

Raises:

Type	Description
ConfigError	the counter returned something other than an int >= 0.

total(messages)

Tokens of the whole list.

Raises:

Type	Description
ConfigError	the counter returned something other than an int >= 0.

Unit dataclass

Messages that are kept, changed or removed together; messages[start:stop].

carries_approval_state(message)

True for a tool result recording an approval or policy decision, or approval raw items.

compress_for(messages, policy, *, format='auto', pinned=None, model_max_tokens=None, ctx=None) async

compress() for any message format: the compressed list in that format, and a report.

Messages kept as they were come back as the same objects; nothing calls the main model.

Raises:

Type	Description
ConfigError	as compress and convert.
MissingDependency	as convert.

compress_for_sync(messages, policy, *, format='auto', pinned=None, model_max_tokens=None, ctx=None)

compress_for for sync code (Flask, Celery, scripts, a sync LangGraph node).

Raises:

Type	Description
ConfigError	as compress_for.
MissingDependency	as compress_for.

enforce_invariants(before, after, protection, *, step, allow_system_prompt_shortening=False)

after with every change that breaks an invariant skipped (the input message kept).

Rules: protected messages stay as they were; a tool round is kept whole if any of it is kept; a step may not add system or tool messages or tool calls; added messages land after the leading system messages, outside tool rounds and before the last unit. With allow_system_prompt_shortening, an unpinned system message may get shorter content in place (nothing else about it may change, and it is never removed).

Raises:

Type	Description
ConfigError	after is not a list of Message.

estimate_tokens(messages)

About chars/4 per message plus a fixed per-message overhead.

has_approval_marker(native)

True when a native message (any format) holds an approval or confirmation item.

Looks for an item type naming an approval, an MCP approval_request_id, or a request-confirmation function name, in dicts, lists and plain or pydantic objects.

recall_tool(store)

The recall_context(id) tool: the full text of a result saved in store by this run.

It reads only results saved under the calling context's owner scope (see owner_scope).

split_units(messages, protection=None)

messages as units in order; protected follows protection (default: of messages).

agent_adapter.context.formats

Compress any framework's message list: convert it to Message, compress, convert back.

Format	Covers
agent_adapter	Message lists (run_tool_loop, CallableModel, your own code)
langchain	LangChain and LangGraph BaseMessage lists
openai	chat dicts and Responses items (OpenAI Agents SDK, any OpenAI-compatible API)
anthropic	Anthropic Messages dicts (Anthropic, Claude Agent SDK)
gemini	Gemini contents, dicts or google.genai objects (Gemini, Google ADK)

Framework integrations add their own formats with register_format when imported.

FORMATS = ('agent_adapter', *_CONVERTERS) module-attribute

FormatName = Literal['auto', 'agent_adapter', 'langchain', 'openai', 'anthropic', 'gemini'] module-attribute

Pinned = Callable[[Any], bool] module-attribute

Conversion

Bases: Generic[T]

Native messages as messages; back() turns a compressed list into natives again.

Messages the compression kept as they were come back as the very same native objects; changed ones keep every field of their native except the changed text.

back(messages)

Natives for messages (this conversion's messages, compressed or not).

convert(messages, format='auto', *, pinned=None)

A Conversion: .messages to compress, .back(compressed) for the native list.

pinned(native) marks messages no step may change or remove.

Raises:

Type	Description
ConfigError	unknown format, agent_adapter (no conversion needed), or an item that is not a message of the format.
MissingDependency	a framework format needs its library to build new messages.

detect_format(messages)

The format of a message list; plain {"role", "content": str} dicts count as openai.

Raises:

Type	Description
ConfigError	the items mix formats or are not messages of any known format.

register_format(name, converter, *, detect=None)

Make format=name use converter; with detect, "auto" finds it too.

Integrations call this when they are imported, so context never imports a framework. Registering the same converter class again (a module reload) replaces it.

Raises:

Type	Description
ConfigError	the name is empty or built in, the converter is not a Converter, detect is not callable, or another converter already has the name.

registered_names()

Names of the registered formats, in registration order.

Also exported here, documented under the canonical path: compress_for, compress_for_sync.