Build an LLM Evaluation Pipeline

Score LLM outputs with exact match, embeddings, and Claude-as-judge.

Introduction

30 Second Summary

When you change a prompt, swap a model, or update your retrieval pipeline, how do you know if things got better or worse? Most teams rely on gut feel and manual spot-checking, which breaks down the moment you ship changes at any real pace.

In this project, you will build a production-style LLM evaluation pipeline that automatically scores model outputs using three methods, stores every result for historical comparison, and visualizes trends in an interactive dashboard.

What You'll Build

A multi-method evaluation system with a FastAPI backend, three scoring engines (exact match, embedding similarity, and LLM-as-judge), SQLite experiment tracking, and a Streamlit dashboard for exploring results.

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

A multi-method scoring engine that evaluates LLM outputs using deterministic matching, OpenAI embedding similarity, and Claude-as-judge scoring.
A FastAPI evaluation backend that triggers runs, orchestrates scoring, and serves results through a REST API.
An interactive Streamlit dashboard visualizing pass rates, score distributions, and run-over-run comparisons with historical tracking in SQLite.
Secret Mission: Add automated regression detection that flags when a new run's scores drop below a configurable threshold compared to the previous baseline.

Are there any prerequisites?

This project assumes you have experience writing Python and are familiar with calling LLM APIs. You will need an OpenAI API key and an Anthropic API key with credits available for embedding and chat completion calls.

You should also have Cursor installed as your code editor and Python 3.10 or higher on your machine.

Before We Start

Before you can build an evaluation pipeline, you need a working Python environment with all the right libraries installed.

In this step, you will set up your project directory, install every dependency your pipeline needs, and confirm that your API keys are working.

In this step, get ready to:

Create a project directory and Python virtual environment.
Install all Python dependencies for the pipeline.
Create the project file structure and verify your API keys.

Create your project directory and virtual environment

Every Python project should live in its own folder with its own virtual environment. A virtual environment isolates your project's packages so they do not conflict with other projects on your machine.

Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.

First, confirm that Python 3.10 or higher is installed on your machine.

macOS

Check your Python version by running:

python3 --version

Windows

Check your Python version by running:

python --version

✔️ I see version 3.10 or higher

You are good to go. Move on to the next instruction below.

ⓧ I see an older version

Your system has a Python version below 3.10. This project requires 3.10+ because of features used by FastAPI and Streamlit.

Visit python.org/downloads and download the latest Python installer for your operating system.
Run the installer and follow the prompts.
On Windows, check the box that says Add Python to PATH before clicking Install Now.
Close and reopen your terminal, then re-run the version check command.

macOS ships an older Python

macOS often ships with an outdated system Python. After installing a newer version, use python3.12 or python3.13 directly if python3 --version still shows the old one.

ⓧ Command not found

Python is not installed on your machine yet.

Visit python.org/downloads and download the latest Python installer.
Run the installer and follow the prompts.
On Windows, check the box that says Add Python to PATH before clicking Install Now.
Close and reopen your terminal, then run the version check command again to confirm it works.

Move to your Desktop by running this command:

cd ~/Desktop

Create your project folder and move into it by running these commands:

mkdir eval_pipeline
cd eval_pipeline

Create and activate a virtual environment by running:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
.\venv\Scripts\activate

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

What does (venv) mean?

The (venv) prefix means your virtual environment is active. Every pip install command you run from here will install packages into this isolated environment, not your system Python.

Install all the project dependencies by running:

pip install fastapi==0.139.0 uvicorn==0.49.0 openai==2.44.0 anthropic==0.116.0 streamlit==1.58.0 numpy pandas

What are these packages?

FastAPI is the web framework for your evaluation API. Uvicorn is the server that runs it.

OpenAI and Anthropic are the Python SDKs for calling the LLM APIs. Streamlit powers your dashboard. numpy and pandas handle math and data tables.

The install takes about 30-60 seconds depending on your internet connection.

Seeing a permissions error?

Make sure (venv) is showing in your terminal prompt. If it is not, re-run the activation command above.

help me fix a permissions error when running pip install in my virtual environment

Create the project file structure

Your pipeline is made up of several focused modules. Each file handles one responsibility.

Create all your project files by running:

macOS

touch main.py models.py scorers.py database.py runner.py dashboard.py dataset.json

Windows

New-Item main.py, models.py, scorers.py, database.py, runner.py, dashboard.py, dataset.json

Confirm the files exist by running:

ls

You should see main.py, models.py, scorers.py, database.py, runner.py, dashboard.py, and dataset.json in the output.

What does each file do?

models.py defines the data shapes (Pydantic models) for test cases, scores, and runs.
scorers.py holds the three scoring methods (exact match, similarity, LLM judge).
database.py manages SQLite storage for experiment history.
runner.py orchestrates a full evaluation run across all test cases.
main.py is the FastAPI app with API endpoints.
dashboard.py is the Streamlit dashboard for visualizing results.
dataset.json stores the benchmark test cases.

Create your .env file and verify API keys

Your pipeline calls both the OpenAI and Anthropic APIs. You will store the keys in a .env file so they stay out of your code.

Open the eval_pipeline folder in Cursor.
Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
Type Cursor and press Enter to open it.
In Cursor, click File in the top menu bar.
Click Open Folder.
Navigate to Desktop > eval_pipeline and click Open.

Create a new file by pressing Cmd+N (macOS) or Ctrl+N (Windows).
Save it immediately as .env by pressing Cmd+S (macOS) or Ctrl+S (Windows) and naming it .env.
Paste the following into your .env file, replacing the placeholder values with your actual keys:

OPENAI_API_KEY=your-openai-api-key-here
ANTHROPIC_API_KEY=your-anthropic-api-key-here

Save the file.

Why a .env file?

Storing API keys in a .env file keeps them separate from your code. This way, you never accidentally commit secrets to version control.

Now verify that both keys work. Switch back to your terminal (make sure (venv) is still active).

Load your keys into the terminal session and test the OpenAI connection by running:

macOS

export $(cat .env | xargs)
python3 -c "from openai import OpenAI; c = OpenAI(); print('OpenAI OK:', c.models.list().data[0].id)"

Windows

Get-Content .env | ForEach-Object { $k,$v = $_ -split '=',2; [Environment]::SetEnvironmentVariable($k,$v,'Process') }
python -c "from openai import OpenAI; c = OpenAI(); print('OpenAI OK:', c.models.list().data[0].id)"

You should see a line starting with OpenAI OK: followed by a model name.

Test the Anthropic connection by running:

macOS

python3 -c "import anthropic; c = anthropic.Anthropic(); r = c.messages.create(model='claude-sonnet-4-6', max_tokens=10, messages=[{'role':'user','content':'Hi'}]); print('Anthropic OK:', r.content[0].text)"

Windows

python -c "import anthropic; c = anthropic.Anthropic(); r = c.messages.create(model='claude-sonnet-4-6', max_tokens=10, messages=[{'role':'user','content':'Hi'}]); print('Anthropic OK:', r.content[0].text)"

You should see a line starting with Anthropic OK: followed by a short greeting from Claude.

Seeing an authentication error?

Double-check that you pasted the full key into .env with no extra spaces or line breaks. Make sure you re-ran the export command after saving changes.

help me debug an API key authentication error in my eval pipeline setup

✔️ Awesome, I've got everything!

Great. Double check you have saved your .env file.

ⓧ I'd like to double check the full code

OPENAI_API_KEY=your-openai-api-key-here
ANTHROPIC_API_KEY=your-anthropic-api-key-here

Your environment is ready. Next up, you will define the data models and benchmark dataset that your evaluation pipeline scores against.

