Building Production-Ready LangGraph Agents with Langfuse: Traces, Prompt Registry, Evals, and Beyond

A complete guide to instrumenting multi-step AI agents with open-source LLM observability — from your first trace to a CI/CD-gated prompt deployment pipeline.

Tags: LangGraph · Langfuse · LLM Observability · Python · AI Agents · MLOps

Reading time: ~25 minutes

The Problem with Agentic AI in Production

You’ve built a LangGraph agent. It works brilliantly on your laptop. You ship it.

Then the questions start:

• “Why did it give a wrong answer for this query?”
• “How much did that run cost us?”
• “Which prompt change last Tuesday broke the accuracy?”
• “Is the agent stuck in a loop for some users?”

Without observability, these questions are unanswerable. You’re flying blind — and with agents that can execute multi-step loops, tool calls, and branching conditional logic, that’s a dangerous place to be.

This is precisely the gap Langfuse fills. It’s an open-source LLM engineering platform that transforms your opaque agent runs into structured, inspectable, continuously-improvable traces. Combined with LangGraph’s stateful graph-based agent framework, you get a full production stack for building and operating reliable AI agents.

In this article we’ll go deep on the complete integration: building a real LangGraph Research Agent, wiring up Langfuse observability with zero boilerplate, managing prompts as versioned artifacts via the Prompt Registry, setting up automated evaluations, tracking cost and latency, and building a CI/CD-safe promotion workflow for prompt changes.

Prerequisites

Before we start, make sure you have:

• Python 3.11+
• An OpenAI API key (or any supported LLM provider)
• A Langfuse account (cloud.langfuse.com — free tier available) or a self-hosted instance

pip install langgraph langchain-openai langfuse tavily-python

Set your environment variables:

export OPENAI_API_KEY="sk-..."
export LANGFUSE_PUBLIC_KEY="pk-lf-..."
export LANGFUSE_SECRET_KEY="sk-lf-..."
export LANGFUSE_HOST="https://cloud.langfuse.com"
# For the web search tool (optional but recommended)
export TAVILY_API_KEY="tvly-..."

Part 1: Understanding LangGraph and Why It Needs Observability

What is LangGraph?

LangGraph is a library for building stateful, multi-actor applications with LLMs. Unlike simple chain-of-thought prompting, LangGraph lets you define your agent as a directed graph — with nodes (steps), edges (transitions), conditional branching, cycles (loops), and a shared typed state that flows through the entire run.

This is powerful because real-world agent tasks don’t fit a linear sequence:

• A research agent might need to search, evaluate quality, and loop back for more searches if the results are thin.
• A coding agent might generate code, run tests, and iteratively fix failures.
• A customer support agent might route to different knowledge bases depending on the query type.

LangGraph handles all of this natively. But this expressiveness comes with a debugging cost: when something goes wrong in a cyclic, multi-LLM-hop graph, traditional logging gives you almost nothing useful.

Why vanilla logging isn’t enough

Consider what you’d need to debug a failed research agent run:

1. What was the initial question and state?
2. Which branch did the router take?
3. What exact prompt (with all variables filled in) went to the LLM at each step?
4. What did the LLM return, and how many tokens did it use?
5. Did the critic node trigger a loop? How many times?
6. What was the final state, and did it match the expected output?
7. How long did each step take, and what was the total cost?

print() statements can't answer this. A structured observability platform can — and Langfuse is purpose-built for exactly this data model.

Part 2: Introducing Langfuse

Langfuse captures your agent runs as a trace — a hierarchical tree of events that mirrors your graph’s execution.

The Langfuse data model

Trace (one per agent run)
├── Span: router_node
│   └── Generation: ChatOpenAI (gpt-4o)
│       ├── input: [prompt messages]
│       ├── output: "web"
│       ├── model: gpt-4o
│       ├── usage: {prompt_tokens: 142, completion_tokens: 3}
│       └── cost: $0.00029
├── Span: web_search_node
│   └── Span: tavily_search (tool call)
├── Span: synthesize_node
│   └── Generation: ChatOpenAI (gpt-4o)
│       ├── input: [compiled prompt + search results]
│       ├── output: "SVB collapsed because..."
│       ├── usage: {prompt_tokens: 891, completion_tokens: 243}
│       └── cost: $0.0021
└── Span: critic_node
    └── Generation: ChatOpenAI (gpt-4o)

Every generation stores the full prompt, completion, model name, token usage, latency, and computed cost. Every span captures the node’s input state, output state, and wall-clock time. This gives you complete reproducibility — you can re-read exactly what happened in any run at any time.

Core Langfuse features

Tracing and observability — The above tree, for every run, searchable by session, user, tag, date range, model, score, and custom metadata.

Prompt Registry — Store prompts as versioned artifacts with deployment labels. Fetch them at runtime and track exactly which version produced which generation.

Evaluations — LLM-as-judge scoring, human annotation queues, and dataset-based regression testing.

Analytics dashboards — Cost per trace, token usage over time, p50/p95/p99 latency, error rates, score distributions, and per-user breakdowns.

Sessions and users — Group traces by session and user ID to analyze multi-turn conversations and per-user quality.

