:::tip 🎮 Interactive Playground Visualize this concept: Try the LLM Observability demo on the EngineersOfAI Playground - no code required. :::

Langfuse - Open-Source LLM Observability

The Compliance Blocker

Your team has been using LangSmith for three months and loves it. Trace visibility is excellent. The evaluation pipeline is catching regressions. Then legal gets involved.

You are building a healthcare AI product - a clinical documentation assistant for hospital systems. The legal team has two words: "data residency." Every patient encounter, every physician note, every clinical query must stay within the European Union. LangSmith's cloud offering processes data on US servers. GDPR Article 44 prohibits transferring personal data to the US without appropriate safeguards. Your DPO (Data Protection Officer) says the current setup is non-compliant.

The alternatives are stark: rebuild your observability stack from scratch (estimated three to four engineering weeks and ongoing maintenance), purchase LangSmith Enterprise with a custom data processing agreement and EU-region hosting (significant cost and negotiation time), or deploy Langfuse - an MIT-licensed, self-hostable LLM observability platform - on a Frankfurt-region server this afternoon.

You spin up a Hetzner CPX42 in Germany, run docker-compose up, and within two hours you have full trace visibility on your clinical documentation assistant. All data stays within the EU. You own the storage. Your DPO signs off. The total infrastructure cost: approximately €180/month for the server and managed PostgreSQL. Zero per-seat fees, zero per-trace fees, and complete auditability of the entire data pipeline.

This is Langfuse's primary value proposition: the open-source, self-hostable observability platform for teams where data sovereignty, cost predictability, or vendor independence are non-negotiable requirements.

What Langfuse Is

Langfuse is an open-source LLM engineering platform created in 2023 by Max Deichmann and Marc Klingen. The core is MIT-licensed. It covers:

• Tracing: Full input/output capture for every LLM call, with token usage and computed costs
• Scores: Quality scores from automated evaluators, rule-based checks, or human reviewers
• Datasets: Versioned evaluation datasets with support for running experiments
• Prompt Management: Version-controlled prompts with labels (production, staging) pulled at runtime
• Cost Analytics: Per-model, per-user, per-feature, per-team cost breakdowns
• User Analytics: Segment traces by user ID, session, custom metadata for product analytics

Langfuse vs LangSmith - When to Choose Each

Feature	Langfuse	LangSmith
Open Source	Yes (MIT)	No
Self-Hosted	Free (Docker / Kubernetes)	Enterprise paid tier
Data Residency	Full control - your infrastructure	US cloud by default
Pricing	Free self-hosted; $30/mo cloud	~$39/user/month and up
Framework	Framework-agnostic	Optimized for LangChain
Auto-tracing	Decorator + manual SDK	LangChain env var (zero-config)
Prompt Hub	Yes (labels + versioning)	Yes (Prompt Hub)
Annotation Queues	Yes	Yes
Embedding Visualization	No	No (use Phoenix)

Choose Langfuse when: data residency requirements, cost at scale, vendor independence, or a non-LangChain stack.

Choose LangSmith when: deeply invested in LangChain, prefer managed SaaS, or need tight ecosystem integration with LangChain tooling.

Langfuse Architecture

Langfuse stores traces in PostgreSQL (relational) and aggregates analytics in ClickHouse (columnar). ClickHouse is optional but essential for production performance above ~100K traces - without it, analytics queries hit PostgreSQL directly and become slow at scale. Deploy ClickHouse from day one; migrating later requires a data backfill.

Self-Hosting: Docker Compose Setup

# docker-compose.yml - Production Langfuse deployment
version: "3.8"

services:
  langfuse-server:
    image: langfuse/langfuse:2.44.0   # ALWAYS pin the version - never use :latest
    depends_on:
      postgres:
        condition: service_healthy
    ports:
      - "3000:3000"
    environment:
      DATABASE_URL:     "postgresql://langfuse:${POSTGRES_PASSWORD}@postgres:5432/langfuse"
      NEXTAUTH_URL:     "${LANGFUSE_HOST}"
      NEXTAUTH_SECRET:  "${NEXTAUTH_SECRET}"   # generate: openssl rand -hex 32
      SALT:             "${LANGFUSE_SALT}"      # generate: openssl rand -hex 32
      # Optional: enable SSO
      # AUTH_GOOGLE_CLIENT_ID: "${GOOGLE_CLIENT_ID}"
      # AUTH_GOOGLE_CLIENT_SECRET: "${GOOGLE_CLIENT_SECRET}"
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:3000/api/public/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  postgres:
    image: postgres:15-alpine
    environment:
      POSTGRES_USER:     langfuse
      POSTGRES_PASSWORD: "${POSTGRES_PASSWORD}"
      POSTGRES_DB:       langfuse
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U langfuse"]
      interval: 10s
      timeout: 5s
      retries: 5
    restart: unless-stopped

