langfuse
langfuse is the Python SDK for Langfuse, the open source LLM engineering platform. It records traces of your LLM application (nested spans and generations with model, token usage, and cost per call) on top of OpenTelemetry, and ships drop-in integrations: import langfuse.openai instead of openai and every call is traced, or add a callback handler for LangChain. Beyond tracing, the same SDK covers prompt management with versioning and client-side caching, datasets and experiments for offline evaluation, scores for LLM-as-a-judge or user feedback, and a full REST API client. It talks to Langfuse Cloud or a self-hosted server.
The strongest open source option for LLM observability, with real momentum and honest self-hosting, but budget time for the server you must run or pay for, and write new code against v4 only since older examples mislead.
Use it if
- You need to see what your LLM app actually did in production: nested traces with per-generation token usage, cost, and latency
- You want tracing without touching call sites: the langfuse.openai wrapper and the LangChain CallbackHandler are genuinely drop-in
- You manage prompts outside code: get_prompt gives versioned, labeled prompts with caching so a prompt tweak does not need a deploy
- Data residency matters: the whole platform is MIT licensed and self-hostable, unlike LangSmith
- You expect pip install to be enough: the SDK is useless without a server, meaning a Langfuse Cloud account or a self-hosted deployment that involves Postgres, ClickHouse, Redis, and S3-compatible storage
- You dislike churn: the SDK was rewritten for v3 in 2025 and again for v4 in March 2026, so most blog posts and older answers show APIs that no longer exist, and the migration guide is required reading
- You are deep in the LangChain ecosystem and want first-party everything: LangSmith is built by the same team as your framework
- You just want to eyeball prompts during development: printing them or using logfire is less machinery than running an observability platform
Setup reality
pip install langfuse is trivial; everything else is not. You need LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and LANGFUSE_BASE_URL (defaults to cloud.langfuse.com), and self-hosting the current server is a multi-service compose file with Postgres, ClickHouse, Redis, and blob storage. The SDK batches events in a background thread, so short scripts, cron jobs, and serverless functions silently lose traces unless you call langfuse.flush() or shutdown() before exit. Because v4 is OpenTelemetry underneath, integrating with an existing OTel setup works but requires reading the propagation docs carefully.
Patterns
Create nested spans and generationstrace-spans-generations
# env: LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_BASE_URL
from langfuse import get_client
langfuse = get_client()
with langfuse.start_as_current_observation(as_type="span", name="process-request") as span:
span.update(input={"query": "user question"})
with langfuse.start_as_current_observation(as_type="generation", name="llm-response", model="gpt-4o") as generation:
# your LLM call here
generation.update(output="Generated response")
span.update(output="Processing complete")
langfuse.flush()This is the v4 API; v3's start_as_current_generation and the v2 trace() API are gone. Context managers close spans automatically.
Trace functions with the @observe decoratorobserve-decorator
from langfuse import observe, get_client
@observe()
def retrieve(query: str) -> list[str]:
return ["doc1", "doc2"]
@observe()
def answer(query: str) -> str:
docs = retrieve(query)
return f"answer based on {len(docs)} docs"
answer("what is caching?")
get_client().flush()Nested decorated calls become nested spans automatically. Function args and return values are captured as input/output unless you disable that.
Trace OpenAI calls with the drop-in wrapperopenai-drop-in
from langfuse.openai import openai # instead of: import openai
client = openai.OpenAI()
resp = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello"}],
)
print(resp.choices[0].message.content)One import change traces every call with model, usage, and cost. Works for sync, async, and streaming clients.
Trace LangChain with the CallbackHandlerlangchain-callback
from langfuse.langchain import CallbackHandler
handler = CallbackHandler()
chain = prompt | model | parser
result = chain.invoke(
{"topic": "observability"},
config={"callbacks": [handler]},
)Pass the handler per invocation via config; chains, tools, and retrievers inside the run all appear as nested observations.
Attach user, session, and tags to all spansuser-session-attributes
from langfuse import get_client, propagate_attributes
langfuse = get_client()
with propagate_attributes(user_id="user-123", session_id="chat-42", tags=["beta"]):
with langfuse.start_as_current_observation(as_type="span", name="handle-message") as span:
... # everything in here carries user_id and session_idpropagate_attributes replaces the v3 pattern of update_trace for these fields; sessions group traces into conversations in the UI.
Score output for evals or user feedbackscore-a-trace
from langfuse import get_client
langfuse = get_client()
with langfuse.start_as_current_observation(as_type="span", name="qa") as span:
span.update(output="the answer")
span.score(name="user-feedback", value=1, data_type="BOOLEAN")
# or later, detached, using ids you stored:
langfuse.create_score(name="accuracy", value=0.9, trace_id="<trace-id>")Scores drive the dashboards and eval charts. Numeric, boolean, and categorical data types are supported.
Fetch and compile a managed promptprompt-management
from langfuse import get_client
langfuse = get_client()
prompt = langfuse.get_prompt("movie-critic", label="production")
text = prompt.compile(movie="Dune 2")
# prompt.config can carry model settings alongside the template
model = prompt.config.get("model", "gpt-4o")Prompts are cached client-side with a TTL, so a fetch failure falls back to cache instead of taking your app down. Labels let you promote versions without code changes.
Do not lose traces in scripts and serverlessflush-short-lived
from langfuse import get_client
langfuse = get_client()
def handler(event, context):
with langfuse.start_as_current_observation(as_type="span", name="lambda-job") as span:
span.update(output="done")
langfuse.flush() # blocks until queued events are sent
return {"ok": True}Events are batched in a background thread. Without flush() (or shutdown() at process exit) short-lived processes exit before anything is delivered.
Run an experiment against a datasetdatasets-experiments
from langfuse import get_client, Evaluation
langfuse = get_client()
dataset = langfuse.get_dataset("qa-golden-set")
def my_task(*, item, **kwargs):
return my_app(item.input)
def exact_match(*, input, output, expected_output, **kwargs):
return Evaluation(name="exact_match", value=float(output == expected_output))
result = dataset.run_experiment(
name="prompt-v2",
task=my_task,
evaluators=[exact_match],
)
print(result.format())Each item run is traced and linked to the dataset item, so regressions are diffable in the UI between experiment runs.
Use the raw REST API clientrest-api-client
from langfuse import get_client
langfuse = get_client()
traces = langfuse.api.trace.list(limit=10)
for t in traces.data:
print(t.id, t.name, t.latency)langfuse.api (and async_api) exposes the full platform API for anything the high-level SDK does not wrap, like bulk exporting traces.
Alternatives
| Package | Registry | Pick it when |
|---|---|---|
| langsmith | PyPI | You are committed to LangChain/LangGraph and want the vendor's own tracing and eval platform. |
| arize-phoenix | PyPI | You want open source, OpenTelemetry-based tracing and evals you can run locally with a single container. |
| logfire | PyPI | You want general application observability from the Pydantic team where LLM calls are one signal among many. |