Legal & Compliance Q&A API with RAG

Build a cited Q&A API that answers compliance questions from legal docs.

Introduction

30 Second Summary

When someone asks a legal question at work, getting the wrong answer can have serious consequences. AI tools that guess or hallucinate in this space are not just unhelpful. They are dangerous.

In this project, you will build a Q&A API that answers compliance questions using only the content of uploaded legal documents. Every answer includes source citations, and the system refuses to speculate when it does not have enough information.

What You'll Build

A FastAPI-powered Legal & Compliance Q&A service that ingests documents, stores them as vector embeddings in ChromaDB, and generates cited answers using Claude.

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

  • A document ingestion endpoint that chunks legal text by section structure instead of arbitrary token counts.
  • A RAG pipeline with ChromaDB for vector storage and retrieval of relevant sections.
  • A Q&A endpoint that enforces citation discipline and refuses to answer when documents lack sufficient context.
  • Secret Mission: a faithfulness evaluator that programmatically checks whether each citation actually supports its claim.

Are there any prerequisites?

You should have basic Python knowledge and some familiarity with REST APIs. No prior experience with vector databases or RAG is required.

Build the Legal Document Chunker

Your environment is ready and connected to the Claude API. Now you need a way to break legal documents into meaningful pieces that your RAG system can search.

Most text-splitting tools chop documents into fixed-size token windows. That approach destroys the structure of legal documents, splitting clauses mid-sentence and separating definitions from the obligations that reference them. In this step, you'll build a chunker that respects section boundaries.

In this step, get ready to:
  • Create a chunker module that splits legal text by numbered section headings.
  • Attach metadata to each chunk including document name, section number, title, and type classification.
  • Test the chunker against a sample compliance document.
Create the chunker module and data structure

Each chunk your RAG system stores needs to carry information about where it came from. You'll define this shape as a Python dataclass, then build the splitting logic around it.

  • Create a new file called chunker.py in your project folder.
  • Add the imports and Chunk dataclass by pasting this code:
import re
from dataclasses import dataclass


@dataclass
class Chunk:
    text: str
    doc_name: str
    section_id: str
    section_title: str
    chunk_type: str

What does this code do?

The @dataclass decorator automatically generates an __init__ method and other useful methods for your class. You just declare the fields with type annotations.

  • text stores the actual section content.
  • doc_name tracks which document this chunk came from.
  • section_id and section_title record the section's position and heading in the document.
  • chunk_type classifies the content (definition, obligation, right, or general).
  • Save chunker.py.

You won't see a change yet. The next chunk adds the functions that use this dataclass.

Add metadata classification and chunking logic

Legal documents contain different types of content. Definitions establish what terms mean. Obligations state what must happen. Your chunker needs to tag each section so the retrieval system can prioritize the right type of content.

  • In chunker.py, add the classify_chunk() function below the Chunk dataclass:
def classify_chunk(text: str) -> str:
    """Classify a chunk as definition, obligation, or general based on content."""
    lower = text.lower()
    if '"' in text and "means" in lower:
        return "definition"
    if "shall" in lower or "must" in lower:
        return "obligation"
    if "right to" in lower or "have the right" in lower:
        return "right"
    return "general"

What does this code do?

This function uses keyword heuristics to classify legal text into categories:

  • Sections containing quoted terms with "means" are definitions (e.g., '"Personal Data" means...').
  • Sections with "shall" or "must" are obligations that mandate specific behaviour.
  • Sections mentioning "right to" describe rights granted to data subjects.
  • Everything else is classified as general.
  • Now add the main chunk_legal_document() function below classify_chunk():
def chunk_legal_document(text: str, doc_name: str) -> list[Chunk]:
    """Split a legal document into chunks by section structure."""
    section_pattern = re.compile(
        r"^(\d+(?:\.\d+)?)\s+(.+?)$", re.MULTILINE
    )

    matches = list(section_pattern.finditer(text))
    chunks = []

    for i, match in enumerate(matches):
        section_id = match.group(1)
        section_title = match.group(2).strip()
        start = match.start()
        end = matches[i + 1].start() if i + 1 < len(matches) else len(text)

        section_text = text[start:end].strip()

        if len(section_text) < 20:
            continue

        chunk = Chunk(
            text=section_text,
            doc_name=doc_name,
            section_id=section_id,
            section_title=section_title,
            chunk_type=classify_chunk(section_text),
        )
        chunks.append(chunk)

    return chunks

What does this code do?

This function uses a regex pattern to find numbered section headings (like "1. Introduction" or "2.1 Personal Data"). It then slices the document at each heading boundary.

  • section_pattern matches lines starting with a number (optionally followed by a decimal), then a space, then the heading text.
  • For each match, the function captures everything from that heading to the start of the next heading.
  • Sections shorter than 20 characters are skipped as they likely contain no meaningful content.
  • Each section becomes a Chunk with its classification determined by classify_chunk().

