Embedding a LangChain Agent in FastAPI: Tools, Manual Search, Streaming

Once an application grows past a handful of screens, its documentation starts to sprawl, and users stop reading it. A 20-page manual that explains features, configuration and domain rules is valuable, but only if people can get an answer out of it without hunting. An assistant built into the product can close that gap: it answers "how does this work?" from the manual and "what is in my project?" from the application's own data.

This guide walks through a compact, working version of such an assistant. You will wire a LangChain agent into a FastAPI service, give it one tool that reads application data and a second tool that searches a PDF manual stored in ChromaDB, pass the authenticated user into the agent at request time, keep conversation history with a LangGraph checkpointer, and stream the answer back to the client. Along the way we point out the gaps in the minimal code that you need to close before it runs cleanly, and what to change before it goes to production.

The scenario and the moving parts

Picture a team that builds a tool for designing photovoltaic installations. The product began small, then accumulated panels, inverters, production estimates, rooftop layouts and a long list of design rules. Its manual now runs to more than 20 pages. Two kinds of questions keep coming up:

• Questions about the product itself: what a feature does, where a setting lives, which rule applies. The answers are in the documentation.
• Questions about the user's own work: which inverter they picked, how much power their system produces, which roofs they have drawn. The answers live in the application's database, and no model knows them by default.

Retrieval-Augmented Generation (RAG) handles the first kind: index the manual, look up the relevant passages when a question arrives, and hand them to the model as context. Tools handle the second kind: small functions the model may call to fetch data from the application. Combine the two and you get an assistant that can both explain the product and reason about a specific project.

The real system behind this scenario has many more tools and much more domain logic. What follows is deliberately reduced to the core so the architecture stays visible:

• FastAPI exposes the HTTP API and resolves who the caller is.
• LangChain's agent (running on LangGraph) owns the reasoning loop and the conversation state.
• An LLM interprets each request and decides whether it needs outside information.
• Tools give the agent controlled access to application functionality.
• RAG lets the agent search the documentation.
• ChromaDB stores the manual's chunks and performs vector search.
• Streaming delivers tokens to the client while the answer is still being generated.

The value of this shape is that the assistant lives inside an existing full-stack application with real users and real data, rather than sitting next to it as a generic chatbot.

Project layout

Each concern gets its own package: HTTP routing, authentication, agent logic, tools and the RAG pipeline. That keeps every file small and makes it obvious where a new capability belongs.

project/
│
├── main.py
├── .env
├── .gitignore
│
├── auth/
│   ├── __init__.py
│   └── dependencies.py
│
├── routers/
│   ├── __init__.py
│   └── chat.py
│
├── llm/
│   ├── __init__.py
│   ├── agent.py
│   ├── context.py
│   ├── orchestrator.py
│   ├── prompts.py
│   ├── provider.py
│   │
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── demo_tool.py
│   │   └── manual_tool.py
│   │
│   └── rag/
│       ├── __init__.py
│       ├── config.py
│       ├── context.py
│       │
│       ├── ingestion/
│       │   ├── __init__.py
│       │   ├── loader.py
│       │   ├── chunker.py
│       │   ├── chroma.py
│       │   └── indexer.py
│       │
│       └── retrieval/
│           ├── __init__.py
│           └── retriever.py
│
├── scripts/
│   ├── __init__.py
│   └── index_manual.py
│
├── docs/
│   └── manual.pdf
│
└── chroma_data/ (*generated locally, not commited or deployed)

What each part is for:

• main.py builds the FastAPI app.
• auth/ holds the mocked authentication dependency.
• routers/ holds the HTTP endpoints.
• llm/ contains everything that concerns the agent.
• llm/tools/ holds the functions the agent may call.
• llm/rag/ holds the retrieval pipeline, split into ingestion/ (load, chunk and index the PDF) and retrieval/ (query ChromaDB).
• scripts/ holds commands you run by hand, such as indexing the manual.
• docs/ holds the source PDF.
• chroma_data/ is generated locally and should never be committed or deployed.

You will not create all of these at once. The build order is: the API layer, then the model, then tools, then the RAG pipeline, then context, streaming and history.

Step 1: A FastAPI skeleton with a mocked user

Start by installing everything the project will need. The list covers the web server, LangChain and LangGraph, the OpenAI-compatible chat client, ChromaDB, the PDF loader and text splitters, and python-dotenv for configuration.

pip install fastapi uvicorn langchain langgraph langchain-openai chromadb langchain-community langchain-text-splitters pypdf python-dotenv

The model will be reached through OpenRouter, so create a .env file in the project root that holds the API key.

OPENROUTER_API_KEY=your_api_key_here

