Industrial Document Q&A API

Build a RAG API that answers questions about machinery SOPs using Gemini AI.

Introduction

30 Second Summary

Every factory floor has thick procedure manuals that operators need to reference in seconds. Finding the right page during an emergency or maintenance task costs time and creates safety risks.

In this project, you will build a local API that reads a machinery Standard Operating Procedure manual and answers questions about it using AI. You'll feed it document data, and it returns structured JSON reports with step-by-step instructions and safety warnings.

What You'll Build

You'll build an Industrial Document Q&A API powered by Google Gemini, ChromaDB, and FastAPI that ingests SOP documents and answers natural-language questions with structured, typed responses.

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

  • A FastAPI web server with /ingest and /ask endpoints that process industrial SOP documents.
  • A ChromaDB vector store that chunks, indexes, and retrieves relevant sections from a mock Machinery SOP Manual.
  • Pydantic-structured AI responses that return consistent, typed industrial reports using the free Gemini 3.5 Flash model.
  • Secret Mission: Add a /history endpoint that logs all previous queries with timestamps and confidence scores.

Are there any prerequisites?

You need basic Python familiarity (variables, functions, f-strings). No prior experience with AI, APIs, or web servers is required.

Before We Start

Before you can build your Industrial Document Q&A API, you need a working Python environment with all the right libraries installed.

In this step, you will install Python, set up a clean project folder, pull in the libraries your API depends on, and grab a free API key from Google.

In this step, get ready to:
  • Install Python and create your project folder with a virtual environment.
  • Install FastAPI, ChromaDB, google-genai, Pydantic, and python-dotenv.
  • Get a free Gemini API key and store it in a .env file.
Install Python and create your project folder

Your API is built with Python, so you need Python 3.10 or higher installed on your machine.

  • Open your search bar by pressing Cmd+Space (macOS) or the Windows key (Windows).
  • Type Terminal and press Enter to open it.
  • Check your Python version by running this command:

macOS

python3 --version

Windows

python --version

✔️ I see version 3.10 or higher

You are good to go. Move on to the next instruction below.

ⓧ I see an older version

Your system has an older Python that will not work for this project. You need 3.10 or higher.

  • Visit python.org/downloads and download the latest Python 3 installer.
  • Run the installer. On Windows, check the box that says Add python.exe to PATH before clicking Install Now.
  • Close and reopen your terminal, then re-run the version check command above.

macOS default Python

macOS ships with an older Python by default. After installing a newer version, use python3 (not python) to invoke it.

ⓧ Command not found

Python is not installed yet. Let's fix that.

  • Visit python.org/downloads and download the latest Python 3 installer for your operating system.
  • Run the installer. On Windows, check the box that says Add python.exe to PATH before clicking Install Now.
  • Close and reopen your terminal, then run the version check command above to confirm the install.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir industrial-qa-api
cd industrial-qa-api
  • Confirm the folder exists by running:
ls

You should see an empty folder. That means you are in the right place.

A virtual environment keeps this project's libraries separate from other Python projects on your machine.

  • Create and activate your virtual environment by running these commands:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

How do I know the virtual environment is active?

After activating, you will see (venv) at the beginning of your terminal prompt. This confirms all packages you install will stay inside this project.

Seeing a permissions error on Windows?

Windows may block script execution by default. Run Set-ExecutionPolicy -Scope CurrentUser RemoteSigned in PowerShell, then try the activate command again.

help me fix a PowerShell execution policy error when activating my Python virtual environment

  • Open VS Code in your project folder by pressing Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Visual Studio Code and press Enter to open it.
  • In VS Code, click File in the top menu bar.
  • Click Open Folder.
  • Navigate to your Desktop, select the industrial-qa-api folder, and click Open.

Why VS Code?

