Build an LLM API Gateway

Build a FastAPI gateway that routes LLM requests by cost with audit logs.

Introduction

30 Second Summary

Every time an app sends a message to an AI model, something has to decide which model handles it, what happens if that model fails, and how much it all costs. In production, that "something" is an API gateway.

In this project, you will build NeuralGate, a FastAPI-based LLM gateway that routes requests between models based on cost, handles failures with automatic fallback chains, and reports every decision to a live observability dashboard.

What You'll Build

NeuralGate is an async Python API gateway that sits between your applications and LLM providers, making routing decisions, controlling concurrency, and logging every interaction for audit.

By the end of this project, you'll have:

  • A FastAPI gateway with an OpenAI-compatible chat endpoint that routes requests between models based on cost.
  • Structured JSONL audit logs capturing every prompt, response, model decision, cost, and latency.
  • A live Langfuse observability dashboard showing traces, costs, and latency across all gateway requests.
  • Secret Mission: A budget ceiling that tracks cumulative spend and rejects requests when the limit is hit.

Are there any prerequisites?

You need basic-to-intermediate Python scripting skills. No async, FastAPI, or LLM API experience is required.

Set Up Your Environment

NeuralGate needs a few pieces of infrastructure before you can start building. You need Python to write the gateway, API keys to connect to LLM providers, and an observability platform to watch what happens.

In this step, you will install Python 3.12 and Cursor, create the project folder, grab your OpenAI and Langfuse credentials, and run a quick test to confirm everything talks to each other.

In this step, get ready to:
  • Install Python 3.12 and Cursor, then create the project folder with a virtual environment.
  • Get your OpenAI and Langfuse Cloud API keys.
  • Verify the full setup works with a test API call.
Install Python 3.12, Cursor, and create the project

Python is the language your gateway is built in. Cursor is the code editor you will use to write it.

  • Press the Windows key to open your search bar.
  • Type PowerShell and press Enter to open it.
  • Check if Python is already installed by running this command:
python --version

✔️ I see version 3.12 or higher

Python is ready to go. Skip ahead to the Cursor install below.

ⓧ I see an older version

You have Python installed, but the gateway needs 3.12 or higher. Download the latest Python 3.12 installer from the official site.

  • Open python.org/downloads in your browser.
  • Scroll down to the Python 3.12 release schedule section and click the latest 3.12.x release.
  • Download the Windows installer (64-bit) (marked as Recommended).
  • Run the installer. On the first screen, check the box Add python.exe to PATH before clicking Install Now.
  • Close and reopen PowerShell, then run python --version again to confirm it shows 3.12.

ⓧ Command not found

Python is not installed yet. Download it from the official site.

  • Open python.org/downloads in your browser.
  • Scroll down to the Python 3.12 release schedule section and click the latest 3.12.x release.
  • Download the Windows installer (64-bit) (marked as Recommended).
  • Run the installer. On the first screen, check the box Add python.exe to PATH before clicking Install Now.
  • Close and reopen PowerShell, then run python --version again to confirm it shows 3.12.

Now install Cursor, the AI-powered code editor you will use throughout this project.

  • Open cursor.com in your browser.
  • Click the download button and run the installer.
  • Once installed, press the Windows key, type Cursor, and press Enter to open it.

Next, create the project folder on your Desktop so you can find it easily.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create the project folder and move into it by running these commands:
mkdir neuralgate
cd neuralgate
  • Confirm the folder exists by running:
ls

You should see an empty directory listing confirming you are inside the neuralgate folder.

Now set up an isolated Python environment so this project's packages do not interfere with anything else on your system.

  • Create the virtual environment by running:
python -m venv venv
  • Activate the virtual environment by running:
.\venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt.

What is a virtual environment?

A virtual environment is a self-contained folder that holds its own copy of Python and any packages you install. This prevents version conflicts between different projects on the same machine.

Not seeing (venv) in your prompt?

Windows may block script execution by default. Run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned then try the activate command again.

help me activate my Python virtual environment on Windows

With the virtual environment active, create the file that lists all the packages your gateway needs.

  • In Cursor, click File in the top menu bar.
  • Click Open Folder.
  • Select the neuralgate folder on your Desktop, then click Select Folder.
  • Create a new file called requirements.txt by pressing Ctrl+N, then paste the following content:
fastapi==0.138.0
uvicorn[standard]==0.34.0
litellm==1.89.1
python-dotenv==1.1.0
opentelemetry-api==1.33.0
opentelemetry-sdk==1.33.0
opentelemetry-exporter-otlp==0.54b0

What are these packages?

FastAPI is the web framework for your gateway's API. Uvicorn is the async server that runs it. LiteLLM handles routing requests to different LLM providers.

The python-dotenv package loads your API keys from a .env file. The OpenTelemetry packages are dependencies that Langfuse uses under the hood for tracing.

  • Save the file with Ctrl+S. When prompted for a name, type requirements.txt and save it in the neuralgate folder.
  • Install all the packages by running this command in your terminal:
pip install -r requirements.txt

This will download and install all seven packages plus their dependencies. It may take a minute or two.

Seeing a permissions error or install failure?

Make sure (venv) appears in your terminal prompt. If it does not, you are installing globally instead of into the virtual environment. Run .\venv\Scripts\activate first.

help me fix pip install errors in my virtual environment

Get your API keys

The gateway needs two sets of credentials. An OpenAI key lets LiteLLM call GPT models. Langfuse keys connect the gateway to your observability dashboard.

  • Open platform.openai.com in your browser.
  • Sign up for an account (or sign in if you already have one).
  • Navigate to API Keys in the left sidebar.
  • Click Create new secret key.
  • Copy the key and save it somewhere safe. You will not be able to see it again.

Why can't I see my key again?

OpenAI only shows the full key once at creation. After that, it is masked for security. If you lose it, you will need to create a new one.

Now get your Langfuse Cloud credentials.

  • Open cloud.langfuse.com in your browser.
  • Sign up for a free account. No credit card is required.
  • Create a new project (name it anything you like, such as neuralgate).
  • Go to Settings, then API Keys.
  • Click Create new API credentials.
  • Copy both the Public Key (starts with pk-lf-) and the Secret Key (starts with sk-lf-).

What is Langfuse Cloud?

Langfuse is an open-source LLM observability platform. The Cloud version gives you a free Hobby tier with 50k observations per month. It shows you traces, costs, and latency for every request your gateway handles.

Now store all three keys in a .env file so your code can load them securely.

  • In Cursor, create a new file called .env in the root of your neuralgate folder.
  • Paste the following content, replacing the placeholder values with your actual keys:
