Post cover image
langfuse

June 3, 2026

Production-Grade agentic observability: a complete Langfuse Deep Dive

By allglenn

23 min read

You shipped an LLM agent. Now what?

You stayed up late wiring together GPT-4o, a vector database, and a complex routing prompt. You deployed your new RAG application. Users are actually hitting the API. And then… complete silence.

You have no idea if it's actually performing well. Your standard error logs show HTTP statuses, but they tell you absolutely nothing about why your agent gave a bizarre, hallucinated answer to the user who just screenshotted it and tagged your team on Twitter.

This is the reality of moving past simple chat prompts: production is a black box.

You can't console.log() your way through a 15-step RAG pipeline. You can't write standard unit tests for non-deterministic model behaviors. And you definitely can't explain to your CTO why your customer support agent told a user their order "probably exists somewhere in the universe."

The Missing Observability Layer

This is where Langfuse comes in. It provides the deep tracking and telemetry that complex RAG systems and autonomous agents actually need to be maintainable.

Instead of guessing what happened during a user session, you get:

  • Full Traces: Visualize the entire lifecycle of an agent execution — from the initial query rewrite and vector database retrieval steps to the final LLM synthesis.
  • Structured Evaluations: Run automated scoring pipelines to test for hallucinations, relevance, and toxicity.
  • Prompt Management: Version control your prompts independently of your application code deployments.
  • Regression Testing: Build "golden datasets" from real user interactions to ensure model updates don't break existing behavior.

All of this is packed into a single open-source platform.

What We're Building

This guide is a practical blueprint for taking your RAG and agent workflows from brittle prototypes to production-grade deployments. We'll break down the core architecture concepts, walk through a hands-on implementation for an AI customer support agent, and honestly compare Langfuse against alternative tooling so you can choose the right stack for your team.

Let's open up the black box.

What is Langfuse and Why Does It Exist?

Before Langfuse, engineering teams building LLM applications ran into a frustrating truth: traditional debugging tools break when your core application logic is handled by a non-deterministic statistical model.

You can't set a breakpoint in a neural network. You can't guarantee matching outputs for identical inputs. When an agent fails, the bug usually isn't a syntax error ,it's a poorly formatted prompt, a bad document chunking strategy in your vector database, or unexpected behavior from the foundational model itself.

Langfuse was founded in 2023 by Max Langkamp, Marc Klingen, and Clemens Rawert after they hit these exact walls building their own AI apps. They open-sourced the platform under the Apache 2.0 license, and it quickly became one of the most prominent LLM engineering tooling repositories on GitHub.

What problem does it solve?

When you move from simple API calls to multi-step RAG pipelines and autonomous agents, the operational challenges shift entirely:

None

Core Concepts: Traces, Spans, Generations, and Scores

Before writing a single line of code, you need to understand Langfuse's four foundational primitives. If you've ever used OpenTelemetry, these will feel instantly familiar, but they have been tailored specifically for RAG pipelines and autonomous agents.

They dictate how data is structured inside the platform:

None

1. Trace

A Trace represents a single end-to-end operation in your application — typically a single user request. It acts as the root of your entire data tree, capturing vital top-level metadata like a unique ID, custom name, optional user or session IDs, and overall execution duration. Think of it exactly like a request trace in distributed system tracing, except it encapsulates your entire LLM reasoning chain.

2. Span

A Span is any unit of work inside a trace that isn't an explicit LLM call. Spans wrap the engineering logic surrounding your models. Database queries, vector similarity searches, document reranking steps, or external API calls all become spans. Because you can nest spans arbitrarily deep, they allow you to map out complex, multi-step agent behaviors into a clean, visual call tree.

3. Generation

A Generation is a specialized type of span reserved strictly for LLM calls. Langfuse isolates these because they require unique tracking parameters. A generation automatically captures the exact input messages, the model's output, prompt names, and token counts. You never have to compute costs manually; Langfuse references its own managed pricing database to calculate token spend for you in real time.

4. Score

A Score is a quality signal attached directly to a trace, span, or generation. Scores are the mechanism that transforms Langfuse from a passive logging utility into a true evaluation platform. They allow you to attach quantitative metrics based on three sources:

  • Human Reviewers: Production outputs annotated directly within the Langfuse UI.
  • LLM Judges: Automated evaluations running at scale to check relevance or tone.
  • Your Own Code: Rule-based checks executed programmatically (e.g., verifying if an output contains a required disclaimer or is valid JSON).

setup in under 5 minutes

step 1: create a Langfuse account

Go to cloud.langfuse.com, create a project, and grab your API keys from Settings → API Keys.

None

step 2: install the SDK

# Python
pip install langfuse openai anthropic

# Node.js
npm install langfuse openai

step 3: set environment variables

# .env
LANGFUSE_PUBLIC_KEY=pk-lf-xxxxxxxxxxxxxxxx
LANGFUSE_SECRET_KEY=sk-lf-xxxxxxxxxxxxxxxx
LANGFUSE_HOST=https://cloud.langfuse.com  # or your self-hosted URL

step 4: your first trace (drop-in integration)

