Lesson 08e: RAG Graph Service — LangGraph Deep Dive#

"Don't just retrieve and generate — check your work first."

Where This Lives#

backend/app/services/rag_graph_service.py (164 lines). This is the only file in the entire backend that uses LangGraph. If you remember one thing from this lesson, remember that — the multi-agent email workflow (Lesson 12) does not use LangGraph, despite what an earlier draft of this course claimed. This file does.

The Problem This File Solves#

Plain RAG (Retrieval-Augmented Generation) is: retrieve chunks → hand them to the LLM → generate an answer. The failure mode is that if retrieval comes back empty or irrelevant, the LLM still tries to answer, and it can hallucinate an answer that sounds confident but is fabricated.

Corrective RAG adds a grading step between retrieval and generation: an LLM call that judges whether the retrieved chunks actually contain the answer. If they don't, the pipeline refuses instead of generating. This file implements that as an explicit state machine using LangGraph's StateGraph.

The State: RAGGraphState#

class RAGGraphState(TypedDict):
    agent_id: int
    agent_name: str
    question: str
    top_k: int
    language: str
    tone: str
 
    retrieved_chunks: list[dict[str, Any]]
    retrieval_evaluation: dict[str, Any] | None
    answer: str
    sources: list[dict[str, Any]]
 
    route: str

A TypedDict, not a Pydantic model — LangGraph nodes are plain functions that take the current state dict and return a partial dict of updates, which LangGraph merges into the running state. The first six fields are inputs, fixed for the whole run. The rest are filled in as the graph executes.

The Four Nodes#

retrieve_node#

def retrieve_node(state: RAGGraphState) -> dict:
    results = search_agent_documents(
        agent_id=state["agent_id"], query=state["question"], top_k=state["top_k"],
    )
    sources = [...]
    return {"retrieved_chunks": results, "sources": sources}

Calls straight into document_service.search_agent_documents() — the same Chroma query described in the previous lesson. Returns updates for two state keys.

evaluate_node — the "corrective" part#

def evaluate_node(state: RAGGraphState) -> dict:
    retrieved_chunks = state.get("retrieved_chunks", [])
    if not retrieved_chunks:
        return {
            "retrieval_evaluation": {"has_answer": False, "confidence": 0, "reason": "No chunks were retrieved."},
            "route": "refuse",
        }
    evaluation = evaluate_retrieved_context(question=state["question"], contexts=retrieved_chunks)
    route = "answer" if evaluation.get("has_answer") else "refuse"
    return {"retrieval_evaluation": evaluation, "route": route}

Two paths here. If retrieval came back empty, the function short-circuits and routes to "refuse" without calling the LLM at all — this both saves an API call and, as a side effect, guarantees evaluate_retrieved_context (and later generate_rag_answer) are never invoked with an empty context list. Otherwise, it delegates the actual judgment to openai_service.evaluate_retrieved_context(), which asks an LLM: "does this context actually answer the question?" and returns has_answer, confidence, and reason. The route key is what the conditional edge below reads.

answer_node and refusal_node#

def answer_node(state: RAGGraphState) -> dict:
    answer = generate_rag_answer(question=state["question"], contexts=state["retrieved_chunks"], ...)
    return {"answer": answer}
 
def refusal_node(state: RAGGraphState) -> dict:
    return {"answer": "I could not find this information in the uploaded documents."}

answer_node calls openai_service.generate_rag_answer() to actually produce a grounded answer. refusal_node is a pure function — no I/O, no LLM call, just a canned honest refusal. This is the entire point of Corrective RAG: refusing is cheap and safe; hallucinating is not.

The Router Function#

def route_after_evaluation(state: RAGGraphState) -> Literal["answer_node", "refusal_node"]:
    if state.get("route") == "answer":
        return "answer_node"
    return "refusal_node"

This is not a graph node — it's a plain function LangGraph calls after evaluate_node runs, to decide which node executes next. It reads the route key that evaluate_node wrote.

Wiring the Graph#

def build_corrective_rag_graph():
    graph = StateGraph(RAGGraphState)
 
    graph.add_node("retrieve_node", retrieve_node)
    graph.add_node("evaluate_node", evaluate_node)
    graph.add_node("answer_node", answer_node)
    graph.add_node("refusal_node", refusal_node)
 
    graph.add_edge(START, "retrieve_node")
    graph.add_edge("retrieve_node", "evaluate_node")
 
    graph.add_conditional_edges(
        "evaluate_node",
        route_after_evaluation,
        {"answer_node": "answer_node", "refusal_node": "refusal_node"},
    )
 
    graph.add_edge("answer_node", END)
    graph.add_edge("refusal_node", END)
 
    return graph.compile()

graph.add_conditional_edges(source, router_fn, mapping) is the LangGraph primitive for branching: after evaluate_node finishes, LangGraph calls route_after_evaluation(state), gets back a string, and looks that string up in mapping to decide the next node. Every path — success or refusal — converges on END.

The Module-Level Singleton#

corrective_rag_graph = build_corrective_rag_graph()

The graph is built and compiled once, at import time, and reused for every request. compile() validates the graph structure (no dangling nodes, no missing edges) and produces a runnable object — you don't want to pay that cost on every HTTP request.

run_corrective_rag_graph() — the Entry Point#

def run_corrective_rag_graph(agent_id, agent_name, question, top_k, language="English", tone="Professional") -> dict:
    initial_state: RAGGraphState = { ... all fields, with empty/default values for the ones nodes fill in ... }
    final_state = corrective_rag_graph.invoke(initial_state)
    return { "question": question, "answer": final_state.get("answer"), "sources": final_state.get("sources", []), ... }

This is the only function routers/documents.py (specifically the POST /agents/{id}/documents/ask endpoint) actually calls. It builds a fully-populated initial state (LangGraph requires every key the TypedDict declares, even if the value is a placeholder like "" or []), calls .invoke() to run the graph synchronously to completion, and reshapes the final state into the response the API returns.

How This Differs From the Multi-Agent Email Workflow#

Common Mistakes#

• Assuming every LangGraph-flavored workflow in this codebase uses StateGraph. Only this one does.
• Calling evaluate_retrieved_context directly instead of through evaluate_node. You'd lose the empty-chunks short-circuit and could trigger the latent empty-context bug in generate_rag_answer.
• Forgetting initial_state needs every TypedDict key populated. LangGraph does not silently default missing keys the way Pydantic can.

Exercises#

1. Add a fifth node, low_confidence_node, that returns a hedged answer (e.g., "This might be right, but double-check: ...") when retrieval_evaluation["confidence"] is between 0.3 and 0.6. Update route_after_evaluation and the conditional edge mapping accordingly.
2. retrieve_node always fetches top_k chunks even if top_k=0. Trace what evaluate_node does in that case.
3. Draw the graph for a non-corrective RAG pipeline (no evaluation step) and explain, in one paragraph, what class of answer-quality bug this codebase avoids by not doing that.

Key Takeaways#

🎯 This file is the only real LangGraph usage in the backend — a StateGraph with conditional routing. 🎯 "Corrective" means grading retrieval quality before generating, and refusing rather than hallucinating. 🎯 The graph is compiled once at import time and reused across requests. 🎯 The empty-retrieval short-circuit in evaluate_node doubles as a bug-prevention measure for generate_rag_answer.

Next Steps#

👉 Lesson 09: API Routers Overview →

Module 3 · Services Deep Dive