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 covers tracing, prompt management, evaluations, and cost tracking. Its core is MIT-licensed; enterprise modules under ee/ have a separate commercial license. The current Python SDK v4 is built on OpenTelemetry.

The guide starts with the supported installation paths, then follows one Python setup through tracing, prompts, evaluation, cost, privacy, and failure handling.

If you need to decide what the team should measure before choosing the tooling, start with the production LLM observability model. This page stays on installation and day-to-day Langfuse work.

What Langfuse is

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

Tracing. Instrument one user task as a root observation with child model, retrieval, tool, and validation observations. You see only the fields your SDK or integration actually records.

Prompt Management. Store versioned prompts and labels outside the application release, while keeping a local fallback for the path that cannot depend on a network fetch.

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

Cost Tracking. Attribute usage and calculated cost by feature, user, or model when the generation contains the required model and usage fields.

┌─────────────────────────────────────────────────────┐
│                  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.” A trace can show the captured prompt, model response, tool calls, and validation steps for that request. That narrows the reproduction search, provided the sensitive fields were deliberately recorded.

Cost

Long context and repeated tool calls can turn a small per-request cost into a material monthly bill. A trace ties tokens and model usage to the feature and task that caused them, instead of leaving the provider invoice as the first warning.

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.

Why this tutorial uses Langfuse

The useful distinction here is not a copied vendor matrix. Plans and feature lists go stale quickly. Langfuse has an OpenTelemetry-based SDK, a self-hostable MIT-licensed core, and integrations for common model and agent frameworks. Check each vendor’s current documentation before making a procurement decision.

Installation

Two paths are available: Langfuse Cloud or a self-hosted deployment. For a local trial, use the maintained Docker Compose file from the Langfuse repository.

Self-hosted via Docker Compose

The current documentation labels Docker Compose as the simplest local or single-VM path. It is not a high-availability production topology: scaling, backups, and failover remain your responsibility. Langfuse recommends Kubernetes for high availability and high throughput.

Do not copy a static compose file from a tutorial. Clone the upstream repository so the application images, storage services, environment variables, and migrations stay on the same release:

git clone https://github.com/langfuse/langfuse.git
cd langfuse

# Review every CHANGEME value before the first start.
docker compose up -d

The official file currently starts the web and worker services with PostgreSQL, ClickHouse, Redis, and MinIO. Replace all placeholder secrets, keep internal storage ports off the public internet, and plan backups before putting real traces into the instance. The maintained steps and production boundary are in the Docker Compose deployment guide.

After the services report healthy, open http://localhost:3000, create an organization and project, then copy the project-scoped Public Key and Secret Key for the SDK.

Cloud (langfuse.com)

Sign up at cloud.langfuse.com, create a project, and copy its keys. On 24 August 2026, the official pricing page lists Hobby as free with 50,000 units per month, 30 days of data access, and two users. Check that page again before relying on those limits.

Installing the Python SDK

pip install langfuse

SDK versions. These examples target the current Python SDK v4 and Python 3.10+ with from langfuse import get_client. Pin the version in your own lockfile after testing it. The v4 defaults use the current Observations and Metrics APIs, which need a v4 self-hosted server. Review the v3→v4 migration guide before upgrading existing instrumentation.

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 not langfuse.auth_check():
    raise RuntimeError("Langfuse authentication failed")

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 Langfuse v4 trace is the set of observations that share one trace ID. The root observation represents the user-visible task; child observations represent validation, retrieval, tools, and model calls. Overall input and output belong on the root.

Root span: "generate-travel-plan"
│
├── Span: "validate-input" (8ms)
│
├── Span: "retrieve-context" (200ms)
│   └── Span: "vector-search" (180ms)
│
├── Generation: "plan-draft" (model-a, 1800 tokens)
│
├── Generation: "plan-review" (model-b, 600 tokens)
│
└── Span: "format-output" (3ms)

A generation can carry model, input/output, usage, cost, latency, and metadata. The SDK does not invent fields your wrapper never records. Native integrations may populate them; manual wrappers must update the generation explicitly.

@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, propagate_attributes
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)
    model = "your-openai-model-id"
    messages = [
        {"role": "system", "content": f"Context:\n{context_text}"},
        {"role": "user", "content": query},
    ]
    response = openai_client.chat.completions.create(
        model=model,
        messages=messages,
    )
    output = response.choices[0].message.content or ""
    langfuse.update_current_generation(
        model=model,
        input=messages,
        output=output,
        usage_details={
            "input": response.usage.prompt_tokens,
            "output": response.usage.completion_tokens,
        } if response.usage else None,
    )
    return output


@observe()
def handle_request(user_query: str, user_id: str) -> str:
    """Handle one user task in the root observation."""
    with propagate_attributes(user_id=user_id):
        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 observation and trace context
