← Back to DevBytes
Developer Tutorial August 19, 2026 9 min read

How to Audit LLM API Access and Usage

How to Audit LLM API Access and Usage

As large language models (LLMs) become embedded in production applications, the APIs that power them become critical infrastructure. Every prompt sent, every token consumed, and every response generated represents both a cost and a potential security surface. Auditing LLM API access and usage is the practice of systematically recording, analyzing, and governing how your organization interacts with LLM providers like OpenAI, Anthropic, Google, and others. This tutorial walks through what LLM API auditing involves, why it matters, how to implement it, and the best practices that separate a robust audit pipeline from a noisy log dump.

What Is LLM API Auditing?

LLM API auditing is the continuous process of capturing metadata and content related to calls made to language model endpoints, then analyzing that data for security, compliance, cost, and performance insights. Unlike traditional API auditing, LLM auditing must account for the unique characteristics of generative AI: prompts may contain sensitive user data, responses may include hallucinated or harmful content, and usage patterns can reveal prompt injection attempts or data exfiltration.

A complete audit system typically captures the following dimensions for every API call:

Why Auditing LLM API Access Matters

The stakes for LLM API auditing are higher than for most traditional APIs because the data flowing through these endpoints is both valuable and risky. Several concrete concerns drive the need for rigorous auditing:

How to Implement LLM API Auditing

A practical auditing implementation has three layers: a capture layer that records calls, a storage layer that persists audit records, and an analysis layer that turns records into actionable insights. The most effective pattern is to intercept calls at a centralized gateway or wrapper rather than instrumenting each call site individually.

1. Build a Centralized API Wrapper

Instead of calling provider SDKs directly throughout your codebase, route all LLM calls through a single wrapper service. This gives you a single chokepoint for logging, policy enforcement, and transformation. Here is a minimal Python example using FastAPI:

from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel
import time, uuid, json, structlog
from datetime import datetime

app = FastAPI()
logger = structlog.get_logger()

class LLMRequest(BaseModel):
    model: str
    messages: list
    temperature: float = 0.7
    max_tokens: int = 1000
    user_id: str
    feature: str  # which product feature triggered this call

@app.post("/v1/chat")
async def chat_proxy(req: LLMRequest, request: Request):
    audit_id = str(uuid.uuid4())
    start = time.time()

    audit_record = {
        "audit_id": audit_id,
        "timestamp": datetime.utcnow().isoformat(),
        "user_id": req.user_id,
        "feature": req.feature,
        "model": req.model,
        "parameters": {
            "temperature": req.temperature,
            "max_tokens": req.max_tokens,
        },
        "prompt_messages": req.messages,
        "client_ip": request.client.host,
        "api_key_hint": request.headers.get("x-api-key", "")[-4:],
    }

    try:
        # Forward to actual provider (OpenAI shown here)
        from openai import OpenAI
        client = OpenAI()
        response = client.chat.completions.create(
            model=req.model,
            messages=req.messages,
            temperature=req.temperature,
            max_tokens=req.max_tokens,
        )

        latency_ms = int((time.time() - start) * 1000)
        audit_record.update({
            "status": "success",
            "response": response.choices[0].message.content,
            "finish_reason": response.choices[0].finish_reason,
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens,
            "latency_ms": latency_ms,
        })

        logger.info("llm_call_completed", **audit_record)
        return {"audit_id": audit_id, "response": response.choices[0].message.content}

    except Exception as e:
        audit_record.update({
            "status": "error",
            "error_type": type(e).__name__,
            "error_message": str(e),
            "latency_ms": int((time.time() - start) * 1000),
        })
        logger.error("llm_call_failed", **audit_record)
        raise HTTPException(status_code=500, detail=str(e))

This wrapper captures every dimension of the call before and after it hits the provider. The audit_id returned to the caller lets downstream services correlate the LLM call with their own traces.

2. Store Audit Records in a Queryable Backend

Structured logs are a good start, but for real analysis you need a database. For moderate volume, PostgreSQL with JSONB columns works well. For high volume, a columnar store like ClickHouse or a search engine like Elasticsearch is more appropriate. Here is a schema and insertion example for PostgreSQL:

CREATE TABLE llm_audit_log (
    audit_id          UUID PRIMARY KEY,
    timestamp         TIMESTAMPTZ NOT NULL,
    user_id           TEXT NOT NULL,
    feature           TEXT NOT NULL,
    model             TEXT NOT NULL,
    status            TEXT NOT NULL,
    prompt_tokens     INTEGER,
    completion_tokens INTEGER,
    total_tokens      INTEGER,
    latency_ms        INTEGER,
    cost_usd          NUMERIC(10, 6),
    prompt_hash       TEXT,
    response_hash     TEXT,
    metadata          JSONB NOT NULL DEFAULT '{}'::jsonb
);

CREATE INDEX idx_audit_user_time ON llm_audit_log(user_id, timestamp DESC);
CREATE INDEX idx_audit_feature_time ON llm_audit_log(feature, timestamp DESC);
CREATE INDEX idx_audit_model_time ON llm_audit_log(model, timestamp DESC);
CREATE INDEX idx_audit_metadata ON llm_audit_log USING GIN(metadata);

Insert records asynchronously so logging never blocks the response path. A simple background task approach using FastAPI:

