Add Caveman to Arize Phoenix: trace and grade compression

This guide adds Caveman Middleware to a Python agent that Arize Phoenix already traces. Phoenix stays exactly as it is: same register() call, same project, same collector. You wrap the agent, run your dataset once per mode, and let Phoenix's evaluators tell you whether the shorter tool output hurt the answers. For the background, read Caveman vs Arize Phoenix.

The example uses a LangChain agent, because that adapter handles recovery for you and Phoenix traces it out of the box. The same shape works for the other supported frameworks.

What do you need before you start?

You need a Phoenix instance (phoenix serve locally is fine), a Phoenix dataset of real inputs with expected answers, and an agent whose tools return something big: a file read, a JSON API response, a long log. Short tool results will show no change.

Check versions, because each adapter declines and passes the call through untouched outside its range. caveman-middleware needs Python 3.13 or newer. The LangChain adapter accepts langchain >=1.4 <2, langchain-core >=1.6 <2 and langgraph >=1.2 <2.

Step 1: Start the Caveman runtime

The middleware sends candidates to a local runtime that does the compression and keeps originals on disk. The Caveman CLI installs and starts it:

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

Plain caveman start runs in record mode and changes nothing, whatever the client asks for, so set compress here. The baseline arm will switch to record on the client side.

Step 2: Wrap the agent and keep Phoenix as it is

Install the packages next to your app:

pip install arize-phoenix-otel arize-phoenix-client openinference-instrumentation-langchain
pip install caveman-sdk "caveman-middleware[langchain]==0.1.0a1"

with_caveman_agent rewrites the keyword arguments you pass to create_agent. It adds the Caveman middleware and the caveman_retrieve tool, and returns a new dict:

import os
from phoenix.otel import register
from langchain.agents import create_agent
from caveman_cloud.middleware import MiddlewareRuntime, Scope
from caveman_middleware.langchain import with_caveman_agent

register(project_name="support-agent", auto_instrument=True)  # unchanged

runtime = MiddlewareRuntime(mode=os.environ.get("CAVEMAN_CLIENT_MODE", "compress"))
if runtime.mode != "off":
    runtime.ready()  # raises if the runtime is unreachable

def build_agent(model, tools, conversation_id):
    scope = Scope("support", conversation_id, "main", "0")
    return create_agent(**with_caveman_agent(
        {"model": model, "tools": tools}, runtime=runtime, scope=scope))

Scope is the identity of one conversation. Its four parts key the stored originals, so give each conversation its own id and keep it stable across turns.

If you call the OpenAI client directly instead, with_caveman_openai_tools from caveman_middleware.openai is the equivalent. Phoenix's OpenAI instrumentor records that client's shortened request body.

Step 3: Run one experiment per mode

Use Phoenix's own experiment runner, with the same dataset and the same evaluator for both arms:

import os, uuid
from phoenix.client import Client

px = Client()
dataset = px.datasets.get_dataset(dataset="support-tickets")
mode = os.environ["CAVEMAN_CLIENT_MODE"]

def task(example):
    agent = build_agent(model, tools, str(uuid.uuid4()))  # your model and tools
    result = agent.invoke({"messages": [{"role": "user", "content": example["input"]["question"]}]})
    return result["messages"][-1].content

def has_answer(output, expected):
    return expected["answer"].lower() in output.lower()

px.experiments.run_experiment(
    dataset=dataset,
    task=task,
    evaluators=[has_answer],
    experiment_name=f"caveman-{mode}",
    experiment_metadata={"caveman_mode": mode},
)

CAVEMAN_CLIENT_MODE=record python experiment.py
CAVEMAN_CLIENT_MODE=compress python experiment.py

Swap has_answer for whatever judge you already trust. The point is that both arms get the same one.

How do you check it worked?

Open both experiments in Phoenix and compare them example by example. Then check four things.

1. Scores. The compress experiment should hold the record experiment's evaluator results. A drop on specific examples tells you which tool output lost a detail the model needed.
2. Which copy the trace shows. Search the spans for caveman: shortened. In a LangChain agent the model span shows the shortened copy and the tool span shows the full original, since your graph state keeps it.
3. Recovery calls. Count caveman_retrieve tool spans. Each one carries the original back and costs tokens, so include them in the total.
4. Tokens. Compare provider-reported input tokens across the whole task, retries included. Caveman's local estimates are labelled inferred and are not a bill.

To see the per-call decision, pass on_report=print to MiddlewareRuntime. You want applied or reused. A skipped report carries a reason such as no_candidate or unsupported_version, listed in the middleware docs.

How do you roll back?

Remove the wrapper: create_agent(model=model, tools=tools). with_caveman_agent returns a new dict and mutates nothing, so the agent is back to what it was. To switch it off without a deploy, set CAVEMAN_CLIENT_MODE=off. Phoenix never changed, so there is nothing to restore there, and the baseline experiment stays in Phoenix for later comparisons.

Keep the runtime up until live conversations finish, since a handle the model already holds resolves only against stored originals. Grants last 24 hours from their last use.

Where to go next

• Caveman vs Arize Phoenix for which layer sees what
• Python middleware docs for every Python adapter
• Phoenix client for datasets and experiments
• Agent evaluations for designing the judge