GraphRAG paso a paso
Tutorial práctico de GraphRAG: cómo definir una ontología, extraer entidades y relaciones con un LLM, cargarlas en Neo4j y resolver preguntas multi-hop con búsqueda local y global.
Sigo con este post la serie dedicada GraphRAG. Si no sabes qué es un grafo o Neo4js, te invito a que leas el primer artículo de la serie. Además, tengo que advertir que es una entrada atípica de este blog, en la que bajo mucho al detalle del código. Si solo te interesa la parte teórica, te recomiendo el tercer artículo de la serie. En cualquier caso, todo el código del tutorial, muy anotado, está en git hub, donde quizás resulte más cómodo de consultar. Y sin más preámbulos, al lío.
1. Las limitaciones de RAG
La técnica RAG, Retrieval-Augmented Generation, existía ya antes de la gran explosión de la IA generativa a finales de 2023. En un paper de 2020, Patrick Lewis y coautores ya hablaban de las posibilidades de esta técnica, que en esencia consiste en consultar información externa relevante antes de responder, para producir respuestas más precisas, actualizadas y fundamentadas. Por ejemplo, un agente que buscase en Wikipedia información relacionada con el prompt planteado antes de enviárselo al modelo, estaría utilizando RAG; al igual que otro que consulte el clima o el estado del tráfico.
El único requisito para que pueda considerarse RAG es que busque la información fuera del prompt, pues de lo contrario sería lo que se conoce como ICL, In Context Learning, donde el modelo infiere el patrón a partir de ejemplos o instrucciones dentro del contexto que se le pasa.
Esta recuperación de contenidos puede hacerse de muchas maneras. Una habitual es mediante vectores semánticos. Los documentos se dividen en trozos, cada trozo se convierte en un vector numérico, en un embedding y se guarda en una base vectorial. Cuando se le formula una pregunta, se busca qué trozos tienen vectores más parecidos y esos textos se pasan como contexto al modelo para que responda de manera más precisa y, sobre todo, con información relacionada que no trae aprendida de fábrica, como las FAQs de una web comercial.
RAG funciona bien cuando los fragmentos de información añadidos se bastan por sí mismos para proporcionar el contexto adecuado a una cuestión. Imaginemos, por ejemplo, un sistema agéntico sobre animales en el que cada animal fuera una unidad de información. Al preguntar sobre los chimpancés, el sistema podría buscar los fragmentos correspondientes a las voces de «mamíferos», «primates» y «homini» para enriquecer el contexto.
Sin embargo, esta técnica se queda corta con las llamadas preguntas multi-hop, que son las preguntas que requieren combinar dos o más piezas de información para llegar a la respuesta. No se responden con un solo dato directo, sino haciendo varios saltos de razonamiento o búsqueda. Por ejemplo, si preguntamos quién escribió Cien años de soledad, la respuesta es simple: Gabriel García Márquez. Pero si le pidiéramos a un modelo que nos dijera «en qué país nació el autor de Cien años de soledad», tendría que dar un salto: primero tendría que identificar a Márquez y luego saber que nació en Colombia.
Para terminar de entender esta limitación podemos pensar en un sistema agéntico basado en RAG tradicional sobre la saga de George R. R. Martin de Juego de tronos, que tan mal remató HBO cuando la llevó a la pequeña pantalla. Si le preguntáramos «cuáles son las principales facciones en conflicto y qué las mueve», recuperaría tropecientos mil fragmentos de forma bastante aleatoria, saturaría el contexto y es probable que la respuesta fuera incorrecta. Sin embargo, con graphRAG este tipo de cuestiones multi-hop son más fáciles de resolver.
Vamos a ver cómo funciona con un ejercicio práctico en el que nos basaremos precisamente en esta obra que, a estas alturas, o ya se conoce, por lo que darían igual los posibles spoilers, o imagino que no se querrá conocer jamás, por lo que también darían igual :P.
2. Plan de trabajo y entorno
En abril de 2024, Microsoft publicó el paper fundacional de la técnica: From Local to Global: A Graph RAG Approach to Query-Focused Summarization, de Darren Edge et al. Era una propuesta compleja que ha dado lugar a distintas variantes más o menos sofisticadas, como LightRAG, de la que hablaré otro día.
Para este tutorial vamos a ver un sistema más sencillo que el original de Microsoft, pero que nos permitirá entender los conceptos básicos, como las ontologías y los sistemas de búsqueda. El plan es el siguiente:
- Paso 1 - Preparación del entorno. Instalaremos todo lo que vamos a necesitar.
- Paso 2 - Definición de las ontologías. Prepararemos las reglas del sistema.
- Paso 3 - Extracción de las entidades y las relaciones. Con las reglas listas, sacaremos las distintas entidades siguiendo las reglas de las ontologías.
- Paso 4 - Carga en Neo4j. Llevaremos a esta base de datos las relaciones.
- Paso 5 - Ejecución de consultas. Comprobaremos que todo funciona lanzando consultas de distinta naturaleza.
Ahora vamos a dejar todo listo antes de escribir ni una línea de lógica de GraphRAG.
Lo primero es preparar una instancia en la nube gratuita de Neo4j. También podríamos hacerlo con Neo4js [← pendiente de aclarar], pero es algo más pesado y para este tutorial nos vale el tier gratuito de esta base de datos.
Como vimos antes, vamos a la consola, creamos una cuenta si no la tuviéramos y creamos una nueva instancia eligiendo AuraDB Free. Copiamos el usuario y el api key que nos da y seguimos en lo que se va creando.
Vamos a necesitar un modelo. Podemos usar cualquiera, pero para este tipo de ejercicios nos debería valer con DeepSeek, que es prácticamente gratuito. Vamos a la plataforma y generamos un api key.
Vamos ahora con las dependencias de python:
- langchain: Se puede utilizar otro framework para trabajar con agentes, pero creo que este es más versátil y sencillo.
- langchain-deepseek: El paquete de langchain para trabajar con DeepSeek. Cambiar si se usa otro modelo, claro.
- langchain-neo4j: El conector específico de LangChain para Neo4j. Nos da clases ya hechas para conectarnos a la base de datos y ejecutar Cypher desde Python de forma muy cómoda.
- neo4j: El driver oficial de Neo4j para Python. Langchain-neo4j internamente depende de este paquete para hacer la conexión real por debajo.
- networkx: Opcional, una librería de grafos en Python que usaremos para trabajar con comunidades sin usar el plugin de pago.
- python-dotenv: Para cargar las variables de entorno.
Ya sea con pip o con uv, incorporamos todo al proyecto.
uv init graphrag-got
cd graphrag-got
uv add langchain langchain-neo4j langchain-deepseek neo4j networkx python-dotenv- Corpus de textos
En la vida real, le pasaríamos a la pipeline todas las novelas para que las troceara y preparase las entidades, lo que vamos a ver en el siguiente epígrafe, pero para este tutorial basta con algo muchísimo más resumido.
Vamos a usar un corpus generado por IA que narra el conflicto por el control de Poniente entre varias casas nobles, unidas por matrimonios políticos y rotas por traiciones. No hace falta que te adelante quién trama qué, el propio texto lo cuenta mejor que yo:
La Casa Stark, señora del Norte desde Invernalia, mantenía una alianza sólida con la Casa Tully de Aguasdulces, sellada por el matrimonio entre Ned Stark y Catelyn Tully. Gracias a ese matrimonio, cuando estalló el conflicto conocido como la Guerra de los Cinco Reyes, la Casa Tully se puso del lado de los Stark casi automáticamente, aportando tropas y el control de la importante encrucijada de Aguasdulces.
En el sur, la Casa Lannister de Roca Casterly había asegurado su influencia sobre el Trono de Hierro mediante otro matrimonio político: Cersei Lannister se casó con el rey Robert Baratheon. Aunque el trono pertenecía formalmente a la Casa Baratheon, en la práctica eran los Lannister quienes controlaban la política de Desembarco del Rey a través de Cersei y su padre, Tywin Lannister.
Tras la muerte de Ned Stark, ejecutado por orden de la Casa Lannister, su hijo Robb Stark fue proclamado Rey en el Norte y lideró la rebelión de los Stark y los Tully contra los Lannister. Robb ganó varias batallas importantes, pero cometió un error político grave: rompió un compromiso matrimonial pactado con la Casa Frey, señores del Cruce, a cambio de casarse por amor con otra mujer.
La Casa Frey, humillada, decidió aliarse en secreto con la Casa Lannister. Junto con la Casa Bolton —que hasta entonces había sido vasalla leal de los Stark dentro del Norte—, los Frey tendieron una trampa durante una celebración de boda: la conocida como la Boda Roja. Allí, la Casa Bolton y la Casa Frey asesinaron a Robb Stark, a Catelyn Stark y a gran parte del ejército del Norte, rompiendo así la alianza Stark-Tully y entregando el control del Norte a la Casa Bolton como nuevos señores, ahora aliados de los Lannister.
Mientras tanto, en Desembarco del Rey, la Casa Lannister buscó reforzar su posición aliándose con la poderosa Casa Tyrell de Altojardín. Esta alianza se selló mediante el matrimonio de Margaery Tyrell, primero con el rey Joffrey Baratheon y, tras el asesinato de este durante su propia boda, con su hermano menor Tommen Baratheon. Gracias a este pacto, la Casa Tyrell aportó su ejército y sus riquezas para sostener el reinado de los Lannister frente a otros pretendientes al trono, como Stannis Baratheon.
Sin embargo, esta alianza Lannister-Tyrell resultó frágil: Cersei Lannister, desconfiada del creciente poder de los Tyrell, orquestó la destrucción de la Fe Militante y del Gran Septo de Baelor, matando en el proceso a Margaery Tyrell y a gran parte de su familia, poniendo fin abruptamente a la alianza y dejando a los Lannister sin el apoyo militar de Altojardín justo cuando más lo necesitaban.
Sansa Stark, hija superviviente de Ned y Catelyn, fue utilizada como pieza política en varios matrimonios forzados que reflejaban los cambios de poder: primero fue prometida y casada con Tyrion Lannister para atar a los Stark a la causa Lannister, y más tarde, tras escapar de Desembarco del Rey, fue casada a la fuerza con Ramsay Bolton, hijo de la Casa Bolton, en un intento de legitimar el control de los Bolton sobre el Norte ante los ojos de los antiguos vasallos Stark.
Lejos de Poniente, Daenerys Targaryen, última superviviente de la dinastía que antaño gobernó los Siete Reinos, formó su propia coalición reuniendo a los guerreros Dothraki mediante su matrimonio con el khal Drogo, y posteriormente liberando y reclutando a los Inmaculados, un ejército de soldados esclavos liberados en las ciudades de Essos. Con tres dragones y estas fuerzas combinadas, Daenerys se convirtió en una potencia militar independiente de cualquier casa de Poniente.
En el extremo norte, Jon Snow ascendió a Lord Comandante de la Guardia de la Noche, la orden encargada de vigilar el Muro. Bajo su mando, la Guardia de la Noche identificó a los Caminantes Blancos y su Ejército de los Muertos como una amenaza existencial para todo Poniente, muy por encima de cualquier disputa entre casas nobles. Consciente de que la Guardia de la Noche no podía enfrentar sola a esa amenaza, Jon viajó a Rocadragón, la fortaleza ancestral de la Casa Targaryen, para pedir ayuda a Daenerys. Ese viaje marcó el primer encuentro entre ambos: Daenerys, inicialmente reacia a ceder su ejército sin que Jon la reconociera como reina, terminó aceptando una alianza tras comprobar la veracidad de la amenaza, uniendo así a la Casa Targaryen, la Guardia de la Noche y, poco después, a los restos de la Casa Stark del Norte —ya liberada del control Bolton— en una única coalición contra los muertos.
Ante el avance de esta coalición Stark-Targaryen, Cersei Lannister prometió públicamente enviar tropas para ayudar contra los Caminantes Blancos, pero en privado decidió no cumplir su palabra: en su lugar, contrató en secreto a la Compañía Dorada, un grupo de mercenarios de Essos, para reforzar exclusivamente la defensa de la Casa Lannister una vez que sus rivales del Norte quedaran debilitados por la guerra contra los muertos. De este modo, hacia el desenlace del conflicto, Poniente quedó dividido en tres bloques principales: la Casa Lannister, ahora aislada y apoyada solo por la Compañía Dorada; la coalición formada por la Casa Stark, la Casa Targaryen y la Guardia de la Noche; y los Caminantes Blancos, enemigos de ambos bandos humanos por igual.4. Ontologías
Con todo listo, es el momento de definir las ontologías, que en este ámbito son el manual de instrucciones, el mapa mental que le das al sistema para definir qué cosas existen en tu mundo y cómo se les permite relacionarse entre sí.
Una ontología se compone de tres elementos fundamentales:
-
Clases o categorías. Define qué tipos de cosas existen, son las etiquetas generales para agrupar los conceptos, como
Personaje,CasaNoble,Region,Organizacionen nuestro tutorial oPaciente,Síntoma,MedicamentoyMédicoen un hospital. -
Relaciones permitidas. Cómo se pueden conectar las distintas clases, con qué verbos o flechas. Por ejemplo, se podría permitir:
- [
Personaje] -(PERTENECE_A)-> [CasaNoble]
Pero prohibir por absurdo
- [
CasaNoble] -(CASADO_CON)-> [Region]
- Propiedades o atributos. ¿Qué características tienen? Son los datos específicos que describen a cada clase. En nuestro caso, para la clase
Personaje, las propiedades podrían sernombre,edad,estado (Vivo/Muerto); para la claseRegionpodrían serTipo (Reino/Ciudad/Fortaleza).
Definir la ontología, que quizás sea la parte más divertida de todo el invento, al igual que lo es el diseño de las tablas en MySQL, es fundamental para que no sea un caos. Por ejemplo, si no tuviéramos definida la relación PADRE_DE y lo dejáramos a la libre decisión del modelo, en un mismo sistema podrían aparecer las relaciones PROGENITOR_DE y PAPÁ_DE para referirse a la misma relación.
En nuestro caso, por ejemplo, podríamos definir estas clases:
Personaje: Ned Stark, Cersei Lannister, Jon Snow, Daenerys...CasaNoble: Stark, Lannister, Tully, Bolton, Tyrell, Baratheon, Targaryen.Organizacion: Guardia de la Noche, Inmaculados, Compañía Dorada, Dothraki, Caminantes Blancos.Region: Norte, Poniente, Essos, Desembarco del Rey, Rocadragón.Evento: Boda Roja, Guerra de los Cinco Reyes.
Para las relaciones vamos a establecer una relación unidireccional de dominio -> rango.
[Dominio] -RELACION-> [Rango]Donde:
- Dominio es el tipo de nodo que puede estar en el origen de la flecha, el sujeto.
- Rango es el tipo de nodo que puede estar en el destino de la flecha, el objeto.
Esto permite definir claramente las relaciones que están permitidas. Por ejemplo, la relación PERTENECE_A solo puede tener como dominio un Personaje y como rango una CasaNoble, lo que descartaría de partida relaciones PERTENECE_A entre un Evento y una Organizacion.
Además, vamos a definir si una relación es simétrica o no. Esto es, si cuando es cierta en un sentido, automáticamente es cierta en el sentido contrario.
Por ejemplo, si esta relación fuera simétrica:
Casa Stark -ALIADA_DE-> Casa TullySignifica que necesariamente también es verdad que:
Casa Tully -ALIADA_DE-> Casa StarkPor último, para cada relación y cada clase vamos a añadir el parámetro comment, donde irá una frase corta en lenguaje natural que explica qué significa esa clase o relación, dirigida tanto a un humano que lea el JSON como al propio LLM cuando se lo pasemos en el prompt de extracción.
En las relaciones, por ejemplo:
{
"nombre": "PERTENECE_A",
"dominio": "Personaje",
"rango": "CasaNoble",
"simetrica": false,
"comment": "El personaje es miembro de esa casa noble por nacimiento o adopción"
}
...Esta manera de tratar la ontología se inspira en el estándar RDF/OWL, aunque no lo sigue con total fidelidad, sobre el que no voy a profundizar ahora para no despistarnos aún más del tema.
Por último, nos quedaría por definir las propiedades o atributos. Al igual que con las relaciones, para cada propiedad definimos su tipo esperado que puede ser:
- un string libre
- o un enum con valores cerrados, para evitar que el LLM invente variantes distintas de lo mismo, como
vivo,Vivo,con vidayestá vivopara el mismo hecho.
Así, para la clase Personaje, sus propiedades podrían quedar de esta manera:
"Personaje": {
"comment": "Un individuo con nombre propio (ej. Ned Stark, Cersei Lannister, Jon Snow)",
"propiedades": {
"nombre": "string, obligatorio",
"estado": "enum: vivo | muerto | desconocido",
"titulo": "string opcional, ej. 'Lord Comandante', 'Rey en el Norte', 'Khal'"
}
}Nuestra ontología, por lo tanto, podría ser más o menos así:
{
"clases": {
"Personaje": {
"comment": "Un individuo con nombre propio (ej. Ned Stark, Cersei Lannister, Jon Snow)",
"propiedades": {
"nombre": "string, obligatorio",
"estado": "enum: vivo | muerto | desconocido",
"titulo": "string opcional, ej. 'Lord Comandante', 'Rey en el Norte', 'Khal'"
}
},
"CasaNoble": {
"comment": "Una casa noble de Poniente (ej. Casa Stark, Casa Lannister, Casa Bolton)",
"propiedades": {
"nombre": "string, obligatorio",
"sede": "string opcional, ej. 'Invernalia', 'Roca Casterly'",
"lema": "string opcional, si se menciona en el texto"
}
},
"Organizacion": {
"comment": "Cualquier grupo colectivo que no es una casa noble: militar, mercenario, o incluso un bando no humano (ej. Guardia de la Noche, Inmaculados, Compañía Dorada, Dothraki, Caminantes Blancos)",
"propiedades": {
"nombre": "string, obligatorio",
"tipo": "enum: militar | mercenaria | religiosa | sobrenatural | otra"
}
},
"Region": {
"comment": "Un lugar geográfico o político (ej. Norte, Essos, Desembarco del Rey, Rocadragón)",
"propiedades": {
"nombre": "string, obligatorio",
"tipo": "enum: reino | ciudad | fortaleza | continente"
}
},
"Evento": {
"comment": "Un suceso histórico puntual dentro de la narrativa (ej. Boda Roja, Guerra de los Cinco Reyes)",
"propiedades": {
"nombre": "string, obligatorio",
"tipo": "enum: batalla | traicion | matrimonio | asesinato | otro"
}
}
},
"relaciones": [
{
"nombre": "CASADO_CON",
"dominio": "Personaje",
"rango": "Personaje",
"simetrica": true,
"comment": "Matrimonio entre dos personajes"
},
{
"nombre": "PERTENECE_A",
"dominio": "Personaje",
"rango": "CasaNoble",
"simetrica": false,
"comment": "El personaje es miembro de esa casa noble por nacimiento o adopción"
},
{
"nombre": "ES_MIEMBRO_DE",
"dominio": "Personaje",
"rango": "Organizacion",
"simetrica": false,
"comment": "El personaje pertenece a una organización (no casa noble)"
},
{
"nombre": "LEAL_A",
"dominio": ["Personaje", "CasaNoble"],
"rango": ["CasaNoble", "Organizacion"],
"simetrica": false,
"comment": "Un personaje, o una casa noble vasalla de otra, mantiene lealtad hacia una casa u organización, sin ser necesariamente miembro formal (ej. un vasallo, un juramentado, una casa menor bajo el mando de una casa mayor)"
},
{
"nombre": "ALIADA_DE",
"dominio": ["CasaNoble", "Organizacion"],
"rango": ["CasaNoble", "Organizacion"],
"simetrica": true,
"comment": "Alianza activa entre dos entidades colectivas"
},
{
"nombre": "TRAICIONA_A",
"dominio": "CasaNoble",
"rango": "CasaNoble",
"simetrica": false,
"comment": "Una casa rompe una alianza o lealtad previa hacia otra de forma activa"
},
{
"nombre": "EN_GUERRA_CON",
"dominio": ["CasaNoble", "Organizacion"],
"rango": ["CasaNoble", "Organizacion"],
"simetrica": true,
"comment": "Conflicto armado activo entre dos entidades colectivas"
},
{
"nombre": "GOBIERNA",
"dominio": ["Personaje", "CasaNoble"],
"rango": "Region",
"simetrica": false,
"comment": "El personaje o la casa noble ejerce control político o militar sobre una región"
},
{
"nombre": "PARTICIPA_EN",
"dominio": ["Personaje", "CasaNoble"],
"rango": "Evento",
"simetrica": false,
"comment": "El personaje o casa tomó parte activa en ese evento"
},
{
"nombre": "CONTRATA",
"dominio": "CasaNoble",
"rango": "Organizacion",
"simetrica": false,
"comment": "Una casa noble paga o recluta a una organización militar externa"
},
{
"nombre": "SE_ENCUENTRA_CON",
"dominio": "Personaje",
"rango": "Personaje",
"simetrica": true,
"comment": "Dos personajes se conocen o interactúan directamente por primera vez"
}
],
"atributos_de_relacion": {
"momento": {
"tipo": "string_libre",
"comment": "Ubicación temporal aproximada de la relación dentro de la narrativa, en lenguaje natural (ej. 'antes de la Boda Roja', 'después de la Boda Roja', 'al inicio del conflicto'). Se usa para poder distinguir relaciones contradictorias que ocurrieron en momentos distintos, como una alianza que luego se convierte en traición.",
"obligatorio": false
}
}
}5. Extracción con un modelo
Ahora hay que preparar un script en python que recorra cada uno de los textos del corpus. La idea es que vaya troceando el texto y se lo vaya enviando al modelo para que prepare la lista de relaciones. En la vida real igual son PDFs, páginas de Confluence o cualquier otra fuente documental. El origen da igual, incluso pueden ser audios o imágenes. Lo importante es que se pueda dividir para que sea manejable.
En nuestro caso vamos a utilizar la separación entre párrafos y el resultado tendrá tres listados:
entidades: la lista de nodos que van a existir en el grafo, cada uno con su tipo y sus propiedades como:Jon Snow, tipoPersonaje, contitulo: "Lord Comandante".relaciones: la lista de tripletas (sujeto-relación-objeto) que van a ser las aristas que conectan esos nodos, comoJon Snow — SE_ENCUENTRA_CON — Daenerys.
Y de propina, una tercera lista con fines de auditoría:
relaciones_descartadas: las tripletas que el LLM propuso pero que no pasaron la validación contra la ontología, junto con el motivo del rechazo.
Desgloso el código, que puedes leer completo en git hub.
Importamos todo lo que vamos a necesitar.
import json
from pathlib import Path
from dotenv import load_dotenv
from langchain_deepseek import ChatDeepSeek
from langchain_core.prompts import ChatPromptTemplateCargamos las variables definidas en el archivo .env (entre ellas, DEEPSEEK_API_KEY) como si fueran variables de entorno normales del sistema operativo. ChatDeepSeek las lee automaticamente, no hace falta pasarlas a mano.
load_dotenv()Definimos las rutas de los archivos en los que va trabajar el script.
ONTOLOGY_PATH = Path("ontology.json")
CORPUS_PATH = Path("corpus.txt")
OUTPUT_PATH = Path("extraction.json")Cargamos la ontología y la devolvemos como un diccionario de python.
def load_ontology() -> dict:
with open(ONTOLOGY_PATH, encoding="utf-8") as f:
return json.load(f)Cargamos el corpus y lo troceamos en chunk partiendo el texto por saltos de línea dobles (\n\n). No usamos ninguna librería de text-splitting, como las que trae LangChain, porque el corpus ya esta naturalmente segmentado en unidades temáticas razonables, y añadir una librería mas aquí sería complejidad innecesaria para este tamaño de texto.
def load_chunks() -> list[str]:
text = CORPUS_PATH.read_text(encoding="utf-8")
# Separamos por parrafo, quitamos espacios sobrantes y descartamos
# posibles fragmentos vacios que puedan colarse al hacer split.
paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
return paragraphsConvertimos el JSON de la ontología en un bloque de texto legible, pensado para ser leído tanto por un humano como por el LLM dentro del prompt del sistema. No le pasamos el JSON bruto al modelo porque entiende mejor instrucciones en lenguaje natural con estructura tipo lista que un JSON anidado, y además así controlamos exactamente que información de la ontología se expone. Por ejemplo, aquí no exponemos el bloque atributos_de_relacion, que se maneja aparte en las reglas del prompt.
def build_ontology_description(ontology: dict) -> str:
"""
Convierte el JSON de la ontología en un bloque de texto legible,
pensado para ser leído tanto por un humano como por el LLM dentro
del prompt del sistema.
"""
lines = ["CLASES PERMITIDAS:"]
for class_name, data in ontology["clases"].items():
# Listamos los nombres de las propiedades esperadas para esa clase,
# para que el LLM sepa que campos rellenar en "propiedades".
props = ", ".join(data["propiedades"].keys())
lines.append(f"- {class_name}: {data['comment']} (propiedades: {props})")
lines.append("\nRELACIONES PERMITIDAS:")
for rel in ontology["relaciones"]:
# El dominio/rango puede ser un string ("Personaje") o una lista
# (["CasaNoble", "Organizacion"]) según cómo se definió en la
# ontología. Normalizamos ambos casos al mismo formato de texto
# "A | B" para que quede legible en el prompt.
domain = rel["dominio"] if isinstance(rel["dominio"], str) else " | ".join(rel["dominio"])
range_ = rel["rango"] if isinstance(rel["rango"], str) else " | ".join(rel["rango"])
lines.append(f"- {rel['nombre']}: [{domain}] -> [{range_}] — {rel['comment']}")
return "\n".join(lines)Luego preparamos el prompt para que el modelo prepare los listados en función de la ontología. Aunque en la vida real habría quizás que trabajar en inglés, por didáctica dejo estos ejemplos en español a propósito, ya que el corpus con el que trabajamos está en este idioma y queremos que las entidades/relaciones extraídas conserven los nombres tal como aparecen en el texto original.
El SYSTEM_PROMPT recibe un placeholder {ontology_description} que se rellena en tiempo de ejecución con el resultado de build_ontology_description(). Las llaves dobles [object Object] son necesarias porque este texto se usa como plantilla de LangChain, que usa llaves simples para variables, y el JSON de ejemplo que mostramos dentro también usa llaves. Hay que escaparlas duplicándolas para que LangChain no las confunda con variables.
El prompt de usuario es más simple: solo inserta el fragmento de texto concreto que queremos procesar en esta llamada.
Algo así:
# --- Plantillas de prompt ---
SYSTEM_PROMPT = """Eres un sistema de extracción de información para construir un grafo de conocimiento.
Debes identificar ÚNICAMENTE entidades y relaciones que encajen EXACTAMENTE en la
siguiente ontología. Si una relación o entidad del texto no encaja en ninguna clase o relación permitida, NO la incluyas.
{ontology_description}
REGLAS ESTRICTAS:
1. Usa EXCLUSIVAMENTE los nombres de clase y relación tal como aparecen arriba (mismo formato, mayúsculas incluidas en las relaciones).
2. Respeta el dominio y rango de cada relación: el tipo del sujeto debe coincidir con el dominio, el tipo del objeto con el rango.
3. Si detectas que una relación ocurrió en un momento distinguible de la narrativa (por ejemplo, antes o después de un evento clave), añade el campo "momento" con una frase corta describiéndolo. Si no es relevante, omite el campo.
4. Responde ÚNICAMENTE con un JSON válido, sin texto adicional, sin explicaciones, sin bloques de código markdown.
FORMATO DE SALIDA (JSON estricto):
{{
"entidades": [
{{"nombre": "...", "tipo": "...", "propiedades": {{}}}}
],
"relaciones": [
{{"sujeto": "...", "tipo_sujeto": "...", "relacion": "...", "objeto": "...", "tipo_objeto": "...", "momento": "..."}}
]
}}
"""
USER_PROMPT = """Extrae entidades y relaciones del siguiente fragmento de texto:
---
{fragment}
---
Responde solo con el JSON."""Por si acaso el modelo devolviera el json envuelto en ```json ... ```, tal y como hacen algunos modelos aunque le pidamos explícitamente que no use bloques de código markdown, preparamos una función que detecte y quite esa envoltura si aparece, para que json.loads() no falle.
def clean_json_response(text: str) -> str:
text = text.strip()
if text.startswith("```"):
# Nos quedamos con el contenido entre el primer y segundo ```
text = text.split("```")[1]
# Si el bloque empezaba con ```json, quitamos también la palabra "json"
if text.startswith("json"):
text = text[4:]
return text.strip()La siguiente función es de las más complejas e importantes. Sirve para validar una relación extraída por el modelo contra la ontología. Esta función es el corazón de la fase de validación: no confiamos ciegamente en que el modelo vaya a respetar las reglas solo porque se lo pedimos en el prompt. Aquí comprobamos en código que efectivamente:
- La relación realmente exista en la ontología (que no se haya inventado un nombre nuevo).
- El tipo del sujeto sea compatible con el dominio declarado.
- El tipo del objeto sea compatible con el rango declarado.
Al final, devuelve una tupla (es_valida, motivo). Si es_valida es False, motivo explica por qué, para poder guardarlo junto a la relación descartada y revisarlo después.
def validate_relation(rel: dict, ontology: dict) -> tuple[bool, str]:
# Buscamos la definición de esta relación dentro de la lista de
# relaciones permitidas por su nombre exacto (ej. "ALIADA_DE").
definition = next((r for r in ontology["relaciones"] if r["nombre"] == rel.get("relacion")), None)
if definition is None:
# El LLM devolvió un nombre de relación que no existe en la ontología.
return False, f"Relación desconocida: {rel.get('relacion')}"
# El dominio/rango en la ontología puede ser un string suelto
# ("Personaje") o una lista (["CasaNoble", "Organizacion"]).
# Normalizamos siempre a lista para poder usar "in" de forma uniforme.
domain = definition["dominio"]
domain = [domain] if isinstance(domain, str) else domain
range_ = definition["rango"]
range_ = [range_] if isinstance(range_, str) else range_
if rel.get("tipo_sujeto") not in domain:
return False, f"'{rel.get('relacion')}' no admite sujeto de tipo '{rel.get('tipo_sujeto')}' (dominio válido: {domain})"
if rel.get("tipo_objeto") not in range_:
return False, f"'{rel.get('relacion')}' no admite objeto de tipo '{rel.get('tipo_objeto')}' (rango válido: {range_})"
# Si llegamos aquí, la relación existe y respeta dominio y rango.
return True, ""Y ya la función principal, donde orquestamos todo el flujo.
def main():
# --- Paso 1: cargar los insumos ---
ontology = load_ontology()
chunks = load_chunks()
ontology_description = build_ontology_description(ontology)
# --- Paso 2: configurar el modelo ---
# temperature=0 porque queremos extracción determinista y consistente,
# no creatividad. Para tareas de extracción estructurada, la
# temperatura alta solo añade variabilidad indeseada entre ejecuciones.
llm = ChatDeepSeek(model="deepseek-chat", temperature=0)
# ChatPromptTemplate.from_messages construye la plantilla de conversación
# que le enviaremos al modelo: un mensaje de sistema (con las reglas y
# la ontología) y un mensaje de usuario (con el fragmento a analizar).
prompt = ChatPromptTemplate.from_messages([
("system", SYSTEM_PROMPT),
("user", USER_PROMPT),
])
# El operador | encadena la plantilla de prompt con el modelo:
# esto es la sintaxis de LCEL (LangChain Expression Language).
# Al invocar "chain", primero se rellena el prompt con las variables
# y después el resultado se envía automáticamente al LLM.
chain = prompt | llm
# Listas donde iremos acumulando los resultados de TODOS los párrafos,
# no solo del último procesado.
all_entities = []
all_relations = []
discarded_relations = []
# --- Paso 3: procesar cada párrafo (chunk) por separado ---
for i, chunk in enumerate(chunks, start=1):
print(f"\n--- Procesando párrafo {i}/{len(chunks)} ---")
# Invocamos la cadena: esto rellena el prompt con la descripción
# de la ontología y el fragmento actual, y hace la llamada real
# a la API de DeepSeek. response.content contiene el texto
# devuelto por el modelo.
response = chain.invoke({
"ontology_description": ontology_description,
"fragment": chunk,
})
# Intentamos interpretar la respuesta como JSON. Si el modelo
# devolvió algo que no es JSON válido (puede pasar, los LLMs no
# son perfectos), lo registramos y pasamos al siguiente párrafo
# en vez de detener todo el script.
try:
data = json.loads(clean_json_response(response.content))
except json.JSONDecodeError:
print(" Respuesta no es JSON válido, se omite este párrafo.")
continue
# Si alguna clave falta en la respuesta, usamos lista vacía por defecto.
entities = data.get("entidades", [])
relations = data.get("relaciones", [])
# Las entidades NO se validan en este script (ver docstring inicial),
# así que se añaden directamente al resultado global.
all_entities.extend(entities)
print(f" Entidades detectadas: {len(entities)}")
# Cada relación SÍ pasa por validate_relation antes de aceptarse.
valid_count = 0
for rel in relations:
is_valid, reason = validate_relation(rel, ontology)
if is_valid:
all_relations.append(rel)
valid_count += 1
else:
# Guardamos la relación descartada junto con el motivo,
# en vez de simplemente ignorarla. Esto es clave para poder
# auditar después qué intentó hacer el LLM y por qué se
# rechazó (por ejemplo, para detectar fallos de diseño en
# la propia ontología, como nos pasó con LEAL_A y GOBIERNA).
discarded_relations.append({**rel, "motivo_descarte": reason})
print(f" Relación descartada: {reason}")
print(f" Relaciones válidas: {valid_count}/{len(relations)}")
# --- Paso 4: guardar el resultado consolidado ---
result = {
"entidades": all_entities,
"relaciones": all_relations,
"relaciones_descartadas": discarded_relations,
}Ale op, ejecutamos el script y nos debería dar algo así:
{
"entidades": [
{
"nombre": "Casa Stark",
"tipo": "CasaNoble",
"propiedades": {
"nombre": "Casa Stark",
"sede": "Invernalia",
"lema": ""
}
},
{
"nombre": "Casa Tully",
"tipo": "CasaNoble",
"propiedades": {
"nombre": "Casa Tully",
"sede": "Aguasdulces",
"lema": ""
}
},
{
"nombre": "Ned Stark",
"tipo": "Personaje",
"propiedades": {
"nombre": "Ned Stark",
"estado": "",
"titulo": ""
}
},
{
"nombre": "Jon Snow",
"tipo": "Personaje",
"propiedades": {}
},
{
"nombre": "Guardia de la Noche",
"tipo": "Organizacion",
"propiedades": {
"tipo": "orden militar"
}
},
{
"nombre": "Caminantes Blancos",
"tipo": "Organizacion",
"propiedades": {
"tipo": "ejército no humano"
}
},
{
"nombre": "Ejército de los Muertos",
"tipo": "Organizacion",
"propiedades": {
"tipo": "ejército no humano"
}
},
{
"nombre": "Rocadragón",
"tipo": "Region",
"propiedades": {
"tipo": "fortaleza"
}
},
{
"nombre": "Casa Targaryen",
"tipo": "CasaNoble",
"propiedades": {}
},
{
"nombre": "Poniente",
"tipo": "Region",
"propiedades": {
"tipo": "continente"
}
}
[... muchos registros más ...]
],
"relaciones": [
{
"sujeto": "Casa Stark",
"tipo_sujeto": "CasaNoble",
"relacion": "GOBIERNA",
"objeto": "Norte",
"tipo_objeto": "Region",
"momento": ""
},
{
"sujeto": "Casa Stark",
"tipo_sujeto": "CasaNoble",
"relacion": "ALIADA_DE",
"objeto": "Casa Tully",
"tipo_objeto": "CasaNoble",
"momento": "antes de la Guerra de los Cinco Reyes"
},
{
"sujeto": "Ned Stark",
"tipo_sujeto": "Personaje",
"relacion": "PERTENECE_A",
"objeto": "Casa Stark",
"tipo_objeto": "CasaNoble",
"momento": ""
},
{
"sujeto": "Catelyn Tully",
"tipo_sujeto": "Personaje",
"relacion": "PERTENECE_A",
"objeto": "Casa Tully",
"tipo_objeto": "CasaNoble",
"momento": ""
},
{
"sujeto": "Ned Stark",
"tipo_sujeto": "Personaje",
"relacion": "CASADO_CON",
"objeto": "Catelyn Tully",
"tipo_objeto": "Personaje",
"momento": "antes de la Guerra de los Cinco Reyes"
},
{
"sujeto": "Casa Targaryen",
"tipo_sujeto": "CasaNoble",
"relacion": "EN_GUERRA_CON",
"objeto": "Caminantes Blancos",
"tipo_objeto": "Organizacion",
"momento": "hacia el desenlace del conflicto"
},
{
"sujeto": "Guardia de la Noche",
"tipo_sujeto": "Organizacion",
"relacion": "EN_GUERRA_CON",
"objeto": "Caminantes Blancos",
"tipo_objeto": "Organizacion",
"momento": "hacia el desenlace del conflicto"
}
[... muchos registros más ...]
],
"relaciones_descartadas": [
{
"sujeto": "Casa Lannister",
"tipo_sujeto": "CasaNoble",
"relacion": "EN_GUERRA_CON",
"objeto": "Stannis Baratheon",
"tipo_objeto": "Personaje",
"momento": "durante la Guerra de los Cinco Reyes",
"motivo_descarte": "'EN_GUERRA_CON' no admite objeto de tipo 'Personaje' (rango válido: ['CasaNoble', 'Organizacion'])"
},
{
"sujeto": "Jon Snow",
"tipo_sujeto": "Personaje",
"relacion": "GOBIERNA",
"objeto": "Guardia de la Noche",
"tipo_objeto": "Organizacion",
"momento": "tras ascender a Lord Comandante",
"motivo_descarte": "'GOBIERNA' no admite objeto de tipo 'Organizacion' (rango válido: ['Region'])"
}
]
}Como decía, en el git puedes ver el ejemplo completo.
6. Carga en Neo4j
Si trabajáramos contra el listado en bruto que se ha generado, que solo es una lista plana de entidades y relaciones, un archivo de texto sin capacidad de navegación, las consultas serían una fatiga. Por ejemplo, si quisiéramos responder cómo se conocieron Jon Snow y Daenerys directamente sobre ese JSON, tendríamos que escribir código a mano para buscar todas las relaciones donde aparezca «Jon Snow», luego todas donde aparezca cada una de esas entidades relacionadas, y así sucesivamente. Sin embargo, eso mismo es lo que hace una base de datos de grafos de forma nativa y eficiente, así que vamos a llevar el listado a Neo4j.
Para eso, vamos a preparar un script que coja el archivo extraction.json generado en la fase anterior y lo cargue en Neo4j, convirtiendo la lista plana de entidades y relaciones en un grafo real y consultable.
Durante la carga, también va a resolver las entidades duplicadas que surgen de procesar el corpus párrafo a párrafo, como sucede por ejemplo con «Casa Stark», que aparecía varias veces como registros separados. Cuando se encuentre con estos casos, los va a fusionar en un único nodo mediante una versión normalizada del nombre.
Además, para las relaciones que la ontología marca como simétricas, como una alianza, donde si A es aliado de B entonces B es aliado de A, generará automáticamente la arista en ambos sentidos, aunque el texto original solo la mencione en una dirección.
El resultado será un grafo en el que cada personaje, casa noble u organización será un único nodo con sus propiedades, conectado por relaciones navegables, listo para las fases de detección de comunidades y consulta.
Pero antes de ver el script, tenemos que añadir tres variables a .env:
NEO4J_URI=neo4j+s://... RESTO URL.
NEO4J_USERNAME=EL ID DE USUARIO QUE NOS DIO AL CREARLA
NEO4J_PASSWORD=LA PASSWORD QUE NOS DIO AL CREARLAPara ver cuál es exactamente la url, seleccionamos la instancia y luego "connection details".
Y ahora ya sí vamos con el script. Al igual que antes,
puedes verlo completo en git hub, que es más fácil de leer.
Aquí lo voy desglosando.
Cargamos todo, definimos las rutas de la ontología y la extracción que preparamos antes y cargamos las variables de entorno de Neo4j.
import json
import os
import unicodedata
from pathlib import Path
from dotenv import load_dotenv
from neo4j import GraphDatabase
load_dotenv()
ONTOLOGY_PATH = Path("ontology.json")
EXTRACTION_PATH = Path("extraction.json")
NEO4J_URI = os.environ["NEO4J_URI"]
NEO4J_USERNAME = os.environ["NEO4J_USERNAME"]
NEO4J_PASSWORD = os.environ["NEO4J_PASSWORD"]Añadimos una función sin más que recupera un json y lo devuelve.
def load_json(path: Path) -> dict:
with open(path, encoding="utf-8") as f:
return json.load(f)Añadimos una función para normalizar un nombre para usarlo como clave de fusión de entidades. Pasamos a minúsculas y quitamos tildes (via unicodedata), para que «Daenerys» y «daenerys» se traten como la misma entidad. No quitamos espacios internos porque equiparar «Jon Snow» y «Jon» requeriría fuzzy matching o un paso adicional con el modelo, que dejo fuera para no complicar aún más el tutorial.
def normalize_name(name: str) -> str:
text = name.strip().lower()
# Descompone caracteres acentuados (é -> e + acento) y luego nos quedamos solo con los caracteres que no son marcas diacríticas, lo que efectivamente elimina las tildes sin tocar el resto del texto.
text = unicodedata.normalize("NFKD", text)
text = "".join(c for c in text if not unicodedata.combining(c))
return textCon la siguiente función recorremos la ontología y extraemos solo los nombres de las relaciones que son simétricas, como ALIADA_DE, CASADO_CON, EN_GUERRA_CON, SE_ENCUENTRA_CON. Las que marcamos con "simetrica": true en el JSON.
def build_symmetric_relation_names(ontology: dict) -> set[str]:
return {rel["nombre"] for rel in ontology["relaciones"] if rel.get("simetrica")}La siguiente función es la más importante, la que nos va a servir para cargar las entidades como nodos en Neo4j. La muestro entera directamente y luego explico algunos detalles.
def load_entities(tx, entities: list[dict]):
for entity in entities:
node_type = entity.get("tipo")
name = entity.get("nombre")
if not node_type or not name:
continue
properties = entity.get("propiedades", {}) or {}
properties["nombre"] = name
query = f"""
MERGE (n:{node_type} {{normalizedName: $normalized_name}})
SET n += $properties
"""
tx.run(
query,
normalized_name=normalize_name(name),
properties=properties,
)Vamos línea por línea.
def load_entities(tx, entities: list[dict]):tx es una transacción de Neo4j (la recibe de fuera, no la crea esta función). Todo lo que hagamos con tx dentro de esta función ocurre dentro de esa transacción. Si algo falla a mitad, Neo4j puede revertirlo entero.
for entity in entities:
node_type = entity.get("tipo")
name = entity.get("nombre")
if not node_type or not name:
continueRecorremos cada entidad del JSON. Si a alguna le falta el tipo o el nombre (datos imprescindibles para crear un nodo válido), la saltamos con continue en vez de dejar que el script explote, ya que una entidad mal formada no debería tirar abajo toda la carga.
properties = entity.get("propiedades", {}) or {}
properties["nombre"] = nameTomamos las propiedades que vengan en el JSON (estado, titulo, sede...) y les añadimos también el nombre, ya que vamos a usar una versión normalizada del nombre como identificador interno del nodo (sin tildes, en minúsculas), pero queremos conservar el nombre original («Daenerys Targaryen», con mayúsculas y tildes) como un dato visible del nodo, para mostrarlo después en las consultas.
query = f"""
MERGE (n:{node_type} {{normalizedName: $normalized_name}})
SET n += $properties
"""Aquí está el corazón de la función: la consulta Cypher. Vamos a desglosarla.
- Lo primero es destacar que estamos usando
MERGEen vez deCREATE, ya queCREATEcrearía un nodo nuevo cada vez, sin importar si ya existe uno igual, por lo que habríamos acabado con «Casa Stark» repetida 5 veces.MERGEen cambio comprueba si ya existe un nodo, en cuyo caso lo usa y, si no existe, lo crea. La condición aquí esnormalizedName: $normalized_name. Por eso todas las apariciones de «Casa Stark», aunque vengan de párrafos distintos, terminan fusionándose en un único nodo. - El segundo detalle importante es
{node_type}insertado con f-string, no como parámetro. Es decir,node_type, comoPersonajeoCasaNoble, se mete directamente en el texto de la consulta, mientras quenormalized_namese pasa como parámetro ($normalized_name). Esto es así porque los parámetros solo pueden sustituir valores, como un nombre o un número, nunca nombres de etiquetas (labels) o tipos de relación. Por eso la etiqueta del nodo tiene que insertarse a mano en el texto de la consulta, como cuando en JavaScript se genera un objeto al vuelo. Esto podría asustar en otro escenario en tanto que podría utilizarse como una especie de inyección SQL, pero aquí es seguro porquenode_typeno viene de texto libre del usuario, sino de un valor que ya pasó por la validación contra la ontología, así que solo puede ser uno de los 5 tipos que definimos. Vamos, tendría que ser el propio modelo quien tuviera intenciones malvadas. - Por último, hay que destacar ese
SET n += $properties, que es la que permite que se realice la fusión sin perder datos. Si solo fuera un=en vez de un+=, cada vez que Neo4j toque ese nodo reemplazaría todas sus propiedades por las nuevas, borrando lo que hubiera antes.
Por último queda la llamada final:
tx.run(
query,
normalized_name=normalize_name(name),
properties=properties,
)tx.run() ejecuta la consulta Cypher sustituyendo $normalized_name y $properties por los valores reales que le pasamos. normalize_name(name) es la función que vimos antes (minúsculas, sin tildes), es la pieza que decide qué cuenta como la misma entidad para efectos de fusión.
Al final, lo que obtenemos es algo así. En el caso de que entities contenga estas tres entradas llegadas de párrafos distintos:
{"nombre": "Casa Stark", "tipo": "CasaNoble", "propiedades": {"sede": "Invernalia"}}
{"nombre": "Casa Stark", "tipo": "CasaNoble", "propiedades": {}}
{"nombre": "Casa Stark", "tipo": "CasaNoble", "propiedades": {"lema": "Se acerca el invierno"}}En Neo4j se convertirían en un único nodo:
(:CasaNoble {normalizedName: "casa stark", nombre: "Casa Stark", sede: "Invernalia", lema: "Se acerca el invierno"})Con las propiedades de las tres apariciones combinadas, gracias al +=.
Seguimos.
Ahora toca sacar las relaciones y para eso podemos preparar una función que recorra cada relación extraída (sujeto-relación-objeto) y genere la arista correspondiente entre los dos nodos ya existentes en Neo4j: primero busca ambos nodos por su nombre normalizado (MATCH), y luego crea la conexión entre ellos con MERGE para no duplicar la arista si ya existía, guardando también el atributo momento si lo hay. Si la relación es de las marcadas como simétricas en la ontología (como ALIADA_DE), repite el mismo proceso pero invirtiendo sujeto y objeto, para dejar la arista creada en ambos sentidos sin depender de que el texto original la hubiera mencionado dos veces.
def load_relations(tx, relations: list[dict], symmetric_names: set[str]):
for rel in relations:
relation_name = rel.get("relacion")
subject_name = rel.get("sujeto")
object_name = rel.get("objeto")
if not relation_name or not subject_name or not object_name:
continue
# El atributo "momento" es opcional; si no viene, guardamos cadena vacía para mantener el esquema de propiedades uniforme.
moment = rel.get("momento", "")
# Cypher no permite parametrizar el tipo de relación, así que lo insertamos con f-string. Es seguro aquí porque relation_name viene siempre de la validación contra la ontología, no de texto libre arbitrario.
query = f"""
MATCH (s {{normalizedName: $subject_name}})
MATCH (o {{normalizedName: $object_name}})
MERGE (s)-[r:{relation_name}]->(o)
SET r.momento = $moment
"""
tx.run(
query,
subject_name=normalize_name(subject_name),
object_name=normalize_name(object_name),
moment=moment,
)
# Si la relación es simétrica, repetimos la operación invirtiendo sujeto y objeto, para dejar también la arista en sentido contrario.
if relation_name in symmetric_names:
tx.run(
query,
subject_name=normalize_name(object_name),
object_name=normalize_name(subject_name),
moment=moment,
)Por último, preparamos una función para orquestar todo el tinglado.
def main():
ontology = load_json(ONTOLOGY_PATH)
extraction = load_json(EXTRACTION_PATH)
symmetric_names = build_symmetric_relation_names(ontology)
entities = extraction.get("entidades", [])
relations = extraction.get("relaciones", [])
print(f"Entidades a cargar (antes de fusionar duplicados): {len(entities)}")
print(f"Relaciones a cargar: {len(relations)}")
print(f"Relaciones simétricas en la ontología: {sorted(symmetric_names)}")
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))
with driver.session() as session:
# execute_write envuelve la operación en una transacción: si algo falla a mitad de la carga, Neo4j revierte los cambios de ese bloque en vez de dejar el grafo a medio cargar.
session.execute_write(load_entities, entities)
session.execute_write(load_relations, relations, symmetric_names)
# Contamos cuántos nodos y relaciones distintos quedaron en la base de datos tras la fusión, para comparar contra los totales en bruto de arriba y ver cuánto se redujo por deduplicación.
node_count = session.run("MATCH (n) RETURN count(n) AS total").single()["total"]
relation_count = session.run("MATCH ()-[r]->() RETURN count(r) AS total").single()["total"]
driver.close()
print("\n=== RESUMEN ===")
print(f"Nodos finales en Neo4j: {node_count}")
print(f"Relaciones finales en Neo4j: {relation_count}")
if __name__ == "__main__":
main()Ahora ya solo faltaría ejecutarlo todo.
python load-graph.py
Entidades a cargar (antes de fusionar duplicados): 76
Relaciones a cargar: 62
Relaciones simétricas en la ontología: ['ALIADA_DE', 'CASADO_CON', 'EN_GUERRA_CON', 'SE_ENCUENTRA_CON']
=== RESUMEN ===
Nodos finales en Neo4j: 45
Relaciones finales en Neo4j: 71Y si todo ha ido bien, en la bbdd veremos todas las entidades y sus aristas.
7. Búsquedas
7.1. Tipos de búsqueda
Con todo este engranaje, ya podríamos realizar una primera búsqueda mediante esta técnica, pero antes debemos saber que hay dos maneras de consultar un grafo GraphRAG: Local Search y Global Search.
Local Search se usa para preguntas sobre entidades concretas y sus conexiones directas. Por ejemplo: ¿cómo se conocieron Jon Snow y Daenerys? Sabemos exactamente de qué dos nodos partir, y la respuesta está en el vecindario cercano de esos nodos, entendiendo por vecindario en este contexto el conjunto de nodos que están conectados a un nodo, directa o indirectamente, dentro de un número limitado de saltos (hops). Así, un vecindario a 1 salto son todos los nodos conectados directamente a un nodo, sin pasar por ningún intermediario; a 2 saltos, son los del salto 1, más todos los nodos conectados a esos, sin pasar por un tercer intermediario.
Por ejemplo, preguntar cómo se conocieron Jon Snow y Daenerys sería una Local Search donde el grafo se recorre por proximidad.
[Jon Snow] --ES_MIEMBRO_DE--> [Guardia de la Noche]
|
[Jon Snow] --SE_ENCUENTRA_CON--> [Daenerys] --momento: "en Rocadragón"
|
[Daenerys] --ALIADA_DE--> [Casa Targaryen]El sistema solo mira lo que está cerca de esos dos nodos ancla. Todo lo demás del grafo (Casa Lannister, la Boda Roja, la Compañía Dorada...) es irrelevante para la pregunta y ni se toca.
Global Search, en cambio, no parte de ningún nodo, resume el grafo entero por bloques. Son consultas para preguntas que no apuntan a ninguna entidad concreta del grafo, sino que piden una vista de conjunto, como qué casas nobles tienen más alianzas y cuáles están más aisladas o qué patrones de traición aparecen a lo largo de la historia.
Como veremos, en este caso, en vez de recorrer el grafo en el momento de preguntar, el sistema lee resúmenes de comunidades generados de antemano y los combina mediante un proceso map-reduce para sintetizar la respuesta final. Vamos ahora con las primeras.
7.2. Local Search
Para implementar la búsqueda vamos a preparar un script que nos conecte con Neo4j. La clave está en esta sentencia:
GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))Ese objeto driver será la conexión maestra desde python hacia la base de datos. Ojo, no es que abra la conexión, simplemente preparará el objeto con la URI y las credenciales ya configuradas para usarse cuando se necesite.
Por lo demás, la pipeline del script sigue los siguientes pasos:
- identifica las entidades ancla mencionadas en la pregunta,
- recupera su vecindario desde Neo4j,
- convierte ese subgrafo en texto legible,
- y por último le pasa ese texto como contexto a un segundo modelo que redacta la respuesta final.
Es literalmente el patrón Retrieval-Augmented Generation aplicado a un grafo en vez de a texto plano: primero se recupera contexto relevante, después se genera la respuesta apoyándose en ese contexto.
Como en las otras ocasiones,
el script completo está en git.
Aquí lo voy desglosando.
Cargamos todo lo que vamos a necesitar.
import json
import os
import unicodedata
from dotenv import load_dotenv
from langchain_deepseek import ChatDeepSeek
from langchain_core.prompts import ChatPromptTemplate
from neo4j import GraphDatabase
load_dotenv()
NEO4J_URI = os.environ["NEO4J_URI"]
NEO4J_USERNAME = os.environ["NEO4J_USERNAME"]
NEO4J_PASSWORD = os.environ["NEO4J_PASSWORD"]Para no complicar más las cosas, dejamos la pregunta fija, que en la vida real se cogería de vía rest o similar.
QUESTION = "¿Cómo se conocieron Jon Snow y Daenerys?"Indicamos cuántos saltos de distancia recorremos desde cada entidad ancla. Lo dejamos como constante configurable arriba del todo para que sea fácil experimentar subiendo o bajando el radio de búsqueda y observar cómo cambia el contexto recuperado y, en consecuencia, la respuesta final.
MAX_HOPS = 2Añadimos un método para normalizar los nombres de antes (minúsculas, sin tildes). Es imprescindible reutilizar exactamente la misma lógica aquí, porque los nodos en Neo4j se guardaron con esa clave normalizada, y si buscáramos con una normalización distinta, simplemente no encontraríamos coincidencias. En la vida real, estaría situado en un archivo de utils.
def normalize_name(name: str) -> str:
text = name.strip().lower()
text = unicodedata.normalize("NFKD", text)
text = "".join(c for c in text if not unicodedata.combining(c))
return textPreparamos un prompt que hace de extractor de entidades ligero. En vez de intentar detectar nombres propios con reglas de texto (mayúsculas, expresiones regulares, etc.), delegamos esa tarea en el propio modelo, que entiende de forma mucho más robusta que «Jon» y «Jon Snow» se refieren a la misma clase de cosa (un nombre propio a buscar en el grafo), incluso si la pregunta usa una forma abreviada o coloquial.
ENTITY_EXTRACTION_SYSTEM_PROMPT = """Identifica los nombres propios de personajes, casas nobles, organizaciones, regiones o eventos que se mencionan en la pregunta del usuario. Responde ÚNICAMENTE con un JSON de la forma {"entidades": ["nombre1", "nombre2"]}, sin texto adicional."""
ENTITY_EXTRACTION_USER_PROMPT = """Pregunta: {question}"""Le pedimos al modelo que identifique los nombres propios mencionados en la pregunta, para usarlos como nodos de partida (nodos ancla) al recorrer el grafo. El resultado de esta función es la lista de nombres que luego intentaremos localizar como nodos reales en Neo4j.
def extract_anchor_entities(chain, question: str) -> list[str]:
response = chain.invoke({"question": question})
text = response.content.strip()
# Igual que antes, nos protegemos por si el modelo envuelve la respuesta en un bloque de código markdown pese a habérselo pedido explícitamente que no lo haga.
if text.startswith("```"):
text = text.split("```")[1]
if text.startswith("json"):
text = text[4:]
data = json.loads(text.strip())
return data.get("entidades", [])Recuperamos para cada entidad ancla el vecindario, es decir, todas las relaciones alcanzables hasta max_hops saltos, en cualquier dirección. Devuelve una lista de tripletas (sujeto, relación, objeto, momento) ya en formato simple de diccionario python, lista para convertir a texto en el siguiente paso, sin que el resto del script tenga que preocuparse de los tipos propios del driver de Neo4j.
def fetch_neighborhood(session, anchor_names: list[str], max_hops: int) -> list[dict]:
triples = []
for name in anchor_names:
# La sintaxis [*1..max_hops] es la que permite un recorrido de longitud variable en Cypher: en vez de fijar el patrón a exactamente un salto, le decimos a Neo4j que explore caminos de entre 1 y max_hops relaciones de longitud, en cualquier dirección (por eso no ponemos flecha -> ni <- en el patrón).
result = session.run(
f"""
MATCH (anchor {{normalizedName: $normalized_name}})
MATCH path = (anchor)-[*1..{max_hops}]-(neighbor)
UNWIND relationships(path) AS rel
RETURN startNode(rel).nombre AS subject, type(rel) AS relation, endNode(rel).nombre AS object, rel.momento AS moment
""",
normalized_name=normalize_name(name),
)
# UNWIND relationships(path) descompone cada camino completo (que puede tener 1 o 2 relaciones seguidas) en sus relaciones individuales, para poder leer el sujeto y el objeto de cada tramo por separado, en vez de quedarnos solo con el nodo final del camino.
for record in result:
triples.append(dict(record))
return triplesVarias entidades ancla pueden traer la misma relación repetida; por ejemplo, si Jon Snow y Daenerys comparten un vecino en su vecindario a 2 saltos, esa relación compartida se recuperaría una vez desde cada ancla. Quitamos esos duplicados exactos antes de construir el contexto final, para no inflar el prompt con información redundante.
def deduplicate_triples(triples: list[dict]) -> list[dict]:
seen = set()
unique = []
for triple in triples:
key = (triple["subject"], triple["relation"], triple["object"])
if key not in seen:
seen.add(key)
unique.append(triple)
return uniqueAl igual que antes, convertimos la lista de tripletas en texto legible en lenguaje natural, para incluirlo como contexto en el prompt final del modelo generador. No usamos el JSON en bruto aquí porque un LLM redacta mejores respuestas cuando el contexto ya viene en forma de frases simples, en vez de tener que interpretar una estructura de datos anidada.
def format_context(triples: list[dict]) -> str:
lines = []
for triple in triples:
line = f"- {triple['subject']} {triple['relation']} {triple['object']}"
# El atributo "momento" es opcional; solo lo añadimos a la línea si realmente tiene contenido, para no ensuciar el contexto con paréntesis vacíos.
if triple.get("moment"):
line += f" ({triple['moment']})"
lines.append(line)
return "\n".join(lines)Enviamos a un segundo modelo de la pipeline todo el tinglado para que redacte la respuesta final. Le indicamos además de forma explícita que no se invente información fuera del contexto y que admita que le falta.
ANSWER_SYSTEM_PROMPT = """Eres un asistente que responde preguntas sobre Juego de Tronos usando ÚNICAMENTE la información del siguiente contexto, extraído de un grafo de conocimiento. No inventes información que no esté en el contexto. Si el contexto no es suficiente para responder, dilo explícitamente.
CONTEXTO (tripletas del grafo):
{context}
"""
ANSWER_USER_PROMPT = """Pregunta: {question}"""Y con todo listo, ya solo nos falta la función principal para orquestar toda la pipeline.
def main():
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))
# Reutilizamos la misma instancia de ChatDeepSeek para las dos llamadas del pipeline (extracción de entidades y generación de la respuesta final), ya que ambas comparten el mismo modelo y configuración; solo cambia la plantilla de prompt que se le encadena en cada caso.
llm = ChatDeepSeek(model="deepseek-chat", temperature=0)
extraction_prompt = ChatPromptTemplate.from_messages([
("system", ENTITY_EXTRACTION_SYSTEM_PROMPT),
("user", ENTITY_EXTRACTION_USER_PROMPT),
])
extraction_chain = extraction_prompt | llm
print(f"Pregunta: {QUESTION}\n")
# Paso 1: identificar los nodos de partida a partir de la pregunta.
anchor_entities = extract_anchor_entities(extraction_chain, QUESTION)
print(f"Entidades ancla identificadas: {anchor_entities}")
# Paso 2: recuperar el vecindario de esas entidades desde Neo4j.
with driver.session() as session:
triples = fetch_neighborhood(session, anchor_entities, MAX_HOPS)
driver.close()
unique_triples = deduplicate_triples(triples)
print(f"Tripletas recuperadas del vecindario (tras deduplicar): {len(unique_triples)}")
# Paso 3: convertir el subgrafo recuperado en texto legible.
context = format_context(unique_triples)
print(f"\n--- Contexto recuperado ---\n{context}\n")
# Paso 4: generar la respuesta final apoyándose en ese contexto.
answer_prompt = ChatPromptTemplate.from_messages([
("system", ANSWER_SYSTEM_PROMPT),
("user", ANSWER_USER_PROMPT),
])
answer_chain = answer_prompt | llm
response = answer_chain.invoke({"context": context, "question": QUESTION})
print("--- Respuesta final ---")
print(response.content)
if __name__ == "__main__":
main()Ale op, si lo ejecutamos, con todos los prints que hemos puesto nos dirá algo así.
Pregunta: ¿Cómo se conocieron Jon Snow y Daenerys?
Entidades ancla identificadas: ['Jon Snow', 'Daenerys']
Tripletas recuperadas del vecindario (tras deduplicar): 11
--- Contexto recuperado ---
- Jon Snow ES_MIEMBRO_DE Guardia de la Noche (antes de ser Lord Comandante)
- Guardia de la Noche ALIADA_DE Casa Stark (durante la coalición Stark-Targaryen)
- Guardia de la Noche ALIADA_DE Casa Targaryen (durante la coalición Stark-Targaryen)
- Casa Stark ALIADA_DE Guardia de la Noche (durante la coalición Stark-Targaryen)
- Casa Targaryen ALIADA_DE Guardia de la Noche (durante la coalición Stark-Targaryen)
- Guardia de la Noche EN_GUERRA_CON Caminantes Blancos (hacia el desenlace del conflicto)
- Guardia de la Noche EN_GUERRA_CON Ejército de los Muertos (durante el mando de Jon Snow)
- Caminantes Blancos EN_GUERRA_CON Guardia de la Noche (hacia el desenlace del conflicto)
- Ejército de los Muertos EN_GUERRA_CON Guardia de la Noche (durante el mando de Jon Snow)
- Jon Snow SE_ENCUENTRA_CON Daenerys (en Rocadragón)
- Daenerys SE_ENCUENTRA_CON Jon Snow (en Rocadragón)
--- Respuesta final ---
Según la información disponible, Jon Snow y Daenerys se encontraron en Rocadragón.7.3. Comunidades
Para ver cómo funciona la búsqueda global, antes debemos saber qué es una comunidad, también conocida como clúster o módulo. Una comunidad es un grupo de nodos que están mucho más conectados entre sí que con el resto del grafo y lo interesante es que esta etiqueta no la pone un ser humano a mano, sino que la identifica un algoritmo que descubre matemáticamente, mirando solo la estructura de conexiones, quién está unido a quién, sin saber nada del significado de esos nodos. Es decir, no somos nosotros quienes definimos una comunidad Stark-Targaryen, sino un algoritmo que advierte que entre las dos casas, entre los dos conjuntos de nodos, se producen muchas conexiones.
Entonces, los pasos son:
- Detectar las comunidades mediante un algoritmo.
- Preparar resúmenes sintéticos sobre las mismas.
- Formular las cuestiones globales.
Para identificar las comunidades podemos usar varios algoritmos. Nosotros vamos a usar Louvain, que detecta comunidades optimizando la modularidad: una métrica que mide si un grupo de nodos tiene más conexiones internas de las esperables por azar y menos conexiones hacia fuera. Funciona en rondas: primero mueve nodos entre comunidades vecinas mientras eso mejore la modularidad, y luego colapsa cada comunidad encontrada en un único súper-nodo para repetir el proceso sobre un grafo más simple, hasta que ya no hay mejoras posibles.
El número de comunidades no se decide de antemano, sino que es una consecuencia de maximizar la modularidad, no un parámetro que nosotros elijamos. Es el mismo principio que usa Leiden, un algoritmo derivado de Louvain que corrige algunas de sus limitaciones y que es el que emplea la implementación de referencia de Microsoft GraphRAG (a través de la librería graspologic). Nosotros nos quedamos con Louvain y con una librería open source más ligera, NetworkX, ya que para el tamaño de este tutorial el resultado es equivalente y el código es más sencillo de seguir.
En el script, es el que se encuentra en esta línea
communities = nx.algorithms.community.louvain_communities(graph, seed=42)Recibe el objeto graph (un nx.Graph, no dirigido, que construimos justo antes con build_networkx_graph()) y devuelve una lista de conjuntos (set), donde cada conjunto contiene los identificadores de los nodos que quedaron agrupados en esa comunidad.
Por ejemplo, con nuestro grafo, el resultado podría verse así (simplificado):
[
{"casa stark", "casa tully", "jon snow", "daenerys", "guardia de la noche", ...},
{"casa lannister", "casa tyrell", "compañía dorada", ...},
{"caminantes blancos", "ejercito de los muertos"},
]El resto del script sirve para traer el grafo completo desde Neo4j, reconstruirlo en memoria con NetworkX, detectar comunidades con el algoritmo de Louvain, guardar de vuelta en Neo4j a qué comunidad pertenece cada nodo, y generar con DeepSeek un resumen en lenguaje natural de cada comunidad detectada.
La versión completa, en el git.
Aquí desglosada.
Cargamos todo.
import json
import os
from collections import defaultdict
from pathlib import Path
import networkx as nx
from dotenv import load_dotenv
from langchain_deepseek import ChatDeepSeek
from langchain_core.prompts import ChatPromptTemplate
from neo4j import GraphDatabase
load_dotenv()
OUTPUT_PATH = Path("communities.json")
NEO4J_URI = os.environ["NEO4J_URI"]
NEO4J_USERNAME = os.environ["NEO4J_USERNAME"]
NEO4J_PASSWORD = os.environ["NEO4J_PASSWORD"]Traemos todos los nodos y relaciones de Neo4j como listas de diccionarios simples de Python, en vez de como los objetos propios del driver (Node, Relationship), para poder pasárselos directamente a NetworkX sin conversiones intermedias.
def fetch_graph(session) -> tuple[list[dict], list[dict]]:
# normalizedName es el mismo identificador que usamos como clave de fusión en la Fase 3, así que lo reutilizamos aquí como identificador de nodo dentro de NetworkX: así garantizamos que el mismo nodo en Neo4j y en el grafo de NetworkX comparten exactamente la misma clave.
nodes_result = session.run("""
MATCH (n)
RETURN n.normalizedName AS id, n.nombre AS name, labels(n) AS labels
""")
nodes = [dict(record) for record in nodes_result]
# Aquí solo recuperamos el tipo de relación (type(r)), no sus propiedades como "momento", porque para la detección de comunidades lo único que le importa a Louvain es que existe una conexión entre dos nodos, no los detalles de esa conexión.
edges_result = session.run("""
MATCH (s)-[r]->(o)
RETURN s.normalizedName AS source, o.normalizedName AS target, type(r) AS relation_type
""")
edges = [dict(record) for record in edges_result]
return nodes, edgesConvertimos las listas planas de nodos y relaciones traídas de Neo4j en un grafo de NetworkX no dirigido, listo para que Louvain lo analice. Usamos un grafo no dirigido a propósito: Louvain detecta comunidades por densidad de conexiones, sin importar el sentido de la flecha, así que para efectos de "quién está en el mismo bando" da igual que el dato original fuera Stark ALIADA_DE Tully o Tully ALIADA_DE Stark, es la misma señal estructural de que ambos nodos están conectados.
def build_networkx_graph(nodes: list[dict], edges: list[dict]) -> nx.Graph:
graph = nx.Graph()
# Añadimos primero todos los nodos, aunque algunos puedan quedar sin ninguna relación (nodos aislados); NetworkX los mantiene igualmente en el grafo, aunque Louvain los tratará casi con seguridad como su propia comunidad de un solo miembro.
for node in nodes:
graph.add_node(node["id"], name=node["name"], labels=node["labels"])
# add_edge crea automáticamente los nodos si no existieran ya (no debería pasar aquí, porque ya los añadimos arriba), y si la misma arista se añade dos veces, NetworkX simplemente actualiza sus propiedades en vez de duplicarla, ya que un grafo simple de NetworkX no permite aristas paralelas entre el mismo par de nodos.
for edge in edges:
graph.add_edge(edge["source"], edge["target"], relation_type=edge["relation_type"])
return graphLa siguiente función ejecuta el algoritmo de Louvain sobre el grafo de NetworkX para detectar comunidades, usando resolution para controlar si prefiere pocas comunidades grandes o muchas pequeñas, y seed=42 para que el resultado sea reproducible entre ejecuciones; luego convierte el resultado (una lista de conjuntos de nodos) en un diccionario {nodo: id_de_comunidad}, formato mucho más práctico para consultar después a qué comunidad pertenece cualquier nodo concreto.
def detect_communities(graph: nx.Graph, resolution: float = 1.0) -> dict[str, int]:
communities = nx.algorithms.community.louvain_communities(graph, resolution=resolution, seed=42)
node_to_community = {}
# enumerate nos da un índice numérico (0, 1, 2...) para cada comunidad detectada, que usamos como su identificador.
for community_id, members in enumerate(communities):
for node_id in members:
node_to_community[node_id] = community_id
return node_to_communityEscribimos de vuelta en Neo4j la propiedad communityId de cada nodo, para que el resultado de la detección de comunidades quede persistido en la base de datos y disponible para futuras consultas Cypher, sin tener que volver a ejecutar Louvain cada vez que alguien quiera saber a qué comunidad pertenece un nodo.
def save_communities_to_neo4j(session, node_to_community: dict[str, int]):
for node_id, community_id in node_to_community.items():
session.run(
"MATCH (n {normalizedName: $node_id}) SET n.communityId = $community_id",
node_id=node_id,
community_id=community_id,
)Reorganizamos la lista plana de nodos en un diccionario {id_de_comunidad: [nodos]}, que es la forma que necesitamos para generar un resumen por comunidad en el siguiente paso: para escribir el resumen de la comunidad 0, necesitamos tener ya agrupados todos sus miembros juntos, no dispersos en una lista plana.
def group_members_by_community(nodes: list[dict], node_to_community: dict[str, int]) -> dict[int, list[dict]]:
grouped = defaultdict(list)
for node in nodes:
community_id = node_to_community.get(node["id"])
# Un nodo podría quedar sin comunidad asignada solo si estuviera completamente desconectado del resto del grafo y Louvain no lo hubiera procesado; esta comprobación es una salvaguarda defensiva para ese caso extremo.
if community_id is not None:
grouped[community_id].append(node)
return groupedVamos ya con los prompts. En este caso, le pediremos al modelo que actúe como analista, no como extractor de datos a diferencia de antes. Aquí el objetivo es una síntesis interpretativa en lenguaje natural, así que el tono de las instrucciones es deliberadamente distinto.
SUMMARY_SYSTEM_PROMPT = """Eres un analista que resume comunidades detectadas en un grafo de conocimiento sobre Juego de Tronos. Se te dará una lista de entidades (personajes, casas, organizaciones, etc.) que el algoritmo de detección de comunidades agrupó por estar densamente conectadas entre sí. Tu trabajo es escribir un resumen breve (2-4 frases) en español que explique qué representa este grupo: por ejemplo, si son una facción, una alianza, un bando enfrentado a otro, etc. Basa el resumen únicamente en los nombres de las entidades que se te dan, sin inventar información que no puedas inferir razonablemente de esos nombres."""
SUMMARY_USER_PROMPT = """Entidades de esta comunidad:
{members}
Escribe el resumen."""Llamamos a DeepSeek para generar el resumen en lenguaje natural de una comunidad concreta, a partir de la lista de nombres de sus miembros y su tipo (Personaje, CasaNoble...), que le da al modelo una pista adicional sobre la naturaleza de cada entidad sin tener que consultar sus propiedades completas.
def generate_community_summary(llm, chain, members: list[dict]) -> str:
# labels(n) en Neo4j devuelve una lista (por si un nodo tuviera varias etiquetas a la vez), así que tomamos el primer elemento, ya que en nuestra ontología cada nodo tiene exactamente un tipo.
member_names = ", ".join(f"{m['name']} ({m['labels'][0]})" for m in members)
response = chain.invoke({"members": member_names})
return response.content.strip()Y por último orquestamos toda la pipeline con la siguiente función.
def main():
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USERNAME, NEO4J_PASSWORD))
# Todo el trabajo que necesita la base de datos (leer el grafo y luego escribir communityId) se hace dentro de una única sesión, para reutilizar la misma conexión en vez de abrir y cerrar sesiones innecesariamente.
with driver.session() as session:
nodes, edges = fetch_graph(session)
print(f"Nodos traídos de Neo4j: {len(nodes)}")
print(f"Relaciones traídas de Neo4j: {len(edges)}")
graph = build_networkx_graph(nodes, edges)
node_to_community = detect_communities(graph, resolution=0.5)
num_communities = len(set(node_to_community.values()))
print(f"Comunidades detectadas: {num_communities}")
save_communities_to_neo4j(session, node_to_community)
driver.close()
# A partir de aquí ya no necesitamos la conexión a Neo4j: el resto del script trabaja solo con los datos que ya trajimos a memoria (nodes, node_to_community) y con llamadas al LLM.
grouped = group_members_by_community(nodes, node_to_community)
# temperature=0.3 es una diferencia deliberada respecto a antes (donde usamos 0 para extracción determinista): aquí queremos que el resumen suene natural y bien redactado, no una transcripción mecánica, así que permitimos un poco más de variabilidad sin llegar a los niveles altos que fomentarían que el modelo invente información no presente en los nombres.
llm = ChatDeepSeek(model="deepseek-chat", temperature=0.3)
prompt = ChatPromptTemplate.from_messages([
("system", SUMMARY_SYSTEM_PROMPT),
("user", SUMMARY_USER_PROMPT),
])
chain = prompt | llm
communities_output = []
# sorted(grouped.items()) simplemente nos asegura que los resúmenes se impriman y se guarden en orden de id de comunidad (0, 1, 2...), en vez de en un orden arbitrario dependiente de cómo Python recorra el diccionario internamente.
for community_id, members in sorted(grouped.items()):
print(f"\n--- Generando resumen de la comunidad {community_id} ({len(members)} miembros) ---")
summary = generate_community_summary(llm, chain, members)
print(f" {summary}")
communities_output.append({
"community_id": community_id,
"members": [m["name"] for m in members],
"summary": summary,
})
# Este archivo, communities.json, es exactamente lo que la Fase 6 (Global Search) va a leer en vez de tener que volver a consultar Neo4j o recalcular las comunidades desde cero.
OUTPUT_PATH.write_text(json.dumps(communities_output, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"\nGuardado en: {OUTPUT_PATH}")
if __name__ == "__main__":
main()Ejecutamos el script y en communities.json nos debería generar algo así:
{
"community_id": 0,
"members": [
"Invernalia"
],
"summary": "Esta comunidad está centrada en la región de Invernalia, el hogar ancestral de la Casa Stark en el Norte de Poniente. Representa el núcleo territorial y político de los Stark, siendo un punto clave en la lucha por el control del Norte. Su presencia como única entidad sugiere que el grupo se enfoca en la importancia estratégica y simbólica de esta región dentro del conflicto general."
},
{
"community_id": 1,
"members": [
"Aguasdulces"
],
"summary": "Esta comunidad está compuesta únicamente por la región de Aguasdulces, lo que sugiere que representa un área geográfica específica dentro del mundo de Juego de Tronos. Al ser una sola entidad, no forma una facción o alianza, sino que probablemente actúa como un punto de referencia territorial en el grafo de conocimiento."
},
{
"community_id": 2,
"members": [
"Casa Lannister",
"Cersei Lannister",
"Robert Baratheon",
"Desembarco del Rey",
"Tywin Lannister",
"Casa Tyrell",
"Margaery Tyrell",
"Joffrey Baratheon",
"Tommen Baratheon",
"Fe Militante",
"Compañía Dorada"
],
"summary": "Esta comunidad representa el núcleo del poder político y militar en torno al Trono de Hierro durante gran parte de la serie, centrado en la alianza entre la Casa Lannister y la Casa Tyrell. Incluye a figuras clave como Cersei, Tywin, Joffrey y Tommen, así como a Margaery Tyrell, y está vinculada a la capital, Desembarco del Rey. La presencia de la Fe Militante y la Compañía Dorada sugiere que este grupo también abarca fuerzas religiosas y mercenarias que apoyan o desafían el control de la capital."
},
{
"community_id": 3,
"members": [
"Casa Baratheon"
],
"summary": "Esta comunidad está compuesta exclusivamente por la Casa Baratheon, una de las grandes casas nobles de Poniente. Representa a la facción que reclama el Trono de Hierro, destacando por su linaje real y su símbolo del venado coronado. Su presencia solitaria sugiere que el algoritmo la identifica como un grupo cohesionado y distinto, posiblemente en oposición a otras casas o facciones rivales."
},
[...] y así todo el resto de comunidades7.4. Global search
El último script del tutorial es muy sencillo.
Implementa el patrón map-reduce sobre los resúmenes de comunidades generados antes evaluando la relevancia de cada comunidad frente a la pregunta y sintetizando las más relevantes en una única respuesta final.
Como siempre, empezamos cargando las cosas y seteando las constantes.
import json
from pathlib import Path
from dotenv import load_dotenv
from langchain_deepseek import ChatDeepSeek
from langchain_core.prompts import ChatPromptTemplate
load_dotenv()
COMMUNITIES_PATH = Path("communities.json")
QUESTION = "¿Cuáles son las principales facciones en conflicto y qué las mueve?"
# Solo se envían a la fase "reduce" las comunidades cuya relevancia (puntuada de 0 a 10 en la fase "map") sea igual o superior a este umbral. Esto es lo que debería filtrar de forma natural las comunidades de una sola región geográfica que vimos antes, sin tener que limpiarlas a mano del archivo communities.json.
RELEVANCE_THRESHOLD = 5Cargamos las comunidades.
def load_communities() -> list[dict]:
with open(COMMUNITIES_PATH, encoding="utf-8") as f:
return json.load(f)Preparamos el prompt de la fase map. Se ejecuta una vez por cada comunidad, de forma independiente, sin que una llamada sepa nada de las demás comunidades. Le pedimos al modelo que puntúe la relevancia y que extraiga solo la información pertinente a la pregunta, no que redacte ya una respuesta final, ya que eso corresponde a la fase reduce.
MAP_SYSTEM_PROMPT = """Eres un analista evaluando si el resumen de una comunidad de un grafo de conocimiento es relevante para responder una pregunta concreta. Se te dará la pregunta y el resumen de una comunidad. Debes devolver ÚNICAMENTE un JSON con esta forma: {{"relevancia": <número entero de 0 a 10>, "extracto": "<una o dos frases con la información de esta comunidad que sea útil para responder la pregunta, o cadena vacía si la relevancia es baja>"}}. No incluyas texto adicional fuera del JSON."""
MAP_USER_PROMPT = """Pregunta: {question}
Resumen de la comunidad:
{summary}
Miembros de la comunidad: {members}
Responde solo con el JSON."""Evaluamos una única comunidad frente a la pregunta, devolviendo su relevancia y un extracto de información útil. Esta es la unidad de trabajo que se repite una vez por comunidad en la fase map.
def map_community(chain, question: str, community: dict) -> dict:
response = chain.invoke({
"question": question,
"summary": community["summary"],
"members": ", ".join(community["members"]),
})
text = response.content.strip()
if text.startswith("```"):
text = text.split("```")[1]
if text.startswith("json"):
text = text[4:]
data = json.loads(text.strip())
return {
"community_id": community["community_id"],
"relevance": data.get("relevancia", 0),
"extract": data.get("extracto", ""),
}Este prompt es la fase reduce. Se ejecuta una sola vez con los extractos ya filtrados de las comunidades relevantes, y su trabajo es sintetizar todo eso en una respuesta final coherente y bien redactada, en vez de simplemente concatenar los extractos sueltos.
REDUCE_SYSTEM_PROMPT = """Eres un asistente que responde preguntas sobre Juego de Tronos a partir de varios extractos de información, cada uno proveniente de una comunidad distinta de un grafo de conocimiento. Sintetiza estos extractos en una única respuesta coherente y bien redactada en español, organizada por facción cuando tenga sentido. No inventes información que no esté en los extractos proporcionados."""
REDUCE_USER_PROMPT = """Pregunta: {question}
Extractos relevantes recopilados:
{extracts}
Redacta la respuesta final."""Combinamos los extractos de las comunidades relevantes en una única respuesta final, delegando la síntesis en el LLM en vez de simplemente concatenar texto.
def reduce_extracts(chain, question: str, mapped_communities: list[dict]) -> str:
extracts_text = "\n".join(
f"- (Comunidad {c['community_id']}, relevancia {c['relevance']}/10): {c['extract']}"
for c in mapped_communities
)
response = chain.invoke({"question": question, "extracts": extracts_text})
return response.contentY ya el orquestador final.
def main():
communities = load_communities()
print(f"Comunidades cargadas: {len(communities)}")
print(f"Pregunta: {QUESTION}\n")
llm = ChatDeepSeek(model="deepseek-chat", temperature=0)
map_prompt = ChatPromptTemplate.from_messages([
("system", MAP_SYSTEM_PROMPT),
("user", MAP_USER_PROMPT),
])
map_chain = map_prompt | llm
# Fase "map": evaluamos cada comunidad de forma independiente. Aquí lo hacemos de forma secuencial (un bucle simple) para mantener el script fácil de seguir; en un sistema en producción, estas llamadas serían buenas candidatas a paralelizarse, ya que ninguna depende del resultado de las demás.
mapped_communities = []
for community in communities:
result = map_community(map_chain, QUESTION, community)
mapped_communities.append(result)
print(f"Comunidad {result['community_id']}: relevancia {result['relevance']}/10")
# Filtramos por el umbral de relevancia antes de pasar a la fase "reduce", y ordenamos de mayor a menor relevancia para que la comunidad más pertinente aparezca primero en el contexto que le demos al LLM final.
relevant_communities = [c for c in mapped_communities if c["relevance"] >= RELEVANCE_THRESHOLD]
relevant_communities.sort(key=lambda c: c["relevance"], reverse=True)
print(f"\nComunidades relevantes (relevancia >= {RELEVANCE_THRESHOLD}): {len(relevant_communities)} de {len(communities)}")
reduce_prompt = ChatPromptTemplate.from_messages([
("system", REDUCE_SYSTEM_PROMPT),
("user", REDUCE_USER_PROMPT),
])
reduce_chain = reduce_prompt | llm
# Fase "reduce": una única llamada final que sintetiza los extractos ya filtrados en una respuesta coherente.
answer = reduce_extracts(reduce_chain, QUESTION, relevant_communities)
print("\n--- Respuesta final ---")
print(answer)
if __name__ == "__main__":
main()Y al final, si ejecutamos el script, nos debería quedar algo así.
Comunidades cargadas: 13
Pregunta: ¿Cuáles son las principales facciones en conflicto y qué las mueve?
Comunidad 0: relevancia 2/10
Comunidad 1: relevancia 0/10
Comunidad 2: relevancia 8/10
Comunidad 3: relevancia 8/10
Comunidad 4: relevancia 9/10
Comunidad 5: relevancia 2/10
Comunidad 6: relevancia 8/10
Comunidad 7: relevancia 1/10
Comunidad 8: relevancia 8/10
Comunidad 9: relevancia 8/10
Comunidad 10: relevancia 0/10
Comunidad 11: relevancia 8/10
Comunidad 12: relevancia 0/10
Comunidades relevantes (relevancia >= 5): 7 de 13
--- Respuesta final ---
Las principales facciones en conflicto en Juego de Tronos son diversas y están motivadas por el control del Trono de Hierro, el poder regional y la supervivencia frente a amenazas sobrenaturales. A continuación, se organizan por facción según los extractos proporcionados:
- **Casa Stark y sus aliados**: Liderados por la Casa Stark, junto con la Casa Tully y la Guardia de la Noche, buscan controlar el Norte y defender los reinos humanos. Se enfrentan a la Casa Frey y la Casa Bolton, motivadas por alianzas rotas y la lucha por el dominio septentrional, lo que culmina en la Boda Roja. Además, se oponen a los Caminantes Blancos y el Ejército de los Muertos, una amenaza existencial que trasciende las disputas políticas.
- **Casa Lannister y Casa Tyrell**: Esta alianza controla el Trono de Hierro desde Desembarco del Rey, motivada por mantener el poder central. Se enfrentan a la Fe Militante (una fuerza religiosa que desafía su autoridad) y a la Compañía Dorada (mercenarios que apoyan a otros reclamantes), así como a facciones rivales como los Baratheon y los Targaryen.
- **Casa Baratheon**: Reclaman el Trono de Hierro por su linaje real, representado por el venado coronado. Dentro de esta facción, Stannis Baratheon actúa como un reclamante solitario desde Dragonstone, enfrentado tanto a los Lannister como a su hermano Renly, con seguidores leales que lo convierten en una fuerza beligerante independiente.
- **Daenerys Targaryen y sus seguidores**: Esta facción, compuesta por exiliados, mercenarios y aventureros de Essos, incluye a los Dothraki y los Inmaculados como base militar. Su motivación principal es reclamar el Trono de Hierro, operando desde fuera de Poniente y representando una amenaza directa a las facciones establecidas.
- **Caminantes Blancos y el Ejército de los Muertos**: Aunque no son una facción política tradicional, representan una fuerza sobrenatural que amenaza a todas las demás, incluyendo a la alianza Stark-Tully, la Guardia de la Noche y los Targaryen, en el contexto de la Guerra de los Cinco Reyes.
En resumen, las facciones se mueven por el control territorial (Norte, Trono de Hierro), la legitimidad dinástica (Baratheon, Targaryen), la supervivencia frente a lo sobrenatural (Caminantes Blancos) y alianzas estratégicas o religiosas (Lannister-Tyrell, Fe Militante).Bueno, es un tutorial extenso y con algunos pasajes complicados, pero espero que haya quedado más claro cómo funciona una pipeline GraphRAG clásica. Otro día vemos distintas variantes, ya sí desde un punto de vista más teórico.