Langfuse: A Step-by-Step LLM Observability Tutorial

An LLM application in production with no observability is a black box. A user got a bad answer — but which prompt fired, how many tokens were burned, did something fail midway? HTTP 200 tells you nothing about quality; the model can return a perfectly valid JSON blob full of nonsense.

Langfuse is an open-source (MIT) platform for LLM observability: tracing, prompt management, evaluations, cost tracking. Self-hosted with no limits, 22,000+ stars on GitHub. The Python SDK v4 is built on OpenTelemetry — compatible out of the box with any existing OTEL instrumentation.

This tutorial covers the full path from installation to production, with Python code for every component.

What Langfuse is

Langfuse solves four problems that standard APM tools (Datadog, Grafana) don’t cover for LLM applications:

Tracing. Every user request breaks down into a tree of operations: LLM calls, retrieval, preprocessing, postprocessing. You see the input/output of each step, latency, model, token counts.

Prompt Management. Prompts live in Langfuse, not in your codebase. Versioning, labels (production/staging), one-click rollback. A prompt change ships without a deploy.

Evaluations. Automated quality scoring: LLM-as-a-Judge, human annotations, custom scores via API. Numbers instead of “seems to be working.”

Cost Tracking. Automatic cost calculation for every LLM call, by model and token count. Breakdowns by feature, user, model.

┌─────────────────────────────────────────────────────┐
│                  LLM Observability                   │
├──────────┬──────────┬──────────────┬────────────────┤
│ Tracing  │   Cost   │   Prompt     │  Evaluation    │
│          │ Tracking │  Management  │                │
├──────────┼──────────┼──────────────┼────────────────┤
│ What     │ How much │ Which prompt │ How good was   │
│ happened │ it cost  │ is live      │ the response   │
└──────────┴──────────┴──────────────┴────────────────┘

Why you need Langfuse

Without observability, three things stay in the blind spot.

Debugging

A user reports: “the AI gave a wrong answer.” Without tracing, you open the code, drop in print statements, try to reproduce the issue. With Langfuse, you open the trace by user_id and see the exact prompt, the model’s response, every intermediate step. The problem is localized in minutes.

Cost

One suboptimal prompt can cost hundreds of dollars a month. GPT-5.5 with a 10,000-token context runs 3-5 cents per call. At 1,000 users a day, the gap between a good prompt and a bad one shows up clearly on the bill. Without tracking, you find out after the fact — from the OpenAI invoice.

Quality

“Seems better” isn’t a metric. You changed the prompt — did quality actually go up or down? Without evaluations it’s a matter of faith; with evaluations it’s a number: average relevance score before and after the change.

Langfuse vs. alternatives

Langfuse isn’t tied to a framework. OpenAI, Anthropic, open-source models, LangChain, LlamaIndex, LiteLLM — all through one SDK.

Installation

Two paths: self-hosted (Docker) or cloud (langfuse.com). For this tutorial we’ll start with self-hosted — full control, no limits.

Self-hosted via Docker Compose

Minimum requirements: Docker, 4 cores, 16 GB RAM (Langfuse’s recommendation). Langfuse v3 runs two application containers (web + worker) plus four storage services: PostgreSQL, ClickHouse, Redis, and S3-compatible storage (MinIO for local). For an overview of the simpler v2 architecture and of monitoring Langfuse itself, see the overview article.

Dev vs. production. The compose file below is for local runs and getting familiar with the platform. It is not meant for production — no high availability, no automated backups, no scaling. For production, Langfuse recommends Kubernetes. If you do run compose in production anyway, replace every default secret (NEXTAUTH_SECRET, SALT, PostgreSQL and MinIO passwords) with random values (openssl rand -base64 32). The current official compose file is always available via curl -LO https://langfuse.com/docker-compose.yml.

# docker-compose.yml (for local development and evaluation)
# Latest version: curl -LO https://langfuse.com/docker-compose.yml

