Graphify-Labs/graphify: Turn any codebase into a queryable knowledge graph

Guide for converting codebases with docs, SQL schemas, configs and PDFs into a local, deterministic, queryable knowledge graph using AST parsing and graph construction.

Средний

Алексей Кузнецов

Системный администратор1 ноября 2026 г.5 мин чтения

```bash
mkdir -p graphify-kg/{raw,processed,graph}
cd graphify-kg
python -m venv venv
source venv/bin/activate
pip install ast tree-sitter networkx
```

```python
# parse_codebase.py
import ast
import json
from pathlib import Path

def extract_entities(tree, filepath):
    entities = []
    for node in ast.walk(tree):
        if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
            entities.append({
                "type": "class" if isinstance(node, ast.ClassDef) else "function",
                "name": node.name,
                "file": str(Path(filepath).relative_to(Path(__file__).parent.parent)),
                "line": node.lineno,
                "col_offset": node.col_offset
            })
    return entities

def main():
    root = Path(__file__).parent.parent
    all_entities = []
    for py_file in root.rglob("*.py"):
        entities = extract_entities(ast.parse(py_file.read_text()), str(py_file))
        all_entities.extend(entities)
    
    # Save to JSON
    with open(root / "processed" / "entities.json", "w") as f:
        json.dump(all_entities, f, indent=2)
    
    print(f"Extracted {len(all_entities)} entities")

if __name__ == "__main__":
    main()
```

```python
# parse_docs_sql.py
import re
from pathlib import Path

def parse_docs(text, filepath):
    """Extract function-like descriptions from documentation."""
    # Find function-like blocks in docs
    pattern = r"(\s*def\s+(\w+)\s*\([^)]*\)\s*->\s*\w+):\s*\n(.*?)(?=\n\S|\n\n)"
    matches = re.findall(pattern, text, re.DOTALL)
    
    entities = []
    for func_name, signature, body in matches:
        entities.append({
            "type": "function",
            "name": func_name,
            "file": str(Path(filepath).relative_to(Path(__file__).parent.parent)),
            "line": 0,  # would need line number extraction
            "description": body.strip()
        })
    return entities

def parse_sql_schema(sql):
    """Extract table and column information from SQL."""
    tables = []
    # Match CREATE TABLE statements
    table_match = re.search(r"CREATE TABLE\s+(\w+)", sql, re.IGNORECASE)
    if table_match:
        table_name = table_match.group(1)
        # Extract column definitions
        cols = re.findall(r"(\w+)\s*\(([^)]+)\)", sql)
        for col, defn in cols:
            tables.append({
                "table": table_name,
                "columns": [col]
            })
    return tables

def main():
    root = Path(__file__).parent.parent
    all_docs = []
    all_sql = []
    
    for doc_file in root.rglob("*.md"):
        all_docs.extend(parse_docs(doc_file.read_text(), doc_file))
    
    for sql_file in root.rglob("*.sql"):
        all_sql.extend(parse_sql_schema(sql_file.read_text()))
    
    # Merge with AST entities
    with open(root / "processed" / "entities.json", "r") as f:
        existing = json.load(f)
    
    # Add doc and SQL entities
    for d in all_docs:
        existing.append({
            "type": "document",
            "name": d["name"],
            "file": str(Path(doc_file).relative_to(Path(__file__).parent.parent)),
            "line": 0,
            "description": d["description"]
        })
    
    for s in all_sql:
        existing.append({
            "type": "schema",
            "name": s["table"],
            "columns": s["columns"],
            "file": str(Path(sql_file).relative_to(Path(__file__).parent.parent))
        })
    
    with open(root / "processed" / "entities.json", "w") as f:
        json.dump(existing, f, indent=2)
    
    print(f"Parsed {len(existing)} total entities")

if __name__ == "__main__":
    main()
```

