Build a Document Q&A API with RAG

Build a fully async RAG API with LangGraph, ChromaDB, and local LLMs.

Introduction

30 Second Summary

Every time you search a help center or ask a chatbot about your account, there is a system behind the scenes that finds the right information and generates an answer just for your question. That system is called a RAG pipeline.

In this project, you will build a Document Q&A API that lets users upload text documents and ask natural-language questions about them. Everything runs locally on your machine with no cloud dependencies.

What You'll Build

A fully async RAG API powered by FastAPI, LangGraph, and ChromaDB that processes documents in the background and answers questions with thread-based conversation memory.

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

  • A FastAPI application with async endpoints, background task processing, and thread-aware conversations.
  • A LangGraph StateGraph with async nodes, conditional routing, and an AsyncSqliteSaver for persistent thread memory.
  • A ChromaDB vector store that indexes uploaded documents with nomic-embed-text embeddings via Ollama.
  • Secret Mission: Add a streaming response endpoint that streams the LLM answer token-by-token using Server-Sent Events.

Are there any prerequisites?

You should have basic Python knowledge and familiarity with REST APIs and async/await.

You also need Ollama installed with DeepSeek V4 already pulled. This project will guide you through pulling the embedding model and setting up the rest.

Before We Start

Before writing any application code, you need a clean project structure and the right tools installed locally.

In this step, you'll create the project directory, set up a Python virtual environment, install all dependencies, and pull the embedding model that powers document search.

In this step, get ready to:
  • Create the project directory with a modular folder structure.
  • Set up a virtual environment and install all Python dependencies.
  • Pull the nomic-embed-text embedding model in Ollama.
Check your Python version

This project requires Python 3.10 or higher because LangGraph and langchain-openai depend on it.

  • Check your Python version by running this command in your terminal:
python3 --version

✔️ I see version 3.10 or higher

Fedora 44 ships with Python 3.14 by default, so you should be all set.

Move on to the next substep.

ⓧ I see an older version

You have Python installed but it is below 3.10. Install a newer version alongside it.

  • Install Python 3.14 by running:
sudo dnf install python3.14
  • Verify the new version is available:
python3.14 --version

Use python3.14 instead of python3 for the rest of this project if your system default is still the older version.

ⓧ Command not found

Python is not installed on your system. Install it now.

  • Install Python by running:
sudo dnf install python3
  • Confirm the installation by running:
python3 --version

You should see a version number of 3.10 or higher.

Create the project directory and folder structure

A well-organized project separates concerns into modules from the start. You'll create directories for application code, tests, and persistent data.

  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder, its sub-directories, and empty init files by running these commands:
mkdir doc-qa-api
cd doc-qa-api
mkdir -p app tests data
touch app/__init__.py tests/__init__.py
  • Confirm the structure was created by running:
ls

You should see app, tests, and data listed in the output.

Why this folder structure?

The app/ directory holds your FastAPI application code, split into focused modules (endpoints, graph pipeline, vector store, config).

The tests/ directory mirrors app/ with one test file per module. The data/ directory stores your ChromaDB vector database and SQLite checkpoint file at runtime.

Don't see the folders?

Make sure you ran cd doc-qa-api before running ls. If you're still in ~/Desktop, you'll see the doc-qa-api folder itself instead of its contents.

Help me troubleshoot

Set up the virtual environment and install dependencies

A virtual environment isolates this project's packages from your system Python. This prevents version conflicts with other projects on your machine.

  • Create and activate a virtual environment by running these commands:
python3 -m venv venv
source venv/bin/activate

Your terminal prompt now shows (venv) at the beginning, confirming the virtual environment is active.

Not seeing (venv) in your prompt?

Make sure you ran source venv/bin/activate (not just the python3 -m venv venv part). The first command creates the environment. The second command activates it.

Help me troubleshoot

  • Create the requirements.txt file with all project dependencies. Copy this into a new file called requirements.txt in the doc-qa-api folder:
fastapi[standard]==0.136.3
langgraph==1.2.4
langchain-openai==1.2.2
langchain-text-splitters==1.1.2
chromadb==1.5.9
langgraph-checkpoint-sqlite==3.1.0
aiosqlite==0.20.0
pytest==8.3.5
pytest-asyncio==0.25.0
httpx==0.28.1
sse-starlette==2.3.3

What are these packages?

  • fastapi[standard] is the async web framework that serves your API endpoints.
  • langgraph is the graph-based agent orchestration framework that controls your RAG pipeline.
  • langchain-openai provides ChatOpenAI and OpenAIEmbeddings classes that connect to Ollama's OpenAI-compatible endpoint.
  • chromadb is the vector database that stores and retrieves document embeddings locally.
  • langgraph-checkpoint-sqlite and aiosqlite provide async SQLite checkpointing for thread-based conversation memory.
  • pytest, pytest-asyncio, and httpx form the testing stack for async endpoints and pipeline verification.
  • Install all dependencies by running:
pip install -r requirements.txt

This takes a minute or two as it downloads and installs all packages. You should see a final message confirming all packages were successfully installed.

Seeing installation errors?

If you see errors about missing build tools, install the Python development headers first:

sudo dnf install python3-devel gcc

Then re-run pip install -r requirements.txt.

Help me troubleshoot

Pull the nomic-embed-text embedding model

Your RAG pipeline needs an embedding model to convert text into vectors for similarity search. The nomic-embed-text model runs locally via Ollama and produces 768-dimensional vectors optimized for document retrieval.

  • Pull the embedding model by running:
ollama pull nomic-embed-text

The download is approximately 300MB. Once complete, you should see a success message confirming the model is ready.

  • Verify the model is available by running:
ollama list

You should see both deepseek-v4 and nomic-embed-text listed in the output.

Why nomic-embed-text?

Since DeepSeek V4 occupies most of your GPU memory, you need an embedding model that runs efficiently on CPU. The nomic-embed-text model is compact (~300MB) and ranks in the top 10 for open-source models on the MTEB benchmark.

It handles document-length text well and runs primarily on CPU, so it won't compete with your LLM for VRAM.

Ollama command not found?

Make sure Ollama is running. You can start it with ollama serve in a separate terminal window. If Ollama itself is not installed, follow the official install guide at ollama.com/download.

Help me troubleshoot

✔️ Awesome, I've got everything!

Great. Make sure you've saved requirements.txt and your virtual environment is still active (you see (venv) in your prompt).

ⓧ I'd like to double check the full code

Here is the complete requirements.txt file. The app/__init__.py and tests/__init__.py files should be empty.

fastapi[standard]==0.136.3
langgraph==1.2.4
langchain-openai==1.2.2
langchain-text-splitters==1.1.2
chromadb==1.5.9
langgraph-checkpoint-sqlite==3.1.0
aiosqlite==0.20.0
pytest==8.3.5
pytest-asyncio==0.25.0
httpx==0.28.1
sse-starlette==2.3.3

Your folder structure should look like this:

doc-qa-api/
├── app/
│   └── __init__.py
├── tests/
│   └── __init__.py
├── data/
├── venv/
└── requirements.txt

