proveai-sdk 0.3.3

Testing framework for headless multi-agent LLM pipelines — capture, snapshot, mock, verify.

ProveAI SDK

Testing framework for multi-agent LLM pipelines. Capture real traces, snapshot the known-good I/O, and let proveai snapshot verify become your CI gate. No backend service, no eval set to maintain — one pip install, one CLI binary, zero hosted infra.

The 30-minute activation arc:

1. pip install proveai-sdk
2. Decorate 3–5 agents with @monitor.
3. Capture one trace with Trace(...).
4. proveai snapshot init — pins each agent's prompt + I/O + tool calls behind content hashes at proveai/snapshots/latest.json, plus the invocation context (interpreter, argv, cwd) so verify can re-run later.
5. Edit prompts / models / agent code as you iterate.
6. proveai snapshot verify — re-runs your pipeline against the pinned snapshot, grades each agent (unchanged / drifted / regressed), and exits 1 on regression. Drop it into pre-commit or GitHub Actions. (Like pytest -u for agents.)

The legacy capture-and-replay flow (proveai run, proveai baseline, proveai experiment) is still here for cohort-scale regression testing; the snapshot surface above is the headline.

Deeper docs: INSTRUMENTATION.md covers every step in detail (provider registration, @monitor topology + kinds, the LangChain Runnable adapter, per-agent snapshots with downstream auto-target expansion, common gotchas). Runnable companion: examples/instrumentation.py.

---

Quick Start — 30 minutes from install to CI gate

1. Install + decorate

pip install proveai-sdk        # or: pip install 'proveai-sdk[pytest]'

from proveai import monitor, Trace, FileExporter

@monitor("classify", model="gpt-4o-mini")
def classify(ticket: str) -> str:
    return llm.invoke(f"Classify: {ticket}")

@monitor("draft", parent_agent="classify", model="gpt-4o-mini")
def draft(category: str) -> str:
    return llm.invoke(f"Draft a reply for: {category}")

@monitor("qa_check", parent_agent="draft", model="claude-haiku-4-5")
def qa_check(reply: str) -> str:
    return llm.invoke(f"QA the reply: {reply}")

2. Capture a trace

with Trace(exporters=[FileExporter()]) as t:
    cat = classify("My order arrived broken")
    reply = draft(cat)
    qa_check(reply)

OTLP JSON + manifest + outcomes land under ./proveai/traces/.

3. Pin it as a snapshot

$ proveai snapshot init
Wrote snapshot 'latest' v1 — 3 agent(s) · trace 84a9dced82e2…

4. Wire it into CI

Three steps. The whole .github/workflows/proveai.yml is auto-generated:

# 1. Commit the snapshot file (it's small, byte-stable, reviewable in PRs).
git add proveai/snapshots/latest.json
git commit -m "pin snapshot"

# 2. Scaffold the workflow from the embedded template.
proveai ci init
# Wrote .github/workflows/proveai.yml
# Next: Add a repo secret named `OPENROUTER_API_KEY`.

# 3. Add the secret to the repo (Settings → Secrets and variables → Actions).