services:
  langfuse-web:
    image: langfuse/langfuse:3
    ports:
      - "3000:3000"
    environment:
      DATABASE_URL: postgresql://postgres:postgres@db:5432/langfuse
      NEXTAUTH_SECRET: your-secret-key-change-me  # openssl rand -base64 32 — MUST change in production
      SALT: your-salt-change-me                    # openssl rand -base64 32 — MUST change in production
      NEXTAUTH_URL: http://localhost:3000
      CLICKHOUSE_URL: http://clickhouse:8123
      REDIS_HOST: redis
      LANGFUSE_S3_EVENT_UPLOAD_BUCKET: langfuse
      LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: http://minio:9000
      LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: minio
      LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY: miniopassword  # change in production
      LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: "true"
    depends_on:
      - db
      - clickhouse
      - redis
      - minio

  langfuse-worker:
    image: langfuse/langfuse:3
    environment:
      DATABASE_URL: postgresql://postgres:postgres@db:5432/langfuse
      NEXTAUTH_SECRET: your-secret-key-change-me
      SALT: your-salt-change-me
      CLICKHOUSE_URL: http://clickhouse:8123
      REDIS_HOST: redis
      LANGFUSE_S3_EVENT_UPLOAD_BUCKET: langfuse
      LANGFUSE_S3_EVENT_UPLOAD_ENDPOINT: http://minio:9000
      LANGFUSE_S3_EVENT_UPLOAD_ACCESS_KEY_ID: minio
      LANGFUSE_S3_EVENT_UPLOAD_SECRET_ACCESS_KEY: miniopassword
      LANGFUSE_S3_EVENT_UPLOAD_FORCE_PATH_STYLE: "true"
    command: ["node", "packages/worker/dist/index.js"]
    depends_on:
      - db
      - clickhouse
      - redis
      - minio

  db:
    image: postgres:16
    environment:
      POSTGRES_PASSWORD: postgres  # change in production
      POSTGRES_DB: langfuse
    volumes:
      - langfuse_pg:/var/lib/postgresql/data

  clickhouse:
    image: clickhouse/clickhouse-server:24
    volumes:
      - langfuse_ch:/var/lib/clickhouse

  redis:
    image: redis:7
    volumes:
      - langfuse_redis:/data

  minio:
    image: minio/minio
    command: server /data --console-address ":9001"
    environment:
      MINIO_ROOT_USER: minio
      MINIO_ROOT_PASSWORD: miniopassword  # change in production
    volumes:
      - langfuse_minio:/data

volumes:
  langfuse_pg:
  langfuse_ch:
  langfuse_redis:
  langfuse_minio:

docker compose up -d

Langfuse is reachable at localhost:3000 after 2-3 minutes. Create an account, organization, and project. In the project settings, copy the Public Key and Secret Key — you’ll need them for the SDK.

Cloud (langfuse.com)

Sign up at cloud.langfuse.com. Create a project, copy the keys. The Hobby plan gives you 50,000 units per month for free — enough for development and small projects.

Installing the Python SDK

pip install langfuse

SDK versions. This article is written for SDK v4 (OpenTelemetry-based, from langfuse import get_client). SDK v4 shipped in March 2026 and has been the standard since — an unpinned pip install langfuse installs v4. Key breaking changes vs. v3: update_current_trace() was split into propagate_attributes(), set_current_trace_io(), and set_current_trace_as_public(); start_span() / start_generation() became start_observation(as_type=...); DatasetItemClient.run() was removed in favor of dataset.run_experiment(); Pydantic v1 is no longer supported. Core patterns (@observe, get_client, update_current_span) work unchanged. Full list — v3→v4 migration guide.

The SDK picks up configuration automatically from environment variables:

export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_BASE_URL="http://localhost:3000"  # self-hosted
# export LANGFUSE_BASE_URL="https://cloud.langfuse.com"  # cloud EU
# export LANGFUSE_BASE_URL="https://us.cloud.langfuse.com"  # cloud US

Check the connection:

from langfuse import get_client

langfuse = get_client()
# If the environment variables are set, the client is ready to go

Tracing

Set up tracing first. Without it, the rest of Langfuse doesn’t work: evaluations attach to traces, cost tracking is calculated from generations inside traces.

Trace structure

A trace in Langfuse is a tree. At the top level is the trace (one user request). Inside are spans (intermediate steps) and generations (LLM calls).

