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.
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:
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:
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:
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.
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
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:
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.