volumes:
  postgres_data:

# Initial setup
cp .env.example .env
# Fill in POSTGRES_PASSWORD, NEXTAUTH_SECRET, LANGFUSE_SALT, LANGFUSE_HOST

docker-compose up -d
# UI available at http://localhost:3000

# Create your first account via the UI
# Go to Settings → API Keys → Create new key
# Copy: LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY

Production Kubernetes Deployment

# helm/values.yaml for Langfuse Helm chart
langfuse:
  replicaCount: 3        # 3 replicas for high availability
  image:
    tag: "2.44.0"       # pin version in production

  resources:
    requests:
      cpu:    "500m"
      memory: "512Mi"
    limits:
      cpu:    "2000m"
      memory: "2Gi"

  env:
    DATABASE_URL:
      secretKeyRef:
        name: langfuse-secrets
        key:  database-url
    NEXTAUTH_SECRET:
      secretKeyRef:
        name: langfuse-secrets
        key:  nextauth-secret
    SALT:
      secretKeyRef:
        name: langfuse-secrets
        key:  salt

  ingress:
    enabled: true
    annotations:
      kubernetes.io/ingress.class: nginx
      cert-manager.io/cluster-issuer: letsencrypt-prod
    hosts:
      - host: langfuse.yourcompany.com
        paths: ["/"]
    tls:
      - secretName: langfuse-tls
        hosts:       ["langfuse.yourcompany.com"]

  # PostgreSQL (use managed RDS/Cloud SQL in production)
  postgresql:
    enabled: false   # use external managed DB

  externalDatabase:
    host:     rds-langfuse.cluster.eu-west-1.rds.amazonaws.com
    port:     5432
    database: langfuse
    existingSecret: langfuse-db-secret

SDK Setup and Configuration

pip install langfuse anthropic

# observability/langfuse_client.py
import os
from langfuse import Langfuse
import anthropic

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com"),
    # For self-hosted: host="https://langfuse.yourcompany.com"
    #
    # Performance tuning for high-throughput applications:
    # flush_at=50,           # batch size before flushing (default: 15)
    # flush_interval=0.5,    # max seconds between flushes (default: 0.5)
    # max_retries=3,         # retries on network failure
    timeout=10,
)

client = anthropic.Anthropic()

Tracing: Three Levels of the Hierarchy

Langfuse's tracing model has three levels that map to what actually happens in an LLM pipeline:

Trace is the root object - one end-to-end user interaction with user_id, session_id, and overall inputs/outputs.

Span represents a non-LLM step: vector retrieval, database query, API call to an external service. Use spans to measure where latency comes from. If a trace takes 3 seconds, spans might reveal it is 2.8s in vector search and 0.2s in LLM synthesis - the bottleneck is retrieval, not generation.

Generation is a specialized span for LLM calls - it additionally records model name, parameters, token usage (input + output), and calculated cost (Langfuse maps model name to per-token pricing automatically).

Low-Level API (Maximum Control)

# tracing/manual_tracing.py
import anthropic
import time
from langfuse import Langfuse
from datetime import datetime

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)
client = anthropic.Anthropic()