• validate_input and search_context become nested spans
• generate_response becomes a generation and records its model and usage
• 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="your-openai-model-id",
        input=[{"role": "user", "content": "Cafes in Moscow"}],
    ) as generation:
        response = call_llm("Cafes in Moscow")
        generation.update(
            output=response,
            usage_details={"input": 42, "output": 128},
        )

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

langfuse.flush()

What tracing surfaces immediately

Three patterns that show up on day one:

Hidden retries. Retry logic may call the model more than once. Separate generation observations reveal each recorded attempt and its usage.

Model mismatch. One endpoint should call model-b but is still calling model-a. Filtering by the model field surfaces the wrong route.

Latency bottleneck. In a four-step chain, one step eats 80% of the time. Without spans, you only see the total. With them, you can see which step needs work.

Prompt Management

A remote prompt registry separates prompt releases from application releases. That can shorten a rollback, but it also introduces a fetch and cache policy that must be tested.

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", type="chat")

# Compile with variables
compiled = prompt.compile(
    destination="Moscow",
    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="your-openai-model-id",
    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", type="chat")

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

# A specific version by number
v3_prompt = langfuse.get_prompt("travel-assistant", type="chat", 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. Production picks up the label change after its SDK cache refreshes - no deploy

A/B testing prompts

import random
from langfuse import get_client, observe

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", type="chat", label="production"
        )
    else:
        prompt = langfuse.get_prompt(
            "travel-assistant", type="chat", 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="your-openai-model-id",
        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. A fresh process has no local prompt cache yet, so give the SDK a fallback for that first failed fetch.

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


prompt = langfuse.get_prompt(
    "travel-assistant",
    type="chat",
    label="production",
    fallback=FALLBACK_MESSAGES,
)
compiled = prompt.compile(user_query="Recommend two cafes")

The SDK exposes prompt.is_fallback if you need a metric for fallback use. Pre-fetching important prompts at startup is another option. Choose one policy and test a cold start with the Langfuse API unavailable.

Caching

The SDK caches prompts in memory with a 60-second default TTL. After the TTL expires, it can serve the stale value while refreshing it in the background. Set a longer TTL only when that freshness trade-off fits the release process:

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

Set cache_ttl_seconds=0 in development when every call must fetch the current value. This cache is separate from provider-side prompt caching.

Authenticated MCP server

Langfuse exposes project data through a Streamable HTTP MCP endpoint. The current docs recommend its CLI or agent skill when an approved agent can run shell commands; MCP is the alternative for tools that cannot. The endpoint has read and write tools by default.

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

Create a project-scoped key for this integration, keep the encoded value out of the repository, and connect only an approved client. If the client should be read-only, allow-list the lookup tools and leave out write tools. Recheck the canonical MCP server reference because its tool list and setup snippets change with the platform.

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 - the exact evaluator model and version
• 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_current_trace(
        name="not-empty",
        value=bool(response.strip()),
        data_type="BOOLEAN",
    )

    # Automated check: length within reasonable bounds
    langfuse.score_current_trace(
        name="length-ok",
        value=50 < len(response) < 5000,
        data_type="BOOLEAN",
    )

    # Format check (if we expect JSON)
    try:
        import json
        json.loads(response)
        langfuse.score_current_trace(
            name="valid-json", value=True, data_type="BOOLEAN"
        )
    except json.JSONDecodeError:
        langfuse.score_current_trace(
            name="valid-json", value=False, data_type="BOOLEAN"
        )

    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.create_score(
        trace_id=trace_id,
        name="user-feedback",
        value=thumbs_up,
        data_type="BOOLEAN",
        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. For the experiment design around those runs, use the separate prompt A/B testing workflow.

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-model-b",
    task=my_task,
)

langfuse.flush()

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

Cost Tracking

Langfuse can calculate cost when a generation has usage data and a model that matches a configured price definition. Verify the result for every model release. Unknown or self-hosted models need an explicit definition or cost_details; never assume a new model is already priced correctly.

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 the model IDs your application actually sends
• 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

Do not iterate old fetch_traces() examples: that helper is not the v4 aggregate data path. Query cost and usage through Metrics API v2, group by the stable user or feature dimension you propagated, and write the aggregate to your billing database. Keep the billing job idempotent and reconcile it against provider invoices. The current request schema is in the Metrics API documentation.

Automatic vs. manual cost calculation

For custom or self-hosted models, record usage and the calculated cost explicitly:

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": count_tokens(query),
            "output": count_tokens(response),
        },
        cost_details={
            "input": estimate_input_cost(query),
            "output": estimate_output_cost(response),
        },
    )

Framework integrations

