Building an “Adaptive” RAG Pipeline: LangGraph & Reasoning Models (2026 Guide)

from typing import Literal
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage

# The "Memory" of our Agent
class GraphState(TypedDict):
    question: str
    generation: str
    documents: list[str]
    search_needed: bool # 2026 Logic: Did we fail to find info internally?

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# Define the routing logic
class RouteQuery(BaseModel):
    """Route a user query to the most relevant datasource."""
    datasource: Literal["vectorstore", "web_search", "general_chat"] = Field(
        ...,
        description="Given a user question, choose to route it to web search, vectorstore, or general chat."
    )

llm = ChatOpenAI(model="gpt-4o", temperature=0)
structured_router = llm.with_structured_output(RouteQuery)

def route_question(state):
    """
    Route question to web search or vectorstore.
    """
    print("---ROUTING QUESTION---")
    question = state["question"]
    source = structured_router.invoke([("system", "You are an expert router."), ("human", question)])
    
    if source.datasource == "web_search":
        print("---ROUTE: WEB SEARCH---")
        return "web_search"
    elif source.datasource == "vectorstore":
        print("---ROUTE: VECTOR STORE---")
        return "retrieve"
    else:
        print("---ROUTE: GENERAL CHAT---")
        return "generate_direct"

from langchain_community.tools.tavily_search import TavilySearchResults

web_search_tool = TavilySearchResults(k=3)

def web_search(state):
    """
    Web search based on the re-phrased question.
    """
    print("---WEB SEARCH---")
    question = state["question"]
    docs = web_search_tool.invoke({"query": question})
    # Join results into a context string
    web_results = "\n".join([d["content"] for d in docs])
    return {"documents": [web_results], "question": question}

def retrieve(state):
    """
    Retrieve documents from Vector DB.
    """
    print("---RETRIEVE---")
    question = state["question"]
    # Assuming 'retriever' is already defined from your Pinecone/Chroma setup
    documents = retriever.invoke(question)
    return {"documents": documents, "question": question}

def grade_documents(state):
    """
    Determines whether the retrieved documents are relevant to the question.
    """
    print("---CHECK RELEVANCE---")
    question = state["question"]
    documents = state["documents"]
    
    # (Insert Grader LLM Logic Here - similar to previous tutorials)
    
    # Logic: If 0 documents are relevant, set flag to search web
    if not relevant_docs:
        print("---DECISION: ALL DOCS BAD, FALLBACK TO WEB---")
        return "web_search" # Cyclic Loop!
    
    return "generate"

from langgraph.graph import END, StateGraph

workflow = StateGraph(GraphState)

# Define Nodes
workflow.add_node("web_search", web_search)
workflow.add_node("retrieve", retrieve)
workflow.add_node("generate", generate) # Defined elsewhere

# Define Entry Point (The Router)
workflow.set_conditional_entry_point(
    route_question,
    {
        "web_search": "web_search",
        "retrieve": "retrieve",
        "generate_direct": "generate"
    },
)

# Define Edges
workflow.add_edge("web_search", "generate")
workflow.add_edge("retrieve", "generate")
workflow.add_edge("generate", END)

# Compile
app = workflow.compile()