Add .env to .gitignore right away. A key that lands in version control once should be considered leaked.

The application entry point

main.py stays tiny. It creates the app and registers the chat router; nothing about the model or the agent belongs here.

from fastapi import FastAPI
from routers.chat import chat_router

app = FastAPI(
    title="AI Agent Demo",
)
app.include_router(chat_router)

A first chat endpoint

In routers/chat.py, define a router under the /chat prefix with a single POST route. For now it only echoes the incoming message, which is enough to confirm the plumbing works before any AI is involved.

from fastapi import APIRouter

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
):
    return {
        "message": message,
    }

Note that message: str on a POST route without a body model makes FastAPI read it from the query string. That is convenient for trying things in Swagger UI, but for a real client you would normally accept a JSON body defined with a Pydantic model, because query strings end up in access logs and have practical length limits.

A mock authentication dependency

A production app would verify a session cookie or a JWT and load the user from the database. Here, a custom header stands in for all of that. Create auth/dependencies.py with a small MockUser dataclass and a get_current_user function that reads the X-Demo-User header and rejects the request with a 401 when it is missing.

from dataclasses import dataclass
from fastapi import Header, HTTPException

@dataclass
class MockUser:
    id: str
    name: str

def get_current_user(
    x_demo_user: str | None = Header(default=None),
) -> MockUser:
    if x_demo_user is None:
        raise HTTPException(
            status_code=401,
            detail="Missing X-Demo-User header",
        )
    return MockUser(
        id=x_demo_user,
        name=x_demo_user,
    )

FastAPI's dependency injection now hands the user to the endpoint. Declaring a parameter with Depends(get_current_user) is all it takes.

from fastapi import APIRouter, Depends
from auth.dependencies import (
    MockUser,
    get_current_user,
)

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    return {
        "user": current_user.name,
        "message": message,
    }

A client identifies itself by sending a header like this one:

X-Demo-User: user-123

On every request, FastAPI runs get_current_user() first and passes the resulting MockUser into ask_ai. The important design point is the division of labor: FastAPI owns authentication, and the AI layer simply receives a user object it can trust. Later, that user object is what lets tools return data belonging to the right person. Swapping the mock for real authentication later only changes this one dependency.

Step 2: Connecting a model through OpenRouter

With a working endpoint and a known caller, the service needs a model. OpenRouter exposes an OpenAI-compatible API, so LangChain's ChatOpenAI class works against it once you point base_url at OpenRouter and pass your OpenRouter key.

Put this in llm/provider.py. It loads .env, fails fast with a clear error if the key is absent, and wraps the key in Pydantic's SecretStr so it is not printed by accident in logs or reprs.

import os
from dotenv import load_dotenv
from pydantic import SecretStr
from langchain_openai import ChatOpenAI

load_dotenv()

api_key = os.getenv(
    "OPENROUTER_API_KEY",
)
if not api_key:
    raise RuntimeError(
        "OPENROUTER_API_KEY environment variable is not set."
    )

model = ChatOpenAI(
    model="YOUR_MODEL",
    api_key=SecretStr(api_key),
    base_url="https://openrouter.ai/api/v1",
)

Because the key comes from the environment, it never appears in source code. At this stage you could already send prompts to the model and get replies, but that is a plain LLM call. The goal is an agent that can decide on its own when it needs a tool.

Picking a model

The model argument is just an OpenRouter model identifier, so you can swap models without touching the rest of the code. When comparing candidates, check:

• tool-calling support, which the agent depends on;
• streaming support;
• the size of the context window;
• rate limits;
• whether a free tier exists.

OpenRouter offers some models free of charge, which is handy while you experiment. The catalogue changes regularly, so browse the current list and filter for free models rather than relying on a fixed recommendation. Whatever you choose goes straight into the constructor:

model = ChatOpenAI(
    model="YOUR_MODEL_ID",
    api_key=SecretStr(api_key),
    base_url="https://openrouter.ai/api/v1",
)

For instance, if the catalogue lists an identifier such as the one below (an example taken at the time of writing; it may no longer be available), you would pass that exact string as model:

google/gemma-4-26b-a4b-it:free

Keep in mind that free models are shared capacity. They get rate-limited or become unavailable, often at the worst moment during a demo. Switching to another model, or attaching your own provider key through OpenRouter, usually fixes that. For production, choose on reliability, capability, latency and cost, not on price alone.

Step 3: From a model to an agent

A direct model call is a single hop: the user's text goes in, a completion comes out. An agent inserts a decision loop. The model looks at the request, decides whether it can answer immediately or needs something first, calls a tool if necessary, reads the result, and only then produces the final answer. Roughly:

• plain call: user, then LLM, then response;
• agent: user, then agent, then the LLM decides what is needed, then a tool call or retrieval if required, then the response.

The agent module

In llm/agent.py, create_agent from LangChain builds the agent from a model and a system prompt. Under the hood it produces a LangGraph graph that runs the model-and-tools loop for you.

from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT

agent = create_agent(
    model=model,
    system_prompt=SYSTEM_PROMPT,
)

This agent has no tools yet, so it behaves much like the bare model. First, give it instructions.

The system prompt

llm/prompts.py holds a short prompt that tells the model what it is for, forbids inventing data, and says which kind of question maps to which kind of lookup.

SYSTEM_PROMPT = """
You are an AI assistant for our demo application.
You help users understand the application and navigate the system.
Never invent data.
When information about the demo system
is required, use the available application tools.
When answering questions about the application,
use the documentation search tool.
Always answer in clear, conversational language.
""".strip()

The prompt sets up two sources of truth:

• application data (what is in the user's account) comes from application tools;
• application knowledge (how the product works) comes from the documentation search.

One clarification before going further. Tools and RAG are introduced separately below because that makes each easier to understand, but in the finished agent the documentation search is itself a tool. There is no second mechanism: the agent sees a list of callable functions, and searching the manual is one of them.

Step 4: The first tool

A tool is a function the agent is permitted to call. This is what makes the architecture scale: rather than stuffing every piece of application data into the prompt, you expose specific operations and let the model request them only when a question needs them.

For the demo, llm/tools/demo_tool.py defines a tool that returns a fixed block of project information.

from langchain.tools import tool

@tool
def get_my_demo_data() -> str:
    """
    Return information about the demonstration data.
    This is just for demo data. But in production, make a more detailed instruction.
    """

    return """
    Project: Aperture Analytics Dashboard
    Owner: Jordan Lee
    Status: In Progress
    Team size: 6
    Budget: $84,000
    Deadline: 2026-11-15
    Description: An internal dashboard for visualizing customer usage
    metrics, built with FastAPI and React, integrating with the
    company's data warehouse.
    """.strip()

Two details matter here. The @tool decorator turns an ordinary Python function into a LangChain tool, deriving its name and argument schema from the signature. And the docstring becomes the tool's description, which is what the model reads when deciding whether to call it. In a real system that description deserves real care: say precisely what the tool returns, when it is appropriate, and when it is not. A vague description is one of the most common reasons an agent calls the wrong tool or none at all.

Register the tool by passing it to create_agent:

from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import get_my_demo_data

agent = create_agent(
    model=model,
    tools=[
        get_my_demo_data,
    ],
    system_prompt=SYSTEM_PROMPT,
)

Your code never decides when the function runs. If a user asks "What demo data do I have?", the model recognizes that it needs account-specific information and calls get_my_demo_data(). If the user asks "What is a demo?", no lookup is needed and the model answers directly. That choice happens on every turn, inside the agent loop.

Step 5: Building the RAG pipeline for the manual

The agent can now fetch application data, but it still knows nothing about how the product works. Pasting a 20-page manual into the system prompt would waste tokens on every request and become painful to keep up to date. RAG avoids both problems.

It is worth being precise about what RAG is and is not. Nothing is trained or fine-tuned on the documentation. At query time the system searches the manual for the passages most relevant to the question and supplies those passages to the model as context, and the model answers from them.

The pipeline has two phases:

1. Ingestion, which runs separately from the web app: load the PDF, split it into chunks, and store the chunks with their embeddings in ChromaDB.
2. Retrieval, which runs inside a request: take the question, search ChromaDB, collect the best chunks, and pass them to the model.

If you want a broader view of the concepts, the blog's overview of retrieval that fetches fresh knowledge on demand covers them in more depth; here we stay on the implementation.

Loading the PDF

llm/rag/ingestion/loader.py uses LangChain's PyPDFLoader, which turns each page of the PDF into a Document.

from pathlib import Path
from langchain_community.document_loaders import PyPDFLoader

PDF_PATH = Path("docs/manual.pdf")

def load_manual():
    loader = PyPDFLoader(
        str(PDF_PATH),
    )
    documents = loader.load()
    return documents

A Document carries two things: the extracted text in page_content, and a metadata dictionary describing where that text came from. The metadata is what later allows an answer to cite a page. Conceptually, each loaded page looks like this:

Document
├── page_content
│   └── "To create a new demo data..."
│
└── metadata
    ├── source: docs/manual.pdf
    └── page: 12

PyPDFLoader typically records both a zero-based page index and a human-readable page_label. The context builder below uses page_label, which matches the page numbers readers see in the PDF.

Splitting pages into chunks

Searching whole pages, let alone the whole document as one block, gives coarse results. llm/rag/ingestion/chunker.py splits the documents with RecursiveCharacterTextSplitter.

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
)
from langchain_core.documents import Document

