feat(inference): add context and output compression for inference requests

docs/docs/concepts/compression_configuration.mdx

@@ -0,0 +1,95 @@
+# Context & Output Compression
+
+## Overview
+
+OGX can automatically reduce the number of tokens sent to and generated by inference
+providers through the `CompressionConfig`. This addresses two related problems: growing
+conversation histories (especially in agentic/tool-calling workloads) push more and more
+tokens into every request, and requests that don't cap output length can generate more
+completion tokens than needed.
+
+Compression is applied uniformly to every chat completion request handled by
+`InferenceRouter` — both direct `/v1/chat/completions` calls and internal calls made by
+the Responses API orchestrator — regardless of which inference provider ultimately serves
+the request. It is **disabled by default**; enabling it changes nothing about a request
+until you opt in.
+
+## Configuration Structure
+
+Compression is configured through a single top-level `compression:` section in your run
+config (the same level as `server:` or `vector_stores:`), not per-provider:
+
+```yaml
+compression:
+  enabled: true
+
+  # Context compression: applied to outgoing messages before every inference call
+  max_context_tokens: 8000
+  dedupe_tool_outputs: true
+  max_tool_output_tokens: 1000
+  summarize_dropped_turns: true
+  summarization_model: null
+  summarization_prompt: "You are compressing older parts of a conversation..."
+
+  # Output token reduction
+  max_output_tokens: 1024
+
+  # Token counting
+  tokenizer_encoding: null
+  model_tokenizer_mappings:
+    llama: cl100k_base
+    mistral: cl100k_base
+```
+
+## Configuration Fields
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `enabled` | `bool` | `false` | Master switch. When `false`, none of the fields below have any effect. |
+| `max_context_tokens` | `int \| None` | `None` | Token budget for outgoing messages. When exceeded (after dedup/truncation below), the oldest conversation turns are dropped (and optionally summarized) until the request fits, or only the most recent turn remains. `None` disables windowing entirely. |
+| `dedupe_tool_outputs` | `bool` | `true` | Collapses exact-duplicate tool-call outputs (e.g. the same search result returned twice in one conversation) to a short placeholder instead of sending the full text twice. |
+| `max_tool_output_tokens` | `int \| None` | `None` | Truncates any single tool output exceeding this many tokens, appending a truncation marker. `None` disables this check. |
+| `summarize_dropped_turns` | `bool` | `false` | When turns are windowed out for exceeding `max_context_tokens`, summarize them via an LLM call and splice the summary back in as a system message, instead of silently dropping them. |
+| `summarization_model` | `str \| None` | `None` | Model used for the summarization call. Defaults to the same model as the original request. |
+| `summarization_prompt` | `str` | _(built-in prompt)_ | Prompt used to instruct the model when summarizing dropped turns. |
+| `max_output_tokens` | `int \| None` | `None` | Default cap applied to `max_tokens`/`max_completion_tokens`. If a request leaves both unset, this value is used; if a request sets either above this value, it's clamped down. `None` disables output capping. |
+| `tokenizer_encoding` | `str \| None` | `None` | Server-level default [tiktoken](https://github.com/openai/tiktoken) encoding name (e.g. `"o200k_base"`, `"cl100k_base"`) used for token counting when the model name doesn't resolve to a known encoding. |
+| `model_tokenizer_mappings` | `dict[str, str]` | common model families | Maps model name prefixes to tiktoken encoding names, used as a fallback when tiktoken can't resolve the model directly (e.g. local/fine-tuned models served via Ollama or vLLM). |
+
+## How It Works
+
+1. **Tool-output dedup and truncation** run first and are cheap — they never remove a
+   message outright (doing so would break the tool_call/tool_response pairing that
+   providers validate), only shrink its content.
+2. **Token counting** uses `tiktoken` when the model can be resolved to a known encoding,
+   falling back to a character-based estimate otherwise.
+3. **Windowing** only kicks in if the message list is still over `max_context_tokens`
+   after dedup/truncation. Messages are grouped into "turns" (a user message plus every
+   assistant/tool message that follows it, up to the next user message) so that a turn is
+   always dropped as a whole unit — this guarantees tool calls and their responses are
+   never split apart. The oldest turns are dropped first; the most recent turn is always
+   kept, even if it alone exceeds the budget.
+4. **Summarization**, if enabled, replaces the dropped turns with a single system message
+   containing an LLM-generated summary, via a request that bypasses compression itself
+   (so the summarization call isn't recursively compressed). If the summarization call
+   fails for any reason, the turns are dropped without a summary rather than failing the
+   original request.
+
+## Example: Long Agent Conversation
+
+```yaml
+compression:
+  enabled: true
+  max_context_tokens: 16000
+  dedupe_tool_outputs: true
+  max_tool_output_tokens: 2000
+  summarize_dropped_turns: true
+  max_output_tokens: 2048
+```
+
+With this configuration, a long-running agent conversation that accumulates many tool
+calls will have repeated tool outputs collapsed, oversized tool outputs truncated, and —
+once the running conversation exceeds 16,000 tokens — its oldest turns replaced with a
+short summary rather than either failing the request or silently losing context. Every
+outgoing request is also capped at 2,048 output tokens unless the caller explicitly asks
+for more (in which case it's clamped down to 2,048).

docs/sidebars.ts

@@ -83,6 +83,7 @@ const sidebars: SidebarsConfig = {
             'concepts/vector_stores_configuration',
           ],
         },
+        'concepts/compression_configuration',
         'concepts/distributions',
         'concepts/resources',
       ],

src/ogx/core/datatypes.py

@@ -11,6 +11,7 @@
 from typing import Annotated, Any, Literal, Self
 from urllib.parse import urlparse
 
+import tiktoken
 from pydantic import BaseModel, Field, SecretStr, field_validator, model_validator
 
 from ogx.core.access_control.datatypes import AccessRule, RouteAccessRule
@@ -715,6 +716,105 @@ class VectorStoresConfig(BaseModel):
     )
 
 
