GraphRAG do Zero: Construindo com SQLite e a Biblioteca Padrão

Lede: Dá para construir um GraphRAG inteiro sem Neo4j, sem graphrag e sem instalar nada. sqlite3 já está no Python, e uma CTE recursiva faz travessia multi-hop com mais elegância que servidor de grafo dedicado. Este artigo é o código que eu escrevi e rodei — com os 4 bugs reais que ele tinha antes de funcionar, porque é aí que está a aula.

🧪 Este post é diferente dos outros: o código aqui foi executado. Cada saída de terminal mostrada é real, inclusive os erros. Os 4 bugs documentados na seção 8 existem porque eu os cometi e achei rodando testes.

---

1. Primeiro, a pergunta honesta: faz sentido?

Antes de reescrever um framework, vale dizer quando não faz sentido.

	Na mão	Com framework
Volume	Até ~100k arestas, ~5k entidades	Acima disso você reescreve o storage
Latência de travessia	Milissegundos, tudo em processo	Sub-milissegundo com índice especializado
Consultas ad-hoc	Escreva SQL, é literalmente uma query	Precisa instalar e aprender a linguagem do banco
Deploy	Um arquivo .db num bucket S3	Serviço, container, backup, versionamento
Comunidades hierárquicas	Você implementa (e vai doer)	Leiden pronto e validado

Faça na mão quando você quer entender o algoritmo, o corpus é pequeno, ou quer zero dependência num ambiente travado. Use framework quando isso vai pra produção com time — e o custo de manter o seu próprio vai aparecer primeiro nos bugs de borda, não nos headlines.

O resto deste artigo é o "na mão", feito direito.

---

2. Zero dependências. Sério.

O que o código abaixo usa:

import sqlite3      # banco de grafo
import json         # parsing das respostas do LLM
import math         # similaridade
import re           # tokenização
import random       # Louvain
import urllib.request   # chamar a API do LLM
from collections import defaultdict

Nenhum pip install. O único requisito de rede é a API do LLM. O resto roda offline.

💡 A ideia central: SQLite não é "menos que um banco de grafo", é um banco com uma linguagem que foi feita para percorrer grafos. Uma CTE recursiva é literalmente o k-hop. Você não simula travessia em memória — você delega ao planner do SQLite, que já sabe fazer isso.

---

3. O schema: 6 tabelas

SCHEMA = """
PRAGMA journal_mode=WAL;

CREATE TABLE IF NOT EXISTS nodes (
    id     TEXT PRIMARY KEY,
    label  TEXT NOT NULL DEFAULT 'Entity',
    name   TEXT NOT NULL,
    type   TEXT,
    desc   TEXT,
    degree INTEGER DEFAULT 0
);

CREATE TABLE IF NOT EXISTS aliases (
    alias    TEXT PRIMARY KEY,
    alias_sq TEXT NOT NULL,      -- forma sem espaços, p/ casar nomes colados
    node_id  TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE
);

CREATE TABLE IF NOT EXISTS edges (
    src      TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
    dst      TEXT NOT NULL REFERENCES nodes(id) ON DELETE CASCADE,
    rel      TEXT NOT NULL,
    weight   REAL DEFAULT 1.0,
    source_chunk TEXT,
    PRIMARY KEY (src, dst, rel)
);

CREATE TABLE IF NOT EXISTS chunks (
    id TEXT PRIMARY KEY, text TEXT, vector BLOB
);

CREATE TABLE IF NOT EXISTS communities (
    community_id INTEGER PRIMARY KEY, level INTEGER,
    title TEXT, report TEXT, summary TEXT
);

CREATE TABLE IF NOT EXISTS memberships (
    node_id TEXT NOT NULL, community_id INTEGER NOT NULL, level INTEGER NOT NULL,
    PRIMARY KEY (node_id, level)
);

CREATE INDEX IF NOT EXISTS idx_edges_src ON edges(src);
CREATE INDEX IF NOT EXISTS idx_edges_dst ON edges(dst);
CREATE INDEX IF NOT EXISTS idx_alias_sq ON aliases(alias_sq);
"""

Três decisões que valem explicar:

