Practical notes: Build a Self-Correcting AI Agent with Self-RAG & LangGraph
Operable walkthrough of Practical notes: Build a Self-Correcting AI Agent with Self-RAG & LangGraph: contracts, checks, and drop-in code slots for teams shipping this pattern.
This article is published in English.
Sources: Build a Self-Correcting AI Agent with Self-RAG & LangGraph
This walkthrough rebuilds the path from raw materials to a working system for: Build a Self-Correcting AI Agent with Self-RAG & LangGraph. The focus is operable steps, explicit checks, and code that you can drop into a repo without guessing intent.
Why Standard RAG Pipelines Are Brittle
For Why Standard RAG Pipelines Are Brittle, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
The Linear Pipeline: Retrieve and Assume
For The Linear Pipeline: Retrieve and Assume, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
[User Query] ──> [Vector Database] ──> [Top-K Chunks] ──> [LLM Prompt] ──> [Final Answer]
Anatomy of a Silent Failure
For Anatomy of a Silent Failure, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
The Root Cause: Blind Trust
For The Root Cause: Blind Trust, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
The Solution: Introducing Self-Reflective Loops
For The Solution: Introducing Self-Reflective Loops, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap. For The Solution: Introducing Self-Reflective Loops, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
From Blind Retrieval to Self-Reflection: The Self-RAG Pattern
When working through From Blind Retrieval to Self-Reflection: The Self-RAG Pattern, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
The Three Pillars of Self-Correction
When working through The Three Pillars of Self-Correction, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
[ Query ]
│
▼
1. Is Retrieval Needed? ──(No)──► Generate directly from parametric memory
│
(Yes)
▼
[ Retrieve ]
│
▼
2. Are Docs Relevant? ──(No)──► Reformulate search query & retry
│
(Yes)
▼
[ Generate ]
│
▼
3. Is Answer Faithful? ──(No)──► Regenerate or extract better context
│
(Yes)
▼
[ Final Output ]
Reflection Tokens as Control Signals
When working through Reflection Tokens as Control Signals, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Cache stable system instructions and tool schemas. Re-sending identical preamble is a common source of burn. When working through Reflection Tokens as Control Signals, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
The Architecture of Self-Reflection
The Architecture of Self-Reflection works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
graph TD
Start[User Query] --> DecideNeed{Is Retrieval Needed?}
DecideNeed -- No --> GenerateDirect[Generate Output]
DecideNeed -- Yes --> Retrieve[Retrieve Documents]
Retrieve --> GradeDocs[Grade Document Relevance]
GradeDocs -- Irrelevant --> Fallback[Rewrite Query / Fallback]
GradeDocs -- Relevant --> GenerateRef[Generate Response with Context]
GenerateRef --> GradeFaith[Grade Faithfulness & Hallucinations]
GradeFaith -- Hallucinated --> GenerateRef
GradeFaith -- Faithful --> GradeUtility[Grade Utility]
GradeUtility -- Low Utility --> Rewrite[Refine Response]
GradeUtility -- High Utility --> End[Final Answer]
Comparing Pipelines and Workflows
Comparing Pipelines and Workflows works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
Architecting with LangGraph: From Linear Chains to State Machines
Architecting with LangGraph: From Linear Chains to State Machines works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move. Architecting with LangGraph: From Linear Chains to State Machines works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
Linear RAG: [Query] ──> [Retrieve] ──> [Generate] ──> [Response] (No correction)
Self-RAG: [Query] ──> [Retrieve] ──> [Evaluate] ──┬─ (Valid) ──> [Response]
▲ │
└─────── (Invalid) ──────┘ (Loop & Refine)
The Limits of Linear Chains
For The Limits of Linear Chains, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
Demystifying the AI State Machine
For Demystifying the AI State Machine, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
┌──────────────────────┐
│ Start │
└──────────┬───────────┘
▼
┌──────────────┐ ┌──────────────────────┐
│ Formulate │◄─┤ Retrieve Docs ├─┐
│ New Query │ └──────────────────────┘ │
└──────▲───────┘ │
│ ▼
┌──────┴───────┐ ┌──────────────────────┐ │
│ Re-Retrieve │ │ Grade Relevance │─┘
└──────────────┘ └──────────┬───────────┘
▼
┌──────────────────────┐
│ Generate Answer │
└──────────┬───────────┘
▼
┌──────────────────────┐
│ Grade Hallucination │
└──────────────────────┘
Inside the LangGraph Core Components
For Inside the LangGraph Core Components, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap. For Inside the LangGraph Core Components, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
Implementing the Central Graph State
When working through Implementing the Central Graph State, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
from typing import List, TypedDict
class GraphState(TypedDict):
"""
Represents the central state of our Self-RAG agent.
This TypedDict serves as the single source of truth passed between
each node in the LangGraph execution flow.
"""
question: str # The original user query
generation: str # The most recent answer generated by the LLM
documents: List[str] # Relevant text passages retrieved from our database
steps: List[str] # An audit trail tracking the sequence of execution steps
Building the Graph, Part 1: Defining State and Action Nodes
When working through Building the Graph, Part 1: Defining State and Action Nodes, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
Understanding State-Based Orchestration
When working through Understanding State-Based Orchestration, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface. When working through Understanding State-Based Orchestration, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
+-------------------------------------------------------------+
| GRAPH STATE |
| - question: str |
| - documents: List[Document] |
| - generation: str |
| - relevance_grade: str |
+-------------------------------------------------------------+
|
+----------------------+----------------------+
| | |
v v v
+--------------+ +--------------+ +--------------+
| Retrieve | | Grade Docs | | Generate |
| Node | | Node | | Node |
+--------------+ +--------------+ +--------------+
Designing the Shared State Ledger
Designing the Shared State Ledger works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
from typing import List, Dict, Any
from typing_extensions import TypedDict
from langchain_core.documents import Document
class GraphState(TypedDict):
"""
Represents the state of our Self-RAG graph.
Attributes:
question: The user query string.
documents: A list of retrieved document chunks.
generation: The generated answer from the LLM.
relevance_grade: The evaluation score of our documents ('yes' or 'no').
"""
question: str
documents: List[Document]
generation: str
relevance_grade: str
The Retrieve Node: Interfacing with Knowledge Bases
The Retrieve Node: Interfacing with Knowledge Bases works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_core.embeddings import FakeEmbeddings
# Initialize a mock vector store to simulate a production knowledge base
embeddings = FakeEmbeddings(size=1536)
vectorstore = InMemoryVectorStore(embeddings)
# Seed our vector store with sample documentation chunks
vectorstore.add_documents([
Document(page_content="LangGraph is a library for building stateful, multi-actor applications with LLMs.", metadata={"source": "docs"}),
Document(page_content="Self-RAG is an architecture that introduces self-reflection and grading into retrieval chains.", metadata={"source": "paper"}),
Document(page_content="Vector databases index text embeddings to enable fast semantic similarity search.", metadata={"source": "tech-blog"})
])
# Create a retriever interface
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})
def retrieve(state: GraphState) -> Dict[str, Any]:
"""
Retrieve documents from the vector database based on the current question.
Args:
state (GraphState): The current state of the graph.
Returns:
dict: An update to the state containing the retrieved documents.
"""
print("--- NODE: RETRIEVING DOCUMENTS ---")
question = state["question"]
# Perform the semantic vector search
retrieved_docs = retriever.invoke(question)
# Return the updated document list to the graph state
return {"documents": retrieved_docs}
The Generate Node: Context-Aware Synthesis
The Generate Node: Context-Aware Synthesis works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move. The Generate Node: Context-Aware Synthesis works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.language_models.fake import FakeListLLM
# Set up a mock LLM that simulates generating structured responses
# In production, replace this with: from langchain_openai import ChatOpenAI
llm = FakeListLLM(responses=[
"Self-RAG is an advanced architecture that uses self-reflection and document grading to improve retrieval accuracy."
])
def generate(state: GraphState) -> Dict[str, Any]:
"""
Generate an answer using the retrieved documents as grounding context.
Args:
state (GraphState): The current state of the graph.
Returns:
dict: An update to the state containing the generated answer.
"""
print("--- NODE: GENERATING ANSWER ---")
question = state["question"]
documents = state["documents"]
# Format the retrieved documents into a single context block
context = "\n\n".join([doc.page_content for doc in documents])
# Construct the instruction prompt
prompt = ChatPromptTemplate.from_template(
"You are an assistant for question-answering tasks.\n"
"Use the following pieces of retrieved context to answer the question.\n"
"If you do not know the answer, say that you do not know.\n"
"Question: {question}\n"
"Context: {context}\n"
"Answer:"
)
# Format the prompt with our state variables and run the model
chain = prompt | llm
generation = chain.invoke({"question": question, "context": context})
# Save the generated response to the state
return {"generation": generation}
The Grade Documents Node: The Quality Inspector
For The Grade Documents Node: The Quality Inspector, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
+-------------------+
| Retrieved Docs |
+-------------------+
|
v
+-----------------------+
| Is Doc Relevant? |
+-----------------------+
/ \
Yes / \ No
/ \
v v
+-------------------+ +------------------+
| Keep in State | | Filter / Discard |
| (Pass to Gen) | | (Trigger Loop) |
+-------------------+ +------------------+
from pydantic import BaseModel, Field
from langchain_core.output_parsers import JsonOutputParser
class DocumentGrade(BaseModel):
"""Binary score for document relevance assessment."""
binary_score: str = Field(
description="Relevance score for the document. Must be 'yes' or 'no'."
)
def grade_documents(state: GraphState) -> Dict[str, Any]:
"""
Assess whether the retrieved documents are relevant to the user question.
Filters out any irrelevant documents from the state.
Args:
state (GraphState): The current state of the graph.
Returns:
dict: Updated documents list and a global relevance grade flag.
"""
print("--- NODE: GRADING RETRIEVED DOCUMENTS ---")
question = state["question"]
documents = state["documents"]
# Use a mock parser to simulate LLM grading
# In production, use: llm.with_structured_output(DocumentGrade)
parser = JsonOutputParser(pydantic_object=DocumentGrade)
filtered_docs = []
has_relevant_doc = "no"
# Evaluate each document individually
for doc in documents:
# Construct a strict grading prompt
grade_prompt = (
f"Question: {question}\n"
f"Document: {doc.page_content}\n"
"Assess if this document is relevant to the question. Output JSON with key 'binary_score' as 'yes' or 'no'."
)
# Simulating LLM evaluation logic (in production, run the LLM chain here)
# For our mock run, if the query terms overlap with the document, mark 'yes'
is_relevant = any(term.lower() in doc.page_content.lower() for term in question.split())
score = "yes" if is_relevant else "no"
if score == "yes":
print(f" -> DOCUMENT GRADING: [RELEVANT] - Source: {doc.metadata.get('source')}")
filtered_docs.append(doc)
has_relevant_doc = "yes"
else:
print(f" -> DOCUMENT GRADING: [NOT RELEVANT] - Source: {doc.metadata.get('source')}")
# Update state with filtered documents and set the control flag
return {
"documents": filtered_docs,
"relevance_grade": has_relevant_doc
}
Building the Graph, Part 2: Wiring Decisions with Conditional Edges
For Building the Graph, Part 2: Wiring Decisions with Conditional Edges, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
+------------------+
| route_question |
+--------+---------+
|
+--------------+--------------+
| |
[web_search] [retrieve]
| |
+--------------> +------------+-------------+
| grade_documents |
+------------+--------------+
|
+-----+-----+
| Decision |
+-----+-----+
|
+-----------------+-----------------+
| |
(transform_query) (generate)
^ |
| +------+------+
| | Decision |
| +------+------+
| |
+--- [hallucination / unsat] -------+------> [__end__]
The Entry Point Router: Inspecting the Query First
For The Entry Point Router: Inspecting the Query First, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap. For The Entry Point Router: Inspecting the Query First, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
# router.py
from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from state import AgentState
# Define the structured output schema for our router
class RouteQuery(BaseModel):
"""Route a user query to the most authoritative datasource."""
datasource: Literal["vectorstore", "web_search"] = Field(
...,
description="Given a user question choose to route it to web search or a vectorstore.",
)
def route_question(state: AgentState) -> str:
"""
Routes the initial question to either the vectorstore retrieval node or web search.
Args:
state (AgentState): The current graph state containing the user query.
Returns:
str: The next node to execute ("vectorstore" or "web_search").
"""
print("--- ROUTING QUESTION ---")
question = state["question"]
# Initialize our LLM with structured output mapping to our Pydantic schema
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_router = llm.with_structured_output(RouteQuery)
# Create the routing system prompt
system_prompt = """You are an expert router directing user questions to a vectorstore or web search.
The vectorstore contains highly technical documentation regarding LangGraph, LangChain, and state machines.
Use the vectorstore for these specialized topics. Otherwise, use web_search for general knowledge or current events."""
route_prompt = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{question}"),
])
# Run the router chain
router_chain = route_prompt | structured_router
decision = router_chain.invoke({"question": question})
if decision.datasource == "web_search":
print("--- ROUTING TO WEB SEARCH ---")
return "web_search"
print("--- ROUTING TO VECTORSTORE ---")
return "retrieve"
The Context Gate: Evaluating Document Relevance
When working through The Context Gate: Evaluating Document Relevance, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
# context_gate.py
from state import AgentState
def decide_to_generate(state: AgentState) -> str:
"""
Evaluates the document grading state to choose the next step.
If any documents were graded as relevant, we proceed to answer generation.
If all documents were deemed irrelevant, we rewrite the query to try again.
Args:
state (AgentState): The current state of the agent.
Returns:
str: Next node to execute ("generate" or "transform_query").
"""
print("--- ASSESSING DOCUMENT RELEVANCE ---")
# Retrieve the decision variable set by the grading node
run_web_search = state.get("run_web_search", False)
if run_web_search:
# No documents were relevant, we must rewrite and search the web
print("--- DECISION: ALL DOCUMENTS IRRELEVANT, ROUTING TO QUERY TRANSFORMATION ---")
return "transform_query"
# We found relevant documents, we can proceed safely to generation
print("--- DECISION: RELEVANT DOCUMENTS FOUND, ROUTING TO GENERATE ---")
return "generate"
The Self-Correction Loop: Verifying Faithfulness and Utility
When working through The Self-Correction Loop: Verifying Faithfulness and Utility, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
+-------------------------+
| Answer Generated |
+------------+------------+
|
v
+--------------------------------+
| Is the answer grounded in |
| retrieved context? |
+----------------+---------------+
|
+---------------+---------------+
| |
[No] [Yes]
| |
v v
+-------------------+ +---------------------+
| Re-route to: | | Is the answer |
| "generate" | | actually useful? |
+-------------------+ +----------+----------+
|
+--------------+--------------+
| |
[No] [Yes]
| |
v v
+-------------------+ +------------------+
| Re-route to: | | Re-route to: |
| "transform_query" | | "__end__" |
+-------------------+ +------------------+
# self_correction.py
from typing import Literal
from pydantic import BaseModel, Field
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from state import AgentState
class GradeHallucinations(BaseModel):
"""Binary score for hallucination present in generation."""
binary_score: Literal["yes", "no"] = Field(
...,
description="Answer is grounded in the facts, 'yes' or 'no'",
)
class GradeAnswer(BaseModel):
"""Binary score to assess if answer addresses question."""
binary_score: Literal["yes", "no"] = Field(
...,
description="Answer addresses the question, 'yes' or 'no'",
)
def grade_generation_v_documents_and_question(state: AgentState) -> str:
"""
Grades the generation for hallucinations and utility.
Args:
state (AgentState): The current graph state.
Returns:
str: Next step path ("supported" -> check utility, "not supported" -> regenerate, "not useful" -> rewrite query)
"""
print("--- CHECKING FOR HALLUCINATIONS ---")
question = state["question"]
documents = state["documents"]
generation = state["generation"]
# Initialize evaluator LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ----------------------------------------
# STEP 1: Groundedness (Hallucination) Check
# ----------------------------------------
hallucination_grader = llm.with_structured_output(GradeHallucinations)
hallucination_system_prompt = """You are an objective grader assessing whether an LLM generation is grounded in / supported by a set of retrieved facts.
Give a binary score of 'yes' if the generation is strictly grounded in and supported by the provided facts. Otherwise, give 'no'."""
hallucination_prompt = ChatPromptTemplate.from_messages([
("system", hallucination_system_prompt),
("human", "Set of facts: \n\n {documents} \n\n LLM generation: {generation}"),
])
# Format document contents into a single string context
doc_context = "\n\n".join([doc.page_content for doc in documents])
hallucination_chain = hallucination_prompt | hallucination_grader
res = hallucination_chain.invoke({"documents": doc_context, "generation": generation})
if res.binary_score == "no":
print("--- DECISION: HALLUCINATION DETECTED! RE-TRYING GENERATION ---")
return "generate"
print("--- DECISION: GENERATION IS GROUNDED IN DOCUMENTS ---")
# ----------------------------------------
# STEP 2: Utility (Answer Relevance) Check
# ----------------------------------------
answer_grader = llm.with_structured_output(GradeAnswer)
answer_system_prompt = """You are an objective grader assessing whether an LLM generation addresses and resolves the user's question.
Give a binary score of 'yes' if the answer directly addresses and completely answers the user's question. Otherwise, give 'no'."""
answer_prompt = ChatPromptTemplate.from_messages([
("system", answer_system_prompt),
("human", "User question: \n\n {question} \n\n LLM generation: {generation}"),
])
answer_chain = answer_prompt | answer_grader
res_answer = answer_chain.invoke({"question": question, "generation": generation})
if res_answer.binary_score == "yes":
print("--- DECISION: ANSWER IS USEFUL AND ACCURATE. ENDING GRAPH ---")
return "useful"
print("--- DECISION: ANSWER DOES NOT FULLY ADDRESS QUESTION. REWRITING QUERY ---")
return "not useful"
Assembling and Compiling the StateGraph
When working through Assembling and Compiling the StateGraph, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface. When working through Assembling and Compiling the StateGraph, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
# graph_builder.py
from langgraph.graph import END, StateGraph
from state import AgentState
# Import node functions from Part 1
from nodes import retrieve, generate, transform_query, web_search, grade_documents
# Import conditional routing edges
from router import route_question
from context_gate import decide_to_generate
from self_correction import grade_generation_v_documents_and_question
# Initialize our state-based graph
workflow = StateGraph(AgentState)
# ----------------------------------------
# 1. Register All Computational Nodes
# ----------------------------------------
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)
workflow.add_node("web_search", web_search)
# ----------------------------------------
# 2. Configure Node Routing & Transitions
# ----------------------------------------
# The entry point of our execution is dynamic, decided by the routing edge
workflow.set_conditional_entry_point(
route_question,
{
"web_search": "web_search",
"retrieve": "retrieve",
},
)
# After retrieval, we must immediately grade our retrieved documents
workflow.add_edge("retrieve", "grade_documents")
# Once documents are graded, the Context Gate decides where to go
workflow.add_conditional_edges(
"grade_documents",
decide_to_generate,
{
"transform_query": "transform_query",
"generate": "generate",
},
)
# If query transformation occurs, we execute a web search to gather new data
workflow.add_edge("transform_query", "web_search")
# After web searching, we feed those newly acquired documents back to the grade step
workflow.add_edge("web_search", "grade_documents")
# Once generation occurs, our Self-Correction edge decides the final quality check
workflow.add_conditional_edges(
"generate",
grade_generation_v_documents_and_question,
{
"generate": "generate", # Loop back to fix hallucination
"not useful": "transform_query", # Loop back to rewrite query and search
"useful": END, # Exit cleanly
},
)
# ----------------------------------------
# 3. Compile and Run the Graph
# ----------------------------------------
app = workflow.compile()
# visualize.py
from graph_builder import app
try:
# Print the plain text visualization of the state machine architecture
app.get_graph().print_ascii()
except Exception as e:
print(f"ASCII generation failed: {e}")
+----------------+
| __start__ |
+-------+--------+
|
+---------+---------+
v v
+------------+ +------------+
| web_search | | retrieve |
+-----+------+ +-----+------+
| |
| v
| +-----+------+
| | grade_docs |
| +-----+------+
| |
+---------+---------+
|
v
+---------+---------+
| decide_to_gen |
+----+---------+----+
| |
+----------+ +----------+
v v
+-----+------+ +-----+------+
| transform_q| | generate |
+-----+------+ +-----+------+
| |
| v
| +---------+---------+
| | self_correct |
| +----+----+----+----+
| | | |
+--------------------------+ | +---> [__end__]
v
(regenerate)
Engineering Synthesis
Engineering Synthesis works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
Real-World Applications
Real-World Applications works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
Advanced Customer Support Bots
Advanced Customer Support Bots works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move. Advanced Customer Support Bots works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
Automated Fact-Checking Systems
For Automated Fact-Checking Systems, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
Dynamic Recommendation Engines
For Dynamic Recommendation Engines, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
Scientific Research Assistants
For Scientific Research Assistants, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap. For Scientific Research Assistants, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
When Should You Use Which Approach?
When working through When Should You Use Which Approach?, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
Production Guardrails: Costs, Latency, and Observability
When working through Production Guardrails: Costs, Latency, and Observability, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface.
Balacing Cost and Latency with Asymmetric Routing
When working through Balacing Cost and Latency with Asymmetric Routing, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Measure recall on a fixed question set before tuning prompts. Prompt churn rarely fixes a weak retrieval surface. When working through Balacing Cost and Latency with Asymmetric Routing, write down the contract first: required inputs, success signal, and what happens on partial failure. That checklist keeps later code changes honest. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
Hard-Stop Circuit Breakers in LangGraph State
Hard-Stop Circuit Breakers in LangGraph State works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
from typing import Dict, TypedDict, Literal
# Define graph state with a loop counter to act as a circuit breaker
class GraphState(TypedDict):
question: str
generation: str
documents: list[str]
loop_step: int # Tracks iterations to prevent runaway execution costs
MAX_RETRIES = 3
def grade_generation_route(state: GraphState) -> Literal["web_search", "finalize", "max_retries_exceeded"]:
"""
Evaluates the quality of generation and routes the agent accordingly.
Why: Acts as a deterministic circuit breaker when LLMs get stuck in correction loops.
"""
current_step = state.get("loop_step", 0)
# 1. Evaluate circuit breaker before making any expensive LLM calls
if current_step >= MAX_RETRIES:
return "max_retries_exceeded"
# 2. Simulated grader check (In production, this queries your grading model)
is_accurate = False # Assume we failed the grade in this step
if is_accurate:
return "finalize"
else:
# Route back for retrieval/generation rewrite
return "web_search"
Structured Prompts for Deterministic Grading
Structured Prompts for Deterministic Grading works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion. Budget tokens per turn and per session. Agentic tools expand context aggressively; hard caps keep demos from becoming surprise invoices.
from langchain_core.pydantic_v1 import BaseModel, Field
class GradeDocument(BaseModel):
"""Binary score for document relevance assessment."""
binary_score: str = Field(
description="Relevance score for the document. Must be 'yes' or 'no'."
)
explanation: str = Field(
description="A brief, single-sentence justification for the given score."
)
Tracing Graph Execution with Observability Engines
Tracing Graph Execution with Observability Engines works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Record timings and token or query cost next to functional results. Cost visibility early prevents surprise bills when the path moves from demo to shared environments. Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move. Tracing Graph Execution with Observability Engines works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope. Document the happy path and the recovery path together. Retries, human gates, and dead-letter handling are part of the product, not later polish.
[User Query] ──> [retrieve] ──> [grade_documents] (Score: No)
│
▼
[finalize] <── [generate] <── [web_search] (Updated State: Loop Count = 1)
What a State-Driven Workflow Reveals
For What a State-Driven Workflow Reveals, define the inputs, the owner of the step, and the exit criteria before changing code. Operators should be able to re-run the step from a known checkpoint without guessing hidden state. Prefer small, testable units over sprawling scripts. When a step fails, the failure should point at a single responsibility rather than a tangled pipeline. Cite the passages that actually grounded the answer. Without citations, operators cannot tell hallucination from an indexing gap.
Operational checklist
Operational checklist works best when treated as a measurable surface. Capture one golden transcript, one failure case, and the rollback note before expanding scope.
Keep configuration outside application code. Environment files, secret stores, and feature flags belong in one place operators can audit without reading the whole graph.
Separate chunking policy from retrieval policy. Changing one should not force a rewrite of the other when quality metrics move.
Put human approval on edges that spend money or change production data. Compile-time wiring does not equal business completeness.
Write a short runbook: how to rotate keys, how to drain the queue, how to roll back the last ingest.
Treat this stage as a contract between inputs and validated outputs. Name the artifacts, define success checks, and refuse silent partial completion.
Before promoting the stack, freeze versions, capture a golden transcript for the critical path, and confirm rollback steps. Shared environments need rate limits, tenancy checks, and a clear owner for secret rotation. Prefer boring reliability over clever one-off demos.
Batch note for eeb69aedbcbc: keep provider keys out of the repo, set a per-session token ceiling, and store transcripts next to the eval fixtures so later model swaps stay comparable.