Define the Evaluation Dataset and Data Models

Your environment is ready with all dependencies installed and both API keys verified.

Before you can score LLM outputs, you need a clear contract for what a test case looks like, how results are structured, and what configuration options exist. In this step, you will define Pydantic data models, build a benchmark dataset of customer support questions, and write a loader that validates the data on read.

In this step, get ready to:

Define Pydantic data models for test cases, scores, and evaluation runs.
Create a benchmark dataset of 10 customer support Q&A test cases.
Write a dataset loader function that validates and returns typed objects.

Define the Pydantic data models

Every piece of data flowing through the pipeline needs a strict shape. Pydantic models act as contracts that validate data at runtime, catching malformed inputs before they corrupt your results.

In Cursor, open models.py from the eval_pipeline folder.
Add the imports and the core evaluation types by pasting this code:

from pydantic import BaseModel
from typing import Optional
import uuid
from datetime import datetime


class TestCase(BaseModel):
    id: str
    input: str
    expected_output: str
    criteria: list[str] = []
    metadata: dict = {}


class ScoreResult(BaseModel):
    exact_match: float
    similarity: float
    judge_score: float
    judge_reasoning: str
    passed: bool


class EvalResult(BaseModel):
    test_case_id: str
    model_output: str
    scores: ScoreResult

What does this code do?

You are defining three data models that represent the core evaluation pipeline types.

TestCase defines what a single test looks like: an input question, the expected answer, keyword criteria for exact-match scoring, and optional metadata.
ScoreResult holds the output of all three scoring methods (exact match, embedding similarity, and LLM judge) plus a combined pass/fail verdict.
EvalResult ties a test case to the model's actual output and its scores.

You won't see a change yet. The next chunk adds the run configuration models that use these types.

Below the EvalResult class, add the run configuration and summary models:

class RunConfig(BaseModel):
    model: str = "gpt-4o-mini"
    prompt_template: str = "Answer the following customer support question helpfully and concisely:\n\n{input}"
    temperature: float = 0.0


class RunSummary(BaseModel):
    run_id: str
    timestamp: str
    config: RunConfig
    total_cases: int
    pass_rate: float
    mean_exact: float
    mean_similarity: float
    mean_judge: float
    results: list[EvalResult] = []


class RunRequest(BaseModel):
    config: RunConfig = RunConfig()

What does this code do?

RunConfig holds the settings for an evaluation run: which model to test, what prompt template to use, and the temperature. Defaults are set so you can trigger a run with zero configuration.
RunSummary captures everything about a completed run: its ID, timestamp, config, aggregate metrics (pass rate and mean scores), and the full list of individual results.
RunRequest is a thin wrapper used by the API endpoint to accept optional configuration from the caller.

Save models.py.
Verify the models load correctly by running this in your terminal:

python -c "from models import TestCase, RunSummary; print(list(TestCase.model_fields.keys()))"

You should see ['id', 'input', 'expected_output', 'criteria', 'metadata'] printed in the terminal.

Seeing a ModuleNotFoundError?

Make sure your terminal is inside the eval_pipeline directory where models.py lives.
Confirm your virtual environment is activated. You should see its name in parentheses at the start of your terminal prompt.

Help me debug this ModuleNotFoundError when importing my Pydantic models.

✔️ Awesome, I've got everything!

Great. Make sure you have saved your file before moving on.

ⓧ I'd like to double check the full code

from pydantic import BaseModel
from typing import Optional
import uuid
from datetime import datetime


class TestCase(BaseModel):
    id: str
    input: str
    expected_output: str
    criteria: list[str] = []
    metadata: dict = {}


class ScoreResult(BaseModel):
    exact_match: float
    similarity: float
    judge_score: float
    judge_reasoning: str
    passed: bool


class EvalResult(BaseModel):
    test_case_id: str
    model_output: str
    scores: ScoreResult


class RunConfig(BaseModel):
    model: str = "gpt-4o-mini"
    prompt_template: str = "Answer the following customer support question helpfully and concisely:\n\n{input}"
    temperature: float = 0.0


class RunSummary(BaseModel):
    run_id: str
    timestamp: str
    config: RunConfig
    total_cases: int
    pass_rate: float
    mean_exact: float
    mean_similarity: float
    mean_judge: float
    results: list[EvalResult] = []


class RunRequest(BaseModel):
    config: RunConfig = RunConfig()

Create the benchmark dataset

An evaluation pipeline is only as useful as the test cases it runs against. You will create a 10-case benchmark for a customer support Q&A task, where each case has an input question, an ideal answer, and keyword criteria for deterministic scoring.

Open dataset.json in Cursor.
Paste the following 10 test cases:

[
  {
    "id": "tc_001",
    "input": "What is your refund policy?",
    "expected_output": "We offer a full refund within 30 days of purchase. After 30 days, we can provide store credit. Please contact support with your order number to initiate a refund.",
    "criteria": ["refund", "30 days", "store credit"],
    "metadata": {"category": "refund"}
  },
  {
    "id": "tc_002",
    "input": "How do I reset my password?",
    "expected_output": "To reset your password, click 'Forgot Password' on the login page. You'll receive an email with a reset link. The link expires in 24 hours.",
    "criteria": ["forgot password", "email", "reset link"],
    "metadata": {"category": "account"}
  },
  {
    "id": "tc_003",
    "input": "Do you ship internationally?",
    "expected_output": "Yes, we ship to over 50 countries. International shipping takes 7-14 business days. Shipping costs vary by destination and are calculated at checkout.",
    "criteria": ["international", "7-14", "checkout"],
    "metadata": {"category": "shipping"}
  },
  {
    "id": "tc_004",
    "input": "Can I cancel my subscription?",
    "expected_output": "You can cancel your subscription at any time from your account settings. Go to Settings > Subscription > Cancel. You'll retain access until the end of your billing period.",
    "criteria": ["cancel", "account settings", "billing period"],
    "metadata": {"category": "subscription"}
  },
  {
    "id": "tc_005",
    "input": "What payment methods do you accept?",
    "expected_output": "We accept Visa, Mastercard, American Express, PayPal, and Apple Pay. All payments are processed securely through our encrypted payment system.",
    "criteria": ["visa", "paypal", "secure"],
    "metadata": {"category": "payment"}
  },
  {
    "id": "tc_006",
    "input": "How do I track my order?",
    "expected_output": "You can track your order using the tracking number sent to your email after shipping. Visit our tracking page or the carrier's website to see real-time updates.",
    "criteria": ["tracking number", "email", "real-time"],
    "metadata": {"category": "shipping"}
  },
  {
    "id": "tc_007",
    "input": "Is there a free trial available?",
    "expected_output": "Yes, we offer a 14-day free trial with full access to all features. No credit card required to start. You can upgrade to a paid plan at any time during or after the trial.",
    "criteria": ["14-day", "free trial", "no credit card"],
    "metadata": {"category": "subscription"}
  },
  {
    "id": "tc_008",
    "input": "How do I contact customer support?",
    "expected_output": "You can reach our support team via email at support@example.com, live chat on our website (available 9am-6pm EST), or by phone at 1-800-555-0123.",
    "criteria": ["email", "live chat", "phone"],
    "metadata": {"category": "support"}
  },
  {
    "id": "tc_009",
    "input": "What is your return process for damaged items?",
    "expected_output": "For damaged items, please take a photo of the damage and contact us within 48 hours of delivery. We'll send a prepaid return label and issue a full refund or replacement once we receive the item.",
    "criteria": ["photo", "48 hours", "prepaid return label"],
    "metadata": {"category": "refund"}
  },
  {
    "id": "tc_010",
    "input": "Can I change my shipping address after placing an order?",
    "expected_output": "You can change your shipping address within 2 hours of placing your order. Go to your order history and click 'Edit Address'. After 2 hours, please contact support as the order may have already been processed.",
    "criteria": ["2 hours", "order history", "contact support"],
    "metadata": {"category": "shipping"}
  }
]

