Add Caveman to Langfuse: setup and quality check

You do not move anything out of Langfuse. Caveman Middleware goes into the agent code, Langfuse keeps tracing, and a Langfuse dataset experiment tells you whether the shorter tool output changed any answers. This guide uses a Python LangChain agent with the Langfuse CallbackHandler, which is the setup where the trace shows exactly what the model read. For how the two tools fit together, see Caveman vs Langfuse.

1. Install and start the runtime

The Python packages need Python 3.13 or newer. The runtime is installed by the Caveman CLI, which needs Node.js 22.13 or newer.

pip install langfuse caveman-sdk "caveman-middleware[langchain]==0.1.0a1"
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}'

mode must say compress. Plain caveman start runs record, which measures and changes nothing.

2. Wrap the agent and keep the Langfuse handler

with_caveman_agent takes the keyword arguments you already pass to create_agent and returns a new dict with Caveman's agent middleware and the caveman_retrieve recovery tool added. The Langfuse handler goes in the run config as before.

import os, uuid
from langchain.agents import create_agent
from langfuse import get_client, Evaluation
from langfuse.langchain import CallbackHandler
from caveman_cloud.middleware import MiddlewareRuntime, Scope
from caveman_middleware.langchain import with_caveman_agent

MODE = os.environ.get("CAVEMAN_CLIENT_MODE", "record")  # "record" = baseline, "compress" = candidate, "off" = rollback
runtime = MiddlewareRuntime(endpoint="http://127.0.0.1:8787", mode=MODE)
if MODE != "off":  # ready() raises in off mode
    runtime.ready()
langfuse = get_client()

def my_task(*, item, **kwargs):
    # One scope per conversation. Recovery handles only resolve under the same scope.
    scope = Scope("eval", str(uuid.uuid4()), "main", "0")
    agent = create_agent(**with_caveman_agent(
        {"model": model, "tools": tools},  # your existing model and tools
        runtime=runtime, scope=scope))
    result = agent.invoke(
        {"messages": [{"role": "user", "content": item.input}]},
        {"callbacks": [CallbackHandler()], "recursion_limit": 12},
    )
    return result["messages"][-1].content

In production, build the scope from your real conversation ID so the model can recover originals across turns. scope_from_config(config, namespace=...) does that from a LangGraph thread_id.

3. Run the same experiment twice

Make a Langfuse dataset of 20 to 50 tasks where a big tool result carries the answer: a failing test buried in a long CI log, one field in a large JSON response, a clause deep in an extracted PDF. Give each item an expected output. Then add an evaluator and run:

def contains_expected(*, output, expected_output, **kwargs):
    ok = bool(expected_output) and expected_output.lower() in output.lower()
    return Evaluation(name="contains_expected", value=1.0 if ok else 0.0)

dataset = langfuse.get_dataset("heavy-tool-output")
dataset.run_experiment(name=f"caveman-{MODE}", task=my_task, evaluators=[contains_expected])
langfuse.flush()
runtime.close()

CAVEMAN_CLIENT_MODE=record python trial.py
CAVEMAN_CLIENT_MODE=compress python trial.py

record sends every byte unchanged, so the baseline runs through the same code path as the candidate. Swap the string match for an LLM-as-a-judge evaluator if your answers are free text.

4. Check it worked

Open the two dataset runs side by side in Langfuse and read four things:

• Scores. The compress run should match the record run. Open every item that flipped; an average can hide one bad regression.
• The LLM generation input. In the compress run, shortened tool results start with [caveman: shortened; exact original via caveman_retrieve handle=…]. With the LangChain adapter this is the copy the model actually received. If you never see it, nothing was compressed.
• Input tokens for the whole task. Compare totals per item, not the first call. A caveman_retrieve call is an extra round trip that carries the original back.
• Recovery calls. Count caveman_retrieve tool spans. A few are healthy. The model asking for the original on almost every item means the short view is dropping what your tasks need.

For the local decision on each call, pass on_report=print to MiddlewareRuntime. Statuses read applied, reused, recorded or skipped with a reason such as no_candidate.

If you use the Langfuse OpenAI drop-in instead

With a raw OpenAI client you own the tool loop, so use with_caveman_openai_tools. It adds caveman_retrieve to your tool definitions and gives you a dispatch table that runs it. The plain with_caveman_openai wrapper has no recovery executor, and the runtime will not shorten text it cannot give back, so on its own it sends every call unchanged.

from langfuse.openai import OpenAI
from caveman_middleware.openai import with_caveman_openai_tools

loop = with_caveman_openai_tools(OpenAI(), runtime=runtime, scope=scope,
    protocol="openai-chat", tools=tools, functions=functions)  # functions: {tool name: callable(args dict)}
response = loop.client.chat.completions.create(model=model_name, messages=messages, tools=loop.tools)
# Run every tool call the model makes through loop.functions[name](arguments), caveman_retrieve included.

Send loop.tools on every request; a call without the recovery tool goes out uncompressed. The Python OpenAI adapter acts only on openai >=3.10 <4, so confirm your Langfuse SDK supports that OpenAI version before relying on this path.

Read these traces differently. Caveman rewrites the HTTP request below chat.completions.create, the method Langfuse patches, so the generation input shows your original messages while the usage shows what the provider billed for the shorter request. For streamed calls, set stream_options={"include_usage": True}; otherwise Langfuse tokenizes the recorded original and overcounts.

Roll back

• Fastest: run with CAVEMAN_CLIENT_MODE=off, which gives the runtime client mode="off". Nothing reaches the runtime and calls pass through untouched.
• Code: delete the with_caveman_agent(...) call and pass your original dict to create_agent. The wrapper returned a new dict and mutated nothing, so behavior is exactly as before.
• Runtime: restart it without CAVEMAN_MODE=compress and it goes back to record.

Langfuse needs no change in any of these. Your traces, prompts, datasets and scores stay where they were.

More: Python middleware docs, supported frameworks, Langfuse experiments via SDK, Langfuse LangChain integration, and the agent evaluations guide.