Datasets — Curated input/expected-output sets for evaluation experiments and CI/CD safety gates.

Part 3: Building the Research Agent

We’ll build a Research Agent that handles fact-heavy user questions. The agent:

1. Routes the question to web search or a local knowledge base.
2. Searches via the Tavily API or a vector store.
3. Synthesizes a structured answer from search results.
4. Critiques its own answer for confidence and completeness.
5. Loops back for more research if confidence is below threshold (max 3 iterations).
6. Returns the final answer.

This is a realistic pattern for production research or Q&A agents — the kind that show up in customer support bots, internal knowledge assistants, and AI research tools.

Agent state

The state is the single source of truth that flows through every node:

from typing import TypedDict, Annotated, List, Optional
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    # Core inputs
    question: str

    # Conversation history (messages accumulate across nodes)
    messages: Annotated[List[BaseMessage], operator.add]

    # Routing decision
    route: Optional[str]  # "web" or "kb"

    # Retrieved content
    search_results: str
    sources: List[str]

    # Output
    answer: str
    confidence: float

    # Loop control
    iterations: int
    max_iterations: int

Using Annotated[List[BaseMessage], operator.add] tells LangGraph to append to the messages list across nodes rather than overwrite it — critical for conversation history to accumulate correctly.

Node implementations

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_community.tools.tavily_search import TavilySearchResults
import json

llm = ChatOpenAI(model="gpt-4o", temperature=0)
search_tool = TavilySearchResults(max_results=5)
def router_node(state: AgentState) -> dict:
    """
    Classify the question and decide the retrieval strategy.
    Returns: route = 'web' for current events/recent facts,
                     'kb' for stable domain knowledge.
    """
    system = SystemMessage(content="""You are a query router for a research agent.
Classify the user's question as either:
- 'web': requires current events, recent news, or live data
- 'kb': can be answered from established knowledge bases
Reply with ONLY a JSON object: {"route": "web", "reasoning": "..."}""")

    human = HumanMessage(content=f"Question: {state['question']}")
    result = llm.invoke([system, human])

    try:
        parsed = json.loads(result.content)
        route = parsed.get("route", "web")
    except json.JSONDecodeError:
        route = "web"  # default to web on parse failure

    return {
        "route": route,
        "messages": [result]
    }

def web_search_node(state: AgentState) -> dict:
    """
    Execute web search via Tavily and format results.
    """
    try:
        results = search_tool.invoke(state["question"])

        # Format results into a readable context block
        formatted = []
        sources = []
        for i, r in enumerate(results, 1):
            formatted.append(
                f"[{i}] {r['title']}\n"
                f"URL: {r['url']}\n"
                f"Content: {r['content'][:500]}..."
            )
            sources.append(r['url'])

        search_results = "\n\n".join(formatted)
    except Exception as e:
        search_results = f"Search failed: {str(e)}. Falling back to general knowledge."
        sources = []

    return {
        "search_results": search_results,
        "sources": sources
    }

def kb_lookup_node(state: AgentState) -> dict:
    """
    Query a local knowledge base (placeholder - plug in your vector store here).
    In production: call ChromaDB, Pinecone, Weaviate, etc.
    """
    # Placeholder for vector store retrieval
    search_results = (
        f"[Knowledge Base Results for: {state['question']}]\n"
        "Retrieved 3 relevant documents from internal KB.\n"
        "Document 1: ..."
    )
    return {
        "search_results": search_results,
        "sources": ["internal-kb"]
    }

def synthesize_node(state: AgentState) -> dict:
    """
    Synthesize a comprehensive answer from search results.
    Uses a prompt fetched from Langfuse Prompt Registry (see Part 4).
    """
    system = SystemMessage(content="""You are an expert research synthesizer.
Your job is to produce a clear, accurate, well-structured answer based on the
provided search results. Always:
- Cite your sources using [1], [2] notation
- Be factually precise - don't embellish beyond what sources say
- Acknowledge uncertainty when sources conflict or are incomplete
- Structure longer answers with clear sections""")

    human = HumanMessage(content=f"""Research results:
{state['search_results']}
Based on the above, answer this question comprehensively:
{state['question']}
This is research iteration {state['iterations'] + 1}.""")

    result = llm.invoke([system, human])

    return {
        "answer": result.content,
        "messages": [result]
    }

def critic_node(state: AgentState) -> dict:
    """
    Self-evaluate the answer quality. Return a confidence score 0.0-1.0.
    If confidence < 0.7 and iterations < max, the graph will loop.
    """
    system = SystemMessage(content="""You are a critical evaluator of AI-generated research answers.
Score the answer on:
1. Factual accuracy (are claims supported by the sources?)
2. Completeness (does it fully address the question?)
3. Clarity (is it well-structured and easy to understand?)
Reply with ONLY valid JSON: {"confidence": 0.85, "weaknesses": ["..."], "needs_more_research": false}""")

    human = HumanMessage(content=f"""Question: {state['question']}
Answer to evaluate:
{state['answer']}
Search results used:
{state['search_results'][:1000]}...""")

    result = llm.invoke([system, human])

    try:
        parsed = json.loads(result.content)
        confidence = float(parsed.get("confidence", 0.8))
        needs_more = parsed.get("needs_more_research", False)
    except (json.JSONDecodeError, ValueError):
        confidence = 0.8
        needs_more = False

    # Loop if confidence is low AND we haven't hit the iteration limit
    should_loop = (
        confidence < 0.7 
        and needs_more 
        and state["iterations"] < state.get("max_iterations", 3)
    )

    return {
        "confidence": confidence,
        "iterations": state["iterations"] + 1,
        "messages": [result],
        "should_loop": should_loop
    }