Your environment is ready. Next up, you'll configure the connection to Ollama and verify that both models respond correctly before building the pipeline.

Configure Ollama and Project Settings

Your environment is installed and ready to go. Now you need to tell your application how to talk to Ollama and where to store its data.

Hardcoding connection URLs, model names, and file paths throughout a codebase makes it fragile. A single config module gives you one place to change settings. Environment variable overrides let you swap values for testing without editing code.

In this step, get ready to:
  • Create a centralized configuration module for all project settings.
  • Verify DeepSeek V4 responds through Ollama's OpenAI-compatible endpoint.
  • Verify nomic-embed-text generates vector embeddings correctly.
Create the configuration module

Every setting your API needs lives in one file. This includes the Ollama connection URL, model names, chunking parameters, and database paths.

  • Create app/config.py inside your doc-qa-api folder by pasting the following code:
"""Centralized configuration for the Document Q&A API."""

import os

# Ollama connection settings
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434/v1")
OLLAMA_API_KEY = "ollama"  # Ollama accepts any string as API key

# Model settings
LLM_MODEL = os.getenv("LLM_MODEL", "deepseek-v4")
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "nomic-embed-text")

# Chunking settings
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
CHUNK_SEPARATORS = ["\n\n", "\n", ". ", " ", ""]

# Retrieval settings
RETRIEVAL_K = 4
MAX_RETRIES = 2

# ChromaDB settings
CHROMA_PERSIST_PATH = os.getenv("CHROMA_PERSIST_PATH", "./data/chroma_db")

# Checkpoint settings
CHECKPOINT_DB_PATH = os.getenv("CHECKPOINT_DB_PATH", "./data/checkpoints.db")

What does this code do?

  • OLLAMA_BASE_URL points to Ollama's OpenAI-compatible endpoint. Using os.getenv() with a default means you can override it in tests without editing source code.
  • OLLAMA_API_KEY is set to "ollama" because Ollama requires the field but ignores the value.
  • CHUNK_SIZE and CHUNK_OVERLAP control how documents are split. 500 characters with 50 overlap balances context against precision.
  • CHUNK_SEPARATORS defines the priority order for splitting. Double newlines are tried first, single newlines next, then sentences, then words.
  • CHROMA_PERSIST_PATH and CHECKPOINT_DB_PATH store data in ./data/ so both databases live in a gitignored folder.
  • Save app/config.py.
  • Confirm the file exists by running:
ls app/config.py

You should see app/config.py printed in the output.

Why environment variables for some settings?

Settings that might change between environments (local dev vs CI vs production) use os.getenv() with sensible defaults. For example, tests can pass CHECKPOINT_DB_PATH=":memory:" to use an in-memory database instead of writing to disk.

Fixed values like CHUNK_SIZE and MAX_RETRIES are constants that rarely change per environment. They stay as plain assignments.

Verify DeepSeek V4 responds via Ollama

Before writing any application code, you need to confirm that DeepSeek V4 is loaded and responding through the OpenAI-compatible API.

  • Test the chat completions endpoint by running this command:
curl http://localhost:11434/v1/chat/completions \
    -H "Content-Type: application/json" \
    -d '{
        "model": "deepseek-v4",
        "messages": [
            {
                "role": "user",
                "content": "Say hello in one word."
            }
        ]
    }'

You should see a JSON response containing a choices array with a message from the model. The first response may take a few seconds while Ollama loads the model into memory.

Not getting a response?

  • Make sure Ollama is running. Open a separate terminal and run ollama serve if it is not already active.
  • Confirm the model is pulled by running ollama list and checking for deepseek-v4 in the output.
  • If you see a connection refused error, check that nothing else is using port 11434.

Help me troubleshoot why Ollama is not responding to my curl request.

Verify nomic-embed-text generates embeddings

The RAG pipeline depends on vector embeddings to find relevant document chunks. You need to confirm nomic-embed-text returns embedding vectors through the same endpoint.

  • Test the embeddings endpoint by running this command:
curl http://localhost:11434/v1/embeddings \
    -H "Content-Type: application/json" \
    -d '{
        "model": "nomic-embed-text",
        "input": "Hello world"
    }'

You should see a JSON response containing a data array with an embedding field holding an array of floating-point numbers. This is the 768-dimensional vector representation of the input text.

Seeing an empty or error response?

  • Confirm the model is available by running ollama list and looking for nomic-embed-text.
  • If the model is missing, pull it by running ollama pull nomic-embed-text.

Help me troubleshoot why nomic-embed-text is not returning embeddings through Ollama.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/config.py and both curl commands returned valid JSON responses.

ⓧ I'd like to double check the full code

Here is the complete app/config.py file:

"""Centralized configuration for the Document Q&A API."""

import os

# Ollama connection settings
OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://localhost:11434/v1")
OLLAMA_API_KEY = "ollama"  # Ollama accepts any string as API key

# Model settings
LLM_MODEL = os.getenv("LLM_MODEL", "deepseek-v4")
EMBEDDING_MODEL = os.getenv("EMBEDDING_MODEL", "nomic-embed-text")

# Chunking settings
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
CHUNK_SEPARATORS = ["\n\n", "\n", ". ", " ", ""]

# Retrieval settings
RETRIEVAL_K = 4
MAX_RETRIES = 2

# ChromaDB settings
CHROMA_PERSIST_PATH = os.getenv("CHROMA_PERSIST_PATH", "./data/chroma_db")

# Checkpoint settings
CHECKPOINT_DB_PATH = os.getenv("CHECKPOINT_DB_PATH", "./data/checkpoints.db")

Both models are confirmed working. Next up, you will build the vector store layer that chunks documents and stores their embeddings in ChromaDB.

Build the Vector Store Layer

In the previous step, you confirmed that both DeepSeek V4 and nomic-embed-text respond correctly through Ollama. Now it is time to put nomic-embed-text to work by building the layer that stores and retrieves document chunks.

A vector store turns plain text into numerical representations (embeddings) that capture meaning. When a user asks a question, the store finds the chunks whose embeddings are closest to the question's embedding. This is the core of any RAG pipeline.

In this step, get ready to:
  • Create a vector store module backed by ChromaDB and nomic-embed-text.
  • Implement async functions for adding and retrieving document chunks.
  • Write tests that verify chunking and similarity search.
Initialize ChromaDB and the embedding model

The vector store module needs three pieces: a persistent database to hold embeddings, an embedding model to generate them, and a text splitter to break documents into meaningful chunks.

  • Create a new file called app/vector_store.py in your editor.
  • Add the following imports and module-level initialization:
"""Vector store layer using ChromaDB and nomic-embed-text via Ollama (async)."""

import chromadb
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

from app.config import (
    OLLAMA_BASE_URL,
    OLLAMA_API_KEY,
    EMBEDDING_MODEL,
    CHUNK_SIZE,
    CHUNK_OVERLAP,
    CHUNK_SEPARATORS,
    CHROMA_PERSIST_PATH,
    RETRIEVAL_K,
)

# Initialize ChromaDB with persistent storage
chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_PATH)
collection = chroma_client.get_or_create_collection(name="documents")

