Agentic RAG in LangGraph: Graders, Conditional Routing and Web Search Fallback

A semantic RAG pipeline built from LangChain's core pieces, a FastAPI backend and a Streamlit chatbot front end, handles PDF uploads, turns the documents into embeddings and answers questions through a chat interface. It also has a basic limit: LangChain's primitives on their own give you linear chains, a straight line from prompt through LLM to output parser. Anything needing conditional branches, loops, dynamic routing or decisions made by the LLM falls outside that shape, and simple if/else statements only buy a shallow, predefined flow.

LangGraph addresses this. The result here is an agentic workflow with these properties:

• conditional branches and loops built right into the pipeline
• an LLM that chooses between web search and RAG retrieval, with no hard-coded rule
• transitions managed through LangGraph nodes and edges
• LLM-based graders that judge retrieved context and generated answers
• routing that reacts to those verdicts
• state carried across transitions by a StateGraph
• the finished graph plugged into a FastAPI backend to power a production-style chatbot

In the end you have an agentic RAG system that answers from either the vector store or the web.

LangGraph terms

Four ideas cover the vocabulary. The StateGraph orchestrates execution. State is a typed dictionary storing and updating the pipeline's data. Nodes are Python functions that change state and act as the graph's units of computation. Edges are the transitions between nodes, and conditional edges enable both branching and loops.

Configuration

Put the API keys in a .env file at the project root:

OPENAI_API_KEY="<Your OpenAI API Key>"
VECTOR_DB=pinecone
Hugging_Face_TOKEN=<your huggingface token>
PINECONE_API_KEY=<your Pinecone API Key>
PINECONE_ENVIRONMENT=gcp-starter
LANGSMITH_TRACING=true
LANGSMITH_ENDPOINT="https://api.smith.langchain.com"
LANGSMITH_API_KEY="<Your LangSmith API Key"
LANGSMITH_PROJECT="agent-ai"
TAVILY_API_KEY=<Your TAVILY API Key>

Then read them in a Pydantic BaseSettings class:

class BaseConfig(BaseSettings):
    OPENAI_API_KEY: Optional[str]
    PINECONE_API_KEY: Optional[str]
    TAVILY_API_KEY: Optional[str]

    """Loads the dotenv file. Including this is necessary to get
    pydantic to load a .env file."""
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

Project layout

The folder structure looks like this:

project/
├── rag_process/
│   ├── __init__.py
│   ├── extractor.py
|   └── service.py
│
├── graphs
│   │── chains.py       # llm chains
│   ├── nodes.py        # graph nodes. Defined as Python functions
│   ├── graph.py        # graph workflow
│   ├── consts.py       # define node names as constants
│   └── state.py        # typed dictionary to store graph states
|── schemas.py          # pydantic data models
|── upload.py           # python functions to upload files asynchronously
├── main.py             # fastAPI app
├── config.py           # pydantic BaseConfig
│── dependencies.py     # dependency inject to fastAPI endpoint function
├── test_chains.py      # pytests
├── .env                # environment variable definition
├── client.py           # streamlit client
├── .gitignore
└── requirements.txt     # package requirements

The retrieval chains, the FastAPI service and the Streamlit client carry over from the earlier pipeline. New here are the graph, its state class and the graders that check retrieved documents for relevance, detect hallucinations and judge whether an answer is relevant to the question.

The LLM chains

The chains live in graphs/chains.py and use LangChain Expression Language (LCEL). Seven are defined:

• retrieve_documents pulls the top-K similar documents from Pinecone
• generate_answer writes an answer from the question plus context
• retrival_grader decides whether retrieved documents are relevant
• web_search runs an internet search and returns context
• answer_grader checks that the answer addresses the question
• hallucination_grader checks that the answer is grounded in the context
• select_routes sends questions about agents, prompt engineering and adversarial attacks to RAG and everything else to web search

Unit tests for chains

Testing matters more than usual in agentic systems because LLM output is nondeterministic and can drift as models get updated, vendors tune them or embeddings change. Tests in test_chains.py use pytest with pytest.mark.asyncio for the async chains. A positive test for RAG retrieval looks like this:

@pytest.mark.asyncio
async def test_rag_retrieval_yes() -> None:
    prompt = "what is prompt engineer ?"
    searched_documents = await retrieve_documents(prompt)
    searched_document = searched_documents[0]
    # logger.info(f"searched_content: {searched_document.page_content}")
    retrieval_grade = await retrieval_grader(searched_document, prompt)
    assert retrieval_grade

