Build a RAG Application with Python and Ollama

Imagine: You upload your course handbook and ask, “How many days do I have to submit the assignment?”

• Retrieval: Find the handbook paragraphs about submission deadlines.
• Generation: Ask the model to answer using those paragraphs.

ollama pull llama3.2
ollama pull nomic-embed-text

ollama-rag/
├── app.py
├── requirements.txt
└── templates/
    └── index.html

# Windows
python -m venv .venv
.venv\Scripts\activate

# macOS / Linux
python3 -m venv .venv
source .venv/bin/activate

Flask>=3.0,<4.0
requests>=2.31,<3.0
numpy>=1.26,<3.0
pypdf>=5.0,<7.0

python -m pip install -r requirements.txt

from pathlib import Path
import io

import numpy as np
import requests
from flask import Flask, render_template, request
from pypdf import PdfReader

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 10 * 1024 * 1024  # 10 MB upload limit

OLLAMA_URL = "http://localhost:11434"
CHAT_MODEL = "llama3.2"
EMBED_MODEL = "nomic-embed-text"

# Starter storage: this is cleared whenever the app restarts.
DOCUMENT_CHUNKS = []
CHUNK_VECTORS = None


def extract_text(filename, file_bytes):
    """Extract text from a supported uploaded file."""
    suffix = Path(filename).suffix.lower()

    if suffix == ".txt":
        return file_bytes.decode("utf-8", errors="replace")

    if suffix == ".pdf":
        reader = PdfReader(io.BytesIO(file_bytes))
        pages = [page.extract_text() or "" for page in reader.pages]
        return "\n".join(pages)

    raise ValueError("Only .txt and .pdf files are supported.")


def split_into_chunks(text, chunk_size=900, overlap=150):
    """Split text into overlapping character chunks."""
    text = " ".join(text.split())
    if not text:
        return []

    chunks = []
    start = 0

    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end].strip()

        if chunk:
            chunks.append(chunk)

        if end >= len(text):
            break

        start = end - overlap

    return chunks


def embed_texts(texts):
    """Ask Ollama to create one embedding vector per text."""
    if not texts:
        return []

    response = requests.post(
        f"{OLLAMA_URL}/api/embed",
        json={"model": EMBED_MODEL, "input": texts},
        timeout=120,
    )
    response.raise_for_status()
    return response.json()["embeddings"]


def add_document(filename, file_bytes):
    """Extract, chunk, and index one uploaded document."""
    global CHUNK_VECTORS

    text = extract_text(filename, file_bytes)
    chunks = split_into_chunks(text)

    if not chunks:
        raise ValueError(
            f"No readable text found in {filename}. "
            "If it is a scanned PDF, OCR is needed."
        )

    vectors = embed_texts(chunks)
    DOCUMENT_CHUNKS.extend(
        {"source": filename, "text": chunk} for chunk in chunks
    )

    new_vectors = np.asarray(vectors, dtype=np.float32)
    if CHUNK_VECTORS is None:
        CHUNK_VECTORS = new_vectors
    else:
        CHUNK_VECTORS = np.vstack([CHUNK_VECTORS, new_vectors])

    return len(chunks)


def retrieve(question, top_k=4):
    """Return the most similar document chunks for a question."""
    if not DOCUMENT_CHUNKS or CHUNK_VECTORS is None:
        return []

    question_vector = np.asarray(
        embed_texts([question])[0], dtype=np.float32
    )

    # Cosine similarity: compare directions, not vector lengths.
    vector_norms = np.linalg.norm(CHUNK_VECTORS, axis=1)
    question_norm = np.linalg.norm(question_vector)

    safe_vector_norms = np.maximum(vector_norms, 1e-12)
    safe_question_norm = max(float(question_norm), 1e-12)

    scores = (CHUNK_VECTORS @ question_vector) / (
        safe_vector_norms * safe_question_norm
    )

    best_indices = np.argsort(scores)[::-1][:top_k]
    return [DOCUMENT_CHUNKS[int(i)] for i in best_indices]


def generate_answer(question, chunks):
    """Generate an answer using retrieved context."""
    if not chunks:
        return "Please upload a readable PDF or TXT document first."

    context = "\n\n".join(
        f"Source: {item['source']}\n{item['text']}"
        for item in chunks
    )

    prompt = f"""
You are a helpful document assistant.
Answer the question using only the CONTEXT below.
If the context does not contain the answer, say you could not find it
in the uploaded documents. Do not invent facts.
Mention the source filename when useful.

CONTEXT:
{context}

QUESTION:
{question}

ANSWER:
""".strip()

    response = requests.post(
        f"{OLLAMA_URL}/api/chat",
        json={
            "model": CHAT_MODEL,
            "messages": [{"role": "user", "content": prompt}],
            "stream": False,
        },
        timeout=180,
    )
    response.raise_for_status()
    return response.json()["message"]["content"].strip()


