aleonsa Langfuse Observability Patterns
Production patterns for tracing, evaluating, and monitoring LLM pipelines with Langfuse.
When to Activate
- Adding observability to LLM calls (Anthropic, OpenAI)
- Tracing multi-step agent pipelines
- Managing and versioning prompts
- Tracking cost and latency per tenant/user
- Building evaluation datasets from production traces
- Debugging LLM failures or unexpected outputs
Setup
# app/core/observability.py
from langfuse import Langfuse
from langfuse.decorators import langfuse_context
from functools import lru_cache
@lru_cache(maxsize=1)
def get_langfuse() -> Langfuse:
return Langfuse(
public_key=settings.LANGFUSE_PUBLIC_KEY,
secret_key=settings.LANGFUSE_SECRET_KEY,
host=settings.LANGFUSE_HOST, # Default: cloud.langfuse.com
)
# .env
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_HOST=https://cloud.langfuse.com # or self-hosted
Tracing with Decorators (Recommended)
Basic Trace
from langfuse.decorators import observe, langfuse_context
@observe()
async def generate_reply(conversation_id: str, messages: list[dict]) -> str:
# Automatically creates a trace named "generate_reply"
response = await anthropic_client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=messages,
)
return response.content[0].text
Nested Spans
@observe() # Creates root trace
async def handle_user_message(conversation_id: str, message: str):
# Each @observe() call within creates a child span
context = await retrieve_context(conversation_id, message)
reply = await generate_reply(conversation_id, context)
await save_reply(conversation_id, reply)
return reply
@observe(name="retrieve_context") # Custom span name
async def retrieve_context(conversation_id: str, query: str) -> list[dict]:
# Embed + vector search
...
Add Metadata to Trace
@observe()
async def generate_reply(conversation_id: str, messages: list[dict]) -> str:
# Tag the current trace with metadata
langfuse_context.update_current_trace(
name="generate_reply",
user_id=current_user.uid,
session_id=conversation_id,
tags=["production", "v2"],
metadata={
"tenant_id": current_tenant.id,
"model": "claude-sonnet-4-6",
"message_count": len(messages),
},
)
response = await anthropic_client.messages.create(...)
return response.content[0].text
Native SDK Integration
Anthropic with Langfuse
from langfuse.anthropic import anthropic as langfuse_anthropic
# Drop-in replacement — wraps the Anthropic client
client = langfuse_anthropic.Anthropic()
# All calls are automatically traced
response = client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
messages=[{"role": "user", "content": prompt}],
metadata={
"langfuse_user_id": user_id,
"langfuse_session_id": session_id,
"langfuse_tags": ["production"],
},
)
OpenAI with Langfuse
from langfuse.openai import openai as langfuse_openai
client = langfuse_openai.OpenAI()
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
metadata={
"langfuse_user_id": user_id,
"langfuse_session_id": session_id,
},
)
Multi-Tenant Trace Isolation
Always tag traces with user_id and a custom tenant_id metadata field. This enables per-tenant cost analysis.
@observe()
async def process_request(
request_data: dict,
user: FirebaseUser,
tenant: Tenant,
):
langfuse_context.update_current_trace(
user_id=user.uid, # Langfuse user_id (built-in)
session_id=request_data.get("conversation_id"),
metadata={
"tenant_id": tenant.id, # Custom — for filtering in UI
"tenant_plan": tenant.plan,
"environment": settings.ENVIRONMENT,
},
tags=[tenant.plan, settings.ENVIRONMENT],
)
...
Prompt Management
Fetch Prompt from Langfuse
from langfuse import Langfuse
langfuse = get_langfuse()
def get_prompt(name: str, variables: dict) -> str:
"""Fetch versioned prompt from Langfuse. Falls back to latest if not pinned."""
prompt = langfuse.get_prompt(name) # Fetches with caching (5min TTL)
return prompt.compile(**variables)
# Usage
system_prompt = get_prompt("support_agent_v2", {
"tenant_name": tenant.name,
"language": "es",
})
Register Prompt via API
langfuse.create_prompt(
name="support_agent_v2",
prompt="You are a support agent for {{tenant_name}}. Reply in {{language}}.",
is_active=True,
labels=["production"],
)
Cost Tracking
Langfuse automatically tracks token usage and cost when using the native integrations. For custom models or manual tracking:
@observe()
async def call_model(prompt: str, model: str) -> str:
start = time.time()
response = await my_custom_llm(prompt)
latency_ms = int((time.time() - start) * 1000)
# Manually log usage
langfuse_context.update_current_observation(
usage={
"input": count_tokens(prompt),
"output": count_tokens(response),
"unit": "TOKENS",
},
model=model,
model_parameters={"temperature": 0.7, "max_tokens": 1024},
latency=latency_ms,
)
return response
Evaluation / Scoring
from langfuse import Langfuse
langfuse = get_langfuse()
# Score a specific trace after human review or automated eval
async def score_response(trace_id: str, score: float, comment: str | None = None):
langfuse.score(
trace_id=trace_id,
name="quality", # Metric name
value=score, # 0.0 to 1.0
comment=comment,
data_type="NUMERIC",
)
# Boolean scoring (correct / incorrect)
async def score_correctness(trace_id: str, correct: bool):
langfuse.score(
trace_id=trace_id,
name="correct",
value=1 if correct else 0,
data_type="BOOLEAN",
)
Automated Evaluation with LLM-as-Judge
@observe(name="eval_response_quality")
async def evaluate_quality(question: str, answer: str, trace_id: str) -> float:
eval_prompt = f"""Rate the quality of this answer from 0-10.
Question: {question}
Answer: {answer}
Reply with only a number."""
response = await anthropic_client.messages.create(
model="claude-haiku-4-5-20251001", # Cheap model for eval
max_tokens=10,
messages=[{"role": "user", "content": eval_prompt}],
)
score = float(response.content[0].text.strip()) / 10
langfuse.score(trace_id=trace_id, name="llm_quality", value=score)
return score
FastAPI Middleware — Request-Level Tracing
from langfuse.decorators import langfuse_context
import uuid
@app.middleware("http")
async def langfuse_trace_middleware(request: Request, call_next):
trace_id = str(uuid.uuid4())
request.state.langfuse_trace_id = trace_id
# Set trace metadata available to all @observe() calls in this request
langfuse_context.update_current_trace(
metadata={
"path": request.url.path,
"method": request.method,
},
)
response = await call_next(request)
# Flush at end of request (important for serverless)
langfuse_context.flush()
return response
Flush in Serverless Environments
Cloud Run, Lambda, and similar serverless environments may kill the process before background flushes complete.
# app/main.py
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
# Flush all pending Langfuse events before shutdown
get_langfuse().flush()
# Per-request flush for short-lived invocations
@router.post("/webhooks/process")
async def process_webhook(payload: dict):
await handle_payload(payload)
get_langfuse().flush() # Ensure traces are sent before response
return {"status": "ok"}
Useful Queries in Langfuse UI
- Cost by tenant: Filter by
metadata.tenant_id, group by date - Latency P95 by model: Group traces by
model, chartlatency - Error rate: Filter by
level=ERROR, count by day - Prompt comparison: Create experiments with different prompt versions
Debug Mode
import logging
logging.getLogger("langfuse").setLevel(logging.DEBUG) # Verbose SDK logs
# Or check if traces are being sent
langfuse = get_langfuse()
print(langfuse.auth_check()) # Should print {"status": "success"}
Reference Skills
- LLM call patterns → skill:
llm-engineering - FastAPI middleware → skill:
fastapi-patterns - Multi-tenant context → skill:
multi-tenant-saas - Cost-aware pipelines → skill:
cost-aware-llm-pipeline