Phase 1: Document Ingestion

This page demonstrates the ingestion pipeline built in Phase 1 using the same synthetic data as the unit tests in tests/test_ingest.py. Every output here is produced by the live rag.ingest module — not hardcoded.

Related issues: #1 Load CSV (closed), #2 Chunk documents (closed), #3 Embed and store in ChromaDB (closed), #4 Spot-check chunk quality (closed).

1 Step 1 — Load speeches from CSV

load_speeches_csv() reads each row of the speeches CSV and converts it to a LangChain Document with the speech text as page_content and all columns (date, year, speakers, title) attached as metadata.

Code

import sys
sys.path.insert(0, "..")

import pandas as pd
from io import StringIO
from rag.ingest import load_speeches_csv, chunk_documents

# Same synthetic data used in test_load_speeches_csv and test_chunk_documents
SAMPLE = pd.DataFrame({
    "date":     ["2023-01-15", "2023-02-20"],
    "year":     [2023, 2023],
    "speakers": ["Christine Lagarde", "Philip Lane"],
    "title":    ["Monetary Policy Speech", "Economic Outlook"],
    "subtitle": ["January Meeting", "February Update"],
    "contents": [
        "The ECB reaffirms its commitment to price stability. " * 20,
        "Economic conditions remain challenging. " * 20,
    ],
})

import tempfile, os
with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False, newline="") as f:
    SAMPLE.to_csv(f, index=False)
    tmp_path = f.name

documents = load_speeches_csv(tmp_path)
os.unlink(tmp_path)

print(f"Speeches loaded: {len(documents)}")

Speeches loaded: 2

1.1 Loaded documents

Code

from IPython.display import HTML, display

rows = "".join(
    f"<tr><td>{i}</td><td>{d.metadata['speakers']}</td><td>{d.metadata['date']}</td>"
    f"<td>{d.metadata['title']}</td><td>{len(d.page_content)} chars</td></tr>"
    for i, d in enumerate(documents)
)

display(HTML(
    "<table style='width:100%;border-collapse:collapse'>"
    "<thead><tr style='border-bottom:2px solid #dee2e6'>"
    "<th>#</th><th>Speaker</th><th>Date</th><th>Title</th><th>Content length</th>"
    "</tr></thead><tbody>" + rows + "</tbody></table>"
))

#	Speaker	Date	Title	Content length
0	Christine Lagarde	2023-01-15	Monetary Policy Speech	1060 chars
1	Philip Lane	2023-02-20	Economic Outlook	800 chars

1.2 Sample metadata

Code

import json
print(json.dumps(documents[0].metadata, indent=2))

{
  "date": "2023-01-15",
  "year": "2023",
  "speakers": "Christine Lagarde",
  "title": "Monetary Policy Speech",
  "subtitle": "January Meeting",
  "source": "/tmp/tmps5oywwc4.csv",
  "row_index": 0
}

2 Step 2 — Chunk documents

chunk_documents() splits each Document into smaller overlapping chunks using RecursiveCharacterTextSplitter. Metadata is preserved on every chunk so retrieved chunks can always be traced back to their source speech.

Code

chunks = chunk_documents(documents, chunk_size=100, chunk_overlap=10)

print(f"Source documents : {len(documents)}")
print(f"Chunks produced  : {len(chunks)}")
print(f"Avg chunk length : {sum(len(c.page_content) for c in chunks) / len(chunks):.0f} chars")

Source documents : 2
Chunks produced  : 20
Avg chunk length : 96 chars

2.1 Chunk lengths

Code

lengths = [len(c.page_content) for c in chunks]

bars = ""
max_len = max(lengths)
for i, l in enumerate(lengths):
    width = int((l / max_len) * 300)
    bars += (
        f"<tr><td style='padding:2px 8px'>chunk {i}</td>"
        f"<td><div style='background:#4c8bf5;height:16px;width:{width}px;border-radius:2px'></div></td>"
        f"<td style='padding:2px 8px'>{l} chars</td></tr>"
    )

display(HTML(
    "<table style='border-collapse:collapse'>" + bars + "</table>"
))