Trace: "generate-travel-plan"
│
├── Span: "validate-input" (8ms)
│
├── Span: "retrieve-context" (200ms)
│   └── Span: "vector-search" (180ms)
│
├── Generation: "plan-draft" (GPT-5.5, 1800 tokens, $0.015)
│
├── Generation: "plan-review" (GPT-5.4-Mini, 600 tokens, $0.001)
│
└── Span: "format-output" (3ms)

For every generation you get: input/output (the full prompt and response), model, tokens (input/output/total), cost in dollars, latency, metadata.

@observe — the main tracing approach

The @observe decorator is the simplest way to instrument code. It creates a span for the function, captures arguments, the return value, and execution time. Nesting is automatic, via OpenTelemetry context propagation.

from langfuse import get_client, observe
from openai import OpenAI

openai_client = OpenAI()
langfuse = get_client()


@observe()
def validate_input(user_query: str) -> str:
    """Validate and normalize the user query."""
    cleaned = user_query.strip().lower()
    if len(cleaned) < 3:
        raise ValueError("Query too short")
    return cleaned


@observe()
def search_context(query: str) -> list[str]:
    """Search the knowledge base for relevant context."""
    # Your search logic: vector DB, Elasticsearch, etc.
    results = vector_db.search(query, top_k=5)
    return [doc.text for doc in results]


@observe(as_type="generation")
def generate_response(query: str, context: list[str]) -> str:
    """LLM call to generate the response."""
    context_text = "\n".join(context)
    response = openai_client.chat.completions.create(
        model="gpt-5.5",
        messages=[
            {"role": "system", "content": f"Context:\n{context_text}"},
            {"role": "user", "content": query},
        ],
    )
    return response.choices[0].message.content


@observe()
def handle_request(user_query: str, user_id: str) -> str:
    """Main request handler. Creates the root trace."""
    query = validate_input(user_query)
    context = search_context(query)
    response = generate_response(query, context)
    return response


# Call it
result = handle_request("Recommend cafes in Moscow", user_id="user-123")
langfuse.flush()

What happens:

• handle_request creates the root trace
• validate_input, search_context, generate_response become nested spans
• generate_response is marked as_type="generation" — it shows up as an LLM call with model and token counts
• Nesting is automatic — OpenTelemetry propagation

Updating the current span

Inside an @observe-decorated function, you can attach metadata via update_current_span(). In SDK v4, user_id and session_id are passed through propagate_attributes() — they automatically propagate to every child observation:

from langfuse import get_client, observe, propagate_attributes

langfuse = get_client()


@observe()
def process_with_metadata(query: str, user_id: str) -> str:
    # user_id and session_id — via propagate_attributes (SDK v4)
    with propagate_attributes(
        user_id=user_id,
        metadata={"source": "api", "version": "2.1"},
        tags=["production", "feature:search"],
    ):
        result = do_something(query)

        # Update output if you need a custom value
        langfuse.update_current_span(output={"processed_result": result})
        return result

Low-level API

For cases where the decorator is awkward (dynamic span creation, event loop integration):

from langfuse import get_client

langfuse = get_client()

# Context manager — automatic nesting
with langfuse.start_as_current_observation(name="process-request") as span:
    span.update(input={"query": "Cafes in Moscow"})

    # Nested span
    with langfuse.start_as_current_observation(name="search-places") as search_span:
        results = search_places("Cafes in Moscow")
        search_span.update(output={"count": len(results)})

    # Generation
    with langfuse.start_as_current_observation(
        as_type="generation",
        name="llm-response",
        model="gpt-5.5",
        input=[{"role": "user", "content": "Cafes in Moscow"}],
    ) as generation:
        response = call_llm("Cafes in Moscow")
        generation.update(
            output=response,
            usage_details={"input_tokens": 42, "output_tokens": 128},
        )

    span.update(output={"response": response})

langfuse.flush()

What tracing surfaces immediately

Three patterns that show up on day one:

Hidden retries. Retry logic calls the model 2-3 times per request. Without tracing you only see the final answer. With tracing, every call shows up with its cost.

Model mismatch. One endpoint should be calling GPT-5.4-Mini but is hitting GPT-5.5 instead. Filtering by the model field in Langfuse surfaces the mismatch immediately.

Latency bottleneck. In a four-step chain, one step eats 80% of the time. Without spans, you only see the total. With them, you see exactly where to optimize.