OPENAI_API_KEY=your-openai-api-key-here
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key-here
LANGFUSE_SECRET_KEY=your-langfuse-secret-key-here
  • Save the file with Ctrl+S.

Keep your .env file private

Never commit this file to version control. If you add a .gitignore later, include .env in it.

Verify the setup works

Before building the full gateway, confirm that LiteLLM can reach OpenAI using your key. A quick test script will prove everything is wired up correctly.

  • In Cursor, create a new file called test_setup.py in the root of your neuralgate folder.
  • Paste the following code:
"""Quick test: verify LiteLLM can call OpenAI with your API key."""

import litellm
from dotenv import load_dotenv

# Load API keys from .env
load_dotenv()

# Make a simple completion call through LiteLLM
response = litellm.completion(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Say hello in one sentence."}],
    max_tokens=50,
)

# Print the model's response
print(response.choices[0].message.content)

# Print the cost of this single request
cost = litellm.completion_cost(completion_response=response)
print(f"Cost: ${cost:.6f}")

What does this code do?

The script loads your .env file so LiteLLM can find your OpenAI key. It sends a simple prompt to gpt-4o-mini and prints the response text plus the cost of that single request.

If you see a response and a cost figure, your Python environment, packages, and API key are all working together.

  • Save test_setup.py.
  • Run the test script by executing this command in your terminal:
python test_setup.py

You should see a short greeting from the model followed by a cost of a fraction of a cent (something like $0.000030).

Seeing an authentication or connection error?

Double-check that your .env file is in the same folder as test_setup.py (the root of neuralgate). Make sure there are no spaces around the = sign in each line.

If you see a billing-related error, ensure your OpenAI account has credits loaded at platform.openai.com.

help me debug my LiteLLM test script failing on Windows

✔️ Awesome, I've got everything!

Great. Double check you have saved both requirements.txt and .env in the root of your neuralgate folder.

ⓧ I'd like to double check the full code

Here is exactly what requirements.txt should contain:

fastapi==0.138.0
uvicorn[standard]==0.34.0
litellm==1.89.1
python-dotenv==1.1.0
opentelemetry-api==1.33.0
opentelemetry-sdk==1.33.0
opentelemetry-exporter-otlp==0.54b0

And here is exactly what .env should contain (with your own keys substituted):

OPENAI_API_KEY=your-openai-api-key-here
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key-here
LANGFUSE_SECRET_KEY=your-langfuse-secret-key-here

Your environment is ready. Next up, you will build the actual gateway that routes requests, handles failures, and tracks every penny.

Build the Gateway Foundation

In Step 1, you set up your environment with Python, Cursor, and all the dependencies NeuralGate needs.

Now it's time to build the actual gateway. You'll create a FastAPI application with an OpenAI-compatible chat endpoint. When a request comes in, the gateway routes it through LiteLLM's Router, which picks the cheapest available model and returns the response along with cost and latency metadata.

In this step, get ready to:
  • Create the configuration file that defines your model routing strategy.
  • Build the main FastAPI gateway application with a chat completions endpoint.
  • Test the gateway by sending a request through the Swagger UI.
Create the configuration file

The configuration file centralises all routing decisions in one place. It defines which models are available, how the router picks between them, and what happens when a model fails.

  • In Cursor, create a new file called config.py inside your neuralgate folder.
  • Paste the following code into config.py:
"""NeuralGate configuration: model routing and gateway settings."""

MODEL_CONFIG = {
    "model_list": [
        {
            "model_name": "fast",
            "litellm_params": {
                "model": "gpt-4o-mini",
            },
        },
        {
            "model_name": "quality",
            "litellm_params": {
                "model": "gpt-4o",
            },
        },
    ],
    "fallbacks": [{"quality": ["fast"]}],
    "routing_strategy": "cost-based-routing",
    "num_retries": 2,
}

GATEWAY_SETTINGS = {
    "max_concurrent_requests": 10,
    "budget_limit_usd": 1.00,
}

What does this code do?

MODEL_CONFIG defines two model groups. "fast" maps to gpt-4o-mini (cheapest option) and "quality" maps to gpt-4o (best output).

The fallbacks key tells LiteLLM: if the "quality" model fails, try "fast" instead. The routing_strategy of "cost-based-routing" ensures the cheapest deployment is always preferred. GATEWAY_SETTINGS sets system-wide limits you'll use later for concurrency and budgeting.

  • Save config.py.

Seeing a red underline on the import later?

If Cursor shows an import warning for config in another file, make sure you saved config.py at the top level of your neuralgate folder (not inside a subfolder).

help me fix import errors in my project

Create the main gateway application

The main application file wires everything together. It initialises the LiteLLM Router on startup, defines the request and response shapes, and exposes the chat completions endpoint.

Before creating main.py, you need a placeholder for the audit logger module. The full audit logger is built in a later step, but the import needs to resolve for the server to start.

  • Create a new file called audit_logger.py inside your neuralgate folder.
  • Paste this placeholder code into audit_logger.py:
"""Placeholder for the audit logger (built in a later step)."""


class AuditLogger:
    def __init__(self, log_path: str):
        pass
  • Save audit_logger.py.

Now create the gateway itself. You'll build main.py in three chunks: the application setup, the data models, and the endpoint logic.

  • Create a new file called main.py inside your neuralgate folder.
  • Set up the imports, environment loading, and application lifecycle by pasting this code into main.py:
"""NeuralGate: LLM API Gateway with smart routing and observability."""

import asyncio
import os
import time
import uuid
from contextlib import asynccontextmanager

import litellm
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from litellm import Router
from pydantic import BaseModel

from audit_logger import AuditLogger
from config import GATEWAY_SETTINGS, MODEL_CONFIG

# Load environment variables from .env file
load_dotenv()

# Concurrency control: limits simultaneous LLM requests
semaphore = asyncio.Semaphore(GATEWAY_SETTINGS["max_concurrent_requests"])