+DEFAULT_COMPRESSION_SUMMARIZATION_PROMPT = (
+    "You are compressing older parts of a conversation to save context space. Summarize the "
+    "conversation turns provided below into a concise handoff note for the same assistant to "
+    "continue the conversation. Preserve important facts, decisions, user preferences, and any "
+    "outstanding tasks. Omit pleasantries and redundant detail."
+)
+
+
+class CompressionConfig(BaseModel):
+    """Configuration for request-side context compression and output token capping,
+    applied uniformly to every inference call via InferenceRouter."""
+
+    enabled: bool = Field(
+        default=False,
+        description="Enable request-side context compression and output token capping. Disabled by default.",
+    )
+    max_context_tokens: int | None = Field(
+        default=None,
+        description=(
+            "Token budget for outgoing messages. When exceeded (after dedup/truncation), the oldest "
+            "conversation turns are windowed out (and optionally summarized). None disables windowing."
+        ),
+    )
+    dedupe_tool_outputs: bool = Field(
+        default=True,
+        description="Collapse exact-duplicate tool-call outputs to a short placeholder before sending to the model.",
+    )
+    max_tool_output_tokens: int | None = Field(
+        default=None,
+        description="Truncate any single tool output exceeding this many tokens. None disables truncation.",
+    )
+    summarize_dropped_turns: bool = Field(
+        default=False,
+        description="Summarize windowed-out turns via an LLM call instead of silently dropping them.",
+    )
+    summarization_model: str | None = Field(
+        default=None,
+        description="Model used to summarize dropped turns. If not set, uses the same model as the request.",
+    )
+    summarization_prompt: str = Field(
+        default=DEFAULT_COMPRESSION_SUMMARIZATION_PROMPT,
+        description="Prompt used to instruct the model to summarize windowed-out conversation turns.",
+    )
+    max_output_tokens: int | None = Field(
+        default=None,
+        description="Default cap applied to max_tokens/max_completion_tokens when the request leaves both unset.",
+    )
+    tokenizer_encoding: str | None = Field(
+        default=None,
+        description=(
+            "Default tiktoken encoding name for token counting (e.g. 'o200k_base', 'cl100k_base'). "
+            "Applied as a server-level default. If not set, encoding is resolved from the model name "
+            "via tiktoken, then model-family prefix mappings, then character-based estimation."
+        ),
+    )
+    model_tokenizer_mappings: dict[str, str] = Field(
+        default_factory=lambda: {
+            "llama": "cl100k_base",
+            "mistral": "cl100k_base",
+            "claude": "cl100k_base",
+            "gemma": "cl100k_base",
+            "qwen": "cl100k_base",
+            "phi": "cl100k_base",
+            "deepseek": "cl100k_base",
+        },
+        description=(
+            "Map model name prefixes to tiktoken encoding names. Used as a heuristic fallback when "
+            "tiktoken cannot resolve the model name directly. Matching is case-insensitive on the "
+            "model name after stripping any provider prefix (e.g., 'ollama/llama3.2:3b' matches 'llama')."
+        ),
+    )
+
+    @field_validator("tokenizer_encoding")
+    @classmethod
+    def validate_tokenizer_encoding(cls, v: str | None) -> str | None:
+        if v is not None:
+            try:
+                tiktoken.get_encoding(v)
+            except ValueError:
+                raise ValueError(
+                    f"Failed to resolve tokenizer_encoding '{v}'. "
+                    "Must be a valid tiktoken encoding name (e.g. 'o200k_base', 'cl100k_base')."
+                ) from None
+        return v
+
+    @field_validator("model_tokenizer_mappings")
+    @classmethod
+    def validate_model_tokenizer_mappings(cls, v: dict[str, str]) -> dict[str, str]:
+        for prefix, enc_name in v.items():
+            try:
+                tiktoken.get_encoding(enc_name)
+            except ValueError:
+                raise ValueError(
+                    f"Failed to resolve model_tokenizer_mappings['{prefix}'] = '{enc_name}'. "
+                    "Must be a valid tiktoken encoding name (e.g. 'o200k_base', 'cl100k_base')."
+                ) from None
+        return v
+
+
 class QuotaPeriod(StrEnum):
     """Time period for request quota enforcement."""
 
