Designing Production‑Grade Observability and Evaluation with Langfuse + RAGAS for LLM Applications

Why Observability and Evaluation Matter for LLM/RAG Systems

Traditional RAG and Agent development relies on manual, eye‑ball judgment of a few conversation samples, which suffers from tiny sample sizes, strong subjectivity, no quantitative metrics (recall, faithfulness, relevance, hallucination rate), and an inability to pinpoint whether failures arise in retrieval, query rewriting, or prompting. This makes version iteration a blind process and prevents stable, industrial‑scale deployment.

Langfuse: An Open‑Source LLM‑Native Observability Platform

Langfuse is a German‑hosted, OTel‑native platform that provides full‑stack tracing, metrics, dataset management, and automated evaluation for LLM applications. It offers two deployment modes: a SaaS cloud service ( https://cloud.langfuse.com) for quick prototyping and a self‑hosted Docker‑Compose stack (PostgreSQL, Redis, ClickHouse, MinIO) for production environments with strict data‑off‑site policies.

Core capabilities are:

Trace : root node representing a complete user request, automatically building a tree of child spans.

Metrics : aggregated QPS, error rates, latency percentiles, token usage, model call counts.

Dataset : manual or programmatic labeling of bad cases, storing input and output for later evaluation.

Evaluation : built‑in UI for human scoring and integration with external frameworks such as RAGAS or DeepEval.

Key Terminology

Langfuse follows the OpenTelemetry model:

Trace – the top‑level request (e.g., a chat session).

Span – any intermediate step (query rewrite, vector retrieval, re‑ranking, tool call).

Generation – the leaf node that records the actual LLM API call, including prompt, response, token usage, and latency.

Installation and Environment Setup

Install the core SDK and optional integrations:

# Core SDK
pip install langfuse
# LangChain support
pip install langchain langchain-core langchain-community
# LangGraph support
pip install langgraph
# DashScope (or OpenAI) API client
pip install dashscope
# RAGAS evaluation (optional)
pip install ragas datasets

Langfuse Python SDK version 3 is stable; version 4 introduces a changed @observe semantics. Set environment variables for the chosen deployment:

# SaaS example (PowerShell)
$env:LANGFUSE_PUBLIC_KEY="pk-xxxx"
$env:LANGFUSE_SECRET_KEY="sk-xxxx"
$env:LANGFUSE_BASE_URL="https://cloud.langfuse.com"
$env:DASHSCOPE_API_KEY="sk-xxxx"

# Self‑hosted example
$env:LANGFUSE_BASE_URL="http://127.0.0.1:3000"
# Public/secret keys are generated in the local UI

When running in China, suppress noisy OpenTelemetry logs and configure an HTTP proxy:

import logging, os
logging.getLogger("opentelemetry").setLevel(logging.CRITICAL)
os.environ["HTTP_PROXY"] = "http://127.0.0.1:7890"
os.environ["HTTPS_PROXY"] = "http://127.0.0.1:7890"

Instrumenting Code with @observe

The @observe decorator automatically creates a Span (or Generation) for the wrapped function, captures arguments, return values, execution time, and any exception, and links child spans to the parent via OpenTelemetry context. Three usage patterns: @observe(name="business flow") – creates the root Trace. @observe() – creates a regular Span for intermediate steps. @observe(as_type="generation") – creates a Generation node for direct LLM calls.

Example hierarchy generated by the decorator:

Trace → multi_step_agent_main_trace (total request)
├─ Span → qa_sub_task
│   └─ Generation → invoke_llm (first model call)
├─ Span → summary_sub_task
│   └─ Generation → invoke_llm (second model call)
└─ Span → translate_sub_task
    └─ Generation → invoke_llm (third model call)

Adding Business Context (user_id, session_id, tags, metadata)

Because the pure decorator cannot inject dynamic attributes, use OpenTelemetry’s current Span to set custom fields before the business logic runs:

from opentelemetry import trace as otel_trace
from langfuse import get_client

def production_llm_pipeline(question, user_id, session_id, tag_list=None, extra_metadata=None):
    current_span = otel_trace.get_current_span()
    if current_span.is_recording():
        current_span.set_attribute("langfuse.user.id", user_id)
        current_span.set_attribute("langfuse.session.id", session_id)
        if tag_list:
            current_span.set_attribute("langfuse.tags", tag_list)
    if extra_metadata:
        get_client().update_current_span(metadata=extra_metadata)
    # Business logic – calls to functions decorated with @observe()
    final_answer = qa_sub_task(question)
    return final_answer

Multi‑user, multi‑turn example:

global_session_id = f"chat_session_{uuid.uuid4().hex[:8]}"
request_queue = [
    ("user_00101", "Python的核心优势是什么？"),
    ("user_00101", "相较于Java有哪些短板？"),
    ("user_00202", "零基础入门推荐哪门后端语言？")
]
for uid, query in request_queue:
    resp = production_llm_pipeline(
        question=query,
        user_id=uid,
        session_id=global_session_id,
        tag_list=["online_prod", "langchain_chat", "v1.0.5"],
        extra_metadata={"request_ip": "127.0.0.1", "app_channel": "web_h5", "timestamp": str(uuid.uuid1())}
    )
    print(f"[User {uid}] Q: {query}
A: {resp}
")
get_client().flush()

Integrating with LangGraph (Agent State Machines)

Wrap the entire graph invocation with a root @observe(as_type="chain") to capture the whole workflow, and decorate each node function with a plain @observe() (or as_type="generation" for LLM calls). This yields a single Trace whose child Spans mirror the graph’s execution order, making it easy to locate bottlenecks in complex agents.

Dataset Management and Human‑In‑The‑Loop Bad‑Case Labeling

In the Langfuse UI, open a Trace, click “Add to Dataset”, and optionally assign a score (1‑5) and free‑form notes. The Trace’s input becomes the dataset item’s input, and output becomes expected_output. Multiple bad cases can be aggregated into a named Dataset (e.g., “Java‑Q&A Bad Cases”).

RAGAS Quantitative Evaluation

RAGAS provides four core metrics for RAG pipelines without requiring a ground‑truth answer:

Context Recall – does the retrieved context contain all needed facts?

Context Precision – proportion of useful information in the retrieved documents.

Faithfulness – does the LLM answer stay grounded in the provided context (detects hallucinations)?

Answer Relevancy – how well does the answer address the original question?

Typical workflow:

Export a Langfuse Dataset (or create items programmatically) containing question, answer, contexts, and optional ground_truth.

Load the data into a HuggingFace Dataset and run ragas.evaluate(...) with the four metrics.

Write the resulting scores back to the original Trace using

client.score(trace_id=..., name="ragas/faithfulness", value=...)

.

Code example for creating dataset items from a Trace:

from langfuse import Langfuse
client = Langfuse()
trace_id = "your_trace_id"
client.create_dataset_item(
    dataset_name="rag_eval_set",
    input={"question": question},
    expected_output={"answer": answer, "contexts": contexts},
    trace_id=trace_id
)

Running RAGAS and writing scores back:

from langfuse import Langfuse
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall

client = Langfuse()
raw = client.get_dataset("rag_eval_set")
items = []
for it in raw.items:
    items.append({
        "question": it.input["question"],
        "answer": it.output["answer"],
        "contexts": it.output["contexts"],
        "ground_truth": it.expected_output.get("ground_truth", "")
    })
hf_ds = Dataset.from_list(items)
result = evaluate(hf_ds, metrics=[faithfulness, answer_relevancy, context_precision, context_recall])
for i, row in result.to_pandas().iterrows():
    client.score(trace_id=raw.items[i].trace_id, name="ragas/faithfulness", value=row["faithfulness"])
    # repeat for other metrics as needed

Best‑Practice Checklist for Full‑Stack RAG Observability

Root Trace : record user_id, session_id, tags, and any custom metadata for filtering and audit.

Query Rewrite Span : capture original_query and rewritten_query to verify preprocessing.

Vector Retrieval Span : log search_query, list of retrieved_docs, and similarity scores for recall/precision analysis.

Re‑ranking Span : store the final contexts sent to the LLM.

Generation Span : record full prompt, answer, token usage, model name, and latency for faithfulness and cost tracking.

Choosing Between Two Integration Approaches

Both approaches achieve observability, but they differ in flexibility and production readiness:

Pure @observe (Method 1) : minimal code changes, ideal for prototypes, demos, or short‑lived scripts. Does not support dynamic user/session attributes or tags.

Decorator + OTel Context Injection (Method 2) : slightly more code (attribute injection once at entry), supports per‑request user_id, session_id, tags, and custom metadata. Recommended for web services, APIs, and long‑running production systems.

Flush Strategies for Reliable Data Delivery

Langfuse buffers data locally and sends it in batches. Call get_client().flush() in the following scenarios:

Short‑lived scripts or one‑off jobs – flush at the end to avoid loss.

Interactive chat loops – flush after each turn for near‑real‑time visibility.

Web services – flush in a request‑lifecycle hook (e.g., FastAPI after_request).

Long‑running background workers – rely on the SDK’s background thread; manual flush only needed on forced termination.

Common Pitfalls and Solutions

Cloud SaaS timeouts : add the logging‑level suppression snippet and configure an HTTP proxy, or switch to the self‑hosted Docker Compose deployment.

Missing data on process exit : always invoke get_client().flush() before the script ends.

Generation token usage not captured : manually call update_current_generation(...) if the LLM SDK does not expose usage.

Broken parent‑child relationships : ensure every function that participates in the workflow is decorated with @observe; otherwise the context chain will be interrupted.

Context propagation across threads or async tasks : copy the OTel context with context.copy() before spawning new threads or asyncio.create_task, then attach it in the child.

Putting It All Together – The Production Loop

Instrument the code (root Trace + per‑step Spans) and inject business attributes.

Run the service; Langfuse records full execution traces.

During review, flag Bad Cases, add them to a Dataset, and optionally add human scores.

Export the Dataset, run RAGAS to obtain quantitative metrics for each version.

Write the RAGAS scores back to the original Traces, enabling side‑by‑side comparison of model, prompt, or retrieval changes.

Iterate based on data‑driven insights rather than intuition.

This closed‑loop of observability → issue localization → dataset‑driven evaluation → metric‑backed iteration provides a scalable, data‑centric foundation for LLM/RAG production systems.