Examples

Capture a Decision

The @capture decorator records a lightweight decision dict and hands it to an exporter. It does not persist a native DecisionSnapshot itself.

from briefcase import capture

@capture(decision_type="summarize")
def summarize(text: str) -> str:
    # Replace with your real model call, e.g. client.responses.create(...).
    return text[:280]

result = summarize("Long document text...")

Configure an Exporter

briefcase.observe() wires up an exporter in one line and returns it, so @capture decisions are actually emitted. Pass "console", "memory", a *.jsonl path, or a BaseExporter instance.

import briefcase

mem = briefcase.observe("memory")  # or "console", or "decisions.jsonl"


@briefcase.capture(decision_type="summarize", async_capture=False)
def summarize(text: str) -> str:
    return text[:280]


summarize("Long document text...")
print(mem.records[0])

The stock exporters live in briefcase.exporters: ConsoleExporter (JSON lines to stderr), JSONLFileExporter (append to a file), and MemoryExporter (collect in .records). For full control, subclass BaseExporter and register it with setup() or pass it to observe().

from briefcase import setup, capture
from briefcase.exporters import BaseExporter


class PrintExporter(BaseExporter):
    async def export(self, decision) -> bool:
        print(decision)
        return True

    async def flush(self) -> None:
        ...

    async def close(self) -> None:
        ...


setup(exporter=PrintExporter())


@capture()
def classify(text: str) -> str:
    return "billing"

See Exporters for the full reference.

Persist and Replay a Snapshot

The native runtime layer is separate from @capture. Call init() to start the native runtime, build a DecisionSnapshot, save it to a SqliteBackend, then replay it with the ReplayEngine.

from briefcase import DecisionSnapshot, Input, Output, init
from briefcase.storage import SqliteBackend
from briefcase.replay import ReplayEngine

init()  # start the native runtime before persisting

# Record a classify_ticket decision from the support-triage agent.
decision = (
    DecisionSnapshot("classify_ticket")
    .with_module("support_service")
)
decision.add_input(Input("ticket_text", "My invoice is wrong", "string"))
decision.add_output(Output("category", "billing", "string").with_confidence(0.93))

# Persist it. SqliteBackend.in_memory() is handy for examples and tests.
storage = SqliteBackend.in_memory()
decision_id = storage.save_decision(decision)

# Replay the recorded decision against the stored snapshot.
# Modes: "strict" (exact match) or "tolerant" (the default).
engine = ReplayEngine(storage)
result = engine.replay(decision_id, "strict")
print(result.status, result.outputs_match)

Measure Drift Across Outputs

DriftCalculator.calculate_drift() scores the consistency of a set of outputs and reports the consensus value and any outliers.

from briefcase.drift import DriftCalculator

calculator = DriftCalculator()
outputs = ["billing", "billing", "account", "billing", "billing"]

metrics = calculator.calculate_drift(outputs)
print(f"Consistency: {metrics.consistency_score:.3f}")
print(f"Agreement:   {metrics.agreement_rate:.3f}")
print(f"Consensus:   {metrics.consensus_output}")
print(f"Status:      {metrics.get_status(calculator)}")

Estimate Cost and Check a Budget

from briefcase.cost import CostCalculator

calculator = CostCalculator()

estimate = calculator.estimate_cost("gpt-4", 1000, 500)
print(f"Total: ${estimate.total_cost:.4f}")

status = calculator.check_budget(85.0, 100.0)
print(f"{status.percent_used:.1f}% used - {status.status}")

Redact PII

from briefcase.sanitize import Sanitizer

sanitizer = Sanitizer()
result = sanitizer.sanitize("Contact support at support@example.com")

print(result.sanitized)
print(f"{result.redaction_count} redaction(s)")
for redaction in result.redactions:
    print(redaction.pii_type, redaction.start_position, redaction.end_position)

Correlate a Multi-Agent Workflow

briefcase_workflow is a context manager that links every agent that runs inside it under one workflow ID.

from unittest.mock import Mock

from briefcase.correlation import briefcase_workflow

client = Mock()  # replace with a real Briefcase client

with briefcase_workflow("content_pipeline", client) as workflow:
    print(f"Workflow: {workflow.workflow_id}")

    workflow.register_agent("retriever", "retrieval")
    workflow.register_agent("summarizer", "generation")
    workflow.register_agent("reviewer", "moderation")

Track and Compare Model Versions with oci-bai

Push images through the oci-bai gateway and use the CLI to inspect commits, compare versions, and search the catalog:

docker tag my-base:latest localhost:8080/rl-gym-env:cuda-base
docker push localhost:8080/rl-gym-env:cuda-base

docker tag my-candidate:latest localhost:8080/rl-gym-env:cartpole
docker push localhost:8080/rl-gym-env:cartpole

# Inspect and compare
oci-bai --repo rl-gym-env log cartpole
oci-bai --repo rl-gym-env diff cuda-base cartpole --depth package
oci-bai --repo rl-gym-env diff cuda-base cartpole --depth bench

# Search the catalog
oci-bai search "format==safetensors cuda>=12.4"

See the Quick Start for the end-to-end walkthrough and CLI Reference for every command and flag. oci-bai is in private beta — contact support@briefcaseai.org for access.

More Examples

See the examples directory for complete, runnable scripts covering basic usage, prompt validation, lakeFS versioning, and multi-agent correlation.

Next steps

Quickstart Install Briefcase and run your first capture in a few minutes.

Exporters Choose where the decisions in these examples are sent.