Prompt Management

The standard approach: the prompt is hardcoded. Changing it means a PR, code review, merge, deploy. An A/B test means two deploys. Langfuse pulls prompts out of the codebase.

Creating a prompt in the UI

In the Langfuse UI: Prompts → New Prompt. Two types:

Text prompt — a single string with variables:

You are a travel assistant. Recommend places in {{destination}}.
Take preferences into account: {{preferences}}.

Chat prompt — an array of messages:

[
  {"role": "system", "content": "You are a travel assistant for {{destination}}."},
  {"role": "user", "content": "{{user_query}}"}
]

Variables in double curly braces {{variable}} get substituted at compile time.

Loading and compiling a prompt

from langfuse import get_client

langfuse = get_client()

# Load a prompt (defaults to the version labeled "production")
prompt = langfuse.get_prompt("travel-assistant")

# Compile with variables
compiled = prompt.compile(
    destination="Moscow",
    preferences="vegetarian cuisine",
    user_query="Recommend cafes downtown",
)

# compiled is a string (for a text prompt) or a list of messages (for a chat prompt)
# Use it in the LLM call
response = openai_client.chat.completions.create(
    model="gpt-5.5",
    messages=compiled,
)

Versioning and labels

Every save creates a new version (1, 2, 3…). Labels attach to specific versions:

• production — the current live version. get_prompt("name") returns it by default.
• staging — the version under test.
• latest — the most recently created version.

# Production version (default)
prod_prompt = langfuse.get_prompt("travel-assistant")

# Staging version for testing
staging_prompt = langfuse.get_prompt("travel-assistant", label="staging")

# A specific version by number
v3_prompt = langfuse.get_prompt("travel-assistant", version=3)

Workflow:

1. Create a new prompt version in the UI
2. Attach the staging label
3. Test it in a staging environment
4. Results look good → reassign the production label to that version
5. The next production request picks up the new prompt — no deploy

A/B testing prompts

import random
from langfuse import get_client, observe, propagate_attributes

langfuse = get_client()


@observe()
def generate_with_ab_test(query: str) -> str:
    """A/B test two prompt versions."""
    variant = "A" if random.random() < 0.5 else "B"

    if variant == "A":
        prompt = langfuse.get_prompt("travel-assistant", label="production")
    else:
        prompt = langfuse.get_prompt("travel-assistant", label="staging")

    compiled = prompt.compile(user_query=query)

    # Tag for filtering in analytics
    langfuse.update_current_span(
        tags=[f"ab-test:prompt-{variant}"],
        metadata={"prompt_variant": variant},
    )

    response = openai_client.chat.completions.create(
        model="gpt-5.5",
        messages=compiled,
    )
    return response.choices[0].message.content

In the Langfuse UI, filter by the ab-test:prompt-A vs. ab-test:prompt-B tag and compare scores and cost.

Fallback pattern

Langfuse is an external dependency. If it’s unavailable, you still need a prompt.

FALLBACK_MESSAGES = [
    {"role": "system", "content": "You are a travel assistant. Help with recommendations."},
    {"role": "user", "content": "{user_query}"},
]


def get_prompt_safe(name: str, **variables) -> list[dict]:
    """Load a prompt with a fallback to a hardcoded version."""
    try:
        prompt = langfuse.get_prompt(name, label="production")
        return prompt.compile(**variables)
    except Exception:
        # Langfuse unavailable — use the fallback
        return [
            {**msg, "content": msg["content"].format(**variables)}
            for msg in FALLBACK_MESSAGES
        ]

This pattern is mandatory in production. Langfuse speeds up iteration, but it must not be a single point of failure.

Caching

The SDK caches prompts client-side. By default, no TTL is set — the prompt is fetched on every get_prompt() call. Set a TTL for production:

# Cache the prompt for 5 minutes
prompt = langfuse.get_prompt("travel-assistant", cache_ttl_seconds=300)

cache_ttl_seconds controls the client-side cache — how long the SDK uses the local copy instead of hitting the server. It has no effect on provider-side caching (Anthropic prompt caching, OpenAI caching) — the two mechanisms are independent.

MCP server: Langfuse right inside your IDE

