Build an Observable AI Pipeline

Trace, evaluate, and experiment on a multi-step LLM pipeline with Langfuse.

Introduction

30 Second Summary

Every team building AI features eventually hits the same problem: something breaks and nobody can explain why. Without visibility into each step, fixing AI feels like guessing in the dark.

In this project, you will build a multi-step AI pipeline that automatically traces every call, scores response quality, and compares prompt changes through structured experiments.

What You'll Build

A production-style question-answering pipeline with two stages (classification and generation), fully instrumented with Langfuse tracing, automated evaluation, and dataset-driven experiments.

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

  • A multi-step AI pipeline traced end-to-end in Langfuse, showing every call's inputs, outputs, latency, and cost.
  • Custom evaluator functions that automatically score response quality using LLM-as-a-Judge.
  • A dataset-driven experiment workflow that compares two prompt variants side-by-side with quantitative scores visible in the Langfuse dashboard.
  • Secret Mission: A hallucination guardrail that detects and flags ungrounded responses before they reach the user.

Are there any prerequisites?

You'll need basic Python knowledge and comfort with pip and environment variables. No prior experience with observability tools or LLM evaluation is required.

Set Up Your Environment

Before you can trace LLM calls or run evaluations, you need the right tools wired together.

In this step, you'll install the Python packages that power your pipeline, create free accounts on Langfuse Cloud and OpenAI, and store your credentials in a local .env file so the SDK can find them automatically.

In this step, get ready to:
  • Install the Python packages your project needs.
  • Create a free Langfuse Cloud account and an OpenAI API key.
  • Configure environment variables and verify the connection.
Install Python dependencies

Your project uses three packages: langfuse for observability, openai for calling models, and python-dotenv for loading credentials from a file.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir langfuse-pipeline
cd langfuse-pipeline
  • Install all three packages by running:
pip install langfuse openai python-dotenv

What does this command do?

This downloads and installs langfuse (the tracing SDK and OpenAI wrapper), openai (the official OpenAI Python client), and python-dotenv (which loads .env files into your environment automatically).

  • Confirm the install succeeded by running:
pip show langfuse

You should see package details including the version number and install location.

Seeing a "command not found" error?

If pip is not found, try pip3 instead. On some systems, Python 3's package manager is accessed as pip3.

If you see a permissions error, add --user at the end of the install command to install into your user directory.

Help me troubleshoot this pip install error

Create your Langfuse Cloud account and get API keys

Langfuse Cloud is where your trace data lives. The free Hobby tier gives you 50,000 observations per month, which is more than enough for this project.

  • Open https://cloud.langfuse.com in your browser.
  • Sign up with your preferred method (Google, GitHub, Microsoft, or email/password).
  • Once signed in, create a new project. Give it any name you like (e.g. observable-pipeline).
  • Navigate to your project's Settings, then select API Keys.
  • Click Create new API credentials.
  • Copy the Secret Key (starts with sk-lf-) and the Public Key (starts with pk-lf-). Save them somewhere safe for the next substep.

What is Langfuse?

Langfuse is an open-source LLM engineering platform for observability. It records every call your application makes to an LLM, including the prompt, the model's response, token usage, latency, and cost.

You'll use it throughout this project to trace your pipeline, evaluate response quality, and run experiments comparing different prompts.

Now get your OpenAI API key.

  • Open https://platform.openai.com and sign in (or create an account if you don't have one).
  • Navigate to API Keys in the left sidebar.
  • Click Create new secret key.
  • Copy the key immediately. You won't be able to see it again.

Keep your keys out of code

Never paste API keys directly into your Python files. The .env file you create next keeps credentials local and out of version control.

Configure environment variables and verify the connection

Both the Langfuse SDK and OpenAI SDK read credentials from environment variables. You'll store them in a .env file that python-dotenv loads at runtime.

  • Create a file called .env inside your langfuse-pipeline folder. Paste the following content, replacing the placeholder values with your real keys:
OPENAI_API_KEY=your-api-key-here
LANGFUSE_SECRET_KEY=your-langfuse-secret-key-here
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key-here
LANGFUSE_BASE_URL=https://cloud.langfuse.com

What does each variable do?

  • OPENAI_API_KEY authenticates your requests to OpenAI's models.
  • LANGFUSE_SECRET_KEY and LANGFUSE_PUBLIC_KEY tell the Langfuse SDK which project to send trace data to.
  • LANGFUSE_BASE_URL points to the Langfuse Cloud EU region. If you chose the US region during sign-up, use https://us.cloud.langfuse.com instead.
  • Save the .env file.
  • Verify your Langfuse connection by running this quick test:
python -c "from dotenv import load_dotenv; load_dotenv(); from langfuse import get_client; lf = get_client(); lf.flush(); print('Connection successful!')"

You should see Connection successful! printed in your terminal. This confirms that your credentials are valid and the SDK can reach Langfuse Cloud.

Seeing an authentication error?

Double-check that your .env file is in the same directory where you ran the command. The load_dotenv() call looks for .env in the current working directory.

Make sure there are no extra spaces around the = signs in your .env file. The format should be KEY=value with no spaces.

If you signed up for the US region of Langfuse Cloud, change LANGFUSE_BASE_URL to https://us.cloud.langfuse.com.

Help me fix my Langfuse authentication error

✔️ Awesome, I've got everything!

Great. Double check you've saved your .env file.

ⓧ I'd like to double check the full code

Here is the complete .env file for comparison:

OPENAI_API_KEY=your-api-key-here
LANGFUSE_SECRET_KEY=your-langfuse-secret-key-here
LANGFUSE_PUBLIC_KEY=your-langfuse-public-key-here
LANGFUSE_BASE_URL=https://cloud.langfuse.com

Your environment is ready. Next up, you'll build a multi-step AI pipeline and see every LLM call traced in your Langfuse dashboard.

Build and Trace a Multi-Step AI Pipeline

Your environment is connected and ready to go. Now it's time to build something real.

In this step, you'll create a two-stage AI pipeline that classifies a question into a domain and then generates an answer. Every LLM call will be automatically traced in Langfuse, giving you full visibility into inputs, outputs, token usage, and cost.

In this step, get ready to:
  • Create a multi-step pipeline with a question classifier and answer generator.
  • Instrument both stages with Langfuse's @observe() decorator for automatic tracing.
  • View your traces in the Langfuse dashboard with model, token, and cost data.
Set up the pipeline file

The pipeline needs two key integrations from Langfuse: a drop-in OpenAI wrapper that automatically captures every API call, and the @observe() decorator that groups calls into a structured trace.

  • Create a new file called pipeline.py in your project folder.
  • Add the following imports and client setup to the top of pipeline.py:
import os
from dotenv import load_dotenv
from langfuse.openai import OpenAI
from langfuse import observe, get_client

load_dotenv()

client = OpenAI()
langfuse = get_client()

What does this code do?

from langfuse.openai import OpenAI replaces the standard OpenAI import with Langfuse's drop-in wrapper. This wrapper automatically records every API call (prompt, model, output, tokens, cost) without changing how you write code.

get_client() creates a singleton Langfuse client that reads your environment variables and handles sending trace data in the background.

  • Save pipeline.py.

You won't see a change yet. The next chunk adds the pipeline functions that actually make LLM calls.

Build the classifier and generator stages

A production AI pipeline rarely makes a single LLM call. Splitting the work into stages (classify, then generate) gives you granular visibility into where time and tokens are spent.

  • Below the langfuse = get_client() line in pipeline.py, add the classifier and generator functions:
@observe()
def classify_question(question: str) -> str:
    """Classify a question into a domain category."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Classify the following question into exactly one category: 'python', 'javascript', 'devops', or 'general'. Respond with only the category name.",
            },
            {"role": "user", "content": question},
        ],
        temperature=0,
    )
    return response.choices[0].message.content.strip().lower()


@observe()
def generate_answer(question: str, category: str, system_prompt: str) -> str:
    """Generate an answer to a technical question."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {
                "role": "user",
                "content": f"[Category: {category}]\n\nQuestion: {question}",
            },
        ],
        temperature=0.7,
    )
    return response.choices[0].message.content

What does this code do?

classify_question() sends the user's question to gpt-4o-mini and gets back a single-word category (python, javascript, devops, or general). Setting temperature=0 makes classification deterministic.

generate_answer() takes the question plus its category and produces a detailed answer. The category is prepended to the user message so the model has routing context.

The @observe() decorator on each function creates a nested span inside the parent trace. Langfuse automatically captures the function name, inputs, outputs, and timing for each span.

  • Save pipeline.py.

You won't see output yet. The next chunk adds the orchestrator and runs the pipeline.

Orchestrate the pipeline and verify your traces

The orchestrator function ties the two stages together and becomes the top-level trace in Langfuse. A main block at the bottom runs a few sample questions so you can see real traces immediately.

  • Below the generate_answer function in pipeline.py, add the orchestrator and main block:
@observe()
def run_pipeline(question: str, system_prompt: str = None) -> dict:
    """Run the full question-answering pipeline."""
    if system_prompt is None:
        system_prompt = (
            "You are a helpful technical assistant. "
            "Provide clear, concise answers to programming questions. "
            "Include code examples when relevant."
        )

    category = classify_question(question)
    answer = generate_answer(question, category, system_prompt)

    return {"question": question, "category": category, "answer": answer}


if __name__ == "__main__":
    questions = [
        "How do I read a file in Python?",
        "What is a closure in JavaScript?",
        "How do I set up a CI/CD pipeline with GitHub Actions?",
    ]

    for q in questions:
        result = run_pipeline(q)
        print(f"\n{'='*60}")
        print(f"Question: {result['question']}")
        print(f"Category: {result['category']}")
        print(f"Answer: {result['answer'][:200]}...")

    langfuse.flush()
    print("\n\nDone! Check your Langfuse dashboard for traces.")

What does this code do?

run_pipeline() is the top-level function that calls classify_question() then generate_answer() in sequence. Because all three functions have @observe(), Langfuse creates a parent trace for run_pipeline with nested child spans for each stage.

langfuse.flush() at the end ensures all trace data is sent to Langfuse before the script exits. Without it, short-lived scripts might exit before the background queue finishes exporting.

  • Save pipeline.py.
  • Run the pipeline by executing this command in your terminal:
python pipeline.py

You should see output for each of the three questions, showing the classified category and a truncated answer.

  • Switch to your Langfuse dashboard in your browser.
  • Click on one of the run_pipeline traces to expand it.
  • You should see nested spans for classify_question and generate_answer, each showing model, tokens, latency, and cost.

Why is tracing valuable here?

Without tracing, a multi-step pipeline is a black box. You can't tell whether a bad answer came from a misclassification or a weak generation prompt.

With Langfuse traces, you see exactly what went into each LLM call and what came out. You can compare latency across stages, track cost per question, and spot regressions over time.

Not seeing traces in Langfuse?

Make sure your .env file has the correct LANGFUSE_SECRET_KEY, LANGFUSE_PUBLIC_KEY, and LANGFUSE_BASE_URL values from your Langfuse project settings.

Check your terminal output for errors. If you see an authentication error, regenerate your API keys in the Langfuse dashboard under Settings > API Keys.

Traces can take a few seconds to appear. Refresh the page if you don't see them immediately. Help me debug why my traces aren't showing up in Langfuse.

✔️ Awesome, I've got everything!

Great. Double check you've saved pipeline.py and confirmed your traces in Langfuse before moving on.

ⓧ I'd like to double check the full code

import os
from dotenv import load_dotenv
from langfuse.openai import OpenAI
from langfuse import observe, get_client

load_dotenv()

client = OpenAI()
langfuse = get_client()


@observe()
def classify_question(question: str) -> str:
    """Classify a question into a domain category."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Classify the following question into exactly one category: 'python', 'javascript', 'devops', or 'general'. Respond with only the category name.",
            },
            {"role": "user", "content": question},
        ],
        temperature=0,
    )
    return response.choices[0].message.content.strip().lower()


@observe()
def generate_answer(question: str, category: str, system_prompt: str) -> str:
    """Generate an answer to a technical question."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {
                "role": "user",
                "content": f"[Category: {category}]\n\nQuestion: {question}",
            },
        ],
        temperature=0.7,
    )
    return response.choices[0].message.content


@observe()
def run_pipeline(question: str, system_prompt: str = None) -> dict:
    """Run the full question-answering pipeline."""
    if system_prompt is None:
        system_prompt = (
            "You are a helpful technical assistant. "
            "Provide clear, concise answers to programming questions. "
            "Include code examples when relevant."
        )

    category = classify_question(question)
    answer = generate_answer(question, category, system_prompt)

    return {"question": question, "category": category, "answer": answer}


if __name__ == "__main__":
    questions = [
        "How do I read a file in Python?",
        "What is a closure in JavaScript?",
        "How do I set up a CI/CD pipeline with GitHub Actions?",
    ]

    for q in questions:
        result = run_pipeline(q)
        print(f"\n{'='*60}")
        print(f"Question: {result['question']}")
        print(f"Category: {result['category']}")
        print(f"Answer: {result['answer'][:200]}...")

    langfuse.flush()
    print("\n\nDone! Check your Langfuse dashboard for traces.")

Your pipeline is fully instrumented. Next up, you'll build an evaluation layer that automatically judges the quality of these responses using LLM-as-a-Judge.