def answer_question_with_tracing(
    user_id: str,
    session_id: str,
    query: str,
    feature: str = "qa-bot",
) -> dict:
    """
    Full manual tracing with all three levels: Trace, Span, Generation.
    Use this when you need maximum control over what gets recorded.
    """
    # Create the root trace
    trace = langfuse.trace(
        name="question-answering",
        user_id=user_id,
        session_id=session_id,
        input={"query": query},
        metadata={
            "feature":  feature,
            "version":  "2.1",
            "env":      os.getenv("ENVIRONMENT", "production"),
        },
        tags=["rag", "production"],
    )

    try:
        # ── Step 1: Vector retrieval (Span) ───────────────────────────────
        retrieval_span = trace.span(
            name="vector-retrieval",
            input={"query": query, "k": 5},
        )
        start_retrieval = time.monotonic()

        # Simulate retrieval (replace with your vector DB call)
        retrieved_docs = [
            {"content": "Doc about topic A...", "source": "kb-v3", "score": 0.91},
            {"content": "Doc about topic B...", "source": "kb-v3", "score": 0.87},
        ]
        retrieval_latency = time.monotonic() - start_retrieval

        retrieval_span.end(
            output={
                "num_docs":     len(retrieved_docs),
                "avg_score":    0.89,
                "latency_ms":   round(retrieval_latency * 1000, 2),
            }
        )

        # ── Step 2: LLM call (Generation) ────────────────────────────────
        context_text = "\n\n".join(d["content"] for d in retrieved_docs)
        messages = [{
            "role": "user",
            "content": f"Context:\n{context_text}\n\nQuestion: {query}"
        }]

        generation = trace.generation(
            name="synthesis",
            model="claude-opus-4-6",
            model_parameters={"temperature": 0.0, "max_tokens": 1024},
            input=messages,
            metadata={"prompt_version": "v2.3"},
        )

        response = client.messages.create(
            model="claude-opus-4-6",
            max_tokens=1024,
            temperature=0.0,
            system=(
                "Answer using only the provided context. "
                "If the answer isn't in the context, say so explicitly."
            ),
            messages=messages,
        )
        answer = response.content[0].text

        generation.end(
            output=answer,
            usage={
                "input":  response.usage.input_tokens,
                "output": response.usage.output_tokens,
                "unit":   "TOKENS",
            },
            level="DEFAULT",
        )

        # ── Update the root trace with final output ───────────────────────
        trace.update(
            output={"answer": answer},
            metadata={
                "total_input_tokens":  response.usage.input_tokens,
                "total_output_tokens": response.usage.output_tokens,
            },
        )

        # CRITICAL: flush in scripts/serverless - background thread may not flush before exit
        langfuse.flush()

        return {
            "answer":   answer,
            "trace_id": trace.id,
        }

    except Exception as e:
        trace.update(
            output={"error": str(e)},
            level="ERROR",
        )
        langfuse.flush()
        raise

High-Level @observe Decorator (Recommended)

The @observe decorator is the recommended approach for most applications - it handles the span/generation nesting automatically:

# tracing/decorator_tracing.py
import anthropic
import json
from langfuse.decorators import langfuse_context, observe
from langfuse import Langfuse

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)
client = anthropic.Anthropic()


@observe(name="embed-query")
def embed_query(query: str) -> list[float]:
    """Embedding step - auto-traced as a span."""
    langfuse_context.update_current_observation(
        metadata={"embedding_model": "voyage-3", "query_length": len(query)}
    )
    # In production: call your embedding API
    return [0.1] * 1024


@observe(name="retrieve-context")
def retrieve_context(query: str, k: int = 5) -> list[dict]:
    """Retrieval step - auto-traced as a span."""
    embedding = embed_query(query)  # auto-nested as child span

    # In production: query your vector store
    docs = [
        {"content": f"Relevant passage about {query[:30]}...", "score": 0.88}
        for _ in range(k)
    ]

    langfuse_context.update_current_observation(
        input={"query": query, "k": k},
        output={"num_docs": len(docs), "avg_score": 0.88},
        metadata={"index": "knowledge-base-v3"},
    )
    return docs


@observe(name="synthesize-answer")
def synthesize_answer(query: str, context: list[dict]) -> str:
    """Synthesis step - auto-traced as a generation."""
    context_text = "\n\n".join(d["content"] for d in context)
    messages = [{
        "role": "user",
        "content": f"Context:\n{context_text}\n\nQuestion: {query}"
    }]

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        temperature=0.0,
        system="Answer using only the provided context. Be precise.",
        messages=messages,
    )
    answer = response.content[0].text

    # Record token usage so Langfuse calculates cost
    langfuse_context.update_current_observation(
        model="claude-opus-4-6",
        model_parameters={"temperature": 0.0, "max_tokens": 1024},
        input=messages,
        output=answer,
        usage={
            "input":  response.usage.input_tokens,
            "output": response.usage.output_tokens,
        },
    )
    return answer


@observe(name="rag-pipeline")
def rag_pipeline(user_id: str, session_id: str, query: str) -> dict:
    """
    Root pipeline - creates the parent trace.
    Children (retrieve-context, synthesize-answer) are auto-nested.
    """
    # Tag the root trace with user and session context
    langfuse_context.update_current_trace(
        user_id=user_id,
        session_id=session_id,
        tags=["rag", "production"],
        metadata={"feature": "document-qa"},
    )

    context = retrieve_context(query)    # creates nested child spans
    answer  = synthesize_answer(query, context)  # creates nested generation

    # Update root trace output
    langfuse_context.update_current_trace(
        input={"query": query},
        output={"answer": answer},
    )

    trace_id = langfuse_context.get_current_trace_id()

    return {
        "answer":   answer,
        "trace_id": trace_id,
    }

Scoring and Feedback Integration

# feedback/langfuse_feedback.py
import anthropic
import json
import re
from langfuse.decorators import langfuse_context, observe
from langfuse import Langfuse
from fastapi import FastAPI
from pydantic import BaseModel

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)
client = anthropic.Anthropic()
app = FastAPI()