This is where Langfuse earns its "zero config" reputation. If you're using OpenAI or Anthropic, replace your import and you're done:

import os, base64
from dotenv import load_dotenv
load_dotenv()

# One-time OTEL setup → sends traces to Langfuse
# (langfuse.anthropic was removed in v4; this is the v4 equivalent)
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry import trace
from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor

auth = base64.b64encode(
    f"{os.environ['LANGFUSE_PUBLIC_KEY']}:{os.environ['LANGFUSE_SECRET_KEY']}".encode()
).decode()
provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(
        endpoint=f"{os.environ['LANGFUSE_BASE_URL']}/api/public/otel/v1/traces",
        headers={"Authorization": f"Basic {auth}"}
    )
))
trace.set_tracer_provider(provider)
AnthropicInstrumentor().instrument()

# Before
# from anthropic import Anthropic

# After — nothing else changes
from anthropic import Anthropic

client = Anthropic()

message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "What is LLM observability?"}]
)

print(message.content[0].text)
# ✅ A trace now appears in your Langfuse dashboardThat’s it. No configuration, no wrappers, no extra parameters. Every call is automatically traced with full inputs, outputs, model, tokens, and cost.

what you'll see in the dashboard

Within seconds of running that code, your Langfuse dashboard will show:

  • The exact messages sent to the model
  • The model's complete response
  • Latency (time to first token + total)
  • Token counts (prompt, completion, total)
  • Cost in USD (automatically calculated)
  • A timeline view of all spans
None

building structured traces with @observe()

The drop-in integration is great for simple scripts. For real applicationsRAG pipelines, agents, multi-step chains , you need structured traces that show you exactly what happened at each step.

Langfuse's @observe() decorator is the primary tool for this. It wraps any Python function and creates a span in the trace tree.

basic decorator usage

from dotenv import load_dotenv
load_dotenv()

from langfuse.decorators import observe, langfuse_context
from anthropic import Anthropic

client = Anthropic()

# --- Mock RAG dependencies ---
class _Doc:
    def __init__(self, content, score):
        self.page_content = content
        self.score = score

class _VectorDB:
    _docs = [
        _Doc("Returns are accepted within 30 days of purchase with receipt.", 0.95),
        _Doc("Items must be unused and in original packaging for a full refund.", 0.88),
        _Doc("Sale items are final sale and cannot be returned.", 0.76),
        _Doc("Online orders can be returned by mail or in-store.", 0.71),
        _Doc("Refunds are processed within 5–7 business days.", 0.65),
    ]
    def similarity_search(self, query, k=5):
        return self._docs[:k]

class _CrossEncoder:
    def rank(self, query, docs):
        return docs  # passthrough for demo

vector_db = _VectorDB()
cross_encoder = _CrossEncoder()
# --- End mocks ---

@observe()
def retrieve_documents(query: str, top_k: int = 5):
    docs = vector_db.similarity_search(query, k=top_k)

    langfuse_context.update_current_observation(
        metadata={
            "num_docs_retrieved": len(docs),
            "top_score": docs[0].score if docs else None
        }
    )
    return [doc.page_content for doc in docs]

@observe()
def rerank_documents(query: str, docs: list):
    reranked = cross_encoder.rank(query, docs)
    return reranked[:3]

@observe()
def answer_question(user_query: str, user_id: str):
    langfuse_context.update_current_trace(
        user_id=user_id,
        tags=["rag", "production"],
        session_id=f"session_{user_id}"
    )

    docs = retrieve_documents(user_query)
    top_docs = rerank_documents(user_query, docs)

    context = "\n\n".join(top_docs)

    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": f"Answer based on this context:\n{context}\n\nQuestion: {user_query}"
            }
        ]
    )

    return message.content[0].text

answer = answer_question("What is our return policy?", user_id="user_42")
print(answer)
# ✅ Root trace + child spans appear in your Langfuse dashboard
None None

This kind of visibility is the difference between debugging in the dark and having a full picture of your pipeline.

async support

Langfuse's decorators work seamlessly in async applications (FastAPI, etc.):

import asyncio
from dotenv import load_dotenv
load_dotenv()

from langfuse.decorators import observe, langfuse_context
from anthropic import AsyncAnthropic

client = AsyncAnthropic()

# --- Mock RAG dependencies ---
class _Doc:
    def __init__(self, content, score):
        self.page_content = content
        self.score = score

DOCS = [
    _Doc("Returns are accepted within 30 days of purchase with receipt.", 0.95),
    _Doc("Items must be unused and in original packaging for a full refund.", 0.88),
    _Doc("Sale items are final sale and cannot be returned.", 0.76),
    _Doc("Online orders can be returned by mail or in-store.", 0.71),
    _Doc("Refunds are processed within 5–7 business days.", 0.65),
]
# --- End mocks ---

@observe()
async def async_retrieve(query: str, top_k: int = 5):
    await asyncio.sleep(0)  # simulate async DB call
    docs = DOCS[:top_k]

    langfuse_context.update_current_observation(
        metadata={
            "num_docs_retrieved": len(docs),
            "top_score": docs[0].score if docs else None
        }
    )
    return [doc.page_content for doc in docs]

