Production RAG with LangGraph and Langfuse: Part 1, Every Stage as a Traceable Graph

Your RAG assistant works in the demo. A support engineer asks "how do I rotate an API key?", it pulls the right doc, and the answer is clean. Two weeks after launch, a customer asks the same thing and gets a confident answer about a feature that was deprecated last year. You open your logs and find a single line: one function call in, one string out. You cannot tell whether retrieval pulled the wrong document, whether the reranker buried the right one, or whether the prompt was fine and the model ignored its context.

That is the real problem this guide solves. RAG is not one step. It is a pipeline of six or seven stages, and any of them can fail silently. When the whole pipeline is a single opaque function, debugging is guesswork. When it is a graph of named nodes with a trace span on each one, debugging is reading.

This is a two-part series. Part 1 builds a production RAG system as a LangGraph state graph and wires Langfuse into every stage. Part 2 does the same for tool-using agents. We use one concrete example throughout Part 1: a customer-support assistant answering over a company's product docs and resolved support tickets.

Why a graph instead of a chain?#

The simplest RAG implementation is a straight line: embed the query, search, stuff the results into a prompt, call the model. You can write that as a single function or an LCEL chain, and for a throwaway demo you should. The problem is that production RAG is not a straight line. It has branches ("did we retrieve enough good context, or should we rewrite the query and try again?") and loops ("the answer isn't grounded in the sources, go back and retrieve differently").

LangGraph models computation as a state graph: a set of nodes that read and write a shared state object, connected by edges that can be conditional. This is the right shape for RAG because it makes three things explicit that a chain hides: the state that flows between stages, the branching decisions, and the boundaries of each stage. Those boundaries are exactly what you want to trace.

Here is the support assistant's query-time graph. Each box is a node, the solid edges are the happy path, and the dashed edges are the two corrective loops that re-query on weak retrieval or an ungrounded answer.

The core building blocks are StateGraph (the container), nodes (functions that transform state), edges (START, END, and conditional routers), and .compile() (which turns the definition into a runnable and attaches a checkpointer). The official mental model is documented in the LangGraph docs.

What LangGraph gives you (and where each piece fits)#

Before the walkthrough, here is the whole toolbox in one place. Each row is a LangGraph primitive, the RAG phase where it earns its place, and what it actually does for you. Treat this table as the roadmap: every phase section below maps back to a row here.

The rest of Part 1 walks these phases in order. Read it top to bottom the first time; after that, jump to the row you need.

Phase 0: Design the state#

The state is the most important decision in the whole graph, and the easiest to get wrong. It is the typed object every node receives and updates. Get it right and each node is a small, testable function. Get it wrong and you end up smuggling data through closures and globals.

For our support assistant the state holds the original question, a possibly-rewritten query, the retrieved documents, a grounding verdict, and the final answer. In LangGraph you declare it as a TypedDict, and you can attach a reducer to any field to control how updates merge, for example appending to a list instead of overwriting it.

from typing import TypedDict, Annotated, Literal
from operator import add

class RAGState(TypedDict):
    question: str
    search_query: str
    documents: Annotated[list, add]   # reducer: accumulate across retries
    grounded: bool
    retries: int
    answer: str

The Annotated[list, add] is the key detail. Because the support assistant may retrieve more than once (corrective loop in Phase 6), we want each retrieval to add documents to the running set rather than replace it. Reducers are where LangGraph encodes "how should concurrent or repeated writes to this field combine," and they are documented under state reducers.

Phase 1: Ingestion and indexing#

Ingestion (loading documents, chunking, embedding, and writing to a vector store) happens offline, before any query. LangGraph does not own this phase; your ingestion job runs on a schedule and populates the index. We cover the mechanics of chunking, embedding models, and index choice in depth in RAG From Development to Monitoring, Part 1 and Build a RAG Search Pipeline, so we will not repeat them here.

What matters for the graph is the retriever interface ingestion produces. Our support assistant indexes two corpora (product docs and resolved support tickets) into a vector store, exposed as a LangChain retriever. The query-time graph only needs that retriever; everything upstream of it is a separate pipeline.

from langchain_anthropic import ChatAnthropic
from langchain_openai import OpenAIEmbeddings
from langchain_postgres import PGVector

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = PGVector(
    embeddings=embeddings,
    collection_name="support_kb",
    connection="postgresql+psycopg://...",
)
retriever = vectorstore.as_retriever(search_kwargs={"k": 8})

llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)

We use Claude Sonnet 4.6 (claude-sonnet-4-6) as the generation model. It is fast and strong enough for grounded support answers, while leaving Opus for the harder reasoning in Part 2.

Phase 2: Query understanding#

A user's raw question is often a poor search query. "It's broken after the update, help" has almost no retrievable signal. The first node in the query-time graph rewrites the question into a clean search query, optionally expanding it. This is a small LLM call, and it is the first thing you will want to see in a trace when retrieval misses.

def rewrite_query(state: RAGState) -> dict:
    prompt = (
        "Rewrite the user's support question into a focused search query "
        "for a product documentation and ticket knowledge base. "
        "Keep product names and error terms. Question: " + state["question"]
    )
    rewritten = llm.invoke(prompt).content
    return {"search_query": rewritten, "retries": 0}

A node is just a function that takes state and returns a partial update. LangGraph merges that dict into the state using the reducers from Phase 0. The conditional routing on which retrieval strategy to use (docs vs tickets vs both) can also live here, expressed later as a conditional edge.

Phase 3: Retrieval#

Retrieval is where most "the answer was wrong" bugs actually live, and it is invisible without tracing. The node runs the search and writes the documents into state.

def retrieve(state: RAGState) -> dict:
    docs = retriever.invoke(state["search_query"])
    return {"documents": docs}

That looks trivial, but the trace is anything but. Once Langfuse is wired (Phase 9), this node's span shows you the exact query that was searched, the top-k documents returned, and their similarity scores. When a customer gets a wrong answer, you open the trace and immediately see whether the right document was even in the retrieved set. If it was not, the bug is in retrieval or indexing; if it was, the bug is downstream. That single distinction saves hours.

For the support assistant we use hybrid search (keyword + vector) because ticket text is full of exact error codes that pure semantic search blurs. The mechanics are covered in Experimenting with Hybrid Search; here it is just a different retriever behind the same node.

Phase 4: Reranking and filtering#

The top 8 vector hits are rarely in the right order. A reranker (a cross-encoder that scores each document against the query directly) reorders them so the most relevant land first, and lets you drop the rest. This is also the first place a conditional edge appears: after reranking, the graph decides whether the context is good enough to answer, or whether it should rewrite and retry.

from langchain_cohere import CohereRerank

reranker = CohereRerank(model="rerank-english-v3.0", top_n=4)

def rerank(state: RAGState) -> dict:
    reranked = reranker.compress_documents(
        documents=state["documents"], query=state["search_query"]
    )
    return {"documents": reranked}

def enough_context(state: RAGState) -> Literal["generate", "rewrite_query"]:
    top_score = state["documents"][0].metadata.get("relevance_score", 0)
    if top_score < 0.5 and state["retries"] < 2:
        return "rewrite_query"   # weak context: loop back and try again
    return "generate"

enough_context is a router: it returns the name of the next node. We wire it with add_conditional_edges when we assemble the graph. This is the difference between a pipeline that gives up and one that recovers, covered in the conditional edges docs.

Phase 5: Generation#

Now the model answers, grounded in the reranked context. The prompt's job is to force grounding: answer only from the provided sources, cite them, and say "I don't know" when the sources don't cover the question. This last instruction is what stops the confident-but-wrong answers from the intro.

def generate(state: RAGState) -> dict:
    context = "\n\n".join(
        f"[{i+1}] {d.page_content}" for i, d in enumerate(state["documents"])
    )
    prompt = (
        "Answer the support question using ONLY the sources below. "
        "Cite sources as [1], [2]. If the sources do not answer it, say so.\n\n"
        f"Sources:\n{context}\n\nQuestion: {state['question']}"
    )
    answer = llm.invoke(prompt).content
    return {"answer": answer}

In the Langfuse trace this node is a generation span: it records the full prompt, the completion, the model name, token counts, and cost. When you later ask "why is this assistant expensive," the answer is a sum over these spans.

Phase 6: Grounding check (corrective RAG)#

This is the phase that separates a hello-world RAG from a production one. After generation, a second model call grades whether the answer is actually supported by the retrieved documents. If it is not, the graph loops back to retrieve with a different query rather than shipping a hallucination. This pattern is Corrective RAG (Yan et al., arXiv:2401.15884), and the related Self-RAG (arXiv:2310.11511) formalizes the self-grading idea.