@@ -891,6 +991,11 @@ class StackConfig(BaseModel):
         description="Configuration for vector stores, including default embedding model",
     )
 
+    compression: CompressionConfig | None = Field(
+        default=None,
+        description="Configuration for request-side context compression and output token capping applied to all inference calls.",
+    )
+
     connectors: list[ConnectorInput] = Field(
         default_factory=list,
         description="List of connectors to register at stack startup",

src/ogx/core/routers/__init__.py

@@ -70,6 +70,7 @@ async def get_auto_router_impl(
         )
         await inference_store.initialize()
         api_to_dep_impl["store"] = inference_store
+        api_to_dep_impl["compression_config"] = run_config.compression
     elif api == Api.vector_io:
         api_to_dep_impl["vector_stores_config"] = run_config.vector_stores
         api_to_dep_impl["inference_api"] = deps.get(Api.inference)

src/ogx/core/routers/inference.py

@@ -15,9 +15,10 @@
 from pydantic import TypeAdapter
 
 from ogx.core.access_control.access_control import is_action_allowed
-from ogx.core.datatypes import ModelWithOwner
+from ogx.core.datatypes import CompressionConfig, ModelWithOwner
 from ogx.core.request_headers import get_authenticated_user
 from ogx.log import get_logger
+from ogx.providers.utils.inference.compression import maybe_compress_request
 from ogx.providers.utils.inference.inference_store import InferenceStore
 from ogx.telemetry.inference_metrics import (
     create_inference_metric_attributes,
@@ -85,10 +86,12 @@ def __init__(
         self,
         routing_table: RoutingTable,
         store: InferenceStore | None = None,
+        compression_config: CompressionConfig | None = None,
     ) -> None:
         logger.debug("Initializing InferenceRouter")
         self.routing_table = routing_table
         self.store = store
+        self.compression_config = compression_config
 
     async def initialize(self) -> None:
         logger.debug("InferenceRouter.initialize")
@@ -260,6 +263,25 @@ async def openai_chat_completion(
             params.tool_choice = None
             params.tools = None
 
+        if self.compression_config and self.compression_config.enabled:
+            summarize = None
+            if self.compression_config.summarize_dropped_turns:
+
+                async def summarize(request: OpenAIChatCompletionRequestWithExtraBody) -> OpenAIChatCompletion:
+                    # Call the provider directly rather than self.openai_chat_completion so
+                    # this internal summarization request isn't recursively compressed.
+                    summarize_provider, summarize_resource_id = await self._get_model_provider(
+                        request.model, ModelType.llm
+                    )
+                    request.model = summarize_resource_id
+                    response = await summarize_provider.openai_chat_completion(request)
+                    assert isinstance(response, OpenAIChatCompletion)
+                    return response
+
+            await maybe_compress_request(
+                params, self.compression_config, model_id=request_model_id, summarize=summarize
+            )
+
         if params.stream:
             response_stream = await provider.openai_chat_completion(params)