Langfuse ships an MCP server — manage the platform straight from Claude Code, Cursor, and other AI assistants. Since May 2026 the MCP covers more than prompts: observations, scores, datasets, annotation queues, metrics, and comments.

{
  "mcpServers": {
    "langfuse": {
      "type": "http",
      "url": "https://your-langfuse.com/api/public/mcp",
      "headers": {
        "Authorization": "Basic base64(publicKey:secretKey)"
      }
    }
  }
}

Your AI assistant sees current prompts, traces, and scores, proposes changes, and applies them through MCP — no switching between IDE and browser.

Evaluations

Tracing shows what happened. Evaluations show how good it was.

Three approaches to scoring

LLM-as-a-Judge. One model scores another model’s answers. Scales well, but costs money and isn’t always accurate.

Human Annotations. Manual scoring through the Langfuse UI. Accurate, but doesn’t scale. Good for calibrating LLM-as-Judge.

Custom Scores via SDK. Programmatic scoring through the API: regex checks, metrics computed in code, user feedback. Fast, cheap, covers the mechanical checks.

LLM-as-a-Judge

Configured in the Langfuse UI: Evaluation → New Evaluator.

Parameters:

• Template — the prompt for the judge model (relevance, helpfulness, toxicity, correctness)
• Model — which model does the judging (GPT-5.5, Claude Sonnet)
• Target — which traces to score (filter by tags, name, date)
• Score name — the metric’s name (e.g., relevance)
• Score type — Numeric (0-1), Categorical, Boolean

Langfuse runs the judge model against every matching trace and records a score. The dashboard shows score distribution, trend over time, correlation with other metrics.

Evaluators can also be configured via API / MCP server, not just the UI. You can score not just the whole trace but individual observations (generation, span) — in a four-step pipeline, that means a separate evaluator per step. Score types: Numeric (0-1), Categorical, Boolean — chosen when the evaluator is created.

Custom Scores via SDK

For automated checks in code:

from langfuse import get_client, observe

langfuse = get_client()


@observe()
def generate_and_evaluate(query: str) -> str:
    """Generate a response with automatic scoring."""
    response = call_llm(query)

    # Automated check: response is not empty
    langfuse.score(
        name="not-empty",
        value=1 if len(response.strip()) > 0 else 0,
        data_type="NUMERIC",
    )

    # Automated check: length within reasonable bounds
    langfuse.score(
        name="length-ok",
        value=1 if 50 < len(response) < 5000 else 0,
        data_type="NUMERIC",
    )

    # Format check (if we expect JSON)
    try:
        import json
        json.loads(response)
        langfuse.score(name="valid-json", value=1, data_type="NUMERIC")
    except json.JSONDecodeError:
        langfuse.score(name="valid-json", value=0, data_type="NUMERIC")

    return response

User feedback as a score

def record_user_feedback(trace_id: str, thumbs_up: bool, comment: str = ""):
    """Record user feedback as a score in Langfuse."""
    langfuse.score(
        trace_id=trace_id,
        name="user-feedback",
        value=1 if thumbs_up else 0,
        data_type="NUMERIC",
        comment=comment,
    )
    langfuse.flush()

Datasets: regression testing for prompts

A dataset is a set of input/expected_output pairs. Change a prompt, run the dataset, compare scores against the previous version.

from langfuse import get_client

langfuse = get_client()

# Create a dataset
langfuse.create_dataset(name="travel-queries-v1")

# Add test cases
test_cases = [
    {
        "input": {"query": "Cafes in central Moscow"},
        "expected_output": "A list of 5+ cafes with addresses and ratings",
    },
    {
        "input": {"query": "Budget hotels in Kazan"},
        "expected_output": "A list of hotels under $30/night with descriptions",
    },
    {
        "input": {"query": "A 3-day Golden Ring route"},
        "expected_output": "A detailed route with stops and logistics",
    },
]

for case in test_cases:
    langfuse.create_dataset_item(
        dataset_name="travel-queries-v1",
        input=case["input"],
        expected_output=case["expected_output"],
    )

# Run an experiment (SDK v4: run_experiment API)
dataset = langfuse.get_dataset("travel-queries-v1")