Why section-based chunking over fixed-size splitting?

Legal documents have inherent structure. Numbered sections contain complete clauses and cross-references. Splitting by section boundaries keeps clauses intact and preserves meaning.

Fixed-size chunking (splitting every 500 tokens) would cut clauses mid-sentence and separate definitions from the obligations that reference them. Research shows section-aware chunking outperforms fixed-size chunking by 8-15 points of recall on legal document corpora.

  • Save chunker.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file before moving on.

ⓧ I'd like to double check the full code

Here is the complete chunker.py so far:

import re
from dataclasses import dataclass


@dataclass
class Chunk:
    text: str
    doc_name: str
    section_id: str
    section_title: str
    chunk_type: str


def classify_chunk(text: str) -> str:
    """Classify a chunk as definition, obligation, or general based on content."""
    lower = text.lower()
    if '"' in text and "means" in lower:
        return "definition"
    if "shall" in lower or "must" in lower:
        return "obligation"
    if "right to" in lower or "have the right" in lower:
        return "right"
    return "general"


def chunk_legal_document(text: str, doc_name: str) -> list[Chunk]:
    """Split a legal document into chunks by section structure."""
    section_pattern = re.compile(
        r"^(\d+(?:\.\d+)?)\s+(.+?)$", re.MULTILINE
    )

    matches = list(section_pattern.finditer(text))
    chunks = []

    for i, match in enumerate(matches):
        section_id = match.group(1)
        section_title = match.group(2).strip()
        start = match.start()
        end = matches[i + 1].start() if i + 1 < len(matches) else len(text)

        section_text = text[start:end].strip()

        if len(section_text) < 20:
            continue

        chunk = Chunk(
            text=section_text,
            doc_name=doc_name,
            section_id=section_id,
            section_title=section_title,
            chunk_type=classify_chunk(section_text),
        )
        chunks.append(chunk)

    return chunks
Test the chunker against a sample document

You need a real legal document to verify the chunker works correctly. You'll create a sample data privacy policy, then run the chunker against it to confirm it splits by section and attaches the right metadata.

  • Create a new file called sample_policy.txt in your project folder.
  • Paste the following sample compliance document into it:
COMPANY DATA PRIVACY POLICY
Effective Date: January 1, 2025

1. Introduction
This Data Privacy Policy ("Policy") governs how Acme Corporation ("Company," "we," "us," or "our") collects, uses, stores, and protects personal data. This Policy applies to all employees, contractors, and third-party service providers who process personal data on behalf of the Company.

2. Definitions
2.1 "Personal Data" means any information relating to an identified or identifiable natural person, including but not limited to names, email addresses, identification numbers, location data, and online identifiers.
2.2 "Processing" means any operation performed on personal data, including collection, recording, storage, retrieval, use, disclosure, or deletion.
2.3 "Data Subject" means the identified or identifiable natural person to whom the personal data relates.
2.4 "Data Controller" means the entity that determines the purposes and means of processing personal data.

3. Data Collection
3.1 The Company shall collect personal data only for specified, explicit, and legitimate purposes.
3.2 Data collection must be limited to what is necessary for the stated purpose (data minimization principle).
3.3 Personal data shall be collected directly from the data subject where practicable. Where data is collected from third parties, the data subject must be informed within 30 days.

4. Data Storage and Security
4.1 Personal data shall be stored in encrypted form using AES-256 encryption at rest.
4.2 Access to personal data shall be restricted to authorized personnel only, following the principle of least privilege.
4.3 Data retention shall not exceed the period necessary for the processing purpose. Default retention period is 3 years unless a longer period is required by law.
4.4 All data storage systems must undergo annual security audits conducted by an independent third party.

5. Data Subject Rights
5.1 Data subjects have the right to access their personal data within 30 days of a written request.
5.2 Data subjects have the right to rectification of inaccurate personal data without undue delay.
5.3 Data subjects have the right to erasure ("right to be forgotten") when the data is no longer necessary for its original purpose.
5.4 Data subjects have the right to data portability in a structured, commonly used, machine-readable format.

6. Data Breach Notification
6.1 In the event of a personal data breach, the Company shall notify the relevant supervisory authority within 72 hours of becoming aware of the breach.
6.2 If the breach is likely to result in a high risk to the rights and freedoms of data subjects, those individuals must be notified without undue delay.
6.3 All data breaches must be documented in the Company's breach register, including the nature of the breach, categories of data affected, and remedial actions taken.