# Initialize embeddings pointing to Ollama's OpenAI-compatible endpoint
embeddings = OpenAIEmbeddings(
    base_url=OLLAMA_BASE_URL,
    api_key=OLLAMA_API_KEY,
    model=EMBEDDING_MODEL,
)

# Text splitter for chunking documents
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=CHUNK_SIZE,
    chunk_overlap=CHUNK_OVERLAP,
    separators=CHUNK_SEPARATORS,
)

What does this code do?

  • chromadb.PersistentClient creates a local database that persists embeddings to disk at ./data/chroma_db.
  • OpenAIEmbeddings points at Ollama's OpenAI-compatible endpoint so it uses nomic-embed-text locally instead of calling OpenAI.
  • RecursiveCharacterTextSplitter breaks long documents into 500-character chunks with 50-character overlap, splitting at natural boundaries like paragraph breaks and sentence ends.

Why these chunking settings?

Chunk size directly impacts RAG quality. Too large (1000+ characters) and you dilute relevant info with noise. Too small (100 characters) and you lose context.

Starting at 500 characters with 50 character overlap is a strong baseline. The semantic separators (\n\n > \n > . > ) ensure chunks break at natural boundaries so edge information is not lost.

You will not see a change yet. The async functions in the next section use these objects.

Implement async chunking and retrieval functions

With the initialization in place, you need two async functions: one to chunk and embed a document, and one to query the store for relevant chunks.

  • Below the text_splitter initialization, add the add_document function:

async def add_document(text: str, doc_id: str) -> int:
    """Chunk and add a document to the vector store. Returns number of chunks added."""
    if not text.strip():
        return 0

    chunks = text_splitter.split_text(text)

    for i, chunk in enumerate(chunks):
        chunk_id = f"{doc_id}_chunk_{i}"
        embedding = await embeddings.aembed_query(chunk)
        collection.add(
            ids=[chunk_id],
            embeddings=[embedding],
            documents=[chunk],
            metadatas=[{"doc_id": doc_id, "chunk_index": i}],
        )

    return len(chunks)

What does this function do?

  • It guards against empty input by returning 0 immediately if the text is blank.
  • text_splitter.split_text(text) breaks the document into overlapping chunks.
  • await embeddings.aembed_query(chunk) generates an embedding for each chunk asynchronously, preventing the event loop from blocking while Ollama processes.
  • Each chunk is stored in ChromaDB with its embedding, full text, and metadata linking it back to the parent document.

Why use aembed_query instead of embed_query?

Using await embeddings.aembed_query() instead of the synchronous embeddings.embed_query() prevents blocking the event loop while waiting for Ollama to return embeddings.

In background tasks, this is critical for throughput. Multiple documents can be processed concurrently without one embedding call freezing everything else.

You will not see a change yet. The test at the end of this step exercises this function.

  • Below add_document, add the retrieve_documents function:

async def retrieve_documents(query: str, k: int = RETRIEVAL_K) -> list[str]:
    """Retrieve the top-k most relevant document chunks for a query."""
    if collection.count() == 0:
        return []

    query_embedding = await embeddings.aembed_query(query)
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=min(k, collection.count()),
    )
    return results["documents"][0] if results["documents"] else []

What does this function do?

  • It returns an empty list immediately if no documents have been indexed yet.
  • It embeds the query using the same model that embedded the documents, ensuring the vectors live in the same space.
  • collection.query() performs similarity search and returns the top-k closest chunks.
  • min(k, collection.count()) prevents ChromaDB from erroring when fewer chunks exist than requested.
  • Save app/vector_store.py.
  • Verify the module is importable by running:
python -c "from app.vector_store import add_document, retrieve_documents; print('Imports OK')"

You should see Imports OK printed with no errors.

Seeing an ImportError?

  • Make sure you are running the command from the doc-qa-api project root (not from inside the app/ folder).
  • Check that your virtual environment is activated. The chromadb and langchain-openai packages must be installed.
  • Confirm app/__init__.py exists (even if empty). Python needs it to treat app/ as a package.

Help me debug this ImportError in my vector store module.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/vector_store.py before moving on to writing tests.

ⓧ I'd like to double check the full code

"""Vector store layer using ChromaDB and nomic-embed-text via Ollama (async)."""

import chromadb
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

from app.config import (
    OLLAMA_BASE_URL,
    OLLAMA_API_KEY,
    EMBEDDING_MODEL,
    CHUNK_SIZE,
    CHUNK_OVERLAP,
    CHUNK_SEPARATORS,
    CHROMA_PERSIST_PATH,
    RETRIEVAL_K,
)

# Initialize ChromaDB with persistent storage
chroma_client = chromadb.PersistentClient(path=CHROMA_PERSIST_PATH)
collection = chroma_client.get_or_create_collection(name="documents")

# Initialize embeddings pointing to Ollama's OpenAI-compatible endpoint
embeddings = OpenAIEmbeddings(
    base_url=OLLAMA_BASE_URL,
    api_key=OLLAMA_API_KEY,
    model=EMBEDDING_MODEL,
)

# Text splitter for chunking documents
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=CHUNK_SIZE,
    chunk_overlap=CHUNK_OVERLAP,
    separators=CHUNK_SEPARATORS,
)


async def add_document(text: str, doc_id: str) -> int:
    """Chunk and add a document to the vector store. Returns number of chunks added."""
    if not text.strip():
        return 0

    chunks = text_splitter.split_text(text)

    for i, chunk in enumerate(chunks):
        chunk_id = f"{doc_id}_chunk_{i}"
        embedding = await embeddings.aembed_query(chunk)
        collection.add(
            ids=[chunk_id],
            embeddings=[embedding],
            documents=[chunk],
            metadatas=[{"doc_id": doc_id, "chunk_index": i}],
        )

    return len(chunks)


async def retrieve_documents(query: str, k: int = RETRIEVAL_K) -> list[str]:
    """Retrieve the top-k most relevant document chunks for a query."""
    if collection.count() == 0:
        return []

    query_embedding = await embeddings.aembed_query(query)
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=min(k, collection.count()),
    )
    return results["documents"][0] if results["documents"] else []
Write tests for the vector store

Automated tests catch regressions early and confirm your async embedding logic works end-to-end against the live Ollama instance.

  • Create a new file called tests/test_vector_store.py and add the following test cases:
"""Tests for the vector store layer."""

import pytest
from app.vector_store import add_document, retrieve_documents


class TestAddDocument:
    @pytest.mark.asyncio
    async def test_add_document_returns_chunk_count(self):
        text = "Python is a programming language. " * 50
        count = await add_document(text, "test-doc-1")
        assert count > 0

    @pytest.mark.asyncio
    async def test_add_empty_document_returns_zero(self):
        count = await add_document("", "test-doc-empty")
        assert count == 0


class TestRetrieveDocuments:
    @pytest.mark.asyncio
    async def test_retrieve_returns_list(self):
        await add_document("FastAPI is an async web framework. " * 50, "test-doc-2")
        results = await retrieve_documents("What is FastAPI?")
        assert isinstance(results, list)
        assert len(results) > 0

