modelgenius-sdk

Capture your production LLM calls as local snapshot files, and validate them against the ModelGenius snapshot format — entirely on your machine, with zero network calls and zero telemetry.

ModelGenius answers "is it safe to switch LLM models, and is it worth it?" by replaying your real production requests against candidate models and comparing quality, cost, and latency. The input to that analysis is a set of snapshots: your requests, the parameters they ran with, and (ideally) the responses your current model actually produced. This SDK is the capture side. It has one runtime dependency (jsonschema) and never imports openai, anthropic, or litellm — it duck-types your existing client objects.

Three ways to adopt it:

1. Wrap your client — one line at startup; every call is spooled to local JSONL as it happens, with real outputs, token usage, and latency attached.
2. Bring your own logs — if you already log LLM traffic, write it in the snapshot format and use modelgenius-sdk validate to check it before uploading.
3. Bring your OpenTelemetry traces — if your app is already instrumented with the OTel GenAI semantic conventions (OpenLLMetry, OpenLIT, vendor SDKs, …), convert an OTLP/JSON export with modelgenius-sdk convert otlp — no app changes needed. Requires message content capture to be enabled in your instrumentation (it is off by default in OTel; metadata-only spans convert to an actionable error). Legacy flattened OpenLLMetry attributes (gen_ai.prompt.N.*) are also understood. Two gaps are inherent to the conventions and flagged rather than guessed: response_format JSON schemas and (often) tool definitions are not carried in traces — converted records note enrichment_needed in metadata and the validator reminds you to supply them at upload.

Install

pip install modelgenius-sdk

Quickstart

OpenAI (sync or async client; streaming supported):

import modelgenius_sdk
from openai import OpenAI

client = modelgenius_sdk.wrap_openai(OpenAI(), workflow_name="support_ticket_triage")
# use `client` exactly as before — every chat.completions call is snapshotted locally

Anthropic (sync or async client; messages.create streaming supported):

import modelgenius_sdk
from anthropic import Anthropic

client = modelgenius_sdk.wrap_anthropic(Anthropic(), workflow_name="math_tutoring")

LiteLLM (success callback):

import litellm
import modelgenius_sdk

litellm.success_callback.append(modelgenius_sdk.make_litellm_callback(workflow_name="ap_automation"))

Validate what you captured (or logs you wrote yourself):

modelgenius-sdk validate ./modelgenius_snapshots

The wrapper returns the same client object and never changes call behavior: your arguments pass through untouched, return values and exceptions propagate unchanged, and any internal capture error is swallowed (logged once), never raised into your app.

What gets captured

One JSON object per call, appended to a local JSONL spool file: the request (messages, model, temperature, tools, response_format, and the rest of the request parameters), plus a captured_output block with the real response — output text, tool calls, token usage, latency, and, via the LiteLLM callback, cost. See docs/snapshot_format.md for the full format, and examples/ for validated sample files.

The more you capture, the more the analysis can do:

You provide	You unlock
captured_output	Baseline scored from real production outputs — no baseline replay, real latency/cost preserved
usage + cost_usd	Honest savings baseline from actual spend
workflow_name + call_name (or workflow_id)	Precise workflow grouping instead of content-based inference
tools / response_format	Deterministic output-contract checks on every candidate
model	Automatic baseline identification

Full table with grouping precedence: docs/snapshot_format.md.

Configuration

Everything can be set in code via CaptureConfig, or via environment variables:

from modelgenius_sdk import CaptureConfig, wrap_openai

config = CaptureConfig(
    spool_dir="/var/log/myapp/snapshots",
    workflow_name="support_ticket_triage",
    call_name="classify_ticket",
    sample_rate=0.25,          # capture 25% of calls
)
client = wrap_openai(OpenAI(), config)

Environment variable	Effect	Default
MODELGENIUS_CAPTURE	Set to 0, false, or no (case-insensitive) to disable capture entirely — the kill switch	enabled
MODELGENIUS_CAPTURE_DIR	Spool directory for snapshot files	./modelgenius_snapshots
MODELGENIUS_CAPTURE_SAMPLE_RATE	Fraction of calls to capture, 0.0–1.0	1.0
MODELGENIUS_CAPTURE_DEBUG	Set to 1 to make internal capture errors raise instead of being swallowed (for tests/debugging)	off

Explicit CaptureConfig values win over environment variables. The sampling decision is made once per call, before dispatch, so a streamed response is either fully captured or not at all.

Redaction and filtering

You stay in control of what lands on disk:

def redact(record: dict):
    # Called with the full record just before it is written.
    for message in record.get("messages", []):
        if isinstance(message.get("content"), str):
            message["content"] = scrub_pii(message["content"])
    if record.get("metadata", {}).get("tenant_tier") == "enterprise":
        return None  # returning None drops the record entirely
    return record

def only_prod_traffic(request_kwargs: dict) -> bool:
    # Called with the request kwargs before capture is attempted.
    return request_kwargs.get("user") != "internal-smoke-test"

config = CaptureConfig(
    workflow_name="support_ticket_triage",
    redact=redact,
    request_filter=only_prod_traffic,
    sample_rate=0.1,
)

Spool files and rotation

Records are appended to <spool_dir>/snapshots-<YYYYMMDD>-<pid>-<seq>.jsonl — one JSON object per line, flushed after every write, safe across processes (each PID writes its own file). When the current file exceeds max_file_bytes (default 64 MiB) the writer rotates to the next sequence number. The spool directory is created lazily on first write. Point modelgenius-sdk validate (or the ModelGenius upload step) at the directory when you are ready.

Trust posture

• Local-only. The library makes no network calls of its own — the only network traffic is your unmodified provider call. Uploading snapshots to ModelGenius is a separate, deliberate step you take elsewhere.
• No telemetry. Nothing is phoned home. Ever.
• Fail-open. Capture errors never raise into your application and never alter the provider call, its return value, or its exceptions.
• You control the data. Redaction hook, request filter, sampling, and a one-variable kill switch (MODELGENIUS_CAPTURE=0).
• Small and auditable. A handful of modules, one dependency, no provider SDK imports — read the whole thing in one sitting.

CLI

modelgenius-sdk validate PATH [--json] [--strict] [--quiet]
modelgenius-sdk convert otlp PATH [-o OUTPUT.jsonl] [--json] [--quiet]
modelgenius-sdk version

validate accepts a .jsonl file, a .json file (object or array), or a directory of both. Exit code 0 means every record is accepted by the same semantics as the ModelGenius upload path; warnings point out optional fields that would unlock capability. --strict turns warnings into failures (useful in CI — this repo validates its own examples/ on every push).

convert otlp converts an OTLP/JSON trace export (a .json payload or a collector file-exporter .jsonl) into snapshot JSONL and validates the result in one step. Exit code 0 means every GenAI span converted and validated; per-span problems (e.g. content capture disabled in your instrumentation) are reported individually and exit 1.

Limitations (v0)

• Failed provider calls are not captured (only successful responses).
• OpenAI: chat.completions only; the responses API surface is not wrapped yet.
• Anthropic: messages.create (including stream=True) is captured; the messages.stream() helper is not yet.
• LiteLLM: the callback is sync-only (litellm accepts sync callbacks for async paths).

Roadmap

• Direct upload client (modelgenius-sdk push) as an explicit, opt-in step
• Continuous monitoring mode (rolling capture windows + scheduled re-evaluation)
• TypeScript SDK
• OpenAI responses surface and Anthropic messages.stream() capture

License

Apache-2.0.