def grade_grounding(state: RAGState) -> dict:
    check = (
        "Is the ANSWER fully supported by the SOURCES? Reply 'yes' or 'no'.\n\n"
        f"SOURCES:\n{[d.page_content for d in state['documents']]}\n\n"
        f"ANSWER: {state['answer']}"
    )
    verdict = llm.invoke(check).content.strip().lower()
    return {"grounded": verdict.startswith("yes")}

def route_after_grading(state: RAGState) -> Literal["__end__", "rewrite_query"]:
    if state["grounded"] or state["retries"] >= 2:
        return "__end__"
    return "rewrite_query"   # ungrounded: correct course

In the trace, this loop is visible as repeated retrieve→generate→grade cycles under one parent trace, so you can see exactly how many corrections a hard question took. Now we assemble everything:

from langgraph.graph import StateGraph, START, END

builder = StateGraph(RAGState)
builder.add_node("rewrite_query", rewrite_query)
builder.add_node("retrieve", retrieve)
builder.add_node("rerank", rerank)
builder.add_node("generate", generate)
builder.add_node("grade_grounding", grade_grounding)

builder.add_edge(START, "rewrite_query")
builder.add_edge("rewrite_query", "retrieve")
builder.add_edge("retrieve", "rerank")
builder.add_edge("generate", "grade_grounding")

builder.add_conditional_edges("rerank", enough_context)
builder.add_conditional_edges("grade_grounding", route_after_grading)

from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(checkpointer=MemorySaver())

Phase 7: Persistence and memory#

Support conversations are multi-turn: "how do I rotate a key?" is often followed by "and does that invalidate the old one immediately?" The second question only makes sense with the first in context. LangGraph handles this with checkpointers, which persist the full state after every node, keyed by a thread_id. Pass the same thread id and the graph resumes where it left off; the conversation has memory for free.

config = {"configurable": {"thread_id": "support-session-8842"}}
graph.invoke({"question": "How do I rotate an API key?"}, config)
graph.invoke({"question": "Does that invalidate the old one immediately?"}, config)

MemorySaver keeps state in memory and is fine for development. In production you swap it for PostgresSaver, which persists checkpoints to your database so a server restart does not lose conversations and so any worker can resume any thread. The interface is identical (only the constructor changes), and it is documented under persistence.

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    checkpointer.setup()
    graph = builder.compile(checkpointer=checkpointer)