Framework integrations change on their own release schedules. Pick one integration at the boundary where calls already pass, pin its packages, and inspect a success, stream, tool call, and provider error before enabling it broadly.

OpenAI Python wrapper

The maintained OpenAI integration wraps the OpenAI client while keeping its interface:

from langfuse.openai import OpenAI

client = OpenAI()
response = client.chat.completions.create(
    model="your-openai-model-id",
    messages=[{"role": "user", "content": "Recommend two cafes"}],
)

The wrapper can record model-call fields supplied by the provider. It does not make it safe to export raw prompts. Apply the same capture and masking policy used by manual observations, then verify the resulting field set in Langfuse.

Other stacks

Do not combine snippets from different SDK generations. Start from the current Langfuse integrations index, and record the exact package versions in your deployment manifest.

Dashboards and analytics

Langfuse dashboards and Metrics API expose the data that has reached the platform. Freshness depends on the SDK, server version, and ingestion path; verify it before using a dashboard for alerts.

Core metrics

Latency. Response time distribution across traces. Compare percentiles by model, feature, or release rather than relying on one average.

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, a session shows the captured traces that share its session_id in time order. Missing or unsampled traces remain missing; a session is not an audit log by itself.

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.

There is no sound request-count table for choosing a rate. Start from the debugging and evaluation cases you must retain, estimate their data volume, and test whether random sampling hides rare failures. Changing the environment variable requires the process to reload its configuration; do not assume a running client picks it up dynamically.

PII masking

LLM traces can contain prompts, responses, tool arguments, and personal data. Decide what is allowed to leave the application, then mask or omit everything else before export.

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

This manual pattern is useful when one function owns all sensitive input. It does not inspect attributes created by third-party OpenTelemetry instrumentation. For new Python v4 setups, Langfuse recommends the export-stage mask_otel_spans hook. Keep it deterministic and fast; an invalid result can drop a span or an export batch.

Use the smallest data path that works:

1. disable capture with capture_input=False, capture_output=False when raw IO is not needed;
2. write only a deliberately masked subset for the observations you own;
3. use mask_otel_spans for attributes exported through the Langfuse client, and configure redaction separately for every other exporter.

See the current masking contract before implementing the hook. A regex example is not a complete PII policy.

Retention

Payload size, media, metadata, and sampling determine storage. Measure your own trace mix instead of applying a per-trace estimate from another workload.

Do not delete directly from copied ClickHouse table names. Langfuse has supported UI and API paths for deleting traces, and its retention feature removes old traces, observations, scores, and media together. Availability depends on the Cloud plan or self-hosted license. Without a policy, self-hosted event data is retained indefinitely by default. Read the current retention and deletion documentation, test against non-production data, and verify the result after deletion.

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 short-lived Python jobs and serverless Python handlers, flush before the runtime is frozen or terminated. Bound that work to the handler’s remaining time; do not turn a telemetry outage into an unbounded retry.

Team access

Hierarchy: Organizations → Projects.

• Organization - a team or a company
• Project - a specific application or environment
• Each project has its own API keys; API keys are not tied to a user
• Organization roles are standard, while fine-grained project roles depend on the current Cloud plan or a self-hosted Enterprise license

Pattern for multi-environment setups:

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

Different keys and projects reduce accidental mixing between environments. User access still follows organization and project roles; verify it with a least-privilege account. The current availability matrix is in the RBAC documentation.

Monitoring Langfuse itself

Langfuse is an external dependency. Health check:

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

Add it to your monitoring. If application startup or a request fetches prompts from Langfuse, exercise the cached or local fallback during an outage test.

Self-hosted vs. Cloud

The real trade-off is operational ownership, not a guessed traces-per-month crossover.

Choose self-hosting when a documented data boundary or network requirement outweighs the work of operating PostgreSQL, ClickHouse, Redis, object storage, web, and worker services. Choose Cloud when the managed operational path matters more than owning that stack. Price alone is not enough; include backups, upgrades, incident response, and the security features your team actually needs.

Changing the SDK endpoint moves new telemetry. Moving historical traces, prompts, or datasets is a separate migration project: inventory the required objects, use supported exports and APIs, test referential links, and keep a rollback copy. Do not promise a transparent historical migration without running it.

Wrapping up

A sensible first pass is small:

1. Install - pip install langfuse + Docker Compose (or Cloud)
2. Tracing - add @observe() to your core functions
3. Cost tracking - record model and usage fields, then verify the calculated cost
4. Prompt management - move one prompt out of code and into Langfuse
5. Evaluations - set up an LLM-as-Judge for a critical endpoint

Start with one endpoint. Review representative success, failure, retry, and sensitive data cases before expanding the instrumentation.

Links:

• Langfuse GitHub - source code and self-hosted setup
• Langfuse Docs - documentation
• Python SDK - current overview
• 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.