Architecting Agentic RAG Systems with LangGraph and FastAPI

Retrieval-Augmented Generation (RAG) has evolved beyond simple vector search and single-prompt generation. Production-grade systems require decision-making loops, fallback strategies, and self-correcting mechanisms.

In this deep dive, we'll build an Agentic RAG Pipeline using LangGraph for orchestration and control flow, combined with a FastAPI backend for a production-ready API.

Why Agentic RAG?

Standard RAG architectures assume the search query is perfect and the retrieved documents are always relevant. When these assumptions fail, standard systems generate hallucinations or unhelpful answers.

Agentic RAG introduces loop controls:

1. Query Router: Decides whether Vector Search, Web Search, or direct LLM reasoning is needed.

1. Document Grader: Checks retrieved document chunks for relevance.

1. Hallucination Grader: Evaluates the generated response against document facts.

1. Answer Grader: Assesses whether the response answers the user query.

Orchestrating the Graph with LangGraph

LangGraph models stateful, multi-actor applications as a graph (nodes and edges). This makes loops and conditional routing straightforward to define and test.

Let's look at the state definition and routing graph:

from typing import List, TypedDict
from langgraph.graph import StateGraph, END

class AgentState(TypedDict):
    question: str
    generation: str
    documents: List[str]
    web_search: bool

# Define nodes
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve_docs)
workflow.add_node("grade_documents", grade_docs)
workflow.add_node("generate", generate_answer)
workflow.add_node("web_search", web_search_fallback)

# Build edges
workflow.set_entry_point("retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "generate": "generate",
        "web_search": "web_search"
    }
)
workflow.add_edge("web_search", "generate")
workflow.add_conditional_edges(
    "generate",
    grade_generation,
    {
        "hallucination": "generate", # Retry generation
        "useful": END,
        "not_useful": "web_search"
      }
)
app = workflow.compile()

Step-by-Step Node Execution

• `retrieve`: Performs vector similarity search on our database (using Cosine Similarity on pgvector).

• `grade_documents`: Loops through all chunks and prompts an LLM to output yes or no depending on relevance. If any chunks are irrelevant, we trigger a web_search fallback.

• `generate`: Generates response based on valid document chunks.

• `grade_generation`: Double checks if generated output is factually grounded in the documents. If a hallucination is detected, the graph retraces its steps.

Setting up FastAPI for Streaming Output

To deliver a premium UI experience, we want to stream our LLM response token-by-token. FastAPI supports Server-Sent Events (SSE) via the StreamingResponse wrapper.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI()

class QueryRequest(BaseModel):
    question: str

@app.post("/api/chat")
async def chat_endpoint(req: QueryRequest):
    async def response_generator():
        # Iterate through graph steps asynchronously
        async for event in graph_app.astream({"question": req.question}):
            for node_name, state in event.items():
                if "generation" in state:
                    yield f"data: {state['generation']}\n\n"
                    
    return StreamingResponse(response_generator(), media_type="text/event-stream")

Production Takeaways

1. Evaluation is Key: Always run regression tests using tools like Ragas to evaluate retrieval accuracy and generation quality before deploying graph modifications.

1. Context Compression: Instead of dumping raw documents, use LLM compressors to extract key facts first, optimizing context window usage and reducing latency.

1. Structured Tool Outputs: Leverage Pydantic schemas in LangChain tool calling to ensure reliable inputs and outputs.