okovia — OkOvia Python SDK

Measure the cost and margin of every AI operation from your backends, workers, queues, and AI-infrastructure code. OkOvia prices each model call and GPU-second and attributes it to a product operation.

Server-side only — use a secret (vik_sec_…) or ingest-only (vik_ing_…) key. Never ship this in a browser or mobile app (use the web tag or the Swift SDK there).

Install

pip install okovia

Usage

from okovia import OkoviaClient

client = OkoviaClient(
    secret_key="vik_sec_xxx",
    project_id="project_123",
    endpoint="https://api.okovia.com",
)

client.record_usage(
    operation_id="op_checkout_7K9x",
    provider="openai",
    model_name="gpt-4o",
    input_tokens=1200,
    output_tokens=300,
    cache_write_tokens=2048,   # cost-category fields the pricing engine uses
    reasoning_tokens=128,
    stream_status="complete",
)

Time a step automatically

with client.step(operation_id="op_1", step_name="rag", provider="openai") as step:
    result = call_your_llm()
    step.add_metric("input_tokens", result.usage.input_tokens)
    step.add_metric("output_tokens", result.usage.output_tokens)
# an unhandled exception inside the block marks the event status="error"

Correlate browser context with backend usage (FastAPI/Starlette)

from okovia import OkoviaCorrelationMiddleware, current_operation_id

app.add_middleware(OkoviaCorrelationMiddleware)
# ... then in a request handler:
client.record_usage(operation_id=current_operation_id() or "op_fallback", ...)

On-device hashing for recommendations (privacy-safe)

from okovia import hash_prompt_prefix

# Salted digest of the repeated prompt prefix — content never leaves the
# process; only the hash is sent. Feeds the "prompt caching off" rule.
client.record_usage(
    operation_id="op_1",
    prompt_prefix_hash=hash_prompt_prefix(prompt, salt="your-project-salt"),
    ...
)

Evaluate quality — only the score travels (0.4.0)

Quality evaluation runs inside your process; the model output never leaves your app. Only the resulting score joins the pipeline, on the same operation_id as the cost:

import okovia
from okovia.evaluators import json_valid, refusal_detected, truncated

okovia.configure(secret_key="vik_sec_...", project_id="...",
                 endpoint="https://api.okovia.com")

okovia.evaluate("op_1", model_output,
                evaluators=[json_valid(), refusal_detected(), truncated()])

Built-in heuristic evaluators (stdlib, zero cost): json_valid, schema_match(schema), refusal_detected, language_match(lang), pii_leak_in_output, truncated. All score in [0, 1], higher is better.

LLM-as-judge with your own provider key — the judge call is reported as a regular usage event, so the cost of measuring quality is measured:

from okovia.evaluators import LLMJudge, JudgeVerdict

def my_judge(output: str) -> JudgeVerdict:
    response = my_provider_call(output)          # your key, your process
    return JudgeVerdict(score=parse_score(response),
                        provider="openai", model="gpt-5-mini",
                        input_tokens=response.usage.input_tokens,
                        output_tokens=response.usage.output_tokens)

okovia.evaluate("op_1", model_output,
                evaluators=[LLMJudge("helpfulness_judge", my_judge)])

client.evaluate(...) works the same on an existing OkoviaClient. Inside a request handled by OkoviaCorrelationMiddleware, pass operation_id=None and the current operation is used.

Run under the remote policy — control plane (0.5.0)

Publish a policy in the console (fallback chains, budget caps, quality floors, routing) and let the SDK enforce it inside your process — OkOvia never proxies your calls:

import okovia

okovia.configure(secret_key="vik_sec_...", project_id="...",
                 endpoint="https://api.okovia.com")

result = okovia.run_with_policy(
    "chat",
    models={
        "gpt-4o": lambda: call_openai("gpt-4o", prompt),
        "gpt-4o-mini": lambda: call_openai("gpt-4o-mini", prompt),
    },
)
result.value       # your callable's return
result.model_used  # after routing / downgrade / fallback

# Attach the decisions to your usage event:
with client.step(operation_id=op, step_name="answer") as step:
    result = client.run_with_policy("chat", models={...})
    step.apply_policy(result)   # model_used + policy_version/fallback_from/...

Semantics: exceeded budget caps alert (proceed, marked), downgrade (start from the cheapest chain model) or block (OkoviaPolicyBlockedError before any call); fallback chains retry the next model on error/timeout. Fail-open: no config, no policy, or an unreachable endpoint mean plain execution — OkOvia being down never blocks you. Caps are soft ceilings with bounded staleness (~5 min server state + 300s SDK TTL).

Quality floors (0.6.0). When the console declares policy.quality_floor.<feature>.min_score and the measured eval average falls below it, run_with_policy can escalate: list "quality_floor" in the feature's fallback triggers and the call starts from the chain's first (preferred) model instead of the routed/cheaper one — result.quality_action == "escalate". Without the trigger it only marks the result ("alert"). Budget actions always win over quality.

Command line (okovia)

Installing the package also installs the okovia CLI — for the places SDKs don't reach naturally: shell scripts, cron jobs, CI pipelines, GPU batch workers.

export OKOVIA_INGEST_KEY="vik_ing_..."
export OKOVIA_PROJECT_ID="..."
export OKOVIA_ENDPOINT="https://api.okovia.com"

okovia doctor --send-test      # validate credentials + connectivity end to end

okovia track --provider openai --model gpt-4o \
             --input-tokens 1200 --output-tokens 340 --feature nightly_batch

okovia track --gpu-seconds 142.5 --provider runpod --model a100 \
             --feature training_job                 # unit type is inferred

okovia costs                   # cost by feature, straight from the API

track infers the unit type from the flags you pass (tokens → tokens, --gpu-seconds → GPU, --images → image generation, none → api_call) and prints the ingestion result as JSON, so it composes with jq.

Privacy

Prompts and completions are never collected. Only usage metadata and salted hashes leave your process; sensitive fields are rejected client-side before sending.

Your own redaction list (0.6.0): fields listed under SDK → Privacy → redact_fields in the console are stripped from raw_usage_json and eval metadata before anything is sent — case-insensitive, at any depth, no release needed. The SDK reads them from the same cached /v1/config the control plane uses (one request per 5 minutes, fail-open: no published config means no extra redaction).

Compatibility

The historical viking_metering package and VikingMeteringClient name remain importable for existing code:

from viking_metering import VikingMeteringClient  # still works

License

MIT — free while OkOvia is in beta.