```python
# build_graph.py
import networkx as nx
from pathlib import Path
import json

def build_knowledge_graph():
    G = nx.DiGraph()
    
    # Add entities from entities.json
    with open("processed/entities.json") as f:
        entities = json.load(f)
    
    for e in entities:
        G.add_node(
            id=str(e["type"] + "_" + e["name"]),
            label=e["name"],
            file=e["file"],
            type=e["type"]
        )
    
    # Add edges based on relationships
    # 1. Function calls (simplified - would need actual call graph)
    # 2. Class inheritance
    # 3. SQL table references in code
    
    # Example: connect function to its file
    for e in entities:
        if e["type"] == "function":
            G.add_edge(
                source=str(e["id"]),
                target=str(e["file"]),
                edge_type="defined_in"
            )
    
    # 4. Table references in SQL
    with open("processed/graph/table_refs.json") as f:
        table_refs = json.load(f)
    
    for ref in table_refs:
        G.add_edge(
            source=str(ref["table"]),
            target=str(ref["file"]),
            edge_type="referenced_by"
        )
    
    # 5. Configuration file references
    with open("processed/graph/configs.json") as f:
        configs = json.load(f)
    
    for cfg in configs:
        G.add_edge(
            source=str(cfg["name"]),
            target=str(cfg["file"]),
            edge_type="configured_by"
        )
    
    # Validate graph
    print(f"Graph has {G.number_of_nodes()} nodes and {G.number_of_edges()} edges")
    
    # Export to RDF (Turtle format)
    return G

if __name__ == "__main__":
    G = build_knowledge_graph()
```

```python
# validate_graph.py
import networkx as nx
from pathlib import Path
import json

def validate_graph(G):
    errors = []
    
    # Check for disconnected components
    components = list(nx.connected_components(G))
    if len(components) > 1:
        errors.append(f"Graph has {len(components)} disconnected components")
    
    # Check node types
    for node, data in G.nodes(data=True):
        if data["type"] not in ["class", "function", "table", "config"]:
            errors.append(f"Unknown node type: {node}")
    
    # Check edge types
    edge_types = set(G.edges(data=dict()).keys())
    expected = {"defined_in", "referenced_by", "configured_by"}
    if not expected.issubset(edge_types):
        errors.append(f"Unexpected edge types: {edge_types - expected}")
    
    if errors:
        print("Validation errors:")
        for e in errors:
            print(f"  - {e}")
        return False
    else:
        print("Graph validation passed")
        return True

def export_graph(G, output_dir):
    # Export as JSON-LD
    with open(output_dir / "knowledge_graph.ld+", "w") as f:
        # Convert to RDF-like structure
        nodes = []
        for node_id, data in G.nodes(data=True):
            nodes.append({
                "type": data["type"],
                "id": node_id,
                "label": data["label"],
                "file": data.get("file", "")
            })
        
        edges = []
        for u, v, data in G.edges(data=True):
            edges.append({
                "from": data["source"],
                "to": data["target"],
                "type": data["edge_type"]
            })
        
        f.write(json.dumps({"nodes": nodes, "edges": edges}, indent=2))
    
    print(f"Exported graph to {output_dir} / knowledge_graph.ld+")

if __name__ == "__main__":
    G = build_knowledge_graph()
    validate_graph(G)
    export_graph(G, Path("processed/graph"))
```

Graphify-Labs/graphify: Turn any codebase into a queryable knowledge graph

Guide for converting codebases with docs, SQL schemas, configs and PDFs into a local, deterministic, queryable knowledge graph using AST parsing and graph construction.

Предварительные требования

• Access to a codebase with source code, documentation, SQL schemas, configuration files, and PDFs
• Local development environment with Python and AST parsing libraries (e.g., ast, tree-sitter)
• Basic understanding of graph databases or knowledge graph construction

Шаг 1: Initialize project structure

Create a dedicated directory for the knowledge graph project. Set up a Python virtual environment and install required parsing libraries (ast, tree-sitter, networkx). Create folders for raw inputs (docs, SQL, configs) and processed outputs (graph, metadata).

