# API reference: llmslim

> Every public name in the llmslim package, with its signature and docstring.

Signatures, type annotations and docstrings of the 80 public names in llmslim, generated from the source of llmslim 0.12.0.

Category: API Reference · Updated: October 2026 · 36 min read

## About this reference

Each entry is a name you import from llmslim: compression, context planning, the agent context runtime, cache planning, token counting and the rewrite engine. It shows the full signature with its type annotations, and the docstring. The page is generated from the source of llmslim 0.12.0 by scripts/api_reference.py, so it changes only when the code does.

## compress()

Compress text, retaining approximately target_ratio of its tokens.

```python
def compress(
    text: str,
    target_ratio: float = 0.5,
    query: Optional[str] = None,
    mode: Optional[str] = None,
    detect_content: bool = False,
    strategy: str = 'extractive',
    provider: Optional[Any] = None,
    required_keywords: Optional[List[str]] = None,
    constraints: Optional[List[str]] = None,
    template_name: Optional[str] = None,
    **kwargs,
) -> CompressionResult:
    """Compress ``text``, retaining approximately ``target_ratio`` of its tokens.

    This is the main entry point for the library -- a single function
    call that performs semantic chunking, extractive ranking, and
    reassembly (or optional semantic rewriting / hybrid optimization).

    Args:
        text: The prompt or document to compress.
        target_ratio: Fraction of original tokens to retain (e.g. ``0.5``
            for a ~50% reduction, ``0.3`` for a ~70% reduction).
        query: Optional query string for relevance-aware compression
            (useful for RAG contexts).
        mode: Optimization mode name (e.g. ``"balanced"``,
            ``"quality"``, ``"aggressive"``, ``"rag"``, ``"chat"``,
            ``"system"``, ``"code"``, ``"documentation"``).  Use
            ``"auto"`` with ``detect_content=True`` for fully automatic
            optimisation.  See :func:`llmslim.modes.list_modes`.
        detect_content: When ``True``, analyse the input to detect its
            content type (JSON, YAML, Python, chat, etc.) and populate
            telemetry fields on the result.  Adds < 1 ms overhead.
        strategy: Optimization strategy ("extractive", "rewrite", "hybrid").
            Defaults to "extractive".
        provider: Optional BaseRewriteProvider instance or callable when
            using "rewrite" or "hybrid" strategy.
        required_keywords: Optional list of keywords that must be present
            in the rewritten prompt.
        constraints: Optional list of prompt constraints passed to rewrite prompt.
        template_name: Optional template name override.
        **kwargs: Passed to :class:`ContextCompressor` if provided (e.g.
            ``max_chunk_tokens``, ``similarity_threshold``,
            ``preserve_patterns``). If no kwargs are given, a shared
            default compressor instance is reused for efficiency.

    Returns:
        A :class:`CompressionResult` with ``.compressed_text``,
        ``.reduction_percent``, and other statistics.

    Example:
        >>> from llmslim import compress
        >>> result = compress(my_long_prompt, target_ratio=0.5)
        >>> send_to_llm(result.compressed_text)
        >>> print(f"Saved {result.tokens_saved} tokens ({result.reduction_percent}%)")

    .. versionchanged:: 0.3.0
       Added ``strategy``, ``provider``, ``required_keywords``, ``constraints``,
       and ``template_name`` parameters.
    """
```

## ContextCompressor

Compresses text via semantic chunking + extractive sentence ranking.

```python
class ContextCompressor:
    """Compresses text via semantic chunking + extractive sentence ranking.

    Example:
        >>> compressor = ContextCompressor()
        >>> result = compressor.compress(long_text, target_ratio=0.5)
        >>> print(result.compressed_text)
        >>> print(result.summary())

    Args:
        embedding_backend: Custom :class:`EmbeddingBackend` or string name
            ('tfidf' / 'fast' / 'semantic' / 'sentence-transformers').
            Defaults to fast TF-IDF backend.
        max_chunk_tokens: Soft token cap per semantic chunk.
        similarity_threshold: Cosine similarity below which a new chunk
            is started. Lower values create fewer, larger chunks.
        min_tokens_for_compression: Texts at or below this token count
            are returned unchanged.
        weights: Optional overrides for ranking weights -- see
            ``llmslim.ranking.DEFAULT_WEIGHTS``.
        preserve_patterns: Optional list of regex strings; any sentence
            matching one of these is always retained (e.g.
            ``[r"API_KEY", r"^System:"]``).
        mode: Optimization mode name (e.g. ``"balanced"``,
            ``"quality"``, ``"aggressive"``).  When set, mode-specific
            ranking weights are applied.  Use ``"auto"`` together with
            ``detect_content=True`` for fully automatic optimisation.
        detect_content: When ``True``, analyse the input to detect its
            content type and populate telemetry fields on the result.
        context_role: Default provenance/trust role for text compressed by
            this instance (``ContextRole`` or an equivalent string such as
            ``"system"``, ``"rag"``, ``"tool"``).  Untrusted roles
            (``rag``/``tool``/``assistant``) are priority-capped so that
            injected imperative language cannot hijack the force-kept tier.
            Defaults to ``ContextRole.GENERAL`` (legacy v0.3.0 behaviour).
            May be overridden per call in :meth:`compress`.
        selection: ``"auto"`` (default) ranks whole units against the query
            when one is given, and keeps line structure for code, logs and
            other line-structured text; ``"focus"`` always uses query-focused
            selection; ``"proportional"`` restores the v0.7 behaviour (every
            chunk keeps its share of the budget).
        relevance: How query-focused selection scores relevance: ``"lexical"``
            (default; BM25 with feedback, no model), ``"semantic"`` (adds a
            small cross-encoder reranker on CPU; needs the ``semantic``
            extra), or a callable ``scorer(query, texts) -> scores``.

    .. versionchanged:: 0.8.0
       Query-focused selection and line-preserving compression are the default
       for queries and line-structured text (``selection="auto"``).
    """

    def __init__(
        self,
        embedding_backend: Optional[EmbeddingBackend | str] = None,
        max_chunk_tokens: int = 300,
        similarity_threshold: Optional[float] = None,
        min_tokens_for_compression: int = DEFAULT_MIN_TOKENS_FOR_COMPRESSION,
        weights: Optional[Dict[str, float]] = None,
        preserve_patterns: Optional[Sequence[str]] = None,
        mode: Optional[str] = None,
        detect_content: bool = False,
        context_role: Union[ContextRole, str] = ContextRole.GENERAL,
        selection: str = 'auto',
        relevance: Any = 'lexical',
    ):
        ...

    def compress(
        self,
        text: str,
        target_ratio: float = 0.5,
        query: Optional[str] = None,
        strategy: str = 'extractive',
        provider: Optional[Any] = None,
        required_keywords: Optional[List[str]] = None,
        constraints: Optional[List[str]] = None,
        template_name: Optional[str] = None,
        context_role: Optional[Union[ContextRole, str]] = None,
    ) -> CompressionResult:
        """Compress ``text``, retaining approximately ``target_ratio`` of its tokens."""
```

## ContextRole

Provenance / trust role for the text being compressed.

```python
class ContextRole(str, Enum):
    """Provenance / trust role for the text being compressed.

    The role controls how aggressively imperative language is protected
    during selection.  See ``docs/phase-1/IMPLEMENTATION_PLAN.md`` §3 for
    the full trust model and §13 for its explicit security limitations.

    * ``SYSTEM`` / ``DEVELOPER`` -- trusted developer-authored
      instructions.  Safety-critical directives may be hard-locked
      (Priority 4) and force-retained.
    * ``USER`` -- the end user's own turn.  Imperative wording may reach
      Priority 3 but not the safety-critical Priority 4 tier.
    * ``ASSISTANT`` / ``TOOL`` / ``RAG`` -- untrusted content (prior model
      output, tool/function results, retrieved documents).  Priority is
      capped so that injected imperatives cannot hijack the force-kept
      budget.
    * ``GENERAL`` -- default.  Preserves v0.3.0 behaviour exactly for
      callers that do not specify a role.

    ``ContextRole`` subclasses ``str``, so the enum members compare equal
    to their string values and either form may be passed anywhere a role
    is accepted.
    """

    SYSTEM = 'system'
    DEVELOPER = 'developer'
    USER = 'user'
    ASSISTANT = 'assistant'
    TOOL = 'tool'
    RAG = 'rag'
    GENERAL = 'general'
```