VS Code has a built-in terminal panel that lets you run commands without leaving the editor. You will use this for the rest of the project.

  • Open the integrated terminal in VS Code by pressing Ctrl+` (backtick key, top-left of your keyboard). This works on both macOS and Windows.
  • If your virtual environment is no longer active (you do not see (venv) in the prompt), activate it again using the platform-specific command above.
Install project dependencies

Your project relies on five libraries. You will pin each to a specific version inside a requirements.txt file so the install is reproducible.

  • In VS Code's Explorer sidebar, right-click inside the file list and select New File.
  • Name the file requirements.txt and press Enter.
  • Paste the following into requirements.txt:
fastapi[standard]==0.136.3
chromadb==1.5.9
google-genai==2.8.0
pydantic==2.13.4
python-dotenv==1.1.0

What are these libraries?

  • fastapi[standard] is the web framework for your API. The [standard] extra includes a development server.
  • chromadb is the vector database that stores and searches your document chunks.
  • google-genai is the official SDK for calling Google's Gemini AI models.
  • pydantic enforces strict data schemas on your API inputs and outputs.
  • python-dotenv loads environment variables from your .env file so you can keep secrets out of code.
  • Save requirements.txt by pressing Cmd+S (macOS) or Ctrl+S (Windows).
  • Install all the dependencies by running this command in your VS Code terminal:
pip install -r requirements.txt

This will download and install all five libraries plus their sub-dependencies. It may take a minute or two.

What about the ChromaDB download?

ChromaDB uses a small embedding model (about 80 MB) to convert text into vectors. It downloads automatically the first time you use ChromaDB, so do not worry if you see a download later in the project.

  • Verify the install succeeded by running:
pip show fastapi

You should see version information including Version: 0.136.3 in the output.

Install failed with a permissions or build error?

Make sure you see (venv) in your terminal prompt. If you do not, your virtual environment is not active. Re-run the activate command from the previous substep.

On Windows, if you see a build error for ChromaDB, make sure you have the latest pip by running python -m pip install --upgrade pip and try again.

help me fix a pip install error for my industrial QA API dependencies

✔️ Awesome, I've got everything!

Great. Make sure requirements.txt is saved and move on.

ⓧ I'd like to double check the full code

fastapi[standard]==0.136.3
chromadb==1.5.9
google-genai==2.8.0
pydantic==2.13.4
python-dotenv==1.1.0
Get your Gemini API key

Your API calls Google's Gemini AI model to generate answers. The free tier covers everything in this project at no cost.

  • Open your browser and go to Google AI Studio.
  • Sign in with your Google account.
  • Click Get API Key in the left sidebar.
  • Click Create API Key.
  • Copy the key that appears. You will only see it once, so paste it somewhere safe temporarily.

Is the Gemini API free?

Yes. The gemini-3.5-flash model has a free tier with free input and output tokens. You will not be charged for anything in this project.

  • Back in VS Code, create a new file called .env by right-clicking in the Explorer sidebar and selecting New File.
  • Name it .env (including the leading dot) and press Enter.
  • Paste the following into your .env file, replacing your-api-key-here with the key you just copied:
GEMINI_API_KEY=your-api-key-here

Why a .env file?

Storing secrets in a .env file keeps them out of your source code. The python-dotenv library loads these values into your app at runtime without ever hard-coding them.

  • Save .env by pressing Cmd+S (macOS) or Ctrl+S (Windows).

File not showing in the sidebar?

Files starting with a dot are hidden by default on some systems. In VS Code's Explorer, they still appear. If you do not see .env, click the refresh icon at the top of the Explorer panel.

help me find my .env file in VS Code

✔️ Awesome, I've got everything!

Great. Make sure your actual API key is in the file (not the placeholder text) and move on.

ⓧ I'd like to double check the full code

Your .env file should contain exactly one line with your real key in place of the placeholder:

GEMINI_API_KEY=your-api-key-here

Your environment is locked and loaded. Next up, you will build the SOP document data that feeds your Q&A pipeline.

Build the SOP Document Data

Your development environment is set up and ready.

Now it's time to build the knowledge base your AI will read from. You'll practice Python's core data structures (lists, dictionaries, and JSON) to represent a real Machinery SOP Manual. AI models can't process entire documents at once, so you need to break them into smaller, searchable chunks.

In this step, get ready to:
  • Define the SOP Manual data as a Python list of dictionaries.
  • Write a chunking function that splits documents into smaller pieces.
  • Test the full pipeline by running your script and verifying the output.
Define the SOP Manual data

Your Q&A system needs a knowledge base to read from. You'll define this as a Python list of dictionaries, where each dictionary represents one section of a factory's Standard Operating Procedure manual.

  • Create a new file called sop_data.py in your industrial-qa-api folder by right-clicking in the VS Code Explorer sidebar and selecting New File.
  • Paste the following code into sop_data.py:
import json

SOP_MANUAL = [
    {
        "section_id": "SOP-001",
        "title": "CNC Lathe Lockout/Tagout Procedure",
        "content": "Before performing any maintenance on the CNC lathe, the operator must follow the lockout/tagout (LOTO) procedure. Step 1: Notify all affected personnel that the machine will be shut down. Step 2: Shut down the CNC lathe using the normal stopping procedure. Step 3: Isolate the machine from its energy source by turning off the main disconnect switch. Step 4: Apply the lockout device to the energy-isolating device. Step 5: Attach a tagout device with your name, date, and reason for lockout. Step 6: Verify isolation by attempting to restart the machine. The machine should not start. Step 7: Proceed with maintenance only after successful verification.",
        "category": "Safety"
    },
    {
        "section_id": "SOP-002",
        "title": "Hydraulic Press Daily Inspection Checklist",
        "content": "The hydraulic press must be inspected daily before operation. Check hydraulic fluid level in the reservoir - it should be between the minimum and maximum marks. Inspect all hoses and fittings for leaks or damage. Verify that pressure gauges are reading within normal operating range (1500-2000 PSI). Check the emergency stop button for proper function by pressing and verifying the press halts immediately. Inspect the die and bolster plate for cracks or wear. Ensure all safety guards are in place and functioning. Record all findings in the daily inspection log.",
        "category": "Maintenance"
    },
    {
        "section_id": "SOP-003",
        "title": "Conveyor Belt Tension Adjustment",
        "content": "Conveyor belt tension must be checked weekly and adjusted if necessary. Proper tension is achieved when the belt deflects no more than 1% of the span length when pressed at the midpoint. To adjust tension: Step 1: Stop the conveyor and engage the lockout procedure (see SOP-001). Step 2: Locate the tension adjustment bolts on the tail pulley. Step 3: Turn both bolts equally in quarter-turn increments. Step 4: Check alignment by running the belt at slow speed and observing tracking. Step 5: If the belt tracks to one side, tighten the bolt on that side by an eighth turn. Step 6: Run at full speed for 5 minutes and recheck tension and tracking.",
        "category": "Maintenance"
    },
    {
        "section_id": "SOP-004",
        "title": "Emergency Shutdown Procedures",
        "content": "In case of emergency, follow these procedures immediately. For fire: Press the nearest emergency stop button, activate the fire alarm pull station, evacuate via the nearest marked exit, and call emergency services. For chemical spill: Press emergency stop on all nearby machinery, evacuate the immediate area, consult the Safety Data Sheet (SDS) for the spilled material, and notify the safety coordinator. For injury: Press emergency stop on the involved machine, do not move the injured person unless there is immediate danger, call for first aid or emergency medical services, and secure the scene to prevent additional injuries. All emergencies must be reported to the shift supervisor within 15 minutes.",
        "category": "Safety"
    },
    {
        "section_id": "SOP-005",
        "title": "Robotic Arm Calibration Procedure",
        "content": "The robotic arm must be calibrated every 500 operating hours or after any collision event. Step 1: Place the robot in teach mode using the teach pendant. Step 2: Move each joint to its mechanical home position. Step 3: Record the encoder values at each home position. Step 4: Compare recorded values to the reference values in the calibration certificate. Step 5: If any axis deviates more than 0.05 degrees from reference, perform a mastering procedure. Step 6: Run the standard test program at 25% speed and verify all waypoints are within tolerance of plus or minus 0.1mm. Step 7: Document calibration results in the maintenance log and update the next calibration due date.",
        "category": "Calibration"
    }
]

What does this code do?

SOP_MANUAL is a Python list containing 5 dictionaries. Each dictionary represents one section of a factory's Standard Operating Procedure manual.

  • section_id gives each section a unique identifier like SOP-001.
  • title is the human-readable name of the procedure.
  • content holds the full text of that SOP section.
  • category classifies the section (Safety, Maintenance, or Calibration).
  • Save sop_data.py.
  • You should see sop_data.py appear in the VS Code Explorer sidebar alongside your .env file.

File not showing up in the sidebar?

  • Make sure you created the file inside the industrial-qa-api folder, not somewhere else on your system.
  • Check that VS Code has the correct folder open. The folder name at the top of the Explorer sidebar should say industrial-qa-api.

Help me figure out why my file isn't showing in VS Code.

Write the chunking function

AI models and vector databases work best with shorter text passages. Your chunking function will split each long SOP section into smaller pieces while keeping track of which section each chunk came from.

  • Add the following function below the SOP_MANUAL list in sop_data.py:


def chunk_documents(documents: list[dict], max_chunk_length: int = 500) -> list[dict]:
    """Split documents into smaller chunks, preserving metadata."""
    chunks = []
    chunk_counter = 0

    for doc in documents:
        content = doc["content"]
        sentences = content.split(". ")

        current_chunk = ""
        for sentence in sentences:
            test_chunk = current_chunk + sentence + ". " if current_chunk else sentence + ". "

            if len(test_chunk) > max_chunk_length and current_chunk:
                chunks.append({
                    "chunk_id": f"chunk-{chunk_counter:03d}",
                    "text": current_chunk.strip(),
                    "metadata": {
                        "section_id": doc["section_id"],
                        "title": doc["title"],
                        "category": doc["category"]
                    }
                })
                chunk_counter += 1
                current_chunk = sentence + ". "
            else:
                current_chunk = test_chunk

        if current_chunk.strip():
            chunks.append({
                "chunk_id": f"chunk-{chunk_counter:03d}",
                "text": current_chunk.strip(),
                "metadata": {
                    "section_id": doc["section_id"],
                    "title": doc["title"],
                    "category": doc["category"]
                }
            })
            chunk_counter += 1

    return chunks

What does this function do?

chunk_documents() takes the full list of SOP sections and breaks each one into smaller text chunks. It splits on sentence boundaries (the ". " pattern) and keeps building a chunk until it would exceed 500 characters.

  • Each chunk gets a unique chunk_id like chunk-000, chunk-001, etc.
  • The metadata dictionary preserves the original section's ID, title, and category so you can trace any chunk back to its source.
  • This approach keeps sentences intact. No word gets cut in half.

Why chunk instead of indexing the full section?

Vector databases find relevant content by comparing the meaning of a query to stored text passages. Shorter chunks produce more precise matches because the meaning of a 3-sentence passage is more focused than a full page of text.

ChromaDB also has a 16KB limit per record, so chunking prevents oversized documents from being rejected.

  • Save sop_data.py.
  • You won't see a change yet. The next section adds a test block so you can run the script and see the chunks in action.

Seeing a red underline in VS Code?

  • Make sure the function is pasted below the closing ] of the SOP_MANUAL list, not inside it.
  • Check that your indentation uses spaces, not tabs. Python is sensitive to inconsistent indentation.

Help me fix the red underline error in my sop_data.py file.

Test the chunking pipeline

Time to verify everything works together. You'll add a test block that runs when you execute the script directly, printing chunk counts and a sample chunk as formatted JSON.

  • Add the following test block at the very bottom of sop_data.py:


if __name__ == "__main__":
    chunks = chunk_documents(SOP_MANUAL)
    print(f"Total documents: {len(SOP_MANUAL)}")
    print(f"Total chunks: {len(chunks)}")
    print("\nSample chunk:")
    print(json.dumps(chunks[0], indent=2))

What does this test block do?

  • if __name__ == "__main__": makes this code run only when you execute the file directly (not when another file imports it).
  • It calls chunk_documents() on your SOP_MANUAL data, then prints the total document count and chunk count.
  • json.dumps() with indent=2 pretty-prints one chunk so you can visually verify the structure and metadata.
  • Save sop_data.py.
  • Run the script in your terminal by running this command:
python sop_data.py

You should see output showing 5 total documents, a chunk count (around 7-9 chunks depending on sentence boundaries), and a formatted JSON sample chunk with chunk_id, text, and metadata fields.

Seeing a ModuleNotFoundError or NameError?

  • If you see ModuleNotFoundError: No module named 'json', check that you typed import json at the very top of the file.
  • If you see NameError: name 'SOP_MANUAL' is not defined, make sure the test block is below both the list and the function, not above them.
  • Make sure your virtual environment is active. You should see (venv) at the start of your terminal prompt.

Help me debug the error I'm getting when running sop_data.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved sop_data.py and your output shows the document count, chunk count, and a sample chunk.

ⓧ I'd like to double check the full code

Here's the complete sop_data.py file:

import json

SOP_MANUAL = [
    {
        "section_id": "SOP-001",
        "title": "CNC Lathe Lockout/Tagout Procedure",
        "content": "Before performing any maintenance on the CNC lathe, the operator must follow the lockout/tagout (LOTO) procedure. Step 1: Notify all affected personnel that the machine will be shut down. Step 2: Shut down the CNC lathe using the normal stopping procedure. Step 3: Isolate the machine from its energy source by turning off the main disconnect switch. Step 4: Apply the lockout device to the energy-isolating device. Step 5: Attach a tagout device with your name, date, and reason for lockout. Step 6: Verify isolation by attempting to restart the machine. The machine should not start. Step 7: Proceed with maintenance only after successful verification.",
        "category": "Safety"
    },
    {
        "section_id": "SOP-002",
        "title": "Hydraulic Press Daily Inspection Checklist",
        "content": "The hydraulic press must be inspected daily before operation. Check hydraulic fluid level in the reservoir - it should be between the minimum and maximum marks. Inspect all hoses and fittings for leaks or damage. Verify that pressure gauges are reading within normal operating range (1500-2000 PSI). Check the emergency stop button for proper function by pressing and verifying the press halts immediately. Inspect the die and bolster plate for cracks or wear. Ensure all safety guards are in place and functioning. Record all findings in the daily inspection log.",
        "category": "Maintenance"
    },
    {
        "section_id": "SOP-003",
        "title": "Conveyor Belt Tension Adjustment",
        "content": "Conveyor belt tension must be checked weekly and adjusted if necessary. Proper tension is achieved when the belt deflects no more than 1% of the span length when pressed at the midpoint. To adjust tension: Step 1: Stop the conveyor and engage the lockout procedure (see SOP-001). Step 2: Locate the tension adjustment bolts on the tail pulley. Step 3: Turn both bolts equally in quarter-turn increments. Step 4: Check alignment by running the belt at slow speed and observing tracking. Step 5: If the belt tracks to one side, tighten the bolt on that side by an eighth turn. Step 6: Run at full speed for 5 minutes and recheck tension and tracking.",
        "category": "Maintenance"
    },
    {
        "section_id": "SOP-004",
        "title": "Emergency Shutdown Procedures",
        "content": "In case of emergency, follow these procedures immediately. For fire: Press the nearest emergency stop button, activate the fire alarm pull station, evacuate via the nearest marked exit, and call emergency services. For chemical spill: Press emergency stop on all nearby machinery, evacuate the immediate area, consult the Safety Data Sheet (SDS) for the spilled material, and notify the safety coordinator. For injury: Press emergency stop on the involved machine, do not move the injured person unless there is immediate danger, call for first aid or emergency medical services, and secure the scene to prevent additional injuries. All emergencies must be reported to the shift supervisor within 15 minutes.",
        "category": "Safety"
    },
    {
        "section_id": "SOP-005",
        "title": "Robotic Arm Calibration Procedure",
        "content": "The robotic arm must be calibrated every 500 operating hours or after any collision event. Step 1: Place the robot in teach mode using the teach pendant. Step 2: Move each joint to its mechanical home position. Step 3: Record the encoder values at each home position. Step 4: Compare recorded values to the reference values in the calibration certificate. Step 5: If any axis deviates more than 0.05 degrees from reference, perform a mastering procedure. Step 6: Run the standard test program at 25% speed and verify all waypoints are within tolerance of plus or minus 0.1mm. Step 7: Document calibration results in the maintenance log and update the next calibration due date.",
        "category": "Calibration"
    }
]


def chunk_documents(documents: list[dict], max_chunk_length: int = 500) -> list[dict]:
    """Split documents into smaller chunks, preserving metadata."""
    chunks = []
    chunk_counter = 0

    for doc in documents:
        content = doc["content"]
        sentences = content.split(". ")

        current_chunk = ""
        for sentence in sentences:
            test_chunk = current_chunk + sentence + ". " if current_chunk else sentence + ". "

            if len(test_chunk) > max_chunk_length and current_chunk:
                chunks.append({
                    "chunk_id": f"chunk-{chunk_counter:03d}",
                    "text": current_chunk.strip(),
                    "metadata": {
                        "section_id": doc["section_id"],
                        "title": doc["title"],
                        "category": doc["category"]
                    }
                })
                chunk_counter += 1
                current_chunk = sentence + ". "
            else:
                current_chunk = test_chunk

        if current_chunk.strip():
            chunks.append({
                "chunk_id": f"chunk-{chunk_counter:03d}",
                "text": current_chunk.strip(),
                "metadata": {
                    "section_id": doc["section_id"],
                    "title": doc["title"],
                    "category": doc["category"]
                }
            })
            chunk_counter += 1

    return chunks


if __name__ == "__main__":
    chunks = chunk_documents(SOP_MANUAL)
    print(f"Total documents: {len(SOP_MANUAL)}")
    print(f"Total chunks: {len(chunks)}")
    print("\nSample chunk:")
    print(json.dumps(chunks[0], indent=2))

Your SOP Manual data is structured and chunked. Next up, you'll connect to Google's Gemini AI and teach it how to answer questions using this data as context.

Write Your First Gemini Prompt

Your SOP data is chunked and ready to be searched. Now you need an AI model that can read those chunks and answer questions about machine maintenance, safety procedures, and calibration steps.

In this step, you will connect to Google's Gemini 3.5 Flash model using the google-genai SDK. You will write a system instruction that shapes Gemini into a strict SOP assistant, then configure it to return structured JSON that matches a defined schema.

In this step, get ready to:
  • Connect to the Gemini API with your API key and initialize the client.
  • Engineer a system instruction that constrains Gemini to answer only from provided context.
  • Configure structured JSON output with a Pydantic schema and verify the full pipeline.
Set up the Gemini client

Before Gemini can answer questions, your script needs a way to authenticate with the API. The google-genai SDK reads your API key and creates a client object that handles all communication with the model.

  • Create a new file called gemini_client.py in your industrial-qa-api folder.
  • Add the imports and client initialization by pasting this code:
import os
import json
from google import genai
from google.genai import types

GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")

client = genai.Client(api_key=GEMINI_API_KEY)

What does this code do?

The from google import genai line imports the official Gemini SDK. The from google.genai import types gives you access to configuration objects like GenerateContentConfig.

The os.environ.get("GEMINI_API_KEY", "") reads your API key from the environment. If the variable is not set, it falls back to an empty string instead of crashing.

Your script needs the GEMINI_API_KEY environment variable to be set in your terminal session before running. You already have the key saved in your .env file, but for running standalone scripts you need to export it manually.

  • Export your API key in the terminal by running the appropriate command for your operating system:

macOS

  • Run this command, replacing your-api-key-here with your actual Gemini API key:
export GEMINI_API_KEY=your-api-key-here

Windows

  • Run this command, replacing your-api-key-here with your actual Gemini API key:
set GEMINI_API_KEY=your-api-key-here

Why export the key instead of hardcoding it?

Hardcoding API keys in source files is a security risk. If you accidentally commit the file to version control, anyone with access can use your key. Environment variables keep secrets out of your code.

Later in the project, the FastAPI server will use python-dotenv to load from .env automatically. For now, the manual export keeps things simple.

Write the system instruction

A system instruction tells the AI model how to behave across all requests. For your SOP assistant, the instruction must enforce strict rules: only answer from provided context, always cite the source section, flag safety warnings, and rate its own confidence.

  • Add the following system instruction below your client initialization in gemini_client.py:
SYSTEM_INSTRUCTION = """You are an industrial SOP (Standard Operating Procedure) assistant. 
You answer questions ONLY based on the provided context from the machinery SOP manual.
If the context does not contain enough information to answer, say so and set confidence_score low.

For every answer you must:
1. Cite the source section (section_id and title)
2. Include any relevant safety warnings
3. Provide step-by-step instructions if the procedure has steps
4. Rate your confidence from 0.0 to 1.0 based on how well the context answers the question"""

What makes this a good system instruction?

This prompt uses several prompt engineering techniques. It assigns a specific role ("You are an industrial SOP assistant"), sets a hard boundary ("ONLY based on the provided context"), and lists exactly what the output must contain.

The confidence score instruction is key. It forces Gemini to self-assess its answer quality. If the retrieved chunks do not contain enough information, the model admits uncertainty rather than hallucinating.

  • Save gemini_client.py.
Configure structured JSON output

Gemini can return free-form text, but for an API you need predictable, parseable output. By setting a response MIME type and providing a JSON schema, you force Gemini to return structured data that matches your exact field requirements.

The schema comes from a Pydantic model. You need to create models.py with the SOPReport class first, so your Gemini client can reference its schema.

  • Create a new file called models.py in your industrial-qa-api folder.
  • Add the SOPReport model by pasting this code:
from pydantic import BaseModel, Field


class SOPReport(BaseModel):
    answer: str = Field(description="The detailed answer based on the SOP manual")
    source_section: str = Field(description="The section ID and title where the answer was found")
    safety_warnings: list[str] = Field(default_factory=list, description="Any safety warnings related to the answer")
    confidence_score: float = Field(ge=0.0, le=1.0, description="Confidence score between 0 and 1")
    steps: list[str] | None = Field(default=None, description="Step-by-step instructions if applicable")

What does this model do?

Each field in SOPReport defines one piece of the AI's response. The Field() calls add constraints like ge=0.0, le=1.0 (confidence must be between 0 and 1) and descriptions that Gemini uses to understand what each field expects.

Pydantic's .model_json_schema() method converts this class into a JSON Schema dictionary. Gemini reads that schema and structures its output to match exactly.

  • Save models.py.

Now you can build the function that sends questions to Gemini with context and gets back a structured report.

  • Switch back to gemini_client.py. Add these two imports at the top of the file, below the existing imports:
from models import SOPReport
from pydantic import ValidationError
  • Add the query_gemini function below your SYSTEM_INSTRUCTION string:
def query_gemini(question: str, context: str) -> SOPReport:
    """Send a question with context to Gemini and return a structured SOPReport."""
    prompt = f"""Context from SOP Manual:
---
{context}
---

Question: {question}

Respond with a JSON object matching this exact schema."""

    response = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_INSTRUCTION,
            response_mime_type="application/json",
            response_json_schema=SOPReport.model_json_schema(),
            temperature=0.3,
        ),
    )

    return parse_gemini_response(response.text)

What does this function do?

The prompt template wraps the retrieved context between --- markers so Gemini can clearly distinguish the manual content from the question.

The GenerateContentConfig does the heavy lifting. response_mime_type="application/json" forces JSON output. response_json_schema tells Gemini exactly which fields to include. temperature=0.3 keeps responses focused and deterministic for factual SOP answers.

  • Add the parse_gemini_response function below query_gemini:
def parse_gemini_response(raw_json: str) -> SOPReport:
    """Validate raw Gemini JSON against the SOPReport model."""
    try:
        data = json.loads(raw_json)
        return SOPReport(**data)
    except (json.JSONDecodeError, ValidationError):
        return SOPReport(
            answer="Unable to parse AI response. Please try again.",
            source_section="Unknown",
            safety_warnings=["Response parsing failed - verify manually"],
            confidence_score=0.0,
            steps=None,
        )

Why a fallback report?

AI models can occasionally return malformed JSON or unexpected fields. The try/except block catches both JSON parsing errors and Pydantic validation failures. Instead of crashing your API, it returns a safe fallback report with a zero confidence score.

  • Add the test block at the bottom of gemini_client.py to verify the full pipeline:
if __name__ == "__main__":
    test_context = "CNC Lathe Lockout/Tagout Procedure: Step 1: Notify all affected personnel. Step 2: Shut down the CNC lathe."
    result = query_gemini("How do I start the lockout procedure?", test_context)
    print(result.model_dump_json(indent=2))
  • Save gemini_client.py.
  • Run the script to test the full pipeline:
python gemini_client.py

You should see a formatted JSON object with fields like answer, source_section, safety_warnings, confidence_score, and steps. The confidence score should be relatively high since the test context directly contains lockout procedure steps.

Seeing an error about GEMINI_API_KEY?

Make sure you exported your API key in the same terminal session where you are running the script. If you opened a new terminal, you need to activate your virtual environment and export the key again.

Also check that your key is valid by visiting aistudio.google.com and confirming it appears in your API keys list.

Help me debug my Gemini API connection

First model download from ChromaDB

If you see a brief download progress bar on your first run, that is ChromaDB downloading its default embedding model (about 80MB). This is a one-time download and will not appear on subsequent runs.

✔️ Awesome, I've got everything!

Great. Double check you have saved both models.py and gemini_client.py.

ⓧ I'd like to double check the full code

Here is the full gemini_client.py for comparison:

import os
import json
from google import genai
from google.genai import types
from models import SOPReport
from pydantic import ValidationError

GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "")

client = genai.Client(api_key=GEMINI_API_KEY)

SYSTEM_INSTRUCTION = """You are an industrial SOP (Standard Operating Procedure) assistant. 
You answer questions ONLY based on the provided context from the machinery SOP manual.
If the context does not contain enough information to answer, say so and set confidence_score low.

For every answer you must:
1. Cite the source section (section_id and title)
2. Include any relevant safety warnings
3. Provide step-by-step instructions if the procedure has steps
4. Rate your confidence from 0.0 to 1.0 based on how well the context answers the question"""


def query_gemini(question: str, context: str) -> SOPReport:
    """Send a question with context to Gemini and return a structured SOPReport."""
    prompt = f"""Context from SOP Manual:
---
{context}
---

Question: {question}

Respond with a JSON object matching this exact schema."""

    response = client.models.generate_content(
        model="gemini-3.5-flash",
        contents=prompt,
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_INSTRUCTION,
            response_mime_type="application/json",
            response_json_schema=SOPReport.model_json_schema(),
            temperature=0.3,
        ),
    )

    return parse_gemini_response(response.text)


def parse_gemini_response(raw_json: str) -> SOPReport:
    """Validate raw Gemini JSON against the SOPReport model."""
    try:
        data = json.loads(raw_json)
        return SOPReport(**data)
    except (json.JSONDecodeError, ValidationError):
        return SOPReport(
            answer="Unable to parse AI response. Please try again.",
            source_section="Unknown",
            safety_warnings=["Response parsing failed - verify manually"],
            confidence_score=0.0,
            steps=None,
        )


if __name__ == "__main__":
    test_context = "CNC Lathe Lockout/Tagout Procedure: Step 1: Notify all affected personnel. Step 2: Shut down the CNC lathe."
    result = query_gemini("How do I start the lockout procedure?", test_context)
    print(result.model_dump_json(indent=2))

And here is models.py:

from pydantic import BaseModel, Field


class SOPReport(BaseModel):
    answer: str = Field(description="The detailed answer based on the SOP manual")
    source_section: str = Field(description="The section ID and title where the answer was found")
    safety_warnings: list[str] = Field(default_factory=list, description="Any safety warnings related to the answer")
    confidence_score: float = Field(ge=0.0, le=1.0, description="Confidence score between 0 and 1")
    steps: list[str] | None = Field(default=None, description="Step-by-step instructions if applicable")

Your AI assistant is now connected and returning structured reports. Next up, you will build the Pydantic models that define your full request and response schemas.

Define Pydantic Models

Your gemini_client.py already imports SOPReport from a file called models.py. That file doesn't exist yet, which means gemini_client.py can't actually run.

In this step, you'll create models.py and define Pydantic models that enforce a strict schema on every AI response and API request. Instead of trusting that Gemini returns the right shape of data, Pydantic validates it for you automatically.

In this step, get ready to:
  • Define a SOPReport model that structures every AI response with typed fields.
  • Define request models for your API endpoints with validation constraints.
  • Verify the full pipeline works by running your Gemini client with the new models.
Create the SOPReport model

Every time Gemini answers a question, you want the response to include specific fields: the answer itself, which SOP section it came from, any safety warnings, a confidence score, and optional step-by-step instructions. A Pydantic BaseModel enforces this structure and rejects responses that don't match.

  • Create a new file called models.py in your industrial-qa-api folder.
  • Add the following imports and SOPReport model:
from pydantic import BaseModel, Field
from datetime import datetime


class SOPReport(BaseModel):
    answer: str = Field(description="The detailed answer based on the SOP manual")
    source_section: str = Field(description="The section ID and title where the answer was found")
    safety_warnings: list[str] = Field(default_factory=list, description="Any safety warnings related to the answer")
    confidence_score: float = Field(ge=0.0, le=1.0, description="Confidence score between 0 and 1")
    steps: list[str] | None = Field(default=None, description="Step-by-step instructions if applicable")

What does this code do?

  • SOPReport inherits from BaseModel, which gives it automatic validation and JSON serialization.
  • Field(ge=0.0, le=1.0) constrains confidence_score to values between 0 and 1. Any value outside that range raises a validation error.
  • default_factory=list means safety_warnings defaults to an empty list if Gemini doesn't provide one.
  • list[str] | None on steps means the field is optional. Not every SOP answer has step-by-step instructions.

What is Pydantic?

Pydantic is a data validation library for Python. You define the shape of your data as a class, and Pydantic automatically validates any data against that shape at runtime.

FastAPI uses Pydantic natively, so the models you define here will also serve as your API's request and response schemas in the next steps.

Define the request models and query log

Your API will have two endpoints that accept requests: one for ingesting documents and one for asking questions. Each needs its own model to validate incoming data.

  • In models.py, add the following models below your SOPReport class:
class IngestRequest(BaseModel):
    document_id: str = Field(default="sop-manual-v1", description="Identifier for the document set to ingest")


class AskRequest(BaseModel):
    question: str = Field(description="The question to ask about the SOP manual")
    n_results: int = Field(default=3, ge=1, le=10, description="Number of relevant chunks to retrieve")


class QueryLog(BaseModel):
    timestamp: datetime
    question: str
    answer_preview: str = Field(description="First 100 characters of the answer")
    confidence_score: float
    sources_used: list[str]

What does this code do?

  • IngestRequest accepts a document_id string that defaults to "sop-manual-v1" if the caller doesn't provide one.
  • AskRequest requires a question string and constrains n_results to between 1 and 10 with ge=1, le=10. This prevents callers from requesting zero results or overwhelming the system.
  • QueryLog stores a record of each query for history tracking. You'll use this later to build a query history endpoint.
  • Save models.py.

Seeing a red underline on list[str] | None?

The | union syntax requires Python 3.10 or higher. If your editor flags it, confirm you're running the correct Python version in your virtual environment.

help me fix a Python version issue with union type syntax

Verify the models work with your Gemini client

Your gemini_client.py already imports SOPReport from models.py and uses it in parse_gemini_response(). Now that the file exists, the full pipeline should work end to end.

  • Run your Gemini client to test the models by running this command:
python gemini_client.py
  • You should see a JSON object printed in your terminal with fields matching your SOPReport model: answer, source_section, safety_warnings, confidence_score, and steps.

How does parse_gemini_response use SOPReport?

The parse_gemini_response() function in gemini_client.py takes raw JSON from Gemini, parses it, and passes the data to SOPReport(**data). Pydantic then validates every field: it checks that confidence_score is between 0 and 1, that answer is a string, and so on.

If validation fails, the function catches the error and returns a safe fallback report instead of crashing your API.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. Your terminal prompt should show (venv) at the beginning. If it doesn't, activate it again with venv\Scripts\activate on Windows or source venv/bin/activate on macOS.

Also confirm that models.py is saved in the same folder as gemini_client.py. Python looks for imports relative to the file that's running.

help me fix a ModuleNotFoundError when running gemini_client.py

✔️ Awesome, I've got everything!

Great. Double check you've saved models.py and confirmed the JSON output in your terminal.

ⓧ I'd like to double check the full code

Here is the complete models.py file:

from pydantic import BaseModel, Field
from datetime import datetime


class SOPReport(BaseModel):
    answer: str = Field(description="The detailed answer based on the SOP manual")
    source_section: str = Field(description="The section ID and title where the answer was found")
    safety_warnings: list[str] = Field(default_factory=list, description="Any safety warnings related to the answer")
    confidence_score: float = Field(ge=0.0, le=1.0, description="Confidence score between 0 and 1")
    steps: list[str] | None = Field(default=None, description="Step-by-step instructions if applicable")


class IngestRequest(BaseModel):
    document_id: str = Field(default="sop-manual-v1", description="Identifier for the document set to ingest")


class AskRequest(BaseModel):
    question: str = Field(description="The question to ask about the SOP manual")
    n_results: int = Field(default=3, ge=1, le=10, description="Number of relevant chunks to retrieve")


class QueryLog(BaseModel):
    timestamp: datetime
    question: str
    answer_preview: str = Field(description="First 100 characters of the answer")
    confidence_score: float
    sources_used: list[str]

Your data models are locked in. Every AI response and API request now has a strict contract that Pydantic enforces at runtime. Next up, you'll build the vector store that indexes your SOP manual for semantic search.

Index the SOP Manual with ChromaDB

Your Pydantic models give Gemini a strict format to follow. But there's a missing piece. When a question comes in, you need a way to find the most relevant SOP sections to send as context.

This is where ChromaDB comes in. It stores your document chunks as vector embeddings and retrieves the most semantically similar ones when you ask a question. In this step, you'll build the indexing and retrieval layer.

In this step, get ready to:
  • Initialize a ChromaDB in-memory vector store and write an ingestion function.
  • Write a query function that retrieves relevant chunks by semantic similarity.
  • Test the full pipeline by ingesting your SOP data and querying it.
Initialize ChromaDB and write the ingestion function

A vector database turns text into numerical representations (embeddings) and lets you search by meaning instead of exact keywords. ChromaDB handles the embedding automatically using its built-in Sentence Transformer model.

  • Create a new file called vector_store.py in your industrial-qa-api folder.
  • Add the ChromaDB client initialization and the ingest_chunks() function by pasting this code:
import chromadb

# Create an in-memory ChromaDB client (no external server needed)
chroma_client = chromadb.Client()
# Create or connect to a collection named "sop-manual"
collection = chroma_client.get_or_create_collection(name="sop-manual")


def ingest_chunks(chunks: list[dict]) -> int:
    """Add document chunks to the ChromaDB collection."""
    if not chunks:
        return 0

    # Add all chunks to the collection - ChromaDB auto-generates embeddings
    collection.add(
        ids=[chunk["chunk_id"] for chunk in chunks],
        documents=[chunk["text"] for chunk in chunks],
        metadatas=[chunk["metadata"] for chunk in chunks],
    )
    return len(chunks)

What does this code do?

chromadb.Client() creates an in-memory vector database. All data lives in RAM, which means it disappears when you stop the script. This is perfect for development.

get_or_create_collection(name="sop-manual") creates a named collection (like a table in a regular database) to store your document chunks. The ingest_chunks() function takes the chunked dicts from sop_data.py and adds them to the collection. ChromaDB automatically generates embeddings from the text.

What is a vector database?

A traditional database finds records by exact matches (like searching for a specific section ID). A vector database converts text into numbers (vectors) that capture meaning. When you search, it finds the chunks whose meaning is closest to your question.

This is why a query like "How do I lock out a machine?" can find the lockout/tagout section even though it uses different words.

  • Save vector_store.py.
Write the query function

The ingestion function loads data in. Now you need a function that searches the collection and returns the most relevant chunks for a given question.

  • In vector_store.py, add the query_manual() function below the ingest_chunks() function:
def query_manual(question: str, n_results: int = 3) -> list[dict]:
    """Query the SOP manual collection and return relevant chunks."""
    results = collection.query(
        query_texts=[question],
        n_results=n_results,
    )

    # Restructure ChromaDB's response into a clean list of dicts
    retrieved = []
    for i in range(len(results["documents"][0])):
        retrieved.append({
            "text": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
        })
    return retrieved

What does this code do?

collection.query() takes your question, converts it into a vector, and finds the n_results most similar chunks in the collection. It returns a nested structure with separate lists for documents, metadatas, and distances.

The loop restructures that nested response into a cleaner format. Each item in the returned list has a text key (the chunk content) and a metadata key (the section title, ID, and category). This is the format your FastAPI endpoint will consume in the next step.

  • Save vector_store.py.
Test the full retrieval pipeline

Time to see the vector store in action. You'll add a test block that ingests your SOP data and queries it with a real question to verify that ChromaDB returns relevant results.

  • At the bottom of vector_store.py, add this test block:
if __name__ == "__main__":
    from sop_data import SOP_MANUAL, chunk_documents

    # Chunk the SOP manual and ingest into ChromaDB
    chunks = chunk_documents(SOP_MANUAL)
    count = ingest_chunks(chunks)
    print(f"Ingested {count} chunks")

    # Test a semantic query
    results = query_manual("How do I perform a lockout tagout?")
    for r in results:
        print(f"\n[{r['metadata']['title']}]")
        print(r["text"][:150] + "...")

What does this test do?

This imports your SOP data and chunking function, feeds all chunks into ChromaDB, then asks "How do I perform a lockout tagout?". ChromaDB should return the CNC Lathe Lockout/Tagout section as the most relevant result.

The [:150] slice keeps the output readable by only printing the first 150 characters of each chunk.

  • Save vector_store.py.
  • Run the script by entering this command in your terminal:
python vector_store.py
  • You should see output showing the number of ingested chunks, followed by retrieved sections with titles in square brackets and a preview of the chunk text.

What should I see?

The top result should be the CNC Lathe Lockout/Tagout Procedure section. ChromaDB found it by comparing the meaning of your question to the meaning of each chunk. Even though you wrote "lockout tagout" and the manual says "lockout/tagout (LOTO)", the semantic search understands they mean the same thing.

Seeing a slow first run or download messages?

ChromaDB downloads its default embedding model (about 80MB) on first use. This is a one-time download. If you see download progress in your terminal, wait for it to complete, then the query results will appear.

If you see an ImportError, make sure your virtual environment is activated and that you ran pip install chromadb in an earlier step.

Why is my vector store query returning unexpected results?

✔️ Awesome, I've got everything!

Great. Double check you've saved vector_store.py and confirmed the query returns relevant lockout/tagout results.

ⓧ I'd like to double check the full code

import chromadb

chroma_client = chromadb.Client()
collection = chroma_client.get_or_create_collection(name="sop-manual")


def ingest_chunks(chunks: list[dict]) -> int:
    """Add document chunks to the ChromaDB collection."""
    if not chunks:
        return 0

    collection.add(
        ids=[chunk["chunk_id"] for chunk in chunks],
        documents=[chunk["text"] for chunk in chunks],
        metadatas=[chunk["metadata"] for chunk in chunks],
    )
    return len(chunks)


def query_manual(question: str, n_results: int = 3) -> list[dict]:
    """Query the SOP manual collection and return relevant chunks."""
    results = collection.query(
        query_texts=[question],
        n_results=n_results,
    )

    retrieved = []
    for i in range(len(results["documents"][0])):
        retrieved.append({
            "text": results["documents"][0][i],
            "metadata": results["metadatas"][0][i],
        })
    return retrieved


if __name__ == "__main__":
    from sop_data import SOP_MANUAL, chunk_documents

    chunks = chunk_documents(SOP_MANUAL)
    count = ingest_chunks(chunks)
    print(f"Ingested {count} chunks")

    results = query_manual("How do I perform a lockout tagout?")
    for r in results:
        print(f"\n[{r['metadata']['title']}]")
        print(r["text"][:150] + "...")

Your vector store is loaded and answering queries. Next up, you'll wire everything together into a FastAPI server with real HTTP endpoints.

Build the FastAPI Server

You have all the building blocks ready: SOP document data, a ChromaDB vector store, a Gemini client, and Pydantic models.

Now you need to expose all of this through a web API so that any client (a front-end app, a command-line tool, or another service) can ingest documents and ask questions over HTTP. FastAPI gives you typed endpoints, automatic validation, and built-in interactive docs with zero extra configuration.

In this step, get ready to:
  • Create a FastAPI app with a /ingest endpoint that chunks and indexes your SOP manual.
  • Create a /ask endpoint that queries ChromaDB, sends context to Gemini, and returns a structured report.
  • Run the dev server and test both endpoints through the interactive docs.
Set up the FastAPI app and the /ingest endpoint

The first endpoint you need is one that takes your SOP manual data, chunks it, and loads it into ChromaDB. This way, any client can trigger document indexing with a single HTTP request.

  • Create a new file called main.py in your industrial-qa-api folder.
  • Add the imports, app setup, and /ingest endpoint by pasting the following code:
import os
from datetime import datetime, timezone
from dotenv import load_dotenv
from fastapi import FastAPI
from models import IngestRequest, AskRequest, SOPReport, QueryLog
from sop_data import SOP_MANUAL, chunk_documents
from vector_store import ingest_chunks, query_manual
from gemini_client import query_gemini

# Load environment variables from .env file
load_dotenv()

# Create the FastAPI application instance
app = FastAPI(
    title="Industrial Document Q&A API",
    description="Ask questions about machinery SOPs and get structured answers powered by Gemini AI",
    version="1.0.0",
)

# In-memory list to store query history for later retrieval
query_history: list[QueryLog] = []


@app.post("/ingest")
def ingest_documents(request: IngestRequest) -> dict:
    """Chunk and index the SOP manual into ChromaDB."""
    chunks = chunk_documents(SOP_MANUAL)
    count = ingest_chunks(chunks)
    return {
        "status": "success",
        "document_id": request.document_id,
        "chunks_indexed": count,
    }

What does this code do?

  • load_dotenv() reads your .env file and makes GEMINI_API_KEY available as an environment variable. This happens before any other code runs.
  • FastAPI(...) creates your web server with metadata that appears in the auto-generated documentation.
  • query_history is an in-memory list that will log every question asked. You will use this in the Secret Mission later.
  • The /ingest endpoint calls chunk_documents() and ingest_chunks() from your earlier files, then returns a success message with the chunk count.
  • Save main.py.

Seeing import errors?

Make sure main.py is in the same folder as models.py, sop_data.py, vector_store.py, and gemini_client.py. All files must be in the same directory for the imports to resolve.

Help me debug import errors in my FastAPI project

Create the /ask endpoint

The /ask endpoint is where the full RAG pipeline comes together. It receives a question, retrieves relevant chunks from ChromaDB, builds a context string, and sends everything to Gemini for a structured answer.

  • In main.py, add the /ask endpoint below the /ingest function by pasting this code:
@app.post("/ask", response_model=SOPReport)
def ask_question(request: AskRequest) -> SOPReport:
    """Query the SOP manual and get a structured AI-powered answer."""
    # Retrieve relevant chunks from the vector store
    retrieved = query_manual(request.question, request.n_results)

    # Build context string from retrieved chunks
    context_parts = []
    sources_used = []
    for chunk in retrieved:
        section_info = f"[{chunk['metadata']['section_id']}] {chunk['metadata']['title']}"
        context_parts.append(f"{section_info}:\n{chunk['text']}")
        sources_used.append(section_info)

    context = "\n\n".join(context_parts)

    # Send question + context to Gemini and get structured report
    report = query_gemini(request.question, context)

    # Log the query for history tracking
    query_history.append(QueryLog(
        timestamp=datetime.now(timezone.utc),
        question=request.question,
        answer_preview=report.answer[:100],
        confidence_score=report.confidence_score,
        sources_used=sources_used,
    ))

    return report

What does this endpoint do?

This is the complete RAG pipeline in one function:

  • query_manual() retrieves the most relevant SOP chunks from ChromaDB based on semantic similarity to the question.
  • The loop builds a formatted context string with section IDs and titles, so Gemini knows where each piece of information came from.
  • query_gemini() sends the question and context to Gemini, which returns a validated SOPReport with answer, sources, safety warnings, and confidence score.
  • The response_model=SOPReport parameter tells FastAPI to validate and document the response shape automatically.
  • Save main.py.

Not sure where to paste the code?

The @app.post("/ask", response_model=SOPReport) decorator and function should go directly after the ingest_documents function, at the same indentation level (not nested inside another function).

Help me figure out where to add my /ask endpoint

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py with both endpoints.

ⓧ I'd like to double check the full code

import os
from datetime import datetime, timezone
from dotenv import load_dotenv
from fastapi import FastAPI
from models import IngestRequest, AskRequest, SOPReport, QueryLog
from sop_data import SOP_MANUAL, chunk_documents
from vector_store import ingest_chunks, query_manual
from gemini_client import query_gemini

load_dotenv()

app = FastAPI(
    title="Industrial Document Q&A API",
    description="Ask questions about machinery SOPs and get structured answers powered by Gemini AI",
    version="1.0.0",
)

query_history: list[QueryLog] = []


@app.post("/ingest")
def ingest_documents(request: IngestRequest) -> dict:
    """Chunk and index the SOP manual into ChromaDB."""
    chunks = chunk_documents(SOP_MANUAL)
    count = ingest_chunks(chunks)
    return {
        "status": "success",
        "document_id": request.document_id,
        "chunks_indexed": count,
    }


@app.post("/ask", response_model=SOPReport)
def ask_question(request: AskRequest) -> SOPReport:
    """Query the SOP manual and get a structured AI-powered answer."""
    retrieved = query_manual(request.question, request.n_results)

    context_parts = []
    sources_used = []
    for chunk in retrieved:
        section_info = f"[{chunk['metadata']['section_id']}] {chunk['metadata']['title']}"
        context_parts.append(f"{section_info}:\n{chunk['text']}")
        sources_used.append(section_info)

    context = "\n\n".join(context_parts)

    report = query_gemini(request.question, context)

    query_history.append(QueryLog(
        timestamp=datetime.now(timezone.utc),
        question=request.question,
        answer_preview=report.answer[:100],
        confidence_score=report.confidence_score,
        sources_used=sources_used,
    ))

    return report
Run the server and test both endpoints

FastAPI includes a built-in development server with auto-reload. It also generates interactive API documentation where you can test your endpoints directly in the browser.

  • Start the development server by running this command:
fastapi dev main.py

What does this command do?

fastapi dev starts a development server with auto-reload enabled. It detects your app variable in main.py automatically. The server runs at http://127.0.0.1:8000 by default.

  • You should see output in your terminal showing the server is running. Look for a line containing Uvicorn running on http://127.0.0.1:8000.
  • Open your browser and navigate to http://127.0.0.1:8000/docs.

You should see the Swagger UI interactive documentation page with your two endpoints listed: /ingest and /ask.

What is this /docs page?

FastAPI auto-generates interactive API documentation using Swagger UI. You can expand each endpoint, fill in request bodies, and click **Execute** to send real requests to your running server. No need for a separate tool like Postman.

  • Test the /ingest endpoint first. Click on POST /ingest to expand it.
  • Click **Try it out**.
  • Leave the default request body as-is (it already contains "document_id": "sop-manual-v1").
  • Click **Execute**.

You should see a response with "status": "success" and a "chunks_indexed" number showing how many chunks were loaded into ChromaDB.

  • Now test the /ask endpoint. Click on POST /ask to expand it.
  • Click **Try it out**.
  • Replace the request body with this JSON:
{
  "question": "What is the lockout tagout procedure for the CNC lathe?",
  "n_results": 3
}
  • Click **Execute**.

You should see a structured JSON response containing an answer, source_section, safety_warnings, confidence_score, and steps fields. The confidence score should be high (close to 1.0) because the lockout/tagout procedure is well-covered in the SOP data.

Getting a 500 Internal Server Error?

Make sure you ingested documents first by calling /ingest before /ask. ChromaDB needs data loaded before it can return results.

If you see a Gemini API error, check that your .env file contains a valid API key and that the server was restarted after you created the file.

Help me debug my FastAPI 500 error

Your RAG API is live and responding to questions with structured, AI-powered answers grounded in the SOP manual. Next up, you will take on the Secret Mission and add query history tracking to your API.

Secret mission

Add a /history endpoint that logs all queries with timestamps and confidence scores

Your API answers questions, but it has no memory of what was asked before. In production, query logging is essential for debugging, auditing, and spotting usage patterns. Can you add a /history endpoint that tracks every query and lets you filter by confidence?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally and uses the free tier of the Gemini API, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The industrial-qa-api project folder on your Desktop (includes all Python files and the virtual environment).
  • Gemini API key in Google AI Studio.
  • FastAPI development server (running locally in your terminal).

Keep everything running

No action needed. Choose this if you're still actively building or experimenting with the API.

  • Your FastAPI server keeps running as long as the terminal window stays open.
  • The Gemini API free tier has no charges, so your API key can remain active.
  • Your project folder and virtual environment stay on your Desktop.

Pause - I'll come back to this later

Shut down the running server to free up your terminal. Your files stay intact so you can pick up where you left off.

  • Switch to the terminal running your FastAPI server and press Ctrl+C to stop it.
  • Your project folder and all files remain on your Desktop. When you come back, navigate to the folder, activate the virtual environment, and run fastapi dev main.py again.

Delete - I don't want to use this again

Remove all project files and revoke your API key. This frees up disk space and ensures your credentials are no longer active.

  • Switch to the terminal running your FastAPI server and press Ctrl+C to stop it.
  • Delete the project folder by running this command from your Desktop:

macOS

cd ~/Desktop
rm -rf industrial-qa-api

Windows

cd ~\Desktop
rmdir /s /q industrial-qa-api
  • Delete your Gemini API key from Google AI Studio. Go to aistudio.google.com, open the API keys page, find your key, and click the delete (trash) icon next to it.

Nice Work!

Nice Work!

Nice work! You just built a fully functional Industrial Document Q&A API from scratch.

You've learned how to:

  • Built a FastAPI web server with /ingest and /ask endpoints that process industrial SOP documents and return structured JSON responses.
  • Integrated ChromaDB as a vector database to chunk, index, and semantically retrieve relevant sections from a machinery SOP manual.
  • Tied it all together into a working RAG pipeline where users query the API, the vector store retrieves context, and Gemini 3.5 Flash returns Pydantic-validated reports.
  • Secret Mission: Added a /history endpoint that logs all queries with timestamps and confidence scores, filterable by a minimum confidence threshold.

Ready to quiz yourself?