What does this dataset cover?

Each test case has four fields that the scoring engine will use.

input is the customer question sent to the model under test.
expected_output is the ideal answer used for similarity comparison and as a reference for the LLM judge.
criteria is a list of keywords used by the exact-match scorer. If the model's response contains these terms, it scores higher.
metadata stores the category so you can later filter results by topic (refund, shipping, account, etc.).

Save dataset.json.
Verify the JSON is valid by running:

python -c "import json; data = json.load(open('dataset.json')); print(f'{len(data)} test cases loaded')"

You should see 10 test cases loaded in the terminal output.

Seeing a JSONDecodeError?

Check for trailing commas after the last object in the array. JSON does not allow trailing commas.
Make sure every string uses double quotes, not single quotes.

Help me fix my JSON syntax error in dataset.json.

Write the dataset loader function

Loading raw JSON is simple, but you want validation at read time. A typed loader function ensures that every test case conforms to the TestCase schema before it enters the pipeline.

Open runner.py in Cursor.
Add the imports and the loader function by pasting this code:

import json
from pathlib import Path

from models import TestCase


def load_dataset(path: str = "dataset.json") -> list[TestCase]:
    # Load JSON and validate each item against the TestCase schema
    with open(Path(path)) as f:
        data = json.load(f)
    return [TestCase(**item) for item in data]

What does this code do?

load_dataset reads the JSON file and unpacks each dictionary into a TestCase object using TestCase(**item).
If any test case has a missing field or wrong type, Pydantic raises a validation error immediately. This prevents bad data from silently flowing through the pipeline.
The default path points to dataset.json so you can call load_dataset() with no arguments during normal use.

Save runner.py.
Test the loader by running:

python -c "from runner import load_dataset; cases = load_dataset(); print(f'Loaded {len(cases)} cases, first: {cases[0].id}')"

You should see Loaded 10 cases, first: tc_001 confirming that all test cases pass validation.

Seeing a ValidationError?

A Pydantic ValidationError means a field in your JSON does not match the model. Check that every test case in dataset.json has id, input, and expected_output as strings.
Make sure criteria is an array of strings, not a single string.

Help me debug this Pydantic ValidationError when loading my dataset.

✔️ Awesome, I've got everything!

Great. Make sure both files are saved before moving on.

ⓧ I'd like to double check the full code

import json
from pathlib import Path

from models import TestCase


def load_dataset(path: str = "dataset.json") -> list[TestCase]:
    # Load JSON and validate each item against the TestCase schema
    with open(Path(path)) as f:
        data = json.load(f)
    return [TestCase(**item) for item in data]

Your data layer is locked in. You have typed models enforcing structure and a validated benchmark dataset ready to go. Next up, you will build the three scoring methods that evaluate how well model outputs match these expected answers.

Build the Multi-Method Scoring Engine

In Step 2, you defined the data models and benchmark dataset that your pipeline will evaluate against. Now it's time to build the scoring engine that actually judges how good an LLM's output is.

A single scoring method can miss important quality dimensions. Exact keyword matching catches obvious failures fast, but it can't tell if a response is semantically correct in different words. Embedding similarity handles that, but it can't judge tone or helpfulness. That's why production evaluation systems use multiple scoring methods in a funnel, from cheap and fast to expensive and nuanced.

In this step, get ready to:

Implement an exact match scorer that checks for required keywords in model outputs.
Build a similarity scorer using OpenAI embeddings and cosine similarity.
Create an LLM-as-judge scorer powered by Claude for nuanced quality assessment.

Set up the scorers module with API clients

All three scoring methods live in a single file. Two of them need external API clients: OpenAI for embeddings and Anthropic for the LLM judge.

Open scorers.py in your eval_pipeline folder and add the imports and client initialization:

import numpy as np
from openai import OpenAI
import anthropic
import json
import os

# Initialize API clients using environment variables
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
anthropic_client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

What does this code do?

The imports bring in numpy for vector math, the OpenAI and Anthropic SDKs for API calls, and json for parsing structured responses.

Both clients are initialized at module level using the environment variables from your .env file. This means they're ready to use the moment any function in this file is called.

Why two different AI providers?

The pipeline uses OpenAI's model as the system under test, and Claude as the judge. Using a different model family as the judge avoids self-preference bias, where a model rates its own family's outputs more favorably.

Implement the exact match scorer

The first scoring method is the cheapest and fastest. It checks whether required keywords from the test case's criteria list appear in the model's output. No API calls needed.

Add the exact_match_score function below the client initialization in scorers.py:

def exact_match_score(expected: str, actual: str, criteria: list[str]) -> float:
    # If no criteria provided, do a full string comparison
    if not criteria:
        return 1.0 if expected.lower().strip() == actual.lower().strip() else 0.0
    # Count how many criteria keywords appear in the actual output
    actual_lower = actual.lower()
    matches = sum(1 for c in criteria if c.lower() in actual_lower)
    return matches / len(criteria)

What does this code do?

When the test case has a criteria list (like ["refund", "30 days", "store credit"]), the scorer checks how many of those keywords appear in the model's output.

If 2 out of 3 criteria are found, the score is 0.67. If all 3 are found, the score is 1.0. This gives a proportional signal rather than a binary pass/fail.

Save scorers.py.

You won't see a visible change yet. The next two scorers will complete the file, and you'll test all three together at the end of this step.

Implement the similarity scorer

Keyword matching can't tell if a response is correct but uses different wording. Embedding similarity solves this by comparing the semantic meaning of two texts. Two sentences that mean the same thing will have embeddings that point in similar directions, even if they share zero words.

Add the similarity_score function below exact_match_score in scorers.py:

def similarity_score(expected: str, actual: str) -> float:
    # Embed both texts using OpenAI's embedding model
    response = openai_client.embeddings.create(
        input=[expected, actual],
        model="text-embedding-3-small"
    )
    # Extract the embedding vectors
    emb_expected = np.array(response.data[0].embedding)
    emb_actual = np.array(response.data[1].embedding)
    # Compute cosine similarity between the two vectors
    cosine_sim = float(np.dot(emb_expected, emb_actual) / (
        np.linalg.norm(emb_expected) * np.linalg.norm(emb_actual)
    ))
    return cosine_sim

What does this code do?

This function sends both the expected and actual outputs to OpenAI's text-embedding-3-small model in a single API call. The model returns a 1536-dimensional vector for each text.

Cosine similarity measures the angle between two vectors. A score of 1.0 means the texts are semantically identical. A score of 0.0 means they're completely unrelated. In practice, customer support answers that cover the same topic typically score 0.75 or higher.

Save scorers.py.

You won't see a change yet. The next chunk adds the final scorer and then you'll verify everything together.

Implement the LLM judge scorer

The most powerful scoring method uses an LLM as a judge. Instead of checking keywords or vector similarity, it reads the question, the expected answer, and the actual response, then provides a nuanced quality judgment with reasoning.

Add the llm_judge_score function below similarity_score in scorers.py:

def llm_judge_score(input_text: str, expected: str, actual: str) -> dict:
    # Build a rubric prompt that instructs Claude to score the response
    rubric_prompt = f"""You are an evaluation judge for a customer support AI. 
Score the following response on a scale of 1-5 and determine if it passes.

Scoring rubric:
- 5: Perfect response, covers all key points from the expected answer
- 4: Good response, covers most key points with minor omissions
- 3: Acceptable response, covers some key points but misses important ones
- 2: Poor response, misses most key points or is partially incorrect
- 1: Failing response, incorrect, unhelpful, or off-topic

A response PASSES if it scores 3 or above.

Customer Question: {input_text}

Expected Answer: {expected}

Actual Response: {actual}

Respond with ONLY valid JSON in this exact format:
{{"score": <1-5>, "passed": <true/false>, "reasoning": "<brief explanation>"}}"""

    # Call Claude with temperature 0 for reproducible judgments
    message = anthropic_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=256,
        temperature=0,
        messages=[{"role": "user", "content": rubric_prompt}]
    )
    response_text = message.content[0].text
    # Parse the JSON response and normalize score to 0.0-1.0 range
    try:
        result = json.loads(response_text)
        return {
            "score": result["score"] / 5.0,
            "passed": result["passed"],
            "reasoning": result["reasoning"]
        }
    except (json.JSONDecodeError, KeyError):
        return {"score": 0.0, "passed": False, "reasoning": "Failed to parse judge response"}

What does this code do?

This function builds a rubric prompt with the original question, expected answer, and actual model output. It sends this to claude-sonnet-4-6 and asks for a structured JSON judgment.

The raw score (1-5) is normalized to a 0.0-1.0 range by dividing by 5. Temperature 0 ensures the same inputs produce the same score every time, which is critical for evaluation reproducibility.

The try/except block handles cases where Claude returns malformed JSON. Instead of crashing the pipeline, it returns a failing score with a descriptive reason.

Save scorers.py.
Verify the module imports cleanly by running this in your terminal:

python -c "from scorers import exact_match_score, similarity_score, llm_judge_score; print('All scorers imported successfully')"

You should see All scorers imported successfully printed in your terminal.

Seeing an import error?

If you see ModuleNotFoundError: No module named 'numpy', make sure your virtual environment is activated. Run pip install numpy if needed.
If you see an error about OPENAI_API_KEY or ANTHROPIC_API_KEY, make sure your .env file is in the project directory and your environment variables are loaded.
Check that you're running the command from inside the eval_pipeline directory where scorers.py lives.

Help me debug my scorers.py import error

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here's the complete scorers.py file for comparison:

import numpy as np
from openai import OpenAI
import anthropic
import json
import os

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
anthropic_client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))


def exact_match_score(expected: str, actual: str, criteria: list[str]) -> float:
    if not criteria:
        return 1.0 if expected.lower().strip() == actual.lower().strip() else 0.0
    actual_lower = actual.lower()
    matches = sum(1 for c in criteria if c.lower() in actual_lower)
    return matches / len(criteria)


def similarity_score(expected: str, actual: str) -> float:
    response = openai_client.embeddings.create(
        input=[expected, actual],
        model="text-embedding-3-small"
    )
    emb_expected = np.array(response.data[0].embedding)
    emb_actual = np.array(response.data[1].embedding)
    cosine_sim = float(np.dot(emb_expected, emb_actual) / (
        np.linalg.norm(emb_expected) * np.linalg.norm(emb_actual)
    ))
    return cosine_sim


def llm_judge_score(input_text: str, expected: str, actual: str) -> dict:
    rubric_prompt = f"""You are an evaluation judge for a customer support AI. 
Score the following response on a scale of 1-5 and determine if it passes.

Scoring rubric:
- 5: Perfect response, covers all key points from the expected answer
- 4: Good response, covers most key points with minor omissions
- 3: Acceptable response, covers some key points but misses important ones
- 2: Poor response, misses most key points or is partially incorrect
- 1: Failing response, incorrect, unhelpful, or off-topic

A response PASSES if it scores 3 or above.

Customer Question: {input_text}

Expected Answer: {expected}

Actual Response: {actual}

Respond with ONLY valid JSON in this exact format:
{{"score": <1-5>, "passed": <true/false>, "reasoning": "<brief explanation>"}}"""

    message = anthropic_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=256,
        temperature=0,
        messages=[{"role": "user", "content": rubric_prompt}]
    )
    response_text = message.content[0].text
    try:
        result = json.loads(response_text)
        return {
            "score": result["score"] / 5.0,
            "passed": result["passed"],
            "reasoning": result["reasoning"]
        }
    except (json.JSONDecodeError, KeyError):
        return {"score": 0.0, "passed": False, "reasoning": "Failed to parse judge response"}

Your scoring engine is complete. You now have three methods that form an evaluation funnel: exact match catches obvious keyword misses for free, similarity measures semantic closeness cheaply, and the LLM judge handles nuanced quality assessment. Next up, you'll build the runner that orchestrates these scorers across your full dataset and stores everything in a database.

Build the Evaluation Runner and Storage Layer

Your scoring engine can evaluate outputs three different ways. Now you need somewhere to store those scores persistently, and an orchestrator to tie everything together.

In this step, you will build the storage layer in SQLite, create a runner that calls the model and scores each test case, and expose the entire pipeline through FastAPI endpoints.

In this step, get ready to:

Create the SQLite database schema and helper functions for storing runs and results.
Build the evaluation runner that orchestrates test execution and scoring.
Wire up FastAPI endpoints to trigger and retrieve evaluation runs.

Create the SQLite database schema

Every evaluation run produces scores, model outputs, and metadata. Without persistence, that data disappears the moment your script ends. SQLite gives you a zero-config database stored in a single file.

Open database.py in Cursor and add the connection helper and schema initialization:

import sqlite3
import json
from pathlib import Path

DB_PATH = Path("eval_pipeline.db")


def get_connection() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    conn = get_connection()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS eval_runs (
            run_id TEXT PRIMARY KEY,
            timestamp TEXT NOT NULL,
            model TEXT NOT NULL,
            prompt_template TEXT NOT NULL,
            temperature REAL NOT NULL,
            total_cases INTEGER NOT NULL,
            pass_rate REAL NOT NULL,
            mean_exact REAL NOT NULL,
            mean_similarity REAL NOT NULL,
            mean_judge REAL NOT NULL
        );

        CREATE TABLE IF NOT EXISTS eval_results (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            run_id TEXT NOT NULL,
            test_case_id TEXT NOT NULL,
            model_output TEXT NOT NULL,
            exact_score REAL NOT NULL,
            similarity_score REAL NOT NULL,
            judge_score REAL NOT NULL,
            judge_reasoning TEXT NOT NULL,
            passed INTEGER NOT NULL,
            FOREIGN KEY (run_id) REFERENCES eval_runs(run_id)
        );
    """)
    conn.commit()
    conn.close()

What does this code do?

The eval_runs table stores one row per evaluation run with aggregate stats (pass rate, mean scores). The eval_results table stores one row per test case per run, holding the model output and all three individual scores.

Setting row_factory = sqlite3.Row lets you access columns by name instead of index, making query results much easier to work with.

Below init_db(), add the write functions that persist run data and individual results:

def save_run(run_id: str, timestamp: str, model: str, prompt_template: str,
             temperature: float, total_cases: int, pass_rate: float,
             mean_exact: float, mean_similarity: float, mean_judge: float):
    conn = get_connection()
    conn.execute(
        """INSERT INTO eval_runs
           (run_id, timestamp, model, prompt_template, temperature,
            total_cases, pass_rate, mean_exact, mean_similarity, mean_judge)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (run_id, timestamp, model, prompt_template, temperature,
         total_cases, pass_rate, mean_exact, mean_similarity, mean_judge)
    )
    conn.commit()
    conn.close()


