Pradvion Python SDK

Track AI API costs by client, feature, and team. Connect every dollar spent to the business outcome behind it — meetings booked, deals closed, reports generated.

pradvion.com · Docs · Dashboard

Installation

pip install pradvion

OpenAI and Anthropic are optional extras — install only what you use:

pip install "pradvion[openai]"       # OpenAI support
pip install "pradvion[anthropic]"    # Anthropic support
pip install "pradvion[all]"          # Both

Quick Start

import openai
import pradvion

# Initialize once at startup
pradvion.init(api_key="nx_live_...")

# Wrap your OpenAI client — drop-in replacement
client = pradvion.monitor(openai.OpenAI())

# Tag with business context, then call as normal
with pradvion.context(
    feature="resume-summarizer",
    customer_id="customer-001",   # hashed with SHA-256 before sending
    team="hr-team",
    environment="production",
):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Summarize this resume"}],
    )
# Cost, tokens, and latency tracked automatically — zero prompt storage

Both pradvion.monitor() and pradvion.wrap() work — they are aliases.

Supported Providers

import openai, anthropic, pradvion

# OpenAI
openai_client   = pradvion.monitor(openai.OpenAI())
async_client    = pradvion.monitor(openai.AsyncOpenAI())  # async supported

# Anthropic
anthropic_client = pradvion.monitor(anthropic.Anthropic())

# Streaming — tokens accumulate correctly across chunks
stream = openai_client.chat.completions.create(model="gpt-4o", messages=[...], stream=True)
for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="")
# Usage tracked automatically from the final chunk

Context Tagging

Tag all AI calls within a block. All parameters are optional.

with pradvion.context(
    feature="resume-summarizer",   # what the AI is doing
    team="hr-team",                # team that owns this feature
    department="engineering",      # department-level grouping
    customer_id="customer-001",     # auto-hashed with SHA-256
    environment="production",      # filters out test traffic in analytics
    conversation_id="run-abc-123", # groups multi-step agent calls
    project="my-project",          # optional project tag
):
    response = client.chat.completions.create(...)

FastAPI middleware

Set context once per request so every AI call in that request is tagged:

@app.middleware("http")
async def pradvion_middleware(request, call_next):
    user = get_current_user(request)
    pradvion.set_context(
        customer_id=user.company_id,
        environment="production",
    )
    response = await call_next(request)
    pradvion.clear_context()
    return response

Business Signals

Track the outcomes your AI produces — not just the cost.

# After an AI call creates a downstream result, record the signal
pradvion.signal(
    customer_id="customer-001",   # links back to AI costs for this customer
    event="meeting_booked",      # lowercase, snake_case
    quantity=1,
    value=150.00,                # dollar value of the outcome
    feature="outreach-agent",
    environment="production",
)

# Batch version
pradvion.signal_batch([
    {"customer_id": "my-company", "event": "email_sent",     "quantity": 50},
    {"customer_id": "my-company", "event": "meeting_booked", "quantity": 3, "value": 450.0},
    {"customer_id": "my-company", "event": "deal_closed",    "quantity": 1, "value": 12000.0},
])

Pradvion automatically computes cost per meeting booked, margin per customer, and ROI per feature in the Unit Economics dashboard.

Agent / Multi-step Tracking

Group all LLM calls within a single agent run:

run_id = pradvion.new_conversation()  # generates a unique ID

with pradvion.context(
    feature="research-agent",
    customer_id="my-company",
    conversation_id=run_id,
    environment="production",
):
    plan   = client.chat.completions.create(...)   # step 1
    search = client.chat.completions.create(...)   # step 2
    report = client.chat.completions.create(...)   # step 3

# All 3 calls appear in the dashboard linked by conversation_id
# Record the business outcome once the agent completes
if task_completed:
    pradvion.signal("my-company", "report_generated", quantity=1, value=50.0)

Manual Tracking

Track calls from providers not yet wrapped:

import time

