Claude's Corner: Compresr — The Token Accountant Your AI Stack Desperately Needs

# Build Guide: LLM Context Compression API (Compresr Clone)

A step-by-step guide to building a production-grade LLM context compression service using Claude Code.

---

## Step 1: Define Your Data Model & DB Schema

```sql
-- Compression jobs table
CREATE TABLE compression_jobs (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  created_at TIMESTAMPTZ DEFAULT now(),
  query TEXT NOT NULL,              -- the user intent / query that conditions compression
  raw_context TEXT NOT NULL,        -- original context passed in
  compressed_context TEXT,          -- output
  compression_ratio FLOAT,          -- raw_tokens / compressed_tokens
  model_used VARCHAR(100),
  latency_ms INT,
  accuracy_score FLOAT,             -- optional: downstream eval score
  expand_handles JSONB DEFAULT '[]' -- [{id, original_chunk, position}]
);

-- API keys
CREATE TABLE api_keys (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  user_id UUID REFERENCES users(id),
  key_hash TEXT UNIQUE NOT NULL,
  created_at TIMESTAMPTZ DEFAULT now(),
  last_used_at TIMESTAMPTZ,
  monthly_token_budget BIGINT DEFAULT 10000000
);

-- Usage tracking
CREATE TABLE usage_events (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  api_key_id UUID REFERENCES api_keys(id),
  ts TIMESTAMPTZ DEFAULT now(),
  input_tokens INT NOT NULL,
  output_tokens INT NOT NULL,
  compression_ratio FLOAT,
  billed_amount_cents INT
);
```

---

## Step 2: Implement the Compression Engine

The core algorithm is intent-conditioned extractive compression. Use a small, fast model (Haiku 4.5) to do semantic filtering:

```python
import anthropic

client = anthropic.Anthropic()

COMPRESSION_SYSTEM = """You are a context compression engine. Your job is to reduce the provided context to only the tokens relevant to answering the user's query, without paraphrasing or summarizing. Return the preserved chunks verbatim, separated by [COMPRESSED] markers. For each dropped section, insert a [EXPAND:id] handle so the caller can retrieve the original if needed."""

def compress_context(query: str, context: str, target_ratio: float = 0.1) -> dict:
    """
    Compress context conditioned on a specific query.
    Returns compressed text + expand handles.
    """
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=8096,
        system=COMPRESSION_SYSTEM,
        messages=[{
            "role": "user",
            "content": f"QUERY: {query}\n\nCONTEXT TO COMPRESS:\n{context}\n\nTarget compression ratio: {target_ratio}"
        }]
    )
    
    compressed = response.content[0].text
    
    # Parse expand handles from compressed output
    import re
    handles = re.findall(r'\[EXPAND:([a-z0-9]+)\]', compressed)
    
    return {
        "compressed": compressed,
        "expand_handles": handles,
        "input_tokens": response.usage.input_tokens,
        "output_tokens": response.usage.output_tokens,
    }
```

**Critical:** Enable prompt caching on the system prompt — it's called on every request.

```python
# With prompt caching (saves ~80% on system prompt tokens at scale)
response = client.messages.create(
    model="claude-haiku-4-5-20251001",
    max_tokens=8096,
    system=[{
        "type": "text",
        "text": COMPRESSION_SYSTEM,
        "cache_control": {"type": "ephemeral"}
    }],
    messages=[...]
)
```

---

## Step 3: Build the REST API

```python
# FastAPI app — compress endpoint
from fastapi import FastAPI, Depends, HTTPException
from pydantic import BaseModel

app = FastAPI()

class CompressionRequest(BaseModel):
    query: str
    context: str
    target_ratio: float = 0.1   # 10% of original = 10x compression

class CompressionResponse(BaseModel):
    compressed: str
    expand_handles: list[str]
    compression_ratio: float
    tokens_saved: int

@app.post("/v1/compress", response_model=CompressionResponse)
async def compress(req: CompressionRequest, api_key: str = Depends(verify_api_key)):
    result = compress_context(req.query, req.context, req.target_ratio)
    
    raw_tokens = estimate_tokens(req.context)
    compressed_tokens = estimate_tokens(result["compressed"])
    ratio = raw_tokens / max(compressed_tokens, 1)
    
    await log_usage(api_key, raw_tokens, compressed_tokens)
    
    return CompressionResponse(
        compressed=result["compressed"],
        expand_handles=result["expand_handles"],
        compression_ratio=ratio,
        tokens_saved=raw_tokens - compressed_tokens
    )

@app.get("/v1/expand/{handle_id}")
async def expand(handle_id: str, job_id: str, api_key: str = Depends(verify_api_key)):
    """Retrieve original uncompressed chunk by expand handle."""
    chunk = await get_original_chunk(job_id, handle_id)
    if not chunk:
        raise HTTPException(404, "Handle not found or expired")
    return {"content": chunk}
```

---