def chunk_documents(
    documents: list[Document],
) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=150,
        separators=[
            "\n\n",
            "\n",
            ". ",
            " ",
            "",
        ],
    )
    return splitter.split_documents(
        documents,
    )

The splitter aims for chunks of about 1,000 characters with 150 characters of overlap. The separator list is tried in order: it prefers to break at paragraph boundaries, then at line breaks, then at sentence ends, then at spaces, and only as a last resort in the middle of a word. The overlap exists because a single fact can straddle a split point; repeating a little text on either side makes it less likely that the relevant sentence ends up cut in half.

These numbers are starting points, not rules. The right size depends on how your documents are written and how precise retrieval needs to be, so treat them as values to tune against real questions. The blog's piece on chunking that preserves evidence goes further into that trade-off.

A persistent Chroma collection

llm/rag/ingestion/chroma.py opens a PersistentClient that stores data on disk and returns the manual's collection, creating it on first use.

import chromadb
from llm.rag.config import (
    CHROMA_PATH,
    MANUAL_COLLECTION_NAME,
)

def get_chroma_client():
    return chromadb.PersistentClient(
        path=CHROMA_PATH,
    )

def get_manual_collection():
    client = get_chroma_client()
    return client.get_or_create_collection(
        name=MANUAL_COLLECTION_NAME,
    )

The paths and names come from llm/rag/config.py, which reads environment variables and falls back to sensible defaults:

import os

CHROMA_PATH = os.getenv(
    "CHROMA_PATH",
    "./chroma_data",
)
MANUAL_PATH = os.getenv(
    "MANUAL_PATH",
    "docs/manual.pdf",
)
MANUAL_COLLECTION_NAME = os.getenv(
    "MANUAL_COLLECTION_NAME",
    "manual",
)

Add the matching entries to .env:

CHROMA_PATH=./chroma_data
MANUAL_PATH=docs/manual.pdf
MANUAL_COLLECTION_NAME=manual

No embedding model is configured anywhere, and that is intentional for the demo. When a collection is created without an explicit embedding function, Chroma uses its built-in default: every time you add documents, Chroma computes their embeddings itself and stores them next to the text and metadata. The default model runs locally and is fetched on first use, so the first indexing run may pause while it downloads.

The result is a local, persistent vector store in the chroma_data/ directory. Because it is derived entirely from the PDF, add it to .gitignore alongside .env.

The indexing job

llm/rag/ingestion/indexer.py ties the ingestion steps together.

from pathlib import Path
from llm.rag.config import MANUAL_PATH
from llm.rag.ingestion.loader import load_manual
from llm.rag.ingestion.chunker import chunk_documents
from llm.rag.ingestion.chroma import get_manual_collection

def index_manual():
    collection = get_manual_collection()
    if collection.count() > 0:
        print(
            f"Manual already indexed "
            f"({collection.count()} chunks)."
        )
        return
    manual_path = Path(
        MANUAL_PATH,
    )
    if not manual_path.exists():
        raise FileNotFoundError(
            f"Manual not found: {manual_path}"
        )
    documents = load_manual()
    print(
        f"Loaded {len(documents)} pages."
    )
    chunks = chunk_documents(
        documents,
    )
    print(
        f"Created {len(chunks)} chunks."
    )
    collection.add(
        ids=[
            f"manual-chunk-{i}"
            for i in range(len(chunks))
        ],
        documents=[
            chunk.page_content
            for chunk in chunks
        ],
        metadatas=[
            chunk.metadata
            for chunk in chunks
        ],
    )
    print(
        f"Stored {len(chunks)} chunks."
    )

Walk through what it does. It opens the collection and returns early if it already holds chunks, which makes repeated runs harmless. It checks that the PDF exists and raises a clear error otherwise. Then it loads the pages, chunks them, and adds everything to Chroma in one call with stable IDs (manual-chunk-0, manual-chunk-1, and so on), the chunk texts, and their metadata. Progress messages report how many pages and chunks were processed.

One pitfall follows from that early return: if you edit the manual and run the script again, nothing happens, because the collection is not empty. To pick up changes you have to delete the collection (or the chroma_data/ directory) before re-indexing, or replace the guard with logic that upserts or rebuilds deliberately.

The listing does not show scripts/index_manual.py; it only needs to import index_manual and call it. Run it once as a module from the project root:

python -m scripts.index_manual

That single run reads the PDF, chunks it, embeds the chunks and stores them. The terminal output reports the counts. In the reduced demo, the manual is a one-page PDF containing a single rule, "The demo data can only be given to admin users", which is enough to show whether retrieval works. From then on, starting FastAPI does not touch the PDF at all, because the vectors are already persisted.

Querying the collection

Indexing on its own gives the agent nothing; it needs a way to search. llm/rag/retrieval/retriever.py wraps Chroma's query API.

from dataclasses import dataclass
from typing import Any
from llm.rag.ingestion.chroma import (
    get_manual_collection,
)

@dataclass
class RetrievedChunk:
    content: str
    metadata: dict[str, Any]
    distance: float

def retrieve_manual(
    query: str,
    n_results: int = 5,
) -> list[RetrievedChunk]:
    collection = get_manual_collection()
    results = collection.query(
        query_texts=[query],
        n_results=n_results,
        include=[
            "documents",
            "metadatas",
            "distances",
        ],
    )
    documents = results["documents"] or []
    metadatas = results["metadatas"] or []
    distances = results["distances"] or []
    retrieved_chunks = []
    for document, metadata, distance in zip(
        documents[0],
        metadatas[0],
        distances[0],
    ):
        retrieved_chunks.append(
            RetrievedChunk(
                content=document,
                metadata=dict(metadata)
                if metadata else {},
                distance=distance,
            )
        )
    return retrieved_chunks

The function sends the question as query_texts, asks for up to five results, and requests the documents, their metadata and their distances. Chroma returns one list per query, which is why the code reads index [0] of each field; the or [] fallbacks guard against missing fields. Each hit is packaged as a RetrievedChunk dataclass so the rest of the code does not depend on Chroma's response shape.

Because the query is embedded with the same model as the stored chunks, matching is semantic. A question like "How do I add new demo data?" finds passages about creating or granting demo data even if they never use the words "add new". The distance tells you how close each hit is; lower means more similar. That value is useful later if you want to discard weak matches instead of always passing five chunks to the model.

With this in place, the retrieval half of RAG works. What remains is handing the results to the model.

Turning chunks into context

llm/rag/context.py formats the hits into one string the model can read.

from llm.rag.retrieval.retriever import (
    RetrievedChunk,
)

def build_context(
    chunks: list[RetrievedChunk],
) -> str:
    context_parts = []
    for chunk in chunks:
        page = chunk.metadata.get(
            "page_label",
        )
        context_parts.append(
            f"Source: User Guide, page {page}\n"
            f"{chunk.content}"
        )
    return "\n\n---\n\n".join(
        context_parts,
    )

Each chunk is prefixed with a source line naming the user guide and its page label, and chunks are separated by a divider. The source line is what lets the model say where an answer came from, and it gives users a way to check.

Step 6: Exposing the manual as a tool

The pipeline is complete: pages are loaded and chunked, chunks live in ChromaDB, relevant ones can be found and formatted. The agent, however, has no idea any of it exists. This is where the earlier architectural note pays off: documentation search becomes just another tool.

llm/tools/manual_tool.py defines search_user_manual, which takes a query, retrieves five chunks, and returns them as formatted context. If nothing comes back, it returns an explicit message saying the manual does not cover the question, which gives the model something honest to relay instead of an empty string.

from langchain.tools import tool
from llm.rag.context import build_context
from llm.rag.retrieval.retriever import retrieve_manual

@tool
def search_user_manual(
    query: str,
) -> str:
    """
    Search the application user manual.
    Use this tool when the user asks about application
    behavior, instructions, rules, limitations, or
    how something works.
    """
    chunks = retrieve_manual(
        query=query,
        n_results=5,
    )
    if not chunks:
        return (
            "The manual does not contain enough "
            "information to answer this question."
        )
    return build_context(
        chunks,
    )

As before, the docstring is the tool's advertisement to the model. It says to use this tool for questions about behavior, instructions, rules, limitations and how things work, which lines up with the system prompt.

Now register both tools with the agent:

from langchain.agents import create_agent
from llm.provider import model
from llm.prompts import SYSTEM_PROMPT
from llm.tools.demo_tool import (
    get_my_solar_system,
)
from llm.tools.manual_tool import (
    search_user_manual,
)

agent = create_agent(
    model=model,
    tools=[
        get_my_solar_system,
        search_user_manual,
    ],
    system_prompt=SYSTEM_PROMPT,
)

Watch the import in that listing: it refers to get_my_solar_system, a name from the full application, while the demo tool module defines get_my_demo_data. Use get_my_demo_data in both the import and the tools list, or the module will fail to import.

