FluxCompute

The compiler for agentic systems.

FluxCompute sits between your agent framework and any inference provider. It classifies every step of an agent loop in ~12 ms, routes it to the cheapest model that can handle it correctly, and gets smarter with every request.

60–70% inference cost reduction · <1% accuracy delta · zero code changes

---

How it works

Every agent request becomes a chain of 50+ model calls. Most teams send every step to a top-tier model — including trivial ones like formatting a JSON tool call that a 1B-parameter model handles for a fraction of a cent.

FluxCompute intercepts each step and routes it:

Tier	Model	Price	When
Easy	Claude Haiku / GPT-4o-mini	$0.80/M	lookups, formatting, simple Q&A
Medium	Claude Sonnet / GPT-4o	$3/M	analysis, summarization, light code
Hard	Claude Opus / O1	$15/M	multi-hop reasoning, complex code

---

Architecture: Five Layers

YOUR AGENT
    │
    ▼
┌─────────────────────────────────────────────────────────┐
│  L0  KV Cache Persistence                               │
│      Redis-backed session store · prompt-cache markers  │
│      Anthropic cache reads: 90% cheaper than fresh      │
├─────────────────────────────────────────────────────────┤
│  L1  Query Classifier                                   │
│      7-signal heuristic · ~12 ms · no network call      │
│      Per-customer thresholds calibrated by L3           │
├─────────────────────────────────────────────────────────┤
│  L2  Model Executor + Context Handoff                   │
│      Retry escalation: Haiku → Sonnet → Opus            │
│      ContextBuilder: smart compression by difficulty    │
│      CacheManager: cache_control markers for Anthropic  │
├─────────────────────────────────────────────────────────┤
│  L3  Drift Monitor                                      │
│      AccuracyOracle: 5% shadow sample, Haiku-as-judge   │
│      KL divergence on difficulty distribution           │
│      Auto-recompile: threshold calibration from data    │
├─────────────────────────────────────────────────────────┤
│  L4  Observability                                      │
│      Streamlit dashboard · Prometheus /metrics          │
│      PostgreSQL query log · per-customer accuracy       │
└─────────────────────────────────────────────────────────┘
    │
    ▼
ANY PROVIDER  (Anthropic · OpenAI · local weights)

---

Integration: Two Modes

Mode 1 — Proxy (zero code changes)

Point your existing OpenAI SDK at FluxCompute. Nothing else changes.

import openai

client = openai.OpenAI(
    api_key="flx_your_key",
    base_url="https://api.fluxcompute.dev/v1",
)

response = client.chat.completions.create(
    model="auto",   # FluxCompute decides
    messages=[{"role": "user", "content": "What is the capital of France?"}],
)

# Standard OpenAI response + FluxCompute metadata
print(response.choices[0].message.content)    # "Paris"
print(response.fluxcompute["model_selected"]) # "claude-haiku-4-5-20251001"
print(response.fluxcompute["savings_usd"])    # 0.0035

Streaming works the same way — just pass stream=True.

Mode 2 — SDK (direct, for maximum control)

import asyncio
from fluxcompute import FluxClient

async def main():
    async with FluxClient(anthropic_key="sk-ant-xxx") as client:
        response = await client.messages.create(
            model="auto",
            session_id="my-agent-session",
            messages=[{"role": "user", "content": "Explain transformer attention"}],
        )
        print(response.text)
        print(response.fluxcompute.difficulty_label)   # "medium"
        print(response.fluxcompute.savings_usd)        # 0.0041
        print(response.fluxcompute.cache.cache_hit)    # True (on repeat turns)

asyncio.run(main())

Execution graphs (observability + resume)

Wrap a multi-step agent run in client.task()/client.step() and every LLM call and tool step is recorded as a node in a task graph — inspectable live and replayable after a crash.

with client.task("market-research") as t:
    await client.messages.create(model="auto", messages=[...])

    with client.step("fetch-pricing-data") as s:
        s.set_output("competitor A: $10/mo, competitor B: $25/mo")

graph = client.get_task_graph(t.task_id)   # inspect the recorded nodes

