Instrument LiteLLM calls with Langfuse traces

Introduction

When you route requests through LiteLLM to multiple providers in production and a customer reports that responses are slower and worse than yesterday, untraced calls leave you with no way to tell which model served the request, how many tokens it burned, or where the latency went. Teams that ship without per-call trace coverage routinely spend days reproducing intermittent quality drops and overspending on tokens silently — a single missing generation event can hide a regression for an entire release cycle. By the end of this lesson you will be able to wrap LiteLLM calls with the Langfuse @observe() decorator so every completion emits a structured generation event capturing model name, token usage, latency, and cost.

Key Terminology

• Trace: the top-level Langfuse container for a single user request, identified by a trace_id and holding the full tree of spans and generations produced while handling that request.
• Span: a timed unit of work inside a trace, created automatically by the @observe() decorator. Spans nest to form a parent-child hierarchy that mirrors your call graph.
• Generation: a specialized span (@observe(as_type="generation")) that represents an LLM call and carries LLM-specific metadata — model name, prompt/completion token counts, latency, and computed cost — so Langfuse can render usage and cost dashboards.
• @observe() decorator: the Langfuse Python SDK decorator that opens a span (or generation) when a function is entered and closes it when the function returns, propagating parent context automatically.
• CallbackHandler: the LiteLLM-specific Langfuse callback registered on litellm.callbacks / success_callback / failure_callback. It receives LiteLLM completion events and forwards them to Langfuse so success and failure paths are both captured.
• langfuse_context.update_current_observation(): the API used inside an @observe()-decorated function to attach structured metadata (model, usage, latency, cost) to the currently active span or generation.
• completion_cost(): LiteLLM's helper that maps a ModelResponse to a USD cost using its per-model pricing table; returns None for custom or self-hosted models with no pricing entry.
• flush_at / flush_interval: Langfuse client batching parameters that control when buffered events are sent — by event count and by elapsed seconds, respectively.

Concepts

Instrumenting LiteLLM calls with Langfuse rests on three ideas working together. First, hierarchy: a request becomes a trace, every decorated function becomes a span, and LLM calls are promoted to generations so Langfuse can aggregate token usage and cost across calls. Second, declarative capture: the @observe() decorator opens and closes spans around your function automatically, and langfuse_context.update_current_observation() lets the wrapper attach LLM-specific telemetry (model, token usage, latency, cost) without your business code needing to know Langfuse exists. Third, reliability under batching: the CallbackHandler registered on both success_callback and failure_callback ensures failed calls are traced too, while flush_at / flush_interval and an explicit client.flush() at shutdown guarantee that buffered events reach Langfuse before the process exits.

Code Walkthrough

How Langfuse Trace Hierarchy Maps to LLM Calls

Before writing any code, you need a mental model of how Langfuse organizes telemetry. A single user request to your GenAI application becomes a trace. Within that trace, each function decorated with @observe() becomes a span. When a span wraps an LLM call specifically, you promote it to a generation — a specialized span type that Langfuse uses to render token usage dashboards, cost aggregations, and model comparison views. The hierarchy matters because Langfuse uses it to compute end-to-end latency, attribute costs to specific pipeline stages, and let you drill from a slow trace down to the exact generation that caused the bottleneck.

The following diagram illustrates how a single API request decomposes into a Langfuse trace tree when your application uses LiteLLM for model routing:

This Mermaid flowchart maps the lifecycle of a single Langfuse trace as it decomposes a user API request into discrete, measurable spans. Starting from a root trace_id, the diagram shows how request_validation, llm_pipeline, and response_formatting spans nest beneath the trace, while the litellm.completion generation node captures exact token counts (1,847), latency (2.3s), and cost ($0.0142) per call. These granular spans feed directly into the Langfuse Dashboard for P95 latency monitoring and per-model cost attribution—critical for detecting inference regressions before they impact production SLAs.

Loading diagram...

Each generation node carries structured metadata — model, usage.prompt_tokens, usage.completion_tokens, latency_ms, and calculated_cost — that Langfuse indexes for its analytics views. The @observe() decorator handles span creation and parent-child linking automatically, but you must explicitly attach generation-level metadata for the LLM-specific fields.

Initializing the Langfuse Client with LiteLLM Callbacks

The first implementation step is establishing the Langfuse client connection and registering it as a LiteLLM callback handler. The init_langfuse() function below creates a Langfuse client authenticated with LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY environment variables, then injects the Langfuse callback into LiteLLM's global callback list. This approach ensures that every subsequent litellm.completion() call — regardless of where it occurs in your codebase — automatically emits trace data without requiring per-call instrumentation. The function also sets flush_at and flush_interval parameters to control batching behavior, which is critical for high-throughput services where flushing on every call would create unacceptable overhead.

1import os
2import litellm
3from langfuse import Langfuse
4from langfuse.callback import CallbackHandler
5
6def init_langfuse() -> Langfuse:
7    """Initialize Langfuse client and register LiteLLM callback."""
8    client = Langfuse(
9        public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
10        secret_key=os.environ["LANGFUSE_SECRET_KEY"],
11        host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com"),
12        flush_at=20,
13        flush_interval=10,
14    )
15
16    # Register Langfuse as a LiteLLM callback handler
17    langfuse_handler = CallbackHandler(
18        public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
19        secret_key=os.environ["LANGFUSE_SECRET_KEY"],
20        host=os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com"),
21    )
22    litellm.callbacks = [langfuse_handler]
23    litellm.success_callback = ["langfuse"]
24    litellm.failure_callback = ["langfuse"]
25
26    return client