7. Third-Party Data Sharing
7.1 Personal data shall not be shared with third parties without a lawful basis for processing.
7.2 All third-party processors must execute a Data Processing Agreement (DPA) before receiving personal data.
7.3 Cross-border data transfers are permitted only to countries with an adequate level of data protection or under appropriate safeguards such as Standard Contractual Clauses.

8. Compliance and Enforcement
8.1 Violations of this Policy may result in disciplinary action, up to and including termination of employment or contract.
8.2 The Data Protection Officer shall conduct quarterly reviews of Policy compliance.
8.3 Employees must complete annual data protection training within 60 days of policy updates.
  • Save sample_policy.txt.
  • Confirm both files exist by running:
ls

You should see chunker.py and sample_policy.txt in the output.

  • Test the chunker by running this command:
python3 -c "
from chunker import chunk_legal_document

with open('sample_policy.txt', 'r') as f:
    text = f.read()

chunks = chunk_legal_document(text, 'data_privacy_policy')
print(f'Total chunks: {len(chunks)}')

for chunk in chunks[:3]:
    print(f'\n[{chunk.section_id}] {chunk.section_title} ({chunk.chunk_type})')
    print(f'  {chunk.text[:80]}...')
"

You should see output showing the total number of chunks and details for the first three sections, with each one tagged with its section ID, title, and type.

What should I see?

The output shows the chunker found all the sections in the document. Section 1 (Introduction) is classified as "general". Section 2 (Definitions) and its subsections are classified as "definition" because they contain quoted terms with "means". Section 3 (Data Collection) is classified as "obligation" because it uses "shall" and "must".

Seeing a ModuleNotFoundError?

  • Make sure you're running the command from the same directory where chunker.py lives.
  • Check that the file is named exactly chunker.py (not chunker.py.txt or Chunker.py).
  • Confirm your virtual environment is still active (you should see the environment name in your terminal prompt).

Still stuck? Help me debug why Python can't find my chunker module.

Your chunker is working. It splits legal documents by their natural section boundaries and tags each piece with metadata. Next up, you'll store these chunks in a vector database so your API can search them by meaning.

Set Up ChromaDB and Ingest Documents

Your chunker splits legal documents into meaningful sections. Now you need somewhere to store those chunks so your Q&A engine can retrieve the most relevant ones for any question.

In this step, you will set up ChromaDB as your vector database. ChromaDB converts text into vector embeddings automatically and stores them on disk. When you query it with a question, it finds the chunks whose meaning is closest to your query.

In this step, get ready to:
  • Initialize a ChromaDB vector store for legal documents.
  • Ingest the sample privacy policy into the store.
  • Verify retrieval works by running a test query.
Initialize the ChromaDB client and collection

ChromaDB needs two things to work: a client that manages the database on disk, and a collection that holds your documents. A collection in ChromaDB is like a table in a traditional database.

  • Create a new file called db.py in your project folder.
  • Add the following code to set up the ChromaDB client and collection:
import chromadb

COLLECTION_NAME = "legal_documents"


def get_chroma_client():
    """Create a persistent ChromaDB client."""
    return chromadb.PersistentClient(path="./chroma_data")


def get_or_create_collection(client=None):
    """Get or create the legal documents collection."""
    if client is None:
        client = get_chroma_client()
    return client.get_or_create_collection(name=COLLECTION_NAME)

What does this code do?

This sets up the foundation for your vector store.

  • COLLECTION_NAME defines a constant so every function references the same collection.
  • get_chroma_client() creates a PersistentClient that saves data to a ./chroma_data folder on disk. Data persists between script runs.
  • get_or_create_collection() retrieves the collection if it already exists, or creates a new one. This makes the code safe to run multiple times.
  • Save db.py.

Why a PersistentClient?

ChromaDB offers in-memory and persistent modes. The in-memory client loses all data when your program ends. The PersistentClient writes to disk automatically, so your ingested documents survive between sessions.

Build and run the ingestion pipeline

The ingestion pipeline reads a document, chunks it using your chunker from Step 2, then stores the chunks with their metadata in ChromaDB. ChromaDB handles embedding generation automatically using its built-in model.

  • In db.py, add the following function below get_or_create_collection():
def ingest_chunks(chunks, collection=None):
    """Ingest a list of Chunk objects into ChromaDB."""
    if collection is None:
        collection = get_or_create_collection()

    documents = [chunk.text for chunk in chunks]
    metadatas = [
        {
            "doc_name": chunk.doc_name,
            "section_id": chunk.section_id,
            "section_title": chunk.section_title,
            "chunk_type": chunk.chunk_type,
        }
        for chunk in chunks
    ]
    ids = [f"{chunk.doc_name}_{chunk.section_id}" for chunk in chunks]

    collection.upsert(documents=documents, metadatas=metadatas, ids=ids)

    return len(documents)

