Langfuse Observability

Instrument LLM and agent applications with Langfuse so you can see every trace, span, token cost, prompt version, and eval score in one place. This skill targets Langfuse v3 (cloud or self-hosted) and wires up the Python langfuse SDK around the OpenTelemetry backbone, so it fits LLM engineers and backend devs running OpenAI, Anthropic, LangChain, or custom agent loops in production.

When to use it

• You have an agent or RAG pipeline and need to debug why a specific run misbehaved — which retrieval, which tool call, which prompt.
• You want per-generation cost and token accounting across models without hand-rolling a metrics layer.
• You're iterating on prompts and need versioning with production/staging labels instead of redeploying code for copy changes.
• You're building a regression suite: run a dataset, attach scores, and compare two model or prompt versions.

How it works

The core is the @observe() decorator, which wraps any function into a nested trace. LLM calls become generation spans that capture model, input, output, usage_details, and computed cost. Inside an observed function you call langfuse.update_current_trace(...) / update_current_span(...) to attach metadata, user/session IDs, and tags. Because v3 is built on OpenTelemetry, you can also instrument via the OTel SDK or use integrations like langfuse.openai (drop-in) and the LangChain CallbackHandler.

Prompt management uses langfuse.get_prompt("name", label="production"), which returns a versioned prompt object you .compile(**vars). The client caches prompts in memory (cache_ttl_seconds) so a Langfuse outage doesn't take down your app. Datasets pair with dataset.run() / item.run() context managers to link each item's execution to a trace, and langfuse.create_score(...) records LLM-as-judge or custom numeric/categorical scores.

Usage example

curl -fsSL --create-dirs -o ~/.claude/skills/langfuse-observability/SKILL.md https://claudeskil.com/skills/langfuse-observability/SKILL.md
pip install langfuse
export LANGFUSE_PUBLIC_KEY=pk-... LANGFUSE_SECRET_KEY=sk-...
export LANGFUSE_HOST=https://cloud.langfuse.com

from langfuse import observe, get_client

langfuse = get_client()

@observe()
def answer(question: str):
    prompt = langfuse.get_prompt("qa-system", label="production")
    msg = prompt.compile(question=question)
    # ... call your model (langfuse.openai auto-traces) ...
    langfuse.update_current_trace(tags=["qa"], user_id="u_42")
    return "..."

answer("What is OTel?")
langfuse.flush()  # ensure buffered events are sent

Tips & gotchas

• Always langfuse.flush() (or rely on shutdown hooks) in short-lived scripts and serverless — events are batched and otherwise lost on exit.
• Link prompts to generations by passing the prompt object so Langfuse can attribute metrics to a version.
• tracing_enabled=False lets you disable ingestion in local/dev without code changes.
• Self-hosting v3 needs Postgres, ClickHouse, Redis, and S3-compatible storage — it's not a single container.

FAQ

Does tracing add latency? Spans are buffered and flushed in a background thread, so request latency is effectively unaffected.

Can I use it without the decorator? Yes — use the OTel SDK directly, the langfuse.openai wrapper, or the LangChain CallbackHandler.

Cloud vs self-host? Same SDK; only LANGFUSE_HOST changes.