Why a tool-based design holds up as the app grows

The agent never needs one giant prompt describing everything the application knows. It gets narrow, controlled capabilities instead. Adding a feature to the assistant means writing a new tool and registering it; the HTTP layer does not change. The demo keeps exactly one data tool and one documentation tool so the pattern is easy to follow, but the same structure carries a much larger toolset in the full application.

Step 7: Passing the authenticated user into the agent

The demo tool still returns hard-coded data. A real tool must know who is asking, and the application already knows that: FastAPI resolved the user in the auth dependency. The missing piece is carrying that user into the agent run. LangChain calls this runtime context.

Define the shape of the context in llm/context.py:

from dataclasses import dataclass
from auth.dependencies import MockUser

@dataclass
class AgentContext:
    user: MockUser

This object is supplied when the agent is invoked, and the distinction matters. The user is request-scoped information. It belongs to the current HTTP request, not to the conversation, and it should never be stored as a message the model can read or rewrite. Keeping it out of the message history also means a prompt cannot talk the agent into acting as a different user. In the full application, the same context object also carries things like the database session and the ID of the project being edited.

The demo code stops at defining the class, so two connections are left for you to make, and it is worth checking the current LangChain documentation for the exact API. First, declare the schema when building the agent, typically with a context_schema=AgentContext argument to create_agent. Second, have tools read it: in LangChain 1.x a tool can accept a runtime parameter (for example annotated as ToolRuntime[AgentContext]) and read the user from its context attribute, which is hidden from the model's view of the tool's arguments. That is where a real get_my_demo_data would look up records for user.id.

Step 8: An orchestration layer that streams

Rather than calling the agent from inside the router, put the interaction in llm/orchestrator.py. The router then stays focused on HTTP, and the orchestrator owns how a message becomes an agent run.

from collections.abc import Iterator
from langchain_core.messages import (
    AIMessage,
    AIMessageChunk,
    BaseMessage,
    ToolMessage,
)
from langchain_core.runnables import RunnableConfig
from llm.agent import agent
from llm.context import AgentContext

def chat_stream(
    user_message: str,
    user,
) -> Iterator[str]:
    config: RunnableConfig = {
        "configurable": {
            "thread_id": f"user:{user.id}",
        }
    }
    context = AgentContext(
        user=user,
    )
    for chunk, metadata in agent.stream(
        {
            "messages": [
                {
                    "role": "user",
                    "content": user_message,
                }
            ]
        },
        config=config,
        context=context,
        stream_mode="messages",
    ):
        if not isinstance(
            chunk,
            BaseMessage,
        ):
            continue
        if isinstance(
            chunk,
            ToolMessage,
        ):
            continue
        if not isinstance(
            chunk,
            (
                AIMessage,
                AIMessageChunk,
            ),
        ):
            continue
        if isinstance(
            chunk.content,
            str,
        ):
            yield chunk.content

There is a lot in this function, so take it piece by piece.

The thread ID selects the conversation

The first block builds the run configuration:

config = {
    "configurable": {
        "thread_id": f"user:{user.id}",
    }
}

A LangGraph checkpointer stores conversation state keyed by thread_id. Every run with the same thread ID continues the same conversation, and its messages can be read back later. Here the thread ID is derived from the user's ID, which means each user has exactly one conversation. That is fine for a demo; a real application would create proper conversation IDs, allow several per user, and verify on every request that the caller owns the thread being accessed.

The listing assumes a checkpointer exists, but none of the agent snippets pass one. Without it, thread_id has no effect and nothing is remembered between requests. Create a single InMemorySaver in llm/agent.py and hand it to create_agent through its checkpointer argument, so both the agent and the history functions below can import the same instance.

Runtime context travels with the run

Next, the orchestrator wraps the user in the context object:

context = AgentContext(
    user=user,
)

That object is passed as context= to agent.stream(). This is the path by which the identity established in FastAPI reaches the agent and, through it, the tools.

Filtering the stream

agent.stream() is called with stream_mode="messages", which yields pairs of a message chunk and metadata as the model produces tokens. Not everything in that stream should reach the user. The loop skips anything that is not a LangChain message, skips ToolMessage objects (raw tool output such as retrieved manual text), keeps only AI messages and AI message chunks, and yields their content when it is a plain string. Content that some providers deliver as a list of parts is silently dropped by that last check, so if you switch models and see empty responses, that is a place to look.

Step 9: Returning a streaming response

Making chat_stream() a generator was deliberate. Waiting for the complete answer before sending a byte leaves the user staring at a spinner, and LLM answers can take several seconds. Streaming shows the first words almost immediately, which makes the assistant feel far more responsive.