# LiteLLM Router instance (initialized during app startup)
router: Router | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Initialize the LiteLLM Router on startup, clean up on shutdown."""
    global router
    router = Router(
        model_list=MODEL_CONFIG["model_list"],
        fallbacks=MODEL_CONFIG["fallbacks"],
        routing_strategy=MODEL_CONFIG["routing_strategy"],
        num_retries=MODEL_CONFIG["num_retries"],
    )
    yield


app = FastAPI(
    title="NeuralGate",
    description="LLM API Gateway with cost-based routing and observability",
    lifespan=lifespan,
)

What does this code do?

load_dotenv() reads your .env file so your OpenAI key is available to LiteLLM without hardcoding it.

The semaphore caps how many LLM requests run at the same time (10 in this case). This prevents your gateway from overwhelming the upstream provider.

The lifespan function creates the LiteLLM Router when the server boots. It loads your model list, fallback chain, and routing strategy from config.py. Everything before yield runs at startup. Everything after yield runs at shutdown.

  • Add the request and response models below the app declaration in main.py:

class Message(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    model: str = "fast"
    messages: list[Message]
    max_tokens: int = 1024
    temperature: float = 0.7


class ChatResponse(BaseModel):
    id: str
    model: str
    content: str
    usage: dict
    cost_usd: float
    latency_ms: float

What are these models for?

Message represents a single chat message with a role (like "user" or "assistant") and content (the text).

ChatRequest defines what callers send to your gateway. It defaults to the "fast" model group unless the caller specifies otherwise. ChatResponse defines what the gateway returns, including metadata like cost_usd and latency_ms that you won't find in standard OpenAI responses.

  • Add the two endpoint functions below the model classes in main.py:

@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest) -> ChatResponse:
    """Route a chat completion request through the gateway."""
    request_id = str(uuid.uuid4())
    start_time = time.time()

    async with semaphore:
        try:
            response = await router.acompletion(
                model=request.model,
                messages=[m.model_dump() for m in request.messages],
                max_tokens=request.max_tokens,
                temperature=request.temperature,
            )

            latency_ms = (time.time() - start_time) * 1000
            cost = litellm.completion_cost(completion_response=response)

            result = ChatResponse(
                id=request_id,
                model=response.model,
                content=response.choices[0].message.content,
                usage=dict(response.usage),
                cost_usd=round(cost, 6),
                latency_ms=round(latency_ms, 2),
            )

            return result

        except Exception as e:
            raise HTTPException(status_code=502, detail=str(e))


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "models_configured": [m["model_name"] for m in MODEL_CONFIG["model_list"]],
    }

What does the chat endpoint do?

When a request arrives, the endpoint generates a unique request_id and starts a timer. It then acquires the semaphore (waiting if 10 requests are already in flight) and calls router.acompletion() to route the request through LiteLLM.

After the response arrives, it calculates the latency and uses litellm.completion_cost() to compute how much that call cost. The gateway returns all of this metadata alongside the model's actual response. If anything goes wrong, it raises an HTTP 502 error with the failure details.

  • Save main.py.

Seeing red squiggles in Cursor?

Make sure your virtual environment is selected as the Python interpreter. Press Ctrl+Shift+P, type Python: Select Interpreter, and pick the one inside your venv folder.

help me fix Python interpreter or import issues in Cursor

✔️ Awesome, I've got everything!

Great. Make sure you've saved all three files: config.py, audit_logger.py, and main.py.

ⓧ I'd like to double check the full code

Here is the complete config.py:

"""NeuralGate configuration: model routing and gateway settings."""

MODEL_CONFIG = {
    "model_list": [
        {
            "model_name": "fast",
            "litellm_params": {
                "model": "gpt-4o-mini",
            },
        },
        {
            "model_name": "quality",
            "litellm_params": {
                "model": "gpt-4o",
            },
        },
    ],
    "fallbacks": [{"quality": ["fast"]}],
    "routing_strategy": "cost-based-routing",
    "num_retries": 2,
}

GATEWAY_SETTINGS = {
    "max_concurrent_requests": 10,
    "budget_limit_usd": 1.00,
}

Here is the complete main.py:

"""NeuralGate: LLM API Gateway with smart routing and observability."""

import asyncio
import os
import time
import uuid
from contextlib import asynccontextmanager

import litellm
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from litellm import Router
from pydantic import BaseModel

from audit_logger import AuditLogger
from config import GATEWAY_SETTINGS, MODEL_CONFIG

# Load environment variables from .env file
load_dotenv()

# Concurrency control: limits simultaneous LLM requests
semaphore = asyncio.Semaphore(GATEWAY_SETTINGS["max_concurrent_requests"])

# LiteLLM Router instance (initialized during app startup)
router: Router | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Initialize the LiteLLM Router on startup, clean up on shutdown."""
    global router
    router = Router(
        model_list=MODEL_CONFIG["model_list"],
        fallbacks=MODEL_CONFIG["fallbacks"],
        routing_strategy=MODEL_CONFIG["routing_strategy"],
        num_retries=MODEL_CONFIG["num_retries"],
    )
    yield


app = FastAPI(
    title="NeuralGate",
    description="LLM API Gateway with cost-based routing and observability",
    lifespan=lifespan,
)


class Message(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    model: str = "fast"
    messages: list[Message]
    max_tokens: int = 1024
    temperature: float = 0.7


class ChatResponse(BaseModel):
    id: str
    model: str
    content: str
    usage: dict
    cost_usd: float
    latency_ms: float


@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest) -> ChatResponse:
    """Route a chat completion request through the gateway."""
    request_id = str(uuid.uuid4())
    start_time = time.time()

    async with semaphore:
        try:
            response = await router.acompletion(
                model=request.model,
                messages=[m.model_dump() for m in request.messages],
                max_tokens=request.max_tokens,
                temperature=request.temperature,
            )

            latency_ms = (time.time() - start_time) * 1000
            cost = litellm.completion_cost(completion_response=response)

            result = ChatResponse(
                id=request_id,
                model=response.model,
                content=response.choices[0].message.content,
                usage=dict(response.usage),
                cost_usd=round(cost, 6),
                latency_ms=round(latency_ms, 2),
            )

            return result

        except Exception as e:
            raise HTTPException(status_code=502, detail=str(e))


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "models_configured": [m["model_name"] for m in MODEL_CONFIG["model_list"]],
    }
Test the gateway

With your files saved, it's time to start the server and send a real request through your gateway.

  • Start the gateway server by running this command in your terminal (make sure your virtual environment is still active):
python -m uvicorn main:app --reload

What does this command do?

python -m uvicorn runs the Uvicorn ASGI server. main:app tells it to find the app object inside main.py. The --reload flag auto-restarts the server when you save file changes.

  • You should see output indicating the server is running at http://127.0.0.1:8000.
  • Open your browser and go to http://localhost:8000/docs to open the Swagger UI.
  • Click on the POST /v1/chat/completions endpoint to expand it.
  • Click Try it out.
  • Replace the request body with this JSON:
{
  "model": "fast",
  "messages": [
    {"role": "user", "content": "What is an API gateway in one sentence?"}
  ],
  "max_tokens": 100
}
  • Click Execute.
  • You should see a JSON response with model, content, cost_usd, and latency_ms fields populated.