def save_result(run_id: str, test_case_id: str, model_output: str,
                exact_score: float, similarity_score: float,
                judge_score: float, judge_reasoning: str, passed: bool):
    conn = get_connection()
    conn.execute(
        """INSERT INTO eval_results
           (run_id, test_case_id, model_output, exact_score,
            similarity_score, judge_score, judge_reasoning, passed)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
        (run_id, test_case_id, model_output, exact_score,
         similarity_score, judge_score, judge_reasoning, int(passed))
    )
    conn.commit()
    conn.close()

What does this code do?

These two functions handle all writes to the database. save_run() inserts one row into eval_runs with summary statistics for the entire evaluation run.

The save_result() function inserts one row per test case, storing the model's actual output alongside each scorer's result. Notice int(passed) converts the boolean to an integer since SQLite does not have a native boolean type.

Below the write functions, add three query functions for reading data back:

def get_all_runs() -> list[dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM eval_runs ORDER BY timestamp DESC"
    ).fetchall()
    conn.close()
    return [dict(row) for row in rows]


def get_run_results(run_id: str) -> list[dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM eval_results WHERE run_id = ? ORDER BY test_case_id",
        (run_id,)
    ).fetchall()
    conn.close()
    return [dict(row) for row in rows]


def get_run(run_id: str) -> dict | None:
    conn = get_connection()
    row = conn.execute(
        "SELECT * FROM eval_runs WHERE run_id = ?", (run_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None

What does this code do?

These three functions cover the read side. get_all_runs() returns every run sorted by most recent first. get_run_results() fetches individual test case results for a specific run.

The get_run() function returns a single run's summary or None if it does not exist. The dashboard and API will use all three.

Save database.py.

Seeing an import error?

The sqlite3 module is part of Python's standard library. If you get an import error, confirm your virtual environment is active and you are running Python 3.10 or higher.

Help me debug a sqlite3 import error in my evaluation pipeline.

✔️ Awesome, I've got everything!

Great. Double check you have saved database.py before moving on.

ⓧ I'd like to double check the full code

import sqlite3
import json
from pathlib import Path

DB_PATH = Path("eval_pipeline.db")


def get_connection() -> sqlite3.Connection:
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn


def init_db():
    conn = get_connection()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS eval_runs (
            run_id TEXT PRIMARY KEY,
            timestamp TEXT NOT NULL,
            model TEXT NOT NULL,
            prompt_template TEXT NOT NULL,
            temperature REAL NOT NULL,
            total_cases INTEGER NOT NULL,
            pass_rate REAL NOT NULL,
            mean_exact REAL NOT NULL,
            mean_similarity REAL NOT NULL,
            mean_judge REAL NOT NULL
        );

        CREATE TABLE IF NOT EXISTS eval_results (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            run_id TEXT NOT NULL,
            test_case_id TEXT NOT NULL,
            model_output TEXT NOT NULL,
            exact_score REAL NOT NULL,
            similarity_score REAL NOT NULL,
            judge_score REAL NOT NULL,
            judge_reasoning TEXT NOT NULL,
            passed INTEGER NOT NULL,
            FOREIGN KEY (run_id) REFERENCES eval_runs(run_id)
        );
    """)
    conn.commit()
    conn.close()


def save_run(run_id: str, timestamp: str, model: str, prompt_template: str,
             temperature: float, total_cases: int, pass_rate: float,
             mean_exact: float, mean_similarity: float, mean_judge: float):
    conn = get_connection()
    conn.execute(
        """INSERT INTO eval_runs
           (run_id, timestamp, model, prompt_template, temperature,
            total_cases, pass_rate, mean_exact, mean_similarity, mean_judge)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (run_id, timestamp, model, prompt_template, temperature,
         total_cases, pass_rate, mean_exact, mean_similarity, mean_judge)
    )
    conn.commit()
    conn.close()


def save_result(run_id: str, test_case_id: str, model_output: str,
                exact_score: float, similarity_score: float,
                judge_score: float, judge_reasoning: str, passed: bool):
    conn = get_connection()
    conn.execute(
        """INSERT INTO eval_results
           (run_id, test_case_id, model_output, exact_score,
            similarity_score, judge_score, judge_reasoning, passed)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
        (run_id, test_case_id, model_output, exact_score,
         similarity_score, judge_score, judge_reasoning, int(passed))
    )
    conn.commit()
    conn.close()


def get_all_runs() -> list[dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM eval_runs ORDER BY timestamp DESC"
    ).fetchall()
    conn.close()
    return [dict(row) for row in rows]


def get_run_results(run_id: str) -> list[dict]:
    conn = get_connection()
    rows = conn.execute(
        "SELECT * FROM eval_results WHERE run_id = ? ORDER BY test_case_id",
        (run_id,)
    ).fetchall()
    conn.close()
    return [dict(row) for row in rows]


def get_run(run_id: str) -> dict | None:
    conn = get_connection()
    row = conn.execute(
        "SELECT * FROM eval_runs WHERE run_id = ?", (run_id,)
    ).fetchone()
    conn.close()
    return dict(row) if row else None

Build the evaluation runner

The runner is the orchestrator that ties your dataset, model, and scorers together. For each test case, it calls the model, runs all three scorers on the output, and persists the results.

Open runner.py and add the imports, client setup, and helper functions:

import json
import uuid
from datetime import datetime, timezone
from pathlib import Path
from openai import OpenAI
import os

from models import TestCase, RunConfig, ScoreResult, EvalResult, RunSummary
from scorers import exact_match_score, similarity_score, llm_judge_score
from database import save_run, save_result

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


def load_dataset(path: str = "dataset.json") -> list[TestCase]:
    with open(Path(path)) as f:
        data = json.load(f)
    return [TestCase(**item) for item in data]


