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

> Explore the 10-phase data flow architecture in code-graph-rag, transforming source code into graph queries via AST parsing, Cypher ingestion, and LLM generation. Understand the pipeline from code to queries.

- Repository: [Vitali Avagyan/code-graph-rag](https://github.com/vitali87/code-graph-rag)
- Tags: architecture
- Published: 2026-08-13

---

**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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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

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

```

Under the hood, [`cli.py`](https://github.com/vitali87/code-graph-rag/blob/main/cli.py) invokes `_run_graph_sync`, instantiates `MemgraphIngestor`, loads parsers via [`parser_loader.py`](https://github.com/vitali87/code-graph-rag/blob/main/parser_loader.py), and executes `GraphUpdater.run()` to complete the 10-phase pipeline.

### Interactive Natural Language Queries

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

```

The orchestrator processes this through `CypherGenerator.generate`:

```python
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

```python
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

```python
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`](https://github.com/vitali87/code-graph-rag/blob/main/cli.py) and [`parser_loader.py`](https://github.com/vitali87/code-graph-rag/blob/main/parser_loader.py) handle initialization, while [`structure_processor.py`](https://github.com/vitali87/code-graph-rag/blob/main/structure_processor.py) creates the repository hierarchy.
- [`graph_updater.py`](https://github.com/vitali87/code-graph-rag/blob/main/graph_updater.py) orchestrates AST parsing via Tree-sitter and hybrid front-ends, producing nodes and relationships.
- `MemgraphIngestor` in [`services/graph_service.py`](https://github.com/vitali87/code-graph-rag/blob/main/services/graph_service.py) buffers and flushes data using `_flush_node_label_group` for efficient bulk inserts.
- `CypherGenerator.generate` in [`services/llm.py`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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`](https://github.com/vitali87/code-graph-rag/blob/main/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.