@observe(name="scored-generation")
def generate_with_auto_scoring(query: str, user_id: str) -> dict:
    """Generate a response and attach an automated quality score."""
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        temperature=0.0,
        messages=[{"role": "user", "content": query}]
    )
    answer = response.content[0].text

    langfuse_context.update_current_observation(
        model="claude-opus-4-6",
        usage={"input": response.usage.input_tokens, "output": response.usage.output_tokens},
    )

    trace_id = langfuse_context.get_current_trace_id()

    # ── Rule-based auto-scores (zero cost, instant) ───────────────────────
    deflection_phrases = ["check our FAQ", "see our documentation", "visit our website"]
    is_deflecting = any(p in answer.lower() for p in deflection_phrases)

    langfuse.score(
        trace_id=trace_id,
        name="no_deflection",
        value=0.0 if is_deflecting else 1.0,
        comment="Deflection detected" if is_deflecting else "No deflection",
    )

    # Response completeness check
    is_complete = (
        len(answer) > 50
        and not answer.strip().endswith("?")
        and answer.count(".") >= 2
    )
    langfuse.score(
        trace_id=trace_id,
        name="response_completeness",
        value=1.0 if is_complete else 0.5,
    )

    return {"answer": answer, "trace_id": trace_id}


class UserFeedback(BaseModel):
    trace_id: str
    rating: int      # 1-5 stars
    comment: str | None = None
    correction: str | None = None


@app.post("/api/feedback")
async def submit_user_feedback(feedback: UserFeedback):
    """Record explicit user feedback to a Langfuse trace."""
    normalized_score = (feedback.rating - 1) / 4  # convert 1-5 to 0.0-1.0

    langfuse.score(
        trace_id=feedback.trace_id,
        name="user_rating",
        value=normalized_score,
        comment=feedback.comment,
        data_type="NUMERIC",
    )

    # If user provided a correction, record it separately as high-value signal
    if feedback.correction:
        langfuse.score(
            trace_id=feedback.trace_id,
            name="user_correction",
            value=0.0,  # original was wrong
            comment=f"User correction: {feedback.correction[:500]}",
        )

    return {"status": "recorded"}


# ── Async LLM-as-judge evaluation ────────────────────────────────────────────

def run_llm_judge_evaluation(
    trace_id: str,
    question: str,
    answer: str,
    context: str,
) -> None:
    """
    Run LLM-as-judge evaluation asynchronously and attach scores.
    Call this from a background worker, not the critical path.
    """
    judge_prompt = f"""Evaluate this AI response on two dimensions.

Question: {question}
Context provided to the AI: {context[:800]}
AI Response: {answer}

Rate each dimension 0.0-1.0:
1. faithfulness: Does the response stick to the provided context? (1.0 = fully grounded, 0.0 = hallucinated)
2. relevance: Does the response address the question directly? (1.0 = fully relevant, 0.0 = misses the question)

Return JSON only: {{"faithfulness": 0.0-1.0, "relevance": 0.0-1.0, "faithfulness_reason": "...", "relevance_reason": "..."}}"""

    try:
        response = client.messages.create(
            model="claude-haiku-4-5-20251001",  # cheap model for evaluation
            max_tokens=200,
            temperature=0.0,
            messages=[{"role": "user", "content": judge_prompt}]
        )

        raw = response.content[0].text.strip()
        raw = re.sub(r"```json\n?|\n?```", "", raw).strip()
        scores = json.loads(raw)

        # Attach both scores to the trace
        for metric in ["faithfulness", "relevance"]:
            if metric in scores:
                langfuse.score(
                    trace_id=trace_id,
                    name=metric,
                    value=float(scores[metric]),
                    comment=scores.get(f"{metric}_reason", ""),
                )

    except (json.JSONDecodeError, KeyError, Exception) as e:
        # Evaluation failure should never affect the user's response
        print(f"Judge evaluation failed for trace {trace_id}: {e}")

Prompt Management

Langfuse's prompt management decouples prompt changes from code deployments - a change in the prompt does not require a code deployment:

# prompts/langfuse_prompts.py
import os
from langfuse import Langfuse
from langfuse.decorators import langfuse_context, observe
import anthropic

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)
client = anthropic.Anthropic()


# ── Creating and versioning prompts ──────────────────────────────────────────

def create_or_update_prompt(
    name: str,
    template: str,
    config: dict,
    labels: list[str] = None,
) -> None:
    """
    Create a new prompt version. Each call creates a new version.
    The 'production' label controls which version is served to your app.
    """
    langfuse.create_prompt(
        name=name,
        prompt=template,
        config=config,
        labels=labels or ["staging"],   # start in staging, promote to production manually
    )
    print(f"Created new version of prompt '{name}'")