def my_task(*, item, **kwargs):
    """Function run_experiment executes for each item."""
    return run_pipeline(item.input["query"])


# run_experiment automatically creates a run, traces each item,
# and links the results — no manual item.link() needed in v4
dataset.run_experiment(
    name="prompt-v3-gpt5.5",
    task=my_task,
)

langfuse.flush()

In the Langfuse UI: Datasets → travel-queries-v1 → Runs. Compare prompt-v2-gpt5.5 vs. prompt-v3-gpt5.5 per item with scores.

Cost Tracking

Langfuse computes the cost of every LLM call automatically, by model and token count. The built-in pricing table gets updated with every release — new models show up on launch day.

What the dashboard shows

• Total cost for a period (day, week, month)
• Cost per trace — average cost of one user request
• Cost per user — spend by a specific user
• Cost per model — split across GPT-5.5, GPT-5.4-Mini, Claude Sonnet, etc.
• Cost trend — spend over time

Per-feature cost tracking

Adding tags to traces groups cost by feature:

from langfuse import get_client, observe, propagate_attributes

langfuse = get_client()


@observe()
def generate_itinerary(destination: str, user_id: str) -> str:
    with propagate_attributes(user_id=user_id, tags=["feature:itinerary", "tier:premium"]):
        # ... LLM calls
        return result


@observe()
def chat_response(message: str, user_id: str) -> str:
    with propagate_attributes(user_id=user_id, tags=["feature:chat", "tier:free"]):
        # ... LLM calls
        return result

Filtering by the feature:itinerary tag shows how much itinerary generation costs. Chat and recommendations show up separately.

Syncing spend to your own database

def sync_user_spending(user_id: str) -> float:
    """Get a user's total spend from Langfuse."""
    traces = langfuse.fetch_traces(user_id=user_id)
    total_cost = sum(t.total_cost or 0 for t in traces.data)

    # Write to your own DB for limits, billing, analytics
    db.update_user_spending(user_id=user_id, amount=total_cost)
    return total_cost

Needed for freemium credit systems, per-user LLM spend limits, and showing usage back to the user.

Automatic vs. manual cost calculation

Langfuse computes cost automatically if it knows the model. For custom or self-hosted LLMs, set usage manually:

with langfuse.start_as_current_observation(
    as_type="generation",
    name="local-llama",
    model="llama-3.1-70b",
    input=[{"role": "user", "content": query}],
) as generation:
    response = call_local_llm(query)
    generation.update(
        output=response,
        usage_details={
            "input_tokens": count_tokens(query),
            "output_tokens": count_tokens(response),
        },
        # Cost for self-hosted: compute cost
        cost_details={
            "input": 0.001,   # $0.001 per call (your estimate)
            "output": 0.002,
        },
    )

Framework integrations

Langfuse integrates with the main LLM frameworks — automatic instrumentation instead of manually tracing every call.

OpenAI SDK — drop-in replacement

Swap one import — every call gets traced automatically:

# Was:
# from openai import OpenAI

# Now:
from langfuse.openai import OpenAI

client = OpenAI()

# Use it like a regular OpenAI SDK
# Every call automatically lands in Langfuse
response = client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "Recommend cafes in Moscow"}],
)

Captured automatically: prompt and response, model, tokens (input/output/total), cost, latency, errors. Works with streaming, async, function/tool calls.

To nest it inside an @observe trace:

from langfuse import observe
from langfuse.openai import OpenAI

client = OpenAI()


@observe()
def handle_request(query: str) -> str:
    # The OpenAI call automatically becomes a child span
    response = client.chat.completions.create(
        model="gpt-5.5",
        messages=[{"role": "user", "content": query}],
    )
    return response.choices[0].message.content

Anthropic SDK

For Anthropic, use the OpenTelemetry instrumentor:

pip install openinference-instrumentation-anthropic

from anthropic import Anthropic
from openinference.instrumentation.anthropic import AnthropicInstrumentor

# Instrument once at app startup
AnthropicInstrumentor().instrument()

client = Anthropic()

# All calls are automatically traced in Langfuse
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Recommend cafes in Moscow"}],
)

OpenTelemetry spans nest automatically inside @observe traces.

LangChain

from langfuse.langchain import CallbackHandler