What should I see?

The response includes the model's answer plus metadata. The model field shows which model actually handled the request (e.g. gpt-4o-mini). The cost_usd should be a tiny number (fractions of a cent). The latency_ms shows how long the round trip took in milliseconds.

Server won't start or you see an error?

If you see ModuleNotFoundError, make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt. Reactivate it with .\venv\Scripts\activate.

If you see an authentication error from OpenAI, double-check that your .env file has the correct OPENAI_API_KEY value and that you saved it.

help me debug my NeuralGate server startup error

Your gateway is alive. You can route LLM requests, get cost and latency data back, and the fallback chain is ready if a model goes down. Next up, you'll stress-test the concurrency control and verify that fallbacks actually kick in when things break.

Add Concurrency Control and Fallback Testing

Your gateway can route requests and return responses. But what happens when 50 requests hit it at the same time?

Without limits, every incoming request would fire off an outbound call to OpenAI simultaneously. That can exhaust connection pools and crash the process. In this step, you will understand the concurrency control already in your code, prove that fallback chains catch failures automatically, and load-test the gateway with simultaneous requests.

In this step, get ready to:
  • Understand how the asyncio semaphore protects downstream providers.
  • Test the fallback chain by simulating a model failure.
  • Fire 5 concurrent requests and verify they all complete.
Understand the concurrency control

Your main.py already wraps the LLM call inside a semaphore. This is the key line that protects the gateway from overload.

  • In Cursor, open main.py and find this block inside the chat_completions function:
    async with semaphore:
        try:
            response = await router.acompletion(
                model=request.model,
                messages=[m.model_dump() for m in request.messages],
                max_tokens=request.max_tokens,
                temperature=request.temperature,
            )

What does the semaphore do?

The line async with semaphore: acts like a bouncer at a club. Only 10 requests (the value from GATEWAY_SETTINGS["max_concurrent_requests"]) can be inside the async with block at the same time.

If an 11th request arrives while 10 are already in flight, it waits in a queue until one of the 10 finishes. The request is not rejected. It just pauses until a slot opens up. This keeps downstream LLM providers from being overwhelmed.

Why not just let all requests through?

LLM providers enforce their own rate limits. If your gateway blasts 100 simultaneous calls, the provider will reject most of them with HTTP 429 errors. The semaphore prevents this by smoothing out bursts on your side before they reach the provider.

Test the fallback chain

Your config.py defines a fallback rule: if the quality model group fails, route to fast instead. You will now prove this works by breaking the quality model on purpose.

  • First, send a request to the quality model to confirm it works normally. In the Swagger UI at http://localhost:8000/docs, click the POST /v1/chat/completions endpoint.
  • Click Try it out.
  • Replace the request body with this JSON:
{
  "model": "quality",
  "messages": [{"role": "user", "content": "Say hello in one word."}],
  "max_tokens": 50
}
  • Click Execute. You should see a response with "model": "gpt-4o" in the output.

Now break the quality model to trigger the fallback.

  • In config.py, find this line inside the quality model entry:
                "model": "gpt-4o",
  • Temporarily change it to an invalid model name:
                "model": "gpt-4o-BROKEN",
  • Save config.py. Because your server is running with --reload, it will restart automatically.
  • Return to the Swagger UI and send the same quality request again by clicking Execute.

This time, the response should show "model": "gpt-4o-mini" instead. The gateway tried the broken quality model, failed, and automatically fell back to fast.

What does this prove?

The fallbacks list in your config tells the LiteLLM Router: "if quality fails after 2 retries, try fast next." This means a single model going down never takes your entire gateway offline.

  • Restore the original model name in config.py by changing "gpt-4o-BROKEN" back to "gpt-4o".
  • Save config.py and wait for the server to reload.
  • Send one more quality request in Swagger UI to confirm "model": "gpt-4o" appears again.

Still seeing the fallback model?

Make sure you saved config.py after restoring the original model name. Check the terminal running uvicorn for a reload confirmation message.

If the server did not reload, stop it with Ctrl+C and start it again.

Help me debug why my fallback is still triggering after I restored the model name.

Test concurrent requests

Sending one request at a time does not prove the semaphore works. You need to fire multiple requests simultaneously and confirm the gateway handles them without errors.

  • Install the httpx library (an async HTTP client) by running this command in a new terminal:
pip install httpx

Why a new terminal?

Your first terminal is running the uvicorn server. You need a separate terminal session to run test scripts against it. Make sure your virtual environment is activated in the new terminal (.\venv\Scripts\activate).

  • Create a new file called test_concurrent.py in your neuralgate folder and paste this code:
"""Test script: fire 5 concurrent requests to verify concurrency control."""

import asyncio
import httpx


async def send_request(client: httpx.AsyncClient, prompt: str):
    response = await client.post(
        "http://localhost:8000/v1/chat/completions",
        json={
            "model": "fast",
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": 50,
        },
        timeout=30.0,
    )
    data = response.json()
    print(f"Model: {data['model']} | Cost: ${data['cost_usd']:.6f} | Latency: {data['latency_ms']:.0f}ms")
    return data


async def main():
    prompts = [
        "What is 2+2?",
        "Name a color.",
        "What day comes after Monday?",
        "Say hello in French.",
        "What is the capital of Japan?",
    ]

    async with httpx.AsyncClient() as client:
        tasks = [send_request(client, p) for p in prompts]
        results = await asyncio.gather(*tasks)

    total_cost = sum(r["cost_usd"] for r in results)
    print(f"\nTotal cost for 5 requests: ${total_cost:.6f}")


if __name__ == "__main__":
    asyncio.run(main())

What does this script do?

  • It creates 5 different prompts and fires all 5 as async HTTP requests at the same time using asyncio.gather.
  • Each request hits your gateway's /v1/chat/completions endpoint with the fast model.
  • After all 5 finish, it prints the model used, cost, and latency for each, plus the total cost.
  • Since you only sent 5 and the semaphore limit is 10, all 5 pass through immediately. If you sent 15, the first 10 would go through and the remaining 5 would wait until slots opened up.
  • Save test_concurrent.py.
  • Run the test script by executing this command in your second terminal:
python test_concurrent.py

You should see 5 lines of output, each showing the model, cost, and latency, followed by the total cost. All 5 requests should complete successfully.

Seeing connection errors?

