Construir un Knowledge Graph desde Texto con LLMs: Pipeline Completo

[Grego]

Construir un Knowledge Graph desde Texto con LLMs: Pipeline Completo

Transforma datos no estructurados en grafos de conocimiento interactivos usando Python y modelos de lenguaje

---

¿Por Qué Knowledge Graphs?

Los datos no estructurados (artículos, documentos, biografías) contienen información valiosa, pero difícil de consultar programáticamente. Un Knowledge Graph (KG) estructura esa información como una red de entidades conectadas por relaciones, permitiendo:

• Consultas tipo “¿Qué descubrió Marie Curie?”
• Navegación visual de conexiones entre conceptos
• Inferencia de nuevos hechos a partir de relaciones existentes
• Integración con sistemas de RAG (Retrieval-Augmented Generation)

Este artículo presenta un pipeline completo que usa LLMs para extraer automáticamente hechos de texto y construir un Knowledge Graph interactivo.

---

El Concepto: Triples SPO

La unidad fundamental de un Knowledge Graph es el triple SPO (Subject-Predicate-Object):

Cada hecho del texto se descompone en tres partes:

Componente	Rol	Ejemplo
Subject	La entidad principal	marie curie
Predicate	La relación/acción	discovered
Object	La entidad relacionada	radium

Esta estructura se mapea directamente al grafo:

• Subject y Object → Nodos
• Predicate → Edge dirigido (con label)

---

Arquitectura del Pipeline

El proceso completo sigue estos pasos:

Resumen de etapas:

1. Input: Texto no estructurado (cualquier documento)
2. Chunking: Dividir en fragmentos manejables con overlap
3. Extracción LLM: Enviar cada chunk al LLM con prompt SPO
4. Normalización: Limpiar, lowercase, deduplicar triples
5. Construcción: Crear el grafo con NetworkX
6. Visualización: Renderizar interactivamente

---

Setup: Dependencias

pip install openai networkx ipycytoscape ipywidgets pandas

import openai
import json
import networkx as nx
import ipycytoscape
import pandas as pd
import os
import re

---

Configuración del LLM

El pipeline es compatible con cualquier proveedor que use la API de OpenAI:

# Variables de entorno
# export OPENAI_API_KEY='tu-api-key'
# export OPENAI_API_BASE='https://api.openai.com/v1'  # Opcional

api_key = os.getenv("OPENAI_API_KEY")
base_url = os.getenv("OPENAI_API_BASE")  # None para OpenAI estándar

# Crear cliente
client = openai.OpenAI(
    api_key=api_key,
    base_url=base_url
)

# Configuración
llm_model = "gpt-4o"  # o "claude-3-sonnet", "deepseek-v3", etc.
llm_temperature = 0.0  # Determinístico para extracción
llm_max_tokens = 4096

Opciones de modelo:

• OpenAI: gpt-4o, gpt-4o-mini
• Anthropic: claude-3-5-sonnet (via API compatible)
• Local: ollama con cualquier modelo
• Otros: DeepSeek, Mistral, etc.

---

Paso 1: Texto de Entrada

Para este ejemplo, usaremos una biografía de Marie Curie:

texto = """
Marie Curie, born Maria Skłodowska in Warsaw, Poland, was a pioneering 
physicist and chemist. She conducted groundbreaking research on radioactivity. 
Together with her husband, Pierre Curie, she discovered the elements polonium 
and radium. Marie Curie was the first woman to win a Nobel Prize, the first 
person and only woman to win the Nobel Prize twice, and the only person to 
win the Nobel Prize in two different scientific fields. She won the Nobel 
Prize in Physics in 1903 with Pierre Curie and Henri Becquerel. Later, she 
won the Nobel Prize in Chemistry in 1911 for her work on radium and polonium.
"""

print(f"Palabras: {len(texto.split())}")
# Palabras: ~120

---

Paso 2: Chunking con Overlap

Los LLMs tienen límites de contexto. Dividir el texto en chunks permite procesar documentos largos, y el overlap preserva contexto entre fragmentos:

def chunk_text(text: str, chunk_size: int = 150, overlap: int = 30) -> list:
    """
    Divide texto en chunks con overlap.
    
    Args:
        text: Texto a dividir
        chunk_size: Palabras por chunk
        overlap: Palabras de solapamiento entre chunks
    
    Returns:
        Lista de dicts con 'text' y 'chunk_number'
    """
    words = text.split()
    chunks = []
    start = 0
    chunk_num = 1
    
    while start < len(words):
        end = min(start + chunk_size, len(words))
        chunk_text = " ".join(words[start:end])
        chunks.append({
            "text": chunk_text,
            "chunk_number": chunk_num
        })
        
        # Siguiente chunk con overlap
        next_start = start + chunk_size - overlap
        if next_start <= start:
            next_start = start + 1
        start = next_start
        chunk_num += 1
        
        # Safety: evitar loops infinitos
        if chunk_num > len(words):
            break
    
    return chunks

# Aplicar chunking
chunks = chunk_text(texto, chunk_size=150, overlap=30)
print(f"Chunks generados: {len(chunks)}")

# Visualizar
for c in chunks:
    words = len(c['text'].split())
    print(f"  Chunk {c['chunk_number']}: {words} palabras")

Output:

Chunks generados: 1
  Chunk 1: 120 palabras

Para textos cortos, puede resultar en un solo chunk. En documentos largos, verás múltiples chunks con el overlap preservando contexto.

---

Paso 3: Prompt de Extracción SPO

El prompt es crítico. Debe especificar exactamente el formato de salida esperado:

SYSTEM_PROMPT = """
You are an AI expert specialized in knowledge graph extraction.
Your task is to identify and extract factual Subject-Predicate-Object (SPO) 
triples from the given text.
Focus on accuracy and adhere strictly to the JSON output format requested.
"""

USER_PROMPT_TEMPLATE = """
Extract Subject-Predicate-Object (S-P-O) triples from the text below.

**RULES:**
1. Output ONLY a valid JSON array. Each element must have keys: 
   "subject", "predicate", "object"
2. NO text before or after the JSON. NO markdown code fences.
3. Keep predicates concise (1-3 words, verbs preferred)
4. ALL values must be LOWERCASE
5. Replace pronouns (she, he, it) with the actual entity name
6. Be specific (e.g., "nobel prize in physics" not just "nobel prize")
7. Extract ALL distinct factual relationships

**Text:**
{text_chunk}

**Required format:**
[
  {{"subject": "entity1", "predicate": "relation", "object": "entity2"}},
  ...
]

Your JSON:
"""

Reglas clave explicadas:

Regla	Razón
Solo JSON	Facilita parsing automático
Lowercase	Normalización para deduplicación
Resolver pronombres	Evita “she discovered” sin saber quién es “she”
Predicados concisos	Grafos más limpios y navegables
Especificidad	Preserva información importante

---

Paso 4: Extracción con el LLM

def extract_triples_from_chunk(client, chunk: dict, model: str) -> list:
    """
    Extrae triples SPO de un chunk usando el LLM.
    
    Returns:
        Lista de triples validados con 'chunk' source
    """
    prompt = USER_PROMPT_TEMPLATE.format(text_chunk=chunk['text'])
    
    try:
        response = client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": SYSTEM_PROMPT},
                {"role": "user", "content": prompt}
            ],
            temperature=0.0,
            max_tokens=4096
        )
        raw = response.choices[0].message.content.strip()
    except Exception as e:
        print(f"Error en chunk {chunk['chunk_number']}: {e}")
        return []
    
    # Parsear JSON
    try:
        data = json.loads(raw)
        if isinstance(data, dict):
            # Algunos LLMs devuelven {"triples": [...]}
            data = next((v for v in data.values() if isinstance(v, list)), [])
    except json.JSONDecodeError:
        # Fallback: buscar array con regex
        match = re.search(r'\[.*\]', raw, re.DOTALL)
        if match:
            try:
                data = json.loads(match.group())
            except:
                return []
        else:
            return []
    
    # Validar estructura
    valid_triples = []
    for t in data:
        if isinstance(t, dict):
            s = t.get('subject', '')
            p = t.get('predicate', '')
            o = t.get('object', '')
            if all(isinstance(x, str) and x.strip() for x in [s, p, o]):
                valid_triples.append({
                    'subject': s,
                    'predicate': p,
                    'object': o,
                    'chunk': chunk['chunk_number']
                })
    
    return valid_triples

# Procesar todos los chunks
all_triples = []
for chunk in chunks:
    triples = extract_triples_from_chunk(client, chunk, llm_model)
    all_triples.extend(triples)
    print(f"Chunk {chunk['chunk_number']}: {len(triples)} triples extraídos")

print(f"\nTotal triples raw: {len(all_triples)}")

Output ejemplo:

Chunk 1: 18 triples extraídos

Total triples raw: 18

---

Paso 5: Normalización y Deduplicación

Los triples raw necesitan limpieza antes de construir el grafo:

def normalize_triples(raw_triples: list) -> list:
    """
    Normaliza y deduplica triples.
    
    Pasos:
    1. Lowercase y trim
    2. Filtrar vacíos
    3. Deduplicar usando Set
    """
    normalized = []
    seen = set()
    
    stats = {
        'original': len(raw_triples),
        'empty_removed': 0,
        'duplicates_removed': 0
    }
    
    for t in raw_triples:
        # Normalizar
        s = t.get('subject', '').strip().lower()
        p = t.get('predicate', '').strip().lower()
        p = re.sub(r'\s+', ' ', p)  # Espacios múltiples → uno
        o = t.get('object', '').strip().lower()
        
        # Filtrar vacíos
        if not all([s, p, o]):
            stats['empty_removed'] += 1
            continue
        
        # Deduplicar
        key = (s, p, o)
        if key in seen:
            stats['duplicates_removed'] += 1
            continue
        
        seen.add(key)
        normalized.append({
            'subject': s,
            'predicate': p,
            'object': o,
            'source_chunk': t.get('chunk', '?')
        })
    
    print(f"Normalización:")
    print(f"  Original: {stats['original']}")
    print(f"  Vacíos removidos: {stats['empty_removed']}")
    print(f"  Duplicados removidos: {stats['duplicates_removed']}")
    print(f"  Final: {len(normalized)}")
    
    return normalized

# Aplicar normalización
clean_triples = normalize_triples(all_triples)

Output ejemplo:

Normalización:
  Original: 18
  Vacíos removidos: 0
  Duplicados removidos: 2
  Final: 16

---

Paso 6: Construcción del Grafo

Con los triples limpios, construimos el grafo usando NetworkX:

def build_knowledge_graph(triples: list) -> nx.DiGraph:
    """
    Construye un DiGraph de NetworkX desde los triples.
    
    - Subject → Nodo
    - Object → Nodo  
    - Predicate → Edge label
    """
    G = nx.DiGraph()
    
    for t in triples:
        subject = t['subject']
        predicate = t['predicate']
        obj = t['object']
        
        # add_edge automáticamente crea los nodos si no existen
        G.add_edge(subject, obj, label=predicate)
    
    return G

# Construir
kg = build_knowledge_graph(clean_triples)

print(f"Knowledge Graph creado:")
print(f"  Nodos (entidades): {kg.number_of_nodes()}")
print(f"  Edges (relaciones): {kg.number_of_edges()}")

Output ejemplo:

Knowledge Graph creado:
  Nodos (entidades): 15
  Edges (relaciones): 16

---

Paso 7: Visualización Interactiva

Usando ipycytoscape para renderizar el grafo en Jupyter:

def visualize_kg(G: nx.DiGraph):
    """
    Crea visualización interactiva del Knowledge Graph.
    """
    # Convertir a formato Cytoscape
    nodes = []
    edges = []
    
    # Calcular grados para sizing
    degrees = dict(G.degree())
    max_degree = max(degrees.values()) if degrees else 1
    
    for node_id in G.nodes():
        degree = degrees.get(node_id, 0)
        size = 20 + (degree / max_degree) * 40
        nodes.append({
            'data': {
                'id': str(node_id),
                'label': str(node_id),
                'size': size
            }
        })
    
    for i, (u, v, data) in enumerate(G.edges(data=True)):
        edges.append({
            'data': {
                'id': f'edge_{i}',
                'source': str(u),
                'target': str(v),
                'label': data.get('label', '')
            }
        })
    
    # Crear widget
    cyto = ipycytoscape.CytoscapeWidget()
    cyto.graph.add_graph_from_json({
        'nodes': nodes,
        'edges': edges
    })
    
    # Estilo
    cyto.set_style([
        {
            'selector': 'node',
            'style': {
                'label': 'data(label)',
                'background-color': '#6366f1',
                'color': '#ffffff',
                'text-valign': 'center',
                'width': 'data(size)',
                'height': 'data(size)',
                'font-size': '10px'
            }
        },
        {
            'selector': 'edge',
            'style': {
                'label': 'data(label)',
                'curve-style': 'bezier',
                'target-arrow-shape': 'triangle',
                'line-color': '#94a3b8',
                'target-arrow-color': '#94a3b8',
                'font-size': '8px',
                'color': '#64748b'
            }
        },
        {
            'selector': 'node:selected',
            'style': {
                'background-color': '#22c55e',
                'border-width': 2,
                'border-color': '#16a34a'
            }
        }
    ])
    
    # Layout
    cyto.set_layout(name='cose', nodeRepulsion=8000)
    
    return cyto