What does this code do?

This function takes the Chunk objects your chunker produces and stores them in ChromaDB.

  • documents is the list of text content that ChromaDB will embed and index for similarity search.
  • metadatas attaches structured data (document name, section ID, title, type) to each chunk for later filtering and citation.
  • ids gives each chunk a unique identifier combining the document name and section ID.
  • upsert() inserts new chunks or updates existing ones. This makes re-ingestion safe without creating duplicates.
  • Save db.py.
  • Test the ingestion pipeline by running this command in your terminal:
python << 'EOF'
from chunker import chunk_legal_document
from db import ingest_chunks

text = open("sample_policy.txt").read()
chunks = chunk_legal_document(text, "data_privacy_policy")
count = ingest_chunks(chunks)
print(f"Ingested {count} chunks into ChromaDB")
EOF

You should see output like Ingested 20 chunks into ChromaDB (the exact number depends on how many sections the chunker finds).

First run takes longer

The first time you run this, ChromaDB downloads its default embedding model locally. This happens automatically and only needs to happen once. Subsequent runs will be much faster.

Seeing an import error?

Make sure you are running the command from the same folder where chunker.py, db.py, and sample_policy.txt all live. Python needs to find the modules in the current directory.

Also confirm your virtual environment is still active. You should see its name in your terminal prompt.

Help me fix my ChromaDB ingestion error.

  • Confirm the chroma_data/ directory was created by running:
ls chroma_data

You should see database files listed inside the folder.

Verify ingestion with a test query

The whole point of a vector store is retrieval. You need a function that takes a question, searches ChromaDB for the most relevant chunks, and returns them with their metadata.

  • In db.py, add the following function below ingest_chunks():
def query_documents(question: str, n_results: int = 5, collection=None):
    """Query ChromaDB for relevant document chunks."""
    if collection is None:
        collection = get_or_create_collection()

    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],
                "distance": results["distances"][0][i],
            }
        )

    return retrieved

What does this code do?

This function is the retrieval half of your RAG pipeline.

  • query_texts accepts a list of questions. ChromaDB embeds the question and compares it against all stored chunk embeddings.
  • n_results controls how many matching chunks to return. Fewer results means more focused context for the LLM.
  • distance indicates how far each result is from your query in embedding space. Lower distances mean higher relevance.
  • Save db.py.
  • Run a test query to confirm retrieval works:
python << 'EOF'
from db import query_documents

results = query_documents("What are the data retention requirements?")
for r in results[:3]:
    meta = r["metadata"]
    print(f"[Section {meta['section_id']}] {meta['section_title']}")
    print(f"  Type: {meta['chunk_type']}, Distance: {r['distance']:.4f}")
    print()
EOF

You should see 3 results printed, each showing a section ID, title, chunk type, and distance score. The sections about data storage and retention should appear with the lowest distances.

Getting zero results?

If no results come back, the ingestion from the previous substep may not have completed successfully. Re-run the ingestion command first, then try the query again.

If you see a chroma_data/ folder but it seems empty, delete it and re-run ingestion from scratch.

Help me debug my ChromaDB query returning no results.

✔️ Awesome, I've got everything!

Great. Double check you have saved db.py with all three functions.

ⓧ I'd like to double check the full code

Here is the complete db.py file for comparison:

import chromadb

COLLECTION_NAME = "legal_documents"


def get_chroma_client():
    """Create a persistent ChromaDB client."""
    return chromadb.PersistentClient(path="./chroma_data")


def get_or_create_collection(client=None):
    """Get or create the legal documents collection."""
    if client is None:
        client = get_chroma_client()
    return client.get_or_create_collection(name=COLLECTION_NAME)


def ingest_chunks(chunks, collection=None):
    """Ingest a list of Chunk objects into ChromaDB."""
    if collection is None:
        collection = get_or_create_collection()

    documents = [chunk.text for chunk in chunks]
    metadatas = [
        {
            "doc_name": chunk.doc_name,
            "section_id": chunk.section_id,
            "section_title": chunk.section_title,
            "chunk_type": chunk.chunk_type,
        }
        for chunk in chunks
    ]
    ids = [f"{chunk.doc_name}_{chunk.section_id}" for chunk in chunks]

    collection.upsert(documents=documents, metadatas=metadatas, ids=ids)

    return len(documents)


def query_documents(question: str, n_results: int = 5, collection=None):
    """Query ChromaDB for relevant document chunks."""
    if collection is None:
        collection = get_or_create_collection()

    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],
                "distance": results["distances"][0][i],
            }
        )

    return retrieved

Your vector store is loaded and retrieval is working. Next up, you will build the Q&A engine that takes these retrieved chunks and generates cited answers using Claude.