Graph assembly

from langgraph.graph import StateGraph, END

def route_after_router(state: AgentState) -> str:
    """Conditional edge: route to web_search or kb_lookup."""
    return state.get("route", "web")
def route_after_critic(state: AgentState) -> str:
    """Conditional edge: loop back for more research or finish."""
    if state.get("should_loop", False):
        return "router"  # start a new research cycle
    return END
# Build the graph
workflow = StateGraph(AgentState)
# Add nodes
workflow.add_node("router", router_node)
workflow.add_node("web_search", web_search_node)
workflow.add_node("kb_lookup", kb_lookup_node)
workflow.add_node("synthesize", synthesize_node)
workflow.add_node("critic", critic_node)
# Set entry point
workflow.set_entry_point("router")
# Add edges
workflow.add_conditional_edges(
    "router",
    route_after_router,
    {"web": "web_search", "kb": "kb_lookup"}
)
workflow.add_edge("web_search", "synthesize")
workflow.add_edge("kb_lookup", "synthesize")
workflow.add_edge("synthesize", "critic")
workflow.add_conditional_edges(
    "critic",
    route_after_critic,
    {"router": "router", END: END}
)
# Compile
app = workflow.compile()

# Export as ASCII or PNG (requires graphviz)
print(app.get_graph().draw_ascii())

Part 4: Wiring Up Langfuse

Step 1: Initialize the client

from langfuse import Langfuse
from langfuse.callback import CallbackHandler

# Client is initialized from environment variables automatically
# LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_HOST
langfuse = Langfuse()
# Verify connectivity on startup
langfuse.auth_check()  # raises if credentials are wrong

Step 2: Create a per-run callback handler

The CallbackHandler is the bridge between LangChain/LangGraph and Langfuse. Create a new instance for each agent run — this ensures each run gets its own trace.

import uuid

def create_handler(user_id: str, session_id: str = None, extra_tags: list = None) -> CallbackHandler:
    return CallbackHandler(
        # Grouping and attribution
        user_id=user_id,
        session_id=session_id or str(uuid.uuid4()),

        # Filtering in dashboards
        tags=["research-agent", "v2.1"] + (extra_tags or []),

        # Custom metadata - searchable in Langfuse UI
        metadata={
            "agent_version": "2.1",
            "environment": "production",
            "llm_model": "gpt-4o",
        },

        # Trace-level name shown in the UI
        trace_name="research-agent-run"
    )

Step 3: Run with the handler

def run_agent(
    question: str,
    user_id: str = "anonymous",
    session_id: str = None,
    max_iterations: int = 3
) -> tuple[str, str]:
    """
    Run the research agent and return (answer, trace_id).
    """
    handler = create_handler(user_id=user_id, session_id=session_id)

    result = app.invoke(
        input={
            "question": question,
            "messages": [],
            "search_results": "",
            "sources": [],
            "answer": "",
            "confidence": 0.0,
            "iterations": 0,
            "max_iterations": max_iterations
        },
        config={
            "callbacks": [handler],
            "recursion_limit": max_iterations * 5  # safety limit for LangGraph
        }
    )

    # Always flush before the function returns
    # (especially important in serverless environments)
    langfuse.flush()

    trace_id = handler.get_trace_id()
    return result["answer"], trace_id

# Usage
answer, trace_id = run_agent(
    question="What caused Silicon Valley Bank to collapse in March 2023?",
    user_id="user-42",
    session_id="session-abc-123"
)
print(f"Answer: {answer}")
print(f"Trace ID: {trace_id}")
print(f"View trace: https://cloud.langfuse.com/trace/{trace_id}")

That’s the complete integration. Every node invocation, every LLM call, every tool call is now captured in Langfuse — automatically, with zero changes to your node implementations.

What Langfuse captures automatically

Data point Where it comes from Node name, input state, output state LangChain callback on_chain_start/end LLM prompt (all messages) on_llm_start LLM completion text on_llm_end Model name and parameters on_llm_start Token usage (prompt + completion) on_llm_end Computed cost (USD) Model pricing table in Langfuse Latency per span Start/end timestamps Tool call names and inputs on_tool_start Tool call outputs on_tool_end Errors and stack traces on_chain_error, on_llm_error

Adding manual spans for non-LangChain operations

Sometimes you have logic that falls outside LangChain callbacks — a database call, a custom embedding, a business logic calculation. You can wrap these in manual spans:

from langfuse.decorators import observe, langfuse_context

@observe(name="vector_store_retrieval")
def query_vector_store(query: str, top_k: int = 5) -> list[dict]:
    """
    Query your vector store. @observe creates a Langfuse span automatically.
    """
    # Your ChromaDB / Pinecone / Weaviate call here
    results = vector_store.similarity_search(query, k=top_k)

    # Optionally update the span with additional metadata
    langfuse_context.update_current_observation(
        metadata={"top_k": top_k, "results_count": len(results)},
        input={"query": query},
        output={"results": [r.page_content[:100] for r in results]}
    )

    return results