# Visualizar
widget = visualize_kg(kg)
display(widget)

El resultado es un grafo interactivo donde puedes:

• Arrastrar nodos para reorganizar
• Click en nodos para seleccionar
• Zoom in/out con scroll
• Ver las relaciones (predicados) en los edges

---

Ejemplo de Output

Para el texto de Marie Curie, el grafo resultante muestra:

Nodos centrales:

• marie curie (hub principal con múltiples conexiones)
• pierre curie
• nobel prize in physics
• nobel prize in chemistry
• radium, polonium
• warsaw, poland

Relaciones típicas extraídas:

(marie curie) —[born in]→ (warsaw, poland)
(marie curie) —[discovered]→ (radium)
(marie curie) —[discovered]→ (polonium)
(marie curie) —[won]→ (nobel prize in physics)
(marie curie) —[won]→ (nobel prize in chemistry)
(marie curie) —[married to]→ (pierre curie)
(pierre curie) —[discovered]→ (radium)

---

Código Completo

# kg_pipeline.py - Pipeline completo de Knowledge Graph

import openai
import json
import networkx as nx
import os
import re

def chunk_text(text, chunk_size=150, overlap=30):
    words = text.split()
    chunks = []
    start = 0
    num = 1
    while start < len(words):
        end = min(start + chunk_size, len(words))
        chunks.append({"text": " ".join(words[start:end]), "chunk_number": num})
        start = start + chunk_size - overlap
        if start <= 0: start = 1
        num += 1
        if num > len(words): break
    return chunks

def extract_triples(client, chunk, model):
    SYSTEM = "You are an AI expert in knowledge graph extraction."
    USER = f"""Extract SPO triples from this text as JSON array.
Rules: lowercase, no markdown, concise predicates, resolve pronouns.
Format: [{{"subject": "x", "predicate": "y", "object": "z"}}]

Text: {chunk['text']}

JSON:"""
    
    try:
        r = client.chat.completions.create(
            model=model,
            messages=[{"role": "system", "content": SYSTEM},
                      {"role": "user", "content": USER}],
            temperature=0.0, max_tokens=4096
        )
        data = json.loads(r.choices[0].message.content.strip())
        if isinstance(data, dict):
            data = next((v for v in data.values() if isinstance(v, list)), [])
    except:
        return []
    
    return [
        {**t, 'chunk': chunk['chunk_number']}
        for t in data
        if all(t.get(k) for k in ['subject', 'predicate', 'object'])
    ]

def normalize(triples):
    seen = set()
    out = []
    for t in triples:
        key = tuple(t[k].strip().lower() for k in ['subject', 'predicate', 'object'])
        if all(key) and key not in seen:
            seen.add(key)
            out.append({'subject': key[0], 'predicate': key[1], 'object': key[2]})
    return out

def build_graph(triples):
    G = nx.DiGraph()
    for t in triples:
        G.add_edge(t['subject'], t['object'], label=t['predicate'])
    return G

# Main
if __name__ == "__main__":
    client = openai.OpenAI()
    text = "..." # Tu texto aquí
    
    chunks = chunk_text(text)
    raw = [t for c in chunks for t in extract_triples(client, c, "gpt-4o")]
    clean = normalize(raw)
    kg = build_graph(clean)
    
    print(f"Nodos: {kg.number_of_nodes()}, Edges: {kg.number_of_edges()}")

---

Próximos Pasos

Este pipeline básico puede extenderse con:

Mejora	Descripción
Entity Linking	Conectar “Marie Curie” y “M. Curie” al mismo ID
Relationship Clustering	Agrupar “born in” y “was born at”
Persistencia	Guardar en Neo4j o ArangoDB
Evaluación	Medir precision/recall de extracción
Multi-hop Queries	“¿Qué descubrió la esposa de Pierre Curie?”
RAG Integration	Usar el KG para mejorar respuestas de LLMs

---

Recursos

• NetworkX Documentation
• ipycytoscape
• Neo4j Graph Database
• Knowledge Graphs - Wikipedia

---

Publicado en yoDEV.dev — La comunidad de desarrolladores de Latinoamérica