@observe()
async def async_llm_call(query: str, docs: list):
    context = "\n\n".join(docs[:3])

    message = await client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": f"Answer based on this context:\n{context}\n\nQuestion: {query}"
            }
        ]
    )

    langfuse_context.update_current_observation(
        input=query,
        output=message.content[0].text,
        usage={
            "input": message.usage.input_tokens,
            "output": message.usage.output_tokens
        },
        model="claude-sonnet-4-6"
    )

    return message.content[0].text

@observe()
async def async_rag_pipeline(query: str, user_id: str):
    langfuse_context.update_current_trace(
        user_id=user_id,
        tags=["rag", "async", "production"],
        session_id=f"session_{user_id}"
    )

    docs = await async_retrieve(query)
    response = await async_llm_call(query, docs)
    return response

async def main():
    answer = await async_rag_pipeline("What is our return policy?", user_id="user_42")
    print(answer)

asyncio.run(main())
# ✅ Root trace + child spans appear in your Langfuse dashboard

Evaluation and scoring: knowing if your app is actually good

Tracing tells you what happened. Scoring tells you how well it happened. This is the part of Langfuse that separates serious teams from teams that are just logging.

the three types of scores

None

1. rule-based scores (start here)

Rule-based scores are instant, free, and require no LLM calls. Add them inline in your pipeline:

import json
from dotenv import load_dotenv
load_dotenv()

from langfuse.decorators import observe, langfuse_context
from anthropic import Anthropic

client = Anthropic()

def call_llm(query: str) -> str:
    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": query}]
    )
    return message.content[0].text

@observe()
def run_support_pipeline(query: str):
    langfuse_context.update_current_trace(tags=["rule-based-scores"])

    response = call_llm(query)

    # Score: response length
    word_count = len(response.split())
    langfuse_context.score_current_trace(
        name="response_length",
        value=word_count
    )

    # Score: did the model refuse?
    refused = any(phrase in response.lower() for phrase in
                  ["i cannot", "i'm unable", "i don't have access"])
    langfuse_context.score_current_trace(
        name="refusal",
        value=1 if refused else 0
    )

    # Score: is the response valid JSON?
    try:
        json.loads(response)
        langfuse_context.score_current_trace(name="valid_json", value=1)
    except Exception:
        langfuse_context.score_current_trace(name="valid_json", value=0)

    return response

result = run_support_pipeline("What is our return policy?")
print(result)
# ✅ Trace appears in Langfuse with 3 scores: response_length, refusal, valid_json
None

2. LLM-as-judge (for nuanced quality)

For things like helpfulness, tone, factual accuracy, or hallucination detection, you need an LLM to evaluate:

import json
import threading
from dotenv import load_dotenv
load_dotenv()

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
from anthropic import Anthropic

lf = Langfuse()
main_client = Anthropic()
eval_client = Anthropic()  # plain client — eval calls are intentionally untraced

CONTEXT = """
Returns are accepted within 30 days of purchase with receipt.
Items must be unused and in original packaging for a full refund.
Sale items are final sale and cannot be returned.
Online orders can be returned by mail or in-store.
Refunds are processed within 5–7 business days.
"""

def evaluate_hallucination(trace_id: str, output: str, context: str):
    """LLM judge: does the output hallucinate beyond the provided context?"""

    eval_prompt = f"""You are a factual accuracy evaluator.

Given the context and the AI's output, determine if the output contains any
factual claims NOT supported by the context (hallucinations).

Context:
{context}

AI Output:
{output}

Respond ONLY with valid JSON in this exact format:
{{"score": 0.95, "reasoning": "...", "hallucination_detected": false}}

Where score is 1.0 = perfectly grounded, 0.0 = completely hallucinated."""

    result = eval_client.messages.create(
        model="claude-haiku-4-5-20251001",  # cheap model for evals
        max_tokens=512,
        messages=[{"role": "user", "content": eval_prompt}]
    )

    raw = result.content[0].text.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip()
    data = json.loads(raw)

    lf.score(
        trace_id=trace_id,
        name="groundedness",
        value=data["score"],
        comment=data["reasoning"]
    )

    print(f"\n--- Eval result ---")
    print(f"Score:    {data['score']}")
    print(f"Reasoning: {data['reasoning']}")
    print(f"Hallucination detected: {data['hallucination_detected']}")

    return data

@observe()
def run_pipeline(query: str):
    langfuse_context.update_current_trace(tags=["llm-as-judge"])

    message = main_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=f"Answer using only the following context:\n{CONTEXT}",
        messages=[{"role": "user", "content": query}]
    )

    response = message.content[0].text
    trace_id = langfuse_context.get_current_trace_id()

    # Fire evaluation in background — doesn't block the user response
    threading.Thread(
        target=evaluate_hallucination,
        args=(trace_id, response, CONTEXT),
        daemon=True
    ).start()

    return response

result = run_pipeline("What is our return policy?")
print(result)

lf.flush()  # ensure both the trace and score are shipped
None

3. human annotation (your ground truth)

