Why FastAPI Fits This Application Specifically

FastAPI is a Python web framework built around type hints and async support — two properties that map directly onto what Lesson 1's design actually needs. Its request and response models are built on Pydantic, the same library Unit 6's structured outputs use for defining a response schema, so validating what a client sends to this application follows the identical mental model already established for validating what the model returns. And FastAPI's native async support is what makes it possible to use the AsyncOpenAI client from Unit 12, Lesson 6 directly inside a request handler, so one user's document upload or chat request doesn't block the server from handling anyone else's request in the meantime.

Setting Up the Application and Request Models

The application starts with FastAPI's app object and Pydantic models describing exactly what shape each endpoint's request and response take.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import AsyncOpenAI

app = FastAPI()
async_client = AsyncOpenAI()

class UploadDocumentRequest(BaseModel):
    filename: str
    content: str

class UploadDocumentResponse(BaseModel):
    document_id: str
    chunk_count: int

class ChatRequest(BaseModel):
    conversation_id: str
    message: str

class ChatResponse(BaseModel):
    reply: str
    conversation_id: str

Defining these models up front, following Unit 6's discipline around explicit schemas, means FastAPI automatically validates every incoming request against them — a request missing the required message field, or sending the wrong type for conversation_id, is rejected with a clear validation error before the endpoint's own code ever runs, the same reliability benefit structured outputs provide on the model's side of a request, applied here to what arrives from a client.

In-Memory Storage for This Capstone's Scope

Following Lesson 1's explicit scoping decision to keep persistence simple, document chunks, their embeddings, and conversation history are held in memory for this capstone, with Lesson 4 noting where a real deployment would replace this with an actual database.

document_store: dict = {}       # document_id -> list of {"text": ..., "embedding": ...}
conversation_store: dict = {}   # conversation_id -> list of {"role": ..., "content": ...}

Implementing Document Ingestion

The upload endpoint chunks the incoming document text and generates an embedding for each chunk, directly reusing Unit 10's chunking and embedding patterns.

import uuid

def chunk_by_paragraph(text: str) -> list:
    return [p.strip() for p in text.split("\n\n") if p.strip()]

@app.post("/documents/upload", response_model=UploadDocumentResponse)
async def upload_document(request: UploadDocumentRequest):
    chunks = chunk_by_paragraph(request.content)
    if not chunks:
        raise HTTPException(status_code=400, detail="Document contained no usable text.")

    embedding_response = await async_client.embeddings.create(
        model="text-embedding-4",
        input=chunks,
    )

    document_id = str(uuid.uuid4())
    document_store[document_id] = [
        {"text": chunk, "embedding": embedding_data.embedding}
        for chunk, embedding_data in zip(chunks, embedding_response.data)
    ]

    return UploadDocumentResponse(document_id=document_id, chunk_count=len(chunks))

Embedding every chunk in a single embeddings.create() call, passing the full list of chunks as input, is more efficient than a separate call per chunk — the embeddings endpoint accepts a batch of inputs directly, and zip(chunks, embedding_response.data) pairs each original chunk back up with its corresponding embedding vector in the response, preserving the order both were submitted in.

Implementing Retrieval

Given a question, the most relevant stored chunks across all uploaded documents are found using the same cosine-similarity approach from Unit 10, Lesson 3.

import numpy as np

def cosine_similarity(vector_a: list, vector_b: list) -> float:
    a, b = np.array(vector_a), np.array(vector_b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))

async def retrieve_relevant_chunks(query: str, top_k: int = 5) -> list:
    query_embedding_response = await async_client.embeddings.create(
        model="text-embedding-4",
        input=query,
    )
    query_embedding = query_embedding_response.data[0].embedding

    all_chunks = [chunk for chunks in document_store.values() for chunk in chunks]
    if not all_chunks:
        return []

    scored_chunks = [
        (cosine_similarity(query_embedding, chunk["embedding"]), chunk["text"])
        for chunk in all_chunks
    ]
    scored_chunks.sort(key=lambda pair: pair[0], reverse=True)
    return [text for _, text in scored_chunks[:top_k]]

This flattens chunks from every uploaded document into a single searchable collection — a deliberate simplification appropriate for this capstone's scope, where a production system with a large number of documents and users would more likely need per-user or per-collection scoping (and, at real scale, the vectorized NumPy approach from Unit 10, Lesson 3, or a dedicated vector database, rather than scoring every chunk in a Python loop).

