Build an Agentic RAG System

Build a self-correcting RAG pipeline with LangGraph, FAISS, and FastAPI.

Introduction

30 Second Summary

When you ask a question and get a useless answer, you naturally rephrase and try again. AI systems can do the same thing if you give them the right architecture.

In this project, you will build an agentic RAG system that retrieves documents, evaluates their quality, and self-corrects when it gets bad results. You will deploy the whole system as an API endpoint you can demo in interviews.

What You'll Build

An agentic document intelligence system powered by LangGraph that orchestrates a multi-step loop of retrieval, grading, query rewriting, and answer generation over your own documents.

By the end of this project, you'll have:

  • A document ingestion pipeline that chunks text and stores embeddings in a FAISS vector store.
  • A LangGraph agentic workflow with retrieve, grade, rewrite, and generate nodes that form an intelligent self-correcting loop.
  • A FastAPI endpoint that accepts questions and returns cited answers from your documents.
  • Secret Mission: Add an adaptive query router that classifies incoming questions and applies different retrieval strategies for simple lookups vs. complex reasoning.

Are there any prerequisites?

You need basic Python knowledge (functions, classes, dictionaries). No prior LangChain or LangGraph experience is required.

Before We Start

Before We Start...

Set Up Your Environment

Before building your agentic RAG system, you need a clean Python environment with all the right libraries installed. A dedicated project folder and virtual environment keep your dependencies isolated from the rest of your system.

In this step, you will create your project directory, install LangGraph, FAISS, and FastAPI, and configure your OpenRouter API connection.

In this step, get ready to:
  • Create a project directory and Python virtual environment.
  • Install all project dependencies.
  • Set up the OpenRouter API key and configuration file.
Create a project directory and virtual environment

