Inspecting a Synchronous Reference Pipeline and Its Outputs

Hello again. Last lesson defined the contracts between a loader, chunker, embedder, store, retriever, generator, and citation layer. Now you will run a small pipeline and inspect what crosses those boundaries.

This is an offline reference run for one Markdown file and one Python file. It uses only the Python standard library: no API key, vector database, asynchronous code, or LLM is required. Its search vectors and answer generator are deliberately simple substitutes, so you can see the data flow before replacing them with more capable components.

See the handoffs before running them

Watch the selected parts of pixegami’s RAG + Langchain Python Project: Easy AI/Chat For Your Docs. The demonstration uses particular services, which you do not need to install for this lesson. Concentrate instead on what the presenter prints at each boundary.

Our run will expose those same boundaries, with source line ranges as well as filenames.

Run the reference pipeline

Save the following as rag_trace.py, then run python rag_trace.py with Python 3.10 or newer. The script creates its two example source files in a temporary directory, runs synchronously, prints the trace, and removes the temporary files when it finishes. You do not need to supply project files yet.

from collections import Counter
from dataclasses import dataclass
from math import sqrt
from pathlib import Path
from tempfile import TemporaryDirectory
import re


@dataclass
class Document:
    source: str
    text: str


@dataclass
class Chunk:
    chunk_id: str
    source: str
    start_line: int
    end_line: int
    text: str


def make_examples(root: Path) -> None:
    examples = {
        "docs/ingestion.md": (
            "# Ingestion guide\n"
            "\n"
            "## Supported input\n"
            "Only Markdown and Python files are indexed.\n"
            "\n"
            "## Selection\n"
            "The `accepts` function checks the file suffix before loading.\n"
            "Chunks retain their source file and line numbers.\n"
        ),
        "src/loader.py": (
            "from pathlib import Path\n"
            "\n"
            'ALLOWED_SUFFIXES = {".md", ".py"}\n'
            "def accepts(path: Path) -> bool:\n"
            "    return path.suffix.lower() in ALLOWED_SUFFIXES\n"
        ),
    }
    for name, text in examples.items():
        path = root / name
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(text, encoding="utf-8")


def load(root: Path) -> list[Document]:
    documents = []
    for path in sorted(root.rglob("*")):
        if path.is_file() and path.suffix.lower() in {".md", ".py"}:
            documents.append(
                Document(
                    source=path.relative_to(root).as_posix(),
                    text=path.read_text(encoding="utf-8"),
                )
            )
    return documents


def chunk(documents: list[Document]) -> list[Chunk]:
    chunks = []
    for doc in documents:
        lines = doc.text.splitlines()
        start = 0

        # Each run of nonblank lines becomes one chunk.
        for i in range(len(lines) + 1):
            if i == len(lines) or not lines[i].strip():
                if start < i:
                    chunks.append(
                        Chunk(
                            chunk_id=f"C{len(chunks) + 1}",
                            source=doc.source,
                            start_line=start + 1,
                            end_line=i,
                            text="\n".join(lines[start:i]),
                        )
                    )
                start = i + 1
    return chunks


def words(text: str) -> list[str]:
    return re.findall(r"[a-z0-9_]+", text.lower())


def embed(text: str, vocabulary: list[str]) -> list[float]:
    counts = Counter(words(text))
    return [float(counts[word]) for word in vocabulary]


def cosine(left: list[float], right: list[float]) -> float:
    magnitude = sqrt(sum(x * x for x in left)) * sqrt(
        sum(x * x for x in right)
    )
    if magnitude == 0:
        return 0.0
    return sum(x * y for x, y in zip(left, right)) / magnitude


def retrieve(question: str, records, vocabulary, top_k: int = 2):
    query_vector = embed(question, vocabulary)
    scored = [
        (cosine(query_vector, vector), item)
        for item, vector in records
    ]
    return sorted(
        scored,
        key=lambda result: (-result[0], result[1].chunk_id),
    )[:top_k]


def build_prompt(question: str, matches) -> tuple[str, dict[str, Chunk]]:
    lookup = {}
    context_blocks = []

    for number, (_, item) in enumerate(matches, start=1):
        label = f"S{number}"
        lookup[label] = item
        context_blocks.append(
            f"[{label}] {item.source}:{item.start_line}-{item.end_line}\n"
            f"{item.text}"
        )

    prompt = (
        "INSTRUCTIONS:\n"
        "Answer using only the supplied context. Cite the supporting "
        "source label; otherwise say the evidence is insufficient.\n"
        "CONTEXT:\n"
        + "\n\n".join(context_blocks)
        + "\nQUESTION:\n"
        + question
    )
    return prompt, lookup


def example_generator(prompt: str) -> str:
    """A narrow, deterministic stand-in for an LLM; not a general answerer."""
    context, question = prompt.split("CONTEXT:\n", 1)[1].rsplit(
        "\nQUESTION:\n", 1
    )

    if "accepts" not in question.lower() or ".txt" not in question.lower():
        return "I do not have an answer from this reference generator."

    for block in context.strip().split("\n\n"):
        header, _, passage = block.partition("\n")
        if (
            'ALLOWED_SUFFIXES = {".md", ".py"}' in passage
            and "return path.suffix.lower() in ALLOWED_SUFFIXES" in passage
        ):
            label = header.split("]", 1)[0].removeprefix("[")
            return (
                "No. `accepts` checks whether the path's lowercased suffix "
                "belongs to the set containing `.md` and `.py`. "
                f"`.txt` is not in that set. [{label}]"
            )

    return "The retrieved passages do not establish the answer."