The @observe decorator is framework-agnostic — it works anywhere in your Python codebase, not just inside LangGraph nodes.

Part 5: Prompt Registry — Manage Prompts as Versioned Artifacts

The problem with prompts in code

In most codebases, prompts live in Python strings scattered across files. When a prompt underperforms, the fix requires:

1. Edit the Python file
2. Write a PR
3. Wait for review
4. Deploy to staging
5. Test
6. Deploy to production

For prompt engineering — which is inherently iterative and often done by non-engineers (ML engineers, product managers, domain experts) — this cycle is painfully slow.

Langfuse’s Prompt Registry decouples prompts from code: prompts are stored as versioned artifacts in Langfuse, fetched at runtime, and managed entirely through the Langfuse UI or API. You change a prompt, label it production, and the running agent picks it up within 60 seconds — no code change, no deployment.

Creating prompts in the registry

You can create prompts in the Langfuse UI (Settings → Prompts → New Prompt) or programmatically:

# Create a new prompt version
langfuse.create_prompt(
    name="research-synthesizer",
    type="chat",  # "chat" for message arrays, "text" for single string
    prompt=[
        {
            "role": "system",
            "content": """You are an expert research synthesizer.
Your job is to produce clear, accurate, well-structured answers.
Guidelines:
- Cite sources using [1], [2] notation
- Be factually precise - only state what sources support
- Acknowledge uncertainty when sources conflict
- Structure answers with clear sections for complex topics
- Keep responses under {{max_words}} words"""
        },
        {
            "role": "user", 
            "content": """Research results:
{{search_results}}
Question to answer:
{{question}}
Research iteration: {{iteration}}/{{max_iterations}}"""
        }
    ],
    labels=["staging"],  # don't go straight to production
    config={
        "model": "gpt-4o",
        "temperature": 0,
        "max_tokens": 800
    },
    tags=["research", "synthesis"]
)

Variables use {{double_brace}} syntax. The config field stores associated model settings — useful for keeping prompt and model configuration together.

Fetching and using prompts at runtime

# Fetch the production-labeled version
# SDK caches for 60 seconds by default — safe for high-throughput
synth_prompt = langfuse.get_prompt("research-synthesizer")

# Or fetch a specific version for testing
synth_prompt_v3 = langfuse.get_prompt("research-synthesizer", version=3)
# Or fetch by label (useful for A/B testing)
synth_prompt_exp = langfuse.get_prompt("research-synthesizer", label="experiment")

def synthesize_node(state: AgentState) -> dict:
    # Compile the template with runtime values
    # compile() returns a list of message dicts for chat prompts
    compiled_messages = synth_prompt.compile(
        search_results=state["search_results"],
        question=state["question"],
        iteration=str(state["iterations"] + 1),
        max_iterations=str(state.get("max_iterations", 3)),
        max_words="500"
    )

    # Convert to LangChain message objects
    messages = []
    for msg in compiled_messages:
        if msg["role"] == "system":
            messages.append(SystemMessage(content=msg["content"]))
        else:
            messages.append(HumanMessage(content=msg["content"]))

    result = llm.invoke(messages)

    return {
        "answer": result.content,
        "messages": [result]
    }

When the Langfuse callback handler is active, it automatically links each generation to the prompt version that was used. This means in the Langfuse UI, you can filter generations by prompt version and see exactly which version of a prompt produced which outputs — an invaluable capability for diagnosing quality regressions.

The prompt promotion workflow

[Draft in UI]
      │
      ▼
[Label: staging] ──► Test in CI pipeline against eval dataset
      │
      ▼ (score ≥ threshold)
[Label: production] ──► Agent picks up within 60s, no deploy needed
      │
      ▼ (if regression detected)
[Roll back: re-label previous version as production]

Rollback is a single click (or one API call):

# If a prompt regression is detected, roll back instantly
langfuse.create_prompt(
    name="research-synthesizer",
    # ... same content as the last good version ...
    labels=["production"]  # this supersedes the bad version
)

Part 6: Evaluations

Evaluations are how you measure whether your agent is actually good — not just whether it runs without errors, but whether it produces accurate, useful answers.

Langfuse supports three evaluation modes that work together.

Mode 1: LLM-as-judge (automated scoring)

Define scoring rubrics and run an LLM over your production traces asynchronously. This is the workhorse of automated evaluation — cheap enough to run on every trace, accurate enough to detect regressions.

import json
from langfuse import Langfuse

