Data Flow Architecture from Source Code to Graph Queries in code-graph-rag

The code-graph-rag pipeline transforms source code into executable graph queries through a 10-phase architecture spanning CLI orchestration, Tree-sitter AST parsing, buffered Cypher ingestion, and LLM-powered query generation.

The vitali87/code-graph-rag project provides a complete data flow architecture from source code to graph queries, enabling developers to index entire codebases into Memgraph and interrogate them using natural language. This system combines Tree-sitter parsing with hybrid language front-ends to build a comprehensive static code model before leveraging LLMs to generate validated Cypher statements against the graph.

Phase 1: CLI Entry and Repository Structure Discovery

The pipeline initiates when a user invokes cgr start or cgr index from the command line. In codebase_rag/cli.py, the _run_graph_sync function parses CLI options, resolves the repository path, and establishes a Memgraph connection.

Immediately after, codebase_rag/parser_loader.py caches language-specific Tree-sitter parsers (and optional front-ends like LibClang or Roslyn) to avoid repeated initialization overhead.

Next, the StructureProcessor class in codebase_rag/parsers/structure_processor.py executes the identify_structure method. This component walks the repository tree and creates hierarchical nodes—PROJECT, PACKAGE, FOLDER, and FILE—while establishing CONTAINS relationships between them.

Phase 2: AST Parsing and Multi-Pass Processing

With the repository structure mapped, codebase_rag/graph_updater.py coordinates the core parsing logic through _process_files. This phase implements a two-pass strategy:

1. Pass 1: The ProcessorFactory creates language-specific processors that parse each source file into an abstract syntax tree (AST).
2. Pass 2: Definitions (functions, classes, modules) become graph nodes, while calls, imports, and inheritance become relationships.

For languages requiring deeper semantic analysis, the system runs incremental hybrid front-ends. The GraphUpdater calls _run_cpp_frontend or _run_csharp_frontend to capture macro-expansion information and semantic facts that Tree-sitter alone cannot extract. Deferred edges—such as macro-generated calls—are resolved after the main passes complete.

Phase 3: Filtering and Bulk Ingestion to Memgraph

Before database insertion, the FilteringIngestor (a wrapper around MemgraphIngestor defined in codebase_rag/services/graph_service.py) filters out disabled node types and relationships according to the capture configuration.

The MemgraphIngestor buffers entities until the configured batch_size threshold is reached. When flushing, the _flush_node_label_group method converts buffered batches into optimized Cypher MERGE and CREATE statements. These execute via a thread-pooled executor against the Memgraph connection established in Phase 1.

Once ingestor.flush_all() completes in graph_updater.py: run, the graph represents the full static code model including files, symbols, call graphs, and imports.

Phase 4: Natural Language to Cypher Translation

The query phase begins when the user submits a natural language question through the interactive REPL or RPC interface. The create_rag_orchestrator function in codebase_rag/services/llm.py initializes the RAG pipeline.

The CypherGenerator.generate method sends the prompt to the configured LLM and receives a raw Cypher response. This undergoes sanitization via _clean_cypher_response and safety validation through _validate_cypher_read_only to ensure only read operations execute against the database.

Finally, the validated Cypher runs against Memgraph using QueryProtocol.execute_write or fetch_all from services/graph_service.py, returning the result set to the user or feeding it back into the agent for tool calls.

Practical Implementation Examples

Ingesting a Repository via CLI

cgr start --repo-path /path/to/project --update-graph

Under the hood, cli.py invokes _run_graph_sync, instantiates MemgraphIngestor, loads parsers via parser_loader.py, and executes GraphUpdater.run() to complete the 10-phase pipeline.

Interactive Natural Language Queries

cgr start --repo-path /path/to/project
> Which function calls `process_data`?

The orchestrator processes this through CypherGenerator.generate:

cypher = await CypherGenerator(active_projects=["myproject"]).generate(
    "Which function calls `process_data`?"
)

# Cleaned output: MATCH (c:Function)-[:CALLS]->(t:Function {name: "process_data"}) RETURN c.name;

The system executes this via MemgraphIngestor.fetch_all and displays the matching nodes.

Programmatic Ingestion

from codebase_rag.services.graph_service import MemgraphIngestor
from codebase_rag.graph_updater import GraphUpdater
from codebase_rag.parser_loader import load_parsers
from pathlib import Path

repo = Path("/path/to/project")
with connect_memgraph(batch_size=500) as ingestor:
    parsers, queries = load_parsers()
    updater = GraphUpdater(
        ingestor=ingestor,
        repo_path=repo,
        parsers=parsers,
        queries=queries,
        project_name="myproject"
    )
    updater.run()

Key calls include connect_memgraph establishing the database session and updater.run() executing the Pass 1–Pass 3 pipeline.

Direct Cypher Generation

from codebase_rag.services.llm import CypherGenerator

gen = CypherGenerator(active_projects=["myproject"])
cypher = await gen.generate("List all routes exposed by the Flask app.")
print(cypher)

# Output: MATCH (r:Endpoint)-[:EXPOSES]->(h:Handler) RETURN r.path, h.name;

Summary

• The data flow architecture from source code to graph queries in code-graph-rag follows a strict 10-phase pipeline from CLI invocation to Cypher execution.
• cli.py and parser_loader.py handle initialization, while structure_processor.py creates the repository hierarchy.
• graph_updater.py orchestrates AST parsing via Tree-sitter and hybrid front-ends, producing nodes and relationships.
• MemgraphIngestor in services/graph_service.py buffers and flushes data using _flush_node_label_group for efficient bulk inserts.
• CypherGenerator.generate in services/llm.py transforms natural language into validated Cypher using _clean_cypher_response and _validate_cypher_read_only.

Frequently Asked Questions

How does code-graph-rag handle multiple programming languages?

The system uses parser_loader.py to initialize Tree-sitter parsers for each supported language. For C++, C#, and Go, it runs additional front-end passes via graph_updater.py methods like _run_cpp_frontend to capture semantic information that Tree-sitter cannot extract, ensuring accurate graph representation across language boundaries.

What database does code-graph-rag use for storing the code graph?

The architecture targets Memgraph as its graph database. The MemgraphIngestor class in services/graph_service.py manages all database interactions, converting internal node and relationship objects into Cypher MERGE and CREATE statements before executing them through a thread-pooled connection.

How does the system ensure generated Cypher queries are safe?

Safety validation occurs in services/llm.py within the CypherGenerator class. After cleaning the LLM response with _clean_cypher_response, the system runs _validate_cypher_read_only to check for dangerous keywords and ensure only read operations execute against the production graph, preventing accidental data modification during natural language queries.

Can I use code-graph-rag programmatically without the CLI?

Yes. You can import GraphUpdater and MemgraphIngestor directly from codebase_rag.graph_updater and codebase_rag.services.graph_service. By calling load_parsers() and instantiating GraphUpdater with a configured ingestor, you can trigger updater.run() to index repositories entirely within Python scripts, bypassing the cgr CLI entry point.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}