RAG Tutorial in Python: Build a RAG System and a RAG AI Agent, Step by Step

askmydocs/
├── data/                # put your PDF, .txt and .md files here
├── chroma_db/           # created automatically: the vector database
├── config.py            # settings
├── ingest.py            # Step 2: load, chunk and store
├── retriever.py         # Step 3: semantic search
├── rag.py               # Step 4: classic RAG with citations
├── agent.py             # Step 5: RAG agent with tools
├── app.py               # Step 6: Streamlit web app
├── rag_ollama.py        # optional: free local model
├── evaluate.py          # Step 7: retrieval quality
├── requirements.txt
├── .env                 # your API key (never commit this)
└── .gitignore

mkdir askmydocs && cd askmydocs
python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install anthropic chromadb pypdf python-dotenv streamlit
pip freeze > requirements.txt
mkdir data

# copy this file to .env and fill in your key
ANTHROPIC_API_KEY=your-anthropic-api-key
CLAUDE_MODEL=claude-opus-5-5

.env
.venv/
chroma_db/
__pycache__/

# Acme Ltd employee handbook

## Annual leave
Every full-time employee gets 24 days of paid annual leave per calendar year, plus public holidays. Part-time staff get leave in proportion to their hours. Up to 5 unused days can be carried over to the next year, and must be used before 31 March.

## Remote work
Employees may work from home up to 3 days a week after completing their probation period of 3 months. Remote days must be agreed with your manager in advance. Core hours are 10:00 to 15:00 in your local time zone.

## Travel expenses
Submit travel expense claims within 30 days through the finance portal, with photos of all receipts. Economy class is standard for flights under 6 hours. Daily meal allowance while travelling is 40 euros.

## Sick leave
If you are ill, inform your manager before 09:30. A doctor's note is required for absences longer than 3 consecutive days.

# Datajin product FAQ

## Free trial
Datajin offers a 14-day free trial with all Pro features. No credit card is needed to start the trial.

## Pricing and payment
The Pro plan costs 19 dollars per month or 190 dollars per year. We accept Visa, Mastercard, PayPal and bank transfer for annual plans.

## Refunds
If you cancel within 7 days of a payment, you get a full refund. Refunds are processed within 5 to 10 business days to the original payment method.

## Data privacy
Uploaded files are encrypted at rest and deleted automatically 30 days after you remove them from your workspace. We never use customer data to train AI models.

# config.py: every setting in one place
import os
from pathlib import Path

from dotenv import load_dotenv

BASE_DIR = Path(__file__).resolve().parent
load_dotenv(BASE_DIR / ".env")

DATA_DIR = BASE_DIR / "data"          # your PDFs, .txt and .md files
DB_DIR = BASE_DIR / "chroma_db"       # the vector database is saved here
COLLECTION = "askmydocs"

CHUNK_SIZE = 800       # characters per chunk
CHUNK_OVERLAP = 150    # characters shared by neighbouring chunks
TOP_K = 4              # how many chunks to retrieve per search

CLAUDE_MODEL = os.getenv("CLAUDE_MODEL", "claude-opus-5-5")

# ingest.py: load documents, split them into chunks and store them in ChromaDB
import hashlib
import sys
from pathlib import Path

import chromadb
from pypdf import PdfReader

import config

SUPPORTED = {".pdf", ".txt", ".md"}


def get_collection():
    """Open (or create) the persistent vector collection.
    ChromaDB embeds text with its default local model (all-MiniLM-L6-v2): free, no API key."""
    client = chromadb.PersistentClient(path=str(config.DB_DIR))
    return client.get_or_create_collection(config.COLLECTION, metadata={"hnsw:space": "cosine"})


def load_file(path: Path) -> list[dict]:
    """Return one record per page (PDF) or per file (TXT/MD)."""
    if path.suffix.lower() == ".pdf":
        reader = PdfReader(path)
        return [{"text": page.extract_text() or "", "source": path.name, "page": i}
                for i, page in enumerate(reader.pages, start=1)]
    return [{"text": path.read_text(encoding="utf-8", errors="ignore"), "source": path.name, "page": 1}]


