pradvion 0.5.4

Track AI API costs per client, project, and feature

Pradvion Python SDK

Track AI API costs per client, project, and feature. Know exactly what to bill each client.

Installation

pip install pradvion

Quick Start — monitor()

The fastest integration. One line wraps your AI client.

import pradvion
from openai import OpenAI

pradvion.init(api_key="nx_live_YOUR_KEY")
openai = pradvion.monitor(OpenAI())

# All calls tracked automatically — no other changes needed
response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}]
)

Auto-Instrument (Zero Setup)

The fastest way to add Pradvion to existing code:

# Install
pip install pradvion

# Preview changes (no files modified)
pradvion-instrument --dry-run

# Apply instrumentation
pradvion-instrument

# Instrument specific file or directory
pradvion-instrument src/
pradvion-instrument app.py

# Remove instrumentation
pradvion-instrument --undo

What it adds automatically:

# Before
from openai import OpenAI
client = OpenAI()

# After
import os
import pradvion
pradvion.init(api_key=os.environ.get("PRADVION_API_KEY", ""))
from openai import OpenAI
client = pradvion.monitor(OpenAI())

Context Manager

Tag calls with customer, feature, and environment:

with pradvion.context(
    feature="resume-summarizer",
    customer_id="samsung-001",
    environment="production"
):
    response = openai.chat.completions.create(...)

Real-Time Streaming Cost

See estimated cost as tokens arrive:

def on_cost_update(cost: float):
    print(f"\r~${cost:.6f}", end="", flush=True)

for chunk in client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    stream=True,
    on_token=on_cost_update,  # Called every 5 tokens
):
    print(chunk.choices[0].delta.content or "", end="")

Business Signals

Track what your AI creates, not just what it costs:

# After an AI call creates value, record it
pradvion.signal(
    customer_id="samsung-001",
    event="email_generated",
    quantity=1,
    value=0.05,
    metadata={"type": "outbound_sales"}
)

pradvion.signal("samsung-001", "meeting_booked", value=150.00)
pradvion.signal("samsung-001", "report_generated", quantity=1, value=50.00)

Conversation Tracking

Group multi-turn AI calls into a single conversation session:

conv_id = pradvion.new_conversation()  # "conv_a3f9b2c1d4e5"

with pradvion.context(
    customer_id="samsung-001",
    conversation_id=conv_id
):
    turn1 = openai.chat.completions.create(...)  # both calls
    turn2 = openai.chat.completions.create(...)  # same conv_id

Budget Management

from pradvion.budget import get_budget_tracker

tracker = get_budget_tracker()
tracker.set_limit("samsung-001", monthly_usd=500.0, action="warn")
tracker.record_spend("samsung-001", 0.05)
print(f"Remaining: ${tracker.remaining('samsung-001'):.2f}")

Cost Forecasting

from pradvion.forecast import forecast_monthly

result = forecast_monthly(
    actual_spend=150.00,
    days_elapsed=15,
    budget_limit=500.00,
)

print(f"Projected: ${result.projected_monthly:.2f}/mo")
print(f"Will exceed: {result.will_exceed_budget}")

Compare Models

from pradvion.compare import compare_cost

result = compare_cost(input_tokens=1000, output_tokens=500)
print(f"Cheapest: {result.cheapest.model} — ${result.cheapest.total_cost:.6f}")
print(f"Save ${result.savings_vs_most_expensive:.4f} vs most expensive")

Middleware Pattern (FastAPI)

@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

Async Support

async_client = pradvion.monitor(openai.AsyncOpenAI())
response = await async_client.chat.completions.create(...)

Streaming

stream = openai.chat.completions.create(
    model="gpt-4o",
    messages=[...],
    stream=True,
)
for chunk in stream:
    print(chunk.choices[0].delta.content, end="")
# Usage captured automatically from last chunk

Agent / RAG Usage

with pradvion.context(feature="research-agent", customer_id="samsung"):
    # All sub-calls tracked under same context
    search = openai.chat.completions.create(...)
    analyze = openai.chat.completions.create(...)
    report = openai.chat.completions.create(...)

Flush and Shutdown

Auto-flush is on by default (flushes on process exit). For scripts or graceful shutdown:

pradvion.flush()     # send all pending events now
pradvion.shutdown()  # stop background worker

Zero Data Loss

All events are written to a local SQLite queue (~/.pradvion/queue.db) before being sent to Pradvion. If your process crashes or the API is unreachable, events are retried automatically on next startup.

Support

[e-mail] | https://pradvion.com