What This Project Builds

This project combines every technique from this unit into one working system: a retrieval-augmented generation (RAG) application that answers questions using a folder of plain-text or Markdown notes as its knowledge source. It reads every note file from a folder, splits longer notes into smaller pieces before embedding them, builds a searchable in-memory index from those pieces (Lessons 1 through 3), and uses the Responses API to turn the most relevant retrieved pieces into a grounded, cited answer — the same fundamental pattern Unit 9, Lesson 2's file search tool implements internally, built here by hand so every step is visible and adjustable.

Step 1: Reading Notes From a Folder

import os

def load_notes_from_folder(folder_path: str) -> list[dict]:
    notes = []
    for filename in os.listdir(folder_path):
        if filename.endswith((".txt", ".md")):
            filepath = os.path.join(folder_path, filename)
            with open(filepath, "r", encoding="utf-8") as f:
                notes.append({"filename": filename, "text": f.read()})
    return notes

notes = load_notes_from_folder("my_notes")
print(f"Loaded {len(notes)} note files.")

This is the only part of the pipeline that touches the filesystem directly; everything downstream operates on the same {"filename": ..., "text": ...} shape regardless of where the notes originally came from, which is what makes it straightforward to later swap this step for a different source — a database table, a set of pages fetched from an internal wiki — without touching the retrieval logic itself.

Step 2: Splitting Longer Notes Into Chunks

A short note can be embedded as a single unit, but a longer note covering multiple topics produces a better search experience when split into smaller pieces first, so that a query about one specific topic can match the relevant piece directly rather than competing against everything else the note covers.

def chunk_by_paragraph(text: str, max_chunk_size: int = 500) -> list[str]:
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks = []
    current_chunk = ""
    for paragraph in paragraphs:
        if len(current_chunk) + len(paragraph) > max_chunk_size and current_chunk:
            chunks.append(current_chunk.strip())
            current_chunk = ""
        current_chunk += paragraph + "\n\n"
    if current_chunk.strip():
        chunks.append(current_chunk.strip())
    return chunks

Splitting along paragraph breaks, rather than at an arbitrary fixed character count, keeps each resulting chunk as a coherent unit of meaning — a chunk boundary falls between two paragraphs rather than in the middle of a sentence, which produces an embedding that represents one complete thought rather than a fragment of one thought stitched to a fragment of another. This step is what separates a RAG pipeline built for real, longer documents from the short-sentence examples used earlier in this unit to introduce the underlying concepts.

Step 3: Building the Searchable Index

def build_notes_index(client, notes: list[dict]) -> list[dict]:
    indexed_chunks = []
    for note in notes:
        chunks = chunk_by_paragraph(note["text"])
        if not chunks:
            continue
        response = client.embeddings.create(model="text-embedding-4", input=chunks)
        for chunk_text, data in zip(chunks, response.data):
            indexed_chunks.append({
                "text": chunk_text,
                "embedding": data.embedding,
                "source_filename": note["filename"],
            })
    return indexed_chunks

notes_index = build_notes_index(client, notes)
print(f"Indexed {len(notes_index)} chunks from {len(notes)} notes.")

This is a one-time indexing step, following the same indexing-versus-querying separation established across Lessons 1 through 3: it's run whenever the notes folder's contents change, entirely separate from any individual question asked against the index afterward. Retaining source_filename on every indexed chunk is what makes it possible to tell a user which specific note an answer's supporting evidence came from.

Step 4: Searching the Index

def cosine_similarity(vector_a: list[float], vector_b: list[float]) -> float:
    import math
    dot_product = sum(a * b for a, b in zip(vector_a, vector_b))
    magnitude_a = math.sqrt(sum(a * a for a in vector_a))
    magnitude_b = math.sqrt(sum(b * b for b in vector_b))
    if magnitude_a == 0 or magnitude_b == 0:
        return 0.0
    return dot_product / (magnitude_a * magnitude_b)

def search_notes(client, query: str, notes_index: list[dict], top_k: int = 3) -> list[dict]:
    query_vector = client.embeddings.create(model="text-embedding-4", input=query).data[0].embedding
    scored = [
        {**chunk, "score": cosine_similarity(query_vector, chunk["embedding"])}
        for chunk in notes_index
    ]
    scored.sort(key=lambda c: c["score"], reverse=True)
    return scored[:top_k]

results = search_notes(client, "what did I decide about the project deadline", notes_index)
for result in results:
    print(f"{result['score']:.3f} — {result['source_filename']}: {result['text'][:80]}...")

This is Lesson 3's semantic_search() pattern applied directly, with source_filename carried through so each result can be traced back to the note it came from.

Step 5: Synthesizing a Grounded, Cited Answer

The retrieved chunks are raw material for an answer, not the answer itself. The final step passes them to the Responses API with explicit instructions to answer only from the retrieved content and to cite which notes it drew from — the same grounding discipline Unit 7, Lesson 5's document question-answering project and Unit 9, Lesson 2's file-search-backed answering both established.

from pydantic import BaseModel
from enum import Enum

class AnswerConfidence(str, Enum):
    ANSWERED = "answered"
    PARTIALLY_ANSWERED = "partially_answered"
    NOT_FOUND = "not_found"

class NotesAnswer(BaseModel):
    answer: str
    confidence: AnswerConfidence
    source_filenames: list[str]

    model_config = {"extra": "forbid"}