@app.route("/", methods=["GET", "POST"])
def index():
    answer = None
    sources = []
    message = None
    question = ""

    if request.method == "POST":
        action = request.form.get("action", "")

        try:
            if action == "upload":
                files = request.files.getlist("documents")
                files = [f for f in files if f and f.filename]

                if not files:
                    raise ValueError("Choose at least one PDF or TXT file.")

                added_chunks = 0
                uploaded_names = []

                for uploaded_file in files:
                    filename = Path(uploaded_file.filename).name
                    file_bytes = uploaded_file.read()

                    if not file_bytes:
                        continue

                    added_chunks += add_document(filename, file_bytes)
                    uploaded_names.append(filename)

                if not uploaded_names:
                    raise ValueError("The selected files were empty.")

                message = (
                    f"Indexed {added_chunks} text chunks from: "
                    + ", ".join(uploaded_names)
                )

            elif action == "ask":
                question = request.form.get("question", "").strip()

                if not question:
                    raise ValueError("Type a question first.")

                chunks = retrieve(question)
                answer = generate_answer(question, chunks)
                sources = list(dict.fromkeys(
                    item["source"] for item in chunks
                ))

            else:
                raise ValueError("Unknown form action.")

        except requests.ConnectionError:
            message = (
                "Could not connect to Ollama. Make sure Ollama is running "
                "and both required models are installed."
            )
        except requests.HTTPError as exc:
            detail = exc.response.text[:300] if exc.response is not None else ""
            message = f"Ollama returned an HTTP error: {detail}"
        except (ValueError, OSError) as exc:
            message = str(exc)
        except Exception as exc:
            app.logger.exception("Unexpected RAG error")
            message = f"Something went wrong: {exc}"

    return render_template(
        "index.html",
        answer=answer,
        sources=sources,
        message=message,
        question=question,
        chunk_count=len(DOCUMENT_CHUNKS),
    )


if __name__ == "__main__":
    # Local development only. Do not expose this debug server publicly.
    app.run(host="127.0.0.1", port=5000, debug=True)

<!doctype html>
<html lang="en">
<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1">
    <title>Local RAG Assistant</title>
    <style>
        body {
            max-width: 850px;
            margin: 40px auto;
            padding: 0 18px;
            font-family: system-ui, sans-serif;
            line-height: 1.55;
            background: #f5f7fb;
            color: #172033;
        }
        header, section {
            background: white;
            padding: 22px;
            margin-bottom: 18px;
            border-radius: 16px;
            box-shadow: 0 5px 20px #1720330d;
        }
        h1 { margin-bottom: 4px; }
        .muted { color: #65708a; }
        input, button {
            box-sizing: border-box;
            width: 100%;
            margin-top: 10px;
            padding: 12px;
            border-radius: 10px;
            font: inherit;
        }
        input { border: 1px solid #cbd2e1; }
        button {
            border: 0;
            background: #5b4bdb;
            color: white;
            font-weight: 700;
            cursor: pointer;
        }
        .notice {
            padding: 12px;
            border-radius: 10px;
            background: #fff4d6;
            overflow-wrap: anywhere;
        }
        .answer {
            white-space: pre-wrap;
            overflow-wrap: anywhere;
        }
    </style>
</head>
<body>
    <header>
        <h1>🦙 Local RAG Assistant</h1>
        <p class="muted">
            Upload your notes, then ask questions answered from those notes.
        </p>
        <strong>Indexed chunks in this session: {{ chunk_count }}</strong>
    </header>

    <section>
        <h2>📄 1. Upload documents</h2>
        <form method="post" enctype="multipart/form-data">
            <input type="hidden" name="action" value="upload">
            <input
                type="file"
                name="documents"
                accept=".pdf,.txt"
                multiple
                required
            >
            <button type="submit">Upload and index</button>
        </form>
    </section>

    <section>
        <h2>💬 2. Ask a question</h2>
        <form method="post">
            <input type="hidden" name="action" value="ask">
            <input
                type="text"
                name="question"
                placeholder="Ask something from your documents..."
                value="{{ question }}"
                required
            >
            <button type="submit">Find answer</button>
        </form>
    </section>

    {% if message %}
    <section class="notice">
        <strong>Notice:</strong> {{ message }}
    </section>
    {% endif %}

    {% if answer %}
    <section>
        <h2>✨ Answer</h2>
        <div class="answer">{{ answer }}</div>
        {% if sources %}
        <h3>📚 Retrieved from</h3>
        <ul>
            {% for source in sources %}
            <li>{{ source }}</li>
            {% endfor %}
        </ul>
        {% endif %}
    </section>
    {% endif %}
</body>
</html>

ollama list

python app.py