Build the Q&A Engine with Citation Enforcement

Your document chunks are stored in ChromaDB and ready to be queried. The missing piece is a generation layer that takes those retrieved chunks and produces answers with verifiable citations.

In this step, you will build the Q&A engine that connects retrieval to generation. Every answer must trace back to a specific document section, and the system must refuse to answer when the documents lack the information.

In this step, get ready to:
  • Create a context formatting function that prepares retrieved chunks for the prompt.
  • Build a system prompt with strict citation enforcement rules.
  • Wire retrieval and generation into a complete Q&A function and test it.
Create the context formatting function

The query_documents() function in db.py returns raw chunks from ChromaDB. Before passing these to Claude, you need to format them with clear source labels so the model can cite them precisely.

  • Create a new file called qa_engine.py in your project folder.
  • Add the following imports, client helper, and formatting function:
import os
from anthropic import Anthropic


def get_anthropic_client():
    """Create an Anthropic client."""
    return Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))


def format_context(retrieved_chunks: list[dict]) -> str:
    """Format retrieved chunks into context for the prompt."""
    context_parts = []
    for chunk in retrieved_chunks:
        meta = chunk["metadata"]
        context_parts.append(
            f"[Doc: {meta['doc_name']}, Section: {meta['section_id']}] "
            f"({meta['section_title']})\n{chunk['text']}"
        )
    return "\n\n---\n\n".join(context_parts)

What does this code do?

get_anthropic_client() creates a Claude API client using your ANTHROPIC_API_KEY environment variable.

format_context() takes the raw chunks returned by query_documents() and formats each one with a [Doc: X, Section: Y] tag. This tag format is what the model will reference in its citations. Each chunk is separated by --- to clearly delineate section boundaries.

  • Save qa_engine.py.

Import error when you test later?

Make sure qa_engine.py is in the same folder as db.py and chunker.py. If you placed it in a subfolder, Python will not find the imports.

help me fix my qa_engine.py imports

Build the citation enforcement system prompt

In legal Q&A, every claim must trace to a source. A generic prompt would let the model fill gaps with guesses. The system prompt locks down behaviour so the model either cites or refuses.

  • In qa_engine.py, add the following SYSTEM_PROMPT constant between the imports and the get_anthropic_client() function:
SYSTEM_PROMPT = """You are a Legal & Compliance Q&A assistant. You answer questions ONLY based on the provided document context.

STRICT RULES:
1. Every factual assertion must cite its source as [Doc: <document_name>, Section: <section_id>].
2. If the provided context does not contain enough information to answer the question, respond with: "I cannot answer this question based on the provided documents. The available context does not contain information about [topic]."
3. Never infer, extrapolate, or provide legal advice beyond what is explicitly stated in the documents.
4. Distinguish between obligations ("shall", "must") and permissions ("may") precisely.
5. If multiple sections are relevant, cite each one separately.
6. Use temperature 0 reasoning - be precise, not creative."""

Why these rules matter for legal RAG

Rule 1 forces every claim to have a traceable source. Rule 2 is the safe refusal behaviour. Without it, the model would fabricate plausible-sounding legal advice when the documents lack an answer.

Rule 4 distinguishes between legal language that creates obligations ("shall", "must") and language that grants permissions ("may"). Confusing these in a compliance context can have serious consequences.

  • Save qa_engine.py.

Why citation enforcement in the prompt?

Production legal RAG systems enforce citations at the prompt level rather than in post-processing. If the model generates a claim without a source, it is already too late to verify. Instructing the model to cite as it writes produces answers that are verifiable by design.

Wire retrieval and generation into ask_question()

Now you will create the main function that takes a question, formats the retrieved context, sends it to Claude with the system prompt, and returns a structured response with the answer and its sources.

  • Add the following ask_question() function at the bottom of qa_engine.py:
def ask_question(question: str, retrieved_chunks: list[dict]) -> dict:
    """Generate a cited answer from retrieved chunks."""
    client = get_anthropic_client()
    context = format_context(retrieved_chunks)

    user_message = f"""Context from legal documents:

{context}

---

Question: {question}

Provide a precise answer citing specific document sections. If the context does not contain the answer, say so explicitly."""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        temperature=0,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_message}],
    )

    answer_text = response.content[0].text

    return {
        "answer": answer_text,
        "sources": [
            {
                "doc_name": chunk["metadata"]["doc_name"],
                "section_id": chunk["metadata"]["section_id"],
                "section_title": chunk["metadata"]["section_title"],
            }
            for chunk in retrieved_chunks
        ],
        "model": "claude-sonnet-4-20250514",
    }

