RAG con Neo4j y LLMs - Potenciando IA con Grafos de Conocimiento

¿Qué es RAG y por qué combinarlo con Neo4j?

RAG (Retrieval-Augmented Generation) es una técnica que mejora las respuestas de los modelos de lenguaje grandes (LLMs) al proporcionarles contexto relevante recuperado de una base de conocimiento externa. En lugar de depender únicamente del conocimiento contenido en los parámetros del modelo, RAG busca información específica antes de generar una respuesta.

Neo4j, como base de datos de grafos, es ideal para RAG porque:

✅ Contexto relacional: Captura relaciones complejas entre entidades
✅ Navegación eficiente: Permite recorrer relaciones rápidamente
✅ Conocimiento estructurado: Organiza información de forma semántica
✅ Búsqueda contextual: Encuentra información relevante considerando conexiones

Arquitectura de RAG con Neo4j

Usuario → Pregunta
    ↓
1. Embedding de la pregunta
    ↓
2. Búsqueda en Neo4j (Vector Search)
    ↓
3. Recuperación de contexto relacional
    ↓
4. Construcción del prompt con contexto
    ↓
5. LLM genera respuesta enriquecida
    ↓
Respuesta contextualizada → Usuario

Requisitos Previos

# Instalar dependencias necesarias
pip install neo4j langchain openai tiktoken

Necesitarás:

Neo4j 5.11+ (con Vector Index support)
API key de OpenAI (o cualquier LLM compatible)
Python 3.9+

Paso 1: Configurar Neo4j con Vectores

Primero, creamos un índice vectorial en Neo4j para almacenar embeddings:

// Crear índice vectorial para documentos
CREATE VECTOR INDEX document_embeddings IF NOT EXISTS
FOR (d:Document)
ON d.embedding
OPTIONS {
  indexConfig: {
    `vector.dimensions`: 1536,
    `vector.similarity_function`: 'cosine'
  }
}

Paso 2: Cargar Conocimiento en el Grafo

from neo4j import GraphDatabase
import openai

class KnowledgeGraphRAG:
    def __init__(self, uri, user, password, openai_key):
        self.driver = GraphDatabase.driver(uri, auth=(user, password))
        openai.api_key = openai_key

    def close(self):
        self.driver.close()

    def create_embedding(self, text):
        """Genera embedding usando OpenAI"""
        response = openai.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return response.data[0].embedding

    def add_document(self, title, content, metadata=None):
        """Agrega documento al grafo con su embedding"""
        embedding = self.create_embedding(content)

        with self.driver.session() as session:
            session.run("""
                CREATE (d:Document {
                    title: $title,
                    content: $content,
                    embedding: $embedding,
                    metadata: $metadata,
                    created_at: datetime()
                })
            """, title=title, content=content,
                embedding=embedding, metadata=metadata)

    def add_relationship(self, doc1_title, doc2_title, rel_type):
        """Crea relación entre documentos"""
        with self.driver.session() as session:
            session.run("""
                MATCH (d1:Document {title: $doc1})
                MATCH (d2:Document {title: $doc2})
                CREATE (d1)-[r:%s]->(d2)
            """ % rel_type, doc1=doc1_title, doc2=doc2_title)

Paso 3: Implementar Búsqueda Vectorial

def retrieve_context(self, query, top_k=3):
    """Recupera contexto relevante usando búsqueda vectorial"""
    query_embedding = self.create_embedding(query)

    with self.driver.session() as session:
        result = session.run("""
            CALL db.index.vector.queryNodes(
                'document_embeddings',
                $top_k,
                $query_embedding
            ) YIELD node, score

            // Obtener también documentos relacionados
            OPTIONAL MATCH (node)-[r:RELATES_TO|REFERENCES]-(related:Document)

            RETURN
                node.title as title,
                node.content as content,
                score,
                collect(DISTINCT {
                    title: related.title,
                    type: type(r)
                }) as related_docs
            ORDER BY score DESC
        """, query_embedding=query_embedding, top_k=top_k)

        return [dict(record) for record in result]

Paso 4: Generar Respuesta con LLM