What do these tests cover?

  • test_add_document_returns_chunk_count confirms that a long text gets split into multiple chunks and embedded successfully.
  • test_add_empty_document_returns_zero verifies the empty-input guard works.
  • test_retrieve_returns_list indexes a document and then queries it, proving the full add-then-retrieve cycle works end-to-end.

These tests call the real Ollama instance. Make sure Ollama is running before you execute them.

  • Save tests/test_vector_store.py.
  • Run the tests by executing:
pytest tests/test_vector_store.py -v

You should see all three tests pass with green output. The first run may take a few seconds as Ollama generates embeddings.

Tests failing or hanging?

  • Confirm Ollama is running (ollama list should show your models). If Ollama is not running, start it with ollama serve.
  • If you see a PytestUnhandledCoroutineWarning, make sure pytest-asyncio is installed in your virtual environment.
  • If tests hang indefinitely, nomic-embed-text may not be pulled yet. Run ollama pull nomic-embed-text and try again.

Help me debug my failing vector store tests.

Your vector store layer is complete. Documents go in, get chunked, and come back as relevant results. Next up, you will wire this into a LangGraph pipeline that retrieves, grades, and generates answers.

Build the Async LangGraph RAG Pipeline

Your vector store can chunk documents and retrieve relevant pieces. Now you need a reasoning layer that decides what to do with those pieces.

A simple RAG chain retrieves once and generates immediately. But what if the retrieved chunks are irrelevant? LangGraph solves this by turning the pipeline into a stateful graph with conditional routing. If retrieved documents are not useful, the graph rewrites the question and retries. It also persists conversation state per thread so follow-up questions have context.

In this step, get ready to:
  • Define the graph state schema and initialize the LLM connection.
  • Create async node functions for retrieval, grading, rewriting, and generation.
  • Wire nodes into a StateGraph with conditional edges and an AsyncSqliteSaver checkpointer.
Define the graph state and initialize the LLM

Every LangGraph graph operates on a shared state object. You will define this state as a TypedDict that carries the question, retrieved documents, the generated answer, and a retry counter.

  • Create a new file app/rag_graph.py inside your app/ folder.
  • Add the imports, LLM initialization, state schema, and the first node function by pasting this code:
"""Async LangGraph RAG pipeline with checkpointing for thread-based memory."""

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langchain_openai import ChatOpenAI

from app.config import OLLAMA_BASE_URL, OLLAMA_API_KEY, LLM_MODEL, MAX_RETRIES, CHECKPOINT_DB_PATH
from app.vector_store import retrieve_documents

# Initialize the LLM pointing to Ollama's OpenAI-compatible endpoint
llm = ChatOpenAI(
    base_url=OLLAMA_BASE_URL,
    api_key=OLLAMA_API_KEY,
    model=LLM_MODEL,
    temperature=0,
)


# Define the graph state
class RAGState(TypedDict):
    question: str
    documents: list[str]
    generation: str
    retry_count: int


# Async Node: Retrieve documents
async def retrieve(state: RAGState) -> dict:
    """Retrieve relevant documents from the vector store."""
    question = state["question"]
    documents = await retrieve_documents(question)
    return {"documents": documents}

What does this code do?

The ChatOpenAI instance connects to your local Ollama server using the same OpenAI-compatible pattern you used for embeddings. Setting temperature=0 makes responses deterministic, which is important for grading and rewriting steps.

The RAGState TypedDict is the shared data structure every node reads from and writes to. The retry_count field prevents infinite loops when the graph keeps rewriting without finding relevant documents.

The retrieve node is a thin async wrapper around the retrieve_documents function you built in the previous step.

Why LangGraph over a simple chain?

A basic RAG chain retrieves once and generates immediately. If the retrieved chunks are irrelevant, the answer is garbage. LangGraph adds conditional routing so the graph can rewrite the question and retry retrieval.

It also makes the decision-making visible and testable. Each node is a pure async function. You can unit-test grading logic independently from generation logic.

  • You won't see a change yet. The next chunk adds the processing nodes that use this state.
Create the async grading, rewriting, and generation nodes

Each node in the graph is an async def function that receives the current state and returns a dictionary of updates. These nodes call the LLM using await llm.ainvoke() to avoid blocking the event loop.

  • Add the document grading node below the retrieve function in app/rag_graph.py:
# Async Node: Grade documents for relevance
async def grade_documents(state: RAGState) -> dict:
    """Filter documents that are relevant to the question."""
    question = state["question"]
    documents = state["documents"]

    if not documents:
        return {"documents": []}

    relevant_docs = []
    for doc in documents:
        response = await llm.ainvoke(
            [
                {
                    "role": "system",
                    "content": (
                        "You are a relevance grader. Given a question and a document, "
                        "respond with only 'yes' if the document is relevant to answering "
                        "the question, or 'no' if it is not."
                    ),
                },
                {
                    "role": "user",
                    "content": f"Question: {question}\n\nDocument: {doc}",
                },
            ]
        )
        if "yes" in response.content.lower():
            relevant_docs.append(doc)

    return {"documents": relevant_docs}

What does the grading node do?

The grader asks the LLM a simple yes/no question for each retrieved chunk: is this document relevant to the user's question? Only chunks that pass the relevance check move forward to generation.

This filtering step prevents the generator from hallucinating answers based on irrelevant context. If all documents are filtered out, the graph routes to a rewrite step instead.

  • Add the question rewriting and answer generation nodes below the grade_documents function:
# Async Node: Rewrite the question
async def rewrite_question(state: RAGState) -> dict:
    """Rewrite the question to improve retrieval results."""
    question = state["question"]
    retry_count = state.get("retry_count", 0)

    response = await llm.ainvoke(
        [
            {
                "role": "system",
                "content": (
                    "You are a question rewriter. Rewrite the given question to be "
                    "more specific and better suited for retrieving relevant documents "
                    "from a vector store. Return only the rewritten question."
                ),
            },
            {"role": "user", "content": question},
        ]
    )

    return {"question": response.content, "retry_count": retry_count + 1}


# Async Node: Generate answer
async def generate_answer(state: RAGState) -> dict:
    """Generate an answer using the retrieved context."""
    question = state["question"]
    documents = state["documents"]
    context = "\n\n".join(documents) if documents else "No relevant context found."

    response = await llm.ainvoke(
        [
            {
                "role": "system",
                "content": (
                    "You are a helpful assistant that answers questions based on the "
                    "provided context. If the context does not contain enough information "
                    "to answer the question, say so clearly. Keep answers concise and "
                    "grounded in the context."
                ),
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {question}",
            },
        ]
    )

    return {"generation": response.content}

What do these nodes do?

The rewrite_question node asks the LLM to rephrase the original question for better vector search results. It also increments retry_count so the graph knows when to stop retrying.

The generate_answer node joins all relevant document chunks into a single context string and asks the LLM to answer the question based on that context. This is the final node that produces the user-facing response.

  • You won't see a change yet. The next chunk wires these nodes into a working graph.
Wire the graph with conditional routing and checkpointing

With all nodes defined, you need to connect them into a graph that makes routing decisions at runtime. The routing function checks whether relevant documents were found and whether the retry limit has been reached.

  • Add the routing function, graph wiring, and compilation helper below the generate_answer function in app/rag_graph.py:
# Routing function: decide whether to generate or rewrite
def route_after_grading(state: RAGState) -> str:
    """Route based on whether relevant documents were found."""
    documents = state["documents"]
    retry_count = state.get("retry_count", 0)

    if documents:
        return "generate_answer"
    elif retry_count >= MAX_RETRIES:
        return "generate_answer"
    else:
        return "rewrite_question"


# Build the graph
workflow = StateGraph(RAGState)

workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("rewrite_question", rewrite_question)
workflow.add_node("generate_answer", generate_answer)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    route_after_grading,
    {
        "generate_answer": "generate_answer",
        "rewrite_question": "rewrite_question",
    },
)
workflow.add_edge("rewrite_question", "retrieve")
workflow.add_edge("generate_answer", END)


# Compile with AsyncSqliteSaver for thread-based persistence
async def get_compiled_graph():
    """Get the compiled graph with async SQLite checkpointer."""
    checkpointer = AsyncSqliteSaver.from_conn_string(CHECKPOINT_DB_PATH)
    return workflow.compile(checkpointer=checkpointer)

What does this code do?

The route_after_grading function is the decision point. If relevant documents exist, proceed to generation. If not and retries remain, rewrite the question and loop back to retrieval. If retries are exhausted (MAX_RETRIES is 2), generate anyway with whatever context is available.

The add_conditional_edges call is what makes this a graph rather than a linear chain. It inspects state after grading and routes to different nodes based on the result.

The AsyncSqliteSaver checkpointer saves graph state after every step. When you pass a thread_id in config, the graph resumes from the last checkpoint for that thread. This gives each conversation its own persistent memory.

  • Save app/rag_graph.py.
  • Verify the module imports correctly by running:
python -c "from app.rag_graph import get_compiled_graph; print('Graph module loaded')"
  • You should see Graph module loaded printed to the terminal.

Seeing an ImportError?

  • Make sure you are running the command from the doc-qa-api project root (the folder containing the app/ directory).
  • Check that your virtual environment is activated. The langgraph-checkpoint-sqlite package must be installed for the AsyncSqliteSaver import to work.
  • Verify there are no typos in the import paths. The checkpoint import is from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver.

Why is my rag_graph.py failing to import?

✔️ Awesome, I've got everything!

Great. Double check you have saved app/rag_graph.py and confirmed it imports cleanly.

ⓧ I'd like to double check the full code

"""Async LangGraph RAG pipeline with checkpointing for thread-based memory."""

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langchain_openai import ChatOpenAI

from app.config import OLLAMA_BASE_URL, OLLAMA_API_KEY, LLM_MODEL, MAX_RETRIES, CHECKPOINT_DB_PATH
from app.vector_store import retrieve_documents

# Initialize the LLM pointing to Ollama's OpenAI-compatible endpoint
llm = ChatOpenAI(
    base_url=OLLAMA_BASE_URL,
    api_key=OLLAMA_API_KEY,
    model=LLM_MODEL,
    temperature=0,
)


# Define the graph state
class RAGState(TypedDict):
    question: str
    documents: list[str]
    generation: str
    retry_count: int


# Async Node: Retrieve documents
async def retrieve(state: RAGState) -> dict:
    """Retrieve relevant documents from the vector store."""
    question = state["question"]
    documents = await retrieve_documents(question)
    return {"documents": documents}


# Async Node: Grade documents for relevance
async def grade_documents(state: RAGState) -> dict:
    """Filter documents that are relevant to the question."""
    question = state["question"]
    documents = state["documents"]

    if not documents:
        return {"documents": []}

    relevant_docs = []
    for doc in documents:
        response = await llm.ainvoke(
            [
                {
                    "role": "system",
                    "content": (
                        "You are a relevance grader. Given a question and a document, "
                        "respond with only 'yes' if the document is relevant to answering "
                        "the question, or 'no' if it is not."
                    ),
                },
                {
                    "role": "user",
                    "content": f"Question: {question}\n\nDocument: {doc}",
                },
            ]
        )
        if "yes" in response.content.lower():
            relevant_docs.append(doc)

    return {"documents": relevant_docs}


# Async Node: Rewrite the question
async def rewrite_question(state: RAGState) -> dict:
    """Rewrite the question to improve retrieval results."""
    question = state["question"]
    retry_count = state.get("retry_count", 0)

    response = await llm.ainvoke(
        [
            {
                "role": "system",
                "content": (
                    "You are a question rewriter. Rewrite the given question to be "
                    "more specific and better suited for retrieving relevant documents "
                    "from a vector store. Return only the rewritten question."
                ),
            },
            {"role": "user", "content": question},
        ]
    )

    return {"question": response.content, "retry_count": retry_count + 1}


# Async Node: Generate answer
async def generate_answer(state: RAGState) -> dict:
    """Generate an answer using the retrieved context."""
    question = state["question"]
    documents = state["documents"]
    context = "\n\n".join(documents) if documents else "No relevant context found."

    response = await llm.ainvoke(
        [
            {
                "role": "system",
                "content": (
                    "You are a helpful assistant that answers questions based on the "
                    "provided context. If the context does not contain enough information "
                    "to answer the question, say so clearly. Keep answers concise and "
                    "grounded in the context."
                ),
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {question}",
            },
        ]
    )

    return {"generation": response.content}


# Routing function: decide whether to generate or rewrite
def route_after_grading(state: RAGState) -> str:
    """Route based on whether relevant documents were found."""
    documents = state["documents"]
    retry_count = state.get("retry_count", 0)

    if documents:
        return "generate_answer"
    elif retry_count >= MAX_RETRIES:
        return "generate_answer"
    else:
        return "rewrite_question"


# Build the graph
workflow = StateGraph(RAGState)

workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("rewrite_question", rewrite_question)
workflow.add_node("generate_answer", generate_answer)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    route_after_grading,
    {
        "generate_answer": "generate_answer",
        "rewrite_question": "rewrite_question",
    },
)
workflow.add_edge("rewrite_question", "retrieve")
workflow.add_edge("generate_answer", END)


# Compile with AsyncSqliteSaver for thread-based persistence
async def get_compiled_graph():
    """Get the compiled graph with async SQLite checkpointer."""
    checkpointer = AsyncSqliteSaver.from_conn_string(CHECKPOINT_DB_PATH)
    return workflow.compile(checkpointer=checkpointer)

Now write a test file to verify the graph runs end-to-end with a test thread.

  • Create a new file tests/test_rag_graph.py and paste this code:
"""Tests for the async LangGraph RAG pipeline."""

import pytest
from app.rag_graph import get_compiled_graph
from app.vector_store import add_document