What does ask_question() do?

  • format_context() prepares the retrieved chunks with citation tags.
  • The user message combines the formatted context with the question.
  • client.messages.create() sends the request to claude-sonnet-4-20250514 with temperature=0 for deterministic, factual output.
  • The return dictionary includes the answer text, the source metadata for each retrieved chunk, and the model name.
  • Save qa_engine.py.
  • Test the full Q&A pipeline by running this command in your terminal:
python -c "
from db import query_documents
from qa_engine import ask_question

chunks = query_documents('What is the data retention period?')
result = ask_question('What is the data retention period?', chunks)
print(result['answer'])
"

You should see a response that cites specific sections from the privacy policy, such as [Doc: sample_policy.txt, Section: 4.3]. The answer will reference the 3-year default retention period from Section 4.3 of the policy.

Seeing an authentication error?

Make sure your ANTHROPIC_API_KEY environment variable is still set in this terminal session. If you opened a new terminal, you will need to export it again.

help me fix my Anthropic API key setup

✔️ Awesome, I've got everything!

Great. Double check you have saved qa_engine.py and confirmed that your test query returned a cited answer.

ⓧ I'd like to double check the full code

import os
from anthropic import Anthropic

SYSTEM_PROMPT = """You are a Legal & Compliance Q&A assistant. You answer questions ONLY based on the provided document context.

STRICT RULES:
1. Every factual assertion must cite its source as [Doc: <document_name>, Section: <section_id>].
2. If the provided context does not contain enough information to answer the question, respond with: "I cannot answer this question based on the provided documents. The available context does not contain information about [topic]."
3. Never infer, extrapolate, or provide legal advice beyond what is explicitly stated in the documents.
4. Distinguish between obligations ("shall", "must") and permissions ("may") precisely.
5. If multiple sections are relevant, cite each one separately.
6. Use temperature 0 reasoning - be precise, not creative."""


def get_anthropic_client():
    """Create an Anthropic client."""
    return Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))


def format_context(retrieved_chunks: list[dict]) -> str:
    """Format retrieved chunks into context for the prompt."""
    context_parts = []
    for chunk in retrieved_chunks:
        meta = chunk["metadata"]
        context_parts.append(
            f"[Doc: {meta['doc_name']}, Section: {meta['section_id']}] "
            f"({meta['section_title']})\n{chunk['text']}"
        )
    return "\n\n---\n\n".join(context_parts)


def ask_question(question: str, retrieved_chunks: list[dict]) -> dict:
    """Generate a cited answer from retrieved chunks."""
    client = get_anthropic_client()
    context = format_context(retrieved_chunks)

    user_message = f"""Context from legal documents:

{context}

---

Question: {question}

Provide a precise answer citing specific document sections. If the context does not contain the answer, say so explicitly."""

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=1024,
        temperature=0,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_message}],
    )

    answer_text = response.content[0].text

    return {
        "answer": answer_text,
        "sources": [
            {
                "doc_name": chunk["metadata"]["doc_name"],
                "section_id": chunk["metadata"]["section_id"],
                "section_title": chunk["metadata"]["section_title"],
            }
            for chunk in retrieved_chunks
        ],
        "model": "claude-sonnet-4-20250514",
    }

Your Q&A engine is wired up and producing cited answers. Next up, you will wrap this into a proper API with FastAPI so you can send questions over HTTP.

Expose the API with FastAPI

Your RAG pipeline is complete. The chunker, vector store, and Q&A engine all work together to produce cited answers from legal documents.

Right now, the only way to use your system is by running Python scripts directly. In this step, you will wrap everything in a FastAPI web service so any client can submit documents and ask questions over HTTP.

In this step, get ready to:
  • Create a FastAPI app with Pydantic request and response models.
  • Build document ingestion and Q&A endpoints with error handling.
  • Run the API and test both endpoints using FastAPI's interactive docs.
Set up the FastAPI app and Pydantic models

FastAPI needs an app instance to register routes against, and Pydantic models to validate incoming requests and shape outgoing responses.

  • Create a new file called main.py in your project folder.
  • Add the imports and app initialization by pasting this code:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from chunker import chunk_legal_document
from db import ingest_chunks, query_documents
from qa_engine import ask_question

app = FastAPI(
    title="Legal & Compliance Q&A API",
    description="RAG-powered Q&A service for legal and compliance documents with citation enforcement.",
    version="1.0.0",
)

What does this code do?

The first two imports bring in FastAPI itself and Pydantic's BaseModel for data validation.

The local imports pull in the three modules you built in earlier steps: the chunker, the database layer, and the Q&A engine. The FastAPI() call creates your app instance with metadata that appears in the auto-generated docs.

  • Below the app initialization, add the Pydantic models that define what each endpoint accepts and returns:
class IngestRequest(BaseModel):
    document_name: str
    content: str