def generate_response(self, query, context_docs):
    """Genera respuesta usando contexto recuperado"""
    # Construir contexto desde Neo4j
    context_text = "\n\n".join([
        f"Documento: {doc['title']}\n{doc['content']}\n"
        f"Documentos relacionados: {', '.join([r['title'] for r in doc['related_docs']])}"
        for doc in context_docs
    ])

    # Crear prompt aumentado
    prompt = f"""Basándote en el siguiente contexto, responde la pregunta del usuario.

Contexto:
{context_text}

Pregunta: {query}

Respuesta detallada:"""

    # Llamar a OpenAI
    response = openai.chat.completions.create(
        model="gpt-4",
        messages=[
            {"role": "system", "content": "Eres un asistente experto que responde basándose en el contexto proporcionado."},
            {"role": "user", "content": prompt}
        ],
        temperature=0.7,
        max_tokens=500
    )

    return response.choices[0].message.content

def ask(self, query):
    """Método principal de RAG"""
    # 1. Recuperar contexto
    context_docs = self.retrieve_context(query, top_k=3)

    # 2. Generar respuesta
    answer = self.generate_response(query, context_docs)

    # 3. Retornar con fuentes
    return {
        "answer": answer,
        "sources": [doc['title'] for doc in context_docs],
        "relevance_scores": [doc['score'] for doc in context_docs]
    }

Paso 5: Uso Completo del Sistema

# Inicializar
rag = KnowledgeGraphRAG(
    uri="bolt://localhost:7687",
    user="neo4j",
    password="password",
    openai_key="sk-..."
)

# Agregar conocimiento al grafo
rag.add_document(
    title="Introducción a Neo4j",
    content="Neo4j es una base de datos de grafos nativa que almacena datos en nodos y relaciones...",
    metadata={"category": "database", "level": "beginner"}
)

rag.add_document(
    title="Cypher Query Language",
    content="Cypher es el lenguaje de consulta declarativo de Neo4j, diseñado para trabajar con grafos...",
    metadata={"category": "query-language", "level": "intermediate"}
)

# Crear relaciones entre documentos
rag.add_relationship(
    "Introducción a Neo4j",
    "Cypher Query Language",
    "PREREQUISITE"
)

# Hacer preguntas
result = rag.ask("¿Cómo puedo consultar datos en Neo4j?")

print("Respuesta:", result['answer'])
print("\nFuentes:", result['sources'])
print("Relevancia:", result['relevance_scores'])

# Cerrar conexión
rag.close()

Ejemplo de Salida

Respuesta: Para consultar datos en Neo4j, utilizas Cypher Query Language,
el lenguaje declarativo diseñado específicamente para trabajar con grafos.
Cypher te permite hacer MATCH para encontrar patrones, WHERE para filtrar,
y RETURN para especificar qué datos quieres recuperar. Por ejemplo:

MATCH (n:Person)-[:KNOWS]->(friend)
WHERE n.name = 'Alice'
RETURN friend.name

Este query encuentra todos los amigos de Alice en el grafo.

Fuentes: ['Cypher Query Language', 'Introducción a Neo4j']
Relevancia: [0.89, 0.76]

Ventajas de RAG con Neo4j vs Búsqueda Vectorial Simple

Característica	Vectores Simples	Neo4j RAG
Contexto relacional	❌ No	✅ Sí
Navegación de conocimiento	❌ Limitada	✅ Completa
Explicabilidad	⚠️ Baja	✅ Alta
Actualización	⚠️ Compleja	✅ Simple
Razonamiento multi-hop	❌ No	✅ Sí

Casos de Uso Avanzados

1. RAG Multi-hop (Razonamiento en Múltiples Pasos)

def multi_hop_retrieve(self, query, depth=2):
    """Recupera contexto navegando el grafo en profundidad"""
    query_embedding = self.create_embedding(query)

    with self.driver.session() as session:
        result = session.run("""
            CALL db.index.vector.queryNodes(
                'document_embeddings',
                3,
                $query_embedding
            ) YIELD node, score

            // Navegar hasta 2 niveles de profundidad
            CALL apoc.path.subgraphAll(node, {
                maxLevel: $depth,
                relationshipFilter: 'RELATES_TO|REFERENCES|PREREQUISITE'
            })
            YIELD nodes, relationships

            RETURN node.title as main_doc,
                   [n IN nodes | n.title] as related_docs,
                   [r IN relationships | type(r)] as relationship_types,
                   score
        """, query_embedding=query_embedding, depth=depth)

        return [dict(record) for record in result]