## CompressionResult

The result of compressing a piece of text.

```python
class CompressionResult:
    """The result of compressing a piece of text.

    The compressed text is available both as ``.compressed_text`` and via
    ``str(result)``, so it can often be dropped directly into an existing
    prompt-building pipeline.

    .. versionchanged:: 0.3.0
       Added optional fields: ``content_type``, ``content_confidence``,
       ``mode``, ``elapsed_ms``, ``instructions_found``,
       ``instructions_kept``, ``entities_found``, ``entities_kept``,
       ``structure_preserved``.
    """

    original_text: str
    compressed_text: str
    original_tokens: int
    compressed_tokens: int
    target_ratio: float
    sentences_total: int
    sentences_kept: int
    num_chunks: int
    backend: str = ''
    chunk_results: List[ChunkResult] = field(default_factory=list, repr=False)
    content_type: Optional[str] = None
    content_confidence: Optional[float] = None
    mode: Optional[str] = None
    elapsed_ms: Optional[float] = None
    instructions_found: Optional[int] = None
    instructions_kept: Optional[int] = None
    entities_found: Optional[int] = None
    entities_kept: Optional[int] = None
    structure_preserved: Optional[bool] = None
    rewrite_metadata: Optional[Any] = None
    token_counter_used: str = 'tiktoken'
    selection: Optional[str] = None

    @property
    def actual_ratio(self) -> float:
        """Fraction of original tokens retained (lower = more compression)."""

    @property
    def reduction_percent(self) -> float:
        """Percentage reduction in token count."""

    @property
    def tokens_saved(self) -> int:
        ...

    def summary(self) -> str:
        """A human-readable summary of the compression result."""

    def detailed_summary(self) -> str:
        """Extended summary including v0.3 telemetry when available."""

    def to_dict(self) -> Dict[str, Any]:
        """Serialise all fields to a plain dictionary."""
```

## ContentType

Detectable content types.

```python
class ContentType(str, Enum):
    """Detectable content types.

    Each value is a lowercase ``snake_case`` string suitable for use as
    a dictionary key, log label, or serialised identifier.
    """

    GENERAL_TEXT = 'general_text'
    CHAT_CONVERSATION = 'chat_conversation'
    SYSTEM_PROMPT = 'system_prompt'
    RAG_CONTEXT = 'rag_context'
    MARKDOWN = 'markdown'
    PYTHON = 'python'
    JAVASCRIPT = 'javascript'
    TYPESCRIPT = 'typescript'
    SQL = 'sql'
    JSON = 'json'
    YAML = 'yaml'
    XML = 'xml'
    HTML = 'html'
    CONFIG_FILE = 'config_file'
    API_DOCUMENTATION = 'api_documentation'
    RESEARCH_PAPER = 'research_paper'
    TECHNICAL_DOCUMENTATION = 'technical_documentation'
    LOG_FILE = 'log_file'
```

## ContentProfile

Immutable profile produced by analyze.

```python
class ContentProfile:
    """Immutable profile produced by :func:`analyze`.

    Attributes:
        content_type: Primary detected content type.
        confidence: Confidence in ``[0.0, 1.0]`` for the primary type.
        secondary_types: Other content types present (e.g. Markdown
            containing Python code blocks).
        has_structure: ``True`` when parseable structure is detected
            (JSON, YAML, XML, HTML, or Markdown heading hierarchy).
        estimated_tokens: Approximate token count of the input.
        language_hint: Programming language identifier when code is
            detected, otherwise ``None``.
        structure_depth: Maximum nesting / heading depth.  ``0`` for
            unstructured text.
        instruction_density: Fraction of non-empty lines matching
            instruction signals.
        entity_density: Fraction of non-empty lines containing named
            entities or technical identifiers.
    """

    content_type: ContentType
    confidence: float
    secondary_types: Tuple[ContentType, ...] = ()
    has_structure: bool = False
    estimated_tokens: int = 0
    language_hint: Optional[str] = None
    structure_depth: int = 0
    instruction_density: float = 0.0
    entity_density: float = 0.0
```

## analyze()

Analyze *text* and return its ContentProfile.

```python
def analyze(text: str) -> ContentProfile:
    """Analyze *text* and return its :class:`ContentProfile`.

    Runs all heuristic classifiers and selects the type with the
    highest confidence.  Falls back to ``GENERAL_TEXT`` when no
    classifier exceeds the confidence threshold.

    This function is designed to be fast and side-effect-free.  It does
    **not** perform compression or modify the input in any way.

    Args:
        text: The text to analyze.

    Returns:
        A frozen :class:`ContentProfile` describing the input.
    """
```

## list_modes()

Return the names of all available optimisation modes.

```python
def list_modes() -> List[str]:
    """Return the names of all available optimisation modes.

    >>> from llmslim.modes import list_modes
    >>> list_modes()
    ['aggressive', 'balanced', 'chat', 'code', 'documentation', 'quality', 'rag', 'system']
    """
```

## get_mode()

Look up a mode by *name*.

```python
def get_mode(name: str) -> OptimizationMode:
    """Look up a mode by *name*.

    Raises ``ValueError`` for unknown mode names.
    """
```

## compress_chat_messages()

Compress the content of chat messages before sending them to an LLM.

```python
def compress_chat_messages(
    messages: Sequence[Dict[str, str]],
    target_ratio: float = 0.5,
    compressible_roles: Iterable[str] = DEFAULT_COMPRESSIBLE_ROLES,
    min_tokens: int = 60,
    **kwargs,
) -> List[Dict[str, str]]:
    """Compress the ``content`` of chat messages before sending them to an LLM.

    Args:
        messages: A list of ``{"role": ..., "content": ...}`` dicts, in
            the same format used by OpenAI, Anthropic, and Gemini chat
            APIs.
        target_ratio: Fraction of tokens to retain in compressible
            messages.
        compressible_roles: Roles eligible for compression. Defaults to
            ``("user", "assistant")`` -- system prompts are preserved
            as-is.
        min_tokens: Messages below this token count are left untouched
            (compression overhead isn't worth it for short turns).
        **kwargs: Extra keyword arguments forwarded to
            :func:`llmslim.core.compress`.

    Returns:
        A new list of message dicts with ``content`` replaced by
        compressed text where applicable. The input list is not mutated.

    Claude thinking: only string content is compressed, so Anthropic content
    block lists (assistant turns with ``thinking`` blocks, tool results) pass
    through. A thinking block's signature is bound to everything before it,
    so once a message is changed, the signed ``thinking`` and
    ``redacted_thinking`` blocks of every later assistant message are removed
    (the ones before it are kept). An unfinished tool round at the end (the
    latest assistant turn calls tools and no user message follows) keeps its
    thinking: Anthropic expects it back. The output is a pure function of the
    input, so call this on the full history with the same options on every
    request; the messages the thinking was produced after are then sent the
    same way each time.

    Example:
        >>> from llmslim import compress_chat_messages
        >>> compressed = compress_chat_messages(conversation_history, target_ratio=0.5)
        >>> response = openai_client.chat.completions.create(model="gpt-5", messages=compressed)
    """
```

## compress_documents()

Compress a list of retrieved documents/chunks for a RAG pipeline.

