Skip to content

aaa.context

ContextPolicy 

ContextPolicy(
    max_input_tokens: int | Callable[[str | None], int],
    *,
    reserve_output: int = 4096,
    trigger: float = 0.85,
    target: float = 0.6,
    keep_recent: int = 3,
    strategies: Sequence[Strategy] | None = None,
    counter: TokenCounter | None = None,
    summarizer: LLMClient | None = None,
    summarizer_model: str | None = None,
)

Fit conversations into a model's input window.

max_input_tokens is the model's context window (an int, or a function of the model name). Nothing here knows your model's real limit, so say it: a too-large number means overflow errors, a too-small one wastes context. reserve_output is kept free for the reply. strategies run in order until the conversation is under target; the default is cheap-and-lossless first, Summarize only if you give a summarizer client.

observe 

observe(
    model: str | None,
    sent: Sequence[Message],
    tools: Sequence[ToolSpec],
    actual_prompt_tokens: int,
) -> None

Calibrate on a real response: compare the provider's own prompt-token count with our raw estimate.

prepare async 

prepare(
    transcript: Sequence[Message],
    *,
    tools: Sequence[ToolSpec] = (),
    model: str | None = None,
    book: SummaryBook | None = None,
    pressure: float = 1.0,
) -> Prepared

The messages to send for transcript. pressure < 1 assumes the real size is 1/pressure times our estimate, which forces deeper cuts. It is used to retry after the provider itself reported an overflow that our estimate missed: strategies stop as soon as the estimate fits, so re-running them unchanged would just produce the same request.

Strategies

TruncateToolResults 

TruncateToolResults(
    max_chars: int = 8000, head: float = 0.7
)

Shorten any single tool result longer than max_chars, keeping its start and end. Lossless in structure, lossy only inside very large outputs (a whole file, a big query result).

ClearToolResults 

ClearToolResults(
    keep_last: int = 4,
    placeholder: str = "[tool result removed to save context]",
)

Replace the content of old tool results (all but the most recent keep_last) with a placeholder. The calls stay, so the model still knows what it did and can call the tool again; only the bulky output goes.

Summarize 

Summarize(
    client: LLMClient | None = None,
    model: str | None = None,
    *,
    keep_tokens_ratio: float = 0.4,
    chunk_tokens: int = 24000,
    max_summary_tokens: int = 1200,
    summarize: Callable[[str, str | None], Awaitable[str]]
    | None = None,
)

Replace the oldest turns with a model-written summary, kept as [head] [summary] [recent tail]. The summary is incremental (each new one folds in the previous), is persisted by the LLMHandler as an aaa.context.summary event, and is reused by content hash, so a restart or another worker does not pay for it again.

summarizer defaults to a plain call on client/model; supply summarize= to use something else.

DropOldest

Last resort: delete the oldest turn groups (never the head or the recent tail) until the conversation fits, leaving a one-line marker so the model knows something is missing.

Strategy

Bases: Protocol

apply async 

apply(work: Work) -> None

Reduce work.messages in place (replace the list); record work.applied if it changed anything.

Token counting

HeuristicCounter 

HeuristicCounter(
    chars_per_token: float = 3.5,
    per_message: int = 4,
    per_tool_call: int = 8,
)

chars / chars_per_token plus a fixed per-message overhead. The default of 3.5 characters per token errs on the high side for English prose and is about right for code and JSON, so it over-counts a little: the safe direction for a budget.

CalibratedCounter 

CalibratedCounter(
    base: TokenCounter | None = None,
    *,
    smoothing: float = 0.3,
    bounds: tuple[float, float] = (0.4, 3.0),
)

Wraps a counter and scales it by the ratio actual / estimated seen on real responses (a smoothed average, per model), so a systematic error in the estimate stops mattering after a few calls.

observe 

observe(
    model: str | None, estimated: int, actual: int
) -> None

Feed back one real measurement: the estimate we made for a request and the provider's own count. estimated must be the uncalibrated estimate (base.count) of exactly what was sent.

TiktokenCounter 

TiktokenCounter(
    encoding: str = "o200k_base", per_message: int = 4
)

Exact counts for OpenAI-family models via tiktoken (pip install tiktoken). Other providers' models tokenize differently, so pair it with CalibratedCounter rather than trusting it blindly.

TokenCounter

Bases: Protocol

Summaries

SummaryBook dataclass 

SummaryBook(records: list[SummaryRecord] = list())

Summaries known for one conversation (loaded from the log, or made in this process).

best 

best(transcript: Sequence[Message]) -> SummaryRecord | None

The summary covering the longest prefix of transcript that matches by content.

SummaryRecord dataclass 

SummaryRecord(upto: int, prefix_hash: str, text: str)

A summary of T[head_end:upto] of the logical transcript T, valid for any transcript whose first upto messages hash to prefix_hash.

Prepared dataclass 

Prepared(
    messages: list[Message],
    report: dict[str, Any] | None = None,
    new_summaries: list[SummaryRecord] = list(),
    tokens_spent: int = 0,
    cost: float = 0.0,
)

groups 

groups(
    msgs: Sequence[Message], start: int = 0
) -> list[tuple[int, int]]

Half-open index ranges of atomic turn groups from start: an assistant message with tool calls together with its tool results, or any other message on its own.