langfuse = Langfuse()
def score_trace_factual_accuracy(trace_id: str, question: str, answer: str):
    """
    Use GPT-4o to evaluate factual accuracy of an answer.
    Score: 0.0 (completely wrong) to 1.0 (fully accurate).
    """
    judge = ChatOpenAI(model="gpt-4o", temperature=0)

    judge_prompt = f"""You are an expert fact-checker evaluating an AI research agent's answer.
Question: {question}
AI Answer:
{answer}
Evaluate the answer on FACTUAL ACCURACY only:
- 1.0: All claims are accurate and well-supported
- 0.8: Mostly accurate with minor imprecisions
- 0.6: Some inaccuracies or unsupported claims
- 0.4: Multiple significant errors
- 0.2: Mostly inaccurate
- 0.0: Completely wrong or fabricated
Reply ONLY with valid JSON:
{{"score": 0.85, "reasoning": "The answer correctly states X but incorrectly claims Y..."}}"""

    result = judge.invoke([HumanMessage(content=judge_prompt)])

    try:
        parsed = json.loads(result.content)
        langfuse.score(
            trace_id=trace_id,
            name="factual-accuracy",
            value=parsed["score"],
            comment=parsed.get("reasoning", ""),
            data_type="NUMERIC"
        )
    except (json.JSONDecodeError, KeyError) as e:
        print(f"Score parsing failed for trace {trace_id}: {e}")

def score_trace_completeness(trace_id: str, question: str, answer: str):
    """Score whether the answer fully addresses the question."""
    judge = ChatOpenAI(model="gpt-4o", temperature=0)

    judge_prompt = f"""Question: {question}
Answer: {answer}
Does this answer completely address the question? Score 0.0-1.0.
Reply ONLY with JSON: {{"score": 0.9, "reasoning": "..."}}"""

    result = judge.invoke([HumanMessage(content=judge_prompt)])
    parsed = json.loads(result.content)

    langfuse.score(
        trace_id=trace_id,
        name="completeness",
        value=parsed["score"],
        comment=parsed.get("reasoning", ""),
        data_type="NUMERIC"
    )

# Run evaluations on recent traces
# In production: run this as a separate async job (e.g., scheduled Lambda or Celery task)
def run_evaluations_on_recent_traces(limit: int = 50):
    traces = langfuse.fetch_traces(
        tags=["research-agent"],
        limit=limit,
        order_by="timestamp",
        order="desc"
    ).data

    for trace in traces:
        # Skip if already scored
        existing_scores = {s.name for s in trace.scores}
        if "factual-accuracy" in existing_scores:
            continue

        # Extract question and answer from trace metadata
        question = trace.input.get("question", "") if trace.input else ""
        answer = trace.output.get("answer", "") if trace.output else ""

        if question and answer:
            score_trace_factual_accuracy(trace.id, question, answer)
            score_trace_completeness(trace.id, question, answer)

    langfuse.flush()

Mode 2: Human annotation

The Langfuse UI has a built-in annotation queue. You configure rubrics (thumbs up/down, 1–5 stars, categorical labels), assign traces to reviewers, and track annotation progress.

# Define annotation queues programmatically
# (or do this in the UI under Settings → Annotation Queues)
queue = langfuse.create_annotation_queue(
    name="research-agent-weekly-review",
    description="Weekly manual review of 50 random production traces",
    score_configs=[
        {
            "name": "overall-quality",
            "data_type": "NUMERIC",
            "min_value": 1,
            "max_value": 5,
            "description": "Overall answer quality (1=poor, 5=excellent)"
        },
        {
            "name": "citation-quality", 
            "data_type": "CATEGORICAL",
            "categories": ["none", "partial", "complete"],
            "description": "Quality of source citations"
        }
    ]
)

Human annotation is especially important in the first few weeks of a new agent deployment, before you have enough labeled data to trust automated scores. It also catches failure modes that LLM judges miss — like answers that are technically accurate but poorly formatted, or that answer a different question than the user intended.

Mode 3: Datasets and regression testing

A dataset is a curated set of (input, expected_output) pairs that you run your agent against to measure quality systematically. This is the cornerstone of safe prompt promotion.

# Create a gold standard dataset
dataset = langfuse.create_dataset(
    name="research-qa-gold-v1",
    description="100 hand-verified Q&A pairs covering finance, science, and history"
)

# Add items
test_cases = [
    {
        "input": {"question": "What caused SVB's collapse in March 2023?"},
        "expected_output": "SVB failed due to a bank run triggered by losses in its bond portfolio..."
    },
    {
        "input": {"question": "What is the mechanism of mRNA vaccines?"},
        "expected_output": "mRNA vaccines work by delivering genetic instructions..."
    },
    # ... 98 more
]
for case in test_cases:
    langfuse.create_dataset_item(
        dataset_name="research-qa-gold-v1",
        input=case["input"],
        expected_output=case["expected_output"],
        metadata={"category": "finance"}  # optional categorization
    )

Running an experiment against the dataset:

def run_experiment(
    experiment_name: str,
    prompt_label: str = "staging"
) -> dict:
    """
    Run the agent against the entire gold dataset and return aggregate scores.
    """
    dataset = langfuse.get_dataset("research-qa-gold-v1")
    scores = []

    for item in dataset.items:
        # Create handler linked to this dataset run
        handler = CallbackHandler(
            tags=["eval", experiment_name],
            metadata={"experiment": experiment_name, "prompt_label": prompt_label}
        )

        result = app.invoke(
            item.input,
            config={"callbacks": [handler]}
        )

        # Link the trace to the dataset item and run
        item.link(
            trace_or_observation=handler.get_trace_id(),
            run_name=experiment_name,
            run_description=f"Testing prompt label: {prompt_label}"
        )

        # Score against expected output
        if item.expected_output:
            score = evaluate_answer_against_expected(
                answer=result.get("answer", ""),
                expected=item.expected_output
            )

            langfuse.score(
                trace_id=handler.get_trace_id(),
                name="answer-similarity",
                value=score
            )
            scores.append(score)

    langfuse.flush()

    avg_score = sum(scores) / len(scores) if scores else 0
    return {
        "experiment": experiment_name,
        "n_items": len(dataset.items),
        "avg_score": avg_score,
        "passed": avg_score >= 0.80  # your threshold
    }