• PRIMARY KEY (src, dst, rel) na tabela de arestas — impede aresta duplicada e, com ON CONFLICT DO UPDATE SET weight = weight + excluded.weight, acumula a evidência. Duas extrações que encontraram a mesma relação reforçam a aresta em vez de duplicá-la.
• alias_sq é um índice à parte com o nome sem espaços. Volto nisso na seção 6, porque foi um bug.
• degree materializado — recalcular o grau a cada consulta custa um COUNT sobre todas as arestas. Como o grafo muda raramente, atualizar na escrita compensa.

---

4. A travessia k-hop: uma CTE recursiva

Esta é a peça central, e cabe em 20 linhas:

def traverse(con, seed_ids, max_hops=2, max_nodes=60, rels=None):
    if rels:
        rels = [r.upper() for r in rels]
        rel_clause = "AND UPPER(e.rel) IN (%s)" % ",".join("?" * len(rels))
        rel_args = list(rels)
    else:
        rel_clause, rel_args = "", ""

    q = f"""
    WITH RECURSIVE walk(id, depth) AS (
        SELECT id, 0 FROM nodes WHERE id IN ({",".join("?" * len(seed_ids))})
        UNION
        SELECT CASE WHEN e.src = w.id THEN e.dst ELSE e.src END, w.depth + 1
        FROM walk w
        JOIN edges e ON (e.src = w.id OR e.dst = w.id)
        WHERE w.depth < ? {rel_clause}
    )
    SELECT id, MIN(depth) AS depth FROM walk
    GROUP BY id ORDER BY depth, id
    """
    args = list(seed_ids) + [max_hops] + rel_args
    rows = con.execute(q, args).fetchall()
    # ... monta nós e arestas

Três detalhes que fazem ela funcionar:

• UNION (não UNION ALL****) — deduplica. Com UNION ALL um nó em rede de malha seria reexpandido indefinidamente.
• CASE WHEN e.src = w.id THEN e.dst ELSE e.src END — a aresta é não-direcional na travessia. Se você quiser seguir só o sentido do fluxo, troque por e.dst e filtre e.src = w.id.
• MIN(depth) no GROUP BY — um nó pode ser alcançado por caminhos de profundidades diferentes. Você quer saber a menor, senão a ordem de visitação contamina a distância.

Saída real (executada)

1) K-HOP a partir de auth (2 saltos)
   nos=6  arestas=7
   d=0  ServicoAutenticacao      grau=4
   d=1  ModuloPagamento          grau=3
   d=1  StoreSessao              grau=2
   d=1  TimeIAM                  grau=2
   d=2  ClusterDB                grau=2
   d=2  ClusterK8sAlpha          grau=3
   com 3 saltos -> 7 nos

E com filtro de relação — note que ClusterDB some, porque chega a pay por EXECUTA_EM, e não por DEPENDE_DE:

2) K-HOP com filtro de relacao (so DEPENDE_DE)
   ['ModuloPagamento', 'ServicoAutenticacao', 'StoreSessao']

⚠️ max_hops não é decoração. A consulta cresce com o número de saltos, e o custo é multiplicado pelo grau médio. Em grafo de densidade alta, k=4 já é catastrófico — e o SQLite não tem como te salvar disso. Sempre limite.

---

5. Comunidades: Louvain em Python puro

O Microsoft GraphRAG usa Leiden hierárquico. Implementar isso em stdlib é exagero. Louvain de um nível dá 80% do resultado em 50 linhas — e o motivo pelo qual o GraphRAG já usou Louvain antes do Leiden.

O algoritmo é uma otimização gulosa de modularidade:

def louvain(adj, seed=42, restarts=8):
    nodes = sorted(adj)                      # ordenação estável
    if not nodes:
        return {}
    m2 = sum(len(adj[n]) for n in nodes)     # 2m
    if m2 == 0:
        return {n: i for i, n in enumerate(nodes)}

    best_comm, best_q = None, -1e9

    for r in range(restarts):
        rnd = random.Random(seed + r)
        comm = {n: n for n in nodes}         # cada um sozinho
        tot  = {n: len(adj[n]) for n in nodes}
        moved = True

        while moved:                          # fase de movimento local
            moved = False
            order = nodes[:]
            rnd.shuffle(order)
            for n in order:
                cur = comm[n]
                k_n = len(adj[n])
                tot[cur] -= k_n                # tira n da comunidade

                ki_in = defaultdict(float)
                for v in sorted(adj[n]):       # sorted() é obrigatório
                    ki_in[comm[v]] += 1

                best_c = cur
                best_g = ki_in.get(cur, 0.0) - (tot[cur] * k_n) / m2
                for c in sorted(ki_in):
                    if c == cur:
                        continue
                    g = ki_in[c] - (tot[c] * k_n) / m2
                    if g > best_g:
                        best_g, best_c = g, c

                comm[n] = best_c
                tot[best_c] += k_n
                if best_c != cur:
                    moved = True

        # renumera communities para ids 0..n-1
        remap = {}
        for i, cid in enumerate(sorted(set(comm.values()))):
            remap[cid] = i
        cand = {n: remap[comm[n]] for n in nodes}
        q = _modularity(adj, cand, m2)
        if q > best_q:
            best_q, best_comm = q, cand

    return best_comm