```python
def compress_documents(
    documents: Sequence[str],
    query: Optional[str] = None,
    target_ratio: float = 0.5,
    context_role: Union[ContextRole, str] = ContextRole.RAG,
    **kwargs,
) -> List[CompressionResult]:
    """Compress a list of retrieved documents/chunks for a RAG pipeline.

    When ``query`` is provided, the documents share one budget
    (``target_ratio`` of their combined tokens) and it goes where the query
    points: the most relevant documents stay whole and irrelevant ones are
    left out (their ``compressed_text`` is empty). Without a query, each
    document is compressed to ``target_ratio`` on its own.

    Args:
        documents: Retrieved document texts/chunks.
        query: The user's query or search query, for relevance-aware
            ranking.
        target_ratio: Fraction of tokens to retain (across all documents
            when ``query`` is given, per document otherwise).
        context_role: Provenance/trust role for the documents.  Defaults to
            :attr:`ContextRole.RAG` (untrusted) so that imperative wording
            inside retrieved passages cannot be force-kept as a
            hard-locked priority.  This is an intentional v0.3.1 behavioural
            change from v0.3.0 (see the Phase 1 plan §8).  Pass a different
            role (e.g. ``"general"``) to opt out.
        **kwargs: Extra keyword arguments forwarded to
            :func:`llmslim.core.compress`.

    Returns:
        A list of :class:`CompressionResult`, one per input document, in
        the same order.

    Example:
        >>> from llmslim import compress_documents
        >>> results = compress_documents(retrieved_chunks, query=user_query, target_ratio=0.4)
        >>> context = "\n\n".join(r.compressed_text for r in results)
        >>> total_saved = sum(r.tokens_saved for r in results)

    .. versionchanged:: 0.8.0
       With a query, the budget is shared across documents by relevance
       (pass ``selection="proportional"`` for the v0.7 per-document split).
    """
```

## estimate_cost_savings()

Estimate dollar savings from compressing prompts at scale.

```python
def estimate_cost_savings(
    original_tokens: int,
    compressed_tokens: int,
    model: str = 'gpt-5',
    requests_per_day: int = 1000,
) -> CostEstimate:
    """Estimate dollar savings from compressing prompts at scale.

    Args:
        original_tokens: Token count before compression.
        compressed_tokens: Token count after compression.
        model: One of the keys in :data:`MODEL_PRICING`.
        requests_per_day: Expected daily request volume.

    Returns:
        A :class:`CostEstimate` with daily/monthly/annual savings.

    Example:
        >>> from llmslim import compress, estimate_cost_savings
        >>> result = compress(prompt, target_ratio=0.5)
        >>> cost = estimate_cost_savings(
        ...     result.original_tokens, result.compressed_tokens,
        ...     model="gpt-5", requests_per_day=50_000,
        ... )
        >>> print(cost.summary())
    """
```

## list_supported_models()

Return the list of model names recognized by estimate_cost_savings.

```python
def list_supported_models() -> List[str]:
    """Return the list of model names recognized by :func:`estimate_cost_savings`."""
```

## CostEstimate

Estimated cost savings from a compression result.

```python
class CostEstimate:
    """Estimated cost savings from a compression result."""

    model: str
    original_tokens: int
    compressed_tokens: int
    tokens_saved: int
    reduction_percent: float
    requests_per_day: int
    price_per_1k_input: float
    daily_savings_usd: float
    monthly_savings_usd: float
    annual_savings_usd: float

    def summary(self) -> str:
        ...
```

## MODEL_PRICING

Approximate USD prices per 1,000 input and output tokens, by model name. They change often; check the provider's pricing page before relying on them.

```python
MODEL_PRICING: Dict[str, Dict[str, float]] = ...
```

## count_tokens()

Return the number of tokens in text.

```python
def count_tokens(text: str, model: Optional[str] = None) -> int:
    """Return the number of tokens in ``text``.

    Args:
        text: The text to count tokens for.
        model: Reserved for future per-model tokenizer support. Currently
            all models use the same ``cl100k_base``-style estimate, which
            is a close match for GPT-5, Claude, and Gemini prompts.

    Returns:
        Token count (>= 0, or >= 1 for non-empty text).

    Note:
        When ``tiktoken`` is unavailable the function falls back to a
        character-based heuristic and emits a single ``logging.warning``
        (once per process).  Use :func:`get_active_token_counter_name` to
        query which counter is in effect.
    """
```

## count_tokens_batch()

Count tokens for a list of texts.

```python
def count_tokens_batch(texts: List[str], model: Optional[str] = None) -> List[int]:
    """Count tokens for a list of texts."""
```

## BaseRewriteProvider

Abstract base class for LLM rewrite providers.

```python
class BaseRewriteProvider(ABC):
    """Abstract base class for LLM rewrite providers.

    Subclass this to integrate any LLM backend (OpenAI, Anthropic,
    Gemini, Ollama, a local model, etc.) with the llmslim rewrite
    engine.  The core library depends only on this abstraction — it
    never imports any provider SDK.

    Example::

        class MyOpenAIProvider(BaseRewriteProvider):
            name = "openai"

            def __init__(self, client, model="gpt-4o"):
                self._client = client
                self._model = model

            def rewrite(self, request: RewriteRequest) -> str:
                response = self._client.chat.completions.create(
                    model=self._model,
                    messages=[
                        {"role": "system", "content": request.system_prompt or ""},
                        {"role": "user", "content": request.user_prompt or request.text},
                    ],
                    **request.metadata,
                )
                return response.choices[0].message.content

            def is_available(self) -> bool:
                return self._client is not None
    """

    name: str = 'base'

    def rewrite(self, request: RewriteRequest) -> str:
        """Rewrite text according to the request."""

    def is_available(self) -> bool:
        """Return ``True`` if this provider is ready to accept requests."""
```

## CallableProvider

Wrap any callable as a rewrite provider.

```python
class CallableProvider(BaseRewriteProvider):
    """Wrap any callable as a rewrite provider.

    This is the simplest integration path — pass a function or lambda
    that accepts a :class:`RewriteRequest` and returns a string.

    Example::

        from llmslim.rewrite import CallableProvider, RewriteRequest

        def my_rewriter(request: RewriteRequest) -> str:
            # call your LLM here
            return request.text[:int(len(request.text) * request.target_ratio)]

        provider = CallableProvider(my_rewriter, name="truncator")

    Args:
        fn: A callable that takes a :class:`RewriteRequest` and returns
            the rewritten text.
        name: Human-readable name for this provider.
    """

    def __init__(self, fn: Callable[[RewriteRequest], str], name: str = 'custom') -> None:
        ...

    def rewrite(self, request: RewriteRequest) -> str:
        """Delegate to the wrapped callable."""

    def is_available(self) -> bool:
        """Always ``True`` — the callable is available if it was provided."""
```

## RewriteRequest

Encapsulates everything a provider needs to perform a rewrite.

```python
class RewriteRequest:
    """Encapsulates everything a provider needs to perform a rewrite.

    Using a request object (instead of positional arguments) future-proofs
    the provider interface: new fields can be added without breaking
    existing provider implementations.

    Attributes:
        text: The text to rewrite (may be original or pre-compressed).
        target_ratio: Desired fraction of original tokens to retain.
        content_type: Detected or user-specified content type hint
            (e.g. ``"general_text"``, ``"rag_context"``).
        constraints: Free-form constraint strings the provider should
            honour (e.g. ``["preserve all URLs", "keep code blocks"]``).
        template_name: Name of the prompt template being used.
        system_prompt: The assembled system prompt for the LLM call.
        user_prompt: The assembled user prompt for the LLM call.
        metadata: Arbitrary key-value pairs for provider-specific config
            (e.g. ``{"temperature": 0.3, "max_tokens": 2048}``).
    """

    text: str
    target_ratio: float = 0.5
    content_type: Optional[str] = None
    constraints: List[str] = field(default_factory=list)
    template_name: Optional[str] = None
    system_prompt: Optional[str] = None
    user_prompt: Optional[str] = None
    metadata: Dict[str, Any] = field(default_factory=dict)
```

## RewriteEngine

Executes prompt rewrites using a provider, template resolver, and validator.

