Master RAG and Agent Observability with Langfuse v4 in Minutes

AI systems often operate like black boxes—until something breaks. Langfuse v4 transforms unstructured AI workflows into observable, traceable pipelines where every call, latency spike, and API cost becomes visible in real time.

The latest iteration of Langfuse, released in March 2026, introduces a redesigned API that deprecates legacy methods like langfuse_context and update_current_trace. This shift prioritizes clarity and control, allowing developers to monitor RAG implementations and agentic systems with precision. Whether you're running a small prototype or a production-grade AI service, Langfuse’s free cloud tier and self-hosted options ensure scalability without vendor lock-in.

Why Observability Matters for AI Workflows

AI systems rarely fail in isolation. A slow vector search can cascade into delayed responses, while an unoptimized LLM call can inflate cloud costs unexpectedly. Observability bridges the gap between development and operations by providing three core capabilities:

• Tracing: Track execution time and data flow across every step—from embedding generation to final response delivery.
• Cost management: Visualize API usage and associated expenses before they spiral out of control.
• Real-time dashboards: Monitor latency, correctness, and throughput without manual instrumentation.

With Langfuse v4, these features are no longer optional—they’re essential for building reliable AI systems.

Setting Up Langfuse v4 in Three Steps

Before you can monitor AI workflows, you need to configure Langfuse. The process is straightforward but requires attention to environment variables and initialization order.

1\. Install the Langfuse SDK

Begin by installing the Python client and saving your dependencies:

pip install langfuse
pip freeze > requirements.txt

This ensures your environment remains reproducible. Next, create a free account on Langfuse Cloud by signing up with GitHub—no credit card required. After logging in, create a new project and navigate to the API keys section under Settings. Copy the LANGFUSE_PUBLIC_KEY (starts with pk-lf-...) and LANGFUSE_SECRET_KEY (starts with sk-lf-...).

2\. Configure Environment Variables

Update your .env file to include Langfuse credentials alongside your existing AI service keys:

GEMINI_API_KEY=AIza...
DB_HOST=localhost
DB_PORT=5432
DB_NAME=ai_docs
DB_USER=postgres
DB_PASSWORD=secure_password

# Langfuse v4 configuration
LANGFUSE_PUBLIC_KEY=pk-lf-...
LANGFUSE_SECRET_KEY=sk-lf-...
LANGFUSE_HOST=

⚠️ Critical note: Langfuse reads environment variables during initialization. Always call load_dotenv() before get_client() to avoid runtime errors. A common mistake is initializing the Langfuse client before loading the .env file, which results in silent failures.

3\. Create a Dedicated Observability Directory

Organize your project to separate evaluation logic from observability code:

pgvector-tutorial/
├── evals/
│   └── ...
└── observability/
    ├── traced_rag.py
    └── traced_agent.py

This structure keeps tracing logic modular and reusable across different AI components.

Instrumenting RAG Pipelines with Langfuse

Tracing a RAG pipeline involves decorating functions and capturing metadata at critical checkpoints. The goal is to expose the invisible steps that transform a user query into a coherent answer.

Implementing the traced_rag.py Script

Start by importing dependencies and initializing the Langfuse client after loading environment variables:

import os
from dotenv import load_dotenv
from langfuse import get_client, observe

load_dotenv()
langfuse = get_client()

Wrap each pipeline step with the @observe() decorator to automatically generate traces. For example, track embedding generation:

@observe()
def get_embedding(text: str) -> list[float]:
    """Trace the time and input/output of embedding generation."""
    result = client.models.embed_content(
        model="gemini-embedding-001",
        contents=text,
        config=types.EmbedContentConfig(
            task_type="RETRIEVAL_QUERY",
            output_dimensionality=768,
        ),
    )
    return result.embeddings[0].values

Similarly, trace vector database searches and LLM calls:

@observe(name="llm_generate")
def generate_answer(question: str, context: str) -> str:
    """Trace the final LLM generation step."""
    prompt = f"""Answer the question based on the following documents.
# Reference Documents
{context}
# Question
{question}
# Answer (concisely, based on the reference documents)"""
    response = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=prompt,
    )
    return response.text

At the pipeline level, use update_current_span() to add contextual metadata:

@observe(name="rag_pipeline")
def rag_answer(question: str) -> str:
    """Trace the entire RAG pipeline with hierarchical spans."""
    langfuse.update_current_span(
        metadata={"question": question, "tags": ["rag", "production"]}
    )
    docs = search_documents(question, top_k=3)
    context = "\n\n".join([f"[{d['title']}]\n{d['body']}" for d in docs])
    answer = generate_answer(question, context)
    return answer

Run the script to generate traces:

python observability/traced_rag.py

Langfuse automatically flushes traces to the dashboard, where you can analyze latency, token usage, and retrieval quality.

Extending Observability to Agentic Systems

Agentic workflows introduce complexity with multi-step reasoning and tool calls. Langfuse v4 simplifies this by allowing granular tracing of each agent action.

Implementing traced_agent.py

Begin with the same initialization pattern and define a function to generate embeddings. Then, create an agent step function that returns candidates explicitly:

@observe()
def agent_step(question: str) -> list[dict]:
    """Trace a single agent reasoning step."""
    # Perform tool selection, reasoning, and tool execution
    candidates = [...]  # List of possible actions
    return candidates

In the main agent loop, ensure you return and process these candidates:

@observe(name="run_agent")
def run_agent(question: str) -> str:
    """Trace the full agent execution."""
    candidates = agent_step(question)
    # Process candidates and generate final response
    return final_answer

⚠️ Common pitfalls to avoid:

• Forgetting to call load_dotenv() before get_client().
• Not returning candidates from agent_step(), which breaks trace continuity.
• Skipping langfuse.flush() at the end of execution, leading to incomplete data.

Once implemented, your agent’s decision-making process becomes fully traceable, revealing bottlenecks and inefficiencies.

From Setup to Insights: What’s Next for AI Observability?

Langfuse v4 marks a turning point in AI observability by making complex workflows accessible to developers of all skill levels. The days of guessing why an AI system underperformed are fading—replaced by real-time dashboards that highlight latency spikes, cost anomalies, and failed retrievals.

As AI adoption accelerates, observability will evolve from a best practice to a baseline requirement. The next frontier includes automated evaluation integration, anomaly detection, and predictive cost modeling. Whether you’re fine-tuning a chatbot or deploying a multi-agent orchestration system, Langfuse provides the visibility needed to build with confidence.