Configure annotation queues in the Langfuse UI (Settings → Scores → Score Configs) to define custom rubrics:

  • Helpfulness — scale 1 to 5
  • Tone — categorical: professional / neutral / inappropriate
  • Contains PII — boolean
  • Customer would be satisfied — boolean

Then create an annotation queue that automatically routes a random sample of traces (e.g., 5% of production traffic, 100% of traces with quality score < 0.6) to your quality team for review.

Human labels serve two critical purposes:

  1. They are your ground truth for measuring everything else
  2. They let you calibrate your LLM judges , compare judge scores against human scores on the same traces to validate your eval pipeline

prompt management: ship prompt changes without deploying code

This is one of Langfuse's most underrated features. In most teams, changing a prompt means editing a string in code, opening a pull request, waiting for review, deploying a cycle that can take hours or days. Langfuse breaks this coupling entirely.

how it works

None 
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
from anthropic import Anthropic

lf = Langfuse()
client = Anthropic()

def ensure_prompt_exists():
    """Create the prompt in Langfuse if it doesn't exist yet."""
    try:
        lf.get_prompt("customer-support-agent")
    except Exception:
        lf.create_prompt(
            name="customer-support-agent",
            type="text",
            prompt=(
                "You are a helpful customer support agent for AcmeCorp.\n\n"
                "Customer Name: {{customer_name}}\n"
                "Subscription Plan: {{subscription_plan}}\n"
                "Current Date: {{current_date}}\n\n"
                "Help the customer with their question. Be polite, concise, and accurate. "
                "If you don't know something, say so."
            ),
            labels=["production"],
            commit_message="Initial version"
        )
        print("Prompt created in Langfuse.")

@observe()
def handle_support_query(customer_query: str, customer_name: str, plan: str):
    langfuse_context.update_current_trace(tags=["prompt-management"])

    # Fetch current production prompt — cached locally, refreshed in background
    prompt = lf.get_prompt("customer-support-agent")

    # Compile template with variables
    system_message = prompt.compile(
        customer_name=customer_name,
        subscription_plan=plan,
        current_date=datetime.now().strftime("%Y-%m-%d")
    )

    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=system_message,
        messages=[{"role": "user", "content": customer_query}]
    )

    response = message.content[0].text

    # Link this generation to the prompt version in the trace
    langfuse_context.update_current_observation(
        prompt=prompt,
        usage={
            "input": message.usage.input_tokens,
            "output": message.usage.output_tokens
        }
    )

    return response

ensure_prompt_exists()

result = handle_support_query(
    customer_query="How do I cancel my subscription?",
    customer_name="Alice",
    plan="Pro"
)
print(result)

lf.flush()
# ✅ Trace links to prompt version in Langfuse — edit the prompt in the UI,
#    the next call picks up the new version automatically
None
  1. Create the prompt once in Langfuse (ensure_prompt_exists) The prompt lives in Langfuse, not in your code. It uses {{double_braces}} for variables. Labeling it production makes it the default served version.

2. Fetch & compile at runtime

prompt = lf.get_prompt("customer-support-agent")   # fetches production version
system_message = prompt.compile(                    # fills in the variables
    customer_name=customer_name,
    subscription_plan=plan,
    current_date=...
)

Langfuse caches the prompt locally and refreshes it in the background — so there's no latency hit on every request.

3. Link the generation to the prompt version

langfuse_context.update_current_observation(prompt=prompt)

A/B testing prompts in production

import random
from datetime import datetime
from dotenv import load_dotenv
load_dotenv()

from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
from anthropic import Anthropic

lf = Langfuse()
client = Anthropic()

def ensure_prompts_exist():
    """Create control and challenger prompts if they don't exist."""
    for name, text, label in [
        (
            "support-agent",
            (
                "You are a customer support agent for AcmeCorp.\n"
                "Customer: {{customer_name}} | Plan: {{subscription_plan}} | Date: {{current_date}}\n\n"
                "Answer the customer's question helpfully and concisely."
            ),
            "production"
        ),
        (
            "support-agent-v2",
            (
                "You are an expert customer success specialist for AcmeCorp.\n"
                "Customer: {{customer_name}} | Plan: {{subscription_plan}} | Date: {{current_date}}\n\n"
                "Answer the customer's question. Always end with a proactive follow-up offer, "
                "e.g. 'Is there anything else I can help you with today?'"
            ),
            "production"
        ),
    ]:
        try:
            lf.get_prompt(name)
        except Exception:
            lf.create_prompt(name=name, type="text", prompt=text, labels=[label])
            print(f"Created prompt: {name}")

def get_prompt_for_user(user_id: str):
    """Deterministic 50/50 split based on user_id."""
    use_variant_b = hash(user_id) % 100 < 50
    if use_variant_b:
        return lf.get_prompt("support-agent-v2")   # challenger
    else:
        return lf.get_prompt("support-agent")       # control