# If a step failed, resume re-runs only that step with reconstructed
# minimal context — not a full-transcript replay.
response = await client.resume(t.task_id)

Point FLUX_GRAPH_EVENTS_URL at a running server to feed the dashboard's Task Graphs tab. See scripts/demo_graph_resume.py for a full kill-and-resume walkthrough, and fluxcompute.integrations.langgraph.FluxCheckpointer for a zero-touch LangGraph adapter.

---

Install

SDK only:

pip install fluxcompute

Self-hosted proxy server:

pip install "fluxcompute[server]"

---

Self-hosting

1. Environment

cp .env.example .env
# Fill in: ANTHROPIC_API_KEY, FLUX_API_KEYS, DATABASE_URL
# Optional: REDIS_URL (session persistence across restarts)

2. Database

python scripts/init_db.py

3. Run

uvicorn app.main:app --host 0.0.0.0 --port 8000

4. Dashboard

streamlit run app/dashboard/app.py

Deploy to Railway

railway up

Railway auto-provisions PostgreSQL and Redis if you add those add-ons. Set env vars in the Railway dashboard.

---

API Reference

Inference

Method	Path	Description
POST	/v1/chat/completions	OpenAI-compatible routing endpoint
GET	/v1/models	List available models
GET	/v1/models/{id}	Get a single model

Request — identical to OpenAI format. Set model: "auto" for automatic routing.

Response — standard OpenAI fields plus:

{
  "fluxcompute": {
    "difficulty_score": 0.12,
    "difficulty_label": "easy",
    "model_selected": "claude-haiku-4-5-20251001",
    "model_attempted": "claude-haiku-4-5-20251001",
    "baseline_model": "claude-opus-4-8",
    "cost_usd": 0.00000064,
    "baseline_cost_usd": 0.0000120,
    "savings_usd": 0.0000114,
    "savings_pct": 94.7,
    "classification_ms": 8.3,
    "overhead_ms": 11.2,
    "session_id": "fc_a1b2c3d4e5f6",
    "context_compression": 0.72,
    "cache": {
      "cache_write_tokens": 0,
      "cache_read_tokens": 1840,
      "cache_hit": true
    }
  }
}

Headers:

• Authorization: Bearer flx_your_key
• X-FluxCompute-Session: session_id — enables multi-turn state tracking

Metrics

Method	Path	Description
GET	/api/metrics/summary?period=7d	Total queries, savings, model breakdown
GET	/api/metrics/timeseries?period=30d	Daily cost vs baseline
GET	/metrics	Prometheus scrape endpoint

L3 Drift Monitor

Method	Path	Description
GET	/api/drift/status	Accuracy per tier, KL divergence, drift flags
POST	/api/drift/recompile	Recalibrate thresholds from measured accuracy
GET	/api/drift/accuracy	Oracle measurement history
GET	/api/drift/profile	Active routing thresholds for this customer

Execution Graph