A virtual environment gives your project its own isolated set of Python packages. This prevents version conflicts between different projects on your machine.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or Command Prompt (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir agentic-rag
cd agentic-rag
  • Create and activate your Python virtual environment by running the commands for your operating system:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

You should see (venv) appear at the start of your terminal prompt. This confirms your virtual environment is active.

Don't see (venv) in your prompt?

Make sure you ran the activation command for your operating system. On macOS, the command starts with source. On Windows, use the backslash path venv\Scripts\activate.

If you see a PowerShell execution policy error on Windows, run Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser and try activating again.

help me activate my Python virtual environment

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type VS Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop and select the agentic-rag folder, then click Open.

Why use VS Code's integrated terminal?

From this point on, use the terminal inside VS Code instead of the system terminal you just used. Open it with Ctrl+` (backtick key, same shortcut on both macOS and Windows).

The integrated terminal automatically opens in your project folder and keeps everything in one window. You may need to reactivate your virtual environment in this new terminal using the activation command for your OS.

Install all dependencies

Your agentic RAG system relies on several libraries working together. This single install command grabs everything you need.

  • Install all project dependencies by running this command in your VS Code terminal:
pip install langgraph langchain-openrouter faiss-cpu fastapi uvicorn langchain-text-splitters python-dotenv openai

What are these packages?

  • langgraph orchestrates your AI agent as a state machine with nodes and edges.
  • faiss-cpu is a vector similarity search library for fast document retrieval.
  • fastapi and uvicorn power your API endpoint.
  • langchain-text-splitters breaks documents into smaller chunks for embedding.
  • python-dotenv loads environment variables from your .env file.
  • openai is the Python SDK used to call OpenRouter's API (which is OpenAI-compatible).
  • Verify that the core library installed correctly by running:
python -c "from langgraph.graph import StateGraph, START, END; print('All imports OK')"

You should see All imports OK printed in your terminal with no errors.

Seeing an import error?

Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt. If not, reactivate it using the command for your operating system.

If the pip install failed partway through, run the install command again. Pip will skip packages already installed and retry the rest.

help me fix my LangGraph import error

Set up the OpenRouter API key

OpenRouter gives you access to free LLM models through an OpenAI-compatible API. Your project needs an API key to authenticate requests.

  • In the VS Code Explorer sidebar, click the new file icon at the top of your agentic-rag folder.
  • Name the file .env and press Enter.
  • Add your OpenRouter API key to .env by pasting the following, replacing your-api-key-here with your actual key:
OPENROUTER_API_KEY=your-api-key-here

Where do I find my OpenRouter API key?

Go to openrouter.ai/keys and sign in to your account. Click Create Key to generate a new key. Copy it immediately as it will not be shown again.

  • Save .env.

Now create the configuration file that loads this key and defines your model settings.

  • Create a new file called config.py in the same agentic-rag folder.
  • Add the following code to config.py:
import os
from dotenv import load_dotenv

# Load environment variables from the .env file
load_dotenv()

# OpenRouter API credentials and model selection
OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
MODEL_NAME = "openai/gpt-oss-120b:free"

What does this code do?

  • load_dotenv() reads your .env file and loads its values as environment variables.
  • OPENROUTER_API_KEY pulls your key from the environment so it never appears in your code.
  • OPENROUTER_BASE_URL points the OpenAI SDK at OpenRouter's servers instead of OpenAI's.
  • MODEL_NAME selects the free gpt-oss-120b model, which supports tool use and structured output at zero cost.
  • Save config.py.
  • Verify your configuration loads correctly by running this in your terminal:
python -c "from config import OPENROUTER_API_KEY, MODEL_NAME; print(f'Model: {MODEL_NAME}'); print('Key loaded!' if OPENROUTER_API_KEY else 'WARNING: No key found')"

You should see Model: openai/gpt-oss-120b:free and Key loaded! printed.

Seeing 'WARNING: No key found'?

Double-check that your .env file is saved in the same folder as config.py (the agentic-rag root).

Make sure there are no spaces around the = sign in your .env file. The format must be OPENROUTER_API_KEY=sk-or-... with no quotes and no spaces.

help me load my OpenRouter API key in config.py

✔️ Awesome, I've got everything!

Great. Make sure both .env and config.py are saved.

ⓧ I'd like to double check the full code

Here is the complete .env file:

OPENROUTER_API_KEY=your-api-key-here

Here is the complete config.py file:

import os
from dotenv import load_dotenv

load_dotenv()

OPENROUTER_API_KEY = os.getenv("OPENROUTER_API_KEY")
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
MODEL_NAME = "openai/gpt-oss-120b:free"

Your environment is ready with all dependencies installed and your API connection configured. Next up, you will build the document ingestion pipeline that turns raw text into searchable vector embeddings.

Build the Document Ingestion Pipeline

Your environment is ready to go. Now it's time to give your agent something to reason about.

A Retrieval-Augmented Generation system is only as good as its knowledge base. In this step, you will create sample documents, split them into searchable chunks, convert those chunks into embeddings, and store everything in a FAISS vector index. By the end, you will be able to query your index and get back the most relevant pieces of text.

In this step, get ready to:
  • Create sample enterprise documents as your knowledge base.
  • Build a chunking and embedding pipeline that converts text into vector representations.
  • Store embeddings in a FAISS index and verify retrieval with a test query.
Create sample documents

Your RAG system needs a knowledge base to search over. You will create two enterprise-style text files that simulate a real company's internal documents.

  • Create a documents folder inside your agentic-rag project by running this command:
mkdir documents
  • Create a new file called company_policy.txt inside the documents/ folder.
  • Paste this content into company_policy.txt:
Acme Corp Employee Handbook - Key Policies

Paid Time Off (PTO) Policy

All full-time employees receive 20 days of paid time off per calendar year. PTO accrues at a rate of 1.67 days per month. New employees may use accrued PTO after completing their 90-day probationary period.

Unused PTO may be carried over to the following year, up to a maximum of 5 days. Any PTO beyond the 5-day carryover limit will be forfeited on January 1st. Employees must submit PTO requests at least 2 weeks in advance for absences of 3 or more consecutive days.

Remote Work Policy

Acme Corp supports a hybrid work model. Employees may work remotely up to 3 days per week with manager approval. Fully remote arrangements require VP-level approval and are evaluated on a case-by-case basis.

Remote employees must maintain a dedicated workspace with reliable internet (minimum 50 Mbps download speed). All remote work must be performed within the country of employment for tax and legal compliance purposes.

Employees working remotely must be available during core hours (10 AM to 3 PM local time) and respond to messages within 30 minutes during these hours. Remote workers must use the company VPN for all work-related network access.

Information Security Requirements

All employees must complete annual security awareness training within the first quarter of each year. Failure to complete training by March 31st will result in temporary suspension of system access.

Passwords must be at least 16 characters and include uppercase, lowercase, numbers, and special characters. Multi-factor authentication (MFA) is mandatory for all company systems. Employees must never share credentials or use personal devices for accessing classified data.

Data classification follows three tiers: Public, Internal, and Confidential. Confidential data may only be stored on company-approved encrypted storage systems. Any suspected data breach must be reported to the security team within 1 hour of discovery.
  • Create a second file called technical_faq.txt inside the documents/ folder.
  • Paste this content into technical_faq.txt:
Technical Infrastructure FAQ

Q: What is the difference between on-premise and cloud deployment?

On-premise deployment means running all servers and infrastructure within our own physical data centers. We maintain full control over hardware, networking, and security. Cloud deployment uses third-party providers (AWS, Azure, GCP) to host our applications and data on their infrastructure.

On-premise benefits include complete data sovereignty, no recurring cloud costs for stable workloads, and full network control. Cloud benefits include elastic scaling, reduced maintenance overhead, and faster provisioning of new environments.

Q: What are the security requirements for cloud deployments?

Cloud deployments must follow the shared responsibility model. The cloud provider secures the infrastructure layer, while we are responsible for securing our applications, data, and access controls.

All cloud resources must be deployed within approved regions (US-East-1 and EU-West-1 only). Data at rest must be encrypted using AES-256, and data in transit must use TLS 1.3. Network security groups must follow the principle of least privilege with no ingress rules allowing 0.0.0.0/0 except for public-facing load balancers.

Cloud deployments require monthly vulnerability scans and quarterly penetration testing. All findings must be remediated within 30 days for critical severity and 90 days for high severity.

Q: How does on-premise security differ from cloud security?

On-premise security focuses on physical access controls, network perimeter defense, and hardware lifecycle management. We maintain our own firewalls, intrusion detection systems, and physical security (badge access, cameras, mantraps).

Cloud security focuses on identity and access management (IAM), encryption, and API security. We rely on the provider's physical security but must implement strong logical controls. Cloud environments require additional monitoring for misconfigurations using tools like AWS Config or Azure Policy.

The key difference is that on-premise security is perimeter-based (protect the castle walls), while cloud security is identity-based (verify every request regardless of origin). This zero-trust approach is mandatory for all new cloud deployments.

Q: What is our disaster recovery strategy?

On-premise systems maintain hot standby servers in our secondary data center with 4-hour RPO and 8-hour RTO. Cloud systems use multi-AZ deployments with automated failover, achieving 15-minute RPO and 1-hour RTO. All critical systems must be tested for failover quarterly.
  • Save both files.

Why these documents?

Enterprise RAG systems typically process internal policies, technical documentation, and compliance materials. These sample documents give your agent realistic content to search over and reason about.

The content is deliberately rich enough that a single retrieval pass might not always surface the best answer. This is what makes the agentic loop (retrieve, grade, rewrite, retry) valuable.

  • Confirm the folder and files exist by running:
ls documents

You should see company_policy.txt and technical_faq.txt listed in the output.

Build the chunking and embedding pipeline

Your documents are too long to pass directly to an LLM. You need to split them into smaller, overlapping chunks so that only the most relevant pieces get retrieved for each question.

  • Create a new file called ingest.py in the root of your agentic-rag folder.
  • Add the imports and the load_documents function by pasting this code:
import os
import json
import numpy as np
import faiss
from openai import OpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from config import OPENROUTER_API_KEY, OPENROUTER_BASE_URL, MODEL_NAME


def load_documents(directory: str) -> list[dict]:
    documents = []
    for filename in os.listdir(directory):
        if filename.endswith(".txt"):
            filepath = os.path.join(directory, filename)
            with open(filepath, "r", encoding="utf-8") as f:
                content = f.read()
            documents.append({"content": content, "source": filename})
    return documents

What does this code do?

The imports pull in the tools you need: faiss for vector search, numpy for numerical arrays, the OpenAI client (pointed at OpenRouter), and the text splitter from LangChain.

The load_documents function scans a directory for .txt files and returns each one as a dictionary with its content and source filename.

  • Add the chunk_documents function below load_documents:

def chunk_documents(documents: list[dict]) -> list[dict]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,
        chunk_overlap=64,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks = []
    for doc in documents:
        splits = splitter.split_text(doc["content"])
        for i, split in enumerate(splits):
            chunks.append({
                "content": split,
                "source": doc["source"],
                "chunk_id": f"{doc['source']}_chunk_{i}",
            })
    return chunks

What does this code do?

RecursiveCharacterTextSplitter breaks each document into chunks of at most 512 characters, with 64 characters of overlap between consecutive chunks. The overlap ensures that sentences at chunk boundaries are not cut off mid-thought.

Each chunk gets tagged with its source filename and a unique chunk_id so you can trace answers back to specific documents later.

  • Add the get_embeddings function below chunk_documents:

def get_embeddings(texts: list[str]) -> np.ndarray:
    client = OpenAI(
        base_url=OPENROUTER_BASE_URL,
        api_key=OPENROUTER_API_KEY,
    )
    embeddings = []
    batch_size = 20
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = client.embeddings.create(
            model="openai/text-embedding-3-small",
            input=batch,
        )
        for item in response.data:
            embeddings.append(item.embedding)
    return np.array(embeddings, dtype="float32")

What does this code do?

This function converts text into numerical vectors (embeddings) using OpenRouter's API with the openai/text-embedding-3-small model. It processes texts in batches of 20 to stay within rate limits.

The result is a NumPy array of float32 vectors. Each vector captures the semantic meaning of a chunk, so similar content ends up close together in vector space.

  • Save ingest.py.

Seeing import errors?

If your editor shows red underlines on faiss or langchain_text_splitters, make sure your terminal is using the virtual environment. Check the bottom-left of VS Code for the active Python interpreter and select the one inside your venv folder.

help me fix import errors in my ingest.py file

Create the FAISS vector store

With chunking and embedding functions ready, you now need to build the actual search index. FAISS uses inner product similarity on normalized vectors, which is equivalent to cosine similarity.

  • Add the build_faiss_index function below get_embeddings in ingest.py:

def build_faiss_index(chunks: list[dict]) -> None:
    texts = [chunk["content"] for chunk in chunks]
    embeddings = get_embeddings(texts)

    dimension = embeddings.shape[1]
    index = faiss.IndexFlatIP(dimension)
    faiss.normalize_L2(embeddings)
    index.add(embeddings)

    faiss.write_index(index, "faiss_index.bin")
    with open("chunks.json", "w", encoding="utf-8") as f:
        json.dump(chunks, f, ensure_ascii=False, indent=2)

    print(f"Indexed {len(chunks)} chunks with dimension {dimension}")

What does this code do?

This function ties everything together. It embeds all chunks, normalizes the vectors with faiss.normalize_L2 (making inner product equivalent to cosine similarity), and adds them to a FAISS IndexFlatIP index.

It saves two files to disk: faiss_index.bin (the vector index) and chunks.json (the chunk metadata). This means you only need to run ingestion once. Future searches load these files directly.

  • Add the search_index function and the script entry point below build_faiss_index:

def search_index(query: str, top_k: int = 3) -> list[dict]:
    index = faiss.read_index("faiss_index.bin")
    with open("chunks.json", "r", encoding="utf-8") as f:
        chunks = json.load(f)

    query_embedding = get_embeddings([query])
    faiss.normalize_L2(query_embedding)

    scores, indices = index.search(query_embedding, top_k)
    results = []
    for score, idx in zip(scores[0], indices[0]):
        if idx < len(chunks):
            result = chunks[idx].copy()
            result["score"] = float(score)
            results.append(result)
    return results


if __name__ == "__main__":
    docs = load_documents("documents")
    chunks = chunk_documents(docs)
    build_faiss_index(chunks)
    print("Ingestion complete!")

    results = search_index("test query")
    for r in results:
        print(f"[{r['score']:.3f}] {r['source']}: {r['content'][:80]}...")

What does this code do?

search_index loads the saved index from disk, embeds your query, normalizes it, and returns the top-k most similar chunks along with their similarity scores.

The if __name__ == "__main__" block runs the full pipeline when you execute the script directly. It loads documents, chunks them, builds the index, and then runs a test search to confirm everything works.

  • Save ingest.py.
  • Run the ingestion pipeline by executing this command in your terminal:
python ingest.py

You should see output showing how many chunks were indexed, followed by the top 3 results for the test query with their similarity scores and source filenames.

Seeing an authentication or API error?

Make sure your .env file contains a valid OpenRouter API key. The key should start with sk-or-. Double-check that config.py loads it correctly by running:

python -c "from config import OPENROUTER_API_KEY; print(OPENROUTER_API_KEY[:10])"

If this prints None or an empty string, your key is not loading from the .env file. Make sure the file is in the same directory as config.py.

help me debug my OpenRouter API key not loading in config.py

  • Confirm the index files were created by running:
ls faiss_index.bin chunks.json

You should see both files listed. These are your persisted vector store.

✔️ Awesome, I've got everything!

Great. Double check you have saved ingest.py and that both faiss_index.bin and chunks.json exist in your project folder.

ⓧ I'd like to double check the full code

import os
import json
import numpy as np
import faiss
from openai import OpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter
from config import OPENROUTER_API_KEY, OPENROUTER_BASE_URL, MODEL_NAME


def load_documents(directory: str) -> list[dict]:
    documents = []
    for filename in os.listdir(directory):
        if filename.endswith(".txt"):
            filepath = os.path.join(directory, filename)
            with open(filepath, "r", encoding="utf-8") as f:
                content = f.read()
            documents.append({"content": content, "source": filename})
    return documents


def chunk_documents(documents: list[dict]) -> list[dict]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,
        chunk_overlap=64,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks = []
    for doc in documents:
        splits = splitter.split_text(doc["content"])
        for i, split in enumerate(splits):
            chunks.append({
                "content": split,
                "source": doc["source"],
                "chunk_id": f"{doc['source']}_chunk_{i}",
            })
    return chunks


def get_embeddings(texts: list[str]) -> np.ndarray:
    client = OpenAI(
        base_url=OPENROUTER_BASE_URL,
        api_key=OPENROUTER_API_KEY,
    )
    embeddings = []
    batch_size = 20
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        response = client.embeddings.create(
            model="openai/text-embedding-3-small",
            input=batch,
        )
        for item in response.data:
            embeddings.append(item.embedding)
    return np.array(embeddings, dtype="float32")


def build_faiss_index(chunks: list[dict]) -> None:
    texts = [chunk["content"] for chunk in chunks]
    embeddings = get_embeddings(texts)

    dimension = embeddings.shape[1]
    index = faiss.IndexFlatIP(dimension)
    faiss.normalize_L2(embeddings)
    index.add(embeddings)

    faiss.write_index(index, "faiss_index.bin")
    with open("chunks.json", "w", encoding="utf-8") as f:
        json.dump(chunks, f, ensure_ascii=False, indent=2)

    print(f"Indexed {len(chunks)} chunks with dimension {dimension}")


def search_index(query: str, top_k: int = 3) -> list[dict]:
    index = faiss.read_index("faiss_index.bin")
    with open("chunks.json", "r", encoding="utf-8") as f:
        chunks = json.load(f)

    query_embedding = get_embeddings([query])
    faiss.normalize_L2(query_embedding)

    scores, indices = index.search(query_embedding, top_k)
    results = []
    for score, idx in zip(scores[0], indices[0]):
        if idx < len(chunks):
            result = chunks[idx].copy()
            result["score"] = float(score)
            results.append(result)
    return results


if __name__ == "__main__":
    docs = load_documents("documents")
    chunks = chunk_documents(docs)
    build_faiss_index(chunks)
    print("Ingestion complete!")

    results = search_index("test query")
    for r in results:
        print(f"[{r['score']:.3f}] {r['source']}: {r['content'][:80]}...")

Your knowledge base is indexed and searchable. Next up, you will build the agentic graph that retrieves these chunks, grades their quality, and generates cited answers.

Build the Agentic RAG Graph

Your documents are indexed and searchable. Now you need an intelligent system that does more than a single retrieve-and-answer pass.

In this step, you will build a LangGraph state machine that orchestrates an agentic loop: retrieve relevant chunks, grade their quality, rewrite the query if needed, and only generate an answer once the context is good enough.

In this step, get ready to:
  • Define the graph state schema and all node functions.
  • Wire the graph with conditional edges that route between nodes.
  • Compile and test the agentic loop with a real question.
Define the graph state and nodes

Every LangGraph workflow starts with a state definition. The state is a typed dictionary that flows through each node, carrying the question, retrieved documents, generated answer, and loop counter.

  • Create a new file called agent.py in your agentic-rag folder by adding the following code:
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from openai import OpenAI
from ingest import search_index
from config import OPENROUTER_API_KEY, OPENROUTER_BASE_URL, MODEL_NAME


class AgentState(TypedDict):
    question: str
    documents: list[dict]
    generation: str
    iterations: int
    max_iterations: int


def get_llm_client() -> OpenAI:
    return OpenAI(
        base_url=OPENROUTER_BASE_URL,
        api_key=OPENROUTER_API_KEY,
    )


def retrieve(state: AgentState) -> AgentState:
    question = state["question"]
    documents = search_index(question, top_k=4)
    return {**state, "documents": documents}

What does this code do?

AgentState is a TypedDict that defines the shape of data flowing through the graph. Every node receives this state and returns an updated copy.

get_llm_client() creates a reusable OpenAI client pointed at the OpenRouter API. The retrieve function searches your FAISS index for the top 4 most relevant chunks and stores them in the state.

You will not see any output yet. The next chunk adds the grading logic.

  • Add the document grading logic below the retrieve function:
def grade_documents(state: AgentState) -> AgentState:
    return state


def grade_condition(state: AgentState) -> Literal["generate", "rewrite_query"]:
    client = get_llm_client()
    question = state["question"]
    docs_text = "\n---\n".join([d["content"] for d in state["documents"]])

    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=[{
            "role": "user",
            "content": (
                f"Question: {question}\n\n"
                f"Retrieved documents:\n{docs_text}\n\n"
                "Are these documents relevant to answering the question? "
                "Reply with exactly 'yes' or 'no'."
            ),
        }],
        temperature=0,
        max_tokens=10,
    )
    answer = response.choices[0].message.content.strip().lower()

    if "yes" in answer or state["iterations"] >= state["max_iterations"]:
        return "generate"
    return "rewrite_query"

What does this code do?

grade_documents is a passthrough node. It exists so the graph has a named node to attach the conditional edge to.

grade_condition is the decision function. It asks the LLM whether the retrieved documents are relevant. If the answer is "yes" (or the iteration cap is reached), it routes to generate. Otherwise it routes to rewrite_query so the system can try again with a better question.

You will not see a change yet. The next chunk adds the rewrite and generate nodes.

  • Add the rewrite_query function below grade_condition:
def rewrite_query(state: AgentState) -> AgentState:
    client = get_llm_client()
    question = state["question"]

    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=[{
            "role": "user",
            "content": (
                f"The following question did not retrieve relevant documents: "
                f"'{question}'\n\n"
                "Rewrite it to be more specific and likely to match relevant content. "
                "Return only the rewritten question."
            ),
        }],
        temperature=0.3,
        max_tokens=150,
    )
    new_question = response.choices[0].message.content.strip()
    return {
        **state,
        "question": new_question,
        "iterations": state["iterations"] + 1,
    }

What does this code do?

rewrite_query asks the LLM to reformulate the question so it better matches your document content. It increments the iterations counter each time so the system knows when to stop retrying.

You will not see a change yet. The generate node comes next.

  • Add the generate function below rewrite_query:
def generate(state: AgentState) -> AgentState:
    client = get_llm_client()
    question = state["question"]
    docs_text = "\n---\n".join(
        [f"[Source: {d['source']}] {d['content']}" for d in state["documents"]]
    )

    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=[{
            "role": "system",
            "content": (
                "You are a helpful assistant that answers questions based on "
                "the provided context. Always cite which source document you "
                "used. If the context does not contain the answer, say so."
            ),
        }, {
            "role": "user",
            "content": f"Context:\n{docs_text}\n\nQuestion: {question}",
        }],
        temperature=0.2,
        max_tokens=500,
    )
    answer = response.choices[0].message.content.strip()
    return {**state, "generation": answer}

What does this code do?

generate is the final node. It labels each chunk with its source file, passes them as context to the LLM, and asks for a cited answer. The system prompt instructs the model to always name the document it pulled information from.

  • Save agent.py.

Seeing an import error?

Make sure your virtual environment is activated. The langgraph and openai packages must be installed in the active environment.

help me fix an import error in agent.py

Wire the graph with conditional edges

With all four node functions defined, you can now connect them into a StateGraph. The graph defines the execution order and the decision points that make this system agentic.

  • Add the build_agent function below the generate function in agent.py:
def build_agent():
    workflow = StateGraph(AgentState)

    workflow.add_node("retrieve", retrieve)
    workflow.add_node("grade_documents", grade_documents)
    workflow.add_node("rewrite_query", rewrite_query)
    workflow.add_node("generate", generate)

    workflow.add_edge(START, "retrieve")
    workflow.add_edge("retrieve", "grade_documents")
    workflow.add_conditional_edges(
        "grade_documents",
        grade_condition,
        {"generate": "generate", "rewrite_query": "rewrite_query"},
    )
    workflow.add_edge("rewrite_query", "retrieve")
    workflow.add_edge("generate", END)

    return workflow.compile()

What does this code do?

  • add_node registers each function as a named node in the graph.
  • add_edge creates a fixed connection between two nodes (e.g. START always goes to retrieve).
  • add_conditional_edges attaches grade_condition as a router. Depending on the return value, the graph flows to either generate or rewrite_query.
  • The edge from rewrite_query back to retrieve creates the self-correcting loop.

Why a graph instead of a chain?

A traditional RAG chain runs in a straight line: retrieve, then generate. If the retrieved documents are irrelevant, the answer is bad and there is no recovery.

A graph lets you add decision points. The conditional edge after grading creates a loop that retries with a rewritten question. This self-correcting behavior is what makes the system agentic.

  • Save agent.py.

You will not see output yet. The next chunk compiles the graph and runs a test.

Compile and test the graph

Compiling transforms your graph definition into a runnable object. You will also add a test block so you can run agent.py directly and see the agentic loop in action.

  • Add the following code at the bottom of agent.py, below the build_agent function:
agent = build_agent()


if __name__ == "__main__":
    result = agent.invoke({
        "question": "What is the company policy on remote work?",
        "documents": [],
        "generation": "",
        "iterations": 0,
        "max_iterations": 3,
    })
    print(f"\nAnswer: {result['generation']}")
    print(f"Iterations: {result['iterations']}")

What does this code do?

agent = build_agent() compiles the graph at module level so other files (like your upcoming API) can import and use it directly.

The if __name__ == "__main__" block runs only when you execute the file directly. It invokes the graph with a test question, starting with zero iterations and a cap of 3.

  • Save agent.py.
  • Run the agent by executing this command in your terminal:
python agent.py
  • You should see a printed answer with source citations, plus an iteration count showing how many retrieve-grade cycles ran.

What should I see?

If your question matches the document content well, you will see Iterations: 0 (the grader said "yes" on the first try). If the question is ambiguous or does not match your document phrasing, you may see Iterations: 1 or higher, meaning the system rewrote the query and re-retrieved before generating.

Getting an empty answer or timeout?

Free-tier models on OpenRouter can be slow under load. Wait up to 30 seconds for a response. If the answer is empty, check that your .env file has a valid OPENROUTER_API_KEY and that faiss_index.bin exists in your project folder.

help me debug my agentic RAG graph not returning answers

✔️ Awesome, I've got everything!

Great. Double check that you have saved agent.py and that your test run produced a cited answer.

ⓧ I'd like to double check the full code

Here is the complete agent.py file for comparison:

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from openai import OpenAI
from ingest import search_index
from config import OPENROUTER_API_KEY, OPENROUTER_BASE_URL, MODEL_NAME


class AgentState(TypedDict):
    question: str
    documents: list[dict]
    generation: str
    iterations: int
    max_iterations: int


def get_llm_client() -> OpenAI:
    return OpenAI(
        base_url=OPENROUTER_BASE_URL,
        api_key=OPENROUTER_API_KEY,
    )


def retrieve(state: AgentState) -> AgentState:
    question = state["question"]
    documents = search_index(question, top_k=4)
    return {**state, "documents": documents}


def grade_documents(state: AgentState) -> AgentState:
    return state


def grade_condition(state: AgentState) -> Literal["generate", "rewrite_query"]:
    client = get_llm_client()
    question = state["question"]
    docs_text = "\n---\n".join([d["content"] for d in state["documents"]])

    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=[{
            "role": "user",
            "content": (
                f"Question: {question}\n\n"
                f"Retrieved documents:\n{docs_text}\n\n"
                "Are these documents relevant to answering the question? "
                "Reply with exactly 'yes' or 'no'."
            ),
        }],
        temperature=0,
        max_tokens=10,
    )
    answer = response.choices[0].message.content.strip().lower()

    if "yes" in answer or state["iterations"] >= state["max_iterations"]:
        return "generate"
    return "rewrite_query"


def rewrite_query(state: AgentState) -> AgentState:
    client = get_llm_client()
    question = state["question"]

    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=[{
            "role": "user",
            "content": (
                f"The following question did not retrieve relevant documents: "
                f"'{question}'\n\n"
                "Rewrite it to be more specific and likely to match relevant content. "
                "Return only the rewritten question."
            ),
        }],
        temperature=0.3,
        max_tokens=150,
    )
    new_question = response.choices[0].message.content.strip()
    return {
        **state,
        "question": new_question,
        "iterations": state["iterations"] + 1,
    }


def generate(state: AgentState) -> AgentState:
    client = get_llm_client()
    question = state["question"]
    docs_text = "\n---\n".join(
        [f"[Source: {d['source']}] {d['content']}" for d in state["documents"]]
    )

    response = client.chat.completions.create(
        model=MODEL_NAME,
        messages=[{
            "role": "system",
            "content": (
                "You are a helpful assistant that answers questions based on "
                "the provided context. Always cite which source document you "
                "used. If the context does not contain the answer, say so."
            ),
        }, {
            "role": "user",
            "content": f"Context:\n{docs_text}\n\nQuestion: {question}",
        }],
        temperature=0.2,
        max_tokens=500,
    )
    answer = response.choices[0].message.content.strip()
    return {**state, "generation": answer}


def build_agent():
    workflow = StateGraph(AgentState)

    workflow.add_node("retrieve", retrieve)
    workflow.add_node("grade_documents", grade_documents)
    workflow.add_node("rewrite_query", rewrite_query)
    workflow.add_node("generate", generate)

    workflow.add_edge(START, "retrieve")
    workflow.add_edge("retrieve", "grade_documents")
    workflow.add_conditional_edges(
        "grade_documents",
        grade_condition,
        {"generate": "generate", "rewrite_query": "rewrite_query"},
    )
    workflow.add_edge("rewrite_query", "retrieve")
    workflow.add_edge("generate", END)

    return workflow.compile()


agent = build_agent()


if __name__ == "__main__":
    result = agent.invoke({
        "question": "What is the company policy on remote work?",
        "documents": [],
        "generation": "",
        "iterations": 0,
        "max_iterations": 3,
    })
    print(f"\nAnswer: {result['generation']}")
    print(f"Iterations: {result['iterations']}")

Your agentic RAG graph is working. Next up, you will wrap it in a FastAPI endpoint so anyone can send questions and get cited answers over HTTP.

Expose as a FastAPI Endpoint

Your agentic RAG pipeline can retrieve, grade, rewrite, and generate answers from the command line. That's great for testing, but useless for any other application that wants to consume it.

In this step, you'll wrap the compiled LangGraph agent in a FastAPI server. This gives you a production-ready HTTP endpoint that any frontend, mobile app, or external service can call to ask questions against your documents.

In this step, get ready to:
  • Create a FastAPI application with a POST endpoint that invokes your agent.
  • Add structured response formatting with source references and error handling.
  • Run and test the server using the built-in Swagger UI.
Create the FastAPI application

Your API needs a file to define the server, the request shape, and the endpoint that connects incoming questions to your agent.

  • Create a new file called main.py in your agentic-rag folder.
  • Add the imports, app instantiation, and request model by pasting this code:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent import agent

# Create the FastAPI app with a descriptive title
app = FastAPI(title="Agentic RAG API")


# Define the shape of incoming requests
class QuestionRequest(BaseModel):
    question: str

What does this code do?

FastAPI creates the web server. HTTPException lets you return proper error responses when something goes wrong.

BaseModel from Pydantic defines the expected JSON structure of incoming requests. FastAPI uses this to automatically validate that every request includes a question field.

The from agent import agent line imports your compiled LangGraph agent directly. When this module loads, it triggers build_agent() to run and compile the graph.

  • Save main.py.
Add response formatting and the endpoint

The endpoint needs to invoke your agent, extract the relevant source filenames, and return a structured JSON response that any client can parse.

  • In main.py, add the following code below the QuestionRequest class:
# Define the shape of outgoing responses
class AnswerResponse(BaseModel):
    answer: str
    sources: list[str]
    iterations: int


# POST endpoint that accepts a question and returns a cited answer
@app.post("/ask", response_model=AnswerResponse)
async def ask_question(request: QuestionRequest):
    try:
        result = agent.invoke({
            "question": request.question,
            "documents": [],
            "generation": "",
            "iterations": 0,
            "max_iterations": 3,
        })
        # Deduplicate source filenames from retrieved documents
        sources = list(set(d["source"] for d in result["documents"]))
        return AnswerResponse(
            answer=result["generation"],
            sources=sources,
            iterations=result["iterations"],
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# Simple health check endpoint
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

What does this code do?

  • AnswerResponse defines the JSON shape clients receive: the generated answer, which source documents were used, and how many retrieve-grade cycles the agent ran.
  • The /ask endpoint invokes the compiled agent with the same state shape you used in the command-line test. It passes max_iterations: 3 to cap the rewrite loop.
  • sources = list(set(...)) deduplicates filenames because multiple chunks often come from the same document.
  • The try/except block catches any failure in the agent pipeline and returns a 500 error with the exception message, instead of crashing silently.
  • The /health endpoint is a lightweight check that confirms the server is running without invoking the full agent.
  • Save main.py.

Seeing import errors?

If you see an error about agent not being found, make sure agent.py is in the same folder as main.py. Both files must live in your agentic-rag directory.

help me fix import errors in my FastAPI app

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py with both the request model and the endpoint code.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from agent import agent

app = FastAPI(title="Agentic RAG API")


class QuestionRequest(BaseModel):
    question: str


class AnswerResponse(BaseModel):
    answer: str
    sources: list[str]
    iterations: int


@app.post("/ask", response_model=AnswerResponse)
async def ask_question(request: QuestionRequest):
    try:
        result = agent.invoke({
            "question": request.question,
            "documents": [],
            "generation": "",
            "iterations": 0,
            "max_iterations": 3,
        })
        sources = list(set(d["source"] for d in result["documents"]))
        return AnswerResponse(
            answer=result["generation"],
            sources=sources,
            iterations=result["iterations"],
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/health")
async def health_check():
    return {"status": "healthy"}
Run and test the server

FastAPI apps run via uvicorn, an ASGI server that handles incoming HTTP requests. It also auto-generates interactive API docs you can use to test your endpoint without writing any client code.

  • Start the server by running this command in your terminal:
uvicorn main:app --reload

What does this command do?

  • uvicorn is the ASGI server that runs your FastAPI app.
  • main:app tells uvicorn to look in main.py for the object named app.
  • --reload makes the server restart automatically whenever you save a file. This is for development only.
  • You should see output indicating the server is running at http://127.0.0.1:8000.
  • Open your browser and go to http://127.0.0.1:8000/docs to see the Swagger UI.

What is Swagger UI?

Swagger UI is an interactive API documentation page that FastAPI generates automatically from your endpoint definitions. It lets you send test requests directly from your browser without needing a separate tool like Postman or curl.

  • In the Swagger UI, click on the POST /ask section to expand it.
  • Click Try it out.
  • Replace the example JSON body with a question about your documents:
{
  "question": "What is the company policy on remote work?"
}
  • Click Execute.
  • Wait 10-30 seconds for the agent to process. You should see a JSON response with an answer field containing a cited answer, a sources array listing which document files were used, and an iterations count showing how many retrieve-grade cycles ran.

Try an ambiguous question too

Send a vague question like "Tell me about security" and watch the terminal output. If the agent triggers the rewrite loop, you'll see the iteration count increase in the response. This confirms your agentic loop is working.

Server not starting?

Make sure your virtual environment is activated and that you installed fastapi and uvicorn in Step 1. Also confirm your terminal's working directory is the agentic-rag folder.

help me debug why uvicorn won't start my FastAPI server

Your agentic RAG system is now a fully deployed API. Any application can send questions and get back cited, self-correcting answers. Next up, you'll tackle the Secret Mission and add an adaptive query router that optimizes the pipeline based on question complexity.

Secret mission

Add an Adaptive Query Router

Your RAG agent treats every question the same way, but not all questions are equal. A simple factual lookup doesn't need the full grade-and-rewrite loop. Add an intelligent routing layer that classifies incoming questions and applies different retrieval strategies for each type.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally and uses a free OpenRouter model, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The agentic-rag project folder on your Desktop (contains all Python files, documents, FAISS index, and virtual environment).
  • A running FastAPI/uvicorn server process (if you haven't stopped it yet).

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with your agentic RAG system.

  • Your project files stay on your Desktop in the agentic-rag folder.
  • The OpenRouter free model has no usage charges, so leaving your API key configured costs nothing.
  • You can restart the FastAPI server anytime by activating your virtual environment and running uvicorn main:app --reload.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files stay on disk so you can pick up where you left off.

  • If the uvicorn server is still running, press Ctrl+C in the terminal to stop it.
  • Deactivate your virtual environment by running:
deactivate
  • To resume later, navigate back to the agentic-rag folder, activate the virtual environment, and restart the server.

Delete - I don't want to use this again

Remove all project resources and start fresh. This cannot be undone.

  • If the uvicorn server is still running, press Ctrl+C in the terminal to stop it.
  • Deactivate the virtual environment by running:
deactivate
  • Delete the entire project folder. On macOS/Linux, run:
cd ~/Desktop
rm -rf agentic-rag
  • On Windows, run:
cd ~\Desktop
rmdir /s /q agentic-rag
  • This removes all Python files, the virtual environment, the FAISS index, your documents, and the .env file containing your API key.
  • If you want to revoke your OpenRouter API key, sign in to OpenRouter and delete the key from your account's API keys page.

That's a Wrap!

That's a Wrap!

Nice work! You just built a production-grade agentic RAG system that intelligently retrieves, evaluates, and synthesizes information from your own documents.

You've learned how to:

  • Built a document ingestion pipeline that chunks text and stores embeddings in a FAISS vector store.
  • Created a LangGraph agentic workflow with retrieve, grade, rewrite, and generate nodes, using conditional routing that self-corrects when retrieval quality is poor.
  • Deployed the entire system as a FastAPI endpoint that accepts questions and returns cited answers.
  • Secret Mission: Added adaptive query routing that optimizes for speed on simple questions and thoroughness on complex ones.

Ready to quiz yourself?