How to add Caveman Middleware to a LangGraph agent

You keep LangGraph, your checkpointer and your tools. Caveman goes in as one more entry in the create_agent middleware list, plus a small local runtime. The steps below use Python; the TypeScript adapter has the same shape and is shown in step 4. For the background, read Caveman vs LangGraph.

What you need

• Python 3.13 or newer with langchain >=1.4 <2, langchain-core >=1.6 <2 and langgraph >=1.2 <2. For TypeScript: langchain >=1.5 <2, @langchain/core >=1.2 <2, @langchain/langgraph >=1.4 <2.
• Node.js 22.13 or newer for the caveman CLI that runs the runtime.
• An agent built with create_agent whose tools return a lot of text. A hand-built StateGraph can use with_caveman_model, but that path has no recovery tool.

Step 1: Record a baseline

Run one representative task a few times as it is today, on a fixed model. Save the final answer, whether it passed your check, and the usage_metadata on each AIMessage in the result.

Step 2: Install the packages

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

caveman-middleware only has prereleases so far, so pin the version or pass --pre. The runtime client ships 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: Wrap create_agent

Build one runtime for the process. Then pass your create_agent arguments through with_caveman_agent. It returns a new dict with the Caveman middleware added to middleware and caveman_retrieve added to tools:

from dataclasses import asdict
from caveman_cloud.middleware import MiddlewareRuntime
from caveman_middleware.langchain import with_caveman_agent, scope_from_config
from langchain.agents import create_agent

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()

agent = create_agent(**with_caveman_agent(
    {"model": model, "tools": tools, "checkpointer": checkpointer},
    runtime=runtime,
    scope=lambda config: scope_from_config(config, namespace="support")))

agent.invoke({"messages": messages},
    {"configurable": {"thread_id": conversation_id}, "recursion_limit": 12})

scope can be a fixed Scope or a function of the run config. scope_from_config reads configurable.thread_id, so the thread your checkpointer resumes and the session Caveman stores originals under are the same thing. It raises if thread_id is missing. Keep the namespace to your tenant or app, never shared across end users.

In TypeScript the call is createAgent(withCavemanAgent({ model, tools }, { runtime, scope })), with the runtime from createMiddlewareRuntime in @caveman-ai/sdk/middleware and scopeFromConfig(config, namespace) for the scope.

Two more things:

• Leave room for recovery. A caveman_retrieve call is one more model and tool round. Keep some headroom in recursion_limit.
• Do not register your own caveman_retrieve. The adapter skips adding its tool if one with that name already exists.

Step 5: Check it worked

The on_report callback prints one decision per model call:

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

Then confirm three things against the baseline. The checkpointed state still holds full ToolMessage content. The task passes the same check. Provider input tokens across the whole run are lower, with recovery calls and retries counted in.

If nothing shrinks, the report's reason says why: runtime_unavailable (endpoint did not answer), recorded (runtime in record mode) or unsupported_version (a LangChain package outside the gate).

The Python middleware docs include quickstart.py, which runs the same agent against a scripted model and checks compression and exact recovery without any provider call.

How to roll back

• Pass your original dict to create_agent instead of the wrapped one. with_caveman_agent returns a new dict and mutates nothing.
• Or construct the runtime with mode="off". The middleware 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 threads already in flight can still recover.

Next steps

Measure finished tasks, not single calls: how to measure agent cost. The deployment page covers running the runtime beside a deployed app and keeping scope stable across restarts.