```bash
mkdir -p graphify-kg/{raw,processed,graph}
cd graphify-kg
python -m venv venv
source venv/bin/activate
pip install ast tree-sitter networkx

## Шаг 2: Parse source code with deterministic AST

Write a script that walks the codebase and extracts AST nodes. For each class, function, module, and important type annotation, record its name, location, and relationships. Use deterministic traversal (no random ordering) to ensure reproducibility. Store extracted entities in a structured format (JSON or CSV).

```bash
```python
# parse_codebase.py
import ast
import json
from pathlib import Path

def extract_entities(tree, filepath):
    entities = []
    for node in ast.walk(tree):
        if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
            entities.append({
                "type": "class" if isinstance(node, ast.ClassDef) else "function",
                "name": node.name,
                "file": str(Path(filepath).relative_to(Path(__file__).parent.parent)),
                "line": node.lineno,
                "col_offset": node.col_offset
            })
    return entities

def main():
    root = Path(__file__).parent.parent
    all_entities = []
    for py_file in root.rglob("*.py"):
        entities = extract_entities(ast.parse(py_file.read_text()), str(py_file))
        all_entities.extend(entities)
    
    # Save to JSON
    with open(root / "processed" / "entities.json", "w") as f:
        json.dump(all_entities, f, indent=2)
    
    print(f"Extracted {len(all_entities)} entities")

if __name__ == "__main__":
    main()

## Шаг 3: Parse documentation and SQL schemas

Extract structured information from README, docs, and database schema files. For documentation, use regex or AST to find function signatures, parameters, and descriptions. For SQL schemas, parse CREATE TABLE statements and extract column names, types, constraints, and foreign keys. Combine these into a unified entity list.

```bash
```python
# parse_docs_sql.py
import re
from pathlib import Path

def parse_docs(text, filepath):
    """Extract function-like descriptions from documentation."""
    # Find function-like blocks in docs
    pattern = r"(\s*def\s+(\w+)\s*\([^)]*\)\s*->\s*\w+):\s*\n(.*?)(?=\n\S|\n\n)"
    matches = re.findall(pattern, text, re.DOTALL)
    
    entities = []
    for func_name, signature, body in matches:
        entities.append({
            "type": "function",
            "name": func_name,
            "file": str(Path(filepath).relative_to(Path(__file__).parent.parent)),
            "line": 0,  # would need line number extraction
            "description": body.strip()
        })
    return entities

def parse_sql_schema(sql):
    """Extract table and column information from SQL."""
    tables = []
    # Match CREATE TABLE statements
    table_match = re.search(r"CREATE TABLE\s+(\w+)", sql, re.IGNORECASE)
    if table_match:
        table_name = table_match.group(1)
        # Extract column definitions
        cols = re.findall(r"(\w+)\s*\(([^)]+)\)", sql)
        for col, defn in cols:
            tables.append({
                "table": table_name,
                "columns": [col]
            })
    return tables

def main():
    root = Path(__file__).parent.parent
    all_docs = []
    all_sql = []
    
    for doc_file in root.rglob("*.md"):
        all_docs.extend(parse_docs(doc_file.read_text(), doc_file))
    
    for sql_file in root.rglob("*.sql"):
        all_sql.extend(parse_sql_schema(sql_file.read_text()))
    
    # Merge with AST entities
    with open(root / "processed" / "entities.json", "r") as f:
        existing = json.load(f)
    
    # Add doc and SQL entities
    for d in all_docs:
        existing.append({
            "type": "document",
            "name": d["name"],
            "file": str(Path(doc_file).relative_to(Path(__file__).parent.parent)),
            "line": 0,
            "description": d["description"]
        })
    
    for s in all_sql:
        existing.append({
            "type": "schema",
            "name": s["table"],
            "columns": s["columns"],
            "file": str(Path(sql_file).relative_to(Path(__file__).parent.parent))
        })
    
    with open(root / "processed" / "entities.json", "w") as f:
        json.dump(existing, f, indent=2)
    
    print(f"Parsed {len(existing)} total entities")

if __name__ == "__main__":
    main()

## Шаг 4: Build the knowledge graph

Create a graph representation using NetworkX. Nodes represent entities (classes, functions, tables, configs). Edges represent relationships (calls, dependencies, schema relationships). Use deterministic edge ordering to ensure reproducibility. Validate the graph structure and export to a queryable format (e.g., RDF, JSON-LD, or Graph DB format).

```bash
```python
# build_graph.py
import networkx as nx
from pathlib import Path
import json

def build_knowledge_graph():
    G = nx.DiGraph()
    
    # Add entities from entities.json
    with open("processed/entities.json") as f:
        entities = json.load(f)
    
    for e in entities:
        G.add_node(
            id=str(e["type"] + "_" + e["name"]),
            label=e["name"],
            file=e["file"],
            type=e["type"]
        )
    
    # Add edges based on relationships
    # 1. Function calls (simplified - would need actual call graph)
    # 2. Class inheritance
    # 3. SQL table references in code
    
    # Example: connect function to its file
    for e in entities:
        if e["type"] == "function":
            G.add_edge(
                source=str(e["id"]),
                target=str(e["file"]),
                edge_type="defined_in"
            )
    
    # 4. Table references in SQL
    with open("processed/graph/table_refs.json") as f:
        table_refs = json.load(f)
    
    for ref in table_refs:
        G.add_edge(
            source=str(ref["table"]),
            target=str(ref["file"]),
            edge_type="referenced_by"
        )
    
    # 5. Configuration file references
    with open("processed/graph/configs.json") as f:
        configs = json.load(f)
    
    for cfg in configs:
        G.add_edge(
            source=str(cfg["name"]),
            target=str(cfg["file"]),
            edge_type="configured_by"
        )
    
    # Validate graph
    print(f"Graph has {G.number_of_nodes()} nodes and {G.number_of_edges()} edges")
    
    # Export to RDF (Turtle format)
    return G

if __name__ == "__main__":
    G = build_knowledge_graph()

## Шаг 5: Validate and document the knowledge graph

Run validation checks on the graph (no cycles, all nodes reachable, consistent types). Generate a summary report and export the graph in multiple formats (JSON-LD for RDF, Graph DB format). Include a README that explains how to query the graph and its schema.

```bash
```python
# validate_graph.py
import networkx as nx
from pathlib import Path
import json

def validate_graph(G):
    errors = []
    
    # Check for disconnected components
    components = list(nx.connected_components(G))
    if len(components) > 1:
        errors.append(f"Graph has {len(components)} disconnected components")
    
    # Check node types
    for node, data in G.nodes(data=True):
        if data["type"] not in ["class", "function", "table", "config"]:
            errors.append(f"Unknown node type: {node}")
    
    # Check edge types
    edge_types = set(G.edges(data=dict()).keys())
    expected = {"defined_in", "referenced_by", "configured_by"}
    if not expected.issubset(edge_types):
        errors.append(f"Unexpected edge types: {edge_types - expected}")
    
    if errors:
        print("Validation errors:")
        for e in errors:
            print(f"  - {e}")
        return False
    else:
        print("Graph validation passed")
        return True

def export_graph(G, output_dir):
    # Export as JSON-LD
    with open(output_dir / "knowledge_graph.ld+", "w") as f:
        # Convert to RDF-like structure
        nodes = []
        for node_id, data in G.nodes(data=True):
            nodes.append({
                "type": data["type"],
                "id": node_id,
                "label": data["label"],
                "file": data.get("file", "")
            })
        
        edges = []
        for u, v, data in G.edges(data=True):
            edges.append({
                "from": data["source"],
                "to": data["target"],
                "type": data["edge_type"]
            })
        
        f.write(json.dumps({"nodes": nodes, "edges": edges}, indent=2))
    
    print(f"Exported graph to {output_dir} / knowledge_graph.ld+")

if __name__ == "__main__":
    G = build_knowledge_graph()
    validate_graph(G)
    export_graph(G, Path("processed/graph"))