Make sure your uvicorn server is still running in the first terminal. If it stopped, restart it:

python -m uvicorn main:app --reload

Also confirm the virtual environment is activated in both terminals. You should see (venv) at the start of your terminal prompt.

Help me fix connection errors when running test_concurrent.py.

\u2714\ufe0f Awesome, I've got everything!

Great. Double check you have saved both config.py (restored to the original gpt-4o) and test_concurrent.py.

\u24e7 I'd like to double check the full code

Here is test_concurrent.py in full:

"""Test script: fire 5 concurrent requests to verify concurrency control."""

import asyncio
import httpx


async def send_request(client: httpx.AsyncClient, prompt: str):
    response = await client.post(
        "http://localhost:8000/v1/chat/completions",
        json={
            "model": "fast",
            "messages": [{"role": "user", "content": prompt}],
            "max_tokens": 50,
        },
        timeout=30.0,
    )
    data = response.json()
    print(f"Model: {data['model']} | Cost: ${data['cost_usd']:.6f} | Latency: {data['latency_ms']:.0f}ms")
    return data


async def main():
    prompts = [
        "What is 2+2?",
        "Name a color.",
        "What day comes after Monday?",
        "Say hello in French.",
        "What is the capital of Japan?",
    ]

    async with httpx.AsyncClient() as client:
        tasks = [send_request(client, p) for p in prompts]
        results = await asyncio.gather(*tasks)

    total_cost = sum(r["cost_usd"] for r in results)
    print(f"\nTotal cost for 5 requests: ${total_cost:.6f}")


if __name__ == "__main__":
    asyncio.run(main())

And here is main.py (unchanged from the previous step):

"""NeuralGate: LLM API Gateway with smart routing and observability."""

import asyncio
import os
import time
import uuid
from contextlib import asynccontextmanager

import litellm
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from litellm import Router
from pydantic import BaseModel

from audit_logger import AuditLogger
from config import GATEWAY_SETTINGS, MODEL_CONFIG

# Load environment variables from .env file
load_dotenv()

# Concurrency control: limits simultaneous LLM requests
semaphore = asyncio.Semaphore(GATEWAY_SETTINGS["max_concurrent_requests"])