def split_text(text: str, size: int = config.CHUNK_SIZE, overlap: int = config.CHUNK_OVERLAP) -> list[str]:
    """Split text into overlapping chunks, preferring to end each chunk at a full stop."""
    text = " ".join(text.split())                     # normalise spaces and line breaks
    chunks, start = [], 0
    while start < len(text):
        end = min(start + size, len(text))
        if end < len(text):
            cut = text.rfind(". ", start, end)
            if cut > start + size // 2:               # only if the sentence end is not too early
                end = cut + 1
        chunks.append(text[start:end].strip())
        if end >= len(text):
            break
        start = end - overlap
        space = text.find(" ", start, end)            # start the next chunk on a word boundary
        if space != -1:
            start = space + 1
    return [c for c in chunks if c]


def ingest_files(paths) -> int:
    """Chunk and index the given files. Re-indexing the same file replaces its old chunks."""
    collection = get_collection()
    total = 0
    for path in map(Path, paths):
        if path.suffix.lower() not in SUPPORTED:
            continue
        collection.delete(where={"source": path.name})          # remove the old version first
        ids, docs, metas = [], [], []
        for record in load_file(path):
            for n, chunk in enumerate(split_text(record["text"])):
                key = f'{record["source"]}|{record["page"]}|{n}|{chunk}'
                ids.append(hashlib.sha1(key.encode()).hexdigest())
                docs.append(chunk)
                metas.append({"source": record["source"], "page": record["page"]})
        for i in range(0, len(ids), 100):                         # add in batches
            collection.add(ids=ids[i:i + 100], documents=docs[i:i + 100], metadatas=metas[i:i + 100])
        print(f"  {path.name}: {len(ids)} chunks")
        total += len(ids)
    return total


def list_sources() -> dict[str, int]:
    """Return {file name: number of chunks} for everything in the knowledge base."""
    counts: dict[str, int] = {}
    for meta in get_collection().get(include=["metadatas"])["metadatas"]:
        counts[meta["source"]] = counts.get(meta["source"], 0) + 1
    return dict(sorted(counts.items()))


if __name__ == "__main__":
    folder = Path(sys.argv[1]) if len(sys.argv) > 1 else config.DATA_DIR
    files = [p for p in sorted(folder.rglob("*")) if p.suffix.lower() in SUPPORTED]
    print(f"Indexing {len(files)} files from {folder} ...")
    print(f"Done: {ingest_files(files)} chunks in the knowledge base.")

python ingest.py

# retriever.py: semantic search over the knowledge base
import config
from ingest import get_collection


def search(query: str, k: int = config.TOP_K, source: str | None = None) -> list[dict]:
    """Return the k chunks most similar in meaning to the query."""
    collection = get_collection()
    count = collection.count()
    if count == 0:
        return []
    result = collection.query(
        query_texts=[query],
        n_results=min(k, count),
        where={"source": source} if source else None,
    )
    return [
        {"text": doc, "source": meta["source"], "page": meta["page"], "score": round(1 - dist, 3)}
        for doc, meta, dist in zip(result["documents"][0], result["metadatas"][0], result["distances"][0])
    ]


if __name__ == "__main__":
    import sys
    for hit in search(" ".join(sys.argv[1:]) or "What is this document about?"):
        print(f'[{hit["score"]}] {hit["source"]} p.{hit["page"]}: {hit["text"][:120]}...')

python retriever.py "how many holidays do I get"

# rag.py: classic RAG = retrieve the best chunks, then let Claude answer from them
import anthropic

import config
from retriever import search

client = anthropic.Anthropic(timeout=60.0, max_retries=2)   # reads ANTHROPIC_API_KEY

SYSTEM = (
    "You answer questions using only the numbered sources inside <sources>. "
    "Cite every fact with its source number, like [1] or [2][3]. "
    "If the sources do not contain the answer, reply exactly: "
    "\"I could not find this in your documents.\" Do not use outside knowledge."
)


def build_prompt(question: str, chunks: list[dict]) -> str:
    context = "\n\n".join(
        f'[{i}] (file: {c["source"]}, page {c["page"]})\n{c["text"]}' for i, c in enumerate(chunks, start=1)
    )
    return f"<sources>\n{context}\n</sources>\n\nQuestion: {question}"


