Serving a RAG App with FastAPI

A RAG chain that runs in a notebook is not yet a product. To let a website, a mobile app or another service use it, you must put it behind an HTTP API. FastAPI is a popular Python web framework for this job. It validates input, generates documentation automatically, and supports async code, which suits slow network calls to models and vector stores.

In this lesson you will wrap a LangChain RAG chain in a FastAPI service, load it once at startup, return answers with their sources, stream tokens, and test everything offline with fake models.

Overview / How it works

A web service handles many requests at the same time. A RAG request does three slow things: it embeds the question, searches a vector store, and calls an LLM. All three wait on the network or on a model. If your code blocks while waiting, other users wait too. FastAPI runs on ASGI (Asynchronous Server Gateway Interface), a standard that lets one process handle many waiting requests using Python’s asyncio. A server such as Uvicorn runs the ASGI app.

Every LCEL runnable (a chain built with the | operator) offers both invoke and ainvoke, plus stream and astream. In an async def endpoint you should call the async versions, so the event loop stays free while the chain waits.

The most important design rule is this: build the expensive objects once, not per request. Loading a vector store, creating an embedding client or connecting to a database can take seconds. FastAPI’s lifespan function runs once when the server starts and once when it stops. You build the chain there and keep it in app.state. Every request then reuses it.

The request path

1. The client sends JSON, for example {"question": "What runs FastAPI?"}.
2. FastAPI parses it into a Pydantic model. Invalid input gets an automatic 422 response, and your code never runs.
3. Your endpoint calls await chain.ainvoke(question).
4. The chain retrieves documents, builds a prompt, and calls the LLM.
5. You convert the result into a response model and FastAPI sends it as JSON.

Syntax / Key concepts

app = FastAPI(lifespan=lifespan)

@app.post('/ask', response_model=AskResponse)
async def ask(body: AskRequest, request: Request):
    result = await request.app.state.chain.ainvoke(body.question)
    return ...

Examples

Example 1: a complete RAG endpoint (offline)

This service needs no API key. It uses DeterministicFakeEmbedding (turns text into repeatable pseudo-random vectors), InMemoryVectorStore, and a small function in place of a real LLM. Because the fake embeddings carry no meaning, the retrieved documents are not truly relevant. The goal is to learn the plumbing. To go live, replace the fake parts with a real embeddings class and chat model from a partner package, reading the key from an environment variable.

from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException, Request
from langchain_core.documents import Document
from langchain_core.embeddings import DeterministicFakeEmbedding
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
from langchain_core.vectorstores import InMemoryVectorStore
from pydantic import BaseModel, Field

TEXTS = [
    'FastAPI is a Python web framework built on Starlette and Pydantic.',
    'Uvicorn is an ASGI server that runs FastAPI applications.',
    'A retriever returns the documents most similar to a query.',
]

PROMPT = ChatPromptTemplate.from_template(
    '''Answer using only the context.

Context:
{context}

Question: {question}'''
)


def fake_llm(prompt_value):
    text = prompt_value.to_string()
    context = text.split('Context:', 1)[1].split('Question:', 1)[0]
    return 'Offline demo answer. Context used: ' + ' '.join(context.split())


def build_chain():
    store = InMemoryVectorStore(DeterministicFakeEmbedding(size=64))
    store.add_documents(
        [Document(page_content=t, metadata={'id': i}) for i, t in enumerate(TEXTS)]
    )
    retriever = store.as_retriever(search_kwargs={'k': 2})
    to_prompt = RunnableLambda(
        lambda x: {
            'context': ' '.join(d.page_content for d in x['docs']),
            'question': x['question'],
        }
    )
    return {'docs': retriever, 'question': RunnablePassthrough()} | RunnablePassthrough.assign(
        answer=to_prompt | PROMPT | RunnableLambda(fake_llm)
    )


@asynccontextmanager
async def lifespan(app):
    app.state.chain = build_chain()
    yield


app = FastAPI(title='RAG API', lifespan=lifespan)


class AskRequest(BaseModel):
    question: str = Field(min_length=3, max_length=500)


class AskResponse(BaseModel):
    question: str
    answer: str
    sources: list[str]


@app.get('/health')
async def health():
    return {'status': 'ok'}


@app.post('/ask', response_model=AskResponse)
async def ask(body: AskRequest, request: Request):
    try:
        result = await request.app.state.chain.ainvoke(body.question)
    except Exception:
        raise HTTPException(status_code=502, detail='The RAG pipeline failed.')
    return AskResponse(
        question=result['question'],
        answer=result['answer'],
        sources=[d.page_content for d in result['docs']],
    )

Save this as app.py, install the packages and start the server:

pip install fastapi uvicorn langchain-core
uvicorn app:app --port 8000
curl -s -X POST http://127.0.0.1:8000/ask \
  -H 'Content-Type: application/json' \
  -d '{"question": "What runs FastAPI?"}'

Output (illustrative; the chosen sources vary because the fake embeddings are not meaningful):

{
  "question": "What runs FastAPI?",
  "answer": "Offline demo answer. Context used: Uvicorn is an ASGI server that runs FastAPI applications. A retriever returns the documents most similar to a query.",
  "sources": [
    "Uvicorn is an ASGI server that runs FastAPI applications.",
    "A retriever returns the documents most similar to a query."
  ]
}

Notice three things. The chain is built in lifespan, so it is created once. The chain returns a dictionary holding both docs and answer, because users trust answers more when they can see the sources. And RunnablePassthrough.assign adds the new answer key while keeping the existing keys. FastAPI also gives you interactive documentation at /docs for free.

Example 2: streaming tokens to the client

