pradvion 1.0.1

Track AI API costs per client, project, and feature

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: [e-mail]
• Dashboard: pradvion.com
• Docs: pradvion.com/docs
• Issues: GitHub

---

License

MIT — see LICENSE