11.4 Building a Repository Intelligence Layer

1. Architecture Overview

The Repository Intelligence Layer has three subsystems, each responsible for a distinct phase of the context engineering pipeline:

Without a dedicated intelligence layer, coding agents resort to brute-force strategies: grepping the entire repository on every query, stuffing random files into the prompt, or asking the developer to manually select relevant context. These workarounds fail silently at scale, producing hallucinated edits to functions the agent never actually read.

A coding agent receives the task "add rate limiting to the /api/users endpoint" on a 200,000-line codebase, but its context window holds only 32,000 tokens, roughly 1% of the code. How does it choose which fragments to read? The Repository Intelligence Layer solves this problem. It maintains a continuously updated semantic index of the codebase so that any agent can ask natural-language questions and receive precisely the relevant fragments, packed within a token budget. The mechanism is a three-stage pipeline, illustrated in Figure 11.6. An indexer parses source files into structural chunks and computes embeddings. A retriever runs hybrid search (dense vector similarity plus keyword matching) over those chunks. A packer assembles the top results into a prompt string that respects the model's token limit. Use a RIL when your codebase exceeds what fits in a single prompt (roughly 5,000+ lines); for smaller projects, passing the entire source tree as context is equally effective.

1. Indexer: walks the repository, parses files with tree-sitter (a parser generator that produces concrete syntax trees for source code, enabling language-aware chunking), extracts hierarchical chunks, generates summaries, computes embeddings, and stores everything in a PostgreSQL database with the pgvector extension.
2. Retriever: accepts a natural-language query (plus optional focal file), runs hybrid search (dense + BM25, a classical term-frequency ranking algorithm that scores documents by literal token overlap with the query + structural re-ranking), and returns ranked chunks with their summaries.
3. Packer: takes ranked results and a token budget, resolves dependencies (parent signatures, imports), applies U-shape positioning (placing the most relevant chunks at the beginning and end of the prompt, where LLM attention is strongest), and produces a ready-to-use prompt string.

These three subsystems communicate through the chunk store (PostgreSQL + pgvector). The indexer writes; the retriever reads; the packer reads and assembles. This separation means the indexer can run as a background process (triggered by file changes), while the retriever and packer respond to interactive queries with sub-second latency in typical configurations. On a 185,000-line codebase with over 4,000 indexed chunks, the full retrieve-and-pack pipeline typically completes in under 100 milliseconds on modern hardware, fast enough that the agent never waits for its context. In short: A Repository Intelligence Layer is a search engine that speaks code structure, not just text.

"""
Repository Intelligence Layer: architecture overview.

Components and their responsibilities:

  FileWatcher ──> Indexer ──> PostgreSQL + pgvector
                                   │
  Agent Query ──> Retriever ───────┤
                                   │
                  Packer <─────────┘
                    │
                  Prompt String ──> LLM
"""
from dataclasses import dataclass, field

@dataclass
class RILConfig:
    """Configuration for the Repository Intelligence Layer."""
    repo_root: str
    db_connection: str = "postgresql://localhost:5432/repo_intel"
    embedding_model: str = "all-MiniLM-L6-v2"
    chunk_languages: list[str] = field(default_factory=lambda: ["python"])
    default_budget: int = 32_000    # tokens
    bm25_candidates: int = 50       # top-k from BM25
    dense_candidates: int = 50      # top-k from dense search
    structural_boost: float = 1.5   # multiplier for import-related files
    skip_dirs: set[str] = field(default_factory=lambda: {
        ".git", "node_modules", "__pycache__", ".venv", "venv",
        ".mypy_cache", ".pytest_cache", "dist", "build",
    })

2. The pgvector-Backed Store

In Section 11.2 we used SQLite for the chunk store. For production use, PostgreSQL with the pgvector extension (where pgvector is a PostgreSQL extension that adds a native vector column type and operators for similarity search) adds native vector columns and approximate nearest-neighbor (ANN) search via Hierarchical Navigable Small World (HNSW) indexes. This consolidates structured metadata (chunk content, summaries, file paths) and vector search (embeddings) into a single database, eliminating the need for a separate vector store.