Add Scores and Custom Evaluation

Your pipeline is generating traces for every question it processes. You can see the inputs, outputs, latency, and cost of each LLM call in your Langfuse dashboard.

But how do you know if the answers are actually good? In this step, you'll build an automated evaluation layer using a pattern called LLM-as-a-Judge. Instead of manually reviewing every response, you'll write evaluator functions that ask another LLM to score quality on your behalf.

In this step, get ready to:
  • Write an accuracy evaluator that scores whether answers match expected outputs.
  • Write a relevance evaluator that scores whether answers address the question.
  • Understand how evaluation scores attach to traces in the Langfuse dashboard.
Write a custom evaluator using LLM-as-a-Judge

The LLM-as-a-Judge pattern uses one LLM to evaluate the output of another. You send the original output and a scoring rubric to a judge model, and it returns a structured score with reasoning.

You'll build two evaluators that follow Langfuse's evaluator signature. Each function receives keyword arguments (output, expected_output, input) and returns an Evaluation object with a name, a numeric value, and a comment.

  • Create a new file called evaluate.py in your project folder by running:
touch evaluate.py
  • Confirm the file exists by running:
ls

You should see evaluate.py listed alongside pipeline.py and .env.

  • Open evaluate.py in your editor and add the imports and client setup by pasting this code:
import os
from dotenv import load_dotenv
from langfuse.openai import OpenAI
from langfuse import get_client, observe, Evaluation

load_dotenv()

client = OpenAI()
langfuse = get_client()

What does this setup do?

This mirrors the same pattern from pipeline.py. The OpenAI import from langfuse.openai ensures even your judge model's calls are traced. The Evaluation class is Langfuse's return type for evaluator functions.

You won't see a change yet. The next chunk adds the first evaluator function.

  • Below the client setup in evaluate.py, add the accuracy evaluator function:
def accuracy_evaluator(*, output, expected_output, **kwargs) -> Evaluation:
    """Evaluate if the answer addresses the question correctly using LLM-as-a-Judge."""
    judge_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are an evaluation judge. Score whether the given answer "
                    "correctly addresses the expected answer. "
                    "Respond with a JSON object: {\"score\": <0.0 to 1.0>, \"reasoning\": \"<brief explanation>\"}"
                ),
            },
            {
                "role": "user",
                "content": f"Expected answer: {expected_output}\n\nActual answer: {output}",
            },
        ],
        temperature=0,
    )

    import json

    try:
        result = json.loads(judge_response.choices[0].message.content)
        return Evaluation(
            name="accuracy", value=result["score"], comment=result["reasoning"]
        )
    except (json.JSONDecodeError, KeyError):
        return Evaluation(name="accuracy", value=0.0, comment="Failed to parse judge response")

What does this code do?

The accuracy_evaluator function takes the pipeline's output and an expected answer, then asks gpt-4o-mini to judge how well the actual answer matches the expected one.

The judge returns a JSON object with a score (0.0 to 1.0) and a reasoning explanation. The function wraps this into a Langfuse Evaluation object. If JSON parsing fails, it returns a safe default score of 0.0.

You won't see a change yet. The next chunk adds the second evaluator.

Add the relevance evaluator

Accuracy checks whether the answer matches what you expected. Relevance checks something different: whether the answer actually addresses the question that was asked.

  • Below the accuracy_evaluator function in evaluate.py, add the relevance evaluator:
def relevance_evaluator(*, output, input, **kwargs) -> Evaluation:
    """Evaluate if the answer is relevant to the question asked."""
    judge_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are an evaluation judge. Score whether the answer is relevant "
                    "to the question. A relevant answer directly addresses what was asked. "
                    "Respond with a JSON object: {\"score\": <0.0 to 1.0>, \"reasoning\": \"<brief explanation>\"}"
                ),
            },
            {
                "role": "user",
                "content": f"Question: {input}\n\nAnswer: {output}",
            },
        ],
        temperature=0,
    )

    import json

    try:
        result = json.loads(judge_response.choices[0].message.content)
        return Evaluation(
            name="relevance", value=result["score"], comment=result["reasoning"]
        )
    except (json.JSONDecodeError, KeyError):
        return Evaluation(name="relevance", value=0.0, comment="Failed to parse judge response")

What does this code do?

The relevance_evaluator takes input (the original question) and output (the generated answer), then asks the judge model to score how directly the answer addresses the question.

This evaluator doesn't need an expected_output because it's measuring relevance to the question itself, not correctness against a reference answer. Both evaluators follow the same pattern: send to judge, parse JSON, return Evaluation.

  • Save evaluate.py.
  • Test that the file imports correctly by running:
python -c "from evaluate import accuracy_evaluator, relevance_evaluator; print('Evaluators loaded successfully')"

You should see Evaluators loaded successfully in your terminal.

Seeing an import error?

  • Make sure you saved evaluate.py before running the command.
  • Check that from langfuse import get_client, observe, Evaluation matches exactly. A typo in Evaluation (capital E) will cause an ImportError.
  • Confirm your terminal is in the same folder as evaluate.py. Run ls to verify.

Why can't I import my evaluators?

✔️ Awesome, I've got everything!

Great. Double check you've saved evaluate.py before moving on.

ⓧ I'd like to double check the full code

import os
from dotenv import load_dotenv
from langfuse.openai import OpenAI
from langfuse import get_client, observe, Evaluation

load_dotenv()

client = OpenAI()
langfuse = get_client()


def accuracy_evaluator(*, output, expected_output, **kwargs) -> Evaluation:
    """Evaluate if the answer addresses the question correctly using LLM-as-a-Judge."""
    judge_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are an evaluation judge. Score whether the given answer "
                    "correctly addresses the expected answer. "
                    "Respond with a JSON object: {\"score\": <0.0 to 1.0>, \"reasoning\": \"<brief explanation>\"}"
                ),
            },
            {
                "role": "user",
                "content": f"Expected answer: {expected_output}\n\nActual answer: {output}",
            },
        ],
        temperature=0,
    )

    import json

    try:
        result = json.loads(judge_response.choices[0].message.content)
        return Evaluation(
            name="accuracy", value=result["score"], comment=result["reasoning"]
        )
    except (json.JSONDecodeError, KeyError):
        return Evaluation(name="accuracy", value=0.0, comment="Failed to parse judge response")


def relevance_evaluator(*, output, input, **kwargs) -> Evaluation:
    """Evaluate if the answer is relevant to the question asked."""
    judge_response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are an evaluation judge. Score whether the answer is relevant "
                    "to the question. A relevant answer directly addresses what was asked. "
                    "Respond with a JSON object: {\"score\": <0.0 to 1.0>, \"reasoning\": \"<brief explanation>\"}"
                ),
            },
            {
                "role": "user",
                "content": f"Question: {input}\n\nAnswer: {output}",
            },
        ],
        temperature=0,
    )

    import json

    try:
        result = json.loads(judge_response.choices[0].message.content)
        return Evaluation(
            name="relevance", value=result["score"], comment=result["reasoning"]
        )
    except (json.JSONDecodeError, KeyError):
        return Evaluation(name="relevance", value=0.0, comment="Failed to parse judge response")
View evaluation results in the Langfuse dashboard

Your evaluators are ready to score pipeline outputs. When used with the experiment runner (coming in the next step), they automatically attach scores to each trace.

  • Test one evaluator live by running this command:
python -c "
from evaluate import accuracy_evaluator
result = accuracy_evaluator(
    output='Use the open() function with a with statement to read files safely.',
    expected_output='Use the open() function with a context manager (with statement) to read files safely.'
)
print(f'Score: {result.value}')
print(f'Comment: {result.comment}')
"

You should see a score close to 1.0 and a comment explaining that the answer correctly addresses the expected output.

How do scores connect to traces?