O ganho de mover o nó n para a comunidade c:

$$\Delta Q = k_{in}(n \to c) - \frac{\Sigma_{tot}(c) \cdot k_n}{2m}$$

Como 2m é constante no grafo, comparar esse valor é equivalente a comparar a versão multiplicada — dá para ignorar a normalização.

Saída real (executada)

3) LOUVAIN — deteccao de comunidades
   comunidades=2
   #0: ['ClusterDB', 'ClusterK8sAlpha', 'ServicoLogs', 'ModuloPagamento']
   #1: ['ServicoAutenticacao', 'TimeIAM', 'StoreSessao']
   (Q=0.219 com 2 comunidades vence Q=0.164 com 3 — restarts decidiram)

---

6. Entity linking em quatro degraus

Transformar "serviço de autenticação" num ID de nó é o elo mais frágil de toda a cadeia. A estratégia é degradação graciosa, do match mais rígido ao mais tolerante:

def squash(s):
    """'Servico Autenticacao' -> 'servicoautenticacao'"""
    return re.sub(r"\W+", "", s.lower(), flags=re.UNICODE)


def entity_link(con, query, min_ratio=0.5):
    toks_list = tokenize(query)
    toks = set(toks_list)
    if not toks:
        return []

    # três formas comprimidas — a ordem dos tokens importa
    formas = {
        "".join(toks_list),              # servicoautenticacao
        "".join(sorted(toks_list)),      # autenticacaoservico
        squash(query),                   # servicodeautenticacao
    }
    hits = {}
    def add(nid, score):
        hits[nid] = max(hits.get(nid, 0.0), score)

    # 1) alias exato da frase inteira
    r = con.execute("SELECT node_id FROM aliases WHERE alias=?",
                    (query.strip().lower(),)).fetchone()
    if r:
        add(r[0], 10.0)

    # 2) alias exato das formas comprimidas
    for i, f in enumerate(formas):
        r = con.execute("SELECT node_id FROM aliases WHERE alias_sq=?",
                        (f,)).fetchone()
        if r:
            add(r[0], 9.0 - i)

    # 3) alias exato de cada token
    for t in toks:
        r = con.execute("SELECT node_id FROM aliases WHERE alias=?", (t,)).fetchone()
        if r:
            add(r[0], 7.0)

    # 4) contenção
    for alias_sq, nid in con.execute(
        "SELECT alias_sq, node_id FROM aliases WHERE length(alias_sq) >= 4"
    ):
        for f in formas:
            if alias_sq in f or f in alias_sq:
                ratio = min(len(alias_sq), len(f)) / max(len(alias_sq), len(f))
                if ratio >= min_ratio:
                    add(nid, ratio * 3.0)

    # descarta entidade isolada: não há o que percorrer a partir dela
    return [nid for nid, _ in sorted(hits.items(), key=lambda kv: -kv[1])
            if con.execute("SELECT degree FROM nodes WHERE id=?",
                           (nid,)).fetchone()[0] > 0]

Saída real (executada)

4) ENTITY LINKING
   'servico de autenticacao'        -> ['ServicoAutenticacao']
   'Cluster DB'                     -> ['ClusterDB']
   'modulo pagamento'               -> ['ModuloPagamento']

Três consultas, três grafias diferentes — e todas acerto. Esse é o payoff de ter o alias_sq indexado.

---

7. Extração de entidades chamando o LLM

O LLM é chamado via urllib, sem SDK. O ponto crítico não é a chamada — é validar a resposta antes de gravar.