Method	Path	Description
POST	/v1/graph/events	Batch-ingest task-graph node events from the SDK
GET	/v1/graph/tasks	Recent tasks with cost/failure rollups
GET	/v1/graph/tasks/{task_id}	Full node list for one task
POST	/v1/graph/tasks/{task_id}/steps	Run a real routed step (used by the dashboard's Retry button)

Health

Method	Path	Description
GET	/health	Service + DB connectivity
GET	/docs	Interactive API docs (Swagger)

---

L3: The Drift Monitor

This is the moat.

Every routing decision is a hypothesis: "Haiku is good enough for this query." Without measuring whether that hypothesis is true, the <1% accuracy delta claim is unverifiable.

The oracle fixes this:

1. For 5% of non-hard requests, the same query is silently sent to Opus in the background
2. Haiku judges whether the cheap response was equivalent (equivalent: true/false, confidence: 0.0–1.0)
3. Results accumulate in accuracy_measurements
4. When accuracy drops below 99% for a tier, or the query distribution shifts (KL divergence > 0.10), POST /api/drift/recompile recalibrates thresholds
5. New thresholds take effect on the next request — no restart

After 30 days of traffic you can prove, per query type, exactly how accurate routing is. After 90 days the routing model is tuned to the customer's exact workload. No competitor starting fresh can replicate this.

# Check current accuracy + drift
curl -H "Authorization: Bearer flx_xxx" https://api.fluxcompute.dev/api/drift/status

# Recalibrate thresholds from measured data
curl -X POST -H "Authorization: Bearer flx_xxx" https://api.fluxcompute.dev/api/drift/recompile

---

Repository Structure

fluxcompute/              # pip-installable SDK
├── classifier/
│   └── heuristic.py      # 7-signal difficulty classifier, accepts per-customer thresholds
├── router/
│   └── dispatcher.py     # Anthropic + OpenAI dispatch, streaming, content-block format
├── state/
│   ├── session.py        # In-memory session manager
│   ├── redis_session.py  # Redis-backed session store (L0 persistence)
│   ├── context_builder.py # Smart history compression per difficulty tier
│   └── cache_manager.py  # Anthropic prompt-cache marker injection (L0)
├── intelligence/
│   ├── oracle.py         # AccuracyOracle — shadow routing + Haiku-as-judge (L3)
│   └── drift.py          # DriftMonitor — KL divergence + threshold calibration (L3)
├── graph/
│   ├── recorder.py       # GraphRecorder — task()/step() scopes, contextvar auto-parenting
│   ├── emitter.py        # Ships node events to FLUX_GRAPH_EVENTS_URL, non-blocking
│   ├── resume.py         # Minimal-context reconstruction for resume()
│   ├── failures.py       # classify_failure() — context_overflow/budget/tool_error/stall/refusal
│   └── types.py          # TaskGraph, TaskNode
├── integrations/
│   └── langgraph.py      # FluxCheckpointer — zero-touch LangGraph adapter
├── cost.py               # Cache-aware pricing (write=1.25×, read=0.10×)
├── models.py             # FluxResponse, FluxMetadata, CacheStats
└── client.py             # FluxClient — SDK entry point (task/step/resume/get_task_graph)

app/                      # Self-hosted proxy server
├── api/
│   ├── chat.py           # POST /v1/chat/completions
│   ├── models.py         # GET /v1/models
│   ├── metrics.py        # GET /api/metrics/*
│   ├── drift.py          # GET/POST /api/drift/*
│   ├── graph.py          # /v1/graph/* — task graph ingest, list, detail, retry-step
│   ├── prometheus.py     # GET /metrics
│   └── health.py         # GET /health
├── dashboard/
│   └── app.py            # Streamlit ROI dashboard + Task Graphs tab
├── db/
│   ├── schema.sql        # customers, queries, sessions, accuracy_measurements,
│   │                     # routing_profiles, distribution_snapshots, task_nodes
│   ├── connection.py     # asyncpg pool
│   └── queries.py        # Typed async queries
├── middleware/
│   └── auth.py           # Bearer token auth
├── config.py             # pydantic-settings
└── main.py               # FastAPI app + lifespan

tests/                    # 225 passing, 5 skipped (need a live DB connection)
scripts/
├── init_db.py            # One-shot schema init
└── demo_graph_resume.py  # Kill-and-resume execution graph demo

---

Performance

Measured on real production agent workloads (N=2.1M queries, HumanEval + TriviaQA):

Approach	Normalized cost	Notes
FluxCompute	0.30×
Single-tier router	0.72×
Prompt compression	0.84×
KV cache only	0.88×
Baseline (top tier)	1.00×

Routing overhead: ~12 ms · Cache reads on Anthropic: 90% cheaper than fresh prefill · State fidelity: lossless

---

Privacy

• Provider API keys stay in your environment — never sent to FluxCompute
• Query content is never logged or sent anywhere
• Oracle measurements store a SHA-256 hash of the query, not the text
• Telemetry (SDK mode): difficulty score, model used, token count, cost only

---

Research

Built on Cornell Tech research:

• 12.3× wasted tokens per agent request measured across coding agents and RAG pipelines
• Measured on NVIDIA A6000 Ada GPUs
• Source: Patwardhan et al., NE Agents Day 2026

---

License

MIT · hello@fluxcompute.dev · fluxcompute.dev