def evaluate_answer_against_expected(answer: str, expected: str) -> float:
    """Simple semantic similarity using an LLM judge."""
    judge = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    prompt = f"""Rate how similar these two answers are on a 0.0-1.0 scale:
Expected: {expected[:500]}
Actual: {answer[:500]}
Reply ONLY with a number between 0.0 and 1.0."""

    result = judge.invoke([HumanMessage(content=prompt)])
    try:
        return float(result.content.strip())
    except ValueError:
        return 0.5

CI/CD-gated prompt promotion

Wire the dataset experiment into your deployment pipeline:

# ci_eval.py — run this in your CI/CD pipeline before promoting a prompt
import sys

def ci_gate_prompt_promotion(
    candidate_label: str = "staging",
    min_score: float = 0.80,
    experiment_name: str = None
) -> None:
    """
    Run evaluation and exit with error code if score is below threshold.
    Use in CI: python ci_eval.py -- exits 0 on pass, 1 on fail.
    """
    import datetime
    exp_name = experiment_name or f"ci-eval-{datetime.date.today()}"

    print(f"Running CI evaluation: {exp_name}")
    result = run_experiment(experiment_name=exp_name, prompt_label=candidate_label)

    print(f"Results: avg_score={result['avg_score']:.3f}, threshold={min_score}")

    if not result["passed"]:
        print(f"FAILED: Score {result['avg_score']:.3f} below threshold {min_score}")
        sys.exit(1)

    print(f"PASSED: Promoting prompt to production")
    # Promote prompt: re-label as production
    # (call Langfuse API or do it manually in UI)
if __name__ == "__main__":
    ci_gate_prompt_promotion()

In your GitHub Actions or CI pipeline:

# .github/workflows/promote-prompt.yml
name: Promote Prompt to Production

on:
  workflow_dispatch:
    inputs:
      prompt_name:
        description: 'Prompt name to promote'
        required: true
jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install langgraph langchain-openai langfuse tavily-python
      - name: Run evaluation gate
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          LANGFUSE_PUBLIC_KEY: ${{ secrets.LANGFUSE_PUBLIC_KEY }}
          LANGFUSE_SECRET_KEY: ${{ secrets.LANGFUSE_SECRET_KEY }}
          LANGFUSE_HOST: ${{ secrets.LANGFUSE_HOST }}
        run: python ci_eval.py

Part 7: Cost and Latency Analytics

What Langfuse tracks automatically

Every generation includes token usage reported by the OpenAI API. Langfuse maps this to USD cost using a built-in pricing table (updated regularly) and surfaces it in dashboards.

You can also override or extend the pricing:

# Custom model pricing (useful for local/fine-tuned models)
langfuse.create_model(
    model_name="my-fine-tuned-gpt4",
    match_pattern="my-fine-tuned-*",
    input_price=0.000010,   # $ per token
    output_price=0.000030,  # $ per token
    tokenizer_id="cl100k_base"
)

Key metrics to track

Metric What it tells you Alert threshold Cost per trace Average USD spend per agent run >2× baseline p95 latency Worst-case user-facing delay >15 seconds Error rate % of traces ending in error >2% Avg. iterations Mean loop count per run >2.5 (agent is looping excessively) Tokens per run Total tokens consumed per agent run >4000 (may need prompt optimization) Score drift 7-day rolling average of factual-accuracy Drop >0.05 from baseline

Custom metadata for richer analytics

Pass business-relevant metadata to slice your dashboards in ways that matter:

handler = CallbackHandler(
    user_id=user.id,
    session_id=session.id,
    tags=["research-agent", "production", f"tier-{user.subscription_tier}"],
    metadata={
        "agent_version": "2.1",
        "question_category": classify_question(question),   # "finance", "science", etc.
        "user_country": user.country,
        "client": "web-app",
        "ab_group": "control"  # for A/B experiments
    }
)

Now in Langfuse dashboards you can filter cost, latency, and quality scores by subscription tier, question category, or A/B group — giving you the same segmentation power as a full product analytics platform.

Logging custom scores during a run

You can write scores to a trace at any point — not just after the run completes. This is useful for logging intermediate signals:

def critic_node(state: AgentState) -> dict:
    # ... run critic LLM ...
    confidence = parsed["confidence"]

    # Log confidence as a score on the current trace
    # Requires getting the trace_id from the callback context
    # In practice, pass trace_id through state or use @observe decorator
    langfuse.score(
        trace_id=current_trace_id,
        name="critic-confidence",
        value=confidence,
        data_type="NUMERIC",
        comment=f"Iteration {state['iterations'] + 1}"
    )

    return {
        "confidence": confidence,
        "iterations": state["iterations"] + 1
    }