# LiteLLM Router instance (initialized during app startup)
router: Router | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Initialize the LiteLLM Router on startup, clean up on shutdown."""
    global router
    router = Router(
        model_list=MODEL_CONFIG["model_list"],
        fallbacks=MODEL_CONFIG["fallbacks"],
        routing_strategy=MODEL_CONFIG["routing_strategy"],
        num_retries=MODEL_CONFIG["num_retries"],
    )
    yield


app = FastAPI(
    title="NeuralGate",
    description="LLM API Gateway with cost-based routing and observability",
    lifespan=lifespan,
)


class Message(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    model: str = "fast"
    messages: list[Message]
    max_tokens: int = 1024
    temperature: float = 0.7


class ChatResponse(BaseModel):
    id: str
    model: str
    content: str
    usage: dict
    cost_usd: float
    latency_ms: float


@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest) -> ChatResponse:
    """Route a chat completion request through the gateway."""
    request_id = str(uuid.uuid4())
    start_time = time.time()

    async with semaphore:
        try:
            response = await router.acompletion(
                model=request.model,
                messages=[m.model_dump() for m in request.messages],
                max_tokens=request.max_tokens,
                temperature=request.temperature,
            )

            latency_ms = (time.time() - start_time) * 1000
            cost = litellm.completion_cost(completion_response=response)

            result = ChatResponse(
                id=request_id,
                model=response.model,
                content=response.choices[0].message.content,
                usage=dict(response.usage),
                cost_usd=round(cost, 6),
                latency_ms=round(latency_ms, 2),
            )

            return result

        except Exception as e:
            raise HTTPException(status_code=502, detail=str(e))


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "models_configured": [m["model_name"] for m in MODEL_CONFIG["model_list"]],
    }

Your gateway is now battle-tested. It handles load gracefully and recovers from model failures automatically. Next up, you will add a structured audit log that captures every single request for compliance and debugging.

Add Structured Audit Logging

Your gateway routes requests and handles failures, but right now there is no record of what happened. If something goes wrong in production, you need a trail that shows exactly which model was called, what it cost, and whether it succeeded or failed.

In this step, you will build a structured audit logger that captures every gateway interaction in JSONL format. Each line in the log file is a self-contained JSON object, making it easy for compliance tools and scripts to parse without complex regex.

In this step, get ready to:
  • Create an audit logger module that writes structured JSONL entries.
  • Integrate the logger into your gateway endpoint for both success and error paths.
  • Verify the audit trail by inspecting real log entries.
Create the audit logger module

The audit logger is a standalone module responsible for one thing: appending a JSON entry to a file every time the gateway processes a request. Each entry captures the full context of the interaction.

  • Create a new file called audit_logger.py in your neuralgate project folder.
  • Add the following code to audit_logger.py:
"""Structured JSONL audit logger for NeuralGate."""

import json
from datetime import datetime, timezone
from pathlib import Path


class AuditLogger:
    def __init__(self, log_path: str):
        self.log_path = Path(log_path)
        self.log_path.parent.mkdir(parents=True, exist_ok=True)

    def log_request(
        self,
        request_id: str,
        model_requested: str,
        model_used: str | None,
        messages: list,
        response_content: str | None,
        cost: float,
        latency_ms: float,
        status: str,
        error: str | None = None,
    ):
        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "request_id": request_id,
            "model_requested": model_requested,
            "model_used": model_used,
            "input_messages": [
                m.model_dump() if hasattr(m, "model_dump") else m
                for m in messages
            ],
            "output": response_content,
            "cost_usd": round(cost, 6),
            "latency_ms": round(latency_ms, 2),
            "status": status,
        }
        if error:
            entry["error"] = error

        with open(self.log_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry) + "\n")

What does this code do?

The AuditLogger class handles two responsibilities:

  • The __init__ method takes a file path and automatically creates any missing parent directories using Path.mkdir(parents=True, exist_ok=True). You never have to manually create a logs/ folder.
  • The log_request method builds a Python dictionary with all the request metadata, serializes it to JSON, and appends it as a single line to the .jsonl file. Each line is one complete JSON object. The error field is only included when present, keeping success entries clean.
  • Save audit_logger.py.

You will not see a visible change yet. The next chunk wires this logger into your gateway.

Seeing a syntax error?

  • The str | None type hint requires Python 3.10 or higher. You installed Python 3.12, so this should work. If you see an error, confirm your virtual environment is active and running the correct Python version.
  • Check for mismatched quotes or parentheses in the entry dictionary.

Help me debug a syntax error in my audit_logger.py file.

Integrate the logger into the gateway

Your main.py already imports AuditLogger from the module you just created. Now you need to instantiate it and call it from your endpoint.

  • In main.py, find this line:
semaphore = asyncio.Semaphore(GATEWAY_SETTINGS["max_concurrent_requests"])
  • Add these two lines directly below it:
# Audit logger: writes structured JSONL logs
audit = AuditLogger("logs/audit.jsonl")

What does this do?

This creates a single AuditLogger instance at module level that writes to logs/audit.jsonl. Because it is defined at module level, the same instance is shared across all requests. The logs/ directory is created automatically on first use.

You will not see a change yet. The next chunk adds the actual logging calls.

Next, you need to log every successful and failed request. The success-path logging goes between creating the result and returning it.

  • In the chat_completions function inside your try block, find this line:
            return result
  • Add this logging call directly above return result:
            # Log successful request
            audit.log_request(
                request_id=request_id,
                model_requested=request.model,
                model_used=response.model,
                messages=request.messages,
                response_content=result.content,
                cost=cost,
                latency_ms=latency_ms,
                status="success",
            )

What does this code do?

After a successful LLM completion, this captures everything about the interaction: the model the user requested, the model that actually served it (which may differ due to fallbacks), the full input and output, and the cost and timing data. The status field is set to "success" so you can filter log entries by outcome.

Now add error-path logging so failed requests are also captured.

  • In the same function, find the except Exception as e: block. It currently looks like this:
        except Exception as e:
            raise HTTPException(status_code=502, detail=str(e))
  • Replace that entire except block with the following, which logs the error before re-raising it:
        except Exception as e:
            latency_ms = (time.time() - start_time) * 1000
            audit.log_request(
                request_id=request_id,
                model_requested=request.model,
                model_used=None,
                messages=request.messages,
                response_content=None,
                cost=0,
                latency_ms=latency_ms,
                status="error",
                error=str(e),
            )
            raise HTTPException(status_code=502, detail=str(e))

What does this code do?

When a request fails, you still want a record of it. The error path calculates how long the request took before failing, sets model_used to None (no model served the request), and records the exception message in the error field.

This means your audit log captures both sides of the story. Compliance reviews can see not just what worked, but what failed and why.

  • Save main.py.
  • Your server should auto-reload (if running with --reload). Check your terminal for a successful reload message with no import errors.

Server crashing on reload?

  • Make sure audit_logger.py is saved in the same folder as main.py. The import from audit_logger import AuditLogger expects both files to be at the same level.
  • Check that the class name in audit_logger.py is exactly AuditLogger (capital A, capital L).
  • Confirm your indentation inside the try/except block is consistent. The audit.log_request(...) call should be at the same indent level as result = ChatResponse(...).

My server is crashing after I added the audit logger. Help me figure out why.

✔️ Awesome, I've got everything!

Great. Double check you have saved both audit_logger.py and main.py.

ⓧ I'd like to double check the full code

Here is the full audit_logger.py:

"""Structured JSONL audit logger for NeuralGate."""

import json
from datetime import datetime, timezone
from pathlib import Path


class AuditLogger:
    def __init__(self, log_path: str):
        self.log_path = Path(log_path)
        self.log_path.parent.mkdir(parents=True, exist_ok=True)

    def log_request(
        self,
        request_id: str,
        model_requested: str,
        model_used: str | None,
        messages: list,
        response_content: str | None,
        cost: float,
        latency_ms: float,
        status: str,
        error: str | None = None,
    ):
        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "request_id": request_id,
            "model_requested": model_requested,
            "model_used": model_used,
            "input_messages": [
                m.model_dump() if hasattr(m, "model_dump") else m
                for m in messages
            ],
            "output": response_content,
            "cost_usd": round(cost, 6),
            "latency_ms": round(latency_ms, 2),
            "status": status,
        }
        if error:
            entry["error"] = error

        with open(self.log_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry) + "\n")

Here is the full main.py:

"""NeuralGate: LLM API Gateway with smart routing and observability."""

import asyncio
import os
import time
import uuid
from contextlib import asynccontextmanager

import litellm
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from litellm import Router
from pydantic import BaseModel

from audit_logger import AuditLogger
from config import GATEWAY_SETTINGS, MODEL_CONFIG

# Load environment variables from .env file
load_dotenv()

# Concurrency control: limits simultaneous LLM requests
semaphore = asyncio.Semaphore(GATEWAY_SETTINGS["max_concurrent_requests"])

# Audit logger: writes structured JSONL logs
audit = AuditLogger("logs/audit.jsonl")

# LiteLLM Router instance (initialized during app startup)
router: Router | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Initialize the LiteLLM Router on startup, clean up on shutdown."""
    global router
    router = Router(
        model_list=MODEL_CONFIG["model_list"],
        fallbacks=MODEL_CONFIG["fallbacks"],
        routing_strategy=MODEL_CONFIG["routing_strategy"],
        num_retries=MODEL_CONFIG["num_retries"],
    )
    yield


app = FastAPI(
    title="NeuralGate",
    description="LLM API Gateway with cost-based routing and observability",
    lifespan=lifespan,
)


class Message(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    model: str = "fast"
    messages: list[Message]
    max_tokens: int = 1024
    temperature: float = 0.7


class ChatResponse(BaseModel):
    id: str
    model: str
    content: str
    usage: dict
    cost_usd: float
    latency_ms: float


@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest) -> ChatResponse:
    """Route a chat completion request through the gateway."""
    request_id = str(uuid.uuid4())
    start_time = time.time()

    async with semaphore:
        try:
            response = await router.acompletion(
                model=request.model,
                messages=[m.model_dump() for m in request.messages],
                max_tokens=request.max_tokens,
                temperature=request.temperature,
            )

            latency_ms = (time.time() - start_time) * 1000
            cost = litellm.completion_cost(completion_response=response)

            result = ChatResponse(
                id=request_id,
                model=response.model,
                content=response.choices[0].message.content,
                usage=dict(response.usage),
                cost_usd=round(cost, 6),
                latency_ms=round(latency_ms, 2),
            )

            audit.log_request(
                request_id=request_id,
                model_requested=request.model,
                model_used=response.model,
                messages=request.messages,
                response_content=result.content,
                cost=cost,
                latency_ms=latency_ms,
                status="success",
            )

            return result

        except Exception as e:
            latency_ms = (time.time() - start_time) * 1000
            audit.log_request(
                request_id=request_id,
                model_requested=request.model,
                model_used=None,
                messages=request.messages,
                response_content=None,
                cost=0,
                latency_ms=latency_ms,
                status="error",
                error=str(e),
            )
            raise HTTPException(status_code=502, detail=str(e))


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "models_configured": [m["model_name"] for m in MODEL_CONFIG["model_list"]],
    }