@observe()
def handle_query(query: str, user_id: str, customer_name: str, plan: str):
    prompt = get_prompt_for_user(user_id)

    langfuse_context.update_current_trace(
        user_id=user_id,
        tags=["ab-testing"],
        metadata={"prompt_variant": prompt.name, "prompt_version": prompt.version}
    )

    system_message = prompt.compile(
        customer_name=customer_name,
        subscription_plan=plan,
        current_date=datetime.now().strftime("%Y-%m-%d")
    )

    message = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        system=system_message,
        messages=[{"role": "user", "content": query}]
    )

    response = message.content[0].text

    langfuse_context.update_current_observation(
        prompt=prompt,
        usage={
            "input": message.usage.input_tokens,
            "output": message.usage.output_tokens
        }
    )

    return response

ensure_prompts_exist()

# Simulate multiple users to show the split
users = [
    ("user_01", "Alice", "Pro"),
    ("user_02", "Bob",   "Starter"),
    ("user_03", "Carol", "Pro"),
    ("user_04", "Dan",   "Enterprise"),
]

for user_id, name, plan in users:
    variant = "B (challenger)" if hash(user_id) % 100 < 50 else "A (control)"
    print(f"\n{'='*50}")
    print(f"User: {name} | Variant: {variant}")
    print('='*50)
    result = handle_query(
        query="What happens if I downgrade my plan?",
        user_id=user_id,
        customer_name=name,
        plan=plan
    )
    print(result)

lf.flush()
# ✅ In Langfuse: filter traces by metadata.prompt_variant to compare A vs B
None None None

Filter traces by metadata.prompt_variant in Langfuse to compare average quality scores, latency, and cost between variants. No statistics PhD required.

datasets and experiment runs: catch regressions before users do

Datasets are how you turn "I think it got worse" into "it got 12% worse on question-answering and 3% better on summaries."

from dotenv import load_dotenv
load_dotenv()

from langfuse import Langfuse
from anthropic import Anthropic

lf = Langfuse()
client = Anthropic()

CONTEXT = """
AcmeCorp Support Policy:
- Returns accepted within 30 days of purchase with receipt. No exceptions after 30 days.
- To export data: go to Settings > Export > choose format > Download.
- For iOS crashes: collect device model, iOS version, and steps to reproduce. Escalate P1 if >10 users affected.
- Refunds processed in 5-7 business days after approval.
- Enterprise plans have a dedicated account manager for escalations.
"""

# ── 1. Build the golden dataset ──────────────────────────────────────────────

def ensure_dataset_exists():
    try:
        lf.get_dataset("customer-support-golden-set")
        print("Dataset already exists, skipping creation.")
        return
    except Exception:
        pass

    lf.create_dataset(
        name="customer-support-golden-set",
        description="Hand-curated QA pairs for regression testing"
    )

    test_cases = [
        {
            "input": {"question": "Can I get a refund after 60 days?"},
            "expected": {"must_contain": "30", "must_not_contain": "yes"}
        },
        {
            "input": {"question": "How do I export my data?"},
            "expected": {"must_contain": "Settings", "must_contain_2": "Export"}
        },
        {
            "input": {"question": "The app is crashing on iOS 17"},
            "expected": {"must_contain": "iOS", "asks_details": True}
        },
    ]

    for case in test_cases:
        lf.create_dataset_item(
            dataset_name="customer-support-golden-set",
            input=case["input"],
            expected_output=case["expected"]
        )
    print("Dataset created with 3 items.")

# ── 2. Pipeline under test ────────────────────────────────────────────────────

def support_pipeline(query: str, model: str) -> str:
    message = client.messages.create(
        model=model,
        max_tokens=512,
        system=f"You are a support agent. Answer using only this context:\n{CONTEXT}",
        messages=[{"role": "user", "content": query}]
    )
    return message.content[0].text

# ── 3. Rule-based scorer ──────────────────────────────────────────────────────

def evaluate_against_expected(output: str, expected: dict) -> float:
    output_lower = output.lower()
    checks = []

    if "must_contain" in expected:
        checks.append(expected["must_contain"].lower() in output_lower)
    if "must_contain_2" in expected:
        checks.append(expected["must_contain_2"].lower() in output_lower)
    if "must_not_contain" in expected:
        checks.append(expected["must_not_contain"].lower() not in output_lower)
    if expected.get("asks_details"):
        checks.append("?" in output or "version" in output_lower or "steps" in output_lower)

    return sum(checks) / len(checks) if checks else 0.0

# ── 4. Experiment runner ──────────────────────────────────────────────────────

def run_experiment(run_name: str, model: str):
    dataset = lf.get_dataset("customer-support-golden-set")
    results = []

    for item in dataset.items:
        with item.observe(run_name=run_name) as trace_id:
            lf.trace(id=trace_id, tags=["datasets-experiments"])

            output = support_pipeline(
                query=item.input["question"],
                model=model
            )
            score = evaluate_against_expected(output, item.expected_output)

            lf.score(
                trace_id=trace_id,
                name="meets_criteria",
                value=score
            )

            results.append((item.input["question"], score, output))

    avg = sum(s for _, s, _ in results) / len(results)
    print(f"\nRun '{run_name}'  |  avg score = {avg:.3f}  ({len(results)} items)")
    for question, score, _ in results:
        print(f"  [{score:.1f}] {question}")
    return avg

# ── 5. Compare two models ─────────────────────────────────────────────────────