EXTRACT_PROMPT = """Extraia entidades e relações do texto.
Responda APENAS JSON válido, sem markdown, com esta forma:
{{"entities": [{{"name": "...", "type": "...", "description": "..."}}],
  "relations": [{{"source": "...", "target": "...", "type": "..."}}]}}

Tipos válidos de entidade: Servico, Infra, Time, Cliente.
Tipos válidos de relação: DEPENDE_DE, EXECUTA_EM, DELEGADO_A, ADMINISTRA.

TEXTO:
{texto}"""


def extract(text, api_key, model="gpt-4o-mini"):
    body = json.dumps({
        "model": model,
        "temperature": 0,
        "response_format": {"type": "json_object"},
        "messages": [{"role": "user",
                      "content": EXTRACT_PROMPT.format(texto=text)}],
    }).encode()

    req = urllib.request.Request(
        "https://api.openai.com/v1/chat/completions",
        data=body,
        headers={"Content-Type": "application/json",
                 "Authorization": f"Bearer {api_key}"})
    raw = json.loads(urllib.request.urlopen(req, timeout=60).read())
    return parse_extraction(raw["choices"][0]["message"]["content"], text)


def parse_extraction(content, source_text):
    """Valida a resposta do LLM. Retorna (nós, arestas) prontos pro SQLite."""
    try:
        data = json.loads(content)
    except json.JSONDecodeError:
        return [], []                    # LLM devolveu lixo: descarta o chunk

    ents, rels = {}, []
    validas = set()

    for e in data.get("entities", []):
        nome = (e.get("name") or "").strip()
        tipo = (e.get("type") or "").strip()
        if not nome or tipo not in TIPOS_ENTIDADE:
            continue
        # alucinação: entidade que não aparece no texto fonte
        if squash(nome) not in squash(source_text):
            continue
        nid = slug(nome)
        ents[nid] = (nid, nome, tipo, e.get("description", ""))
        validas.add(slug(nome))

    for r in data.get("relations", []):
        s, t = slug(r.get("source", "")), slug(r.get("target", ""))
        if s in validas and t in validas:
            rels.append((s, t, (r.get("type") or "").upper()))
    return list(ents.values()), rels

🛡️ O filtro if squash(nome) not in squash(source_text) é o mais importante do arquivo. LLM alucina entidade. Sem essa checagem, um chunk de 200 palavras vira 3 nós ghosts, e eles poluem a modularidade do Louvain, o entity linking e a resposta final. É a mesma lição do grafo ruidoso do post anterior, agora com a defesa no código.