# Promote a specific version to production
def promote_to_production(name: str, version_id: int) -> None:
    """Move a tested prompt version to production label."""
    langfuse.update_prompt(
        name=name,
        version=version_id,
        new_labels=["production"],
    )
    print(f"Promoted version {version_id} of '{name}' to production")


# ── Reading prompts at runtime ────────────────────────────────────────────────

def get_compiled_prompt(
    name: str,
    label: str = "production",
    **variables,
) -> str:
    """
    Pull a prompt by label and compile with template variables.
    Raises if prompt or label doesn't exist.
    """
    prompt_obj = langfuse.get_prompt(name, label=label)
    return prompt_obj.compile(**variables)


# ── Using prompts with observation linking ────────────────────────────────────

@observe(name="prompted-generation")
def generate_with_versioned_prompt(
    query: str,
    company: str,
    customer_tier: str,
) -> str:
    """
    Pull prompt from Langfuse and link this generation to the prompt version.
    Langfuse will track quality metrics per prompt version automatically.
    """
    # Pull the production prompt
    prompt_obj = langfuse.get_prompt(
        "customer-support-v3",
        label="production",
    )

    # Compile with template variables
    system_prompt = prompt_obj.compile(
        company=company,
        tier=customer_tier,
    )

    # Link this generation to the prompt version
    # Langfuse can now show you quality metrics broken down by prompt version
    langfuse_context.update_current_observation(prompt=prompt_obj)

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        temperature=0.0,
        system=system_prompt,
        messages=[{"role": "user", "content": query}],
    )
    return response.content[0].text


# Example: define prompt template with {{mustache}} variables
SUPPORT_PROMPT_TEMPLATE = """You are a customer support agent for {{company}}.
Customer tier: {{tier}}

Rules:
- Never say "check our FAQ" or "visit our website"
- For {{tier}} customers, prioritize speed and direct action
- Always provide a concrete next step or timeline
- If you cannot resolve immediately, escalate clearly"""

# Create the prompt (run once to initialize):
# create_or_update_prompt(
#     name="customer-support-v3",
#     template=SUPPORT_PROMPT_TEMPLATE,
#     config={"model": "claude-opus-4-6", "max_tokens": 1024},
#     labels=["staging"],
# )

Datasets and Evaluation

# evals/langfuse_evals.py
import anthropic
import json
from langfuse import Langfuse
from datetime import datetime

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)
client = anthropic.Anthropic()


def setup_golden_dataset(dataset_name: str) -> None:
    """Create a golden evaluation dataset with hand-curated examples."""
    try:
        langfuse.get_dataset(dataset_name)
        print(f"Dataset '{dataset_name}' already exists.")
        return
    except Exception:
        pass

    langfuse.create_dataset(
        name=dataset_name,
        description="Golden evaluation set - curated by engineering team.",
    )

    examples = [
        {
            "input":    {"query": "How do I cancel my subscription?", "tier": "premium"},
            "expected": {"must_contain": ["cancel", "confirmation"], "max_deflections": 0},
        },
        {
            "input":    {"query": "I was charged twice this month.", "tier": "enterprise"},
            "expected": {"must_contain": ["apologize", "refund"], "tone": "urgent"},
        },
        {
            "input":    {"query": "Can I download my invoice?", "tier": "standard"},
            "expected": {"must_contain": ["Settings", "Billing", "Download"], "max_deflections": 0},
        },
    ]

    for ex in examples:
        langfuse.create_dataset_item(
            dataset_name=dataset_name,
            input=ex["input"],
            expected_output=ex["expected"],
        )

    print(f"Created dataset '{dataset_name}' with {len(examples)} examples.")