ensure_dataset_exists()

score_sonnet = run_experiment(
    run_name="sonnet-4-6-baseline",
    model="claude-sonnet-4-6"
)
score_haiku = run_experiment(
    run_name="haiku-4-5-challenger",
    model="claude-haiku-4-5-20251001"
)

print(f"\n{'='*50}")
print(f"Sonnet 4.6  : {score_sonnet:.3f}")
print(f"Haiku 4.5   : {score_haiku:.3f}")
diff = score_haiku - score_sonnet
print(f"Difference  : {diff:+.3f}  ({'haiku better' if diff > 0 else 'sonnet better' if diff < 0 else 'tied'})")
print(f"Cost        : Haiku is ~10x cheaper than Sonnet")

lf.flush()
# ✅ Langfuse UI: Datasets > customer-support-golden-set
#    → side-by-side run comparison with per-item scores

What to look at in Langfuse go to Datasets > customer-support-golden-set:

Runs tab
├── sonnet-4-6-baseline      avg: 1.000
└── haiku-4-5-challenger     avg: 1.000
Per-item breakdown (3 rows)
├── "Can I get a refund after 60 days?"   Sonnet: 1.0 | Haiku: 1.0
├── "How do I export my data?"            Sonnet: 1.0 | Haiku: 1.0
└── "The app is crashing on iOS 17"       Sonnet: 1.0 | Haiku: 1.0
Dataset already exists, skipping creation.

Run 'sonnet-4-6-baseline'  |  avg score = 1.000  (3 items)
  [1.0] The app is crashing on iOS 17
  [1.0] How do I export my data?
  [1.0] Can I get a refund after 60 days?

Run 'haiku-4-5-challenger'  |  avg score = 1.000  (3 items)
  [1.0] The app is crashing on iOS 17
  [1.0] How do I export my data?
  [1.0] Can I get a refund after 60 days?

==================================================
Sonnet 4.6  : 1.000
Haiku 4.5   : 1.000
Difference  : +0.000  (tied)
Cost        : Haiku is ~10x cheaper than Sonnet

Each item links to its full trace so you can read the actual response. The real value shows when you change your prompt or swap models and a score drops , Langfuse flags which items regressed compared to the baseline run.

None None

step-by-step use case: building a production customer support Agent (code provided)

github project

Let's put everything together with a complete, realistic example. We'll build a customer support assistant for a SaaS product that:

  1. Retrieves relevant knowledge base articles
  2. Generates a helpful response
  3. Evaluates quality automatically
  4. Tracks costs per customer plan
None

This project simulates a real-world AI support system for a fictional company called AcmeCorp. When a customer asks a question , about refunds, data exports, billing issues, plan differences, the system automatically finds the most relevant articles from a knowledge base, feeds them to Claude as context, and generates a grounded, accurate answer. The response is returned to the user along with source citations so they know where the answer came from.

Beyond just answering questions, every interaction is fully observable: Langfuse captures the complete trace of what happened, which KB articles were retrieved, which prompt version was used, how many tokens were consumed, and how good the response was across three dimensions (helpfulness, groundedness, tone) scored automatically by a second Claude call running in the background. Users can also submit a thumbs up or down, which becomes a human score attached to the same trace.

The project also includes a quality regression system: a golden dataset of 7 representative support questions runs automatically on every pull request. If a prompt change or model swap causes quality to drop below a threshold, the CI job fails and the PR is blocked, so regressions are caught before they ever reach users.

How it works the full flow

None

Every step above is visible in Langfuse as a nested span tree with scores.

step 1: project setup

pip install langfuse anthropic fastapi uvicorn python-dotenv pinecone-client

Your .env in the project root already has all required keys:

LANGFUSE_PUBLIC_KEY=...
LANGFUSE_SECRET_KEY=...
LANGFUSE_BASE_URL=...
ANTHROPIC_API_KEY=...

Project structure

real_project/
├── config.py          ← clients + prompt bootstrap
├── retrieval.py       ← step 1: find relevant KB articles
├── generation.py      ← step 2: build prompt + call Claude
├── pipeline.py        ← orchestrator: wires steps 1+2, adds scoring
├── api.py             ← FastAPI server: /support/ask + /support/feedback
├── requirements.txt
├── tests/
│   ├── build_dataset.py  ← creates golden test dataset in Langfuse (run once)
│   └── ci_eval.py        ← regression gate: fails if quality drops below threshold
└── .github/
    └── workflows/
        └── eval.yml      ← GitHub Actions: runs ci_eval.py on every PR

Bootstrap Langfuse resources (run once , creates the prompt and golden dataset):

python -c "from config import ensure_prompt_exists; ensure_prompt_exists()"
python tests/build_dataset.py

File-by-file code explanation

config.py : clients and prompt bootstrap

from dotenv import load_dotenv
load_dotenv()

from langfuse import Langfuse
from anthropic import Anthropic

lf = Langfuse()
prod_client = Anthropic()   # traced — all production calls go through @observe decorators
eval_client = Anthropic()   # untraced — eval calls don't pollute your production traces