Building the LangfuseInstrumentor Class

With the client initialized, the next step is encapsulating instrumentation logic in a reusable class. The LangfuseInstrumentor class below wraps LiteLLM's completion() and acompletion() functions with the @observe() decorator, ensuring that every LLM call automatically creates a properly nested Langfuse generation. The class extracts model name, token usage (prompt and completion), wall-clock latency, and estimated cost from LiteLLM's ModelResponse object and attaches them as structured metadata on the Langfuse generation. This pattern keeps your application code clean — callers invoke instrumentor.completion() with the same arguments they would pass to litellm.completion(), and all telemetry happens transparently. The class also handles the edge case where LiteLLM returns None for cost when using custom or self-hosted models.

1import time
2from dataclasses import dataclass, field
3from typing import Any, Optional
4from langfuse.decorators import observe, langfuse_context
5import litellm
6
7@dataclass
8class LangfuseInstrumentor:
9    """Wraps LiteLLM calls with Langfuse trace instrumentation."""
10    client: Any = field(default=None)
11
12    def __post_init__(self):
13        if self.client is None:
14            self.client = init_langfuse()
15
16    @observe(as_type="generation")
17    def completion(
18        self, model: str, messages: list[dict],
19        **kwargs: Any,
20    ) -> dict:
21        """Instrumented synchronous LLM completion."""
22        start = time.perf_counter()
23
24        response = litellm.completion(
25            model=model, messages=messages, **kwargs
26        )
27
28        latency_ms = (time.perf_counter() - start) * 1000
29        usage = response.usage
30        cost = litellm.completion_cost(completion_response=response)
31
32        # Attach structured metadata to the current Langfuse generation
33        langfuse_context.update_current_observation(
34            model=model,
35            usage={
36                "input": usage.prompt_tokens,
37                "output": usage.completion_tokens,
38                "total": usage.total_tokens,
39            },
40            metadata={
41                "latency_ms": round(latency_ms, 2),
42                "cost_usd": cost if cost is not None else 0.0,
43                "litellm_model_id": response.get("model", model),
44            },
45        )
46        return response
47
48    @observe(as_type="generation")
49    async def acompletion(
50        self, model: str, messages: list[dict],
51        **kwargs: Any,
52    ) -> dict:
53        """Instrumented async LLM completion."""
54        start = time.perf_counter()
55
56        response = await litellm.acompletion(
57            model=model, messages=messages, **kwargs
58        )
59
60        latency_ms = (time.perf_counter() - start) * 1000
61        usage = response.usage
62        cost = litellm.completion_cost(completion_response=response)
63
64        langfuse_context.update_current_observation(
65            model=model,
66            usage={
67                "input": usage.prompt_tokens,
68                "output": usage.completion_tokens,
69                "total": usage.total_tokens,
70            },
71            metadata={
72                "latency_ms": round(latency_ms, 2),
73                "cost_usd": cost if cost is not None else 0.0,
74            },
75        )
76        return response
77
78    def shutdown(self):
79        """Flush pending traces before process exit."""
80        self.client.flush()

Do's and Don'ts

Do's

1. ✓Do inject CallbackHandler into litellm.callbacks once at startup inside init_langfuse() — registering the handler globally means every litellm.completion() call anywhere in your codebase emits trace data automatically; per-call instrumentation would require patching every callsite and silently misses any invocation you forget.
2. ✓Do explicitly populate model, usage.prompt_tokens, usage.completion_tokens, latency_ms, and calculated_cost via langfuse_context.update_current_observation() on every @observe()-decorated function that wraps an LLM call — @observe() creates the span and handles parent-child linking, but Langfuse only promotes a span to a generation and indexes it in the Usage, Cost, and P95 Latency dashboard views when these exact field names are present.
3. ✓Do set flush_at=20 and flush_interval=10 in the Langfuse() constructor and call client.flush() from your shutdown hook — the batching parameters keep instrumentation off the per-request hot path, while the explicit flush guarantees the last buffered batch reaches Langfuse before the process exits instead of being silently discarded.

Don'ts

1. ✗Don't bypass @observe() and hand-roll span or trace creation — the decorator propagates trace_id and wires the parent-child hierarchy automatically; bypassing it orphans generation nodes from their parent trace, severing the drill-down path from a slow end-to-end latency in the Langfuse Dashboard to the specific litellm.completion generation that caused the bottleneck.
2. ✗Don't write litellm.completion_cost() results directly into calculated_cost without guarding against None — the function returns None for self-hosted or custom models, and an unguarded assignment propagates NaN into Langfuse cost aggregations, silently corrupting per-model cost attribution until you notice the dashboard totals are wrong.
3. ✗Don't set litellm.success_callback = ["langfuse"] without also setting litellm.failure_callback = ["langfuse"] — omitting the failure callback means provider-level errors such as rate-limit 429s, content-filter rejections, and network timeouts produce no generation event, leaving the failure modes you most need to diagnose in production invisible in your trace history.