class TestRAGGraph:
    @pytest.mark.asyncio
    async def test_graph_returns_generation(self):
        await add_document("LangGraph is a stateful agent framework built on LangChain. " * 30, "graph-test-doc")
        graph = await get_compiled_graph()
        config = {"configurable": {"thread_id": "test-thread-1"}}
        result = await graph.ainvoke(
            {"question": "What is LangGraph?", "documents": [], "generation": "", "retry_count": 0},
            config,
        )
        assert "generation" in result
        assert len(result["generation"]) > 0

    @pytest.mark.asyncio
    async def test_graph_uses_thread_id_for_memory(self):
        graph = await get_compiled_graph()
        config = {"configurable": {"thread_id": "test-thread-memory"}}
        result1 = await graph.ainvoke(
            {"question": "What is LangGraph?", "documents": [], "generation": "", "retry_count": 0},
            config,
        )
        assert result1["generation"]

What do these tests verify?

The first test uploads a document, compiles the graph, and invokes it with a question. It asserts that the graph produces a non-empty generation field, proving the full retrieve-grade-generate flow works.

The second test verifies that passing a thread_id in the config does not break execution. The checkpointer saves state under that thread, enabling conversation continuity in later steps.

  • Save tests/test_rag_graph.py.
  • Run the graph tests by running:
pytest tests/test_rag_graph.py -v
  • You should see both tests pass. The first test takes longer because DeepSeek V4 runs inference locally for grading and generation.

Tests timing out or failing?

  • Make sure Ollama is running. Check with ollama list to confirm both deepseek-v4 and nomic-embed-text models are available.
  • Local LLM inference can be slow. If a test times out, increase the timeout with pytest tests/test_rag_graph.py -v --timeout=120.
  • If you see a database lock error, delete the data/checkpoints.db file and run the test again.

Why are my RAG graph tests failing?

Your RAG pipeline can now retrieve, grade, rewrite, and generate with full thread memory. Next up, you will wrap this graph in a FastAPI application with async endpoints and background document processing.

Create the FastAPI Application with Background Tasks

You have a working RAG graph that retrieves, grades, rewrites, and generates answers with persistent thread memory. Now you need a way for users to actually interact with it.

In this step, you will build the web layer that ties everything together. Document uploads run asynchronously in the background (returning a 202 response instantly), and questions are routed through your LangGraph pipeline with thread-based conversation continuity.

In this step, get ready to:
  • Create a background task module for async document processing.
  • Build a fully async FastAPI application with endpoints for document upload, task polling, and Q&A.
  • Write integration tests and verify the full flow via Swagger UI.
Define the background task module

Before building the API endpoints, you need a module that handles document processing in the background. This keeps the API responsive by returning immediately while the heavy embedding work happens asynchronously.

  • Create a new file called app/tasks.py and add the following code:
"""Background task definitions for async document processing."""

import uuid
from datetime import datetime

from app.vector_store import add_document

# In-memory task status store (use SQLite/Redis for production persistence)
task_store: dict[str, dict] = {}


async def process_document_task(task_id: str, text: str, title: str) -> None:
    """Background task: chunk and embed a document into the vector store."""
    task_store[task_id]["status"] = "processing"
    task_store[task_id]["started_at"] = datetime.utcnow().isoformat()

    try:
        doc_id = str(uuid.uuid4())
        chunks_added = await add_document(text, doc_id)
        task_store[task_id].update({
            "status": "completed",
            "doc_id": doc_id,
            "chunks_added": chunks_added,
            "completed_at": datetime.utcnow().isoformat(),
        })
    except Exception as e:
        task_store[task_id].update({
            "status": "failed",
            "error": str(e),
            "completed_at": datetime.utcnow().isoformat(),
        })

What does this code do?

This module defines the work that happens after the API accepts a document upload:

  • task_store is a simple dictionary that tracks each task's status, timestamps, and results.
  • process_document_task is an async function that updates the task status to "processing", calls add_document() to chunk and embed the text, then marks the task as "completed" or "failed".

By separating task logic from the endpoint handler, you can test background processing independently.

  • Save app/tasks.py.

You won't see a change yet. The next file (app/main.py) imports and uses this module.

Build the FastAPI application

Now you will create the main application file that wires together your RAG graph, background tasks, and HTTP endpoints. The app uses a lifespan context manager to compile the LangGraph graph once at startup.

  • Create a new file called app/main.py and add the imports, lifespan, app instance, and Pydantic models:
"""Async FastAPI application for the Document Q&A API."""

import uuid
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field

from app.config import LLM_MODEL, EMBEDDING_MODEL
from app.rag_graph import get_compiled_graph
from app.tasks import process_document_task, task_store
from app.vector_store import collection


# Lifespan: initialize the compiled graph with checkpointer
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.rag_graph = await get_compiled_graph()
    yield


app = FastAPI(
    title="Document Q&A API",
    description="Async RAG API with thread memory, background processing, powered by LangGraph and DeepSeek V4",
    version="2.0.0",
    lifespan=lifespan,
)


# --- Request/Response Models ---

class DocumentInput(BaseModel):
    text: str = Field(..., min_length=1, description="The document text to index")
    title: str = Field(default="Untitled", description="Document title for reference")


class QuestionInput(BaseModel):
    question: str = Field(..., min_length=1, description="The question to ask")
    thread_id: str | None = Field(default=None, description="Thread ID for conversation continuity")


class TaskAcceptedResponse(BaseModel):
    task_id: str
    status: str
    message: str


class TaskStatusResponse(BaseModel):
    task_id: str
    status: str
    doc_id: str | None = None
    chunks_added: int | None = None
    error: str | None = None


class AnswerResponse(BaseModel):
    answer: str
    question: str
    thread_id: str


class HealthResponse(BaseModel):
    status: str
    llm_model: str
    embedding_model: str
    documents_indexed: int

What does this code do?

  • The lifespan context manager compiles your LangGraph graph with its AsyncSqliteSaver checkpointer once at startup, storing it on app.state so every request can access it.
  • The FastAPI instance is configured with a title, description, and version that appear in the auto-generated docs.
  • The Pydantic models define the exact shape of request bodies and responses, giving you automatic validation and documentation.
  • Save app/main.py.

You won't see a change yet. The next chunk adds the first endpoint you can test.

  • Add the health check endpoint below the models in app/main.py:
# --- Endpoints ---

@app.get("/health", response_model=HealthResponse)
async def health_check():
    """Check API health and Ollama connectivity."""
    return HealthResponse(
        status="healthy",
        llm_model=LLM_MODEL,
        embedding_model=EMBEDDING_MODEL,
        documents_indexed=collection.count(),
    )

What does this code do?

The /health endpoint returns the API status, model names, and how many documents are currently indexed. It acts as a quick diagnostic to confirm the server is running and connected to the right models.

  • Save app/main.py.
  • Start the development server by running this command:
fastapi dev app/main.py
  • In a separate terminal, test the health endpoint by running:
curl http://localhost:8000/health

You should see a JSON response with "status": "healthy", the model names, and "documents_indexed": 0.

Server won't start?

  • Check that Ollama is running. The lifespan initializes the graph which depends on the Ollama connection being available.
  • Make sure your virtual environment is activated. The fastapi CLI is only available inside the venv where you installed fastapi[standard].
  • Verify that app/tasks.py exists. The import in main.py will fail if the file is missing.

Help me debug why my FastAPI server won't start.

  • Add the document upload endpoint below the health check in app/main.py:
@app.post("/documents", response_model=TaskAcceptedResponse, status_code=202)
async def upload_document(doc: DocumentInput, background_tasks: BackgroundTasks):
    """Upload a document for async processing. Returns 202 with a task_id."""
    task_id = str(uuid.uuid4())
    task_store[task_id] = {
        "status": "queued",
        "title": doc.title,
        "created_at": __import__("datetime").datetime.utcnow().isoformat(),
    }

    background_tasks.add_task(process_document_task, task_id, doc.text, doc.title)

    return TaskAcceptedResponse(
        task_id=task_id,
        status="queued",
        message=f"Document '{doc.title}' queued for processing. Poll GET /tasks/{task_id} for status.",
    )

Why return 202 instead of 200?

A 202 Accepted response tells the client that the request was valid but processing has not finished yet. Document chunking and embedding is CPU/GPU-intensive work that would block the response for seconds or minutes.

By offloading to BackgroundTasks, the API returns instantly with a task_id the client can poll. This is the standard async processing pattern for production APIs.

  • Save app/main.py. The dev server auto-reloads.
  • Test the document upload by running:
curl -X POST http://localhost:8000/documents \
  -H "Content-Type: application/json" \
  -d '{"text": "FastAPI is a modern async web framework for Python.", "title": "FastAPI Intro"}'

You should see a JSON response with a task_id, "status": "queued", and a message telling you where to poll.

Getting a 422 validation error?

  • Make sure the Content-Type header is set to application/json. Without it, FastAPI cannot parse the request body.
  • Check that your JSON body includes the text field with at least 1 character. The min_length=1 validator rejects empty strings.

Help me fix my 422 error when posting a document.

  • Add the task polling endpoint below the document upload in app/main.py:
@app.get("/tasks/{task_id}", response_model=TaskStatusResponse)
async def get_task_status(task_id: str):
    """Poll the status of a document processing task."""
    if task_id not in task_store:
        raise HTTPException(status_code=404, detail="Task not found.")

    task = task_store[task_id]
    return TaskStatusResponse(
        task_id=task_id,
        status=task["status"],
        doc_id=task.get("doc_id"),
        chunks_added=task.get("chunks_added"),
        error=task.get("error"),
    )

What does this code do?

This endpoint lets clients poll the status of a background task using the task_id returned by POST /documents. It returns the current status ("queued", "processing", "completed", or "failed") along with metadata like chunks_added once processing finishes.

  • Save app/main.py. The dev server auto-reloads.
  • Poll the task status using the task_id from your previous response by running:
curl http://localhost:8000/tasks/[[TASK_ID="the task_id from your previous curl response"]]

You should see "status": "completed" with a doc_id and chunks_added count.

Status still showing "queued"?

  • Wait a few seconds and poll again. Embedding generation takes time, especially on the first call while Ollama loads the model into memory.
  • Check your server terminal for errors. If you see a connection refused error, make sure Ollama is running with nomic-embed-text available.

Help me figure out why my task is stuck in queued status.

  • Add the question-answering endpoint below the task status endpoint in app/main.py:
@app.post("/ask", response_model=AnswerResponse)
async def ask_question(q: QuestionInput):
    """Ask a question. Provide thread_id for conversation continuity."""
    if collection.count() == 0:
        raise HTTPException(
            status_code=400,
            detail="No documents indexed yet. Upload documents first via POST /documents.",
        )

    thread_id = q.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    result = await app.state.rag_graph.ainvoke(
        {"question": q.question, "documents": [], "generation": "", "retry_count": 0},
        config,
    )

    return AnswerResponse(
        answer=result["generation"],
        question=q.question,
        thread_id=thread_id,
    )

How does thread-based memory work here?

When the client sends a thread_id, the graph resumes from the last checkpoint saved for that thread. This means follow-up questions within the same thread have access to previous conversation context.

If no thread_id is provided, the endpoint generates a fresh one and returns it in the response. The client can then use that ID for subsequent questions in the same conversation.

  • Save app/main.py. The dev server auto-reloads.
  • Test the Q&A endpoint by running:
curl -X POST http://localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{"question": "What is FastAPI?"}'

You should see a JSON response containing an answer grounded in your uploaded document, along with a thread_id you can reuse for follow-up questions.

Getting a 400 error about no documents?

  • Make sure your earlier document upload completed successfully. Poll GET /tasks/{task_id} and confirm it shows "status": "completed" before asking questions.
  • If the task failed, check the error field in the task status response for details.

Help me debug why /ask returns no documents indexed.

✔️ Awesome, I've got everything!

Great. Double check you have saved app/main.py with all four endpoints.

ⓧ I'd like to double check the full code

"""Async FastAPI application for the Document Q&A API."""

import uuid
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field

from app.config import LLM_MODEL, EMBEDDING_MODEL
from app.rag_graph import get_compiled_graph
from app.tasks import process_document_task, task_store
from app.vector_store import collection


# Lifespan: initialize the compiled graph with checkpointer
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.rag_graph = await get_compiled_graph()
    yield


app = FastAPI(
    title="Document Q&A API",
    description="Async RAG API with thread memory, background processing, powered by LangGraph and DeepSeek V4",
    version="2.0.0",
    lifespan=lifespan,
)


# --- Request/Response Models ---

class DocumentInput(BaseModel):
    text: str = Field(..., min_length=1, description="The document text to index")
    title: str = Field(default="Untitled", description="Document title for reference")


class QuestionInput(BaseModel):
    question: str = Field(..., min_length=1, description="The question to ask")
    thread_id: str | None = Field(default=None, description="Thread ID for conversation continuity")


class TaskAcceptedResponse(BaseModel):
    task_id: str
    status: str
    message: str


class TaskStatusResponse(BaseModel):
    task_id: str
    status: str
    doc_id: str | None = None
    chunks_added: int | None = None
    error: str | None = None


class AnswerResponse(BaseModel):
    answer: str
    question: str
    thread_id: str


class HealthResponse(BaseModel):
    status: str
    llm_model: str
    embedding_model: str
    documents_indexed: int


# --- Endpoints ---

@app.get("/health", response_model=HealthResponse)
async def health_check():
    """Check API health and Ollama connectivity."""
    return HealthResponse(
        status="healthy",
        llm_model=LLM_MODEL,
        embedding_model=EMBEDDING_MODEL,
        documents_indexed=collection.count(),
    )


@app.post("/documents", response_model=TaskAcceptedResponse, status_code=202)
async def upload_document(doc: DocumentInput, background_tasks: BackgroundTasks):
    """Upload a document for async processing. Returns 202 with a task_id."""
    task_id = str(uuid.uuid4())
    task_store[task_id] = {
        "status": "queued",
        "title": doc.title,
        "created_at": __import__("datetime").datetime.utcnow().isoformat(),
    }

    background_tasks.add_task(process_document_task, task_id, doc.text, doc.title)

    return TaskAcceptedResponse(
        task_id=task_id,
        status="queued",
        message=f"Document '{doc.title}' queued for processing. Poll GET /tasks/{task_id} for status.",
    )