class IngestResponse(BaseModel):
    message: str
    chunks_created: int


class AskRequest(BaseModel):
    question: str
    n_results: int = 5


class Source(BaseModel):
    doc_name: str
    section_id: str
    section_title: str


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

What do these models do?

  • IngestRequest expects a document name and its full text content.
  • IngestResponse confirms how many chunks were created from the document.
  • AskRequest takes a question string and an optional n_results parameter (defaults to 5 retrieved chunks).
  • Source structures each citation with a document name, section ID, and title.
  • AskResponse bundles the generated answer, its source citations, and the model used.
  • Save main.py.

Seeing a red underline on the imports?

If your editor shows an error on the local imports (from chunker import ...), make sure chunker.py, db.py, and qa_engine.py are in the same folder as main.py.

help me fix import errors in my FastAPI project

Add the document ingestion and Q&A endpoints

With the models defined, you can now write the endpoint functions that wire your RAG pipeline into HTTP routes.

  • Below the Pydantic models in main.py, add the /ingest endpoint:
@app.post("/ingest", response_model=IngestResponse)
def ingest_document(request: IngestRequest):
    """Ingest a legal document into the vector store."""
    chunks = chunk_legal_document(request.content, request.document_name)

    if not chunks:
        raise HTTPException(
            status_code=400,
            detail="No sections found in document. Ensure the document uses numbered sections (e.g., '1. Introduction').",
        )

    count = ingest_chunks(chunks)

    return IngestResponse(
        message=f"Successfully ingested '{request.document_name}'.",
        chunks_created=count,
    )

What does this endpoint do?

When a client POSTs to /ingest, this function chunks the document by legal section structure and stores the embeddings in ChromaDB.

If the chunker finds no numbered sections, it returns a 400 error with a helpful message. Otherwise it confirms how many chunks were stored.

  • Below the /ingest endpoint, add the /ask endpoint:
@app.post("/ask", response_model=AskResponse)
def ask(request: AskRequest):
    """Ask a question about ingested legal documents."""
    retrieved = query_documents(request.question, n_results=request.n_results)

    if not retrieved:
        raise HTTPException(
            status_code=404,
            detail="No documents found. Please ingest documents first using /ingest.",
        )

    result = ask_question(request.question, retrieved)

    return AskResponse(
        answer=result["answer"],
        sources=[Source(**s) for s in result["sources"]],
        model=result["model"],
    )

What does this endpoint do?

The /ask endpoint retrieves relevant chunks from ChromaDB, passes them to the Q&A engine, and returns Claude's cited answer along with structured source metadata.

If no documents have been ingested yet, it returns a 404 error guiding the user to call /ingest first.

  • Save main.py.

Seeing a syntax error on save?

Check that each raise HTTPException(...) call has matching parentheses. The detail string spans multiple lines inside the parentheses, so a missing closing paren is a common mistake.

help me fix a syntax error in my FastAPI endpoint

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py before moving on.

ⓧ I'd like to double check the full code

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from chunker import chunk_legal_document
from db import ingest_chunks, query_documents
from qa_engine import ask_question

app = FastAPI(
    title="Legal & Compliance Q&A API",
    description="RAG-powered Q&A service for legal and compliance documents with citation enforcement.",
    version="1.0.0",
)


class IngestRequest(BaseModel):
    document_name: str
    content: str


class IngestResponse(BaseModel):
    message: str
    chunks_created: int


class AskRequest(BaseModel):
    question: str
    n_results: int = 5


class Source(BaseModel):
    doc_name: str
    section_id: str
    section_title: str


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


@app.post("/ingest", response_model=IngestResponse)
def ingest_document(request: IngestRequest):
    """Ingest a legal document into the vector store."""
    chunks = chunk_legal_document(request.content, request.document_name)

    if not chunks:
        raise HTTPException(
            status_code=400,
            detail="No sections found in document. Ensure the document uses numbered sections (e.g., '1. Introduction').",
        )

    count = ingest_chunks(chunks)

    return IngestResponse(
        message=f"Successfully ingested '{request.document_name}'.",
        chunks_created=count,
    )


@app.post("/ask", response_model=AskResponse)
def ask(request: AskRequest):
    """Ask a question about ingested legal documents."""
    retrieved = query_documents(request.question, n_results=request.n_results)

    if not retrieved:
        raise HTTPException(
            status_code=404,
            detail="No documents found. Please ingest documents first using /ingest.",
        )

    result = ask_question(request.question, retrieved)

    return AskResponse(
        answer=result["answer"],
        sources=[Source(**s) for s in result["sources"]],
        model=result["model"],
    )
Run the API and test both endpoints