Part 8: Sessions, Multi-Turn Agents, and User Tracking

For agents embedded in multi-turn chat applications, session tracking is critical. Group all traces from a single conversation under one session_id:

class ConversationManager:
    def __init__(self, user_id: str):
        self.user_id = user_id
        self.session_id = str(uuid.uuid4())
        self.conversation_history = []

    def ask(self, question: str) -> str:
        """Run a research query, maintaining conversation context."""
        handler = CallbackHandler(
            user_id=self.user_id,
            session_id=self.session_id,  # same session_id for all turns
            tags=["research-agent", "multi-turn"],
            metadata={
                "turn_number": len(self.conversation_history) + 1,
                "has_prior_context": len(self.conversation_history) > 0
            }
        )

        # Include conversation history in state
        result = app.invoke(
            input={
                "question": question,
                "messages": list(self.conversation_history),
                "search_results": "",
                "sources": [],
                "answer": "",
                "confidence": 0.0,
                "iterations": 0,
                "max_iterations": 3
            },
            config={"callbacks": [handler]}
        )

        langfuse.flush()

        # Update history
        self.conversation_history.extend(result.get("messages", []))

        return result["answer"]

In the Langfuse UI, you can view all traces in a session as a timeline — perfect for debugging multi-turn failures where context from earlier turns causes issues later.

Part 9: Self-Hosting Langfuse

Langfuse is fully open-source (MIT license). If data residency, compliance, or cost are concerns, you can self-host.

Quick local setup

git clone https://github.com/langfuse/langfuse.git
cd langfuse

# Copy and configure environment
cp .env.example .env
# Edit .env: set NEXTAUTH_SECRET, DATABASE_URL, etc.
docker compose up -d

Langfuse is available at http://localhost:3000. Set LANGFUSE_HOST=http://localhost:3000 in your agent code.

Production architecture

For production self-hosting, Langfuse requires:

• PostgreSQL (RDS, Aurora, or Supabase) — primary data store for traces and metadata
• ClickHouse — analytical query engine for dashboard aggregations
• Redis — queue for async event processing
• Object storage (S3 or compatible) — for large payloads and exports

The Langfuse team provides a Helm chart for Kubernetes deployments:

helm repo add langfuse https://langfuse.github.io/langfuse-k8s
helm install langfuse langfuse/langfuse \
  --set postgresql.auth.password=yourpassword \
  --set langfuse.nextAuthSecret=yoursecret

Important: Never use the default Docker Compose volumes in production. They are ephemeral — a container restart will wipe your trace history. Always configure external, persistent database backends.

Part 10: Putting It All Together

Here’s the complete production-ready integration in one place:

"""
research_agent.py
Complete LangGraph research agent with full Langfuse observability.
"""

import uuid
import json
import operator
from typing import TypedDict, Annotated, List, Optional
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.graph import StateGraph, END
from langfuse import Langfuse
from langfuse.callback import CallbackHandler
# ─── Initialization ───────────────────────────────────────────────────────────
langfuse = Langfuse()
langfuse.auth_check()
llm = ChatOpenAI(model="gpt-4o", temperature=0)
search_tool = TavilySearchResults(max_results=5)
# ─── Fetch prompts from registry ──────────────────────────────────────────────
# These are fetched once at startup and cached (60s TTL)
# Update prompts in Langfuse UI without redeploying
router_prompt     = langfuse.get_prompt("research-router",      fallback_to_latest=True)
synthesizer_prompt = langfuse.get_prompt("research-synthesizer", fallback_to_latest=True)
critic_prompt     = langfuse.get_prompt("research-critic",      fallback_to_latest=True)
# ─── State definition ─────────────────────────────────────────────────────────
class AgentState(TypedDict):
    question: str
    messages: Annotated[List[BaseMessage], operator.add]
    route: Optional[str]
    search_results: str
    sources: List[str]
    answer: str
    confidence: float
    iterations: int
    max_iterations: int

# ─── Nodes ────────────────────────────────────────────────────────────────────
def router_node(state: AgentState) -> dict:
    compiled = router_prompt.compile(question=state["question"])
    result = llm.invoke(compiled)
    try:
        parsed = json.loads(result.content)
        route = parsed.get("route", "web")
    except json.JSONDecodeError:
        route = "web"
    return {"route": route, "messages": [result]}

def web_search_node(state: AgentState) -> dict:
    try:
        results = search_tool.invoke(state["question"])
        formatted = [
            f"[{i}] {r['title']}\nURL: {r['url']}\n{r['content'][:400]}"
            for i, r in enumerate(results, 1)
        ]
        search_results = "\n\n".join(formatted)
        sources = [r['url'] for r in results]
    except Exception as e:
        search_results = f"Search failed: {e}"
        sources = []
    return {"search_results": search_results, "sources": sources}

def synthesize_node(state: AgentState) -> dict:
    compiled = synthesizer_prompt.compile(
        search_results=state["search_results"],
        question=state["question"],
        iteration=str(state["iterations"] + 1),
        max_iterations=str(state.get("max_iterations", 3)),
        max_words="500"
    )
    result = llm.invoke(compiled)
    return {"answer": result.content, "messages": [result]}

