Langfuse — Stop Debugging LLM Apps Blind

Your support RAG returned a confident answer from the wrong policy doc.

If you have no traces, the postmortem is a Slack thread: "model issue?", "prompt drift?", "embeddings?", "try GPT-whatever." That is not debugging. That is guessing with an API key.

Langfuse is open-source LLM observability. You get traces (one user request), nested spans (each pipeline step), generations (LLM calls with tokens/cost/latency), and scores you attach when a human or eval marks the answer good or bad. The goal is simple: open the bad request and see which step failed.

What you're solving

A typical RAG path looks like:

1. Retrieve top-k chunks
2. Build a prompt from those chunks
3. Call the model
4. Return an answer

Without instrumentation, production only shows you step 4. Langfuse makes 1–3 visible in the same tree, with timing and cost on every LLM call.

Install and credentials

pip install langfuse openai

Set env vars (Cloud or self-hosted — same SDK):

export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_BASE_URL="https://cloud.langfuse.com"   # EU
# US: https://us.cloud.langfuse.com
# JP: https://jp.cloud.langfuse.com

Optional sanity check (not for hot paths):

from langfuse import get_client
get_client().auth_check()

Docs: OpenAI integration (Python), SDK overview.

Drop-in OpenAI: one import change

If you already use the OpenAI Python SDK, Langfuse's drop-in records prompts, completions, usage, cost, latency, and API errors by changing the import:

# before: import openai
from langfuse.openai import openai
# or: from langfuse.openai import OpenAI, AsyncOpenAI

completion = openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "Answer only from the context."},
        {"role": "user", "content": "What is the refund window?"},
    ],
    name="support-answer",  # shows up as the generation name in Langfuse
)
print(completion.choices[0].message.content)

You can also pass Langfuse attributes via metadata (e.g. langfuse_session_id, langfuse_user_id, langfuse_tags) on the same call. See the custom trace properties section in the docs.

Nest the pipeline with @observe()

A single OpenAI call becomes one generation. A RAG pipeline needs a trace tree. Decorate each step:

from langfuse import observe, get_client, propagate_attributes
from langfuse.openai import openai

@observe(name="retrieve")
def retrieve(query: str, k: int = 4) -> list[dict]:
    # your vector store / BM25 / hybrid search
    return search_chunks(query, k=k)

@observe(name="build_prompt")
def build_prompt(query: str, chunks: list[dict]) -> list[dict]:
    context = "\n\n".join(c["text"] for c in chunks)
    return [
        {"role": "system", "content": "Answer only from the context. If missing, say you don't know."},
        {"role": "user", "content": f"Context:\n{context}\n\nQuestion: {query}"},
    ]

@observe(name="rag_answer")
def rag_answer(query: str, user_id: str, session_id: str) -> str:
    with propagate_attributes(
        user_id=user_id,
        session_id=session_id,
        tags=["support-rag"],
        metadata={"pipeline": "support-rag-v2"},
    ):
        chunks = retrieve(query)
        messages = build_prompt(query, chunks)
        completion = openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            name="generate",
        )
        return completion.choices[0].message.content

answer = rag_answer(
    "What is the refund window for annual plans?",
    user_id="u_42",
    session_id="s_99",
)

# Short-lived scripts / serverless: flush before exit
get_client().flush()

Nested @observe calls become parent/child spans automatically. The OpenAI drop-in generations nest under the active observation. Instrumentation docs: Observe wrapper.

The workflow that matters: debug the bad RAG answer

This is the part you actually ship for.

1. Reproduce or open the failing request in Langfuse

Find the trace by user_id / session_id / time window. You should see something like:

• rag_answer (root span)
  • retrieve — input query, output chunks (ids, scores, text snippets)
  • build_prompt — assembled messages
  • generate — model, tokens, cost (USD), latency, completion text

2. Read bottom-up from the answer

If generate looks fluent but wrong, do not start by swapping models. Open retrieve. Check:

• Are the top chunks from the right document / product / locale?
• Are similarity scores suspiciously low or clustered on boilerplate?
• Did a filter drop the policy section you expected?

In the wrong-policy case, the generate span is innocent. The retrieve span shows high-confidence chunks from an older refund FAQ. Root cause: stale index or missing metadata filter — not "the LLM hallucinated."

3. Attach a score so the failure is searchable

Inside the active observation (or afterward via create_score with trace_id):

from langfuse import get_client

langfuse = get_client()

# While still inside an @observe / start_as_current_observation context:
langfuse.score_current_span(
    name="retrieval_relevance",
    value=0.2,
    data_type="NUMERIC",
    comment="Top chunks from outdated refund FAQ; annual-plan policy missing",
)
langfuse.score_current_trace(
    name="answer_correctness",
    value=0,
    data_type="BOOLEAN",
    comment="Wrong refund window cited",
)

Score types include NUMERIC, CATEGORICAL, BOOLEAN, and TEXT. Full API: Scores via SDK.

4. Fix the span that failed, then re-run the same query

New trace should show better retrieve output and a passing score. That pair — before/after traces with scores — is your regression evidence for the next prompt or index change.

Gotchas

• Flush in short-lived processes. The SDK batches in the background. Scripts and serverless handlers should call get_client().flush() (or shutdown()) before exit or you lose the last events.
• Large I/O on @observe. Huge chunk dumps can add overhead. Disable per decorator with capture_input=False / capture_output=False, or via LANGFUSE_OBSERVE_DECORATOR_IO_CAPTURE_ENABLED.
• Region matters. LANGFUSE_BASE_URL must match where you created the project (EU / US / JP / HIPAA / self-host). Wrong base URL looks like "nothing is showing up."
• Assistants API. The OpenAI Assistants path is not fully covered by the drop-in (server-side state). Prefer @observe around your own orchestration, or see Langfuse's Assistants notes in the OpenAI docs.
• Sampling. High-volume apps can set LANGFUSE_SAMPLE_RATE (or Langfuse(sample_rate=...)) so you don't drown the project — but keep 100% on the paths you are actively debugging.

Checklist

1. pip install langfuse openai and set LANGFUSE_PUBLIC_KEY / SECRET_KEY / BASE_URL.
2. Switch to from langfuse.openai import openai for automatic generation logging.
3. Put @observe(name=...) on retrieve / prompt / answer (and any tools).
4. Propagate user_id, session_id, and tags early with propagate_attributes.
5. On a bad answer, open the trace → inspect the retrieve span before blaming the model.
6. score_current_span / score_current_trace so failures are filterable later.
7. flush() in scripts and serverless.

References

• Langfuse — OpenAI SDK integration (Python)
• Instrumentation: @observe, context managers, nesting
• Python SDK overview / env vars
• Scores via API/SDK
• GitHub: langfuse/langfuse

When a production AI answer fails, do you open a trace — or start a Slack thread of guesses?