How to add token compression to Pydantic AI agents

Your Pydantic AI agent keeps its model, tools, dependencies and output types. Caveman goes in as one more entry in capabilities=[...], plus a small local runtime. For the background, read Caveman vs Pydantic AI.

What you need

• Python 3.13 or newer and pydantic-ai-slim >=2.42 <3. Outside that range the capability declines, reports unsupported_version, and passes requests through untouched.
• A model built on Pydantic AI's OpenAIChatModel or AnthropicModel. Other model classes pass through untouched and report unsupported_provider.
• Tools that return plain strings. A dict or Pydantic model return is sent as is.
• Node.js 22.13 or newer for the caveman CLI that runs the runtime.
• An agent that answers in plain text and whose tools return a lot of text. The recovery tool is only offered on text-output runs with automatic tool choice, so start there rather than on a structured output_type agent.

Step 1: Record a baseline

Run one representative task a few times as it is today, on a fixed model. Save the output, whether it passed your check, and result.usage() for each run.

Step 2: Install the packages

pip install caveman-sdk "caveman-middleware[pydantic-ai]==0.1.0a1"

caveman-middleware only has prereleases so far, so pin the version or pass --pre. The runtime client lives in caveman-sdk as caveman_cloud.middleware.

Step 3: Start the local runtime

npm install -g @caveman-ai/cli && caveman setup --install
CAVEMAN_MODE=compress caveman start

Check it from another shell:

curl -s http://127.0.0.1:8787/caveman/v1/middleware/capabilities | jq '{mode, persistent, recovery, retention_seconds}'

You want "mode": "compress" and "recovery": true. Plain caveman start runs record, which measures and changes nothing.

Step 4: Add the capability

The docs example is one line on your Agent:

from pydantic_ai import Agent
from caveman_middleware.pydantic_ai import CavemanCapability

def build(*, model, runtime, scope):
    return Agent(model, capabilities=[CavemanCapability(runtime=runtime, scope=scope)])

Around it you need a runtime and a scope. The runtime is created once per process. CavemanCapability accepts the regular MiddlewareRuntime and converts it to an async view itself:

from dataclasses import asdict
from caveman_cloud.middleware import MiddlewareRuntime, Scope

runtime = MiddlewareRuntime(endpoint="http://127.0.0.1:8787", mode="compress",
    deadline_ms=500, on_report=lambda report: print(asdict(report)))
if runtime.mode != "off":  # ready() raises in off mode
    runtime.ready()

The scope is four strings, Scope(namespace, session_id, branch_id, cache_epoch), and it keys the stored originals. A recovery handle only resolves under the same four, so tie session_id to your conversation. You can pass a fixed Scope, or a function that receives the run context and is resolved once per run. The adapter ships scope_from_run(ctx, namespace=...), which uses the run's native conversation_id. Pydantic AI generates one on the first run and carries it through message_history; pass conversation_id=... to agent.run to use your own:

from caveman_middleware.pydantic_ai import CavemanCapability, scope_from_run

agent = Agent(model, tools=tools, capabilities=[CavemanCapability(
    runtime=runtime, scope=lambda ctx: scope_from_run(ctx, namespace="support"))])

Keep namespace to your tenant or app, never shared across end users. If the model decides to recover an original, that is one extra tool call, so leave headroom in any request limits you set.

Step 5: Check it worked

The on_report callback prints one decision per model request:

• First request: skipped with no_candidate. No tool has returned yet.
• After the first large tool return: applied, with a transform id such as caveman.engine.log.v1.
• Later requests: reused. The same shortened bytes go out again, so the provider's cached prefix holds.

Then compare against the baseline. The agent's messages still hold the full tool returns. The task passes the same check. Provider input tokens in result.usage() are lower across the whole run, with any recovery calls counted in.

If nothing shrinks, read the reason. runtime_unavailable means the endpoint did not answer. recorded means the runtime is in record mode. unsupported_version means pydantic-ai-slim is outside the gate. unsupported_provider means the model is not OpenAIChatModel or AnthropicModel. Requests with structured output go out uncompressed.

How to roll back

• Remove CavemanCapability(...) from the capabilities list. Nothing else changed.
• Or construct the runtime with mode="off". The capability reports disabled and sends the original messages.

Call runtime.close() at shutdown. The runtime process keeps stored originals for 24 hours after their last use, so conversations already in flight can still recover.

Next steps

Judge the change on finished tasks: how to measure agent cost. If you trace with Logfire, compare usage there per run. The Python middleware docs cover the runtime options, and the deployment page covers running the runtime beside a deployed app.