@app.get("/tasks/{task_id}", response_model=TaskStatusResponse)
async def get_task_status(task_id: str):
    """Poll the status of a document processing task."""
    if task_id not in task_store:
        raise HTTPException(status_code=404, detail="Task not found.")

    task = task_store[task_id]
    return TaskStatusResponse(
        task_id=task_id,
        status=task["status"],
        doc_id=task.get("doc_id"),
        chunks_added=task.get("chunks_added"),
        error=task.get("error"),
    )


@app.post("/ask", response_model=AnswerResponse)
async def ask_question(q: QuestionInput):
    """Ask a question. Provide thread_id for conversation continuity."""
    if collection.count() == 0:
        raise HTTPException(
            status_code=400,
            detail="No documents indexed yet. Upload documents first via POST /documents.",
        )

    thread_id = q.thread_id or str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id}}

    result = await app.state.rag_graph.ainvoke(
        {"question": q.question, "documents": [], "generation": "", "retry_count": 0},
        config,
    )

    return AnswerResponse(
        answer=result["generation"],
        question=q.question,
        thread_id=thread_id,
    )
Write integration tests and verify via Swagger UI

Automated tests verify that your endpoints behave correctly without manual curl commands. You will use httpx with pytest-asyncio to exercise the actual async code paths.

  • Create a new file called tests/test_api.py and add the following code:
"""Integration tests for the async FastAPI application."""

import pytest
import httpx
from app.main import app


@pytest.fixture
async def client():
    async with httpx.AsyncClient(app=app, base_url="http://test") as ac:
        yield ac


class TestHealthEndpoint:
    @pytest.mark.asyncio
    async def test_health_returns_200(self, client):
        response = await client.get("/health")
        assert response.status_code == 200
        data = response.json()
        assert data["status"] == "healthy"


class TestDocumentEndpoint:
    @pytest.mark.asyncio
    async def test_upload_returns_202_with_task_id(self, client):
        response = await client.post(
            "/documents",
            json={"text": "Python is a programming language. " * 50, "title": "Python"},
        )
        assert response.status_code == 202
        data = response.json()
        assert "task_id" in data
        assert data["status"] == "queued"


class TestAskEndpoint:
    @pytest.mark.asyncio
    async def test_ask_returns_thread_id(self, client):
        # Upload and wait for processing first
        response = await client.post(
            "/documents",
            json={"text": "LangGraph is a stateful agent framework. " * 50, "title": "LG"},
        )
        task_id = response.json()["task_id"]

        # Poll until complete (simplified for test)
        import asyncio
        for _ in range(30):
            status = await client.get(f"/tasks/{task_id}")
            if status.json()["status"] == "completed":
                break
            await asyncio.sleep(0.5)

        # Ask a question
        response = await client.post("/ask", json={"question": "What is LangGraph?"})
        assert response.status_code == 200
        data = response.json()
        assert "thread_id" in data
        assert len(data["answer"]) > 0

What does this code do?

  • The client fixture creates an async HTTP client that sends requests directly to your FastAPI app without needing a running server.
  • TestHealthEndpoint verifies the health check returns 200 with a "healthy" status.
  • TestDocumentEndpoint verifies document uploads return 202 with a task_id.
  • TestAskEndpoint runs the full flow: upload a document, poll until processing completes, then ask a question and verify the response includes an answer and thread_id.
  • Save tests/test_api.py.
  • Stop your dev server (Ctrl+C) and run the tests by executing:
pytest tests/test_api.py -v

You should see all three tests pass with green output.

Tests failing with connection errors?

  • Make sure Ollama is still running. The tests call add_document and ainvoke which both connect to the local Ollama instance.
  • If test_ask_returns_thread_id times out, the embedding or LLM call may be slow on first run while models load. Increase the loop range or wait time.
  • Check that you are running pytest from the doc-qa-api project root so Python can resolve the app package.

Help me debug my failing API tests.

Now test the API interactively using the auto-generated documentation.

  • Start the server again by running:
fastapi dev app/main.py
  • Open your browser and navigate to http://localhost:8000/docs.

You should see the Swagger UI with all four endpoints listed: /health, /documents, /tasks/{task_id}, and /ask. You can expand any endpoint and click Try it out to send test requests directly from the browser.

Why use Swagger UI?

FastAPI automatically generates interactive API documentation from your Pydantic models and endpoint type hints. This gives you a ready-made testing interface without writing any frontend code. It also serves as living documentation for anyone consuming your API.

Your API is fully functional. Users can upload documents, poll their processing status, and ask questions with conversation memory. Next up, you will add streaming responses that deliver answers token-by-token.

Secret mission

Add Streaming Responses with Server-Sent Events

Your /ask endpoint works, but it makes the user wait for the entire RAG pipeline to finish before showing any output. Real-world chat UIs stream tokens as they are generated. Can you add a streaming endpoint that delivers the answer in real time?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs or subscription charges. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The doc-qa-api/ project directory on your Desktop (all Python source files, tests, and the data/ folder with ChromaDB and SQLite checkpoint databases).
  • A Python virtual environment inside your project directory.
  • The nomic-embed-text model pulled into Ollama during this project.

Keep everything running

No action needed. Choose this if you plan to keep experimenting with your Document Q&A API or want to add more features.

  • Your FastAPI server will remain available whenever you run it again.
  • The nomic-embed-text model stays in Ollama alongside your existing DeepSeek V4 model.
  • ChromaDB data persists in data/chroma_db/ and checkpoint history persists in data/checkpoints.db.

Pause - I'll come back to this later

Shut down any running processes to free up memory. Your files and models stay on disk so you can resume later.

  • If your FastAPI dev server is still running, press Ctrl+C in the terminal to stop it.
  • Deactivate your virtual environment by running:
deactivate
  • Everything else stays on disk. When you return, cd back into doc-qa-api/, reactivate your venv, and start the server again.

Delete - I don't want to use this again

Remove all project resources and start fresh. This frees up disk space used by the project files, vector store data, and the embedding model.

  • If your FastAPI dev server is still running, press Ctrl+C in the terminal to stop it.
  • Deactivate your virtual environment by running:
deactivate
  • Delete the entire project directory by running:
cd ~/Desktop
rm -rf doc-qa-api/
  • Remove the nomic-embed-text model from Ollama by running:
ollama rm nomic-embed-text

What about DeepSeek V4?

DeepSeek V4 was already installed on your machine before this project. Only remove it if you no longer need it for other work.

Nice Work!

Nice Work!

You just built a production-style, fully async Document Q&A API that runs entirely on your local machine. No cloud dependencies, no API bills, and full conversation memory across threads.

Here is what you accomplished:

  • Built a fully async FastAPI server with background document processing (202 Accepted pattern) and thread-aware conversations powered by LangGraph and AsyncSqliteSaver.
  • Designed a LangGraph StateGraph with async nodes, conditional routing, and a retry loop that rewrites queries when no relevant documents are found.
  • Created a ChromaDB vector store with local nomic-embed-text embeddings and DeepSeek V4 for generation, all running via Ollama on your own hardware.
  • Added streaming token-by-token responses using .astream() with Server-Sent Events.

Ready to quiz yourself?