import psycopg
from pgvector.psycopg import register_vector

class PgChunkStore:
    """PostgreSQL + pgvector store for chunks and embeddings."""

    def __init__(self, connection_string: str, embedding_dim: int = 384):
        self.conn = psycopg.connect(connection_string)
        register_vector(self.conn)
        self.embedding_dim = embedding_dim
        self._create_schema()

    def _create_schema(self):
        with self.conn.cursor() as cur:
            cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
            cur.execute(f"""
                CREATE TABLE IF NOT EXISTS chunks (
                    id TEXT PRIMARY KEY,
                    file_path TEXT NOT NULL,
                    chunk_type TEXT NOT NULL,
                    name TEXT NOT NULL,
                    start_line INTEGER,
                    end_line INTEGER,
                    content TEXT NOT NULL,
                    token_estimate INTEGER,
                    parent_id TEXT REFERENCES chunks(id),
                    signature TEXT,
                    docstring TEXT,
                    llm_summary TEXT,
                    embedding vector({self.embedding_dim}),
                    updated_at TIMESTAMPTZ DEFAULT NOW()
                );
                CREATE INDEX IF NOT EXISTS idx_chunks_file
                    ON chunks(file_path);
                CREATE INDEX IF NOT EXISTS idx_chunks_parent
                    ON chunks(parent_id);
            """)
            # HNSW index for fast approximate nearest-neighbor search
            cur.execute(f"""
                CREATE INDEX IF NOT EXISTS idx_chunks_embedding
                    ON chunks
                    USING hnsw (embedding vector_cosine_ops)
                    WITH (m = 16, ef_construction = 64);
            """)
        self.conn.commit()

    def upsert_chunk(self, chunk_id: str, file_path: str, chunk_type: str,
                     name: str, start_line: int, end_line: int,
                     content: str, token_estimate: int,
                     parent_id: str | None, signature: str,
                     docstring: str, llm_summary: str,
                     embedding: list[float]):
        """Insert or update a chunk with all its metadata and embedding."""
        with self.conn.cursor() as cur:
            cur.execute("""
                INSERT INTO chunks (id, file_path, chunk_type, name,
                    start_line, end_line, content, token_estimate,
                    parent_id, signature, docstring, llm_summary,
                    embedding, updated_at)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s,
                        %s::vector, NOW())
                ON CONFLICT (id) DO UPDATE SET
                    content = EXCLUDED.content,
                    token_estimate = EXCLUDED.token_estimate,
                    signature = EXCLUDED.signature,
                    docstring = EXCLUDED.docstring,
                    llm_summary = EXCLUDED.llm_summary,
                    embedding = EXCLUDED.embedding,
                    updated_at = NOW()
            """, (chunk_id, file_path, chunk_type, name,
                  start_line, end_line, content, token_estimate,
                  parent_id, signature, docstring, llm_summary,
                  embedding))

    def vector_search(self, query_embedding: list[float],
                      top_k: int = 20) -> list[dict]:
        """Find the top-k chunks nearest to the query embedding."""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT id, file_path, chunk_type, name, start_line,
                       end_line, content, token_estimate, parent_id,
                       signature, docstring, llm_summary,
                       1 - (embedding <=> %s::vector) AS similarity
                FROM chunks
                ORDER BY embedding <=> %s::vector
                LIMIT %s
            """, (query_embedding, query_embedding, top_k))

            cols = ["id", "file_path", "chunk_type", "name", "start_line",
                    "end_line", "content", "token_estimate", "parent_id",
                    "signature", "docstring", "llm_summary", "similarity"]
            return [dict(zip(cols, row)) for row in cur.fetchall()]

    def get_chunk(self, chunk_id: str) -> dict | None:
        """Retrieve a single chunk by ID."""
        with self.conn.cursor() as cur:
            cur.execute("""
                SELECT id, file_path, chunk_type, name, start_line,
                       end_line, content, token_estimate, parent_id,
                       signature, docstring, llm_summary
                FROM chunks WHERE id = %s
            """, (chunk_id,))
            row = cur.fetchone()
            if row is None:
                return None
            cols = ["id", "file_path", "chunk_type", "name", "start_line",
                    "end_line", "content", "token_estimate", "parent_id",
                    "signature", "docstring", "llm_summary"]
            return dict(zip(cols, row))

    def delete_stale(self, file_path: str):
        """Remove all chunks for a file (before re-indexing it)."""
        with self.conn.cursor() as cur:
            cur.execute("DELETE FROM chunks WHERE file_path = %s", (file_path,))

    def commit(self):
        self.conn.commit()

    def close(self):
        self.conn.close()

3. The Indexing Pipeline

With the chunk store in place, the next question is how to populate it efficiently as files are added and changed.

The indexer orchestrates the full pipeline: file discovery, parsing, chunking, summarization, embedding, and storage. It supports both full re-indexing (for initial setup) and incremental indexing (for file changes detected by a watcher). Embeddings are computed using sentence-transformers, a Python library that wraps pre-trained transformer models to produce fixed-length vector representations of text.

import hashlib
from pathlib import Path
from sentence_transformers import SentenceTransformer

class RepositoryIndexer:
    """Indexes a repository into the pgvector-backed chunk store."""

    def __init__(self, config: RILConfig):
        self.config = config
        self.store = PgChunkStore(config.db_connection)
        self.embedder = SentenceTransformer(config.embedding_model)
        self._file_hashes: dict[str, str] = {}

    def _file_hash(self, path: Path) -> str:
        """Compute a content hash to detect changes."""
        return hashlib.md5(path.read_bytes()).hexdigest()

    def _should_index(self, path: Path) -> bool:
        """Check whether a file needs re-indexing."""
        rel = str(path.relative_to(self.config.repo_root))
        current_hash = self._file_hash(path)
        if self._file_hashes.get(rel) == current_hash:
            return False
        self._file_hashes[rel] = current_hash
        return True

    def _embed_text(self, text: str) -> list[float]:
        """Compute embedding for a text string."""
        return self.embedder.encode(text, normalize_embeddings=True).tolist()

    def index_file(self, file_path: Path):
        """Index a single file: parse, chunk, summarize, embed, store."""
        try:
            source = file_path.read_text(encoding="utf-8")
        except (UnicodeDecodeError, PermissionError):
            return

        rel_path = str(file_path.relative_to(self.config.repo_root))

        # Remove stale chunks for this file
        self.store.delete_stale(rel_path)

        # Build hierarchical chunks
        # build_hierarchy and extract_docstrings are defined in
        # Section 11.2; extract_signature is from Section 11.3.
        hierarchy = build_hierarchy(source, rel_path)
        docstrings = extract_docstrings(source)

        for h_chunk in hierarchy:
            chunk = h_chunk.chunk
            sig = extract_signature(chunk)
            doc = docstrings.get(chunk.name, "")

            # Choose the best text for embedding
            embed_text = doc if doc else sig if sig else chunk.content[:2000]
            embedding = self._embed_text(embed_text)

            self.store.upsert_chunk(
                chunk_id=chunk.id, file_path=rel_path,
                chunk_type=chunk.chunk_type, name=chunk.name,
                start_line=chunk.start_line, end_line=chunk.end_line,
                content=chunk.content, token_estimate=chunk.token_estimate,
                parent_id=h_chunk.parent_id, signature=sig,
                docstring=doc, llm_summary="", embedding=embedding,
            )

            # Index children (methods)
            for child in h_chunk.children:
                cc = child.chunk
                child_sig = extract_signature(cc)
                child_doc = docstrings.get(cc.name.split(".")[-1], "")
                child_embed = child_doc if child_doc else child_sig
                child_embedding = self._embed_text(child_embed or cc.content[:2000])

                self.store.upsert_chunk(
                    chunk_id=cc.id, file_path=rel_path,
                    chunk_type=cc.chunk_type, name=cc.name,
                    start_line=cc.start_line, end_line=cc.end_line,
                    content=cc.content, token_estimate=cc.token_estimate,
                    parent_id=chunk.id, signature=child_sig,
                    docstring=child_doc, llm_summary="",
                    embedding=child_embedding,
                )

    def index_repository(self) -> dict:
        """Full repository indexing pass. Returns statistics."""
        root = Path(self.config.repo_root)
        stats = {"files_indexed": 0, "files_skipped": 0}

        extension_map = {".py": "python", ".js": "javascript", ".ts": "typescript"}
        target_extensions = {
            ext for ext, lang in extension_map.items()
            if lang in self.config.chunk_languages
        }

        for path in root.rglob("*"):
            if not path.is_file():
                continue
            if path.suffix not in target_extensions:
                continue
            if any(part in self.config.skip_dirs for part in path.parts):
                continue

            if self._should_index(path):
                self.index_file(path)
                stats["files_indexed"] += 1
            else:
                stats["files_skipped"] += 1

        self.store.commit()
        return stats

4. The Query Interface

The query interface ties together the retriever and packer from Section 11.3, providing a single method that accepts a natural-language query and returns a ready-to-use context string. This is the interface that coding agents call.

from rank_bm25 import BM25Okapi
import re

class RepositoryIntelligence:
    """The unified query interface for the Repository Intelligence Layer."""

    def __init__(self, config: RILConfig):
        self.config = config
        self.store = PgChunkStore(config.db_connection)
        self.embedder = SentenceTransformer(config.embedding_model)
        self.import_graph: dict[str, set[str]] = {}
        self._bm25: BM25Okapi | None = None
        self._bm25_chunks: list[dict] = []

    def load_import_graph(self):
        """Build the import graph from the indexed repository."""
        # build_import_graph() walks source files and extracts
        # import statements; see Section 11.3 for the implementation.
        self.import_graph = build_import_graph(self.config.repo_root)

    def load_bm25_index(self):
        """Build the BM25 index from all stored chunks."""
        with self.store.conn.cursor() as cur:
            cur.execute("SELECT id, file_path, chunk_type, name, content, "
                        "token_estimate, parent_id, signature, docstring "
                        "FROM chunks")
            cols = ["id", "file_path", "chunk_type", "name", "content",
                    "token_estimate", "parent_id", "signature", "docstring"]
            self._bm25_chunks = [dict(zip(cols, row)) for row in cur.fetchall()]

        tokenized = [self._tokenize(c["content"]) for c in self._bm25_chunks]
        self._bm25 = BM25Okapi(tokenized)

    @staticmethod
    def _tokenize(text: str) -> list[str]:
        """Code-aware tokenization for BM25."""
        text = re.sub(r'([a-z])([A-Z])', r'\1 \2', text)
        return re.findall(r'[a-zA-Z]\w*', text.lower())

    def query(
        self,
        question: str,
        focal_file: str | None = None,
        budget: int | None = None,
    ) -> str:
        """Answer a query with assembled context from the repository.

        Args:
            question: natural-language description of the task
            focal_file: the file being edited (if known), for structural boost
            budget: token budget override (defaults to config.default_budget)

        Returns:
            A formatted prompt string containing relevant code chunks.
        """
        budget = budget or self.config.default_budget

        # 1. Dense retrieval via pgvector
        query_embedding = self.embedder.encode(
            question, normalize_embeddings=True
        ).tolist()
        dense_results = self.store.vector_search(
            query_embedding, top_k=self.config.dense_candidates
        )

        # 2. Sparse retrieval via BM25
        sparse_results = []
        if self._bm25 is not None:
            query_tokens = self._tokenize(question)
            bm25_scores = self._bm25.get_scores(query_tokens)
            top_indices = bm25_scores.argsort()[::-1][:self.config.bm25_candidates]
            sparse_results = [
                (self._bm25_chunks[i], float(bm25_scores[i]))
                for i in top_indices
            ]

        # 3. Fuse with RRF
        dense_as_tuples = [(r, r.get("similarity", 0.0)) for r in dense_results]
        fused = reciprocal_rank_fusion(
            [dense_as_tuples, sparse_results], top_n=30
        )

        # 4. Structural re-ranking
        if focal_file and self.import_graph:
            fused = structural_rerank(
                fused, self.import_graph, focal_file,
                boost_factor=self.config.structural_boost,
            )

        # 5. Pack into context
        packed = pack_context(fused, budget, self.store)

        # 6. Prepend the file tree for orientation
        # build_file_tree() produces a text representation of the
        # directory structure, giving the agent spatial orientation.
        tree = build_file_tree(self.config.repo_root, max_depth=3)
        tree_header = f"## Repository Structure\n```\n{tree}\n```\n\n"
        tree_tokens = len(tree.split()) * 2

        if packed.total_tokens + tree_tokens <= budget:
            return tree_header + "## Relevant Code\n\n" + packed.to_prompt()
        else:
            return "## Relevant Code\n\n" + packed.to_prompt()

    def get_file_summary(self, file_path: str) -> str:
        """Get a summary of all chunks in a specific file."""
        with self.store.conn.cursor() as cur:
            cur.execute(
                "SELECT name, chunk_type, signature, docstring "
                "FROM chunks WHERE file_path = %s ORDER BY start_line",
                (file_path,)
            )
            rows = cur.fetchall()

        lines = [f"## {file_path}\n"]
        for name, chunk_type, signature, docstring in rows:
            lines.append(f"### {chunk_type}: {name}")
            if signature:
                lines.append(f"```python\n{signature}\n```")
            if docstring:
                lines.append(f"> {docstring[:200]}")
            lines.append("")

        return "\n".join(lines)

5. Integration with the Discovery Workbench

The query interface provides the retrieval logic, but it still needs a home within the broader application architecture where agents can discover and invoke it.

The Repository Intelligence Layer plugs into the Discovery Workbench (Chapter 6) as its code understanding module. Chapter 12 wraps it in an MCP server for cross-agent access; here, we register it as a Python service.

class DiscoveryWorkbench:
    """The Discovery Workbench platform (growing across the book).

    This class is a simplified view; the full implementation spans
    multiple chapters. Here we add the repository intelligence module.
    """

    def __init__(self):
        self.modules: dict[str, object] = {}

    def register_module(self, name: str, module: object):
        """Register a service module with the workbench."""
        self.modules[name] = module
        print(f"[Workbench] Registered module: {name}")

    def get_module(self, name: str) -> object:
        return self.modules[name]


def setup_repo_intelligence(workbench: DiscoveryWorkbench,
                            repo_root: str,
                            db_connection: str) -> RepositoryIntelligence:
    """Initialize and register the Repository Intelligence Layer.

    This function:
    1. Creates the RIL configuration
    2. Runs the initial indexing pass
    3. Loads the BM25 and import graph indexes
    4. Registers the RIL with the Discovery Workbench

    Call this once during workbench startup.
    """
    config = RILConfig(repo_root=repo_root, db_connection=db_connection)

    # Index the repository
    indexer = RepositoryIndexer(config)
    stats = indexer.index_repository()
    print(f"[RIL] Indexed {stats['files_indexed']} files "
          f"({stats['files_skipped']} unchanged)")

    # Build the query interface
    ril = RepositoryIntelligence(config)
    ril.load_import_graph()
    ril.load_bm25_index()

    # Register with the workbench
    workbench.register_module("repo_intelligence", ril)

    return ril


# Usage example
workbench = DiscoveryWorkbench()
ril = setup_repo_intelligence(
    workbench,
    repo_root="/path/to/my/project",
    db_connection="postgresql://localhost:5432/repo_intel",
)

# An agent queries the RIL for context
context = ril.query(
    question="Fix the pagination bug in the user list endpoint",
    focal_file="src/api/routes/users.py",
    budget=32_000,
)
print(f"Assembled context: {len(context.split())} words")

6. Incremental Updates with File Watching

A repository is not static. Developers (and coding agents) edit files continuously. The indexer must re-index changed files without re-processing the entire repository. We use the watchdog library to monitor the filesystem and trigger incremental re-indexing when files change.

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, FileModifiedEvent, FileCreatedEvent
import threading

class IndexUpdateHandler(FileSystemEventHandler):
    """Watches for file changes and triggers incremental re-indexing."""

    def __init__(self, indexer: RepositoryIndexer, extensions: set[str] = {".py"}):
        self.indexer = indexer
        self.extensions = extensions
        self._lock = threading.Lock()

    def on_modified(self, event):
        self._handle(event)

    def on_created(self, event):
        self._handle(event)

    def _handle(self, event):
        if event.is_directory:
            return
        path = Path(event.src_path)
        if path.suffix not in self.extensions:
            return
        # Skip files in ignored directories
        if any(part in self.indexer.config.skip_dirs for part in path.parts):
            return

        with self._lock:
            self.indexer.index_file(path)
            self.indexer.store.commit()


def start_file_watcher(indexer: RepositoryIndexer) -> Observer:
    """Start a background file watcher for incremental index updates.

    Returns the Observer instance (call .stop() to shut down).
    """
    handler = IndexUpdateHandler(indexer)
    observer = Observer()
    observer.schedule(handler, indexer.config.repo_root, recursive=True)
    observer.daemon = True
    observer.start()
    print(f"[RIL] File watcher started on {indexer.config.repo_root}")
    return observer

7. End-to-End Walkthrough

Now that every component is in place, from the pgvector store through the hybrid retriever to the file watcher that keeps the index current, we can see how they cooperate on a single realistic query.

The following walkthrough traces a complete query through the system. A coding agent receives the task: "Add rate limiting to the /api/users endpoint." Here is what happens. Figure 11.4.1 illustrates end-to-end query flow through the Repository Intelligence Layer.

1. The agent calls ril.query("Add rate limiting to the /api/users endpoint", focal_file="src/api/routes/users.py", budget=32000).
2. The dense retriever embeds the query and finds chunks semantically related to "rate limiting" and "API endpoint": middleware definitions, decorator patterns, and HTTP handler functions.
3. The BM25 retriever matches on exact tokens: api, users, endpoint, rate, limit. It finds the specific users.py route file and any existing rate-limiting utilities.
4. Reciprocal Rank Fusion (RRF) merges the two result sets. Chunks that appear in both (the users route file, any rate-limiting middleware) rank highest.
5. Structural re-ranking boosts chunks from files imported by users.py (the middleware module, the authentication module, the response serializers) and from files that import users.py (the URL router configuration, the test file).
6. The packer selects chunks that fit in 32,000 tokens, includes parent class signatures for method-level chunks, and arranges them with the most relevant at the beginning and end.
7. The agent receives a context string containing the users route, the middleware base class, the authentication decorator, and the test file, positioned for maximum attention.

Exercises

1. (Conceptual) The Repository Intelligence Layer uses a single embedding model for all code. Describe a scenario where using different embedding models for different file types (e.g., one model for Python, another for SQL, a third for configuration files) would significantly improve retrieval quality. What additional complexity does multi-model embedding introduce for the vector search step?
2. (Coding) Implement the complete RIL pipeline from this section and index a public Python repository of at least 10,000 lines (e.g., Flask). Run 10 natural-language queries against it (e.g., "How does request routing work?", "Where is session handling implemented?", "Find the error handling middleware"). For each query, manually inspect the top-5 returned chunks and rate their relevance on a 1-5 scale. Report the average relevance score and identify the query types where the system performs best and worst.
3. (Analysis) Compare the retrieval quality of three configurations: (a) dense retrieval only, (b) BM25 only, (c) hybrid with RRF. Use the same 10 queries from the previous exercise. For each configuration, compute Recall@5 and MRR (using your manual relevance labels as ground truth). Quantify the improvement that hybrid search provides over each individual retriever. Does the improvement match the theoretical expectation from the "embedding collapse" discussion in Section 11.3?