def resolve_citations(answer: str, lookup: dict[str, Chunk]) -> list[str]:
    labels = list(dict.fromkeys(re.findall(r"\[(S\d+)\]", answer)))
    unknown = [label for label in labels if label not in lookup]
    if unknown:
        raise ValueError(f"Unknown citation labels: {unknown}")

    return [
        f"[{label}] {lookup[label].source}:"
        f"{lookup[label].start_line}-{lookup[label].end_line}"
        f" (chunk {lookup[label].chunk_id})"
        for label in labels
    ]


def main() -> None:
    with TemporaryDirectory(prefix="rag_trace_") as location:
        root = Path(location)
        make_examples(root)
        documents = load(root)
        chunks = chunk(documents)

        print("DOCUMENTS")
        for doc in documents:
            print(f"{doc.source}: {doc.text!r}")

        print("\nCHUNKS")
        for item in chunks:
            print(
                f"{item.chunk_id} {item.source}:"
                f"{item.start_line}-{item.end_line} {item.text!r}"
            )

        # Toy vectors: counts over one vocabulary shared by chunks and queries.
        vocabulary = sorted({
            word for item in chunks for word in words(item.text)
        })
        records = [(item, embed(item.text, vocabulary)) for item in chunks]
        print(f"\nINDEX: {len(records)} records; "
              f"{len(vocabulary)} vector dimensions")

        question = "Does `accepts` return True for a `.txt` path?"
        matches = retrieve(question, records, vocabulary, top_k=2)

        print(f"\nQUESTION: {question}")
        print("RETRIEVAL RESULTS")
        for score, item in matches:
            print(
                f"{item.chunk_id} score={score:.3f} "
                f"{item.source}:{item.start_line}-{item.end_line} "
                f"{item.text!r}"
            )

        prompt, lookup = build_prompt(question, matches)
        print("\nPROMPT\n" + prompt)

        answer = example_generator(prompt)
        print("\nANSWER\n" + answer)

        print("\nCITATIONS")
        for citation in resolve_citations(answer, lookup):
            print(citation)


if __name__ == "__main__":
    main()

The script is intentionally contained in one file so you can run and inspect it now. Its functions still mark replaceable boundaries; later lessons will develop the ingestion, chunking, embedding, retrieval, and generation choices separately.

Read the output as a trace

Start with DOCUMENTS. There should be two: docs/ingestion.md and src/loader.py. Each holds a whole file’s text and its relative path. Nothing has been selected as evidence yet.

In CHUNKS, the blank-line splitter produces several smaller passages. Find the chunk from src/loader.py:3-5. It contains both the allowed suffixes and the accepts function’s return statement. Keeping those lines together matters: the return statement refers to ALLOWED_SUFFIXES, so either line alone would provide weaker evidence. The printed C... ID identifies the stored chunk; the path and line range identify its origin.

The INDEX line reports a record for each chunk and the length of each numeric vector. Here the “embedder” counts words against a shared vocabulary, and the “store” is the in-memory records list. These are real, inspectable numeric vectors, but word-count vectors are not semantic embeddings: they cannot reliably connect different expressions of the same idea. No vector or index survives after the script exits.

For RETRIEVAL RESULTS, look at the ranked scores and the returned text. The question mentions accepts and path, which help the code passage rank highly under this toy search. Importantly, .txt need not occur in a source passage for that passage to answer the question. The relevant code specifies the allowed set; you can infer that .txt is absent. Scores rank possible evidence here; they do not certify that an answer is correct.

Next, read the entire PROMPT. It contains instructions, the retrieved passages with prompt-local labels such as [S1], and the original question. Compare it with RETRIEVAL RESULTS: the generator receives the passage text, not its vector. The labels also have a lookup back to the chunks’ paths and line ranges.

Finally, examine ANSWER and CITATIONS together. For this particular question, the deterministic generator responds only if a retrieved passage contains both the suffix set and the return statement. Its citation label resolves to src/loader.py:3-5. Check those three lines yourself: they support the narrow claim about this accepts function. They would not support a claim that no other part of a larger application ever processes .txt files.

The citation resolver checks that a label in the answer actually refers to a retrieved chunk. It does not automatically verify that every claim is supported. Your manual comparison of answer and source is a separate, necessary check; later lessons will address that problem more systematically.

What this run does—and does not—demonstrate

You have executed both halves of the architecture: file loading, chunking, vector creation, and storage happen before the question; question vectorization, retrieval, prompt construction, answer production, and citation resolution happen for the question. Every stage is synchronous, and you can inspect its output.

There are two deliberate limitations. The blank-line chunker is not yet Markdown- or Python-structure-aware; it happens to keep the needed Python lines together in this tiny file. And example_generator is not an LLM: it recognizes this one question and code pattern, then returns a fixed-form answer or declines. It makes the generation handoff runnable without a provider, but it should not be presented as a general question-answering system.

The important accomplishment is a checkable trace: you can point to the loaded file, the stored chunk, its retrieval score, the exact passage placed in the prompt, and the source location behind the answer.

Key takeaways

A runnable RAG trace is more informative than an architecture diagram alone. Here, a Markdown file and a Python file became identifiable documents and chunks; a question selected ranked chunks; the prompt supplied their words to a generator substitute; and a citation label resolved back to source lines. The toy components expose the contracts without hiding them behind a service.

Next, we move into reliable ingestion: discovering only the intended .md and .py files, and handling the practical ways file loading can fail before any chunk reaches an index.