Push the branch. The workflow runs proveai snapshot verify on every pull request. With no arguments it walks every snapshot in proveai/snapshots/ locally; in CI it auto-targets to only the snapshots whose agents the PR touched (resolved via git diff $GITHUB_BASE_REF...HEAD against each agent's captured source file). A 10-agent pipeline where the PR edited one prompt pays 1/10 of a full re-verify.

The verdict table lands on the workflow run page (via $GITHUB_STEP_SUMMARY) so the reviewer sees it one click from the PR:

Agent	Verdict	Change	Rationale
classify	✓ unchanged	—	—
draft	~ drifted	downstream	judge: equivalent rewording
qa_check	✗ regressed	direct edit	judge: lost the security context

Exit 0 when no agent regressed; exit 1 on any regression (drift alone stays warning-only) — blocks the merge. The Tier-3 LLM judge classifies the gray zone (equivalent → drifted, worse → regressed); verdicts cached per (old_hash, new_hash) pair so re-runs cost nothing. The (direct edit) / (downstream) tag distinguishes root-cause edits from downstream collateral so you can fix the cause and not chase the symptoms.

Other provider? proveai ci init --provider {anthropic|openai|azure-openai|bedrock|custom} swaps the secret block. --full opts out of auto-targeting (good for weekly cron sweeps). --changed previews the CI scope locally without setting CI=true. Intentional drift? Run proveai snapshot update and commit the updated snapshot in the same PR.

verify actually re-runs your pipeline. As of v0.3.0, the default snapshot verify re-invokes the same command you originally captured under (recorded silently at Trace.__enter__), captures a fresh trace in a temp dir, and diffs it against the pinned snapshot. So when you edit a prompt or swap a model and run proveai snapshot verify, you see exactly what changed and what propagated. Cost is proportional to the change footprint — unchanged agents serve from the response cache for free. For cheap fixture-only smoke checks (or pipelines too expensive to re-run on every commit), pass --quick. For "what-if" prompt testing without editing your MAS code, pass --override AGENT=path/to/new_prompt.md. See CLI.md for the full migration table.

Bonus — pytest

from proveai.pytest import snapshot

@snapshot.test("qa_check")
def test_qa_check_stable(snap):
    snap.assert_no_regression("ok")

Same judge, same cache, fails with the judge rationale in the pytest output when an agent regresses.

---

Installation

# From PyPI
pip install proveai-sdk

# With optional extras
pip install 'proveai-sdk[pytest]'             # pytest plugin (@snapshot.test)
pip install 'proveai-sdk[embeddings-local]'   # local Tier-2 embeddings, no API key

# From source (development)
cd proveai-sdk
pip install -e .              # or: uv sync --all-extras for dev/test tooling

Core dependencies: httpx, pydantic, jsonschema, rich, jinja2, markupsafe, pyyaml. Optional: sentence-transformers (local embeddings, ~50ms/call, no API key).

Configure the LLM endpoints

Capture and replay both need to know where to send LLM calls. Drop a .env next to your script (or export these in your shell). A typical setup pointing at OpenRouter for chat + Ollama for local embeddings:

# .env

# --- Tier-1 minimum: an OpenAI-compatible chat endpoint --------------
# Required so `replay_trace` and `llm_judge` know where to dispatch the
# captured `accounts/fireworks/models/...` model. Without these, replay
# raises ValueError("No provider registered for model …").
PROVEAI_PROVIDER_PATTERN=accounts/fireworks/models/*,deepseek-*
PROVEAI_PROVIDER_BASE_URL=https://api.fireworks.ai/inference/v1
FIREWORKS_API_KEY=fw_your_key_here     # auto-discovered via fallback chain

# --- Pin the Tier-3 judge to your endpoint ---------------------------
# Without this, llm_judge defaults to claude-haiku-4-5 and every replay
# emits "judge skip: ANTHROPIC_API_KEY not set" rationales. Setting it
# routes the judge through the same model the rest of the pipeline uses.
PROVEAI_JUDGE_MODEL=accounts/fireworks/models/deepseek-v3p1

# --- Optional: Tier-2 embeddings via Ollama --------------------------
# Skip this and text mismatches escalate from Tier 1 directly to the
# judge. Wiring up a local embedder is faster (~50ms vs cents per call)
# and keeps the diff stack offline for prose-equivalence checks.
PROVEAI_EMBEDDER_MODEL=nomic-embed-text:latest
PROVEAI_EMBEDDER_BASE_URL=http://localhost:11434/v1
PROVEAI_EMBEDDER_API_KEY=ollama

Why each one is required. The SDK ships built-in providers only for gpt-* (OpenAI) and claude-* (Anthropic). Anything else — Fireworks, Together, Groq, Anyscale, Ollama, vLLM, your own deployment — needs to register with the dispatch registry, and the env-driven path is the zero-Python way to do it. For OpenAI / Anthropic users, just set OPENAI_API_KEY or ANTHROPIC_API_KEY and skip the PROVEAI_PROVIDER_* section entirely.

The variables are read on import (and again on every reset_providers()), so any script or notebook that loads .env before importing proveai gets the registrations automatically. python-dotenv works fine; rolling your own _load_local_env() helper is also common (see examples/instrumentation.py).

---

Quick Start — Capture

Decorate your agents, group them in a Trace context, write to disk via FileExporter. Optional Checks auto-label each trace as good / bad / skip.

from proveai import (
    monitor, trace, FileExporter,
    assert_json, regex_match, tool_called, LLMJudge,
)

@monitor("classify", model="gpt-4o-mini", kind="CHAT_MODEL", version="v3")
def classify(ticket: str) -> str:
    return llm.invoke(f"Classify: {ticket}")

@monitor("crm_lookup", parent_agent="classify", kind="TOOL")
def crm_lookup(category: str) -> dict:
    return crm.fetch(category)

@monitor("draft", parent_agent="crm_lookup", model="gpt-4o-mini")
def draft(category: str, history: dict) -> str:
    return llm.invoke(...)

@monitor("qa_check", parent_agent="draft", model="claude-haiku-4-5")
def qa_check(draft_text: str) -> str:
    return llm.invoke(f"Critique: {draft_text}")

exporter = FileExporter("./traces")

with trace("ticket-triage", exporters=[exporter], checks=[
    assert_json("classify"),
    tool_called("crm_lookup"),
    regex_match("draft", r"\[\d+\]"),               # citation marker
    LLMJudge("qa_check", "must catch obvious errors"),
]) as t:
    cat = classify("My order arrived broken")
    history = crm_lookup(cat)
    response = draft(cat, history)
    qa_check(response)

After the with block exits, ./traces/ contains:

./traces/
├── manifest.json            # trace_id → {version, agents, timestamp}
├── outcomes.jsonl           # per-check + composite labels
└── <trace_id>.json          # OTLP JSON for each captured trace

---

Quick Start — Replay (CLI)

Captured traces are executable artifacts. Replay them against new code, a swapped prompt, or a different model, and get per-edge verdicts.

The named-artifact flow uses baselines (trace sets) and experiments (override sets) instead of inline filter+override flags:

# Browse what's in the trace store, with detection status
# (detection is written automatically at capture — no analyze step)
proveai status --filter detection=flagged --limit 10

# Snapshot the failure cohort as a versioned baseline
proveai baseline create reviewer-failures --filter detection=flagged

# Save the candidate fix as a named experiment
proveai experiment new fix-qa-check --override qa_check=prompts/qa_check_v2.md

# Replay the baseline through the experiment, write the report,
# fail CI (exit 1) if any edge regressed
proveai run --baseline reviewer-failures --experiment fix-qa-check

# OR — isolate one captured span (e.g. a single loop iteration) and re-run
# only that span live, holding everything else to recorded outputs
proveai run --hold-at <span_id> --override qa_check=prompts/qa_check_v2.md

The legacy inline form still works for one-off use:

proveai run \
  --filter outcome=good \
  --override qa_check=prompts/qa_check_v2.md \
  --report report.html

Sample summary table (fits in 80 cols):

                  3 trace(s) → report.html
┏━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Trace    ┃ Edge                       ┃ Verdict      ┃ Rationale     ┃
┡━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ a1b2c3d… │ classify→crm_lookup        │ unchanged    │ JSON deep-eq… │
│ a1b2c3d… │ crm_lookup→draft           │ unchanged    │ TOOL call de… │
│ a1b2c3d… │ draft→qa_check             │ regressed    │ judge: lost … │
│ b4d6f8a… │ draft→qa_check             │ regressed    │ judge: missi… │
└──────────┴────────────────────────────┴──────────────┴───────────────┘

Per-edge totals:
  classify→crm_lookup: 3 unchanged
  crm_lookup→draft: 3 unchanged
  draft→qa_check: 1 unchanged, 2 regressed

report.html is a self-contained Jinja2-rendered file (CSS inlined, no JS framework, no CDN) with an aggregate agent graph (edges colored by regression rate), per-trace side-by-side panels, and judge rationales.

---

Public API map

# Capture
from proveai import (
    monitor, trace, Trace, FileExporter, add_exporter,
    Check, RuleCheck, LLMJudge,
    assert_json, regex_match, tool_called, not_contains,
    output_length, schema_match, contract_conformance,
    record_outcome,
)

# Replay engine + diff
from proveai import (
    load_trace, iter_traces,
    replay_span, replay_trace,
    text_diff, llm_judge, attribute_edges,
    CheckResult, ReplayedSpan, ReplayedTrace, EdgeVerdict,
)

# Store + artifacts
from proveai import (
    Store,                                         # project-local layout (./proveai/)
    Detection, AnalyzeResponse,                    # analysis result + persistence
    Baseline,                                      # named, versioned trace_id snapshot
    Experiment,                                    # named override set (YAML)
    Run,                                           # replay+attribution execution dir
    Snapshot, AgentDefinition, ExpectedIO,         # pinned known-good agent I/O
    EngineCache, EngineCacheMiss,                  # offline engine response cache
)

# Mock surface + pytest plugin
from proveai import mock                           # mock.from_trace, mock.responses, mock.failure
from proveai.pytest import snapshot                # @snapshot.test, snap.assert_no_regression

# Provider layer
from proveai import (
    Provider, OpenAICompatibleProvider, AnthropicProvider,
    register_provider, dispatch,
)
from proveai.cache import (
    ResponseCache, CachedProvider, cached_dispatch, default_cache, set_default_cache,
)
from proveai.providers.embedding import (
    LocalEmbeddingAdapter, OpenAICompatibleEmbeddingAdapter, cosine_similarity,
)

# Layman narrative renderers
from proveai.narrative.mast import render as render_detection, narrate_cohort
from proveai.narrative.verdicts import render as render_verdict

# HTML report
from proveai.report import render_html, write_html_report

# Engine integration (original)
from proveai import ProveAI, on_detection

---

Capture surface

@monitor — instrument an agent function

@monitor(
    "agent_name",                # required
    operation="invoke_agent",    # invoke_agent | chat | execute_tool
    description="agent role",
    model="gpt-4o-mini",         # used by replay to dispatch the right Provider
    parent_agent="upstream",     # topology edge: parent → this agent
    version="v3",                # gen_ai.version — filter replays by build
    kind="CHAT_MODEL",           # MLflow taxonomy — drives per-type Tier-1 diff
)
def my_agent(query: str) -> str:
    ...

Valid kind values: CHAIN, AGENT, TOOL, LLM, CHAT_MODEL, RETRIEVER, RERANKER, EMBEDDING, PARSER, MEMORY, UNKNOWN (default).

Behavior:

• Outside a trace() context → runs normally, zero overhead.
• Inside a trace() context → captures input args, output, timing, model, version, kind, parent edge, and error status as a span.
• Sync and async functions both supported.
• Exceptions are re-raised after recording the error.

trace() / Trace — group spans, run checks, fire exporters

from proveai import FileExporter, trace

with trace(
    "my-service",
    exporters=[FileExporter()],            # default: ./proveai/traces
    checks=[...],
) as t:
    ...

On exit:

1. Registered checks run (in order) → t.check_results plus a composite t.outcome (any bad → bad, all good/skip → good).
2. Exporters fire (per-trace then global).
3. Check exceptions are caught — a flaky judge never crashes the trace.

Checks — auto-label each trace

from proveai import (
    Check, RuleCheck,
    assert_json, regex_match, tool_called, not_contains,
    output_length, schema_match, contract_conformance,
    LLMJudge,
)

checks = [
    assert_json("classify"),
    schema_match("classify", {"type": "object", "required": ["category"]}),
    contract_conformance("draft", PydanticDraftModel),
    regex_match("draft", r"\[\d+\]"),
    output_length("draft", min_chars=50, max_chars=2000),
    tool_called("crm_lookup"),
    not_contains("response", "I don't know"),
    LLMJudge("qa_check", "must catch obvious errors", model="gpt-4o-mini"),
]

Every check accepts an optional trigger=callable(trace) -> bool predicate to gate cost: LLMJudge(..., trigger=lambda t: t.passed("json_valid")) skips the LLM call when the cheap rule check already failed.

LLMJudge returns good/bad/skip; malformed LLM responses degrade to skip rather than crashing the trace.

Custom checks subclass Check and implement __call__(trace) -> CheckResult.

FileExporter — three-file on-disk layout

exporter = FileExporter()  # default: ./proveai/traces (honors $PROVEAI_TRACES_DIR)

with trace("svc", exporters=[exporter]) as t:
    ...

Writes (under the configured root, default ./proveai/traces):

• <root>/<trace_id>.json — full OTLP JSON.
• <root>/manifest.json — atomic-write index (trace_id → {version, agents, timestamp}).
• <root>/outcomes.jsonl — append-only log: {trace_id, check_name, label, score, source} with source ∈ {check, composite, manual}.

add_exporter registers a global exporter that fires for every Trace.

record_outcome — manual override

from proveai import record_outcome

record_outcome(trace_id, label="good", score=0.95, reason="thumbs-up from user", root="./proveai/traces")

Appends an outcomes.jsonl row with source="manual". Use when you have a downstream business signal (UI thumbs, conversion, ticket resolution) that the automated checks couldn't see.

---

Replay engine

load_trace / iter_traces — read from the trace store

from proveai import load_trace, iter_traces

trace = load_trace("./traces/<trace_id>.json")

# Filter by composite outcome from outcomes.jsonl
for trace in iter_traces("./traces", filter={"outcome": "good"}):
    ...

# Other dict keys: version, agent. Multiple keys are AND-ed.
iter_traces("./traces", filter={"outcome": "good", "version": "v3"})

# Or pass an arbitrary callable
iter_traces("./traces", filter=lambda t: t.span("retriever") is not None)

Round-trips the OTLP JSON: trace_id, service_name, span attributes, status, timestamps all restored.

replay_span — re-execute one captured span

from proveai import replay_span

replayed = replay_span(span, override={
    "model": "gpt-4o",
    "prompt": "BE CONCISE.",         # prepends/replaces leading system message
    # "messages": [...],              # full replacement
    "temperature": 0.0,               # plus any provider kwargs
})
# replayed.new_output, replayed.model, replayed.latency_ms

Routes via dispatch(model) (the Provider registry) by default; pass provider= to inject a specific one. A CachedProvider wraps the resolved provider so identical replay calls hit the on-disk cache.

replay_trace — re-execute a captured topology

from proveai import replay_trace

replayed = replay_trace(trace, overrides_by_agent={
    "qa_check": {"prompt": open("prompts/qa_check_v2.md").read()},
    "draft":    {"model": "claude-haiku-4-5"},
})

Walks the captured topology (graph.node.parent_id) parent-first via BFS. Each parent's new output is substituted into a child's input wherever the child captured the parent's old output verbatim — covers the common B(input=A()) headless-chain shape. Explicit messages/prompt overrides for a child suppress the substitution.

Returns a ReplayedTrace with replayed_spans (per-span new outputs) and an empty edge_verdicts list — the diff orchestrator fills that in next.

---

Diff stack — three tiers

The orchestrator picks the cheapest verdict tier whose answer it trusts.

Tier	Module	Cost	Decisive on
1 — structural / per-type	text_diff + _per_type_tier1	free	JSON, numeric swaps, whitespace-equal text, RETRIEVER doc sets, TOOL calls, EMBEDDING vectors, PARSER deep-equal
2 — embedding cosine	Embedder + cosine_similarity	~50ms (local)	obvious-match (sim > 0.99 + numbers match) and obvious-divergence (sim < 0.40) prose
3 — LLM judge	llm_judge	cents (cached)	the uncertain band — produces a rationale for the report

text_diff — Tier 1

from proveai import text_diff

text_diff(old, new) -> {"equal": bool, "kind": "text|json|numeric", "unified": str}

• JSON deep-equal (key order ignored, sorted-keys diff).
• Numeric-aware factual-swap defense — "$1,200" vs "$12,000" returns equal=False, kind="numeric".
• Whitespace-normalized text equality otherwise.
• Always emits a unified diff string suitable for terminal display.

Embedding (Tier 2)

from proveai.providers.embedding import (
    LocalEmbeddingAdapter, OpenAICompatibleEmbeddingAdapter, cosine_similarity,
)

embedder = LocalEmbeddingAdapter()                          # sentence-transformers, offline
# or
embedder = OpenAICompatibleEmbeddingAdapter(api_key="...")  # text-embedding-3-small

a = embedder.embed("the cat sat")
b = embedder.embed("the feline rested")
cosine_similarity(a, b)  # → 0.0..1.0

Each EmbeddingResult carries model and version stamps; cosine_similarity raises ValueError on cross-model comparison so the orchestrator can fall back to the judge instead of producing meaningless scores.

llm_judge — Tier 3

from proveai import llm_judge

llm_judge(input_messages, old_output, new_output, model="claude-haiku-4-5") ->
    {"verdict": "equivalent|better|worse|skip", "confidence": 0.0..1.0, "rationale": str}

Single prompt, structured-JSON response, code-fence tolerant. Malformed responses or provider exceptions degrade to skip with confidence 0.0 rather than crashing. Routes via dispatch(model) and is wrapped in CachedProvider automatically — repeat calls hit the response cache.

attribute_edges — the orchestrator

from proveai import attribute_edges

verdicts = attribute_edges(
    trace,                # captured Trace (topology + old outputs)
    replayed,             # ReplayedTrace (new outputs)
    new_trace=None,       # optional: actually-executed topology for path/loop detection
    embedder=embedder,    # optional Tier 2; None → escalate text mismatches direct to judge
    judge_provider=None,  # injected provider for tests; default = dispatch(model)
    judge_model="claude-haiku-4-5",
    cache=None,           # ResponseCache; None → process default
    calibration=None,     # CalibrationReport; None → load proveai.calibration.json if present
)
# Also assigned to replayed.edge_verdicts as a side effect.

Per-edge logic:

if span.kind in {RETRIEVER, TOOL, EMBEDDING, PARSER}:
    use the per-type Tier-1 strategy → verdict (no escalation)

else:                                        # CHAT_MODEL / LLM / AGENT / ... / unset
    text_diff(old, new) →
        json equal      → unchanged
        json/numeric ≠  → regressed                    (decisive)
        text equal      → unchanged                    (whitespace-only)
        text ≠          → escalate to Tier 2

    Tier 2 (if embedder provided) →
        sim > 0.99 AND numeric tokens match → unchanged
        sim < 0.40                          → regressed
        else                                 → escalate to Tier 3

    Tier 3 (judge) →
        equivalent → drifted
        worse      → regressed
        better     → drifted (with note)
        skip       → drifted (confidence 0.0)

Cross-cutting handlers:

• Parallel siblings — children of the same parent whose [start, end] intervals all overlap are diffed as an unordered output set; reordering between parallel runs returns unchanged instead of false-firing as regressed.
• Path divergence (new_trace provided) — parents whose child set differs between old and new produce a single path_changed verdict, not a cascade of regressed ones under the diverged subtree.
• Loop iteration count (new_trace provided) — (parent, child) pairs whose multiplicity changed produce one iteration_count_changed verdict instead of N per-iteration regressions.

EdgeVerdict.status values: unchanged, drifted, regressed, path_changed, iteration_count_changed.

---

Provider layer

Provider protocol + dispatch registry

from proveai import (
    Provider, OpenAICompatibleProvider, AnthropicProvider,
    register_provider, dispatch,
)

# Built-in providers register themselves on import:
#   gpt-*, o1-*, o3-*, o4-*  → OpenAICompatibleProvider
#   claude-*                  → AnthropicProvider

# OpenAICompatibleProvider works against any /chat/completions-shaped endpoint
together = OpenAICompatibleProvider(
    api_key="...", base_url="https://api.together.xyz/v1",
)
register_provider("llama-*", together)

# Local Ollama / vLLM via base_url — zero data egress
local = OpenAICompatibleProvider(api_key="ignored", base_url="http://localhost:11434/v1")
register_provider("ollama:*", local)

provider = dispatch("gpt-4o-mini")     # returns the matching provider
provider.call(messages, "gpt-4o-mini")

Custom providers (Bedrock, Vertex, Cohere, anything genuinely non-OpenAI-shaped) implement the Provider protocol (call + supports) and register_provider(pattern, instance).

Env-driven custom-endpoint registration

When you don't want to write Python at all — drop these into .env:

PROVEAI_PROVIDER_PATTERN=accounts/fireworks/models/*,deepseek-*
PROVEAI_PROVIDER_BASE_URL=https://api.fireworks.ai/inference/v1
FIREWORKS_API_KEY=fw_…       # or PROVEAI_PROVIDER_API_KEY=…

# Optional — for Tier-2 embeddings via Ollama / a self-hosted endpoint:
PROVEAI_EMBEDDER_MODEL=nomic-embed-text:latest
PROVEAI_EMBEDDER_BASE_URL=http://localhost:11434/v1
PROVEAI_EMBEDDER_API_KEY=ollama

# Optional — pin the Tier-3 judge to your custom model:
PROVEAI_JUDGE_MODEL=accounts/fireworks/models/deepseek-v3p1

The SDK reads these on import and registers an OpenAICompatibleProvider for the patterns before the built-in gpt-* / claude-* defaults — so overlapping patterns (e.g. PROVEAI_PROVIDER_PATTERN=gpt-* pointing at a local Ollama URL) override the defaults. PROVEAI_JUDGE_MODEL is read at call time, so a .env loaded after import still wins.

ResponseCache + CachedProvider

File-based response cache keyed by sha256(provider_id + model + messages_json + sorted_kwargs). Default location ~/.proveai-cache/ (override via PROVEAI_CACHE_DIR env var or root=). Disable globally with PROVEAI_NO_CACHE=1.

from proveai.cache import (
    ResponseCache, CachedProvider, cached_dispatch, default_cache, set_default_cache,
)

cache = ResponseCache(root="./.cache")

# Wrap any provider
wrapped = CachedProvider(my_provider, cache=cache)

# Or resolve via dispatch + wrap
provider = cached_dispatch("gpt-4o-mini")              # uses default cache

# Tests / CI-deterministic mode
set_default_cache(ResponseCache(enabled=False))

replay_span, llm_judge, and the CLI run command all consult the cache automatically — repeat calls with identical inputs return the stored response without hitting the LLM.

---

CLI — proveai

proveai --help

Subcommands (full reference in CLI.md):

Subcommand	What it does
snapshot	Headline CI gate. init / show / list / verify / update / diff.
status	Recent traces with outcome + detection + age columns
list	Legacy browse view of the trace store
analyze	Deprecated — exits 2; detection is written automatically at capture time
baseline	Manage named, versioned trace snapshots: create/list/show
experiment	Manage named override sets: new/list/show
run	Replay a baseline against an experiment, attribute edges, write report. Span-level counterfactual via --hold-at.
calibrate	Compute per-agent Tier-2 similarity thresholds
ci	Scaffolds the GitHub Actions gate via ci init, pinned to the installed SDK version, strict judge mode on
validate-fix	Deprecated — exits 2 pointing at proveai run --hold-at

snapshot

The headline CI surface. Six actions on top of one artifact at proveai/snapshots/<name>.json.

# Pin the most recent trace as the 'latest' snapshot.
proveai snapshot init

# Or pin a specific trace under a custom name.
proveai snapshot init weekly-goods --trace-id 84a9dced82e2…

# Browse what's persisted.
proveai snapshot list
proveai snapshot show latest

# Re-run your pipeline against the snapshot, grade each agent. Exit 1 on regressed.
proveai snapshot verify
proveai snapshot verify --quick                          # legacy fixture-check, no re-run
proveai snapshot verify --override qa_check=prompts/qa_v2.md   # what-if (no code edit)
proveai snapshot verify --no-judge                       # hash-only, exit 0/2
proveai snapshot verify --strict-judge                   # exit 2 if any agent went unjudged

# Accept the new state (jest -u).
proveai snapshot update                                  # full refresh
proveai snapshot update --agent qa_check                 # partial

# See what changed (last verify vs. snapshot).
proveai snapshot diff
proveai snapshot diff --side-by-side --agent qa_check

As of v0.3.0, verify re-invokes your pipeline by default — it re-runs the command captured at snapshot init time (interpreter, argv, cwd), captures a fresh trace, and diffs each agent against the snapshot, catching prompt edits, model swaps, and agent-code changes. --quick skips the re-run for the cheap legacy fixture-check; --override AGENT=PATH does what-if prompt testing without editing your MAS. Because auto-replay re-runs your pipeline, side effects fire again — the SDK sets PROVEAI_VERIFY=1 in the subprocess so you can guard them (if os.environ.get("PROVEAI_VERIFY"): ...).

verify writes a full Run artifact (proveai/runs/<run_id>/) with verdicts + metrics + a self-contained report.html. The Tier-3 LLM judge handles non-byte-identical outputs and caches verdicts per (old_hash, new_hash, judge_model) under proveai/snapshots/.judge_cache.json, so the second verify against the same change costs zero LLM calls. PROVEAI_NO_JUDGE=1 short-circuits to hash-only. Snapshots pinned before v0.3.0 (no capture context) fall back to the legacy fixture-check — re-run proveai snapshot init to enable auto-replay.

When the judge can't run (missing key, unreachable provider, malformed reply), verify warns loudly, and --strict-judge (or PROVEAI_STRICT_JUDGE=1, on by default in scaffolded CI workflows) turns that into a failing gate.

Every non-unchanged verdict row is annotated (direct edit) or (downstream) based on a hash comparison between the fresh trace's prompt + model and the snapshot's pin. direct edit is a root cause (you changed this agent yourself); downstream is collateral (its prompt is intact, but upstream input drifted). Same labels appear as a Change column in report.html. See CLI.md → "Change-kind annotation" for the details and limitations.

list

Legacy browse view of the trace store + composite outcomes. Use status (below) for the current store layout.

proveai list --traces ./traces
proveai list --outcome good --version v3
proveai list --no-color                    # for CI logs

Renders a rich table with Trace ID, Version, Agents, Outcome (color-coded), Timestamp.

status

Sibling of list — adds a Detection column (clean / suspect / flagged / —) and an Age column ("5m ago").

proveai status --filter detection=flagged --limit 10
proveai status --filter outcome=bad --filter since=1d

Filter keys: outcome, version, agent, detection (flagged/suspect/clean), since (e.g. 1h, 2d).

analyze (deprecated)

proveai analyze was retired in plan v3.4 / SDK 0.3.1 — it now prints a redirect and exits 2. Detection is written automatically at capture time by FileExporter: every trace that runs checks gets a thin Detection JSON (regression verdict clean / suspect / flagged plus the checks that fired) alongside the trace, so proveai status and --filter detection=… work with no follow-up command.

The status mapping is unchanged: composite outcome bad → flagged; good → clean; warn / skip / absent → suspect. The narrative renderers (proveai.narrative.mast.render and narrate_cohort) remain importable for users who write their own engine integrations.

baseline and experiment

proveai baseline create reviewer-failures --filter detection=flagged
proveai baseline list
proveai baseline show reviewer-failures

proveai experiment new fix-qa-check --override qa_check=prompts/qa_v2.md
proveai experiment list
proveai experiment show fix-qa-check

Both write to <store>/baselines/ and <store>/experiments/ respectively.

validate-fix (deprecated)

validate-fix was retired as a headline command. Counterfactual replay now lives on proveai run:

# Hold every other span; re-execute just the captured span with id <span_id>
proveai run --hold-at <span_id> --override qa_check=prompts/qa_v2.md

# Or replay every iteration of an agent counterfactually
proveai run --counterfactual qa_check --override qa_check=prompts/qa_v2.md

Running proveai validate-fix … now exits 2 with a redirect message.

run

Replay filtered traces, attribute edges, write the report, exit non-zero on regression (CI gate).

proveai run \
  --traces ./traces \
  --filter outcome=good \
  --filter version=v3 \
  --override qa_check=prompts/qa_check_v2.md \
  --override draft=prompts/draft_v2.md \
  --report report.html \
  --embedder local \
  --judge-model claude-haiku-4-5 \
  --no-cache \
  --no-color

Options:

• --filter key=value — repeatable, AND-ed. Supports outcome, version, agent.
• --override AGENT=PATH — repeatable; PATH contents become AGENT's replacement system prompt during replay.
• --counterfactual AGENT[,AGENT…] — replay only the listed agent(s) live; hold the rest to recorded outputs.
• --hold-at SPAN_ID[,SPAN_ID…] — span-level counterfactual; useful for picking one iteration out of a loop.
• --propagate — blast-radius mode for --counterfactual / --hold-at: held agents still re-execute with parent-substituted inputs.
• --report — output HTML path (default report.html).
• --embedder none|local|openai — Tier 2 backend (default none skips Tier 2).
• --judge-model — override the Tier 3 model.
• --no-cache — bypass the on-disk response cache.
• --no-color — strip ANSI for CI.

Exit codes: 0 = no regression, 1 = at least one edge regressed, 2 = argparse / IO error.

calibrate

Compute per-agent Tier-2 similarity thresholds from known-good traces.

proveai calibrate \
  --traces ./traces \
  --out proveai.calibration.json \
  --model text-embedding-3-small

Loads each agent's known-good outputs, computes pairwise cosine similarities within each agent's distribution, and writes recommended (unchanged_threshold, regressed_threshold) per agent. attribute_edges reads proveai.calibration.json automatically when present and falls back to module defaults (0.99 / 0.40) otherwise.

---

HTML report

from proveai.report import write_html_report

# Pairs of (original Trace, ReplayedTrace) — typically built by the run loop above
write_html_report(pairs, "report.html", judge_model="claude-haiku-4-5")

Single self-contained HTML file (CSS inlined, no JS framework, no CDN — opens in any browser without a server):

• Aggregate view at the top — agent-graph SVG with each parent → child edge colored by regression rate across the dataset.
• Per-trace section — side-by-side panels for each agent (input messages, old output, new output, unified diff, judge rationale per edge).
• Footer — totals per status (unchanged/drifted/regressed/path_changed/iteration_count_changed), model + judge model used, timestamp.

---

Engine integration (original surface)

Stream traces to the ProveAI detection engine and react to detected failures. This is the original SDK use case and stays fully supported.

from proveai import ProveAI, monitor, trace, on_detection

client = ProveAI(
    endpoint="http://localhost:19000",
    run_id="run-1",
    customer_id="acme",
    mas_name="research-pipeline",
)

@monitor("researcher", model="gpt-4o")
def researcher(q): return llm.invoke(q)

@monitor("synthesizer", parent_agent="researcher")
def synthesizer(r): return llm.invoke(f"Synthesize: {r}")

@on_detection(severity="critical", min_confidence=0.8)
def alert(response):
    for d in response.diagnoses:
        page_oncall(d.agent_name, d.root_cause)

with trace("research-pipeline") as t:
    r = researcher("AI safety")
    s = synthesizer(r)
    response = client.analyze(t)        # SSE-streamed 7-layer detection

if t.result.detected:
    fix = t.result.propagation[0].remediation

ProveAI methods: health(), analyze(trace) (SSE), analyze_async(trace), analyze_batch([t1, t2]). Filters on @on_detection: severity (critical/high/medium/low) and min_confidence. Handler exceptions are caught and logged.

Engine endpoints used: POST /ingest/traces/stream, POST /ingest/traces, GET /health.

---

Architecture

Capture (in-process, dev or prod)
  ────────────────────────────────────────────────
  @monitor / Trace ──► Check runner ──► FileExporter ──► ./traces/
  ─                    (rules + judge)                  ├── manifest.json
                                                        ├── outcomes.jsonl
                                                        └── <trace_id>.json …

Snapshot verify (CLI, the headline CI gate)
  ────────────────────────────────────────────────
  proveai snapshot init     ── pin agent I/O hashes + capture context
  proveai snapshot verify   ── re-invoke your pipeline (subprocess, v0.3.0+)
        │                      ──► fresh trace ──► per-agent hash + judge diff
        └─► exit 1 if any agent regressed   — the CI gate

Replay (CLI, on-demand, no backend service)
  ────────────────────────────────────────────────
  proveai run --filter outcome=good --override AGENT=PATH
        │
        ├─ iter_traces(filter)  ── from manifest + outcomes.jsonl
        │
        ├─ for each trace:
        │     replay_trace        ── BFS topology, propagate parent outputs
        │       │
        │       └─► Provider dispatch ──► live LLM (or CachedProvider hit)
        │
        │     attribute_edges    ── walk Tier 1 → Tier 2 → Tier 3
        │       │
        │       ├─ text_diff                    — Tier 1, structural
        │       ├─ embedding cosine              — Tier 2, optional, --embedder local|openai
        │       └─ llm_judge                     — Tier 3, cheap default, cached
        │
        │     Returns one EdgeVerdict per edge:
        │       unchanged | drifted | regressed |
        │       path_changed | iteration_count_changed
        │
        └─► write_html_report(pairs)             — self-contained report.html
            terminal summary table               — fits in 80 cols
            exit code = 1 if any edge regressed  — CI gate

Engine integration (parallel use case)
  ────────────────────────────────────────────────
  @monitor / Trace ──► client.analyze(trace) ──► engine SSE ──► AnalyzeResponse
                                                                       │
                                                            on_detection callbacks