def call_model(input_text: str, config: RunConfig) -> str:
    prompt = config.prompt_template.replace("{input}", input_text)
    response = openai_client.chat.completions.create(
        model=config.model,
        temperature=config.temperature,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content

What does this code do?

The load_dataset() function reads your dataset.json and validates each item against the TestCase Pydantic model. If any test case is malformed, Pydantic raises a clear error.

The call_model() function injects the test case input into the prompt template, calls OpenAI's chat completions API, and returns the raw text response. The prompt template and model are configurable per run, so you can compare different prompts head-to-head.

Below call_model(), add the main run_evaluation() function that orchestrates the full pipeline:

def run_evaluation(config: RunConfig, dataset_path: str = "dataset.json") -> RunSummary:
    run_id = str(uuid.uuid4())[:8]
    timestamp = datetime.now(timezone.utc).isoformat()
    dataset = load_dataset(dataset_path)

    results: list[EvalResult] = []

    for test_case in dataset:
        # Call the model under test
        model_output = call_model(test_case.input, config)

        # Run all three scorers
        exact = exact_match_score(test_case.expected_output, model_output, test_case.criteria)
        sim = similarity_score(test_case.expected_output, model_output)
        judge = llm_judge_score(test_case.input, test_case.expected_output, model_output)

        # A test case passes only if all three scorers agree
        passed = exact >= 0.5 and sim >= 0.8 and judge["passed"]

        scores = ScoreResult(
            exact_match=exact,
            similarity=sim,
            judge_score=judge["score"],
            judge_reasoning=judge["reasoning"],
            passed=passed
        )

        result = EvalResult(
            test_case_id=test_case.id,
            model_output=model_output,
            scores=scores
        )
        results.append(result)

        # Persist each result immediately
        save_result(
            run_id=run_id,
            test_case_id=test_case.id,
            model_output=model_output,
            exact_score=exact,
            similarity_score=sim,
            judge_score=judge["score"],
            judge_reasoning=judge["reasoning"],
            passed=passed
        )

    # Compute aggregate statistics
    total = len(results)
    pass_rate = sum(1 for r in results if r.scores.passed) / total
    mean_exact = sum(r.scores.exact_match for r in results) / total
    mean_similarity = sum(r.scores.similarity for r in results) / total
    mean_judge = sum(r.scores.judge_score for r in results) / total

    # Persist the run summary
    save_run(
        run_id=run_id,
        timestamp=timestamp,
        model=config.model,
        prompt_template=config.prompt_template,
        temperature=config.temperature,
        total_cases=total,
        pass_rate=pass_rate,
        mean_exact=mean_exact,
        mean_similarity=mean_similarity,
        mean_judge=mean_judge
    )

    return RunSummary(
        run_id=run_id,
        timestamp=timestamp,
        config=config,
        total_cases=total,
        pass_rate=pass_rate,
        mean_exact=mean_exact,
        mean_similarity=mean_similarity,
        mean_judge=mean_judge,
        results=results
    )

What does this code do?

This is the heart of the pipeline. For each test case, it calls the model, runs all three scorers, and stores the result in SQLite immediately. If the process crashes mid-run, you still have partial results.

The pass logic requires all three scorers to agree: exact match must be at least 0.5, similarity must reach 0.8, and the LLM judge must mark it as passing. This multi-gate approach prevents false positives from any single scorer.

After processing all test cases, the function computes aggregate stats (pass rate and mean scores) and persists the run summary before returning a typed RunSummary object.

Save runner.py.

Seeing a circular import?

Make sure runner.py imports from models, scorers, and database. If you accidentally import from runner inside one of those files, you will get a circular import error. The dependency flow should be one-directional: runner depends on the others, never the reverse.

Help me debug a circular import error in my evaluation pipeline.

✔️ Awesome, I've got everything!

Great. Double check you have saved runner.py before moving on.

ⓧ I'd like to double check the full code

import json
import uuid
from datetime import datetime, timezone
from pathlib import Path
from openai import OpenAI
import os

from models import TestCase, RunConfig, ScoreResult, EvalResult, RunSummary
from scorers import exact_match_score, similarity_score, llm_judge_score
from database import save_run, save_result

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


def load_dataset(path: str = "dataset.json") -> list[TestCase]:
    with open(Path(path)) as f:
        data = json.load(f)
    return [TestCase(**item) for item in data]


def call_model(input_text: str, config: RunConfig) -> str:
    prompt = config.prompt_template.replace("{input}", input_text)
    response = openai_client.chat.completions.create(
        model=config.model,
        temperature=config.temperature,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content


def run_evaluation(config: RunConfig, dataset_path: str = "dataset.json") -> RunSummary:
    run_id = str(uuid.uuid4())[:8]
    timestamp = datetime.now(timezone.utc).isoformat()
    dataset = load_dataset(dataset_path)

    results: list[EvalResult] = []

    for test_case in dataset:
        model_output = call_model(test_case.input, config)

        exact = exact_match_score(test_case.expected_output, model_output, test_case.criteria)
        sim = similarity_score(test_case.expected_output, model_output)
        judge = llm_judge_score(test_case.input, test_case.expected_output, model_output)

        passed = exact >= 0.5 and sim >= 0.8 and judge["passed"]

        scores = ScoreResult(
            exact_match=exact,
            similarity=sim,
            judge_score=judge["score"],
            judge_reasoning=judge["reasoning"],
            passed=passed
        )

        result = EvalResult(
            test_case_id=test_case.id,
            model_output=model_output,
            scores=scores
        )
        results.append(result)

        save_result(
            run_id=run_id,
            test_case_id=test_case.id,
            model_output=model_output,
            exact_score=exact,
            similarity_score=sim,
            judge_score=judge["score"],
            judge_reasoning=judge["reasoning"],
            passed=passed
        )

    total = len(results)
    pass_rate = sum(1 for r in results if r.scores.passed) / total
    mean_exact = sum(r.scores.exact_match for r in results) / total
    mean_similarity = sum(r.scores.similarity for r in results) / total
    mean_judge = sum(r.scores.judge_score for r in results) / total

    save_run(
        run_id=run_id,
        timestamp=timestamp,
        model=config.model,
        prompt_template=config.prompt_template,
        temperature=config.temperature,
        total_cases=total,
        pass_rate=pass_rate,
        mean_exact=mean_exact,
        mean_similarity=mean_similarity,
        mean_judge=mean_judge
    )

    return RunSummary(
        run_id=run_id,
        timestamp=timestamp,
        config=config,
        total_cases=total,
        pass_rate=pass_rate,
        mean_exact=mean_exact,
        mean_similarity=mean_similarity,
        mean_judge=mean_judge,
        results=results
    )

Create the FastAPI endpoints

The runner works, but you need a way to trigger evaluations and retrieve results without editing code every time. FastAPI gives you a clean HTTP interface that any tool or script can call.

Open main.py and add the full API server:

from fastapi import FastAPI, HTTPException
from models import RunRequest, RunSummary
from database import init_db, get_all_runs, get_run, get_run_results
from runner import run_evaluation

app = FastAPI(title="LLM Eval Pipeline", version="1.0.0")


@app.on_event("startup")
def startup():
    # Create database tables on first launch
    init_db()


@app.post("/runs", response_model=RunSummary)
def create_run(request: RunRequest):
    # Trigger a full evaluation run with the given config
    summary = run_evaluation(request.config)
    return summary


@app.get("/runs")
def list_runs():
    # Return all runs sorted by most recent first
    return get_all_runs()


@app.get("/runs/{run_id}")
def get_run_detail(run_id: str):
    # Return a specific run with all individual results
    run = get_run(run_id)
    if not run:
        raise HTTPException(status_code=404, detail="Run not found")
    results = get_run_results(run_id)
    return {"run": run, "results": results}

What does this code do?

The @app.on_event("startup") decorator ensures the database tables exist before any request is handled.

You have three endpoints. POST /runs triggers a new evaluation run. GET /runs lists all historical runs. GET /runs/{run_id} returns a single run with full per-test-case results for debugging.

Save main.py.
Start the server by running this command:

uvicorn main:app --reload

You should see output indicating the server is running at http://127.0.0.1:8000.
Open your browser and navigate to http://127.0.0.1:8000/docs to see the auto-generated Swagger UI with all three endpoints listed.

Server fails to start?

Make sure you are running the command from inside your eval_pipeline project directory where main.py lives. Also confirm your virtual environment is active and that your .env file has both API keys set.

If you see a ModuleNotFoundError, double check that you installed all dependencies from Step 1 in the same virtual environment you are running from now.

Help me debug my FastAPI server that fails to start.

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and that the Swagger UI at /docs shows all three endpoints.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, HTTPException
from models import RunRequest, RunSummary
from database import init_db, get_all_runs, get_run, get_run_results
from runner import run_evaluation

app = FastAPI(title="LLM Eval Pipeline", version="1.0.0")


@app.on_event("startup")
def startup():
    init_db()


@app.post("/runs", response_model=RunSummary)
def create_run(request: RunRequest):
    summary = run_evaluation(request.config)
    return summary


@app.get("/runs")
def list_runs():
    return get_all_runs()


@app.get("/runs/{run_id}")
def get_run_detail(run_id: str):
    run = get_run(run_id)
    if not run:
        raise HTTPException(status_code=404, detail="Run not found")
    results = get_run_results(run_id)
    return {"run": run, "results": results}

Your evaluation pipeline is fully wired. You can trigger runs, score outputs with three methods, and persist everything in SQLite. Next up, you will build a Streamlit dashboard to visualize these results and compare runs side by side.

Build the Streamlit Dashboard

You have a fully functional evaluation backend. Your FastAPI server can trigger runs, score outputs with three methods, and store everything in SQLite.

Raw database rows are hard to interpret at a glance. In this step, you will build a Streamlit dashboard that presents your evaluation data as filterable tables, trend charts, and side-by-side run comparisons.

In this step, get ready to:

Build a run overview page with a model filter and summary table.
Build a run detail view with expandable per-test-case results.
Build comparison visualizations showing score trends and run-vs-run metrics.

Build the run overview page

The overview page is the dashboard's landing view. It loads all evaluation runs from the database, displays them in a table, and lets you filter by model.

Open dashboard.py in Cursor and add the following imports and initialization code:

import streamlit as st
import pandas as pd
from database import get_all_runs, get_run_results, init_db

# Initialize the database on app load
init_db()

# Configure the page layout
st.set_page_config(page_title="LLM Eval Dashboard", layout="wide")
st.title("LLM Evaluation Dashboard")

# Load all runs from the database
runs = get_all_runs()

# Show a message and stop if no runs exist yet
if not runs:
    st.info("No evaluation runs yet. Trigger a run via the API: POST /runs")
    st.stop()

# Convert to a DataFrame for easy manipulation
runs_df = pd.DataFrame(runs)
runs_df["timestamp"] = pd.to_datetime(runs_df["timestamp"])

What does this code do?

The script imports streamlit and pandas, then pulls your helper functions from database.py.

Calling init_db() ensures the tables exist when the dashboard launches independently from the FastAPI server. The st.stop() call halts the script early if no runs are found, avoiding errors in downstream code that expects data.

Below the initialization code, add the sidebar filter and run overview table:

# Sidebar filter for model selection
st.sidebar.header("Filters")
models = runs_df["model"].unique().tolist()
selected_model = st.sidebar.selectbox("Model", ["All"] + models)

if selected_model != "All":
    runs_df = runs_df[runs_df["model"] == selected_model]

# Display the run overview table
st.header("Run Overview")
display_df = runs_df[["run_id", "timestamp", "model", "total_cases",
                       "pass_rate", "mean_exact", "mean_similarity", "mean_judge"]].copy()
display_df["pass_rate"] = (display_df["pass_rate"] * 100).round(1).astype(str) + "%"
display_df["mean_exact"] = display_df["mean_exact"].round(3)
display_df["mean_similarity"] = display_df["mean_similarity"].round(3)
display_df["mean_judge"] = display_df["mean_judge"].round(3)
st.dataframe(display_df, use_container_width=True)

What does this code do?

The sidebar selectbox lets you filter the table by model name. If you run evaluations against multiple models, this isolates each one.

The display_df formatting converts the raw pass_rate float (like 0.7) into a human-readable percentage (like "70.0%"). The other score columns are rounded to three decimal places for readability.

Save dashboard.py.
Launch the dashboard by running this command in a new terminal:

streamlit run dashboard.py

Your browser opens to http://localhost:8501. You should see the "LLM Evaluation Dashboard" title and a table showing your evaluation runs with pass rates and mean scores.

Don't see any runs in the table?

Make sure you triggered at least one evaluation run via your FastAPI server before launching the dashboard. The dashboard reads from the same eval_pipeline.db file.

Check that you are running streamlit run dashboard.py from inside your eval_pipeline project folder so the database file is found.

Help me debug my Streamlit dashboard not showing data.

Build the run detail view

The overview table shows aggregate scores, but you need to drill into individual test cases to understand why a run passed or failed. The detail view lets you select a run and expand each test case to see the model output, judge reasoning, and all three scores.

In dashboard.py, add the run detail section below the overview table code:

# Run detail view with expandable per-test-case results
st.header("Run Details")
selected_run_id = st.selectbox("Select a run", runs_df["run_id"].tolist())

if selected_run_id:
    results = get_run_results(selected_run_id)
    if results:
        results_df = pd.DataFrame(results)
        results_df["passed"] = results_df["passed"].map({1: "PASS", 0: "FAIL"})

        for _, row in results_df.iterrows():
            status_color = "green" if row["passed"] == "PASS" else "red"
            with st.expander(f":{status_color}[{row['passed']}] {row['test_case_id']} - Exact: {row['exact_score']:.2f} | Sim: {row['similarity_score']:.2f} | Judge: {row['judge_score']:.2f}"):
                st.write(f"**Model Output:** {row['model_output']}")
                st.write(f"**Judge Reasoning:** {row['judge_reasoning']}")
                cols = st.columns(3)
                cols[0].metric("Exact Match", f"{row['exact_score']:.2f}")
                cols[1].metric("Similarity", f"{row['similarity_score']:.2f}")
                cols[2].metric("Judge Score", f"{row['judge_score']:.2f}")

What does this code do?

The selectbox at the top lets you pick any run by its ID. For the selected run, the code loads every individual test-case result from the database.

Each test case renders as a collapsible expander. The label shows the pass/fail status in color (green or red), the test case ID, and all three scores at a glance. Expanding a row reveals the full model output, judge reasoning, and score metrics side by side.

Save dashboard.py and return to your browser.
Streamlit auto-reloads the page. Scroll down to the "Run Details" section and select a run from the dropdown. You should see collapsible rows for each test case, color-coded green (PASS) or red (FAIL).

Seeing an error about missing columns?

Double check that the column names in your code match the database schema exactly: exact_score, similarity_score, judge_score, judge_reasoning, and passed.

Help me fix column errors in my Streamlit dashboard.

Build the comparison view

Individual run details help you debug specific failures. Comparison views help you answer the bigger question: are things getting better or worse over time?

In dashboard.py, find the line st.dataframe(display_df, use_container_width=True) at the end of the overview section. Add the following score trends section directly below it, above the st.header("Run Details") line:

# Line chart showing score trends across runs
st.header("Score Trends Over Time")
if len(runs_df) > 1:
    chart_df = runs_df.sort_values("timestamp")[["timestamp", "pass_rate", "mean_exact", "mean_similarity", "mean_judge"]]
    chart_df = chart_df.set_index("timestamp")
    st.line_chart(chart_df)
else:
    st.info("Run at least 2 evaluations to see trends.")

What does this code do?

This section sorts runs by timestamp and plots all four metrics (pass rate, exact match, similarity, judge score) on a single line chart. You can visually spot if scores are trending up or down across runs.

The if len(runs_df) > 1 guard prevents Streamlit from rendering a meaningless chart with a single data point.

Now scroll to the very bottom of dashboard.py (below the Run Details section) and add the run comparison bar chart:

# Side-by-side comparison of any two runs
st.header("Compare Runs")
if len(runs_df) >= 2:
    col1, col2 = st.columns(2)
    run_ids = runs_df["run_id"].tolist()
    with col1:
        run_a = st.selectbox("Run A", run_ids, index=0)
    with col2:
        run_b = st.selectbox("Run B", run_ids, index=min(1, len(run_ids) - 1))

    if run_a != run_b:
        results_a = pd.DataFrame(get_run_results(run_a))
        results_b = pd.DataFrame(get_run_results(run_b))

        compare_df = pd.DataFrame({
            "Metric": ["Pass Rate", "Mean Exact", "Mean Similarity", "Mean Judge"],
            f"Run {run_a}": [
                results_a["passed"].mean(),
                results_a["exact_score"].mean(),
                results_a["similarity_score"].mean(),
                results_a["judge_score"].mean()
            ],
            f"Run {run_b}": [
                results_b["passed"].mean(),
                results_b["exact_score"].mean(),
                results_b["similarity_score"].mean(),
                results_b["judge_score"].mean()
            ]
        })
        st.bar_chart(compare_df.set_index("Metric"))

What does this code do?

Two side-by-side dropdowns let you pick any two runs. The code loads per-test-case results for each, computes aggregate metrics, and renders them as a grouped bar chart.

This makes it easy to see at a glance whether a prompt change or model swap improved or degraded performance across all dimensions.

Save dashboard.py and return to your browser.
You should see the "Score Trends Over Time" line chart between the overview table and run details. If you have two or more runs, the "Compare Runs" section appears at the bottom with a bar chart.

Trigger a second run to see comparisons

If you only have one evaluation run, the trends chart and comparison section will show info messages instead of visualizations. Trigger another run with a different prompt template via your FastAPI endpoint to see the charts populate.

Charts not rendering?

Make sure you have at least 2 evaluation runs stored in the database. The trends chart and comparison section both guard against single-run datasets.

If you see an import error, confirm that pandas is installed in your virtual environment. It was included in the pip install from Step 1.

Help me debug my Streamlit charts not showing.

✔️ Awesome, I've got everything!

Great. Double check you have saved dashboard.py and the dashboard is rendering all four sections: Run Overview, Score Trends, Run Details, and Compare Runs.

ⓧ I'd like to double check the full code

Here is the complete dashboard.py file:

import streamlit as st
import pandas as pd
from database import get_all_runs, get_run_results, init_db

init_db()

st.set_page_config(page_title="LLM Eval Dashboard", layout="wide")
st.title("LLM Evaluation Dashboard")

runs = get_all_runs()

if not runs:
    st.info("No evaluation runs yet. Trigger a run via the API: POST /runs")
    st.stop()

runs_df = pd.DataFrame(runs)
runs_df["timestamp"] = pd.to_datetime(runs_df["timestamp"])

st.sidebar.header("Filters")
models = runs_df["model"].unique().tolist()
selected_model = st.sidebar.selectbox("Model", ["All"] + models)

if selected_model != "All":
    runs_df = runs_df[runs_df["model"] == selected_model]

st.header("Run Overview")
display_df = runs_df[["run_id", "timestamp", "model", "total_cases",
                       "pass_rate", "mean_exact", "mean_similarity", "mean_judge"]].copy()
display_df["pass_rate"] = (display_df["pass_rate"] * 100).round(1).astype(str) + "%"
display_df["mean_exact"] = display_df["mean_exact"].round(3)
display_df["mean_similarity"] = display_df["mean_similarity"].round(3)
display_df["mean_judge"] = display_df["mean_judge"].round(3)
st.dataframe(display_df, use_container_width=True)

st.header("Score Trends Over Time")
if len(runs_df) > 1:
    chart_df = runs_df.sort_values("timestamp")[["timestamp", "pass_rate", "mean_exact", "mean_similarity", "mean_judge"]]
    chart_df = chart_df.set_index("timestamp")
    st.line_chart(chart_df)
else:
    st.info("Run at least 2 evaluations to see trends.")

st.header("Run Details")
selected_run_id = st.selectbox("Select a run", runs_df["run_id"].tolist())

if selected_run_id:
    results = get_run_results(selected_run_id)
    if results:
        results_df = pd.DataFrame(results)
        results_df["passed"] = results_df["passed"].map({1: "PASS", 0: "FAIL"})

        for _, row in results_df.iterrows():
            status_color = "green" if row["passed"] == "PASS" else "red"
            with st.expander(f":{status_color}[{row['passed']}] {row['test_case_id']} - Exact: {row['exact_score']:.2f} | Sim: {row['similarity_score']:.2f} | Judge: {row['judge_score']:.2f}"):
                st.write(f"**Model Output:** {row['model_output']}")
                st.write(f"**Judge Reasoning:** {row['judge_reasoning']}")
                cols = st.columns(3)
                cols[0].metric("Exact Match", f"{row['exact_score']:.2f}")
                cols[1].metric("Similarity", f"{row['similarity_score']:.2f}")
                cols[2].metric("Judge Score", f"{row['judge_score']:.2f}")

st.header("Compare Runs")
if len(runs_df) >= 2:
    col1, col2 = st.columns(2)
    run_ids = runs_df["run_id"].tolist()
    with col1:
        run_a = st.selectbox("Run A", run_ids, index=0)
    with col2:
        run_b = st.selectbox("Run B", run_ids, index=min(1, len(run_ids) - 1))

    if run_a != run_b:
        results_a = pd.DataFrame(get_run_results(run_a))
        results_b = pd.DataFrame(get_run_results(run_b))

        compare_df = pd.DataFrame({
            "Metric": ["Pass Rate", "Mean Exact", "Mean Similarity", "Mean Judge"],
            f"Run {run_a}": [
                results_a["passed"].mean(),
                results_a["exact_score"].mean(),
                results_a["similarity_score"].mean(),
                results_a["judge_score"].mean()
            ],
            f"Run {run_b}": [
                results_b["passed"].mean(),
                results_b["exact_score"].mean(),
                results_b["similarity_score"].mean(),
                results_b["judge_score"].mean()
            ]
        })
        st.bar_chart(compare_df.set_index("Metric"))

Secret mission

Add Automated Regression Detection

Your evaluation pipeline scores runs, but how do you know when things get worse? Right now you'd have to manually compare run summaries to spot quality drops. Build an automated regression detector that flags when a new run's scores fall below the previous baseline, so you catch degradation the moment it happens.

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs at rest. API charges only occur when you actively trigger evaluation runs.

Resources you used:

eval_pipeline project directory on your Desktop (all Python files, dataset.json, and .env).
eval_pipeline.db SQLite database file (created inside the project directory at runtime).
Python virtual environment (inside or alongside the project directory).

Keep everything running

No action needed. Choose this if you plan to keep experimenting with different models, prompt templates, or evaluation datasets.

Your project files stay on your Desktop and cost nothing at rest.
The SQLite database retains all your historical evaluation runs for future comparison.
Each future evaluation run costs approximately $0.10-0.20 in API calls (OpenAI embeddings + chat completions, Anthropic Claude judge calls across 10 test cases).

Pause - I'll come back to this later

Stop any running processes to free up system resources. Your files and database remain intact for when you return.

If the FastAPI server is running, press Ctrl+C in the terminal to stop it.
If the Streamlit dashboard is running, press Ctrl+C in its terminal to stop it.
Deactivate your virtual environment by running:

Delete - I don't want to use this again

Remove all project files, the database, and the virtual environment. There are no cloud resources to clean up.

If the FastAPI server or Streamlit dashboard is running, press Ctrl+C in each terminal to stop them.
Delete the entire project directory by running the command for your operating system below.

macOS

Remove the project folder by running:

cd ~/Desktop
rm -rf eval_pipeline

Windows

Remove the project folder by running:

cd ~\Desktop
Remove-Item -Recurse -Force eval_pipeline

Your OpenAI and Anthropic API keys remain active on their respective platforms. If you no longer need them, you can revoke them from your OpenAI and Anthropic account dashboards.

Nice Work!

Nice work! You just built a production-style LLM evaluation pipeline that scores model outputs with multiple methods, tracks experiments over time, and visualizes results in an interactive dashboard.

You've learned how to:

Build a multi-method scoring engine that combines exact match, embedding similarity, and LLM-as-judge evaluation to assess LLM outputs from multiple angles.
Create a FastAPI backend with SQLite storage that orchestrates evaluation runs, persists every score, and exposes results through a REST API.
Build an interactive Streamlit dashboard for visualizing pass rates, score distributions, and comparing runs side-by-side.
Secret Mission: Added automated regression detection that flags quality drops across runs and surfaces regressed test cases in the dashboard.

Ready to quiz yourself?