driftcast

Cost and trace telemetry for agent frameworks. Captures actual tokens, cost, latency, and errors as your agent runs, persisted to a local SQLite file.

You own your data. DriftCast is content-agnostic by default: it records the shape and economics of execution (token counts, cost, latency, status, structure) — never your prompts, completions, or documents. Everything is written to a local file you control; there is no DriftCast server in the data path. Unlike cloud-coupled tracers (which go dark under Zero Data Retention policies), there is nothing to switch off. See Data ownership.

One decorator, one explicit call. @lens.track turns any function into a tracked run or span automatically — nesting into a call-tree on its own. The only thing you pass by hand is token usage (driftcast.record(...)), because provider response shapes differ across providers and call types (embeddings vs. chat completions). Everything else — cost lookup, latency, structure, persistence — is automatic.

Install

pip install driftcast

Optional local dashboard (Gradio):

pip install "driftcast[dashboard]"

The core SDK is stdlib-only. driftcast[dashboard] adds the local Gradio viewer (driftcast dashboard). Claude Code capture via the OTLP receiver (driftcast-otel) is included in the core install.

Build from source:

git clone https://github.com/AkashRK1216/driftcast
cd driftcast
pip install -e ".[dashboard]"

Usage

Decorate your functions with @lens.track. The outermost decorated call becomes a run (one full pipeline execution); nested decorated calls become spans (individual provider calls), auto-parented into a tree. Inside a span, call driftcast.record(...) once to report token usage.

import driftcast

lens = driftcast.init(project="rag-agent", db_path="./driftcast.db")

@lens.track(model="text-embedding-3-small")
def embed_query(query):
    result = openai_client.embeddings.create(model=EMBED_MODEL, input=query)
    driftcast.record(input_tokens=result.usage.prompt_tokens, output_tokens=0)
    return result

@lens.track(model="gpt-4o-mini")
def generate_answer(query):
    response = openai_client.chat.completions.create(...)
    driftcast.record(
        input_tokens=response.usage.prompt_tokens,
        output_tokens=response.usage.completion_tokens,
    )
    return response.choices[0].message.content

@lens.track                              # the top-level call is the run
def ask(query):
    driftcast.annotate(customer_id="acme")   # business labels onto the run
    embed_query(query)
    return generate_answer(query)

ask("what is the refund policy?")

• @lens.track(model=None, name=None) — the whole API. The outermost decorated call opens a run; nested decorated calls become spans, auto-nested by call depth. On exit each records cost, latency_ms, and status; an exception is recorded as status="error" and re-raised — tracing never masks a real failure. Works on sync and async functions.
• driftcast.record(input_tokens, output_tokens, content=None) — call once inside a @lens.track(model=...) function to report what the provider call consumed. This is the one number you pass by hand (provider usage shapes differ). Pass content= to persist prompt/response only when capture_content=True (see Data ownership).
• driftcast.annotate(**labels) — attach business labels (route, customer_id, judge verdicts…) to the current run's metadata. Always stored, never treated as content.
• driftcast.outcome(accepted, wasted=[...]) — at the end of a run, record your own ground-truth label (what you accepted / what was wasted). Content-free; persisted into the run's metadata.

Data ownership

driftcast.init(..., capture_content=False) is the default. In that mode:

• Content passed via content= is dropped before persistence. Token counts, cost, latency, and structure are still recorded — enough for cost attribution and tracing, with zero prompt/response data at rest.
• Error messages are reduced to the exception type (e.g. RateLimitError), since provider errors can echo input content. The full message is kept only when content capture is on.
• Metadata is stored separately from content, so per-customer attribution (customer_id=...) never requires storing a prompt.

Set capture_content=True to also persist content= payloads for debugging — written only to your local db_path, never transmitted anywhere. This is the design that lets DriftCast run under Zero Data Retention policies where cloud-coupled tracers cannot.

CLI

driftcast summary --db ./driftcast.db [--project rag-agent]

Prints an aggregate report grouped by run and by model — total cost, total tokens, run count, average latency.

Live dashboard

A web dashboard renders the same data as a live, auto-refreshing view (headline totals, runs, per-model cost/tokens/latency). Gradio is an optional extra — the core SDK stays stdlib-only.

pip install -e ".[dashboard]"           # installs gradio
driftcast dashboard --db ./driftcast.db --project rag-agent --port 7861

To pop it automatically alongside an agent's own UI, launch it non-blocking:

import driftcast.dashboard as dashboard

# returns immediately; server runs in a background thread on its own port
dashboard.launch(db_path="./driftcast.db", project="rag-agent", port=7861, block=False)

The RAG test agent does exactly this — running python main.py opens the chat UI and the stats dashboard side by side (dashboard on port 7861, override with DRIFTCAST_DASHBOARD_PORT).

Pricing

src/driftcast/pricing.py is a plain editable $ per 1M tokens dict. Unknown models cost $0.0 and log a warning, so untracked spend is a visible signal to add the model rather than a silent miscalculation.

Storage

SQLite via stdlib sqlite3 — file-based, no extra dependency, matches the "under $100, personal dogfood" scale this targets. Two tables: runs (one row per pipeline execution) and spans (one row per provider call).