def run_evaluation_experiment(
    dataset_name: str,
    experiment_name: str,
    model_version: str = "claude-opus-4-6",
) -> dict:
    """
    Run an evaluation experiment against a Langfuse dataset.
    Each item in the dataset is run through the model and scored.
    Results are stored in Langfuse and viewable in the Experiments UI.
    """
    dataset = langfuse.get_dataset(dataset_name)
    results = []

    for item in dataset.items:
        # item.observe() creates a trace linked to this dataset item
        with item.observe(run_name=experiment_name) as root_trace:
            query  = item.input.get("query", "")
            tier   = item.input.get("tier", "standard")
            expected = item.expected_output or {}

            # Run the model
            response = client.messages.create(
                model=model_version,
                max_tokens=512,
                temperature=0.0,
                system=(
                    f"You are customer support. Tier: {tier}. "
                    "Never deflect to FAQ or website. Own the problem."
                ),
                messages=[{"role": "user", "content": query}],
            )
            answer = response.content[0].text

            # ── Score on multiple metrics ──────────────────────────────────

            # Metric 1: keyword coverage
            must_contain = expected.get("must_contain", [])
            covered = sum(1 for kw in must_contain if kw.lower() in answer.lower())
            keyword_score = covered / len(must_contain) if must_contain else 1.0

            root_trace.score(
                name="keyword_coverage",
                value=keyword_score,
                comment=f"{covered}/{len(must_contain)} required keywords",
            )

            # Metric 2: no deflection
            deflection_phrases = ["check our FAQ", "see our documentation", "visit our website"]
            has_deflection = any(p in answer.lower() for p in deflection_phrases)
            max_deflections = expected.get("max_deflections", 0)

            root_trace.score(
                name="no_deflection",
                value=0.0 if has_deflection else 1.0,
                comment="Deflection detected" if has_deflection else "Clean",
            )

            # Metric 3: LLM quality judge
            judge_response = client.messages.create(
                model="claude-haiku-4-5-20251001",
                max_tokens=100,
                temperature=0.0,
                messages=[{
                    "role": "user",
                    "content": (
                        f"Rate this support response 0.0-1.0 for helpfulness.\n"
                        f"Query: {query}\nResponse: {answer}\n"
                        f"Return JSON: {{\"score\": 0.0-1.0}}"
                    )
                }]
            )
            try:
                judge_score = float(json.loads(judge_response.content[0].text).get("score", 0.5))
            except (json.JSONDecodeError, ValueError):
                judge_score = 0.5

            root_trace.score(
                name="llm_quality_judge",
                value=judge_score,
            )

            results.append({
                "query":         query,
                "answer":        answer[:100],
                "keyword_score": keyword_score,
                "no_deflection": 0.0 if has_deflection else 1.0,
                "judge_score":   judge_score,
            })

    # Compute aggregate summary
    n = len(results)
    summary = {
        "n_examples":    n,
        "keyword_coverage_mean": sum(r["keyword_score"] for r in results) / n if n else 0,
        "no_deflection_mean":    sum(r["no_deflection"] for r in results) / n if n else 0,
        "judge_score_mean":      sum(r["judge_score"]   for r in results) / n if n else 0,
        "experiment_name":       experiment_name,
    }

    print(f"\nExperiment '{experiment_name}' complete:")
    for k, v in summary.items():
        if isinstance(v, float):
            print(f"  {k}: {v:.4f}")
        else:
            print(f"  {k}: {v}")

    return summary

Cost Attribution by Feature and Team

One of Langfuse's most operationally useful features is cost attribution. With structured metadata on every trace, you get monthly cost reports broken down by product area:

# analytics/cost_attribution.py
import os
from langfuse.decorators import langfuse_context, observe
from langfuse import Langfuse
from datetime import datetime, timedelta

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)


@observe(name="ai-search")
def handle_ai_search(user_id: str, query: str, team: str = "discovery") -> list:
    """Tag traces with feature and team for cost attribution."""
    langfuse_context.update_current_trace(
        user_id=user_id,
        metadata={"feature": "ai-search", "team": team, "product": "search-v2"},
    )
    # ... perform search
    return []


@observe(name="document-summary")
def handle_document_summary(user_id: str, doc_id: str) -> str:
    """Different feature - tagged separately for cost breakdown."""
    langfuse_context.update_current_trace(
        user_id=user_id,
        metadata={"feature": "doc-summary", "team": "content", "doc_id": doc_id},
    )
    # ... summarize document
    return ""


def generate_cost_report(days: int = 30) -> dict[str, float]:
    """
    Generate a cost report broken down by feature.
    Requires ClickHouse for large trace volumes.
    """
    traces = langfuse.fetch_traces(
        from_timestamp=datetime.now() - timedelta(days=days),
        limit=5000,    # increase for higher volume
    )

    costs_by_feature: dict[str, float] = {}
    costs_by_team:    dict[str, float] = {}

    for trace in traces.data:
        metadata = trace.metadata or {}
        feature  = metadata.get("feature", "unknown")
        team     = metadata.get("team", "unknown")

        # Sum calculated_cost across all observations in this trace
        trace_cost = 0.0
        if hasattr(trace, "observations") and trace.observations:
            for obs in trace.observations:
                if hasattr(obs, "calculated_cost") and obs.calculated_cost:
                    trace_cost += obs.calculated_cost

        costs_by_feature[feature] = costs_by_feature.get(feature, 0.0) + trace_cost
        costs_by_team[team]       = costs_by_team.get(team, 0.0)       + trace_cost

    total_cost = sum(costs_by_feature.values())

    report = {
        "period_days":      days,
        "total_cost_usd":   round(total_cost, 4),
        "by_feature":       {k: round(v, 4) for k, v in sorted(costs_by_feature.items(), key=lambda x: -x[1])},
        "by_team":          {k: round(v, 4) for k, v in sorted(costs_by_team.items(), key=lambda x: -x[1])},
        "generated_at":     datetime.now().isoformat(),
    }

    return report