def answer(question: str, k: int = config.TOP_K) -> dict:
    chunks = search(question, k)
    if not chunks:
        return {"answer": "The knowledge base is empty. Add documents first.", "sources": []}
    try:
        response = client.messages.create(
            model=config.CLAUDE_MODEL,
            max_tokens=8000,
            output_config={"effort": "low"},          # quick, cheaper answers
            system=SYSTEM,
            messages=[{"role": "user", "content": build_prompt(question, chunks)}],
        )
    except anthropic.APIError as e:
        return {"answer": f"The AI service returned an error: {e}", "sources": chunks}
    if response.stop_reason == "refusal":
        return {"answer": "Sorry, I can't help with that request.", "sources": chunks}
    text = "".join(block.text for block in response.content if block.type == "text")
    return {"answer": text, "sources": chunks}


if __name__ == "__main__":
    import sys
    result = answer(" ".join(sys.argv[1:]) or "Summarise the documents.")
    print(result["answer"], "\n\nSources:")
    for i, s in enumerate(result["sources"], start=1):
        print(f'  [{i}] {s["source"]}, page {s["page"]} (score {s["score"]})')

python rag.py "How long does a refund take?"

# agent.py: a RAG agent. Claude decides when and what to search, and can search several times.
import anthropic
from anthropic import beta_tool

import config
from ingest import list_sources
from retriever import search

client = anthropic.Anthropic(timeout=90.0, max_retries=2)


@beta_tool
def search_documents(query: str, source: str = "") -> str:
    """Search the user's documents and return the most relevant passages with file name and page.

    Args:
        query: What to look for. Use specific keywords. If results are weak, rephrase and search again.
        source: Optional file name to search inside one document only, e.g. "hr_policy.md". Leave empty to search all files.
    """
    hits = search(query, k=config.TOP_K, source=source or None)
    if not hits:
        return "No matching passages found."
    return "\n\n".join(f'(file: {h["source"]}, page {h["page"]}, score {h["score"]})\n{h["text"]}' for h in hits)


@beta_tool
def list_documents() -> str:
    """List every document in the knowledge base and how many chunks it has."""
    sources = list_sources()
    if not sources:
        return "The knowledge base is empty."
    return "\n".join(f"{name}: {n} chunks" for name, n in sources.items())


AGENT_SYSTEM = (
    "You are a careful research assistant for the user's own documents. "
    "Always call search_documents before answering a question about the documents. "
    "For questions with several parts or comparisons, search once per part. "
    "Answer only from what the tools return and cite each fact as (file, page). "
    "If the documents do not contain the answer, say so clearly."
)


def run_agent(question: str, history: list[dict] | None = None) -> dict:
    """Run the agent loop and return the final answer plus the tool calls it made."""
    runner = client.beta.messages.tool_runner(
        model=config.CLAUDE_MODEL,
        max_tokens=8000,
        output_config={"effort": "medium"},
        system=AGENT_SYSTEM,
        tools=[search_documents, list_documents],
        messages=[*(history or []), {"role": "user", "content": question}],
        max_iterations=8,                              # safety limit on tool rounds
    )
    steps, final = [], None
    try:
        for message in runner:                         # one message per round
            final = message
            steps += [f"{b.name}({b.input})" for b in message.content if b.type == "tool_use"]
    except anthropic.APIError as e:
        return {"answer": f"The AI service returned an error: {e}", "steps": steps}
    if final is None or final.stop_reason == "refusal":
        return {"answer": "Sorry, I can't help with that request.", "steps": steps}
    text = "".join(b.text for b in final.content if b.type == "text")
    return {"answer": text, "steps": steps}


if __name__ == "__main__":
    import sys
    result = run_agent(" ".join(sys.argv[1:]) or "Which documents do I have, and what are they about?")
    print("Agent steps:", *result["steps"], sep="\n  ")
    print("\n" + result["answer"])

python agent.py "Compare the refund rules with the free trial. Do I need a credit card to try it?"

# app.py: the Streamlit web app. Run with: streamlit run app.py
from pathlib import Path

import streamlit as st

import config
from agent import run_agent
from ingest import ingest_files, list_sources
from rag import answer

st.set_page_config(page_title="AskMyDocs", page_icon="📚", layout="wide")
st.title("📚 AskMyDocs: chat with your documents")