def ensure_prompt_exists():
    """Create the production system prompt in Langfuse if it doesn't exist."""
    try:
        lf.get_prompt("customer-support-system")
    except Exception:
        lf.create_prompt(
            name="customer-support-system",
            type="text",
            prompt=(
                "You are a helpful customer support agent for AcmeCorp.\n\n"
                "Support tier: {{support_tier}}\n"
                "Customer plan: {{customer_plan}}\n\n"
                "Knowledge base context:\n{{kb_context}}\n\n"
                "Answer the customer's question using only the knowledge base context above. "
                "Be concise, empathetic, and accurate. "
                "If the answer isn't in the context, say so and offer to escalate."
            ),
            labels=["production"],
            commit_message="Initial version"
        )
        print("[config] Prompt 'customer-support-system' created in Langfuse.")
lf = Langfuse()           # Langfuse client — reads keys from env automatically
prod_client = Anthropic() # for production API calls — traced via @observe
eval_client = Anthropic() # for eval/judge calls — intentionally untraced

Two separate Anthropic() instances even though they are identical objects. This is a convention: prod_client calls get captured in Langfuse traces because they are called inside @observe decorated functions. eval_client calls happen outside any trace context, so they never appear as spansyour eval runs don't pollute your production dashboard.

ensure_prompt_exists() creates the system prompt in Langfuse on first run. The prompt uses {{double_braces}} for variables:

"Knowledge base context:\n{{kb_context}}\n\nCustomer plan: {{customer_plan}}"

After the first run, edit this prompt directly in the Langfuse UI under Prompts. The next API call picks up the new version instantly, no code change needed.

retrieval.py : the KB retrieval layer

@observe(name="kb_retrieval")
def retrieve_knowledge_base(query: str, top_k: int = 3) -> list[dict]:
    docs = _keyword_search(query, top_k)
    langfuse_context.update_current_observation(
        metadata={
            "num_results": len(docs),
            "top_similarity_score": docs[0]["score"] if docs else 0,
            ...
        },
        output={"retrieved_titles": [d["title"] for d in docs]}
    )
    return docs

@observe(name="kb_retrieval") turns this function into a child span inside whatever parent trace is active. Langfuse automatically links it to the support_pipeline span from pipeline.py because they run in the same call stack.

langfuse_context.update_current_observation(...) attaches structured data to this specific span not the root trace. In Langfuse you'll see the retrieval span with the metadata and output logged next to it, which lets you debug poor retrieval quality directly from a trace.

_keyword_search() is the mock vector DB. It does simple keyword overlap scoring. To use a real vector DB (Pinecone, pgvector, Weaviate), replace only _keyword_search() everything else stays the same:

def _keyword_search(query, top_k):
    embedding = embed_client.embed(query)
    results = pinecone_index.query(vector=embedding, top_k=top_k)
    return [{"title": r.metadata["title"], "content": r.metadata["content"],
             "url": r.metadata["url"], "score": r.score} for r in results.matches]

generation.py: prompt management and Claude call

def build_system_prompt(docs, customer_plan):
    prompt_template = lf.get_prompt("customer-support-system")   # fetches from Langfuse
    compiled = prompt_template.compile(
        kb_context=kb_context,
        customer_plan=customer_plan,
        support_tier="priority" if customer_plan == "enterprise" else "standard"
    )
    return compiled, prompt_template

lf.get_prompt() fetches the prompt labeled production from Langfuse. It is cached locally and refreshed in the background, so there is no latency hit on every request. The returned prompt_template object tracks which version was fetched.

.compile(...) fills in the {{variables}} and returns a plain string ready to send to Claude.

@observe(name="response_generation")
def generate_support_response(query, docs, customer_plan):
    ...
    message = prod_client.messages.create(
        model=model,
        system=system_prompt,
        messages=[{"role": "user", "content": query}]
    )
    langfuse_context.update_current_observation(
        prompt=prompt_obj,      # ← links this span to the Langfuse prompt version
        model=model,
        usage={"input": message.usage.input_tokens, "output": message.usage.output_tokens}
    )

Passing prompt=prompt_obj to update_current_observation is what links the generation span to the exact prompt version that was used. In Langfuse you can then filter traces by prompt version and see exactly which version produced which response — essential when you roll out a new prompt.

Model routing — haiku for free/pro, sonnet for enterprise:

model = "claude-sonnet-4-6" if customer_plan == "enterprise" else "claude-haiku-4-5-20251001"

Haiku is ~10x cheaper and fast enough for standard cases. Enterprise customers get Sonnet for higher quality.

pipeline.py — the orchestrator

This is the main function that wires everything together and owns the root trace.

@observe(name="support_pipeline")
def handle_support_query(query, customer_id, customer_plan, session_id):

Because this is the outermost @observe, it becomes the root trace in Langfuse. All nested @observe calls (retrieval, generation) automatically become child spans under it.

Step 1 — attach customer context to the root trace:

langfuse_context.update_current_trace(
    user_id=customer_id,
    session_id=session_id,
    tags=[f"plan:{customer_plan}", "channel:support-widget", "env:production"],
)