```python
class RewriteEngine:
    """Executes prompt rewrites using a provider, template resolver, and validator.

    Responsibility: Execution only. Does not pick templates itself (delegates to
    TemplateResolver) nor does it perform validation directly (delegates to RewriteValidator).
    """

    def __init__(
        self,
        provider: BaseRewriteProvider,
        resolver: Optional[TemplateResolver] = None,
        validator: Optional[RewriteValidator] = None,
    ) -> None:
        ...

    def rewrite(
        self,
        text: str,
        target_ratio: float = 0.5,
        content_type: Optional[str] = None,
        template_name: Optional[str] = None,
        constraints: Optional[List[str]] = None,
        required_keywords: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> RewriteResult:
        """Execute rewrite pipeline: format prompt -> call provider -> validate -> return result."""
```

## RewriteMetadata

Grouped metadata for rewrite telemetry within CompressionResult.

```python
class RewriteMetadata:
    """Grouped metadata for rewrite telemetry within CompressionResult.

    Grouped object to keep CompressionResult clean as features evolve.
    """

    strategy: str
    accepted: bool
    fallback_used: bool
    provider_name: str
    template_name: str
    similarity_score: float
    instruction_retention: float
    entity_retention: float
    rewrite_latency_ms: float
    validation_scores: Dict[str, float] = field(default_factory=dict)
    failure_reasons: List[str] = field(default_factory=list)
    error_message: Optional[str] = None

    def to_dict(self) -> Dict[str, Any]:
        """Convert metadata to plain dictionary."""
```

## RewriteValidator

Composite validator that runs the full validation pipeline.

```python
class RewriteValidator:
    """Composite validator that runs the full validation pipeline.

    Orchestrates :class:`StructuralValidator`, :class:`InstructionValidator`,
    :class:`EntityValidator`, and a pluggable :class:`BaseSimilarityValidator`
    (default: :class:`TfidfSimilarityValidator`).

    Args:
        similarity_validator: A :class:`BaseSimilarityValidator` instance.
            Defaults to :class:`TfidfSimilarityValidator`.
        min_similarity: Minimum similarity score to accept.
        min_instruction_retention: Minimum instruction retention rate.
        min_entity_retention: Minimum entity retention rate.

    Example::

        validator = RewriteValidator(min_similarity=0.5)
        result = validator.validate(original_text, rewritten_text)
        if result.passed:
            print("Rewrite accepted")
        else:
            print(f"Rejected: {result.failure_reasons}")
    """

    def __init__(
        self,
        similarity_validator: Optional[BaseSimilarityValidator] = None,
        min_similarity: float = 0.6,
        min_instruction_retention: float = 0.8,
        min_entity_retention: float = 0.7,
    ) -> None:
        ...

    def validate(
        self,
        original: str,
        rewrite: str,
        required_keywords: Optional[List[str]] = None,
    ) -> ValidationResult:
        """Run all validators and return a composite result."""
```

## ValidationResult

Outcome of the rewrite validation pipeline.

```python
class ValidationResult:
    """Outcome of the rewrite validation pipeline.

    Attributes:
        passed: ``True`` if all validators accepted the rewrite.
        similarity_score: Similarity between original and rewrite
            (from the configured :class:`BaseSimilarityValidator`).
        instruction_retention: Fraction of original instructions
            preserved in the rewrite.
        entity_retention: Fraction of original entities preserved.
        token_reduction: Actual compression ratio achieved
            (``compressed_tokens / original_tokens``).
        keyword_retention: Fraction of required keywords found in
            the rewrite.
        failure_reasons: Human-readable explanations of why validation
            failed (empty tuple on success).
        scores: Full per-validator score breakdown.
    """

    passed: bool
    similarity_score: float
    instruction_retention: float
    entity_retention: float
    token_reduction: float
    keyword_retention: float
    failure_reasons: Tuple[str, ...] = ()
    scores: Dict[str, float] = field(default_factory=dict)
```

## ContextSource

Read-only retrieval boundary; sources never decide model-context priority.

```python
class ContextSource(Protocol):
    """Read-only retrieval boundary; sources never decide model-context priority."""

    source_id: str

    async def search(self, query: str, limit: int = 10) -> Sequence[ContextItem]:
        """Return bounded, provenance-bearing candidates."""
```

## ContextSourceError

A context source failed before producing an authoritative result set.

```python
class ContextSourceError(RuntimeError):
    """A context source failed before producing an authoritative result set."""
```

## ContextStore

Explicit persistence boundary; planning never writes automatically.

```python
class ContextStore(Protocol):
    """Explicit persistence boundary; planning never writes automatically."""

    async def save(self, item: ContextItem, namespace: str = 'default') -> None:
        """Persist an explicitly approved context item."""

    async def get(self, item_id: str, namespace: str = 'default') -> Optional[ContextItem]:
        """Return one item when present."""

    async def search(
        self,
        query: str,
        limit: int = 10,
        namespace: str = 'default',
    ) -> Sequence[ContextItem]:
        """Retrieve bounded candidates from explicit persistent state."""

    async def delete(self, item_id: str, namespace: str = 'default') -> bool:
        """Delete an explicitly addressed item and report whether it existed."""
```

## InMemoryContextSource

Deterministic local source useful for applications and tests.

```python
class InMemoryContextSource:
    """Deterministic local source useful for applications and tests."""

    def __init__(self, items: Sequence[ContextItem], source_id: str = 'in-memory') -> None:
        ...

    async def search(self, query: str, limit: int = 10) -> Sequence[ContextItem]:
        ...
```

## InMemoryContextStore

Process-local explicit context store with no automatic ingestion.

```python
class InMemoryContextStore:
    """Process-local explicit context store with no automatic ingestion."""

    def __init__(self) -> None:
        ...

    async def save(self, item: ContextItem, namespace: str = 'default') -> None:
        ...

    async def get(self, item_id: str, namespace: str = 'default') -> Optional[ContextItem]:
        ...

    async def search(
        self,
        query: str,
        limit: int = 10,
        namespace: str = 'default',
    ) -> Sequence[ContextItem]:
        ...

    async def delete(self, item_id: str, namespace: str = 'default') -> bool:
        ...
```

## collect_context_sources()

Collect sources concurrently while making every failure explicit.

```python
async def collect_context_sources(
    sources: Sequence[ContextSource],
    query: str,
    *,
    limit_per_source: int = 10,
    fail_open: bool = False,
) -> ContextSourceResult:
    """Collect sources concurrently while making every failure explicit.

    The default is fail-closed. ``fail_open=True`` is an explicit availability
    choice and returns source IDs plus exception class names, never exception
    messages that could contain credentials or customer data.
    """
```

## plan_context()

Convenience API for one-shot adaptive context planning.

```python
def plan_context(
    *,
    messages: Sequence[Mapping[str, Any]] = (),
    documents: Sequence[Union[str, Mapping[str, Any], ContextItem]] = (),
    tools: Sequence[Union[ToolSchema, Mapping[str, Any]]] = (),
    memories: Sequence[Union[str, Mapping[str, Any], ContextItem]] = (),
    query: str = '',
    model: str = 'generic-128k',
    max_input_tokens: Optional[int] = None,
    reserve_output_tokens: int = 4096,
    safety_margin_tokens: int = 256,
    budget: Optional[Union[int, ContextBudget]] = None,
    policy: Union[PlannerPolicy, PolicyPreset, str, None] = None,
    provider: Optional[BaseRewriteProvider] = None,
    model_profile: Optional[ModelProfile] = None,
    raise_on_infeasible: bool = False,
) -> ContextPlan:
    """Convenience API for one-shot adaptive context planning.

    Optional items are chosen by a deterministic value-maximizing knapsack
    with two documented adjustments (see ``docs/planning/ALGORITHM.md``):

    - Documents, memories and tool results whose ``metadata`` marks them
      superseded (``superseded`` truthy, ``superseded_by`` set, or
      ``valid_until`` in the past) have their value scaled by
      ``llmslim.planning.scoring.SUPERSEDED_VALUE_FACTOR`` (0.05), so they are
      dropped before current evidence under budget pressure.
    - The single most relevant optional, non-superseded evidence item is not
      dropped in favour of strictly lower-relevance evidence when it fits by
      dropping or compressing that evidence.

    Retrieved content is always rendered with ``trusted="false"``; neither
    its wording nor a ``role`` key in a document mapping changes that.
    """
```