with st.sidebar:
    st.header("1. Add documents")
    uploads = st.file_uploader("PDF, TXT or MD files", type=["pdf", "txt", "md"], accept_multiple_files=True)
    if uploads and st.button("Index files", type="primary"):
        config.DATA_DIR.mkdir(exist_ok=True)
        paths = []
        for f in uploads:
            path = config.DATA_DIR / Path(f.name).name        # keep only the file name (safe path)
            path.write_bytes(f.getbuffer())
            paths.append(path)
        with st.spinner("Reading, chunking and embedding..."):
            st.success(f"Indexed {ingest_files(paths)} chunks.")

    st.header("2. Choose a mode")
    mode = st.radio("Mode", ["RAG (fast, one search)", "RAG agent (multi-step)"], label_visibility="collapsed")

    st.header("Knowledge base")
    sources = list_sources()
    st.caption("  \n".join(f"• {name} ({n} chunks)" for name, n in sources.items()) or "Empty: add a file above.")
    if st.button("Clear chat"):
        st.session_state.messages = []

if "messages" not in st.session_state:
    st.session_state.messages = []

for m in st.session_state.messages:
    with st.chat_message(m["role"]):
        st.markdown(m["content"])

if question := st.chat_input("Ask a question about your documents"):
    st.chat_message("user").markdown(question)
    history = st.session_state.messages[-6:]                   # short memory for follow-up questions
    st.session_state.messages.append({"role": "user", "content": question})

    with st.chat_message("assistant"):
        with st.spinner("Searching your documents..."):
            if mode.startswith("RAG agent"):
                result = run_agent(question, history)
            else:
                result = answer(question)
        st.markdown(result["answer"])
        if result.get("sources"):
            with st.expander("Sources used"):
                for i, s in enumerate(result["sources"], start=1):
                    st.markdown(f'**[{i}] {s["source"]}, page {s["page"]}** (similarity {s["score"]})')
                    st.caption(s["text"][:400] + "...")
        if result.get("steps"):
            with st.expander("Agent steps"):
                st.code("\n".join(result["steps"]))
    st.session_state.messages.append({"role": "assistant", "content": result["answer"]})

streamlit run app.py

# rag_ollama.py: the same RAG pipeline, answered by a free local model through Ollama
# 1) install Ollama from ollama.com   2) ollama pull llama3.2   3) pip install ollama
from ollama import chat

from rag import SYSTEM, build_prompt
from retriever import search


def answer_local(question: str, model: str = "llama3.2") -> dict:
    chunks = search(question)
    if not chunks:
        return {"answer": "The knowledge base is empty. Add documents first.", "sources": []}
    response = chat(model=model, messages=[
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": build_prompt(question, chunks)},
    ])
    return {"answer": response.message.content, "sources": chunks}


if __name__ == "__main__":
    import sys
    print(answer_local(" ".join(sys.argv[1:]) or "Summarise the documents.")["answer"])

# install Ollama from ollama.com first
ollama pull llama3.2
pip install ollama
python rag_ollama.py "How many days can I work from home?"

# evaluate.py: measure retrieval quality before you tune prompts or models
from retriever import search

# (question, file that must appear in the top results). Write 10-20 of these for your own documents.
EVAL_SET = [
    ("How many days of annual leave do employees get?", "hr_policy.md"),
    ("Can I work from home, and how many days a week?", "hr_policy.md"),
    ("How do I claim travel expenses?", "hr_policy.md"),
    ("How long does a refund take?", "product_faq.md"),
    ("Which payment methods are accepted?", "product_faq.md"),
    ("Is there a free trial?", "product_faq.md"),
]


def hit_rate(k: int = 4) -> float:
    hits = 0
    for question, expected in EVAL_SET:
        found = [h["source"] for h in search(question, k=k)]
        ok = expected in found
        hits += ok
        print(f'{"✓" if ok else "✗"} {question}  ->  {found}')
    return hits / len(EVAL_SET)


if __name__ == "__main__":
    for k in (1, 4):
        print(f"\nHit rate @{k}: {hit_rate(k):.0%}\n")

python evaluate.py

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8501
CMD ["streamlit", "run", "app.py", "--server.port=8501", "--server.address=0.0.0.0"]

docker build -t askmydocs .
docker run -p 8501:8501 --env-file .env -v "$(pwd)/chroma_db:/app/chroma_db" askmydocs