FastAPI includes a built-in development server and auto-generated interactive documentation. You can test every endpoint directly in your browser without writing a separate client.

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

What does this command do?

The fastapi dev command starts a Uvicorn server with auto-reload enabled. Any time you save a file, the server restarts automatically.

It serves your API at http://127.0.0.1:8000 and generates interactive Swagger docs at /docs.

  • You should see output indicating the server is running, including the URL http://127.0.0.1:8000.
  • Open your browser and go to http://127.0.0.1:8000/docs.

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

  • Click on POST /ask to expand it.
  • Click Try it out.
  • Replace the request body with this JSON:
{
  "question": "What is the data breach notification timeline?",
  "n_results": 5
}
  • Click Execute.

You should see a 200 response with a JSON body containing an answer field (with citations like [Doc: ..., Section: ...]), a sources array, and a model field.

Why test /ask first?

You already ingested sample_policy.txt into ChromaDB in the previous step. The /ask endpoint queries that existing data, so it works immediately without needing to call /ingest first.

  • Now test the /ingest endpoint. Click on POST /ingest to expand it.
  • Click Try it out.
  • Replace the request body with this JSON:
{
  "document_name": "test_doc",
  "content": "1. Overview\nThis is a test document for ingestion.\n\n2. Requirements\n2.1 All systems must be tested quarterly."
}
  • Click Execute.

You should see a 200 response with a message confirming ingestion and a chunks_created count.

Getting a 500 Internal Server Error on /ask?

This usually means your ANTHROPIC_API_KEY environment variable is not set in the terminal session where you ran fastapi dev. Stop the server with Ctrl+C, re-export your key, and restart the server.

help me fix a 500 error when calling my /ask endpoint

Your Legal and Compliance Q&A API is live. You have a fully functional RAG service that ingests documents, retrieves relevant sections, and generates cited answers. Next up, you will push this further with a faithfulness evaluator in the Secret Mission.

Secret mission

Build a Faithfulness Evaluator

Your Q&A API generates cited answers, but how do you know those citations actually support the claims they're attached to? In this secret mission, you'll build a /evaluate endpoint that programmatically checks citation quality by verifying each reference exists in the vector store and using Claude to score whether the source passage truly backs its claim.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing infrastructure costs.

Resources you used:

  • Project folder containing chunker.py, db.py, qa_engine.py, main.py, and sample_policy.txt.
  • ChromaDB persistent data stored in the chroma_data/ directory.
  • Python virtual environment with installed packages (FastAPI, ChromaDB, Anthropic SDK).
  • ANTHROPIC_API_KEY environment variable set in your terminal session.

Keep everything running

No action needed. Choose this if you're still actively building or plan to extend the API further.

  • Your project files stay on disk and you can restart the API anytime with fastapi dev.
  • The chroma_data/ folder retains your ingested documents so you don't need to re-ingest.
  • Your ANTHROPIC_API_KEY environment variable only persists in your current terminal session. You'll need to export it again if you open a new terminal.

Pause - I'll come back to this later

Shut down the running dev server and deactivate your virtual environment to free up system resources.

  • Stop the FastAPI dev server by pressing Ctrl+C in the terminal where it's running.
  • Deactivate your virtual environment by running:
deactivate
  • Your project files and chroma_data/ directory remain on disk. When you're ready to resume, reactivate the virtual environment and re-export your API key.

Delete - I don't want to use this again

Remove all project resources and start fresh. This deletes your code files, the ChromaDB data, and the virtual environment.

  • Stop the FastAPI dev server by pressing Ctrl+C in the terminal.
  • Deactivate your virtual environment by running:
deactivate
  • Navigate up to the parent directory (your Desktop) and remove the entire project folder by running:

macOS

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

Windows

cd ~\Desktop
rmdir /s /q legal-qa-api
  • Unset the API key environment variable in your current terminal session by running:

macOS

unset ANTHROPIC_API_KEY

Windows

Remove-Item Env:ANTHROPIC_API_KEY
  • This removes the variable from your current session only. If you added it to a shell profile file (like .bashrc or .zshrc), remove that line manually.

Nice Work!

Nice Work!

You just built a Legal & Compliance Q&A API that answers questions from real documents with verifiable source citations. This is the kind of production-grade RAG system that high-stakes industries demand.

You've learned how to:

  • Built a legal-aware document chunker that splits compliance text by section structure instead of arbitrary token counts.
  • Created a full RAG pipeline with ChromaDB for vector storage and Claude for citation-enforced answer generation.
  • Exposed the system as a FastAPI service with document ingestion and Q&A endpoints, complete with safe refusal behaviour when documents lack sufficient context.
  • Added a faithfulness evaluator that programmatically checks whether each citation actually supports its claim.

Ready to quiz yourself?