FastAPI's StreamingResponse accepts a generator directly. Update routers/chat.py:

from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from auth.dependencies import (
    MockUser,
    get_current_user,
)
from llm.orchestrator import chat_stream

chat_router = APIRouter(
    prefix="/chat",
    tags=["Chat"],
)

@chat_router.post("")
def ask_ai(
    message: str,
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    return StreamingResponse(
        chat_stream(
            user_message=message,
            user=current_user,
        ),
        media_type="text/plain",
    )

The response is sent as text/plain, with each yielded piece written to the connection as soon as it is produced. Because chat_stream is a regular (synchronous) generator, Starlette iterates it in a worker thread, so it does not block the event loop. If you later need structured events on the client (for example, to show "searching the manual..." while a tool runs), Server-Sent Events are a natural next step.

The full request path now reads: the client posts to /chat, FastAPI authenticates the caller, chat_stream() starts an agent run, the model decides whether to call a tool, any tool executes and returns its result, the model writes the answer, and the tokens stream back to the client.

The key property is that FastAPI never runs the model itself. The router speaks HTTP, the orchestrator drives the agent, the agent decides what information it needs, and tools do the fetching. Each layer can change without disturbing the others.

Step 10: Reading conversation history back

Because the agent is checkpointed, its state is stored after every turn. That makes it possible to show a returning user their previous conversation. Two helper functions do the work.

Reading the raw checkpoint

The first function loads the latest checkpoint for a thread and returns the messages channel, or an empty list if the thread has never been used:

def get_conversation_messages(
    thread_id: str,
) -> list[BaseMessage]:
config: RunnableConfig = {
        "configurable": {
            "thread_id": thread_id,
        }
    }
    checkpoint = checkpointer.get(
        config,
    )
    if checkpoint is None:
        return []
    return checkpoint[
        "channel_values"
    ].get(
        "messages",
        [],
    )

If you copy this listing, fix the indentation of the config assignment: it must be indented inside the function body, otherwise Python raises an error. The function also needs BaseMessage, RunnableConfig and the shared checkpointer instance imported.

What comes back is the agent's raw state, and that includes more than the chat a user remembers. When the agent calls a tool, LangGraph records an AI message containing the tool call and a separate tool message containing the result. Those are implementation details that the frontend should not have to interpret.

Shaping messages for display

The second function builds the user-facing view:

def get_conversation_messages_for_display(
    thread_id: str,
) -> list[dict[str, str]]:
    display = []
    for message in get_conversation_messages(
        thread_id,
    ):
        if isinstance(
            message,
            HumanMessage,
        ):
            content = _extract_text_content(
                message.content,
            )
            if content.strip():
                display.append(
                    {
                        "type": "human",
                        "content": content,
                    }
                )
            continue
        if isinstance(
            message,
            AIMessage,
        ):
            content = _extract_text_content(
                message.content,
            )
            if content.strip():
                display.append(
                    {
                        "type": "ai",
                        "content": content,
                    }
                )
    return display

It keeps only HumanMessage and AIMessage objects, extracts their text, drops any that are empty, and returns simple dictionaries with a type and content. Tool messages never appear because they are neither of the two accepted types. Empty AI messages are dropped too, which matters because an AI message that only requests a tool call usually has no text.

The listing relies on a helper, _extract_text_content, that is not shown. Its job is to return the content unchanged when it is a string and, when it is a list of content parts, to join the text parts together. You also need to import HumanMessage and AIMessage.

The history endpoint

Expose the display view through a GET route in routers/chat.py. It derives the same thread ID the chat endpoint uses and returns it together with the messages.

@chat_router.get("/current")
def get_current_conversation(
    current_user: MockUser = Depends(
        get_current_user,
    ),
):
    thread_id = (
        f"user:{current_user.id}"
    )
    messages = (
        get_conversation_messages_for_display(
            thread_id,
        )
    )
    return {
        "thread_id": thread_id,
        "messages": messages,
    }

A chat UI can call this when the page opens and render the existing conversation before the user types anything:

GET /chat/current

The response looks like this:

{
    "thread_id": "user:user-123",
    "messages": [
        {
            "type": "human",
            "content": "How much demo data do I have?"
        },
        {
            "type": "ai",
            "content": "You currently have 18 demo data."
        }
    ]
}

Why not return the raw state

The agent's internal state and the conversation a user sees are different things. As features are added, the state accumulates tool calls, tool results, intermediate steps, model metadata and other bookkeeping. Returning all of it would couple the frontend to agent internals and could leak tool output you never meant to show. The backend should define what the public conversation history is and return only that.

How a single request flows end to end

With all the pieces in place, it is worth being clear about one property: the model never touches your database or your PDF. It can only ask for a tool to be called. The tool, running as ordinary Python code with ordinary access controls, performs the operation and returns text, and the model uses that text to compose its answer. That boundary is what keeps the assistant safe to embed in an application with real data.

Trying it in Swagger UI

FastAPI generates interactive documentation automatically, so there is no need for a separate client while testing. Start the server:

python -m uvicorn main:app --reload

Then open the interactive API docs served under /docs, set the X-Demo-User header, and try three interactions:

• A data question, such as asking what demo data you have. The agent should decide it needs the demo tool, call it, and answer from the returned project details. In the full application, the same kind of tool queries the user's actual records.
• A documentation question, such as asking who can receive demo data. The agent should call the manual search, retrieve the single indexed rule, and answer that only admin users can.
• The history endpoint, GET /chat/current, which should return the human and AI turns from the previous two questions, without any tool messages.

If the first two produce answers grounded in the tool output and the third shows a clean transcript, every layer is working.

Before you take it to production

The demo simplifies several components on purpose. These are the ones to revisit before real users depend on the service.

Durable conversation state

InMemorySaver is perfect for development, but everything it holds vanishes when the process restarts, and it cannot be shared between several API instances behind a load balancer. Use a checkpointer backed by a database or other durable storage so that conversations survive deployments and every instance sees the same state. For what the in-memory saver actually stores and how, see the blog's walkthrough of how InMemorySaver organizes checkpoints, writes and blobs.

A real vector store deployment

A local Chroma directory is fine for demonstrating the pipeline, but it is not production infrastructure. Run Chroma as a persistent service, or move to a managed vector database that suits your stack. Whatever you choose must be persistent, backed up and reachable from every application instance.

Indexing stays out of the API

The demo already gets one important decision right: indexing is a separate script that you invoke on its own, and the API only retrieves.

python -m scripts.index_manual

The server never loads the PDF, chunks it or computes embeddings at startup. Indexing is offline ingestion; retrieval is part of serving a request. Keeping them apart means the API does not need to detect documentation changes or rebuild anything.

In production, take the next step and run the same indexing code as a dedicated ingestion job, triggered from a deployment pipeline, on a schedule, or by a worker whenever new documentation is uploaded. The architecture stays the same; the job just becomes automated, repeatable and independently deployable. The responsibilities then split cleanly:

• Ingestion job: load documents, chunk them, embed them, update the vector store.
• FastAPI service: accept questions, retrieve relevant chunks, generate responses.
• Vector store: persist the indexed representations used at query time.

Remember the early-return guard in the indexer when you automate it; a job that silently skips re-indexing is worse than no job.

An explicit embedding model

Relying on Chroma's default embedding function keeps the demo free of extra configuration, but a production system should choose and configure its embedding model explicitly. That makes results reproducible and gives you control over quality, cost, latency and where embeddings are computed. One rule is non-negotiable: the same embedding model must be used for indexing and for querying. Changing it means re-indexing everything.

Observability and error handling

Once an agent is live, seeing what it did matters as much as making it work. A single request can involve several model calls, one or more tool calls and a retrieval step before the final answer. Logging only that final answer tells you almost nothing when something goes wrong. Instrument the whole flow:

• Tool calls: which tools ran, with what arguments, and how long each took.
• Model latency: the duration of every LLM request.
• Token usage and cost: essential when one user message can trigger multiple model calls.
• Tool failures: tools should return controlled error messages instead of crashing the request.
• Provider failures: handle rate limits, timeouts and unavailable models gracefully, ideally with a fallback model.
• Run tracing: record the ordered sequence of model calls, tool calls and responses for each run.
• Retrieval quality: if the manual search keeps returning irrelevant chunks, the cause is more often chunking, embeddings, the query or the retrieval settings than the LLM.

The goal is that the agent is never a black box. For any run you should be able to say what it did, which tools it called, how long each step took and where it failed. The specific tooling depends on your stack; the principle does not.

Key takeaways

• You do not need a large AI platform to add a useful assistant to a domain-heavy application. Start with the smallest set of pieces that solves a real problem.
• Treat documentation search as one tool among others. The agent then has a single, uniform way to reach both product knowledge and user data.
• Keep identity in runtime context, not in messages. The user belongs to the request, and tools should read it from there.
• Separate HTTP, orchestration, agent and tools. Each layer stays small, and adding a capability means adding a tool.
• Checkpointed state gives you history almost for free, but return a curated view, not the raw agent state.
• Stream responses, index offline, pin the embedding model, and instrument every step before real users arrive.