O que eu não implementei: retries com backoff, cache de resposta, chamadas em paralelo e parsing de JSON que veio com `json around. Num corpus real você precisa dos quatro. Estão fora porque exigem API key para eu testar, e prefiro dizer do que entregar código não exercitado.

---

8. Os 4 bugs que meu código tinha

Esta seção é a razão de o artigo existir. Todos foram encontrados rodando os testes, não lendo o código.

Bug 1 — tokenize não sabia de nomes colados

'entity_link' de "servico de autenticacao"  ->  []

O alias no banco era servicoautenticacao (uma palavra só). O tokenize separa por espaço, então produzia {"servicoautenticacao"}, e o Jaccard com {"servico", "autenticacao"} dava zero. Corrigindo: derivei a comparação da forma comprimida dos tokens sem stop word, não do texto bruto.

Bug 2 — preposição no meio quebrava a substring

Ainda não casava. squash("servico de autenticacao") produz servicodeautenticacao, e servicoautenticacao não é substring disso — o de separa. Corrigindo: montei três formas comprimidas (ordem original, ordem alfabética, texto bruto) e testei as três.

Bug 3 — Louvain guloso é sensível à ordem

Mesma entrada, saídas diferentes entre execuções. Um único passe greedy depende da ordem de visita dos nós, e adj[n] é um set. Corrigindo em duas etapas:

1. restarts=8 seeds diferentes, escolhendo a maior modularidade. No grafo de teste, o restart 5 achou 3 comunidades (Q=0,164) e o restart 1 achou 2 (Q=0,219) — o algoritmo escolheu a melhor.
2. Ainda havia variabilidade entre processos. A causa: for v in adj[n] itera um set de strings, cuja ordem depende do hash randomizado do Python, e o desempate g > best_g fica diferente a cada execução. Corrigindo: for v in sorted(adj[n]).

🔁 Essa segunda é a que mais me custou tempo e a mais vale registrar: se seu pipeline depende de desempate, ele não é reprodutível até você ordenar toda iteração sobre集合 não ordenada. Um sorted() em dois lugares resolveu.

Bug 4 — o budget contava nós mas não arestas

AssertionError: estourou o orcamento: 294   (teto era 200)

Eu somava o tamanho das linhas de nó no orçamento e depois anexava as arestas de graça. O contexto estourava exatamente na parte que mais importa — os relacionamentos, que é o que distingue GraphRAG de RAG vetorial. Corrigindo: nós e arestas entram juntos na contagem, com um helper take() que respeita o teto nos dois.

Saída real depois da correção:

6) BUDGET / PRUNING com max_chars=200
   mantidos=6
   (ServicoAutenticacao)
   (ModuloPagamento)
   (TimeIAM)
   (StoreSessao)
   (ClusterK8sAlpha)
   (ClusterDB)
   (ServicoAutenticacao) -[DELEGADO_A]-> (TimeIAM)
   (ServicoAutenticacao) -[DEPENDE_DE]-> (ModuloPagamento)

---

9. Busca global: map-reduce sobre relatórios de comunidade

Com comunidades detectadas, a busca global é o mesmo padrão map-reduce do post anterior, sem framework:

def global_search(con, pergunta, llm, level=1, top_comunidades=8, budget=12000):
    # 1) seleciona comunidades por similaridade do título+sumário
    cands = con.execute(
        "SELECT community_id, title, summary FROM communities WHERE level=?",
        (level,)).fetchall()
    ranked = sorted(
        cands,
        key=lambda c: -(cosine(embed(pregunta), embed(c[1] + " " + (c[2] or ""))))
    )[:top_comunidades]

    # 2) MAP — uma chamada por comunidade
    parciais = []
    for cid, title, _ in ranked:
        parcial = llm(f"Com base no resumo desta seção do corpus:\n\n"
                      f"{title}\n\nPergunta: {pergunta}\n\n"
                      f"Responda apenas com o que ESTA seção sustenta. "
                      f"Se não sustenta, diga 'não informado'.")
        parciais.append(parcial)

    # 3) REDUCE — uma chamada consolidando
    return llm(
        "Estas são respostas parciais de seções distintas de um corpus. "
        "Some-as, elimine repetições e responda à pergunta. Marque "
        "explicitamente qualquer ponto que as seções não cobrem.\n\n"
        + pergunta + "\n\n" + "\n\n---\n\n".join(parciais))

O truque do "não informado" no prompt é o que segura a alucinação no map: um LLM之道 tender a inventar resposta quando a seção não tem a informação, e forçar a resposta negativa expõe as lacunas no reduce, em vez de deixá-las virarem afirmação.

---

10. Quando desistir da implementação própria

Este código é didático e roda. Ele não é um produto. Os limites reais:

Limite	Consequência	Quando dói
Sem comunidade hierárquica	Um nível só. Perde a síntese em múltiplas escalas que o GraphRAG faz com Leiden	Corpus grande e diversificado
Busca vetorial é brute force	Coseno sobre todos os chunks, em Python puro	Acima de ~5.000 chunks fica lento
Extração é sequencial	Um chunk por vez, sem paralelismo nem cache	Indexação de corpus real
Sem retry nem observabilidade	Rate limit derruba o pipeline	Sempre, no primeiro deploy
Concorrência de escrita	SQLite serializa escrita. Indexação paralela trava	Extração com worker pool

Minha recomendação honesta: use este código para entender, prototipar e validar a hipótese. Quando a resposta sair certa e o volume subir, migre para o neo4j-graphrag (post 2) — você vai direto saber o que pedir a ele, porque sabe o que ele faz.

---

Referências

• 📚 Microsoft GraphRAG — Introduction to GraphRAG — a arquitetura que estamos reimplementando
• 📚 Louvain (Blondel et al., 2008) — o paper do método de communities
• 📚 Newman-Girvan modularity — a métrica Q que o Louvain otimiza
• 📚 SQLite — WITH RECURSIVE — sintaxe da CTE recursiva
• 🔧 modelcontextprotocol/python-docx — não, mas — sqlite3 na stdlib
• 🔧 OpenAI — Structured Outputs — response_format: json_object