## AdaptiveContextPlanner

Provider-neutral, deterministic adaptive context planner.

```python
class AdaptiveContextPlanner:
    """Provider-neutral, deterministic adaptive context planner.

    The default path is fully local. A rewrite provider is considered only
    when the caller explicitly supplies one; LLMSlim never executes tools.
    """

    def __init__(
        self,
        *,
        policy: Union[PlannerPolicy, PolicyPreset, str, None] = None,
        provider: Optional[BaseRewriteProvider] = None,
        model_profiles: Iterable[ModelProfile] = (),
    ) -> None:
        ...

    def plan(
        self,
        *,
        messages: Sequence[Mapping[str, Any]] = (),
        documents: Sequence[Union[str, Mapping[str, Any], ContextItem]] = (),
        tools: Sequence[Union[ToolSchema, Mapping[str, Any]]] = (),
        memories: Sequence[Union[str, Mapping[str, Any], ContextItem]] = (),
        query: str = '',
        model: str = 'generic-128k',
        max_input_tokens: Optional[int] = None,
        reserve_output_tokens: int = 4096,
        safety_margin_tokens: int = 256,
        budget: Optional[Union[int, ContextBudget]] = None,
        policy: Union[PlannerPolicy, PolicyPreset, str, None] = None,
        provider: Optional[BaseRewriteProvider] = None,
        model_profile: Optional[ModelProfile] = None,
        raise_on_infeasible: bool = False,
    ) -> ContextPlan:
        """Plan the highest-value safe context within a finite model budget."""

    async def aplan(
        self,
        *,
        context_sources: Sequence[Any] = (),
        source_limit: int = 10,
        fail_open_sources: bool = False,
        documents: Sequence[Union[str, Mapping[str, Any], ContextItem]] = (),
        query: str = '',
        **kwargs: Any,
    ) -> ContextPlan:
        """Retrieve explicit context sources and then run the synchronous planner."""
```

## ContextItem

One provenance-bearing logical unit considered by the planner.

```python
class ContextItem:
    """One provenance-bearing logical unit considered by the planner."""

    item_id: str
    content: str
    kind: ContextKind = ContextKind.GENERAL
    role: ContextRole = ContextRole.GENERAL
    original_order: int = 0
    token_count: int = 0
    compressible: bool = True
    required: bool = False
    relevance: Optional[float] = None
    recency: float = 0.0
    priority: float = 0.0
    source: str = 'caller'
    metadata: Mapping[str, Any] = field(default_factory=dict, compare=False)

    def to_dict(self, include_content: bool = True) -> Dict[str, Any]:
        ...
```

## ContextKind

Logical context classes understood by the planner.

```python
class ContextKind(str, Enum):
    """Logical context classes understood by the planner."""

    SYSTEM = 'system'
    DEVELOPER = 'developer'
    USER = 'user'
    ASSISTANT = 'assistant'
    TOOL_RESULT = 'tool_result'
    RAG_DOCUMENT = 'rag_document'
    MEMORY = 'memory'
    TOOL_SCHEMA = 'tool_schema'
    GENERAL = 'general'
```

## ContextBudget

Resolved input budget, including output reservation and safety margin.

```python
class ContextBudget:
    """Resolved input budget, including output reservation and safety margin."""

    max_input_tokens: int
    reserve_output_tokens: int = 4096
    safety_margin_tokens: int = 256
    target_utilization: float = 1.0
    model_context_window: Optional[int] = None

    @property
    def available_input_tokens(self) -> int:
        ...

    def to_dict(self) -> Dict[str, Any]:
        ...
```

## ContextPlan

Complete local planning result and audit record.

```python
class ContextPlan:
    """Complete local planning result and audit record."""

    status: PlanStatus
    final_context: str
    final_messages: Tuple[Mapping[str, Any], ...]
    decisions: Tuple[ContextDecision, ...]
    budget: ContextBudget
    model_profile: ModelProfile
    metrics: PlanMetrics
    validation: PlanValidationResult
    warnings: Tuple[str, ...] = ()
    fallbacks: Tuple[str, ...] = ()

    @property
    def feasible(self) -> bool:
        ...

    def to_dict(self, include_content: bool = True) -> Dict[str, Any]:
        ...
```

## CandidateMethod

Available representations for one logical context item.

```python
class CandidateMethod(str, Enum):
    """Available representations for one logical context item."""

    RAW = 'raw'
    EXTRACTIVE_COMPRESSED = 'extractive_compressed'
    REWRITE_COMPRESSED = 'rewrite_compressed'
    HYBRID_COMPRESSED = 'hybrid_compressed'
    DROP = 'drop'
```

## PlannerPolicy

Compact parameter set shared by the planner's single algorithm.

```python
class PlannerPolicy:
    """Compact parameter set shared by the planner's single algorithm."""

    name: str = PolicyPreset.BALANCED.value
    relevance_weight: float = 0.32
    recency_weight: float = 0.22
    provenance_weight: float = 0.26
    density_weight: float = 0.12
    priority_weight: float = 0.08
    risk_penalty: float = 0.2
    compression_ratios: Tuple[float, ...] = (0.75, 0.5, 0.3)
    allow_drop: bool = True
    allow_rewrite: bool = True
    allow_hybrid: bool = True
    preserve_latest_turns: int = 2
    minimum_entity_retention: float = 0.7
    minimum_instruction_retention: float = 0.8
    target_utilization: float = 1.0
    experimental_selective_tools: bool = False
```

## PolicyPreset

The built-in planner policies, by name.

```python
class PolicyPreset(str, Enum):
    """The built-in planner policies, by name."""

    BALANCED = 'balanced'
    QUALITY_FIRST = 'quality_first'
    COST_FIRST = 'cost_first'
    LATENCY_FIRST = 'latency_first'
```

## ModelProfile

Data-driven context and pricing metadata for a target model.

```python
class ModelProfile:
    """Data-driven context and pricing metadata for a target model."""

    model_id: str
    provider: str
    context_window: int
    input_cost_per_million: Optional[float] = None
    cached_input_cost_per_million: Optional[float] = None
    output_cost_per_million: Optional[float] = None
    currency: str = 'INR'
    capabilities: Tuple[str, ...] = ()
    pricing_as_of: Optional[str] = None
    source: Optional[str] = None
    metadata: Mapping[str, Any] = field(default_factory=dict, compare=False)

    def estimate_input_cost(self, tokens: int) -> Optional[float]:
        ...

    def to_dict(self) -> Dict[str, Any]:
        ...
```

## InfeasibleContextError

Raised by strict callers when mandatory context exceeds the budget.

```python
class InfeasibleContextError(ContextPlanningError):
    """Raised by strict callers when mandatory context exceeds the budget."""

    def __init__(self, plan: ContextPlan) -> None:
        ...
```

## MCPAdaptiveContextPlan

Keeps context allocation and stale-safe catalog hydration state together.

```python
class MCPAdaptiveContextPlan:
    """Keeps context allocation and stale-safe catalog hydration state together.

    This is data only. It deliberately has no execution method; callers retain
    authorization and invocation responsibility through existing host adapters.
    """

    context_plan: ContextPlan
    catalog_plan: CatalogContextPlan
```

## plan_mcp_context()

Measure/select an MCP catalog, then allocate it with other context.

```python
def plan_mcp_context(
    snapshot: ToolCatalogSnapshot,
    *,
    query: str = '',
    mode: PlanMode = PlanMode.FULL,
    top_k: int = 5,
    experimental: bool = False,
    planner: Optional[AdaptiveContextPlanner] = None,
    **planner_kwargs: Any,
) -> MCPAdaptiveContextPlan:
    """Measure/select an MCP catalog, then allocate it with other context."""
```