Production Operations

Retention and Storage Management

At 100K traces/day with full content capture, PostgreSQL grows by approximately 2-5 GB/day depending on prompt length. Plan your storage:

# Check current Langfuse database size
docker-compose exec postgres psql -U langfuse -c "
SELECT
    pg_size_pretty(pg_database_size('langfuse')) as total_size,
    pg_size_pretty(pg_total_relation_size('observations')) as observations_size,
    pg_size_pretty(pg_total_relation_size('traces')) as traces_size;
"

# Set up automated cleanup (runs as a cron job or scheduled task)

# ops/retention.py
from langfuse import Langfuse
from datetime import datetime, timedelta

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)


def delete_old_traces(retention_days: int = 90) -> int:
    """
    Delete traces older than retention_days.
    Run as a weekly cron job to control storage growth.
    """
    cutoff = datetime.now() - timedelta(days=retention_days)

    old_traces = langfuse.fetch_traces(
        to_timestamp=cutoff,
        limit=1000,
    )

    deleted_count = 0
    for trace in old_traces.data:
        try:
            langfuse.delete_trace(trace.id)
            deleted_count += 1
        except Exception as e:
            print(f"Could not delete trace {trace.id}: {e}")

    print(f"Deleted {deleted_count} traces older than {retention_days} days")
    return deleted_count

Common Mistakes

:::danger Always call langfuse.flush() in scripts and Lambda functions The background upload thread may not flush before process exit in short-lived processes. In AWS Lambda, Google Cloud Run jobs, or any script that exits after completing work, add langfuse.flush() as the final line before return or program exit. Failing to flush means your traces are silently dropped - your traces look empty in the UI and you think the SDK isn't working, when actually the process exited before the buffer flushed. :::

:::warning Deploy ClickHouse from the start for production scale Without ClickHouse, analytics queries go directly to PostgreSQL. Above ~100K traces, dashboard queries become slow (10-30 seconds). Above ~1M traces, some queries time out entirely. Deploy ClickHouse from day one - migrating later requires setting up ClickHouse, running a data migration script, and verifying all data transferred correctly. This is a painful retroactive fix. A single ClickHouse node (4 CPU, 8 GB RAM) handles tens of millions of traces. :::

:::danger Never hardcode API keys in source code

# BAD - key in source code, will be committed to git
langfuse = Langfuse(secret_key="sk-lf-xxxx-your-key-here")

# GOOD - from environment variable
langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
)

Langfuse separates public key (safe to include in frontend logs) and secret key (must be protected). The secret key can create traces and read full trace content. Never include it in client-side code or Docker images. :::

:::warning Never pin to :latest in Docker Always pin image versions: langfuse/langfuse:2.44.0. The :latest tag can pull a version with database schema migrations that corrupt your database on startup if you are upgrading across major versions. Always test version upgrades in staging first. The Langfuse changelog documents breaking changes and required migration steps. :::

:::warning Handle langfuse.flush() in serverless cold starts In AWS Lambda with provisioned concurrency or warm containers, the background thread persists. But on cold starts, the thread has not yet buffered traces when the function exits. Always call langfuse.flush() in Lambda handlers regardless - it is idempotent and harmless when there is nothing to flush. :::

Interview Q&A

Q1: When would you choose Langfuse over LangSmith for a production system?

Choose Langfuse when any of these conditions apply: (1) Data residency - you cannot send data to US-hosted servers (GDPR, HIPAA, financial regulation). Deploy Langfuse within your jurisdiction. It runs on any server with Docker. (2) Cost at scale - at millions of traces/day, Langfuse self-hosted on $500/month infrastructure is dramatically cheaper than LangSmith Enterprise ($39+/user/month × team size). The break-even is typically around 10-15 engineers at 100K+ traces/day. (3) Vendor independence - MIT license means no lock-in. If Langfuse shuts down, you still have all your trace data and the source code. (4) Non-LangChain stack - Langfuse works equally well with raw Anthropic SDK, LlamaIndex, custom frameworks, or mixed Python and TypeScript stacks.

Choose LangSmith when: you are already deep in the LangChain ecosystem (zero-config auto-tracing via environment variables), you prefer fully managed infrastructure with an SLA, or you need tight Prompt Hub integration with LangChain's evaluators and chain types.

Q2: Explain Langfuse's three-level tracing hierarchy and when you would use each level.