def answer_from_notes(client, query: str, notes_index: list[dict]) -> NotesAnswer:
    matched_chunks = search_notes(client, query, notes_index)
    context = "\n\n".join(f"[{chunk['source_filename']}] {chunk['text']}" for chunk in matched_chunks)

    response = client.responses.parse(
        model="gpt-5.6-terra",
        instructions=(
            "Answer the question using only the note content provided below. "
            "Cite which files the answer draws from in source_filenames. "
            "If the notes fully answer the question, set confidence to 'answered'. "
            "If they partially address it, set confidence to 'partially_answered'. "
            "If they don't address it at all, set confidence to 'not_found' and don't guess."
        ),
        input=f"Note content:\n{context}\n\nQuestion: {query}",
        text_format=NotesAnswer,
    )
    return response.output_parsed

result = answer_from_notes(client, "what did I decide about the project deadline", notes_index)
print(f"[{result.confidence.value}] {result.answer}")
print(f"Sources: {', '.join(result.source_filenames)}")

The AnswerConfidence enum follows the same tiered-confidence pattern used throughout this course (Unit 6, Unit 7 Lesson 5, Unit 9 Lesson 2), giving the model an honest way to signal that the retrieved notes only partially address the question, or don't cover it at all, rather than fabricating a confident-sounding answer from irrelevant retrieved content — a real risk for any personal notes collection, where many questions may simply never have been written down anywhere.

Step 6: Keeping the Index in Sync With the Folder

Notes get added, edited, and deleted over time, and a stale index silently produces answers based on outdated or missing content. A practical version of this project needs a way to detect that the underlying folder has changed and refresh the index accordingly.

import hashlib

def compute_folder_signature(notes: list[dict]) -> str:
    combined = "".join(f"{note['filename']}:{note['text']}" for note in sorted(notes, key=lambda n: n["filename"]))
    return hashlib.sha256(combined.encode("utf-8")).hexdigest()

def refresh_index_if_changed(client, folder_path: str, current_index: list[dict], last_signature: str | None) -> tuple[list[dict], str]:
    notes = load_notes_from_folder(folder_path)
    new_signature = compute_folder_signature(notes)
    if new_signature == last_signature:
        return current_index, last_signature
    print("Notes folder has changed — rebuilding the index.")
    return build_notes_index(client, notes), new_signature

Computing a simple signature over the folder's combined contents and comparing it against the signature from the last time the index was built is a lightweight way to detect a change without having to track individual file modification times or diff file contents directly — if the signature differs at all, something in the folder changed, and the index is rebuilt from scratch rather than trying to update it incrementally, which is a reasonable trade-off for a small personal notes collection where a full rebuild is fast and infrequent.

Step 7: Testing the Pipeline Without Real API Calls

Following this course's dependency-injection testing pattern, the chunking, filtering, and ranking logic can all be tested independently of real embedding or model calls.

def test_chunk_by_paragraph_splits_on_boundaries():
    text = "First paragraph here.\n\nSecond paragraph here.\n\nThird paragraph here."
    chunks = chunk_by_paragraph(text, max_chunk_size=30)
    assert len(chunks) >= 2
    assert all(chunk.strip() for chunk in chunks)
    print("PASS: chunk_by_paragraph splits long text into multiple non-empty chunks")

def test_search_notes_ranks_by_similarity():
    fake_index = [
        {"text": "closely related note", "embedding": [1.0, 0.0], "source_filename": "a.md"},
        {"text": "unrelated note", "embedding": [0.0, 1.0], "source_filename": "b.md"},
    ]
    query_vector = [0.9, 0.1]
    scored = [{**chunk, "score": cosine_similarity(query_vector, chunk["embedding"])} for chunk in fake_index]
    scored.sort(key=lambda c: c["score"], reverse=True)
    assert scored[0]["source_filename"] == "a.md"
    print("PASS: search correctly ranks the more similar chunk first")

test_chunk_by_paragraph_splits_on_boundaries()
test_search_notes_ranks_by_similarity()

Testing chunking and ranking logic with small, controlled inputs, rather than real notes and real embedding calls, verifies the mechanical correctness of the pipeline quickly and without cost — reserving real embedding and model calls for a smaller set of end-to-end tests confirming the full pipeline produces sensible answers against actual notes and actual questions.

Troubleshooting Checklist

1. Are answers coming back not_found for questions the notes should cover? Check whether relevant content is being split awkwardly across chunk boundaries, or whether top_k in search_notes() is too small to surface the right chunk.
2. Are answers citing the wrong note file? Confirm source_filename is attached correctly during indexing and survives unchanged through search and answer synthesis.
3. Is the index out of sync with recent edits to the notes folder? Confirm refresh_index_if_changed() is actually being called before each question is answered, and that the signature comparison is detecting real changes.
4. Are very short notes producing poor search matches? A note shorter than max_chunk_size becomes a single chunk with no splitting at all — verify chunk_by_paragraph() handles this case correctly rather than producing an empty result.
5. Is confidence always coming back answered even for genuinely unclear questions? Revisit the instructions wording to more explicitly and forcefully request an honest partially_answered or not_found when the retrieved notes don't fully cover the question.

Extending the Project

This project generalizes directly to any personal or team knowledge base stored as plain text files — meeting notes, journal entries, project documentation. Natural extensions, each building on techniques from across this unit and this course, include watching the notes folder for changes automatically rather than checking on each query (extending refresh_index_if_changed() into a background process), replacing the linear-scan search with the NumPy-vectorized version from Lesson 3 as the notes collection grows, and, once the collection grows large enough or the retrieval requirements become sophisticated enough that this hand-built pipeline's maintenance cost outweighs its flexibility, migrating the same underlying notes into Unit 9, Lesson 2's file search tool and vector stores — the exact decision Lesson 4 covers in depth.