## ContextEnvelope

Stable, provenance-bearing input set. Secrets are never accepted as metadata.

```python
class ContextEnvelope:
    """Stable, provenance-bearing input set. Secrets are never accepted as metadata."""

    items: Tuple[ContextItem, ...]
    current_query: str = ''
    session_id: Optional[str] = None
    agent_id: Optional[str] = None
    model: str = 'generic-128k'
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    envelope_id: str = ''

    @classmethod
    def from_bundle(cls, bundle: ContextBundle, **kwargs: Any) -> 'ContextEnvelope':
        ...

    @classmethod
    def from_messages(
        cls,
        messages: Sequence[Mapping[str, Any]],
        *,
        current_query: str = '',
        **kwargs: Any,
    ) -> 'ContextEnvelope':
        ...

    @classmethod
    def from_inputs(
        cls,
        *,
        messages: Sequence[Mapping[str, Any]] = (),
        documents: Sequence[Any] = (),
        memories: Sequence[Any] = (),
        tool_results: Sequence[Any] = (),
        tools: Sequence[Any] = (),
        current_query: str = '',
        **kwargs: Any,
    ) -> 'ContextEnvelope':
        ...

    def to_dict(self, *, include_content: bool = False) -> dict[str, Any]:
        ...
```

## ContextGraph

Edges provide interpretation dependencies; they never grant trust or authority.

```python
class ContextGraph:
    """Edges provide interpretation dependencies; they never grant trust or authority."""

    def __init__(self, item_ids: Iterable[str], edges: Iterable[ContextEdge] = ()) -> None:
        ...

    @classmethod
    def from_items(cls, items: Sequence[ContextItem]) -> 'ContextGraph':
        ...

    def dependencies(self, item_id: str) -> frozenset[str]:
        ...

    def closure(self, selected_ids: Iterable[str]) -> frozenset[str]:
        ...

    def to_dict(self) -> dict[str, Any]:
        ...
```

## ContextEdge

A directed edge of EdgeType kind from one context item id to another.

```python
class ContextEdge:
    """A directed edge of :class:`EdgeType` ``kind`` from one context item id to another."""

    source_id: str
    target_id: str
    kind: EdgeType

    def to_dict(self) -> dict[str, str]:
        ...
```

## EdgeType

How one context item relates to another.

```python
class EdgeType(str, Enum):
    """How one context item relates to another."""

    REPLIES_TO = 'replies_to'
    REFERENCES = 'references'
    SAME_ENTITY = 'same_entity'
    SUPPORTS = 'supports'
    DERIVED_FROM = 'derived_from'
    TOOL_CALL_RESULT = 'tool_call_result'
    MEMORY_SOURCE = 'memory_source'
    SUPERSEDES = 'supersedes'
    CONTRADICTS = 'contradicts'
    DEPENDS_ON = 'depends_on'
```

## ContextPolicy

Caller-owned rules applied before planning: roles never dropped, token caps per kind, source allow and deny lists, stale tool results, and an optional redactor.

```python
class ContextPolicy:
    """Caller-owned rules applied before planning: roles never dropped, token caps per kind, source allow and
    deny lists, stale tool results, and an optional redactor.
    """

    never_drop_roles: FrozenSet[ContextRole] = ...
    max_memory_tokens: Optional[int] = None
    max_rag_tokens: Optional[int] = None
    stale_tool_result_seconds: Optional[int] = None
    allowed_sources: Optional[FrozenSet[str]] = None
    denied_sources: FrozenSet[str] = field(default_factory=frozenset)
    redactor: Optional[Redactor] = field(default=None, compare=False, repr=False)

    def apply(
        self,
        items: Sequence[ContextItem],
        query: str,
        *,
        now: Optional[datetime] = None,
    ) -> Tuple[Tuple[ContextItem, ...], Tuple[str, ...]]:
        """Return retained items and safe decision reasons. Mandatory items fail closed."""
```

## QualityReport

Whether a selected context passed its quality gates, with the metrics and the gates that failed.

```python
class QualityReport:
    """Whether a selected context passed its quality gates, with the metrics and the gates that failed."""

    passed: bool
    quality_floor: float
    metrics: Mapping[str, float]
    failures: Tuple[str, ...] = ()

    def to_dict(self) -> Dict[str, Any]:
        ...
```

## ContextRuntime

Construct one model-visible context; never calls the model or executes tools.

```python
class ContextRuntime:
    """Construct one model-visible context; never calls the model or executes tools."""

    def __init__(
        self,
        *,
        model: str = 'generic-128k',
        objective: str = 'balanced',
        quality_floor: float = 0.8,
        max_input_tokens: Optional[int] = None,
        reserve_output_tokens: int = 4096,
        safety_margin_tokens: int = 256,
        policy: Optional[ContextPolicy] = None,
        model_profile: Optional[ModelProfile] = None,
        max_session_messages: int = 100,
        cache_policy: Optional[CachePolicy] = None,
        cache_manager: Optional[CacheManager] = None,
    ) -> None:
        ...

    def prepare_sync(
        self,
        *,
        session_id: Optional[str] = None,
        user_input: str = '',
        messages: Sequence[Mapping[str, Any]] = (),
        documents: Sequence[Any] = (),
        memories: Sequence[Any] = (),
        tool_results: Sequence[Any] = (),
        tools: Sequence[Any] = (),
    ) -> PreparedContext:
        ...

    async def prepare(
        self,
        *,
        context_sources: Sequence[ContextSource] = (),
        source_limit: int = 10,
        fail_open_sources: bool = False,
        user_input: str = '',
        documents: Sequence[Any] = (),
        **kwargs: Any,
    ) -> PreparedContext:
        ...

    def session(self, session_id: str) -> 'RuntimeSession':
        ...

    def clear_session(self, session_id: str) -> None:
        ...
```

## RuntimeSession

A conversation on a ContextRuntime: keeps a bounded message history and prepares each turn.

```python
class RuntimeSession:
    """A conversation on a :class:`ContextRuntime`: keeps a bounded message history and prepares each turn."""

    runtime: ContextRuntime
    session_id: str

    def record(self, role: str, content: str) -> None:
        ...

    async def prepare(self, user_input: str, **kwargs: Any) -> PreparedContext:
        ...
```

## PreparedContext

Everything prepared for one model turn: the plan, the model input, quality, trace and cache plan.

```python
class PreparedContext:
    """Everything prepared for one model turn: the plan, the model input, quality, trace and cache plan."""

    envelope: ContextEnvelope
    graph: ContextGraph
    plan: ContextPlan
    quality: QualityReport
    model_input: ModelInput
    trace: ContextTrace
    cache_plan: Optional[CachePlan] = None

    def with_provider_telemetry(self, telemetry: CacheTelemetry) -> 'PreparedContext':
        """Attach provider-reported usage without mutating the local estimate."""

    @property
    def feasible(self) -> bool:
        ...

    @property
    def cost(self) -> Mapping[str, Any]:
        ...

    @property
    def explanation(self) -> str:
        ...
```

## ModelInput

Provider-neutral request parts; no credentials or executable callbacks.

```python
class ModelInput:
    """Provider-neutral request parts; no credentials or executable callbacks."""

    model: str
    messages: Tuple[Mapping[str, Any], ...]
    tools: Tuple[Mapping[str, Any], ...]
    context: str
    message_stabilities: Tuple[str, ...] = field(default=(), repr=False)

    def to_dict(self) -> Dict[str, Any]:
        ...
```

## ContextTrace

Local diagnostics. Item IDs are hashed; prompt bodies are absent.