Trace is the root object - one end-to-end user interaction. It carries user_id, session_id, overall input and output, tags, and metadata. Every user request should create exactly one trace. This is what shows up in the trace list in the Langfuse UI.

Span represents a non-LLM processing step: vector retrieval, database query, API call to an external service. Use spans to measure where latency originates. If a trace takes 3 seconds, spans might reveal it is 2.8s in vector search and 0.2s in LLM synthesis - the bottleneck is retrieval. Spans have input, output, start_time, and end_time but no token counting.

Generation is a specialized span for LLM calls. It additionally records model name, model_parameters (temperature, max_tokens), input (the messages array), output (the response text), and usage (input/output tokens). Langfuse uses the model name to automatically calculate calculated_cost using its built-in pricing table - you just record which model you used and how many tokens.

The hierarchy enables cost attribution by step (which call is most expensive?), latency attribution by step (which call is slowest?), and quality scoring at any level (you can score a specific generation or the entire trace).

Q3: How do you implement cost attribution by team and feature in Langfuse at scale?

Tag every trace with structured metadata at the start of the request pipeline: langfuse_context.update_current_trace(metadata={"feature": "ai-search", "team": "discovery"}). This is a non-negotiable convention - enforce it via a shared middleware or base class, not voluntary tagging in each endpoint.

Build a weekly cost report job that queries traces by time range (using langfuse.fetch_traces(from_timestamp=...)) and sums observation.calculated_cost per trace. Group by metadata.feature and metadata.team. Output a table showing which features consumed the most LLM budget.

The critical discipline is enforcement. Without enforcement, large "unknown" buckets emerge (traces with no feature tag) that make attribution useless. Options for enforcement: (1) Middleware layer that sets required metadata fields before calling any downstream @observe function. (2) A custom @observe wrapper that requires feature and team parameters. (3) CI linting that checks for required metadata calls in any function marked with @observe.

At scale, the cost report becomes a product management tool - product teams see how much their features cost per active user and can make optimization decisions (smaller model, shorter prompts, caching).

Q4: What are the operational challenges of self-hosting Langfuse in production?

(1) Database growth: full prompt/response storage grows fast - typically 2-5 GB/day at 100K traces with average prompts. Implement retention policies: delete traces older than 90 days via a weekly cron job calling langfuse.delete_trace(). For very high volume, consider partitioning PostgreSQL tables by month.

(2) ClickHouse operational overhead: ClickHouse is a separate service requiring its own backups, monitoring, and sizing. Without it, analytics are slow. With it, you manage two database systems. The recommendation: deploy ClickHouse from day one on a dedicated node. It handles tens of millions of rows easily on 4 CPU / 8 GB RAM.

(3) Version pinning: Always pin the Docker image version and test migrations in staging before applying to production. The schema migration on startup is irreversible - a failed migration on the wrong version can leave the database in an inconsistent state.

(4) Availability: The Langfuse SDK buffers traces in memory during server outages and retries when the server comes back. But if the application restarts during an outage, buffered traces are lost. For high-value traces, implement a write-ahead log pattern that persists to local disk before uploading.

(5) Blob storage for large payloads: Configure S3 or GCS for payloads above a size threshold (typically 50KB). Without it, very long prompts bloat PostgreSQL row storage and slow queries.

Q5: How does Langfuse's prompt versioning improve production safety compared to storing prompts in code or in a database?

Storing prompts in source code creates hard coupling between prompt changes and code deployments. Every tweak requires PR → review → CI → deployment - a 30-60 minute cycle for what might be a one-word change. Non-engineers (product managers, domain experts) cannot iterate without code repository access.

Storing prompts in a production database (custom table) is faster but lacks versioning, rollback capability, and per-version quality tracking.

Langfuse's prompt management provides: (1) Version history - every create_prompt() call creates a new immutable version. You can always see what every version contained. (2) Label-based routing - labels (production, staging, experiment-v3) control which version is served. Changing the active version is a UI click - no code change, no deployment. (3) Instant rollback - if a new prompt causes quality degradation, reassign the production label to the previous version in seconds. (4) Per-version quality tracking - when you link a generation to a prompt object with langfuse_context.update_current_observation(prompt=prompt_obj), Langfuse tracks quality metrics (scores) broken down by prompt version. You can see "version 3 of the support prompt has 15% lower faithfulness score than version 2" directly in the UI. (5) Non-engineer access - product managers can iterate on prompts in the Langfuse UI without touching the codebase.

The tradeoff: prompt changes are not visible in code review history. Mitigate with naming conventions (semantic version numbers in the prompt name) and Langfuse's built-in audit log that shows who changed which prompt when.