Each Evaluation object you return has three fields: name (like "accuracy" or "relevance"), value (a numeric score from 0.0 to 1.0), and comment (the judge's reasoning).

When you pass these evaluators to Langfuse's experiment runner, it automatically attaches the returned scores to the corresponding traces. You can then view scores on the Scores tab of any trace detail view, or navigate to Scores in the left navigation menu to see analytics across all scored traces.

Score returned 0.0 unexpectedly?

  • A score of 0.0 with the comment "Failed to parse judge response" means the judge model didn't return valid JSON. Check that your OPENAI_API_KEY is still valid in your .env file.
  • If you see a rate limit error, wait a few seconds and try again.

Why is my evaluator returning 0.0 instead of a real score?

Your evaluation layer is complete. In the next step, you'll put these evaluators to work by running a structured experiment that compares two different prompt variants side-by-side with quantitative scores.

Run a Dataset-Driven Experiment

Your evaluators can score individual traces, but how do you know whether a prompt change actually improves overall quality? You need a structured way to run the same set of questions through different prompt variants and compare the results.

In this step, you'll create a test dataset, define two competing system prompts, and use Langfuse's Experiment Runner to compare them side-by-side with quantitative scores.

In this step, get ready to:
  • Create a test dataset of questions with expected answers.
  • Run two experiments comparing different prompt variants.
  • Analyze experiment results side-by-side in the Langfuse dashboard.
Create a test dataset with expected answers

A dataset in Langfuse is a collection of test cases that you run your application against each time you make a change. Each item has an input (the question) and an expected output (what a correct answer looks like).

  • Create a new file called experiment.py in your project folder.
  • Add the imports, client setup, and test dataset by pasting the following code:
import os
from dotenv import load_dotenv
from langfuse.openai import OpenAI
from langfuse import get_client, Evaluation

from pipeline import classify_question, generate_answer
from evaluate import accuracy_evaluator, relevance_evaluator

load_dotenv()

client = OpenAI()
langfuse = get_client()

# Define test dataset
test_data = [
    {
        "input": "How do I read a file in Python?",
        "expected_output": "Use the open() function with a context manager (with statement) to read files safely.",
    },
    {
        "input": "What is a closure in JavaScript?",
        "expected_output": "A closure is a function that has access to variables from its outer (enclosing) scope, even after the outer function has returned.",
    },
    {
        "input": "How do I create a Docker container?",
        "expected_output": "Write a Dockerfile defining the image, then use docker build to create the image and docker run to start a container.",
    },
    {
        "input": "What is the difference between let and var in JavaScript?",
        "expected_output": "let has block scope and cannot be redeclared, while var has function scope and can be redeclared.",
    },
    {
        "input": "How do I handle errors in Python?",
        "expected_output": "Use try/except blocks to catch and handle exceptions. You can catch specific exception types.",
    },
]

What does this code do?

  • The imports pull in classify_question and generate_answer from your pipeline, plus both evaluators from evaluate.py. This lets the experiment reuse your existing pipeline and scoring logic.
  • test_data defines five test cases, each with an input question and an expected_output that describes what a good answer should cover.
  • These expected outputs are not exact strings the model must produce. They're reference answers the evaluator uses to judge quality.
  • Save experiment.py.

Seeing an import error?

Make sure pipeline.py and evaluate.py are in the same folder as experiment.py. Python can only import from files in the same directory or on the module path.

Help me debug my import error in experiment.py

Define prompt variants and run the experiments

The power of dataset-driven experiments is comparing different approaches with the same test cases. You'll define two system prompts and create a task function that connects them to your pipeline.

  • Add the two prompt variants and the task factory below your test_data list in experiment.py:
# Two prompt variants to compare
PROMPT_A = (
    "You are a helpful technical assistant. "
    "Provide clear, concise answers to programming questions. "
    "Include code examples when relevant."
)

PROMPT_B = (
    "You are a senior software engineer explaining concepts to a junior developer. "
    "Start with a one-sentence summary, then provide a detailed explanation with a practical code example. "
    "End with a common pitfall to avoid."
)


def create_task(system_prompt: str):
    """Create a task function for the experiment runner."""

    def task(*, item, **kwargs):
        question = item.input
        category = classify_question(question)
        answer = generate_answer(question, category, system_prompt)
        return answer

    return task

What does this code do?

  • PROMPT_A is a concise assistant style. PROMPT_B is a more detailed, structured teaching style. These represent two approaches a team might debate.
  • create_task() is a factory function that returns a task closure. The Experiment Runner calls this task once per dataset item, passing the item's input and expected output.
  • Inside the closure, item.input gives you the question string. The task classifies it and generates an answer using the specified system prompt, then returns the answer for evaluation.
  • Save experiment.py.

Now you need the main function that uploads the dataset to Langfuse and runs both experiments.

  • Add the run_experiments() function and entry point below create_task() in experiment.py:
def run_experiments():
    """Upload dataset and run experiments comparing two prompt variants."""
    # Create or get the dataset in Langfuse
    dataset = langfuse.create_dataset(
        name="tech-qa-eval",
        description="Technical QA evaluation dataset",
    )

    # Upload test items
    for item in test_data:
        langfuse.create_dataset_item(
            dataset_name="tech-qa-eval",
            input=item["input"],
            expected_output=item["expected_output"],
        )

    # Run experiment with Prompt A
    print("Running experiment with Prompt A...")
    result_a = dataset.run_experiment(
        name="Tech QA - Prompt A (Concise)",
        task=create_task(PROMPT_A),
        evaluators=[accuracy_evaluator, relevance_evaluator],
    )

    # Run experiment with Prompt B
    print("Running experiment with Prompt B...")
    result_b = dataset.run_experiment(
        name="Tech QA - Prompt B (Detailed)",
        task=create_task(PROMPT_B),
        evaluators=[accuracy_evaluator, relevance_evaluator],
    )

    print("\nExperiments complete!")
    print("View results side-by-side in your Langfuse dashboard:")
    print("  -> Navigate to Datasets > tech-qa-eval > Experiments")

    langfuse.flush()


if __name__ == "__main__":
    run_experiments()

What does this code do?

  • langfuse.create_dataset() creates a named dataset in your Langfuse project. If it already exists, Langfuse returns the existing one.
  • The loop uploads each test case as a dataset item with its input and expected output.
  • dataset.run_experiment() is the Experiment Runner. It iterates through every dataset item, calls your task function on each one, then runs the evaluators against the output. Each experiment gets a unique name so you can compare them later.
  • langfuse.flush() ensures all observations are exported before the script exits.
  • Save experiment.py.
  • Run the experiment script by running this command:
python experiment.py

This will take a minute or two. The script runs 5 questions through both prompt variants (10 LLM calls for answers, plus 10 classification calls, plus 20 evaluation calls).

  • Wait for the script to finish. You should see output like:
Running experiment with Prompt A...
Running experiment with Prompt B...

Experiments complete!
View results side-by-side in your Langfuse dashboard:
  -> Navigate to Datasets > tech-qa-eval > Experiments

Script running slowly or timing out?

The experiment makes many API calls (around 40 total). If you see rate limit errors, wait 30 seconds and run the script again. Langfuse's Experiment Runner handles retries for most transient errors.

Help me debug my experiment.py script

✔️ Awesome, I've got everything!

Great. Double check you've saved experiment.py and the script completed successfully.

ⓧ I'd like to double check the full code

import os
from dotenv import load_dotenv
from langfuse.openai import OpenAI
from langfuse import get_client, Evaluation

from pipeline import classify_question, generate_answer
from evaluate import accuracy_evaluator, relevance_evaluator

load_dotenv()

client = OpenAI()
langfuse = get_client()

# Define test dataset
test_data = [
    {
        "input": "How do I read a file in Python?",
        "expected_output": "Use the open() function with a context manager (with statement) to read files safely.",
    },
    {
        "input": "What is a closure in JavaScript?",
        "expected_output": "A closure is a function that has access to variables from its outer (enclosing) scope, even after the outer function has returned.",
    },
    {
        "input": "How do I create a Docker container?",
        "expected_output": "Write a Dockerfile defining the image, then use docker build to create the image and docker run to start a container.",
    },
    {
        "input": "What is the difference between let and var in JavaScript?",
        "expected_output": "let has block scope and cannot be redeclared, while var has function scope and can be redeclared.",
    },
    {
        "input": "How do I handle errors in Python?",
        "expected_output": "Use try/except blocks to catch and handle exceptions. You can catch specific exception types.",
    },
]

# Two prompt variants to compare
PROMPT_A = (
    "You are a helpful technical assistant. "
    "Provide clear, concise answers to programming questions. "
    "Include code examples when relevant."
)

PROMPT_B = (
    "You are a senior software engineer explaining concepts to a junior developer. "
    "Start with a one-sentence summary, then provide a detailed explanation with a practical code example. "
    "End with a common pitfall to avoid."
)


def create_task(system_prompt: str):
    """Create a task function for the experiment runner."""

    def task(*, item, **kwargs):
        question = item.input
        category = classify_question(question)
        answer = generate_answer(question, category, system_prompt)
        return answer

    return task


def run_experiments():
    """Upload dataset and run experiments comparing two prompt variants."""
    # Create or get the dataset in Langfuse
    dataset = langfuse.create_dataset(
        name="tech-qa-eval",
        description="Technical QA evaluation dataset",
    )

    # Upload test items
    for item in test_data:
        langfuse.create_dataset_item(
            dataset_name="tech-qa-eval",
            input=item["input"],
            expected_output=item["expected_output"],
        )

    # Run experiment with Prompt A
    print("Running experiment with Prompt A...")
    result_a = dataset.run_experiment(
        name="Tech QA - Prompt A (Concise)",
        task=create_task(PROMPT_A),
        evaluators=[accuracy_evaluator, relevance_evaluator],
    )

    # Run experiment with Prompt B
    print("Running experiment with Prompt B...")
    result_b = dataset.run_experiment(
        name="Tech QA - Prompt B (Detailed)",
        task=create_task(PROMPT_B),
        evaluators=[accuracy_evaluator, relevance_evaluator],
    )

    print("\nExperiments complete!")
    print("View results side-by-side in your Langfuse dashboard:")
    print("  -> Navigate to Datasets > tech-qa-eval > Experiments")

    langfuse.flush()


if __name__ == "__main__":
    run_experiments()
Analyze experiment results in the Langfuse dashboard

Your experiments are now stored in Langfuse with all their traces, outputs, and evaluation scores. The dashboard lets you compare the two prompt variants side-by-side.

  • Switch back to your Langfuse dashboard in the browser.
  • Click Datasets in the left navigation.
  • Click on the tech-qa-eval dataset.
  • Navigate to the Experiments tab.

You should see both experiment runs listed: Tech QA - Prompt A (Concise) and Tech QA - Prompt B (Detailed).

What should I see?

The experiments table shows high-level metrics for each run, including average accuracy and relevance scores, total cost, and latency. You can compare the two rows to see which prompt variant scored higher overall.

Click into either experiment to see per-item results. Each item shows the question, the generated answer, and the individual accuracy and relevance scores from your evaluators.

Why run experiments instead of eyeballing outputs?

Manually checking a few outputs feels good but does not generalize. A prompt change might improve 3 answers while silently degrading 2 others. Dataset-driven experiments give you aggregate scores across a consistent set of inputs, so you can spot regressions before they reach production.

You now have a repeatable experiment workflow. Any time you want to test a prompt change, you can add it as a new variant, rerun the dataset, and compare scores to make data-driven decisions.

Secret mission

Add a Hallucination Guardrail

Your pipeline traces every call and evaluates quality. But what about answers that sound confident yet contain made-up facts? Build a guardrail that automatically detects hallucinated responses and flags them on the trace before they reach users.

Clean Up Your Resources

Clean Up Your Resources

This project runs locally and uses pay-per-call APIs, so there are no ongoing costs once you stop running the scripts. Decide whether to keep your resources intact, pause to prevent accidental API charges, or delete everything entirely.

Resources you used:

  • Local project files (.env, pipeline.py, evaluate.py, experiment.py, guardrail.py).
  • Python packages (langfuse, openai, python-dotenv).
  • Langfuse Cloud project (traces, datasets, scores, experiment runs).
  • OpenAI API key stored in your .env file.

Keep everything running

No action needed. Choose this if you plan to keep building on this pipeline or experimenting with new evaluators.

  • Your local files stay on disk and can be rerun anytime.
  • Your Langfuse Cloud project remains on the free Hobby tier (50,000 units/month, 30-day retention). No charges accrue.
  • Your OpenAI API key remains active. You are only charged when you run the scripts.

Pause - I'll come back to this later

Disable your API key so no accidental script runs incur charges. Your files and Langfuse data stay intact for when you return.

  • Go to https://platform.openai.com/api-keys.
  • Click the trashcan icon next to the API key you used for this project to delete it.
  • When you come back, create a new API key and update your .env file with the new value.

Delete - I don't want to use this again

Remove all project files, uninstall packages, and clean up your cloud accounts.

  • Delete your local project files by running this command (macOS/Linux):
rm -rf pipeline.py evaluate.py experiment.py guardrail.py .env
  • On Windows, run this instead:
Remove-Item pipeline.py, evaluate.py, experiment.py, guardrail.py, .env
  • Uninstall the Python packages by running:
pip uninstall langfuse openai python-dotenv -y
  • Delete your OpenAI API key:
  • Delete your Langfuse Cloud project:
  • Open your Langfuse dashboard and navigate to your project's Settings page.
  • Scroll to the Danger Zone within the General section.
  • Click Delete and confirm. This revokes all project API keys and permanently removes all traces, datasets, scores, and experiment runs.
  • Delete your Langfuse account (optional):
  • Click your user menu in the bottom-right corner of the Langfuse dashboard.
  • Select Account Settings.
  • Follow the prompts to delete your account.

Nice Work!

Nice Work!

You've just built a production-grade observable AI pipeline with automated evaluation and structured experimentation. This is the exact workflow professional teams use to ship LLM features with confidence.

You've learned how to:

  • Build a multi-step AI pipeline with end-to-end tracing in Langfuse, showing every LLM call's inputs, outputs, latency, and cost.
  • Create custom LLM-as-a-Judge evaluators that automatically score response quality for accuracy and relevance.
  • Run dataset-driven experiments that quantitatively compare prompt variants side-by-side with reproducible results.
  • Secret Mission: Built a hallucination guardrail that catches and flags ungrounded responses before they reach users.

Ready to quiz yourself?