Verify the audit trail

Time to generate some real log entries and confirm the audit trail is working.

  • Open your browser and navigate to http://localhost:8000/docs to open the Swagger UI.
  • Click on the POST /v1/chat/completions endpoint, then click Try it out.
  • Send a request with model set to "fast" and a simple message like "What is 2+2?".
  • Send a second request with model set to "quality" and a different message.
  • Open the log file by running this command in your terminal:
type logs\audit.jsonl

You should see two JSON entries, one per line. Each entry contains request_id, model_requested, model_used, cost_usd, latency_ms, and a status of "success".

What should I see?

Each line looks something like this (compressed to one line in the actual file):

{"timestamp": "2026-06-23T10:15:30.123456+00:00", "request_id": "abc-123", "model_requested": "fast", "model_used": "gpt-4o-mini", "cost_usd": 0.000045, "latency_ms": 812.34, "status": "success"}

The model_requested shows what the caller asked for ("fast" or "quality"), while model_used shows the actual model that served the request (gpt-4o-mini or gpt-4o). This distinction is critical for auditing fallback behavior.

Don't see the log file?

  • Make sure you sent at least one successful request through the Swagger UI after saving both files. The log file is only created on the first write.
  • Confirm you are running the type command from the neuralgate folder. The logs/ directory is created relative to where the server started.
  • If you see an empty file, check your server terminal for errors. A missing import or typo in the audit.log_request(...) call would cause the request to fail without writing a log entry.

Help me figure out why my audit log file is not being created.

Your gateway now has a complete audit trail. Every request that flows through NeuralGate is permanently recorded with timing, cost, and routing decisions. Next up, you will connect Langfuse to get a live observability dashboard on top of this data.

Connect Langfuse for Observability

Your gateway now captures every interaction in a structured audit log. That's great for compliance, but you still can't spot patterns at a glance.

In this step, you'll connect Langfuse to your gateway so every request appears as a trace in a live dashboard. You'll be able to see cost spikes, latency outliers, and model usage patterns in real time.

In this step, get ready to:
  • Configure Langfuse tracing via LiteLLM callbacks.
  • Send varied requests to generate trace data.
  • Explore the Langfuse dashboard to visualize costs and latency.
Configure the Langfuse callback

LiteLLM has a built-in callback system that can send trace data to observability platforms. By registering Langfuse as a callback, every completion that flows through your Router automatically gets reported with its prompt, response, token count, latency, and cost.

  • In main.py, find the line that says # Concurrency control: limits simultaneous LLM requests.
  • Directly above that comment, add the following two callback lines with their comment:
# Configure Langfuse observability via LiteLLM callbacks
litellm.success_callback = ["langfuse"]
litellm.failure_callback = ["langfuse"]

What does this code do?

These two lines tell LiteLLM to send data to Langfuse after every completion call.

  • litellm.success_callback = ["langfuse"] sends trace data (prompt, response, tokens, cost, latency) whenever a request succeeds.
  • litellm.failure_callback = ["langfuse"] sends error details whenever a request fails.
  • LiteLLM automatically reads LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY from your environment variables. No additional configuration is needed.
  • Save main.py.
  • Because you launched the server with --reload, uvicorn detects the change and restarts automatically. Check your terminal for a line confirming the reload.

Server not reloading?

  • Make sure you saved the file. Unsaved changes won't trigger a reload.
  • If the server crashed, check the terminal for an import error. Confirm that litellm is spelled correctly and that your virtual environment is active.
  • help me debug why my server won't restart after adding the Langfuse callbacks

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and the server reloaded successfully.

ⓧ I'd like to double check the full code

"""NeuralGate: LLM API Gateway with smart routing and observability."""

import asyncio
import os
import time
import uuid
from contextlib import asynccontextmanager

import litellm
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from litellm import Router
from pydantic import BaseModel

from audit_logger import AuditLogger
from config import GATEWAY_SETTINGS, MODEL_CONFIG

# Load environment variables from .env file
load_dotenv()

# Configure Langfuse observability via LiteLLM callbacks
litellm.success_callback = ["langfuse"]
litellm.failure_callback = ["langfuse"]

# Concurrency control: limits simultaneous LLM requests
semaphore = asyncio.Semaphore(GATEWAY_SETTINGS["max_concurrent_requests"])

# Audit logger: writes structured JSONL logs
audit = AuditLogger("logs/audit.jsonl")