import asyncio
import asyncpg

async def persist_audit(record: dict):
    conn = await asyncpg.connect("postgresql://audit_user:pass@localhost/audit")
    await conn.execute(
        """
        INSERT INTO llm_audit_log
        (audit_id, timestamp, user_id, feature, model, status,
         prompt_tokens, completion_tokens, total_tokens, latency_ms, metadata)
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
        """,
        record["audit_id"], record["timestamp"], record["user_id"],
        record["feature"], record["model"], record["status"],
        record.get("prompt_tokens"), record.get("completion_tokens"),
        record.get("total_tokens"), record.get("latency_ms"),
        json.dumps(record),
    )
    await conn.close()

# In the wrapper, after building audit_record:
asyncio.create_task(persist_audit(audit_record))

3. Compute Cost and Detect Anomalies

Token counts alone are not enough; you need to translate them into cost. Maintain a pricing table per model and compute cost at audit time. Then run anomaly detection over rolling windows to catch spikes.

PRICING = {
    "gpt-4o":        {"input": 0.000005,   "output": 0.000015},
    "gpt-4o-mini":   {"input": 0.00000015, "output": 0.0000006},
    "claude-3-5-sonnet": {"input": 0.000003, "output": 0.000015},
}

def compute_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    if model not in PRICING:
        return 0.0
    p = PRICING[model]
    return (prompt_tokens * p["input"]) + (completion_tokens * p["output"])

# Anomaly check: flag any single call costing more than $1
def is_cost_anomaly(record: dict) -> bool:
    return record.get("cost_usd", 0) > 1.0

For broader anomaly detection, query the database for rolling statistics and alert when a user or feature exceeds historical baselines:

SELECT
    user_id,
    DATE_TRUNC('hour', timestamp) AS hour,
    SUM(cost_usd) AS hourly_cost,
    COUNT(*) AS call_count
FROM llm_audit_log
WHERE timestamp > NOW() - INTERVAL '24 hours'
GROUP BY user_id, hour
HAVING SUM(cost_usd) > 50.00
ORDER BY hourly_cost DESC;

4. Scan Prompts for Sensitive Data

Auditing is not just about recording; it is about detecting problems. Run prompt content through a redaction or classification step before logging. The example below uses regex for common patterns, but in production you should use a dedicated PII detection library or service.

import re

PATTERNS = {
    "ssn": re.compile(r"\b\d{3}-\d{2}-\d{4}\b"),
    "credit_card": re.compile(r"\b(?:\d[ -]*?){13,16}\b"),
    "email": re.compile(r"\b[\w.-]+@[\w.-]+\.\w+\b"),
    "api_key": re.compile(r"\bsk-[a-zA-Z0-9]{20,}\b"),
}

def scan_prompt(messages: list) -> list:
    findings = []
    for msg in messages:
        content = msg.get("content", "")
        for pii_type, pattern in PATTERNS.items():
            if pattern.search(content):
                findings.append({
                    "pii_type": pii_type,
                    "role": msg.get("role"),
                })
    return findings

# Add to audit_record before persisting:
audit_record["pii_findings"] = scan_prompt(req.messages)
if audit_record["pii_findings"]:
    logger.warning("pii_detected_in_prompt", audit_id=audit_id,
                   findings=audit_record["pii_findings"])

5. Enforce Access Policies

Auditing and policy enforcement are two sides of the same coin. Use the audit data to drive real-time decisions: block calls from users over their quota, deny requests to disallowed models, or redact sensitive content before it reaches the provider.

RATE_LIMITS = {
    "free_tier":  {"calls_per_hour": 100, "tokens_per_day": 50000},
    "pro_tier":   {"calls_per_hour": 1000, "tokens_per_day": 500000},
}

async def check_policy(user_id: str, tier: str, model: str, estimated_tokens: int) -> bool:
    # Check model allowlist
    allowed_models = {"gpt-4o-mini"} if tier == "free_tier" else {"gpt-4o", "gpt-4o-mini"}
    if model not in allowed_models:
        return False

    # Check hourly call rate
    recent_calls = await fetch_recent_call_count(user_id, hours=1)
    if recent_calls >= RATE_LIMITS[tier]["calls_per_hour"]:
        return False

    # Check daily token budget
    daily_tokens = await fetch_daily_token_usage(user_id)
    if daily_tokens + estimated_tokens > RATE_LIMITS[tier]["tokens_per_day"]:
        return False

    return True

Best Practices for LLM API Auditing

Implementing the mechanics above is only half the battle. The following practices ensure your audit system remains useful, secure, and maintainable over time.

Conclusion

Auditing LLM API access and usage is not a one-time setup but an ongoing discipline that grows in importance as your reliance on language models deepens. By centralizing calls through a wrapper, persisting structured records to a queryable store, computing cost and detecting anomalies, scanning for sensitive data, and enforcing policies in real time, you build a system that gives you visibility, control, and accountability. The investment pays off the first time you catch a cost spike before it spirals, detect a prompt injection attempt before it succeeds, or produce an audit trail that satisfies a compliance review. Start simple, instrument everything, and refine your policies as the data teaches you what normal and abnormal look like for your specific applications.

— Ad —

Google AdSense will appear here after approval

← Back to all articles