update_current_trace targets the root trace, not the current span. This is different from update_current_observation which targets the current span. Tags and user_id set here appear at the top level in Langfuse and are filterable across all traces.

Step 2 — rule-based scores (synchronous, zero cost):

langfuse_context.score_current_trace(name="response_length", value=len(response.split()))
langfuse_context.score_current_trace(name="refusal_detected", value=1 if has_refusal else 0)
langfuse_context.score_current_trace(name="cites_kb_article", value=1 if cites_kb else 0)

These run instantly with no extra API call. They give you immediate, cheap signals on every trace.

Step 3 — async LLM judge (non-blocking):

trace_id = langfuse_context.get_current_trace_id()
threading.Thread(target=_run_async_eval, args=(trace_id, ...), daemon=True).start()

get_current_trace_id() captures the trace ID before the function returns. The background thread uses lf.score(trace_id=..., ...) to attach scores to that trace after Claude finishes evaluating — typically 2–3 seconds after the user already got their response. The user never waits for the eval.

Why two separate Anthropic clients for eval? _run_async_eval uses eval_client which is called outside any @observe context. This means those Haiku calls create no spans in Langfuse. If you used prod_client instead, every eval call would appear as a span inside your production traces and pollute your dashboards.

api.py — the FastAPI server

Two endpoints:

POST /support/ask — calls handle_support_query and returns:

{
  "response": "To export your data, go to Settings > Export...",
  "sources": [{"title": "Data Export Guide", "url": "/help/export"}],
  "trace_id": "abc123..."
}

The trace_id is returned to the frontend intentionally — so the UI can attach user feedback to the exact trace that produced this response.

POST /support/feedback — receives the trace_id and a thumbs up/down:

lf.score(
    trace_id=req.trace_id,
    name="user_satisfaction",
    value=1 if req.thumbs_up else 0,
    data_type="BOOLEAN"
)

This calls the Langfuse API directly (not inside a trace context) to attach a human score to a trace that already exists. In Langfuse you can then filter traces by user_satisfaction = 0 to find responses users disliked and investigate what went wrong.

tests/build_dataset.py golden dataset

Creates a dataset in Langfuse called support-bot-golden-v1 with 7 hand-curated test cases: standard policy questions plus historical edge cases (angry customer, API ToS question, etc.).

Run this once:

python tests/build_dataset.py

After that, go to Datasets in Langfuse to see the items. You can add more from the UI or from code as you discover new edge cases in production.

tests/ci_eval.py the regression gate

This is the key to catching regressions before users do. It runs every item in the golden dataset through the live pipeline, waits for the LLM-judge scores, and checks them against a threshold:

THRESHOLD = 0.75   # fail if any metric average drops below this
for item in dataset.items:
    with item.observe(run_name=RUN_NAME) as trace_id:   # links trace to dataset run
        result = handle_support_query(query=item.input["query"], ...)

item.observe(run_name=...) creates a trace and links it to the dataset run in Langfuse. After the eval run, go to Datasets → support-bot-golden-v1 → Runs and you see a side-by-side table: each run as a column, each test case as a row, with scores in every cell. This is how you compare two prompt versions or two models systematically.

The script exits with code 1 if any metric fails — which fails the GitHub Actions job and blocks the PR merge.

.github/workflows/eval.yml CI gate

Triggers on any PR that modifies files in real_project/. Runs ci_eval.py with production API keys from GitHub secrets. If scores regress, the job fails and the PR cannot merge.

Add these secrets to your GitHub repo settings:

  • LANGFUSE_PUBLIC_KEY
  • LANGFUSE_SECRET_KEY
  • LANGFUSE_BASE_URL
  • ANTHROPIC_API_KEY

Running locally

Start the server:

uvicorn api:app --reload --port 8000

Send a test query:

curl -X POST http://localhost:8000/support/ask \
  -H "Content-Type: application/json" \
  -d '{"query": "How do I export my data?", "customer_id": "user_42", "customer_plan": "pro", "session_id": "sess_1"}'

Submit feedback:

curl -X POST http://localhost:8000/support/feedback \
  -H "Content-Type: application/json" \
  -d '{"trace_id": "<id from above>", "thumbs_up": true}'

Run the regression suite:

python tests/ci_eval.py

Langfuse concepts used and where

None

github project

interesting resources and further reading

official documentation and community

going deeper on LLM evaluation

  • RAGAS — framework for RAG-specific evaluation metrics (faithfulness, answer relevancy, context precision) — integrates natively with Langfuse
  • DeepEval — pytest-style unit testing for LLMs, works alongside Langfuse
  • Evals by OpenAI — OpenAI's framework for model evaluation, useful for understanding eval patterns
  • "Your AI Product Needs Evals" — essential blog post by Hamel Husain, the best practical guide to LLM evaluation thinking

LLM ops and production patterns

prompt engineering

  • DSPy — Stanford's framework for programmatically optimizing prompts (instead of writing them manually) — use with Langfuse for observability
  • Instructor — structured outputs from LLMs using Pydantic, integrates cleanly with Langfuse tracing
  • LiteLLM — unified interface for 100+ LLM providers, has native Langfuse integration