```python
class ContextTrace:
    """Local diagnostics. Item IDs are hashed; prompt bodies are absent."""

    trace_id: str
    session_hash: Optional[str]
    planner_version: str
    model: str
    original_tokens: int
    planned_tokens: int
    budget_tokens: int
    decisions: Tuple[Mapping[str, Any], ...]
    quality: QualityReport
    dependency_edges: int
    planning_latency_ms: float
    transformation_latency_ms: float
    validation_latency_ms: float
    estimated_cost_before: Optional[float]
    estimated_cost_after: Optional[float]
    cost_currency: Optional[str]
    warnings: Tuple[str, ...] = ()
    cache: Optional[CachePlan] = None
    provider_cache_telemetry: Optional[CacheTelemetry] = None

    @classmethod
    def from_plan(
        cls,
        envelope: ContextEnvelope,
        plan: ContextPlan,
        quality: QualityReport,
        graph: ContextGraph,
        exclusions: Sequence[str],
    ) -> 'ContextTrace':
        ...

    def to_dict(self) -> Dict[str, Any]:
        ...

    def explain(self) -> str:
        ...
```

## plan_context_v2()

Reduce context only under budget pressure, preserving known dependencies.

```python
def plan_context_v2(
    envelope: ContextEnvelope,
    *,
    max_input_tokens: Optional[int] = None,
    reserve_output_tokens: int = 4096,
    safety_margin_tokens: int = 256,
    quality_floor: float = 0.8,
    objective: str = 'balanced',
    context_policy: Optional[ContextPolicy] = None,
    model_profile: Optional[ModelProfile] = None,
    cache_policy: Optional[CachePolicy] = None,
    cached_stable_digests: Optional[Mapping[str, str]] = None,
) -> Tuple[ContextPlan, QualityReport, ContextGraph, Tuple[str, ...]]:
    """Reduce context only under budget pressure, preserving known dependencies.

    If no quality-safe step can satisfy the budget, return an explicit
    infeasible plan. The selected content is never silently truncated.
    """
```

## render_context_item()

Render one item with an explicit provenance boundary, as the planner puts it in the final context.

```python
def render_context_item(item: ContextItem, content: str) -> str:
    """Render one item with an explicit provenance boundary, as the planner puts it in the final context.

    The block is an ``<llmslim_context id="..." kind="..." role="..." trusted="...">`` element around ``content``.
    Attribute values are escaped, and untrusted content has any envelope tag
    neutralized (see :mod:`llmslim.planning.boundary`), so neither an item ID
    nor untrusted text can close the block or forge a trusted one. Trusted
    content, and untrusted content without an envelope tag, render as given.

    Args:
        item: The item whose id, kind and role label the block. Its role decides whether it is trusted.
        content: The text to render, for example ``item.content`` or a compressed form of it.

    Returns:
        The rendered block, followed by a blank line.
    """
```

## rendered_token_count()

Tokens that content costs in the context once rendered for item, boundary included.

```python
def rendered_token_count(item: ContextItem, content: str) -> int:
    """Tokens that ``content`` costs in the context once rendered for ``item``, boundary included.

    Args:
        item: The item ``content`` belongs to.
        content: The text to render, as for :func:`render_context_item`.

    Returns:
        The token count of the rendered block, from :func:`llmslim.count_tokens`.
    """
```

## generic_request()

Return detached Python request data for a caller-owned model adapter.

```python
def generic_request(prepared: PreparedContext) -> Dict[str, Any]:
    """Return detached Python request data for a caller-owned model adapter."""
```

## sarvam_messages()

Return messages accepted by the existing SarvamProvider.chat interface.

```python
def sarvam_messages(prepared: PreparedContext) -> list[Dict[str, str]]:
    """Return messages accepted by the existing SarvamProvider.chat interface."""
```

## openai_compatible_request()

Return OpenAI-style messages and authoritative schemas without calling a provider.

```python
def openai_compatible_request(prepared: PreparedContext) -> Dict[str, Any]:
    """Return OpenAI-style messages and authoritative schemas without calling a provider."""
```

## make_openai_agents_input_filter()

Build a RunConfig.call_model_input_filter for text-only SDK inputs.

```python
def make_openai_agents_input_filter(
    runtime: ContextRuntime,
    *,
    on_prepared: Optional[Callable[[PreparedContext], None]] = None,
) -> Callable[[Any], Any]:
    """Build a RunConfig.call_model_input_filter for text-only SDK inputs.

    The SDK retains agent execution and tool ownership. Complex SDK input
    items fail closed because rewriting them could break call/result contracts.
    That includes ``reasoning`` items (OpenAI's encrypted reasoning, or Claude
    thinking passed through a model adapter), so the filter never changes the
    history in front of reasoning that is replayed. SDK instructions are
    included in planning and returned unchanged.
    """
```

## CacheCapabilities

The prompt-caching features a provider offers.

```python
class CacheCapabilities:
    """The prompt-caching features a provider offers."""

    provider: str
    automatic_prefix_cache: bool = False
    explicit_prefix_cache: bool = False
    explicit_cached_object: bool = False
    conversation_state: bool = False
    client_managed_kv_cache: bool = False
    cache_usage_reporting: bool = False
    configurable_ttl: bool = False
    cache_prewarm: bool = False
    cache_breakpoints: bool = False
    verified: bool = True
```

## CacheManager

Tenant-scoped in-memory metadata only; no prompt bodies or KV tensors.

```python
class CacheManager:
    """Tenant-scoped in-memory metadata only; no prompt bodies or KV tensors."""

    def __init__(self, *, max_entries: int = 1024) -> None:
        ...

    def observe(
        self,
        cache_plan: CachePlan,
        policy: CachePolicy,
        session_id: Optional[str],
        model: str,
    ) -> CachePlan:
        ...

    def cached_stable_digests(
        self,
        policy: CachePolicy,
        session_id: Optional[str],
        model: str,
    ) -> Mapping[str, str]:
        ...

    def record_provider_reference(
        self,
        policy: CachePolicy,
        session_id: str,
        model: str,
        fingerprint: str,
        *,
        previous_response_id: Optional[str] = None,
        provider_cache_reference: Optional[str] = None,
    ) -> None:
        ...

    def clear(self, *, tenant_id: str, session_id: Optional[str] = None) -> None:
        ...
```

## CachePlan

The cacheable layout of one request: stable prefix, semi-stable and dynamic parts, and expected reuse.

```python
class CachePlan:
    """The cacheable layout of one request: stable prefix, semi-stable and dynamic parts, and expected reuse."""

    provider: str
    mode: str
    conversation_mode: str
    segments: Tuple[CacheSegment, ...]
    prefix_fingerprint: str
    stable_prefix_tokens: int
    semi_stable_tokens: int
    dynamic_suffix_tokens: int
    logical_context_tokens: int
    transmitted_input_bytes: int
    generation: int = 1
    status: CacheStatus = CacheStatus.COLD
    invalidation_reason: Optional[str] = None
    estimated_cache_read_tokens: int = 0
    estimated_cache_saving: Optional[float] = None
    estimated_effective_input_cost: Optional[float] = None
    provider_state_referenced_tokens: int = 0
    stateful_suffix_start: Optional[int] = None
    explanation: str = 'Provider cache usage has not been reported.'
    message_digests: Tuple[str, ...] = field(default=(), repr=False)
    previous_response_id: Optional[str] = field(default=None, repr=False)
    provider_cache_reference: Optional[str] = field(default=None, repr=False)

    def to_dict(self) -> dict[str, Any]:
        ...
```

## CachePolicy

How to plan prompt caching for a provider: mode, tenant, conversation mode, TTL and prices.

```python
class CachePolicy:
    """How to plan prompt caching for a provider: mode, tenant, conversation mode, TTL and prices."""

    mode: str = 'disabled'
    provider: str = 'generic'
    tenant_id: Optional[str] = None
    conversation_mode: str = 'stateless'
    ttl_seconds: Optional[int] = None
    cached_input_rate: Optional[float] = None
    uncached_input_rate: Optional[float] = None
    cache_write_rate: Optional[float] = None
    rendering_version: str = RENDER_VERSION
    request_settings: Mapping[str, Any] = field(default_factory=dict)

    @property
    def capabilities(self) -> CacheCapabilities:
        ...

    def validate_capabilities(self) -> None:
        ...
```

