AI Pipeline Architecture Model
An advanced assistant capable of ingesting massive, multi-file code repositories. It maps structural code relationships (classes, inheritance, functions) rather than just treating code as raw text chunks.
RAG-005: Repo-Level Code RAG — Engineering Blueprint
This architectural blueprint outlines the system design, AST parsing mechanics, graph-vector database schemas, and retrieval mathematics for RAG-005 (Repo-Level Code RAG). This system treats large-scale codebases as interconnected syntax graphs rather than plain text.
Target Ingest Latency: $< 25\text{ms}$ per file for AST slicing, dependency analysis, and metadata generation.
Retrieval Query SLA: $< 120\text{ms}$ total round-trip latency to query, traverse symbol dependencies up to 2 hops, and assemble context.
Memory Constraints: Under $200\text{MB}$ heap footprint for local IDE background runner nodes.
Accuracy Target: 100% scope integrity (ensuring class, method, and function signatures are parsed completely alongside docstrings and inner code blocks without arbitrary truncation).
Instead of character-limit chunking, this architecture splits source files strictly along syntactic boundaries (nodes) and maps their interactions (edges) into a unified graph-vector memory space.
[ Raw Source Code File (JS/TS, PY, RS) ]
│
▼
[ Tree-sitter Multi-Language AST ]
│
┌────────────────────┴────────────────────┐
▼ ▼
[ Extract Semantic Symbols ] [ Map Structural Graph ]
• Class Signatures • Node: File, Class, Method
• Method Signatures • Edge: IMPORTS, CALLS, EXTENDS
• Inner Function Blocks │
│ │
▼ ▼
[ ChromaDB Vector Store ] [ Neo4j Graph DB ]
Vector: Symbol Embedding Nodes & Edges
3.1 Syntactic Code Slicing (AST Scope Preservation)
Let a code file $F$ be represented as a root syntax node $N_{\text{root}}$ parsed via Tree-sitter. We extract a subset of logical syntax nodes $S = {s_1, s_2, \dots, s_k}$ where each $s_i$ represents a syntactic scope boundary (Class, Method, or Function definition).
The character span of each slice is defined by its exact AST coordinate span:
$$\text{Span}(s_i) = [Row_{\text{start}}(s_i), Row_{\text{end}}(s_i)]$$
Unlike sliding window techniques, syntactic slicing ensures that for any extracted code chunk $C_i$:
$$\forall t \in \text{Tokens}(C_i), \quad t \in \text{Scope}(s_i)$$
This preserves structural and programmatic execution context, eliminating the risk of cutting off class definitions or trailing scopes.
3.2 Hybrid Graph-Vector Proximity Ranking
When retrieving code context for a target query $Q$, we perform a two-step topological search. First, we find the closest semantic symbol match in the vector database:
$$n_{\text{anchor}} = \arg\max_{n \in V_{\text{vector}}} \cos(\mathbf{e}_Q, \mathbf{e}_n)$$
Second, we expand the context query to include the structural neighborhood of $n_{\text{anchor}}$ in the Neo4j graph within a hop boundary $H$. Let the hop distance between two nodes $u$ and $v$ be $\text{dist}G(u, v)$. The structural relevance weight $R(v)$ of a node $v$ relative to $n{\text{anchor}}$ is formulated as:
$$R(v) = \cos(\mathbf{e}Q, \mathbf{e}_v) \cdot \beta^{\text{dist}_G(n{\text{anchor}}, v)}$$
Where:
$\beta \in (0.0, 1.0]$ is the structural hop decay coefficient (typically $\beta = 0.5$).
$\text{dist}G(n{\text{anchor}}, v) \le H$, with a maximum traversal limit $H = 2$.
This Python implementation utilizes Tree-sitter to parse raw codebase files, ChromaDB to store semantic symbol embeddings, Neo4j to represent dependency relationships, and LlamaIndex nodes as the common interface wrapper.
import os from typing import List, Dict, Any, Tuple from tree_sitter import Language, Parser import chromadb from neo4j import GraphDatabase from llama_index.core.schema import TextNode
Language.build_library( 'build/my-languages.so', [ 'vendor/tree-sitter-python', 'vendor/tree-sitter-javascript', 'vendor/tree-sitter-typescript' ] )
PY_LANGUAGE = Language('build/my-languages.so', 'python')
class RepoLevelIndexer: def init(self, neo4j_uri: str, neo4j_auth: Tuple[str, str]): # Setup local-first Vector Storage (ChromaDB) self.chroma_client = chromadb.PersistentClient(path="./chroma_db") self.vector_collection = self.chroma_client.get_or_create_collection("symbol_embeddings")
# Setup structural Graph Storage (Neo4j)
self.neo4j_driver = GraphDatabase.driver(neo4j_uri, auth=neo4j_auth)
# Initialize AST Parser
self.parser = Parser()
self.parser.set_language(PY_LANGUAGE)
def extract_symbols_and_deps(self, file_path: str, source_code: str) -> List[Dict[str, Any]]:
"""
Parses the code with Tree-sitter and extracts functional slices using
custom tree queries (S-expressions) to preserve logical scopes.
"""
tree = self.parser.parse(bytes(source_code, "utf8"))
root_node = tree.root_node
# S-expression query to locate class and function declarations
query_str = """
(class_definition
name: (identifier) @class.name) @class.def
(function_definition
name: (identifier) @function.name) @function.def
"""
query = PY_LANGUAGE.query(query_str)
captures = query.captures(root_node)
slices = []
for node, tag in captures:
if tag in ["class.def", "function.def"]:
start_byte = node.start_byte
end_byte = node.end_byte
code_text = source_code[start_byte:end_byte]
# Fetch identifier info if nested
name = "anonymous"
for child in node.children:
if child.type == "identifier":
name = child.text.decode("utf8") if isinstance(child.text, bytes) else child.text
slices.append({
"name": name,
"type": "Class" if "class" in node.type else "Function",
"code_text": code_text,
"file_path": file_path,
"start_line": node.start_point[0],
"end_line": node.end_point[0],
"hash": hash(code_text)
})
return slices
def commit_to_storage_layer(self, file_path: str, slices: List[Dict[str, Any]], imports: List[str]):
"""
Atomically saves embeddings to ChromaDB and constructs graph
relationships and dependency edges inside Neo4j.
"""
with self.neo4j_driver.session() as session:
# First, ensure File Node exists in Graph
session.run(
"MERGE (f:File {path: $path})",
path=file_path
)
for item in slices:
symbol_id = f"{file_path}::{item['name']}"
# Step A: Generate local embeddings and push to ChromaDB
# (Assuming native ChromaDB client defaults to SentenceTransformers)
self.vector_collection.add(
documents=[item["code_text"]],
metadatas=[{"file_path": file_path, "type": item["type"], "name": item["name"]}],
ids=[symbol_id]
)
# Step B: Write Graph Node and CONTAINS edge to Neo4j
session.run("""
MATCH (f:File {path: $file_path})
MERGE (s:Symbol {id: $symbol_id})
ON CREATE SET s.name = $name, s.type = $type, s.start_line = $start_line, s.end_line = $end_line
ON MATCH SET s.start_line = $start_line, s.end_line = $end_line
MERGE (f)-[:CONTAINS]->(s)
""",
file_path=file_path,
symbol_id=symbol_id,
name=item["name"],
type=item["type"],
start_line=item["start_line"],
end_line=item["end_line"]
)
# Step C: Write internal IMPORTS edges between File nodes
for imp in imports:
session.run("""
MATCH (current:File {path: $current_path})
MERGE (target:File {path: $target_path})
MERGE (current)-[:IMPORTS]->(target)
""",
current_path=file_path,
target_path=imp
)
def close(self):
self.neo4j_driver.close()
This dynamic retrieval query uses LlamaIndex definitions to execute semantic lookup, pull related node dependencies up to 2 degrees of separation, and assemble complete context.
Graph Retrieval Routine
// 1. Anchor node is retrieved via vector ID lookup matching Chromadb's anchor MATCH (anchor:Symbol {id: $anchor_symbol_id})
// 2. Traversed adjacent dependency paths (1 to 2 hops away) MATCH path = (anchor)-[r:CALLS|CONTAINS|IMPORTS*1..2]-(dependency) WITH anchor, path, dependency, length(path) AS hop_distance
// 3. Assemble unique list of adjacent contexts prioritizing closer modules ORDER BY hop_distance ASC WITH anchor, collect(distinct { name: dependency.name, type: labels(dependency)[0], id: dependency.id, hop: hop_distance }) AS connections
RETURN anchor.name AS anchor_name, connections LIMIT 10;
6.1 AST Reconstruction Validation (Fidelity Test)
To prove code block integrity, validation passes are executed to guarantee that any extracted syntactical slices compile cleanly back to the identical source context.
import pytest
def test_ast_slice_preserves_ranges(): indexer = RepoLevelIndexer(neo4j_uri="bolt://localhost:7687", neo4j_auth=("neo4j", "test")) source_code = """ class DataPipeline: def execute(self): print("Running operations") """ # Execute Tree-sitter slicer slices = indexer.extract_symbols_and_deps("pipeline.py", source_code)
# Assert class structure is captured exactly
assert len(slices) == 2 # 1 Class Node + 1 Function Node
assert slices[0]["type"] == "Class"
assert "class DataPipeline:" in slices[0]["code_text"]
assert "def execute(self):" in slices[1]["code_text"]
6.2 Offline Network Security Verification
Ensure that codebase metadata parsing, symbol AST generation, and indexing remain completely local on-device without triggering external data leak channels.
def test_parsing_is_fully_offline(): import socket # Sever standard external socket bindings def run_isolated(): raise ConnectionError("External communication forbidden.")
socket.socket = run_isolated
# Executing the parser should trigger zero outbound network errors
indexer = RepoLevelIndexer(neo4j_uri="bolt://localhost:7687", neo4j_auth=("neo4j", "test"))
slices = indexer.extract_symbols_and_deps("local.py", "def test(): pass")
assert len(slices) >= 0
Advanced Engineering Skills Inventory: Repo-Level Code RAG (RAG-005)
This skills inventory maps out the specialized compiler design, graph database modeling, low-latency IPC communication, local-first memory footprints, and VS Code extension architecture competencies required to build and scale Repo-Level Code RAG systems.
Traditional text-splitting chunkers break code structure by truncating class scopes or isolating functions from their context. Building code RAG requires compiler-level syntax parsing.
[ Raw Multi-File Codebase ]
│
[ Tree-sitter AST Generator ]
• Custom S-Expression Queries
• Scope-Preserving Chunk Slices
Tree-Sitter Engineering:
Compiling, caching, and dynamically binding custom language parser modules inside multi-threaded background runtimes using WASM targets.
Writing highly optimized S-expression query filters (.scm files) to capture class, method, variable, and import/export structures precisely:
$$Q_{\text{AST}} = { \text{class_definition}, \text{function_definition}, \text{import_statement} }$$
Using local parser recovery strategies to maintain AST traversal sanity on partially written, uncompiled, or structurally broken code nodes.
Syntax-Aware Context Slicing:
Extracting complete, self-contained syntax scopes (including decorators, docstrings, typing signatures, and bodies) to ensure zero information loss at the chunk boundary.
Preserving AST token ranges to enable bidirectional code highlighting (editor selection to vector lookup, and vice versa).
Code bases are networks of dependencies, not flat lists. Mapping them requires linking dense semantic representations (vectors) with absolute relations (directed graph edges).
Neo4j/SQLite Code Graph Schema:
Designing multi-tenant graph nodes with strict metadata properties (id, name, type, start_line, end_line, content_hash).
Creating directional relationship edges to model structural and computational connections:
$$R = { \text{CONTAINS}, \text{CALLS}, \text{IMPORTS}, \text{INHERITS_FROM}, \text{DEFINED_IN} }$$
Two-Stage Graph-Vector Retrieval:
Executing initial dense vector queries (using ChromaDB or LanceDB) to find semantic start coordinates (anchor nodes).
Traversing adjacent graph paths up to $2$ hops to pull parent context, class blueprints, and helper functions.
Computing dynamic hop decay weights ($R(v)$) inside the database transaction threads to rank nearby code blocks:
$$R(v) = \cos(\mathbf{e}Q, \mathbf{e}_v) \cdot \beta^{\text{dist}_G(n{\text{anchor}}, v)}$$
Indexing large repositories ($>10,000$ files) on every single keystroke degrades CPU performance and editor responsiveness. You must understand stateful delta-tracking.
[ Local Workspace File Watcher ] ──► [ Local SQLite Hash Manifest ] │ ▼ [ AST Delta Node Diffing Logic ] │ ┌──────────────────┴──────────────────┐ ▼ ▼ [ ChromaDB Vector Upsert ] [ Neo4j Orphan Pruning ]
File System Observers:
Registering highly efficient workspace file system watchers (e.g., using vscode.workspace.createFileSystemWatcher or OS-native handlers) that ignore noise like .git, node_modules, and compilation artifact caches.
Building parallel directory walkers that execute with near-native speeds to parse workspace structures on startup.
Stateful Delta Analysis:
Maintaining a local database manifest tracking relative file paths, timestamps, and overall file SHA-256 checksums.
Executing AST node diffing: when a file changes, computing the SHA-256 hash of each individual symbol slice. Only symbols with new hashes are pushed to the embedding model, reducing API usage and compute overhead by over 90%.
Writing relational cleanups to prune orphan references and stale nodes inside Neo4j when variables or methods are renamed or deleted.
The Model Context Protocol (MCP) acts as the secure API bridge between local IDE context and LLM agents (Claude Desktop, Cursor, Windsurf).
JSON-RPC 2.0 stdio Pipelines:
Building asynchronous, lightweight Python/TypeScript servers communicating via standard input/output (stdio) streams.
Handling high-frequency frame serialization and deserialization cleanly to prevent memory leaks and thread blocks under intense tool-polling runs.
MCP Tool Integration:
Registering codebase semantic search engines as functional schemas (JSON schema formats).
Structuring tool parameters so LLM engines can easily build query structures matching:
search_semantic_symbols(query: str)
get_symbol_dependencies(symbol_name: str)
fetch_file_ast(file_path: str)
To expose your RAG backend to developers, you must build high-performance, sandboxed local editor integrations.
[ VS Code Extension Thread ] ──► [ Webview UI Panel ] ──► [ HTML5 Canvas 2D ] │ • Hierarchical Call Graph ▼ • Visualizes active paths [ Workspace Providers ] • Hover, Diagnostics, Definition
Extension Activation & Thread Lifecycles:
Structuring lazy-loading configuration bindings to ensure the extension has 0ms impact on initial editor boot metrics.
Spawning background processes using low-priority runtime queues to keep keyboard inputs and UI loops running at 60 FPS during indexing runs.
Monitoring active heap memory footprints, enforcing a strict 200MB memory ceiling by evicting stale vector segments and clearing read pools.
Sandboxed Webview Interactions:
Packaging and shipping highly responsive, isolated Webview interfaces using Next.js/React and Tailwind CSS.
Exposing event message gateways (postMessage) to securely stream parsed call-graph coordinates, references, and files between the background extension process and the active Webview panel.
To run this project locally, follow these simple steps designed for both developers and users with no prior programming experience.
pip install -r requirements.txt
.env file in a text editor (like Notepad) and paste your API keys:
GEMINI_API_KEY=your_gemini_api_key_here
run_project.bat or run_pito.bat file, or run the following command:
python app.py (or app_gui.py)