chunk 0		94 chars
chunk 1		94 chars
chunk 2		95 chars
chunk 3		94 chars
chunk 4		94 chars
chunk 5		95 chars
chunk 6		94 chars
chunk 7		94 chars
chunk 8		95 chars
chunk 9		94 chars
chunk 10		94 chars
chunk 11		87 chars
chunk 12		99 chars
chunk 13		99 chars
chunk 14		99 chars
chunk 15		99 chars
chunk 16		99 chars
chunk 17		99 chars
chunk 18		99 chars
chunk 19		99 chars

2.2 Sample chunk with metadata

Code

c = chunks[0]
print(f"Content : {c.page_content[:120]}...")
print()
print(f"Metadata: {json.dumps(c.metadata, indent=2)}")

Content : The ECB reaffirms its commitment to price stability. The ECB reaffirms its commitment to price...

Metadata: {
  "date": "2023-01-15",
  "year": "2023",
  "speakers": "Christine Lagarde",
  "title": "Monetary Policy Speech",
  "subtitle": "January Meeting",
  "source": "/tmp/tmps5oywwc4.csv",
  "row_index": 0
}

3 Step 3 — Embed chunks and store in ChromaDB

ingest_to_chromadb() is the end-to-end function: it calls Steps 1 and 2 internally, then embeds every chunk using Ollama (nomic-embed-text) and writes them to a local ChromaDB collection.

Because Ollama is not available at render time, the cell below uses the same mock strategy as tests/test_ingest.py — replacing OllamaEmbeddings and Chroma with lightweight fakes so the pipeline logic can be demonstrated without a running model.

Code

from unittest.mock import MagicMock, patch

# Fake embedding: returns a 4-dim vector for any text (real model uses 768-dim)
class FakeEmbeddings:
    def embed_documents(self, texts):
        return [[0.1, 0.2, 0.3, 0.4]] * len(texts)
    def embed_query(self, text):
        return [0.1, 0.2, 0.3, 0.4]

mock_vectorstore = MagicMock()
mock_vectorstore.__len__ = lambda self: len(chunks)

with patch("rag.ingest.OllamaEmbeddings", return_value=FakeEmbeddings()), \
     patch("rag.ingest.Chroma.from_documents", return_value=mock_vectorstore), \
     patch("os.makedirs"):
    from rag.ingest import ingest_to_chromadb
    import tempfile, os
    with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False, newline="") as f:
        SAMPLE.to_csv(f, index=False)
        tmp_path = f.name
    vs = ingest_to_chromadb(tmp_path, chroma_dir="/tmp/demo_chroma")
    os.unlink(tmp_path)

Loading ECB speeches from /tmp/tmp7awsixra.csv...
Loaded 2 speeches.
Chunking documents...
Created 3 chunks.
Initializing Ollama embeddings (nomic-embed-text)...
Embedding and storing in ChromaDB...
Ingestion complete. ChromaDB persisted to /tmp/demo_chroma
  Total chunks: 3
  Collection: ecb_speeches

3.1 What gets stored

Code

print(f"Chunks embedded and stored : {len(chunks)}")
print(f"ChromaDB collection        : ecb_speeches")
print(f"Embedding model            : nomic-embed-text (768-dim, via Ollama)")
print(f"Embedding dims (mock)      : 4  (real model: 768)")
print()
print("Each stored document contains:")
print("  • page_content  — the chunk text (≤1000 chars by default)")
print("  • metadata      — speaker, date, title, source file, row index")

Chunks embedded and stored : 20
ChromaDB collection        : ecb_speeches
Embedding model            : nomic-embed-text (768-dim, via Ollama)
Embedding dims (mock)      : 4  (real model: 768)

Each stored document contains:
  • page_content  — the chunk text (≤1000 chars by default)
  • metadata      — speaker, date, title, source file, row index

3.2 Pipeline summary

Code