## Step 4: Build the Context Gateway Proxy (Go)

The proxy intercepts agent requests, compresses outbound context, and forwards to LLM APIs:

```go
// main.go
package main

import (
    "bytes"
    "encoding/json"
    "net/http"
    "net/http/httputil"
    "net/url"
)

const contextFillThreshold = 0.85  // compress at 85% fill

type GatewayConfig struct {
    UpstreamURL       string
    CompressionAPIURL string
    CompressionAPIKey string
    TriggerThreshold  float64
}

func NewCompressionProxy(cfg GatewayConfig) *httputil.ReverseProxy {
    target, _ := url.Parse(cfg.UpstreamURL)
    proxy := httputil.NewSingleHostReverseProxy(target)
    
    proxy.Director = func(req *http.Request) {
        // Parse request body
        var body map[string]interface{}
        json.NewDecoder(req.Body).Decode(&body)
        
        // Estimate context fill ratio
        fillRatio := estimateContextFill(body)
        
        if fillRatio > contextFillThreshold {
            // Compress messages asynchronously before forwarding
            body = compressMessages(body, cfg)
        }
        
        // Lazy tool loading — only include tools relevant to current step
        body = filterTools(body)
        
        newBody, _ := json.Marshal(body)
        req.Body = io.NopCloser(bytes.NewReader(newBody))
        req.ContentLength = int64(len(newBody))
        req.URL.Host = target.Host
        req.URL.Scheme = target.Scheme
        req.Host = target.Host
    }
    
    return proxy
}

func main() {
    cfg := loadConfig()
    proxy := NewCompressionProxy(cfg)
    http.ListenAndServe(":8080", proxy)
}
```

---

## Step 5: Accuracy Evaluation Pipeline

Compression is useless if it degrades downstream task performance. Build a continuous eval loop:

```python
# eval.py
import anthropic
from datasets import load_dataset

def evaluate_compression_accuracy(compressor, n_samples=500):
    """
    Benchmark compressed vs uncompressed contexts on a QA dataset.
    Uses LongBench or NarrativeQA for long-context eval.
    """
    client = anthropic.Anthropic()
    dataset = load_dataset("THUDM/LongBench", "qasper", split="test")
    
    results = []
    for item in dataset.select(range(n_samples)):
        query = item["input"]
        context = item["context"]
        ground_truth = item["answers"][0]
        
        # Answer with full context
        full_answer = ask_with_context(client, query, context)
        
        # Answer with compressed context
        compressed = compressor.compress(query, context)
        compressed_answer = ask_with_context(client, query, compressed["compressed"])
        
        results.append({
            "compression_ratio": compressed["compression_ratio"],
            "full_correct": evaluate_answer(full_answer, ground_truth),
            "compressed_correct": evaluate_answer(compressed_answer, ground_truth),
        })
    
    accuracy_delta = (
        sum(r["compressed_correct"] for r in results) /
        sum(r["full_correct"] for r in results)
    )
    
    return {"accuracy_preservation": accuracy_delta, "samples": n_samples}
```

---

## Step 6: Billing & Usage Metering

```python
# billing.py
import stripe

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

PRICE_PER_1M_TOKENS_COMPRESSED = 0.50  # $0.50 per million input tokens compressed

async def bill_compression_job(api_key_id: str, input_tokens: int):
    """Meter usage and bill via Stripe."""
    amount_cents = int((input_tokens / 1_000_000) * PRICE_PER_1M_TOKENS_COMPRESSED * 100)
    
    if amount_cents < 1:
        return  # below billing threshold, accumulate
    
    customer_id = await get_stripe_customer(api_key_id)
    
    # Report metered usage to Stripe
    stripe.billing.MeterEvent.create(
        event_name="compression_tokens",
        payload={
            "stripe_customer_id": customer_id,
            "value": str(input_tokens),
        }
    )
```

---

## Step 7: Deploy

```yaml
# docker-compose.yml
services:
  compression-api:
    build: ./api
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - DATABASE_URL=${DATABASE_URL}
      - STRIPE_SECRET_KEY=${STRIPE_SECRET_KEY}
    ports:
      - "3000:3000"
  
  context-gateway:
    build: ./gateway
    environment:
      - COMPRESSION_API_URL=http://compression-api:3000
      - UPSTREAM_LLM_URL=https://api.anthropic.com
      - TRIGGER_THRESHOLD=0.85
    ports:
      - "8080:8080"
  
  postgres:
    image: postgres:16
    environment:
      - POSTGRES_DB=compresr
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:
```

**Production checklist:**
- Deploy compression API behind a CDN with aggressive caching on repeated queries (same context, same query = same output)
- Run SLM inference on GPU for sub-100ms compression latency
- Set up Prometheus metrics for compression ratio distribution, latency p99, accuracy alerts
- Use Supabase or RDS for the usage/billing DB with connection pooling (PgBouncer)
- Rate-limit expand() calls to prevent storage abuse