Voltar ao blog
12 de maio de 20264 min de leitura

Fusão Automática de Mundos: Quando Dois Contextos se Tornam Um

pythonaialgorithms

Dois mundos no seu grafo de conhecimento falam sobre a mesma coisa mas com nomes diferentes. Um se chama "Machine Learning" e outro "Aprendizado de Máquina". Ambos têm triplas sobre redes neurais, gradient descent e overfitting. Manter os dois separados é desperdício: duplica informação, fragmenta contexto, e impede que inferências cruzem os dados de ambos. A solução é detectar automaticamente quando dois mundos convergiram e fundi-los em um só, combinando seus grafos de forma inteligente.

Implementei fusão automática que roda periodicamente, detecta pares com similaridade coseno acima de 0.94, combina seus grafos com votação de triplas (triplas confirmadas por ambos ganham boost de confiança), recalcula o centroide, e remove o mundo redundante. O resultado é um grafo que se auto-organiza sem intervenção manual.

Guia de tópicos:

  • Por Que Mundos Duplicados Aparecem
  • Detectando Pares para Fusão
  • Votação de Triplas: Resolvendo Conflitos
  • Compactação do Índice Vetorial
  • Exemplo Prático em Python
  • Considerações Finais

Por Que Mundos Duplicados Aparecem

Mundos duplicados surgem naturalmente. O usuário fala sobre o mesmo tema usando vocabulário diferente em conversas separadas. O threshold de criação pode não capturar a similaridade se as primeiras mensagens foram muito diferentes. Ou o tema evoluiu gradualmente: "Python básico" e "Python avançado" começam distantes mas convergem conforme acumulam triplas sobre a mesma linguagem.

A fragmentação não é um bug. É consequência natural de um sistema que cria mundos sob demanda sem taxonomia pré-definida. O importante é ter consolidação automática que resolve a duplicação quando ela se torna evidente.

Detectando Pares para Fusão

A detecção usa o índice vetorial para eficiência. Para cada mundo, busca os vizinhos mais próximos. Se algum par tem similaridade acima de 0.94, é candidato a fusão. O threshold é alto de propósito: você não quer fundir mundos que são apenas "parecidos". Eles precisam ser essencialmente o mesmo tema para justificar a fusão.

A verificação roda a cada N mensagens (no meu caso, a cada 5). Não precisa ser a cada mensagem porque fusão é uma operação relativamente cara e mundos não convergem instantaneamente.

Votação de Triplas: Resolvendo Conflitos

Quando ambos os mundos têm triplas para o mesmo par (sujeito, verbo), a votação decide. Se ambos concordam no objeto dominante, a confiança recebe um boost (confirmação por múltiplas fontes). Se discordam, ambas as hipóteses são mantidas na distribuição com suas respectivas probabilidades.

Isso é mais sofisticado que "pegar a de maior confiança". Manter alternativas preserva a incerteza real. Se um mundo diz "interpretado" com 0.7 e outro diz "compilado" com 0.5, a distribuição resultante reflete essa discordância em vez de silenciá-la.

Compactação do Índice Vetorial

Após várias fusões, o índice vetorial acumula "tombstones" (slots de mundos deletados). Quando a proporção passa de 40%, o índice é reconstruído do zero com apenas os mundos vivos. Isso mantém performance de busca e reduz uso de memória. É O(n) mas só acontece raramente.

Exemplo Prático em Python

import numpy as np

def cosine(a, b):
    n = np.linalg.norm(a) * np.linalg.norm(b)
    return float(np.dot(a, b) / n) if n > 0 else 0.0


def find_fusion_candidates(worlds: dict, threshold=0.94) -> list[tuple]:
    """Find pairs of worlds similar enough to merge."""
    pairs = []
    ids = list(worlds.keys())
    seen = set()
    for i in range(len(ids)):
        for j in range(i+1, len(ids)):
            sim = cosine(worlds[ids[i]]["centroid"], worlds[ids[j]]["centroid"])
            if sim > threshold:
                pair = (ids[i], ids[j])
                if pair not in seen:
                    seen.add(pair)
                    pairs.append((*pair, sim))
    return pairs


def vote_triples(triples_a: list[dict], triples_b: list[dict]) -> list[dict]:
    """Combine triples from two worlds with vote-based confidence boost."""
    vote_map = {}
    for t in triples_a + triples_b:
        key = (t["s"], t["v"], t["o"])
        if key not in vote_map:
            vote_map[key] = []
        vote_map[key].append(t["c"])

    merged = []
    for (s, v, o), confidences in vote_map.items():
        # Boost for confirmation by multiple sources
        bonus = min(0.05 * (len(confidences) - 1), 0.20)
        final_c = float(np.clip(np.mean(confidences) + bonus, 0, 0.95))
        merged.append({"s": s, "v": v, "o": o, "c": final_c,
                       "sources": len(confidences)})
    return merged


def merge_worlds(w1: dict, w2: dict) -> dict:
    """Merge w2 into w1, returning the combined world."""
    w1["triples"] = vote_triples(w1["triples"], w2["triples"])
    w1["centroid"] = (w1["centroid"] + w2["centroid"]) / 2.0
    w1["access_count"] = w1.get("access_count", 1) + w2.get("access_count", 1)
    return w1


# Demo
worlds = {
    "ml_en": {
        "topic": "Machine Learning",
        "centroid": np.array([0.9, 0.3, 0.1]),
        "triples": [
            {"s": "neural network", "v": "tipo", "o": "modelo", "c": 0.85},
            {"s": "gradient descent", "v": "usado_em", "o": "treinamento", "c": 0.9},
        ],
        "access_count": 5,
    },
    "ml_pt": {
        "topic": "Aprendizado de Máquina",
        "centroid": np.array([0.88, 0.32, 0.11]),
        "triples": [
            {"s": "neural network", "v": "tipo", "o": "modelo", "c": 0.8},
            {"s": "overfitting", "v": "problema", "o": "generalização", "c": 0.75},
        ],
        "access_count": 3,
    },
    "cooking": {
        "topic": "Culinária",
        "centroid": np.array([0.1, 0.8, 0.7]),
        "triples": [{"s": "bolo", "v": "requer", "o": "forno", "c": 0.9}],
        "access_count": 2,
    },
}

# Find and execute fusions
pairs = find_fusion_candidates(worlds)
print(f"Fusion candidates: {len(pairs)}")
for w1_id, w2_id, sim in pairs:
    print(f"  Merging '{worlds[w1_id]['topic']}' + '{worlds[w2_id]['topic']}' (sim={sim:.4f})")
    worlds[w1_id] = merge_worlds(worlds[w1_id], worlds[w2_id])
    del worlds[w2_id]
    print(f"  Result: {len(worlds[w1_id]['triples'])} triples, "
          f"{worlds[w1_id]['access_count']} accesses")

print(f"\nWorlds remaining: {[w['topic'] for w in worlds.values()]}")

Considerações Finais

Fusão automática é o que mantém o sistema limpo a longo prazo. Sem ela, o número de mundos cresceria indefinidamente, a busca ficaria mais lenta, e o contexto ficaria fragmentado. Com ela, o sistema se auto-organiza: cria mundos quando precisa, funde quando convergem, remove quando ficam vazios.

O princípio é: permita fragmentação no curto prazo (é mais fácil criar mundos novos do que decidir na hora se algo pertence a um existente), mas consolide no longo prazo (detecte convergência e unifique). Flexibilidade na criação, eficiência na manutenção.

Links indicativos:

Voltar ao blog