2. RAG con Filtrado Temporal

def retrieve_recent_context(self, query, days=30):
    """Recupera solo contexto reciente"""
    query_embedding = self.create_embedding(query)

    with self.driver.session() as session:
        result = session.run("""
            CALL db.index.vector.queryNodes(
                'document_embeddings',
                5,
                $query_embedding
            ) YIELD node, score

            WHERE node.created_at > datetime() - duration({days: $days})

            RETURN node.title, node.content, score
            ORDER BY score DESC
        """, query_embedding=query_embedding, days=days)

        return [dict(record) for record in result]

3. RAG con Personalización de Usuario

def personalized_retrieve(self, query, user_id):
    """Recupera contexto personalizado según preferencias del usuario"""
    query_embedding = self.create_embedding(query)

    with self.driver.session() as session:
        result = session.run("""
            // Encontrar documentos relevantes
            CALL db.index.vector.queryNodes(
                'document_embeddings',
                10,
                $query_embedding
            ) YIELD node, score

            // Boost de documentos relacionados con el usuario
            OPTIONAL MATCH (u:User {id: $user_id})-[:INTERESTED_IN]->(t:Topic)
            OPTIONAL MATCH (node)-[:ABOUT]->(t)

            WITH node, score,
                 CASE WHEN t IS NOT NULL THEN 1.5 ELSE 1.0 END as user_boost

            RETURN node.title, node.content,
                   (score * user_boost) as adjusted_score
            ORDER BY adjusted_score DESC
            LIMIT 5
        """, query_embedding=query_embedding, user_id=user_id)

        return [dict(record) for record in result]

Optimización y Mejores Prácticas

1. Chunking Inteligente

def smart_chunk_document(self, long_text, chunk_size=500, overlap=50):
    """Divide documentos largos en chunks con overlap"""
    words = long_text.split()
    chunks = []

    for i in range(0, len(words), chunk_size - overlap):
        chunk = ' '.join(words[i:i + chunk_size])
        chunks.append(chunk)

    return chunks

def add_long_document(self, title, content):
    """Agrega documento largo en chunks relacionados"""
    chunks = self.smart_chunk_document(content)

    for i, chunk in enumerate(chunks):
        chunk_title = f"{title} - Part {i+1}"
        self.add_document(chunk_title, chunk)

        # Relacionar chunks consecutivos
        if i > 0:
            prev_chunk = f"{title} - Part {i}"
            self.add_relationship(prev_chunk, chunk_title, "NEXT_CHUNK")

2. Caché de Embeddings

from functools import lru_cache

@lru_cache(maxsize=1000)
def create_embedding_cached(self, text):
    """Versión con caché de create_embedding"""
    return self.create_embedding(text)

3. Monitoreo de Relevancia

def ask_with_confidence(self, query, min_score=0.7):
    """RAG con threshold de confianza"""
    context_docs = self.retrieve_context(query, top_k=3)

    # Verificar relevancia mínima
    if not context_docs or context_docs[0]['score'] < min_score:
        return {
            "answer": "No encontré información suficientemente relevante para responder con confianza.",
            "confidence": "low",
            "sources": []
        }

    answer = self.generate_response(query, context_docs)

    return {
        "answer": answer,
        "confidence": "high",
        "sources": [doc['title'] for doc in context_docs],
        "avg_score": sum(d['score'] for d in context_docs) / len(context_docs)
    }

Conclusión

RAG con Neo4j combina lo mejor de dos mundos: la capacidad de generación de LLMs y el conocimiento estructurado en grafos. Esta arquitectura permite:

🎯 Respuestas más precisas basadas en conocimiento específico
🔍 Trazabilidad completa de fuentes y razonamiento
🔄 Actualización dinámica del conocimiento sin reentrenar modelos
🌐 Contexto relacional que los vectores simples no pueden capturar
📊 Explicabilidad de por qué se generó cierta respuesta

El futuro de la IA conversacional está en combinar modelos de lenguaje con bases de conocimiento estructuradas, y Neo4j es la plataforma perfecta para construir estos sistemas.