LLMs are slow, so users like to see words appear as they are generated. Server-Sent Events (SSE) is a simple format for this: each message is a line starting with data: followed by a blank line. The code below uses GenericFakeChatModel, a test model that streams a prepared reply word by word.

import json

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage
from langchain_core.output_parsers import StrOutputParser

app = FastAPI()


def make_chain():
    model = GenericFakeChatModel(
        messages=iter([AIMessage(content='RAG grounds answers in your documents.')])
    )
    return model | StrOutputParser()


@app.get('/stream')
async def stream(q: str):
    chain = make_chain()

    async def events():
        async for token in chain.astream(q):
            yield 'data: ' + json.dumps(token) + '\n\n'
        yield 'data: [DONE]\n\n'

    return StreamingResponse(events(), media_type='text/event-stream')

Output (shape of the stream; chunk boundaries depend on the model):

data: "RAG"

data: " "

data: "grounds"

data: " "
...
data: [DONE]

The endpoint returns immediately with a generator. FastAPI sends each yielded string as soon as it exists. We create the fake model inside the endpoint only because its message iterator can be used once; a real chat model is safe to share. Sending a final [DONE] marker lets the client know the answer is complete.

Example 3: testing the API without a server

FastAPI ships a TestClient that calls your app in-process. Use it as a context manager so the lifespan runs and the chain is built. (It needs the httpx package: pip install httpx.)

from fastapi.testclient import TestClient

from app import app

with TestClient(app) as client:
    ok = client.post('/ask', json={'question': 'What runs FastAPI?'})
    print(ok.status_code, sorted(ok.json()))
    print(len(ok.json()['sources']))
    bad = client.post('/ask', json={'question': 'hi'})
    print(bad.status_code)

200 ['answer', 'question', 'sources']
2
422

The short question hi has fewer than three characters, so Pydantic rejects it with 422 before the chain is ever called. Because the chain is fake and deterministic in shape, these tests are fast and free, which makes them ideal for a CI pipeline.

Under the hood / Step by step

1. uvicorn app:app imports the module and creates the ASGI application.
2. On startup, Uvicorn triggers the lifespan. build_chain() embeds the documents and fills the store, then stores the chain on app.state.
3. A request arrives. FastAPI matches /ask, parses the JSON, and validates it against AskRequest.
4. ainvoke runs the chain’s steps. The retriever and the prompt run first; the model call is the long wait, during which the event loop serves other requests.
5. Plain Python functions wrapped in RunnableLambda are synchronous. LangChain runs them in a worker thread during ainvoke, so they do not freeze the loop, but heavy CPU work will still slow the process.
6. The returned AskResponse is serialised to JSON. On shutdown the code after yield in the lifespan runs, which is where you close connections.

Common Mistakes

1. Building the chain on every request

@app.post('/ask')
async def ask(body: AskRequest):
    chain = build_chain()  # re-embeds all documents each time!
    return await chain.ainvoke(body.question)

This is wrong because loading the store and embeddings repeats on every call, making each request slow and expensive. Build once in lifespan and read it from request.app.state.chain, as in Example 1.

2. Blocking the event loop

@app.post('/ask')
async def ask(body: AskRequest, request: Request):
    return request.app.state.chain.invoke(body.question)  # blocks

Inside async def, a synchronous invoke holds the whole event loop until the model replies, so every other user waits. Use await chain.ainvoke(...). If you must use a blocking library, declare the endpoint with plain def; FastAPI then runs it in a thread pool.

3. Returning raw exceptions

result = await chain.ainvoke(question)  # no error handling

If the provider times out, the client receives a vague 500 error and your logs may lack context. Catch failures, log the details on the server, and return a short HTTPException message such as the 502 in Example 1. Never send API keys, prompts or stack traces to users.

Best Practices

• Create the chain, vector store and model clients once in lifespan, and close them after yield.
• Use async endpoints with ainvoke and astream; use abatch when you must process many inputs together.
• Define strict Pydantic models with length limits. This protects your budget from huge prompts.
• Return sources (and metadata such as a document id) with every answer so users can verify claims.
• Read API keys and settings from environment variables; never hard-code them or return them.
• Set timeouts and a small retry count on the model client, because providers sometimes fail. Check your provider’s documentation for the exact option names.
• Add authentication and rate limiting (for example in a gateway or middleware) before exposing the API publicly, since each call costs money.
• Expose a cheap /health endpoint for load balancers and container orchestrators.
• Run several Uvicorn worker processes for more throughput, but remember each worker builds its own chain. For large indexes use a shared vector database instead of an in-memory store.
• Log the question length, latency and retrieved document ids. Use tracing tools such as LangSmith to inspect failures, and avoid logging private user data.
• Write tests with TestClient and fake models so they run without network access.

Practice Exercises

1. Add a k field to AskRequest (integer, default 2, allowed range 1 to 5) and make the endpoint use it. Hint: build the retriever per request with store.as_retriever(search_kwargs={'k': k}), or pass the setting through the chain’s config.
2. Extend Example 3 with a test that checks GET /health returns {'status': 'ok'}, and a test that sends a 600-character question and expects 422.
3. Combine Examples 1 and 2: add a /ask-stream endpoint that first sends the sources as one SSE event, then streams the answer tokens. Hint: retrieve first, then stream a prompt-plus-model chain.

Summary

• FastAPI turns a LangChain chain into an HTTP service with validation and automatic documentation.
• Build heavy objects once in the lifespan function and store them on app.state.
• Use ainvoke and astream in async endpoints so the event loop is never blocked.
• Pydantic request and response models protect your service and your budget.
• Stream with StreamingResponse and SSE for a responsive user experience.
• Handle errors with HTTPException, and test offline with TestClient and fake models.