Back to Blog
Jan 18, 20264 min readOnuzulike Anthony

Building Knowledge Graphs in SQLite

Concept nodes, edge tables, recursive CTEs for multi-hop traversal, and how Cognix uses a SQLite graph to map relationships across a knowledge base.

AI/MLKnowledge GraphsSQLiteAI/MLGraph

Graph databases are the obvious choice for relationship data — but they're an operational overhead most local-first applications don't need. SQLite handles graph traversal surprisingly well using recursive CTEs, and for knowledge bases where the graph has thousands of nodes rather than millions, it's the right tool.

Schema

Two tables cover the core use case:

sql
CREATE TABLE concept_mentions (
    id          INTEGER PRIMARY KEY,
    concept     TEXT NOT NULL,
    normalized  TEXT NOT NULL,     -- lowercase, stripped
    chunk_id    INTEGER REFERENCES chunks(id),
    file_id     INTEGER REFERENCES raw_files(id),
    source_path TEXT,
    mention_count INTEGER DEFAULT 1,
    first_seen  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    last_seen   TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
 
CREATE TABLE graph_edges (
    id              INTEGER PRIMARY KEY,
    source_concept  TEXT NOT NULL,
    target_concept  TEXT NOT NULL,
    relationship    TEXT,          -- "related_to", "depends_on", "contradicts", etc.
    weight          REAL DEFAULT 1.0,
    source_file     TEXT,
    source_chunk_id INTEGER REFERENCES chunks(id),
    metadata_json   TEXT
);
 
CREATE INDEX idx_edges_source ON graph_edges(source_concept);
CREATE INDEX idx_edges_target ON graph_edges(target_concept);
CREATE INDEX idx_concepts_norm ON concept_mentions(normalized);

Populating the Graph

In Cognix, the intelligence pipeline scans claims for concept co-occurrence:

python
def extract_concepts(text: str) -> list[str]:
    """Extract noun phrases as concept candidates."""
    words = text.lower().split()
    # Simple bigram + unigram extraction (production: use spaCy NP extraction)
    concepts = []
    for i, word in enumerate(words):
        if len(word) > 4 and word.isalpha():
            concepts.append(word)
        if i < len(words) - 1:
            bigram = f"{words[i]} {words[i+1]}"
            if all(len(w) > 3 for w in bigram.split()):
                concepts.append(bigram)
    return list(set(concepts))
 
def build_edges_from_chunk(chunk_text: str, chunk_id: int, source_path: str, conn):
    concepts = extract_concepts(chunk_text)
    # Co-occurring concepts in the same chunk are "related_to"
    for i, src in enumerate(concepts):
        for tgt in concepts[i+1:]:
            conn.execute("""
                INSERT INTO graph_edges (source_concept, target_concept, relationship, source_chunk_id, source_file)
                VALUES (?, ?, 'related_to', ?, ?)
                ON CONFLICT DO UPDATE SET weight = weight + 0.1
            """, (src, tgt, chunk_id, source_path))

Single-Hop Query

Finding all concepts directly related to a given concept:

python
def get_neighbors(concept: str, conn) -> list[dict]:
    rows = conn.execute("""
        SELECT DISTINCT
            CASE WHEN source_concept = ? THEN target_concept ELSE source_concept END AS neighbor,
            relationship,
            MAX(weight) as weight,
            COUNT(*) as edge_count
        FROM graph_edges
        WHERE source_concept = ? OR target_concept = ?
        GROUP BY neighbor, relationship
        ORDER BY weight DESC, edge_count DESC
        LIMIT 20
    """, (concept, concept, concept)).fetchall()
 
    return [{"concept": r[0], "relationship": r[1], "weight": r[2], "count": r[3]}
            for r in rows]

Multi-Hop Traversal with Recursive CTEs

The real power: following relationships to depth N without writing N nested joins.

sql
WITH RECURSIVE graph_walk(concept, depth, path) AS (
    -- Base case: start from the seed concept
    SELECT ?, 0, ?
 
    UNION ALL
 
    -- Recursive case: expand neighbors
    SELECT
        CASE
            WHEN e.source_concept = gw.concept THEN e.target_concept
            ELSE e.source_concept
        END,
        gw.depth + 1,
        gw.path || ' -> ' || CASE
            WHEN e.source_concept = gw.concept THEN e.target_concept
            ELSE e.source_concept
        END
    FROM graph_edges e
    JOIN graph_walk gw ON (e.source_concept = gw.concept OR e.target_concept = gw.concept)
    WHERE gw.depth < 3        -- max depth 3
      AND gw.path NOT LIKE '%' || CASE
            WHEN e.source_concept = gw.concept THEN e.target_concept
            ELSE e.source_concept
          END || '%'           -- prevent cycles
)
SELECT DISTINCT concept, MIN(depth) as depth, path
FROM graph_walk
WHERE depth > 0
ORDER BY depth, concept;

This returns all concepts reachable within 3 hops, with the shortest path to each. The cycle prevention via NOT LIKE is a common SQLite pattern when you can't use proper visited sets.

Python API

python
def get_concept_neighbors(concept: str, depth: int = 3, conn) -> list[dict]:
    rows = conn.execute("""
        WITH RECURSIVE graph_walk(concept, depth, path) AS (
            SELECT ?, 0, ?
            UNION ALL
            SELECT
                CASE WHEN e.source_concept = gw.concept
                     THEN e.target_concept ELSE e.source_concept END,
                gw.depth + 1,
                gw.path || ' -> ' || CASE WHEN e.source_concept = gw.concept
                     THEN e.target_concept ELSE e.source_concept END
            FROM graph_edges e
            JOIN graph_walk gw ON (e.source_concept = gw.concept OR e.target_concept = gw.concept)
            WHERE gw.depth < ? AND gw.path NOT LIKE '%' || CASE WHEN e.source_concept = gw.concept
                     THEN e.target_concept ELSE e.source_concept END || '%'
        )
        SELECT DISTINCT concept, MIN(depth) as min_depth
        FROM graph_walk WHERE depth > 0
        GROUP BY concept ORDER BY min_depth
    """, (concept, concept, depth)).fetchall()
 
    return [{"concept": r[0], "depth": r[1]} for r in rows]

Weighting and Pruning

Edge weights accumulate with each co-occurrence. Prune low-weight edges periodically to keep the graph sparse:

python
def prune_weak_edges(min_weight: float = 0.5, conn):
    conn.execute("DELETE FROM graph_edges WHERE weight < ?", (min_weight,))
    conn.commit()

When to Move to a Real Graph DB

SQLite graph traversal is fine up to ~100k edges and depth ≤ 5. Beyond that, the NOT LIKE cycle prevention becomes expensive and DFS/BFS algorithms in Python start to win. At that scale, reach for Neo4j or LanceDB's graph capabilities. For a personal knowledge base with a few thousand documents, SQLite handles it without adding an operational dependency.