display(HTML(
    "<table style='width:100%;border-collapse:collapse'>"
    "<thead><tr style='border-bottom:2px solid #dee2e6'>"
    "<th style='text-align:left;padding:6px'>Step</th>"
    "<th style='text-align:left;padding:6px'>Function</th>"
    "<th style='text-align:left;padding:6px'>Input → Output</th>"
    "</tr></thead>"
    "<tbody>"
    "<tr><td style='padding:6px'>1</td><td><code>load_speeches_csv()</code></td>"
    "<td>CSV rows → LangChain Documents</td></tr>"
    "<tr><td style='padding:6px'>2</td><td><code>chunk_documents()</code></td>"
    "<td>Documents → overlapping chunks (metadata preserved)</td></tr>"
    "<tr><td style='padding:6px'>3</td><td><code>ingest_to_chromadb()</code></td>"
    "<td>Chunks → embeddings → ChromaDB collection</td></tr>"
    "<tr><td style='padding:6px'>—</td><td><code>load_chromadb()</code></td>"
    "<td>Disk → Chroma instance (for query phase)</td></tr>"
    "</tbody></table>"
))

Step	Function	Input → Output
1	`load_speeches_csv()`	CSV rows → LangChain Documents
2	`chunk_documents()`	Documents → overlapping chunks (metadata preserved)
3	`ingest_to_chromadb()`	Chunks → embeddings → ChromaDB collection
—	`load_chromadb()`	Disk → Chroma instance (for query phase)

4 Step 4 — Spot-check chunk quality

Before embedding, it is worth verifying that the chunks produced by Step 2 are sensible: no empty splits, no degenerate micro-chunks, metadata intact, and lengths within the requested bounds. These checks are codified in tests/test_ingest.py and run on every push via CI.

Code

quality_checks = [
    ("No empty chunks",            all(c.page_content.strip() for c in chunks)),
    ("No chunks shorter than 10 chars", all(len(c.page_content.strip()) >= 10 for c in chunks)),
    ("All required metadata keys present",
        all({"date", "year", "speakers", "title", "source"} <= set(c.metadata) for c in chunks)),
    ("No chunk exceeds chunk_size by >10%",
        all(len(c.page_content) <= 100 * 1.1 for c in chunks)),
]

rows = "".join(
    f"<tr><td>{'✅' if ok else '❌'}</td><td>{label}</td></tr>"
    for label, ok in quality_checks
)
display(HTML(
    "<table style='border-collapse:collapse'>"
    "<thead><tr style='border-bottom:2px solid #dee2e6'>"
    "<th style='padding:6px'>Result</th><th style='padding:6px;text-align:left'>Check</th>"
    "</tr></thead><tbody>" + rows + "</tbody></table>"
))

Result	Check
✅	No empty chunks
✅	No chunks shorter than 10 chars
✅	All required metadata keys present
✅	No chunk exceeds chunk_size by >10%

4.1 Chunk length distribution

Code

lengths = [len(c.page_content) for c in chunks]
bins = [0, 25, 50, 75, 100, 125]
labels_bin = ["0–25", "26–50", "51–75", "76–100", "101+"]
counts = [
    sum(1 for l in lengths if bins[i] <= l < bins[i+1])
    for i in range(len(bins)-1)
]
counts.append(sum(1 for l in lengths if l >= 125))

max_count = max(counts) if counts else 1
bars = "".join(
    f"<tr><td style='padding:2px 8px'>{labels_bin[i]}</td>"
    f"<td><div style='background:#4c8bf5;height:16px;width:{int(counts[i]/max_count*200)}px;"
    f"border-radius:2px;min-width:2px'></div></td>"
    f"<td style='padding:2px 8px'>{counts[i]} chunks</td></tr>"
    for i in range(len(labels_bin))
)
display(HTML("<table style='border-collapse:collapse'>" + bars + "</table>"))
print(f"Min: {min(lengths)} chars  |  Max: {max(lengths)} chars  |  Avg: {sum(lengths)/len(lengths):.0f} chars")

0–25		0 chunks
26–50		0 chunks
51–75		0 chunks
76–100		20 chunks
101+		0 chunks

Min: 87 chars  |  Max: 99 chars  |  Avg: 96 chars

5 Phase 1 complete

All three ingestion steps are implemented, tested, and demonstrated above. The ChromaDB collection (ecb_speeches) is ready for the Phase 2 query pipeline — retrieval + generation.

See Phase 2 issues #5–8.