start = time.monotonic()
try:
    response = my_llm_client.generate(prompt)
    pradvion.get_client().track(
        provider="openai",
        model="gpt-4o",
        input_tokens=response.usage.prompt_tokens,
        output_tokens=response.usage.completion_tokens,
        latency_ms=int((time.monotonic() - start) * 1000),
        status_code=200,
        customer_id="my-company",
        feature="chatbot",
    )
except Exception as e:
    pradvion.track_error(
        provider="openai",
        model="gpt-4o",
        error=str(e),
        status_code=500,
        latency_ms=int((time.monotonic() - start) * 1000),
        customer_id="my-company",
    )
    raise

# Batch version
pradvion.track_batch([
    {"provider": "openai", "model": "gpt-4o",        "input_tokens": 500, "output_tokens": 200, "latency_ms": 800},
    {"provider": "anthropic", "model": "claude-sonnet-4-6", "input_tokens": 300, "output_tokens": 150, "latency_ms": 600},
])

Integrations

LangChain

from langchain_openai import ChatOpenAI
from pradvion.integrations.langchain import PradvionCallback
import pradvion

pradvion.init(api_key="nx_live_...")

callback = PradvionCallback(
    feature="research-chain",
    customer_id="my-company",
    environment="production",
)

# Works with chains, agents, and LCEL runnables
llm = ChatOpenAI(model="gpt-4o", callbacks=[callback])
response = llm.invoke("Summarize this document")

LangGraph

LangGraph uses LangChain under the hood — the same callback works:

from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI
from pradvion.integrations.langchain import PradvionCallback
import pradvion

pradvion.init(api_key="nx_live_...")
run_id = pradvion.new_conversation()

callback = PradvionCallback(
    feature="research-agent",
    customer_id="my-company",
    conversation_id=run_id,
)

llm = ChatOpenAI(model="gpt-4o", callbacks=[callback])

graph = StateGraph(...)
# ... define nodes and edges ...
graph.invoke(inputs)

LlamaIndex

from llama_index.core import Settings
from llama_index.core.callbacks import CallbackManager
from pradvion.integrations.llamaindex import PradvionLlamaCallback
import pradvion

pradvion.init(api_key="nx_live_...")

callback = PradvionLlamaCallback(
    feature="rag-pipeline",
    customer_id="my-company",
    environment="production",
)

# Set globally — all LlamaIndex pipelines are tracked
Settings.callback_manager = CallbackManager([callback])

index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("What are the main findings?")

OpenTelemetry

from pradvion.integrations.otel import PradvionSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

exporter = PradvionSpanExporter(api_key="nx_live_...")
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(exporter))

# Pradvion reads GenAI semantic conventions from OTEL spans
# Compatible with: OpenLLMetry, traceloop, Langfuse OTEL, etc.

init() Options

pradvion.init(
    api_key="nx_live_...",    # required — from Dashboard → Projects → API Keys
    base_url="https://...",   # optional — default: Pradvion cloud
    timeout=5,                # HTTP timeout in seconds (default: 5)
    async_tracking=True,      # background queue, non-blocking (default: True)
    auto_flush=True,          # flush on process exit (default: True)
)

Flush & Shutdown

Auto-flush is on by default (flushes on process exit). In short-lived processes — scripts, Lambda functions, or tests — call flush explicitly:

pradvion.flush(timeout=10.0)  # wait up to 10s for all events to send
pradvion.shutdown()           # flush + stop background worker

Privacy

Pradvion is privacy-first by design:

• Tracks token counts, model names, and latency — nothing else
• Customer IDs are SHA-256 hashed before leaving your server
• Prompts and responses are never captured or transmitted
• All source code is open source and auditable

Requirements

• Python >= 3.9
• No required dependencies — OpenAI and Anthropic are optional extras

Support

• Email: hello@pradvion.com
• Dashboard: pradvion.com
• Docs: pradvion.com/docs
• Issues: GitHub

License

MIT — see LICENSE