Lutflow SDK

The Lutflow SDK gives developers a first, honest handle on what LLM inference costs — and how to keep it in check. Track and enforce your provider spend in-process (OpenAI, Anthropic, Google Gemini), with budget strategies that fit your workflow. It also includes lutflow measure, an offline utility for cost and energy analysis from GPU telemetry exports.

Offline-first — no account, no backend needed to start.

Installation

pip install lutflow            # Core + CLI (measure + in-process enforcement)
pip install lutflow[openai]    # + OpenAI wrapper
pip install lutflow[anthropic] # + Anthropic wrapper
pip install lutflow[gemini]    # + Google Gemini wrapper
pip install lutflow[all]       # All providers

Compatible with Python 3.9+.

What this SDK is today

Capability	Status	Notes
In-process budget enforcement	Ships, real	Tracks token/GPU-time spend in your process and, on breach, raises, warns, runs a callback, or sends `SIGKILL` to the current process.
Provider wrappers	Ships, real	OpenAI, Anthropic, Google Gemini — wrap the client, spend is metered per call.
Pricing tables	Ships, real	Token and GPU-hour pricing with per-model overrides.
`lutflow measure`	Ships, real	Offline utility: cost & energy metrics from a telemetry export. Read-only, no network.

Budget enforcement acts only on the current Python process (raise / warn / callback / SIGKILL of os.getpid()) — it is not cluster-level workload termination. The connected Lutflow platform (fleet-wide cost intelligence) is in private development; see Sentinel for its current assisted-Beta status.

In-process budget enforcement

Wrap your provider client; Lutflow meters every call and enforces the limit locally. This runs entirely in your process — no account, no network.

from lutflow import Client, BudgetStrategy
import openai

client = Client(
    tenant_id="acme",
    budget_usd=10.00,
    on_budget_exceeded=BudgetStrategy.RAISE_ERROR,
)
wrapped = client.wrap(openai.OpenAI())
response = wrapped.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)

print(f"Spent:     ${client.accumulated_cost_usd:.4f}")
print(f"Remaining: ${client.remaining_budget_usd:.4f}")

When the budget is exceeded, the configured strategy fires in this process:

Strategy	Behavior (in-process)
`RAISE_ERROR`	Raises `BudgetExceededError` (default)
`WARN_ONLY`	Logs a warning, continues
`CALLBACK`	Calls a function you provide
`SELF_KILL`	Sends `SIGKILL` to the current process (`os.getpid()`)

A context-manager form is also available:

from lutflow import budget_session

with budget_session(budget_usd=0.50, tenant_id="acme") as session:
    wrapped = session.wrap(openai.OpenAI())
    # ...

GPU-time pricing (self-hosted models)

For self-hosted models (vLLM, TGI, BentoML), price by GPU-time instead of a per-provider API:

from lutflow import Client, PricingMode

client = Client(
    tenant_id="acme",
    budget_usd=5.00,
    pricing_mode=PricingMode.GPU_TIME,
    gpu_type="nvidia-l4",
)
client.start_gpu_timer()
# ... run inference ...
cost = client.stop_gpu_timer()

Quickstart: Measure

The lutflow measure command computes cost and energy metrics per token from a GPU telemetry export. Completely offline and read-only.

lutflow measure --from-telemetry export.jsonl

Command Reference

Usage: lutflow [OPTIONS] COMMAND [ARGS]...

  Lutflow — AI Cost Intelligence for LLM inference.
  Track & enforce your LLM spend in-process; compute cost & energy metrics
  from a telemetry export.

  Start here:
    lutflow measure --from-telemetry <export.jsonl>

Options:
  --version  Show the version and exit.
  --help     Show this message and exit.

Commands:
  measure  Compute cost & energy metrics per token from a GPU telemetry...

Measure Options

lutflow measure --help

Usage: lutflow measure [OPTIONS]

  Compute cost & energy metrics per token from a GPU telemetry export.
  Offline, read-only.

  KPI: joules (and optionally $) per 1000 useful tokens, where useful = prompt
  + generation. Reads a telemetry export (.jsonl or .jsonl.gz) in research or
  production schema, segmented by observed concurrency. Cross-checks against
  an energy counter when one is present. Idle windows (0 useful tokens) are
  undefined.

Options:
  --from-telemetry FILE  Telemetry export (.jsonl or .jsonl.gz): research
                         (cumulative totals) or prod (deltas) format.
                         [required]
  --price-per-hr FLOAT   GPU $/hour to also report $/1M useful tokens (e.g.
                         0.85 for an L4).
  --min-samples INTEGER  Drop concurrency buckets with fewer than N productive
                         intervals (sparse transitions).  [default: 3]
  --help                 Show this message and exit.

Example Output

Concurrency  Samples  J/1k tokens  $/1M tokens
-----------  -------  -----------  -----------
1            42       18.3         $0.12
2            38       12.1         $0.08
4            25       9.4          $0.06

Telemetry Format

The SDK accepts .jsonl or .jsonl.gz files with GPU telemetry in either research or production schema:

Research schema (cumulative totals):

{"timestamp": 1720000000, "gpu_power_watts": 285, "tokens_total": 1000}

Production schema (deltas per interval):

{"timestamp": 1720000000, "gpu_power_watts": 285, "tokens_delta": 50, "interval_ms": 1000}

SDK (PyPI)