## CacheSegment

One context item in a cache plan, with its stability and position in the prompt.

```python
class CacheSegment:
    """One context item in a cache plan, with its stability and position in the prompt."""

    item_hash: str
    kind: str
    stability: str
    tokens: int
    digest: str
    position: int

    def to_dict(self) -> dict[str, Any]:
        ...
```

## CacheStatus

The state of a cache plan: estimated locally, or reported by the provider.

```python
class CacheStatus(str, Enum):
    """The state of a cache plan: estimated locally, or reported by the provider."""

    DISABLED = 'disabled'
    COLD = 'cold'
    ESTIMATED_REUSE = 'estimated_reuse'
    REPORTED_HIT = 'reported_hit'
    REPORTED_MISS = 'reported_miss'
    INVALIDATED = 'invalidated'
```

## CacheTelemetry

Token and latency usage of one request, estimated locally or reported by the provider.

```python
class CacheTelemetry:
    """Token and latency usage of one request, estimated locally or reported by the provider."""

    source: str = 'estimated'
    input_tokens: Optional[int] = None
    cache_read_tokens: Optional[int] = None
    cache_write_tokens: Optional[int] = None
    output_tokens: Optional[int] = None
    provider_cache_hit: Optional[bool] = None
    transmitted_input_bytes: Optional[int] = None
    time_to_first_token_ms: Optional[float] = None
    full_latency_ms: Optional[float] = None

    @property
    def uncached_input_tokens(self) -> Optional[int]:
        ...

    @property
    def cache_hit_ratio(self) -> Optional[float]:
        ...

    def to_dict(self) -> dict[str, Any]:
        ...
```

## PROVIDER_CAPABILITIES

The prompt-caching features of each known provider, by provider name.

```python
PROVIDER_CAPABILITIES: Mapping[str, CacheCapabilities] = ...
```

## estimate_cache_cost()

Rates are caller-supplied per million; read/write subsets never add twice.

```python
def estimate_cache_cost(
    input_tokens: int,
    read_tokens: int,
    write_tokens: int,
    policy: CachePolicy,
) -> Optional[float]:
    """Rates are caller-supplied per million; read/write subsets never add twice."""
```

## cache_cost_breakdown()

Disjoint compression and caching savings against the same raw baseline.

```python
def cache_cost_breakdown(
    original_tokens: int,
    planned_tokens: int,
    telemetry: CacheTelemetry,
    policy: CachePolicy,
) -> dict[str, Optional[float]]:
    """Disjoint compression and caching savings against the same raw baseline."""
```

## compile_cache_plan()

Fingerprint the provider-visible leading prefix without changing model semantics.

```python
def compile_cache_plan(
    plan: ContextPlan,
    model_input: ModelInput,
    policy: CachePolicy,
    material_settings: Optional[Mapping[str, Any]] = None,
) -> CachePlan:
    """Fingerprint the provider-visible leading prefix without changing model semantics.

    The prefix is the tools, in caller order, then the leading messages marked stable (by default, system and
    developer messages), together with the provider, model, rendering version, tenant and settings. The plan's
    segments say how stable each kept context item is. Nothing in ``model_input`` is changed.

    Args:
        plan: The context plan the request was built from.
        model_input: The request parts built from ``plan``.
        policy: The cache policy: mode, provider, tenant, conversation mode and prices.
        material_settings: Request settings that change what the provider caches, such as sampling settings.
            They are part of the fingerprint.

    Returns:
        The cache plan, with status ``DISABLED`` when ``policy.mode`` is ``"disabled"`` and ``COLD`` otherwise.

    Raises:
        ValueError: The policy asks for a caching feature the provider does not offer.
    """
```

## openai_responses_request()

Build a Responses API request; state continuation is opt-in and guarded.

```python
def openai_responses_request(
    prepared: PreparedContext,
    policy: CachePolicy,
    *,
    store: Optional[bool] = None,
) -> dict[str, Any]:
    """Build a Responses API request; state continuation is opt-in and guarded.

    A prior response is referenced only when the new input has exactly one
    appended user turn. This avoids silently duplicating assistant/tool output.
    """
```

## anthropic_messages_request()

Build Claude Messages parameters with plan-derived cache controls.

```python
def anthropic_messages_request(
    prepared: PreparedContext,
    policy: CachePolicy,
    *,
    max_tokens: int = 1024,
) -> dict[str, Any]:
    """Build Claude Messages parameters with plan-derived cache controls."""
```

## gemini_generate_request()

Build generate_content arguments, using only a tenant-scoped cache ref.

```python
def gemini_generate_request(prepared: PreparedContext, policy: CachePolicy) -> dict[str, Any]:
    """Build generate_content arguments, using only a tenant-scoped cache ref."""
```

## gemini_cached_content_request()

Build a Google Gen AI caches.create request, never issue it implicitly.

```python
def gemini_cached_content_request(
    prepared: PreparedContext,
    policy: CachePolicy,
    *,
    minimum_tokens: int = 2048,
) -> dict[str, Any]:
    """Build a Google Gen AI caches.create request, never issue it implicitly."""
```

## parse_cache_telemetry()

Read usage only; never inspect or store response text or provider secrets.

```python
def parse_cache_telemetry(
    provider: str,
    response: Any,
    *,
    transmitted_input_bytes: Optional[int] = None,
    time_to_first_token_ms: Optional[float] = None,
    full_latency_ms: Optional[float] = None,
) -> CacheTelemetry:
    """Read usage only; never inspect or store response text or provider secrets."""
```

## request_bytes()

UTF-8 JSON payload size, excluding transport headers and SDK framing.

```python
def request_bytes(request: Mapping[str, Any]) -> int:
    """UTF-8 JSON payload size, excluding transport headers and SDK framing."""
```

## vllm_chat_request()

Build OpenAI-client chat parameters for a host-owned vLLM server.

```python
def vllm_chat_request(
    prepared: PreparedContext,
    policy: CachePolicy,
    *,
    tenant_cache_salt: Optional[str] = None,
) -> dict[str, Any]:
    """Build OpenAI-client chat parameters for a host-owned vLLM server."""
```

## TransformersKVSession

Prepare and commit model forward calls using unchanged token prefixes.

```python
class TransformersKVSession:
    """Prepare and commit model forward calls using unchanged token prefixes.

    The caller must supply immutable model/tokenizer revision identifiers and
    must call ``clear`` after changing model weights, config, or chat template.
    """

    def __init__(
        self,
        *,
        model: Any,
        tokenizer: Any,
        model_revision: str,
        tokenizer_revision: str,
        renderer_version: str,
        tenant_id: str,
        cache_kind: str = 'dynamic',
        max_cache_len: Optional[int] = None,
    ) -> None:
        ...

    @property
    def cached_tokens(self) -> int:
        ...

    def prepare_forward(
        self,
        *,
        full_token_ids: Sequence[int],
        attention_mask: Sequence[int],
        position_ids: Sequence[int],
        device: str,
        model_revision: str,
        tokenizer_revision: str,
        renderer_version: str,
        tenant_id: str,
    ) -> dict[str, Any]:
        """Return only new token IDs and offsets after validating the cached prefix."""

    def commit_forward(self, *, past_key_values: Any) -> None:
        """Accept a host-produced cache only if its length matches the input."""

    def forward(
        self,
        *,
        full_token_ids: Sequence[int],
        attention_mask: Sequence[int],
        position_ids: Sequence[int],
        device: str,
        model_revision: str,
        tokenizer_revision: str,
        renderer_version: str,
        tenant_id: str,
    ) -> Any:
        """Run one local prefill/continuation and keep its KV cache in memory."""

    def abort_forward(self) -> None:
        ...

    def clear(self) -> None:
        ...
```

## Related

- [API reference: llmslim.memory](https://www.llmslim.app/docs/api-llmslim-memory.md)
- [Getting Started](https://www.llmslim.app/docs/getting-started.md)

Canonical page: https://www.llmslim.app/docs/api-llmslim