How to build a RAG API service with LangChain and FastAPI

A RAG script becomes easier to reuse when another application can call it through HTTP. FastAPI can wrap a LangChain retrieval flow behind a JSON endpoint, so web apps, workers, or automation jobs do not need to import the Python module directly.

The sample service keeps a small in-memory index and uses LangChain runnables to connect request text, retrieval, and response assembly. Pydantic validates the incoming JSON before the chain runs, while the sources field in the response shows which indexed documents were used.

Use this local shape to prove the API boundary before attaching a provider chat model or persistent vector database. Retrieved documents remain untrusted input, so production services should keep model instructions separate from retrieved context and avoid returning secrets, raw prompts, or internal trace details to callers.

1. Install LangChain, FastAPI, Uvicorn, and the numeric dependency used by the local vector store.

   $ python3 -m pip install --upgrade langchain fastapi "uvicorn[standard]" numpy

   InMemoryVectorStore similarity search uses numpy in this local smoke test. Use a virtual environment for project work so API packages do not change the system Python install.
   Related: How to create a virtual environment for LangChain
   Related: How to install LangChain with pip

2. Create app.py with the FastAPI routes and LangChain retrieval chain.

   app.py

   from fastapi import FastAPI, HTTPException
   from langchain_core.documents import Document
   from langchain_core.embeddings import Embeddings
   from langchain_core.runnables import RunnableLambda, RunnablePassthrough
   from langchain_core.vectorstores import InMemoryVectorStore
   from pydantic import BaseModel, Field
    
    
   class KeywordEmbeddings(Embeddings):
       terms = [
           "fastapi",
           "rag",
           "retrieval",
           "context",
           "answer",
           "endpoint",
           "validation",
           "langchain",
       ]
    
       def _embed(self, text: str) -> list[float]:
           lowered = text.lower()
           return [float(lowered.count(term)) for term in self.terms]
    
       def embed_documents(self, texts: list[str]) -> list[list[float]]:
           return [self._embed(text) for text in texts]
    
       def embed_query(self, text: str) -> list[float]:
           return self._embed(text)
    
    
   documents = [
       Document(
           page_content=(
               "FastAPI exposes the RAG workflow through a POST endpoint that "
               "accepts a question and returns JSON to the calling application."
           ),
           metadata={"source": "service-design"},
       ),
       Document(
           page_content=(
               "LangChain retrieves context from the vector store before the "
               "answer is assembled, so the response can include source labels."
           ),
           metadata={"source": "retrieval-flow"},
       ),
       Document(
           page_content=(
               "Pydantic request models validate incoming JSON before the RAG "
               "chain runs."
           ),
           metadata={"source": "request-validation"},
       ),
   ]
    
   vector_store = InMemoryVectorStore.from_documents(documents, KeywordEmbeddings())
   retriever = vector_store.as_retriever(search_kwargs={"k": 2})
    
    
   def format_docs(docs: list[Document]) -> str:
       return "\n".join(
           f"{doc.metadata['source']}: {doc.page_content}" for doc in docs
       )
    
    
   def answer_from_context(payload: dict) -> dict:
       context = payload["context"]
       sources = [
           line.split(":", 1)[0]
           for line in context.splitlines()
           if ":" in line
       ]
       return {
           "answer": (
               "LangChain retrieved "
               f"{', '.join(sources)} and FastAPI returned the answer as JSON "
               "for the calling application."
           ),
           "sources": sources,
       }
    
    
   rag_chain = (
       {
           "context": retriever | RunnableLambda(format_docs),
           "question": RunnablePassthrough(),
       }
       | RunnableLambda(answer_from_context)
   )
    
    
   class AskRequest(BaseModel):
       question: str = Field(min_length=3)
    
    
   class AskResponse(BaseModel):
       answer: str
       sources: list[str]
    
    
   app = FastAPI(title="LangChain RAG API")
    
    
   @app.get("/health")
   def health() -> dict:
       return {"status": "ok", "documents": len(documents)}
    
    
   @app.post("/ask", response_model=AskResponse)
   def ask(payload: AskRequest) -> dict:
       question = payload.question.strip()
       if not question:
           raise HTTPException(status_code=400, detail="Question is required")
       return rag_chain.invoke(question)

   The deterministic embedding class and answer builder keep the smoke test local. Replace them with the embedding model, vector store, and chat model used by the production service after the HTTP contract is proven.

3. Start the development server with Uvicorn.

   $ python3 -m uvicorn app:app --host 127.0.0.1 --port 8000
   INFO:     Started server process [49]
   INFO:     Waiting for application startup.
   INFO:     Application startup complete.
   INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

   The 127.0.0.1 host keeps the development service local to the machine. Bind to a wider interface only behind the deployment controls used for that environment.

4. Check the service health route.

   $ curl -s http://127.0.0.1:8000/health
   {"status":"ok","documents":3}

5. Send a question to the RAG endpoint.

   $ curl -s -X POST http://127.0.0.1:8000/ask \
     -H 'Content-Type: application/json' \
     -d '{"question":"How does LangChain retrieve context for the FastAPI RAG endpoint?"}'
   {"answer":"LangChain retrieved service-design, retrieval-flow and FastAPI returned the answer as JSON for the calling application.","sources":["service-design","retrieval-flow"]}

   The sources array should name the retrieved documents that shaped the answer. If the sources are wrong, test the retriever before changing the API route.

6. Stop the development server after the smoke test.

   CTRL-C
   INFO:     Shutting down
   INFO:     Waiting for application shutdown.
   INFO:     Application shutdown complete.
   INFO:     Finished server process [49]