Implementing the Tool

Following Lesson 1's design, one tool beyond retrieval is included: looking up which documents are currently loaded, a request better served by a direct function than by semantic search over document content.

from agents import function_tool

@function_tool
def list_loaded_documents() -> str:
    """Return the number of documents currently loaded into the assistant."""
    return f"{len(document_store)} document(s) currently loaded."

Implementing the Chat Endpoint

The chat endpoint ties every other piece together: retrieval, conversation history, and the tool, combined into a single request following Lesson 1's designed conversation flow.

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    history = conversation_store.setdefault(request.conversation_id, [])
    relevant_chunks = await retrieve_relevant_chunks(request.message)
    context_block = "\n\n".join(relevant_chunks) if relevant_chunks else "No relevant document content found."

    system_instructions = (
        "Answer the user's question using only the provided document context. "
        "If the context doesn't contain the answer, say so rather than guessing.\n\n"
        f"Document context:\n{context_block}"
    )

    messages = [{"role": "system", "content": system_instructions}] + history + [
        {"role": "user", "content": request.message}
    ]

    try:
        response = await async_client.responses.create(
            model="gpt-5.6-terra",
            input=messages,
            tools=[list_loaded_documents],
        )
    except Exception as e:
        raise HTTPException(status_code=502, detail=f"Upstream model request failed: {e}")

    reply_text = response.output_text
    history.append({"role": "user", "content": request.message})
    history.append({"role": "assistant", "content": reply_text})

    return ChatResponse(reply=reply_text, conversation_id=request.conversation_id)

conversation_store.setdefault(request.conversation_id, []) retrieves the existing history for a known conversation or starts a new empty one, directly implementing Unit 4's conversation-state pattern without needing the platform's own conversation-state mechanism, since history here is explicitly managed for full visibility into what gets sent on each turn — again, a deliberate choice matching this capstone's goal of making every step inspectable. The system instructions are rebuilt fresh on every turn with the current retrieval results, since which document chunks are relevant can change entirely from one question to the next even within the same ongoing conversation.

Translating Upstream Errors Into HTTP Responses

The try/except block around the model call reflects Unit 12, Lesson 1's error-handling guidance, adapted to a web application's specific need to return something meaningful to the calling client rather than simply letting an unhandled exception propagate as a generic server error.

from openai import APIStatusError

@app.post("/chat", response_model=ChatResponse)
async def chat_with_detailed_errors(request: ChatRequest):
    try:
        response = await async_client.responses.create(model="gpt-5.6-terra", input=request.message)
    except APIStatusError as e:
        if e.status_code == 429:
            raise HTTPException(status_code=503, detail="The assistant is temporarily overloaded. Please try again shortly.")
        raise HTTPException(status_code=502, detail="The assistant encountered an upstream error.")

    return ChatResponse(reply=response.output_text, conversation_id=request.conversation_id)

Translating a 429 from the model API into a 503 returned to this application's own client — rather than passing the raw upstream status code straight through — reflects that the meaning of "too many requests" is specific to the relationship between this application and the model API, not necessarily the right status code to describe the relationship between this application and whoever is calling it.

Common Mistakes

Using the synchronous OpenAI client inside an async def FastAPI route, blocking the entire server for the duration of every model call rather than allowing other requests to be handled concurrently, exactly the mistake Unit 12, Lesson 6 warned against.

Rebuilding conversation history or retrieval context inconsistently between turns, such as retrieving document context once and reusing stale results across a multi-turn conversation instead of retrieving fresh context for each new question.

Letting an unhandled exception from the model API propagate as a generic, uninformative server error, rather than catching it and returning a status code and message that's actually meaningful to the calling client.

Storing conversation history and document embeddings without any bound, which works for a capstone's scope but would grow memory usage without limit in a real, long-running deployment.

Best Practices

Use the async client throughout, matching FastAPI's own async request handling, so one slow model call doesn't block the server from serving other users.

Define explicit Pydantic request and response models for every endpoint, gaining automatic request validation the same way Unit 6's structured outputs validate model responses.

Retrieve fresh context on every conversation turn rather than caching it across turns, since relevance can shift entirely from one question to the next.

Translate upstream API errors into status codes and messages meaningful to this application's own clients, rather than passing raw upstream errors through unchanged.