# LiteLLM Router instance (initialized during app startup)
router: Router | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Initialize the LiteLLM Router on startup, clean up on shutdown."""
    global router
    router = Router(
        model_list=MODEL_CONFIG["model_list"],
        fallbacks=MODEL_CONFIG["fallbacks"],
        routing_strategy=MODEL_CONFIG["routing_strategy"],
        num_retries=MODEL_CONFIG["num_retries"],
    )
    yield


app = FastAPI(
    title="NeuralGate",
    description="LLM API Gateway with cost-based routing and observability",
    lifespan=lifespan,
)


class Message(BaseModel):
    role: str
    content: str


class ChatRequest(BaseModel):
    model: str = "fast"
    messages: list[Message]
    max_tokens: int = 1024
    temperature: float = 0.7


class ChatResponse(BaseModel):
    id: str
    model: str
    content: str
    usage: dict
    cost_usd: float
    latency_ms: float


@app.post("/v1/chat/completions")
async def chat_completions(request: ChatRequest) -> ChatResponse:
    """Route a chat completion request through the gateway."""
    request_id = str(uuid.uuid4())
    start_time = time.time()

    async with semaphore:
        try:
            response = await router.acompletion(
                model=request.model,
                messages=[m.model_dump() for m in request.messages],
                max_tokens=request.max_tokens,
                temperature=request.temperature,
            )

            latency_ms = (time.time() - start_time) * 1000
            cost = litellm.completion_cost(completion_response=response)

            result = ChatResponse(
                id=request_id,
                model=response.model,
                content=response.choices[0].message.content,
                usage=dict(response.usage),
                cost_usd=round(cost, 6),
                latency_ms=round(latency_ms, 2),
            )

            audit.log_request(
                request_id=request_id,
                model_requested=request.model,
                model_used=response.model,
                messages=request.messages,
                response_content=result.content,
                cost=cost,
                latency_ms=latency_ms,
                status="success",
            )

            return result

        except Exception as e:
            latency_ms = (time.time() - start_time) * 1000
            audit.log_request(
                request_id=request_id,
                model_requested=request.model,
                model_used=None,
                messages=request.messages,
                response_content=None,
                cost=0,
                latency_ms=latency_ms,
                status="error",
                error=str(e),
            )
            raise HTTPException(status_code=502, detail=str(e))


@app.get("/health")
async def health():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "models_configured": [m["model_name"] for m in MODEL_CONFIG["model_list"]],
    }
Generate trace data

For Langfuse to show meaningful metrics, you need a batch of varied requests flowing through the gateway. You'll send prompts to both model groups so the dashboard captures routing decisions, cost differences, and latency patterns.

  • Open the Swagger UI in your browser at http://localhost:8000/docs.
  • Click the POST /v1/chat/completions endpoint, then click Try it out.
  • Send the following request with the fast model by pasting this into the request body and clicking Execute:
{
  "model": "fast",
  "messages": [{"role": "user", "content": "What is an API gateway?"}],
  "max_tokens": 100
}
  • Send at least 4 more requests with different prompts. Vary the model field between fast and quality to generate diverse trace data:
{
  "model": "quality",
  "messages": [{"role": "user", "content": "Explain observability in 2 sentences."}],
  "max_tokens": 100
}
{
  "model": "fast",
  "messages": [{"role": "user", "content": "What is the capital of France?"}],
  "max_tokens": 50
}
{
  "model": "quality",
  "messages": [{"role": "user", "content": "Write a haiku about servers."}],
  "max_tokens": 50
}
{
  "model": "fast",
  "messages": [{"role": "user", "content": "Name three programming languages."}],
  "max_tokens": 50
}
  • Confirm each request returns a successful response with cost_usd and latency_ms fields populated.
  • Wait 10-20 seconds after your last request. Langfuse ingests traces asynchronously, so there is a short delay before they appear in the dashboard.

Why wait 10-20 seconds?

LiteLLM batches callback data and sends it to Langfuse in the background. The traces won't appear in the Langfuse UI instantly. A short wait ensures all your requests have been ingested and are ready to view.

Requests succeeding but worried about Langfuse?

If your gateway returns responses normally but you're unsure whether Langfuse is receiving data, check your terminal. LiteLLM may log warnings if it cannot reach Langfuse (for example, if your LANGFUSE_PUBLIC_KEY or LANGFUSE_SECRET_KEY values are incorrect in .env).

help me figure out why my Langfuse traces are not appearing

Explore the Langfuse dashboard

Your traces are now flowing into Langfuse Cloud. Time to see what "AI observability" looks like in practice.

  • Open your Langfuse project in the browser at cloud.langfuse.com.
  • Click into your project (the one you created in Step 1).
  • Look at the tracing table. You should see a row for each request you sent. Each entry shows the model used, latency, token count, and cost.
  • Click on any individual trace to expand it. You'll see the full prompt you sent, the model's response, and detailed timing information.

What am I looking at?

Each row in the tracing table is one LLM call that flowed through your gateway. Langfuse captured it automatically because of the two callback lines you added.

Notice the cost column. Requests routed to gpt-4o-mini (the "fast" group) cost significantly less than those routed to gpt-4o (the "quality" group). This is the cost-based routing in action.

  • Navigate to the home dashboard of your project. This shows aggregate metrics across all your requests.
  • Look for the total cost, model usage breakdown, and latency patterns.

Why does this matter?

In production, you might route thousands of requests per hour across multiple models. Without observability, you'd have no idea which model is eating your budget or which requests are slow.

Langfuse gives you that visibility with zero extra code beyond those two callback lines. This is what "AI observability" means in practice: full visibility into every decision your system makes.

Secret mission

Budget Ceiling Enforcement

Add a cumulative cost tracker that rejects requests once a spending limit is reached. You will use thread-safe state management and custom HTTP error responses, building on the concurrency concepts from Step 3.

Clean Up Your Resources

Clean Up Your Resources

This project runs locally on your machine, so there are no ongoing compute costs. The only charges come from OpenAI API calls, which only happen when you actively send requests through the gateway.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • OpenAI API key (used a few cents of API credits).
  • Langfuse Cloud project (free Hobby tier, 50k units/month, data auto-deletes after 30 days).
  • Local neuralgate project folder (virtual environment, code files, logs).

Keep everything running

No action needed. Choose this if you are still actively building or plan to continue to a follow-up project.

  • Your OpenAI API key stays active. You will only be charged when you send new requests.
  • Your Langfuse Cloud project remains available. Trace data is retained for 30 days on the Hobby plan.
  • Your local neuralgate folder stays on your Desktop.

Pause - I'll come back to this later

Shut down the running server and deactivate your OpenAI key to prevent any accidental charges. You can reactivate later.

  • Stop the uvicorn server by pressing Ctrl+C in the terminal where it is running.
  • Go to platform.openai.com and navigate to the API Keys page.
  • Find the key you created for this project and revoke it. You can create a new one later if you return.
  • Langfuse requires no action. The free Hobby tier has no charges, and data auto-deletes after 30 days of inactivity.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

  • Stop the uvicorn server by pressing Ctrl+C in the terminal.
  • Revoke your OpenAI API key. Go to platform.openai.com, navigate to API Keys, find the key you created for this project, and delete it.
  • Delete your Langfuse project. Go to cloud.langfuse.com, open your project, navigate to Settings, scroll to the Danger Zone in the General section, and confirm deletion.
  • Delete the local project folder by running the command for your operating system:

macOS

cd ~/Desktop
rm -rf neuralgate

Windows

cd ~\Desktop
rmdir /s /q neuralgate

That's a Wrap!

That's a Wrap!

Nice work! You just built NeuralGate, a production-patterned LLM API gateway that routes requests between models, controls concurrency, logs every interaction for audit, and reports everything to a live observability dashboard.

You've learned how to:

  • Build an async FastAPI gateway that routes LLM requests between models based on cost-based routing, with automatic fallback chains that reroute failed requests to alternative models.
  • Implement bounded concurrency control using asyncio semaphores and create structured JSONL audit logs capturing every request, response, cost, and routing decision.
  • Connect Langfuse observability to visualize traces, costs, and latency in a live dashboard.
  • Secret Mission: Enforced a budget ceiling that tracks cumulative spend and rejects requests when exhausted.

Ready to quiz yourself?