It asks about prompt engineering and passes the first retrieved document to retrieval_grader. Because that topic is covered by documents already in Pinecone, the grader should return True. The test guards against breakage when embeddings or model behavior shift, catches regressions from model updates early and verifies the routing logic between RAG and web search. A matching negative test confirms the grader returns False and the flow heads to web search.

State and nodes

State is a typed dictionary kept in graphs/state.py:

class GraphState(TypedDict):

    question: str
    generation: str
    web_search: bool
    documents: list[str]

Nodes read and update this state. Defined in graphs/nodes.py, they're ordinary Python functions that read state, call LLM chains and return updated fields. A node used in a conditional edge returns the name of the next node to run. There are seven.

• generate calls generate_answer and returns the answer.
• retrieve calls retrieve_documents and returns the documents.
• search_web calls web_search and appends results to the documents.
• grade_documents keeps the relevant documents and sets web_search=True when needed.
• grade_generation runs the hallucination and answer graders and returns "useful", "not useful" or "not supported".
• decide_to_generate picks WEBSEARCH or GENERATE depending on web_search.
• route_question sends agent-related questions to RAG and others to web search.

Building the StateGraph

The graph itself is assembled in graphs/graph.py, and the finished workflow looks like a branching flow between retrieval, grading, generation and search. Building it takes several steps.

First, give every node a string identifier. Keeping those names as constants in graphs/consts.py makes the graph easier to read and prevents typos:

RETRIEVE = "retrieve"
GRADE_DOCUMENTS = "grade_documents"
GENERATE = "generate"
WEBSEARCH = "websearch"

Next, create the StateGraph and register each node with its function:

workflow = StateGraph(GraphState)

workflow.add_node(RETRIEVE, retrieve)
workflow.add_node(GRADE_DOCUMENTS, grade_documents)
workflow.add_node(GENERATE, generate)
workflow.add_node(WEBSEARCH, search_web)

Each add_node call binds a name to a function, and that separation between names and behavior makes the workflow easier to maintain, visualize and extend.

Third, set a conditional entry point. Rather than hard-coding a start node, LangGraph lets a function inspect the question and pick the first route. Here route_question decides whether to begin with web search or RAG retrieval:

workflow.set_conditional_entry_point(
    route_question,
    {
        WEBSEARCH: WEBSEARCH,
        RETRIEVE: RETRIEVE,
    }
)

That function runs first and returns either "WEBSEARCH" or "RETRIEVE" depending on the topic, and LangGraph uses its return value to choose the starting node. Agent-related questions go to RAG, and general knowledge starts with web search.

Fourth, connect nodes with edges. Static edges always follow one path, while conditional edges choose the next node based on an LLM decision:

workflow.add_edge(RETRIEVE, GRADE_DOCUMENTS)
workflow.add_conditional_edges(
    GRADE_DOCUMENTS,
    decide_to_generate,
    {
        WEBSEARCH: WEBSEARCH,
        GENERATE: GENERATE,
    },
)

workflow.add_conditional_edges(
    GENERATE,
    grade_generation,
    {
        "not supported": GENERATE,
        "useful": END,
        "not useful": WEBSEARCH,
    }
)

workflow.add_edge(WEBSEARCH, GENERATE)
workflow.add_edge(GENERATE, END)

Together these edges form a dynamic workflow in which the LLM picks the initial route, document relevance decides between RAG and web search, and answer quality decides whether to refine, fall back or finish. That structure turns a plain RAG pipeline into a sturdier agentic system.

Finally, compile the graph and optionally draw it:

app = workflow.compile()

app.get_graph().draw_mermaid_png(output_file_path="graphs.png")

Serving it through FastAPI

In dependencies.py, a function imports the compiled graph and invokes it asynchronously:

from graphs.graph import app


async def get_generation(body: RAGRequest=Body(...)) -> dict:
    try:
        generation = await app.ainvoke(body.dict())

        return {
            "answer": generation.get("generation", ""),
            "web_search": generation.get("web_search", False)
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

FastAPI injects it as a dependency, and it returns the generated answer plus a flag for whether it came from RAG or web search.

Trying it out

The Streamlit front end from the earlier pipeline can be reused as is. Asking how to become a millionaire correctly routes to web search, while asking what prompt engineering is routes to the semantic RAG path. That shows dynamic routing along with clear provenance for each answer.

To recap, the LangGraph StateGraph now manages state, the workflow has dynamic transitions with loops and branches through conditional edges, LLM graders cover document relevance, hallucination detection and answer quality, and the API tells the user whether an answer came from RAG or the web. The next step in this line of work is tool use and the ReAct framework, for more agentic behavior still.