langfuse_handler = CallbackHandler()

# Pass into config when invoking the chain
chain = prompt | llm | output_parser

result = chain.invoke(
    {"question": "Recommend cafes"},
    config={"callbacks": [langfuse_handler]},
)

The CallbackHandler captures every step in the chain: prompts, LLM calls, tool calls, retrieval — fully nested.

LlamaIndex

Uses OpenTelemetry-based instrumentation:

pip install openinference-instrumentation-llama-index

from openinference.instrumentation.llama_index import LlamaIndexInstrumentor

# Instrument once at startup
LlamaIndexInstrumentor().instrument()

# All LlamaIndex calls are automatically traced
index = VectorStoreIndex.from_documents(documents)
query_engine = index.as_query_engine()
response = query_engine.query("Cafes in Moscow")

LiteLLM — universal proxy

LiteLLM is a proxy to any LLM provider. One line for Langfuse:

import litellm

litellm.success_callback = ["langfuse"]

# Any provider — automatic tracing
response = litellm.completion(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "Hi"}],
    metadata={
        "trace_name": "chat-response",
        "trace_user_id": "user-123",
        "generation_name": "greeting",
        "tags": ["production", "chat"],
    },
)

LiteLLM sends Langfuse everything: model, tokens, cost, latency, input/output.

Claude Agent SDK

For tracing agents built on the Claude Agent SDK, use the OpenInference instrumentor:

pip install langfuse claude-agent-sdk openinference-instrumentation-claude-agent-sdk

from claude_agent_sdk import Agent, Runner
from openinference.instrumentation.claude_agent_sdk import ClaudeAgentSDKInstrumentor

# Instrument once at app startup
ClaudeAgentSDKInstrumentor().instrument()

agent = Agent(
    name="travel-assistant",
    instructions="You are a travel assistant.",
)

# Traces agent steps: tool calls, model responses, handoffs
result = Runner.run_sync(agent, "Recommend a route around Moscow")

In Langfuse you see every agent step: model calls, tool usage, handoffs between agents.

Dashboards and analytics

Langfuse ships with built-in dashboards for analyzing LLM applications, updated in real time.

Core metrics

Latency. Response time distribution across traces. Filter by model, feature, user. Percentiles: p50, p90, p99. If p99 latency is climbing, you see it immediately.

Cost. Spend by day, model, feature, user. Trend over time. Anomalies (sudden spikes) stand out on the graph.

Quality. Average evaluation score. Trend over time — is quality rising or falling after a prompt change. Broken down by evaluator (relevance, helpfulness, toxicity).

Volume. Trace count by day. Load spikes. Distribution across models and features.

Sessions

Langfuse groups traces into sessions — a chain of requests from one user during one session:

@observe()
def handle_message(message: str, session_id: str, user_id: str) -> str:
    with propagate_attributes(session_id=session_id, user_id=user_id):
        return generate_response(message)

In the UI: Sessions → a specific session. You see every trace for that user in chronological order — the full picture of the interaction.

Filtering and search

Filter by:

• Name — trace or observation name
• Tags — arbitrary tags
• User ID — a specific user
• Model — the LLM model
• Score — traces with a given score (e.g., relevance < 0.5)
• Time range — a date range
• Metadata — arbitrary fields

Typical queries:

• “All traces for user X in the last week” — for debugging
• “Traces with relevance < 0.3” — for analyzing bad responses
• “Traces with cost > $0.10” — for optimizing expensive requests

Production best practices

Sampling

In production, you don’t always need to trace 100% of requests. Under high load, sampling cuts down data volume and cost (cloud) or infrastructure load (self-hosted).

# Via environment variable (recommended):
export LANGFUSE_SAMPLE_RATE=0.1  # 10% of traces

A value between 0 and 1. Sampling operates at the trace level — if a trace is selected, all of its observations get sent; if not, none do.

Recommendations:

• Development: 100% (trace everything)
• Staging: 100%
• Production, < 1,000 req/day: 100%
• Production, 1,000-10,000 req/day: 50%
• Production, > 10,000 req/day: 10-20%

You can change the sampling rate dynamically via the environment variable.

PII masking