def critic_node(state: AgentState) -> dict:
    compiled = critic_prompt.compile(
        question=state["question"],
        answer=state["answer"],
        iteration=str(state["iterations"] + 1)
    )
    result = llm.invoke(compiled)
    try:
        parsed = json.loads(result.content)
        confidence = float(parsed.get("confidence", 0.8))
        needs_more = parsed.get("needs_more_research", False)
    except (json.JSONDecodeError, ValueError):
        confidence = 0.8
        needs_more = False
    should_loop = (
        confidence < 0.7
        and needs_more
        and state["iterations"] < state.get("max_iterations", 3)
    )
    return {
        "confidence": confidence,
        "iterations": state["iterations"] + 1,
        "should_loop": should_loop,
        "messages": [result]
    }

# ─── Graph assembly ────────────────────────────────────────────────────────────
workflow = StateGraph(AgentState)
workflow.add_node("router", router_node)
workflow.add_node("web_search", web_search_node)
workflow.add_node("synthesize", synthesize_node)
workflow.add_node("critic", critic_node)
workflow.set_entry_point("router")
workflow.add_conditional_edges("router",
    lambda s: s.get("route", "web"),
    {"web": "web_search", "kb": "synthesize"})
workflow.add_edge("web_search", "synthesize")
workflow.add_edge("synthesize", "critic")
workflow.add_conditional_edges("critic",
    lambda s: "router" if s.get("should_loop") else END,
    {"router": "router", END: END})
app = workflow.compile()

# ─── Public API ───────────────────────────────────────────────────────────────
def run_research_agent(
    question: str,
    user_id: str = "anonymous",
    session_id: str = None,
    max_iterations: int = 3,
    environment: str = "production"
) -> dict:
    """
    Run the research agent with full Langfuse observability.

    Returns:
        {
            "answer": str,
            "confidence": float,
            "sources": list[str],
            "iterations": int,
            "trace_id": str,
            "trace_url": str
        }
    """
    session_id = session_id or str(uuid.uuid4())

    handler = CallbackHandler(
        user_id=user_id,
        session_id=session_id,
        tags=["research-agent", environment],
        metadata={
            "agent_version": "2.1",
            "environment": environment,
            "max_iterations": max_iterations
        },
        trace_name="research-agent"
    )

    result = app.invoke(
        input={
            "question": question,
            "messages": [],
            "route": None,
            "search_results": "",
            "sources": [],
            "answer": "",
            "confidence": 0.0,
            "iterations": 0,
            "max_iterations": max_iterations
        },
        config={
            "callbacks": [handler],
            "recursion_limit": max_iterations * 5
        }
    )

    langfuse.flush()

    trace_id = handler.get_trace_id()

    return {
        "answer": result["answer"],
        "confidence": result["confidence"],
        "sources": result.get("sources", []),
        "iterations": result["iterations"],
        "trace_id": trace_id,
        "trace_url": f"https://cloud.langfuse.com/trace/{trace_id}"
    }

# ─── Example usage ────────────────────────────────────────────────────────────
if __name__ == "__main__":
    result = run_research_agent(
        question="What were the main causes of the 2023 banking crisis?",
        user_id="demo-user",
        environment="development"
    )

    print(f"\nAnswer:\n{result['answer']}")
    print(f"\nConfidence: {result['confidence']:.0%}")
    print(f"Sources: {', '.join(result['sources'][:3])}")
    print(f"Iterations: {result['iterations']}")
    print(f"\nView trace: {result['trace_url']}")

Summary and Key Takeaways

Building a LangGraph agent that works is the first challenge. Building one you can actually operate in production — debug, improve, and trust — is the second, harder challenge. Langfuse bridges that gap.

Here’s the mental model that makes everything click together:

Build your agent as a LangGraph state graph with well-named nodes and clean state transitions. Make each node do one thing.

Instrument by adding a CallbackHandler to your invoke call. One line of code gives you a complete trace of every run, forever.

Manage prompts in the Prompt Registry from day one. The separation between “what the agent does” (code) and “how it thinks” (prompts) pays dividends immediately — faster iteration, safer changes, clearer blame when things go wrong.

Evaluate continuously. Set up LLM-as-judge scoring on production traces within your first week. Build a gold dataset within your first month. Gate all prompt changes behind dataset evaluations before they reach production.

Operate with dashboards. Watch cost per trace, p95 latency, error rate, and score drift. Set alerts. Treat your agent like a service, not a script.

The feedback loop this creates — run → trace → score → improve prompt → experiment → promote — is what separates toy demos from agents that actually earn user trust over time.

Resources

• Langfuse documentation
• Langfuse GitHub repository — open source, MIT license
• LangGraph documentation
• Langfuse LangChain integration docs
• Prompt Registry guide
• Evaluation documentation
• Self-hosting guide

If this helped you build better agents, consider clapping 👏 and following for more content on LLM engineering, observability, and production AI systems.

All code in this article is Python 3.11+, tested with langfuse>=2.7, langgraph>=0.1, and langchain-openai>=0.1.