For knowledge that should outlive a single conversation (say, a customer's preferred product tier so you bias retrieval toward the right docs), LangGraph offers the store, a separate long-term key-value layer. Checkpointers are short-term per-thread memory; the store is cross-session memory. We use the store more heavily for the agent in Part 2.

Phase 8: Streaming#

A support answer that takes four seconds to appear feels broken even when it is correct. LangGraph streams in several modes via .stream(): messages streams the model's tokens as they generate, updates streams each node's output as it completes, and values streams the full state after each step. For a UI you typically use messages to show the answer appearing, and updates to show "Searching docs… Reranking… Writing answer."

for chunk in graph.stream(
    {"question": "How do I rotate an API key?"},
    config,
    stream_mode="updates",
):
    print(chunk)   # {'retrieve': {...}}, then {'rerank': {...}}, ...

The streaming modes are documented under streaming. The same intermediate steps you stream to the user are the ones Langfuse records as spans. Streaming and tracing are two views of the same node boundaries.

Phase 9: Observability with Langfuse#

Everything above produced clean node boundaries. Langfuse turns those boundaries into a trace tree with one line of wiring. Langfuse provides a LangChain callback handler; you pass it in the run config and it captures every node, every LLM call, and the full input/output of each as a nested span. There is no per-node instrumentation to write.

from langfuse.langchain import CallbackHandler

langfuse_handler = CallbackHandler()   # reads keys from environment

config = {
    "configurable": {"thread_id": "support-session-8842"},
    "callbacks": [langfuse_handler],
    "metadata": {"langfuse_user_id": "cust_4471", "langfuse_session_id": "support-session-8842"},
}
graph.invoke({"question": "How do I rotate an API key?"}, config)

The handler is the current Langfuse v3 integration; you set LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and LANGFUSE_HOST in the environment, and it sends traces over OpenTelemetry. The setup is in the Langfuse LangChain docs. One trace now shows the whole query: rewrite → retrieve (with scores) → rerank → generate (with tokens and cost) → grade, and every corrective loop nested underneath.

What you get on top of free tracing is evaluation. You can attach scores to any trace (a faithfulness or answer-relevance grade from an LLM-as-judge, or a thumbs-up from the user) and Langfuse aggregates them over time so you can see quality trend, not just one answer. And you can build a dataset of representative questions and run the whole graph against it as an offline evaluation before every release, catching regressions before users do. We go deep on RAG evaluation methodology in Evaluating RAG Quality and Add Evaluation to an AI Feature.

Phase 10: Productionizing#

Three things separate a graph that demos from one that runs:

Durability and retries. Network calls to your vector store, reranker, and model all fail occasionally. LangGraph lets you attach a retry policy to individual nodes so a transient failure does not kill the whole request, and the checkpointer means a crash mid-graph can resume from the last completed node rather than restarting. See durable execution.

Cost and latency control. Every LLM node costs money and time. With Langfuse spans you can see that the grounding check in Phase 6 doubles your model calls, a real tradeoff you can now quantify and decide on per-query (for example, skip the check for high-confidence retrievals). Without the trace, that cost is invisible.

Deployment. You can self-host the compiled graph behind any web framework, or deploy it on LangGraph Platform, which manages the checkpointer, scaling, and a built-in API. Either way the graph definition is unchanged. That portability is a real benefit of keeping orchestration in LangGraph rather than scattered across your application code.

Common mistakes and tradeoffs#

The honest tradeoff: a graph is more machinery than a one-liner. For a true linear pipeline with no branching, a chain is less code. The moment you need a conditional ("retry if retrieval is weak") or a loop ("re-query if ungrounded"), which every production RAG eventually does, the graph stops being overhead and starts being the only sane way to keep the logic, and its observability, legible.

Key Takeaways#

• RAG is a pipeline of stages, and any stage can fail silently. Modeling it as a LangGraph state graph makes each stage a named node you can inspect.
• State design is the foundational decision. The fields you carry between nodes are also the fields you will read in a trace when something breaks.
• The conditional edge is what makes RAG production-grade: re-query on weak retrieval, and loop back on ungrounded answers (corrective RAG) instead of shipping a hallucination.
• Checkpointers give multi-turn memory and durability for free; use MemorySaver in dev and PostgresSaver in production.
• Langfuse wires in once at the callback layer and turns every node into a span with inputs, outputs, latency, and cost, plus scores and datasets for evaluation.
• Wire observability before you tune. Until you can watch a real query flow through the graph, every parameter change is a guess.

FAQ#

Do I really need LangGraph for simple RAG?

No. For a genuinely linear retrieve-then-generate flow with no branching, a plain function or an LCEL chain is less code. LangGraph pays off the moment you add a conditional (retry on weak retrieval) or a loop (corrective RAG), which most production systems eventually need.

What is the difference between LangGraph and LCEL?

LCEL composes linear chains of runnables. LangGraph models stateful graphs with branches, loops, and persistence. They are complementary: a LangGraph node can internally be an LCEL chain. Use LCEL for the straight segments and LangGraph for the control flow around them.

Should I use Langfuse or LangSmith?

Both trace LangGraph well. Langfuse is open-source and self-hostable, which matters if you need data to stay in your infrastructure; it also has strong prompt management and dataset-based evaluation. LangSmith is the first-party LangChain option with tight ecosystem integration. The LangGraph code is identical, and only the callback handler changes.

How much does tracing cost in latency?

The Langfuse handler sends spans asynchronously over OpenTelemetry, so it adds negligible latency to the request path. The real cost is storage and the LLM calls for any LLM-as-judge scores you choose to run, which you control.

Should the grounding check run on every query?

It doubles your model calls, so on cost-sensitive systems, gate it: skip it when retrieval confidence is high and only run it when the top reranked score is borderline. Because Langfuse shows you the cost of that node per query, you can make this tradeoff with data instead of guessing.

How do I evaluate the whole graph, not just one answer?

Build a Langfuse dataset of representative questions with expected answers, then run the compiled graph against it on every release. Attach faithfulness and relevance scores to each run so you catch regressions before deploying. See Evaluating RAG Quality for the methodology.