LLM traces contain full prompts and responses — personal data frequently ends up in them. For GDPR/compliance, mask PII before it reaches Langfuse.

import re
from langfuse import observe


def mask_pii(text: str) -> str:
    """Mask personal data."""
    # Email
    text = re.sub(r'[\w.-]+@[\w.-]+\.\w+', '[EMAIL]', text)
    # Phone number (US format)
    text = re.sub(r'\+?1?[\s-]?\(?\d{3}\)?[\s-]?\d{3}[\s-]?\d{4}', '[PHONE]', text)
    # Card number
    text = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '[CARD]', text)
    return text


@observe(capture_input=False, capture_output=False)
def handle_sensitive_request(query: str) -> str:
    """For sensitive data — disable auto-capture of IO."""
    result = generate_response(query)

    # Manually write the masked data
    langfuse.update_current_span(
        input=mask_pii(query),
        output=mask_pii(result),
    )
    return result

Two approaches:

1. capture_input=False, capture_output=False + manually writing masked data
2. LANGFUSE_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED=false — global switch to disable auto-capture

Retention

Traces take up space: one trace with a 1,000-token prompt is 8-12 KB; at 1,000 traces/day, that’s 300-400 MB per month. Set up deletion of old data:

-- For self-hosted: trace data lives in ClickHouse, not PostgreSQL
-- Connect to ClickHouse and run:
ALTER TABLE traces DELETE WHERE created_at < NOW() - INTERVAL 90 DAY;
ALTER TABLE observations DELETE WHERE created_at < NOW() - INTERVAL 90 DAY;
-- (ClickHouse syntax: INTERVAL 90 DAY, no quotes)

In Langfuse Cloud, retention is configured in the UI.

Async and flush

The SDK sends traces asynchronously — the main thread isn’t blocked. In short-lived environments (serverless, scripts), data may not finish sending. Call flush() before exiting:

langfuse = get_client()

# ... your logic ...

# Before exiting — wait for all data to send
langfuse.flush()

For serverless (AWS Lambda, Cloudflare Workers), call flush() at the end of the handler.

Team access

Hierarchy: Organizations → Projects.

• Organization — a team or a company
• Project — a specific application or environment
• Each project has its own API keys and data
• RBAC — available on Cloud Core, Pro, and Enterprise

Pattern for multi-environment setups:

• Project myapp-dev — development
• Project myapp-staging — staging
• Project myapp-prod — production

Different keys per environment keep data isolated.

Monitoring Langfuse itself

Langfuse is an external dependency. Health check:

curl http://localhost:3000/api/public/health

Add it to your monitoring (Zabbix, Uptime Robot, Grafana). If Langfuse is unreachable — fall back to hardcoded prompts.

Self-hosted vs. Cloud

When to self-host

• Regulatory requirements — data can’t leave your perimeter
• Volume > 100,000 units/month (savings vs. Cloud Core at $29/month)
• Infrastructure already in place (Kubernetes, Docker)
• Full control over data and retention required

When to use Cloud

• Fast start with no DevOps overhead
• Small team (1-5 people)
• Volume < 50,000 units/month — the Hobby plan is free
• Need RBAC, SSO, audit logs

Migrating between the two

Langfuse exports and imports data via API. Starting on Cloud and moving to self-hosted later is a realistic path. Prompts and datasets transfer through the SDK.

Wrapping up

Langfuse covers four LLM observability needs in a single tool. The shortest path to first data:

1. Install — pip install langfuse + Docker Compose (or Cloud)
2. Tracing — add @observe() to your core functions
3. Cost tracking — works automatically once tracing is in place
4. Prompt management — move one prompt out of code and into Langfuse
5. Evaluations — set up an LLM-as-Judge for a critical endpoint

Each step stands on its own. Start by tracing a single endpoint, look at a week’s worth of data — that’s enough to tell whether the tool is worth taking further.

Links:

• Langfuse GitHub — source code and self-hosted setup
• Langfuse Docs — documentation
• Python SDK — reference
• Python SDK v3 → v4 migration — breaking changes
• Integrations — OpenAI, Anthropic, LangChain, LlamaIndex
• Prompt Management — managing prompts
• Evaluations — scoring quality

Need help with LLM observability? I help startups ship AI products — belov.works.