"""
=========================================================================
=======
12_b_cargar_pdf_automerging.py - INGESTA PADRE-HIJO (Auto-merging)
=========================================================================
=======
TIPO DE RAG: HIERARCHICAL CHUNKING (Padre-Hijo / Small-to-Big)
EL PROBLEMA:
- Chunks pequeños = búsqueda precisa PERO contexto incompleto
- Chunks grandes = contexto rico PERO búsqueda imprecisa
LA SOLUCIÓN (2 niveles):
[PADRE 2000 chars] ← Lo que recibe el LLM (contexto completo)
|
+-----+-----+
| | |
[H1] [H2] [H3] ← Lo que buscamos (400 chars, precisión)
1. Vectorizamos HIJOS (búsqueda precisa)
2. Guardamos PADRE en metadatos del hijo
3. Al encontrar hijo → entregamos padre al LLM
-------------------------------------------------------------------------
-------
RAPTOR (Versión Avanzada - N niveles):
-------------------------------------------------------------------------
-------
[Resumen Global] ← Nivel 3
|
+-----------+-----------+
| |
[Resumen A] [Resumen B] ← Nivel 2
| |
+-----+-----+ +-----+-----+
[C1] [C2] [C3] [C4] [C5] [C6] ← Nivel 1
Proceso: Chunks → Clustering → LLM resume → Repetir
¿Por qué NO lo usamos?
- Muchas llamadas LLM (costoso). Para 10 PDFs = 50+ llamadas.
- Padre-Hijo (2 niveles) cubre el 90% de casos reales.
¿Cuándo SÍ vale la pena? Libros 500+ páginas, miles de documentos.
=========================================================================
=======
"""
import os
import glob # Para buscar archivos PDF
import sys
import logging
import json
import chromadb
from [Link] import Settings
from sentence_transformers import SentenceTransformer
from openai import OpenAI
from dotenv import load_dotenv
import utils # Incluye leer_pdf_markdown()
# Añadir src al path
[Link]([Link]([Link](__file__)))
import utils
# Configuración
load_dotenv()
utils.setup_logging()
logger = [Link]("ingesta_automerging")
# --- CONFIGURACIÓN DE CHUNKING ---
TAMANO_PADRE = 2000 # Contexto grande para el LLM
TAMANO_HIJO = 400 # Contexto pequeño para buscar (Vector)
SOLAPE_HIJO = 50
# --- CONFIGURACIÓN DB ---
# Configuración de Rutas usando utils
DATA_DIR = utils.data_dir() / "documentos" / "pdf"
[Link](f"Ruta de datos: {DATA_DIR}")
DB_DIR = str(utils.project_root() / "chroma_db")
[Link](f"Ruta Base de Datos: {DB_DIR}")
# ¡OJO! Usamos una colección distinta para no mezclar con la normal
COLLECTION_NAME = "coleccion_automerging"
MODELO_EMBEDDINGS = [Link]("MODELO_EMBEDDINGS", "Qwen/Qwen3-Embedding-
0.6B")
# Archivo JSON con metadatos manuales
ruta_json = utils.data_dir() / "metadatos_pdfs.json"
# Configuración LLM para clasificación automática
LLM_API_KEY = [Link]("LLM_API_KEY", "ollama")
LLM_BASE_URL = [Link]("LLM_BASE_URL", "[Link]
MODELO_CLASIFICADOR = [Link]("MODELO_FAST", "llama3.2")
# LISTA MAESTRA DE CATEGORÍAS (Taxonomía)
CATEGORIAS_VALIDAS = [
"Salud Mental",
"Respiratorias",
"Enfermedades Crónicas"
]
def clasificar_documento(nombre_archivo: str, texto_inicio: str,
client_llm) -> str:
"""
Clasifica un documento usando LLM.
Igual que en 12_cargar_pdf.py para mantener consistencia.
"""
system_prompt = f"""Eres un clasificador de documentos médicos.
Categorías válidas: {', '.join(CATEGORIAS_VALIDAS)}
REGLAS:
- Responde SOLO con el nombre exacto de la categoría.
- Si no encaja en ninguna, responde "General".
- NO expliques tu razonamiento."""
user_prompt = f"""Clasifica este documento:
ARCHIVO: {nombre_archivo}
TEXTO:
"{texto_inicio[:1000]}"
CATEGORIA:"""
try:
resp = client_llm.[Link](
model=MODELO_CLASIFICADOR,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.0
)
categoria_raw = [Link][0].[Link]()
for cat in CATEGORIAS_VALIDAS:
if [Link]() in categoria_raw.lower():
return cat
[Link](f" LLM respondió '{categoria_raw}' (no válida) →
Usando 'General'")
return "General"
except Exception as e:
[Link](f"Fallo al clasificar {nombre_archivo}: {e}")
return "General"
def cargar_texto_pdf(ruta_pdf):
"""
Extrae texto del PDF en formato Markdown.
Usa pymupdf4llm que preserva mejor tablas e imágenes.
"""
return utils.leer_pdf_markdown(ruta_pdf)
def crear_chunks_jerarquicos(texto_completo):
"""
Crea Pares [Hijo, Texto_Padre]. Lógica "Small-to-Big".
1. Primero cortamos el texto en bloques GRANDES (Padres). Estos
tienen el sentido completo.
2. Luego, cada bloque grande lo volvemos a picar en trozos PEQUEÑOS
(Hijos).
¿Por qué?
- Los HIJOS son vectores: Muy específicos, fáciles de encontrar por
similitud.
- Los PADRES son contexto: Lo que realmente necesita leer el LLM.
Al final, guardamos el HIJO en la base de datos, pero le metemos una
nota
en su mochila (metadatos) que contiene el texto del PADRE entero.
"""
chunks_procesados = []
# 1. Cortar en PADRES (Chunks grandes)
# Ejemplo: Páginas enteras o secciones de 2000 caracteres.
chunks_padre = utils.hacer_chunking(texto_completo,
chunk_size=TAMANO_PADRE, overlap=200)
[Link](f"Generated {len(chunks_padre)} Parent Chunks (Big
context).")
for i, texto_padre in enumerate(chunks_padre):
# 2. Cortar cada PADRE en HIJOS (Chunks pequeños)
# Ejemplo: Párrafos de 400 caracteres.
chunks_hijo = utils.hacer_chunking(texto_padre,
chunk_size=TAMANO_HIJO, overlap=SOLAPE_HIJO)
for texto_hijo in chunks_hijo:
# MAGIA: El hijo lleva el texto del padre en su ADNi
(metadata)
chunks_procesados.append({
"texto_vectorizable": texto_hijo, # <--- LO QUE VEMOS
(Indexamos esto)
"texto_completo_padre": texto_padre, # <--- LO QUE
ESCONDEMOS (Contexto Expandido)
"padre_id": i
})
return chunks_procesados
def main():
[Link](" Iniciando Ingesta AVANZADA (Auto-merging)...")
patron = [Link](DATA_DIR, "*.pdf")
#Lista de rutas a todos los PDFs
archivos = [Link](patron)
if not archivos:
[Link]("No se encontraron archivos PDF en la carpeta.")
return
# 1. Setup Chroma
client = [Link](path=DB_DIR)
# Borramos la colección si existe para empezar limpio
try: client.delete_collection(COLLECTION_NAME)
except: pass
collection = client.get_or_create_collection(name=COLLECTION_NAME)
[Link](f"Colección '{COLLECTION_NAME}' creada.")
# 2. Setup Modelo Embeddings
[Link](f"Cargando modelo: {MODELO_EMBEDDINGS}")
model = SentenceTransformer(MODELO_EMBEDDINGS)
# 3. Setup Cliente LLM (para clasificación automática)
client_llm = OpenAI(base_url=LLM_BASE_URL, api_key=LLM_API_KEY)
# 4. Cargar metadatos manuales (JSON) - Prioridad sobre LLM
meta_manual = {}
if [Link](ruta_json):
try:
with open(ruta_json, 'r', encoding='utf-8') as f:
lista_meta = [Link](f)
meta_manual = {item['archivo']: item for item in
lista_meta}
[Link](f"Cargados {len(meta_manual)} metadatos manuales
desde JSON")
except Exception as e:
[Link](f"Error cargando JSON: {e}")
else:
[Link](f"No se encontró {ruta_json} - Se usará solo LLM")
total_chunks = 0
for archivo in archivos:
nombre_archivo = [Link](archivo)
[Link](f"Procesando: {nombre_archivo}...")
texto = cargar_texto_pdf(archivo)
# --- CLASIFICACIÓN HÍBRIDA: JSON > LLM ---
if nombre_archivo in meta_manual:
# Caso A: Metadatos manuales (prioridad)
datos = meta_manual[nombre_archivo]
categoria = [Link]('categoria', 'General')
subcategoria = [Link]('subcategoria', 'General')
[Link](f" [MANUAL] Categoría: {categoria}")
else:
# Caso B: LLM clasifica automáticamente
categoria = clasificar_documento(nombre_archivo, texto,
client_llm)
subcategoria = "General"
[Link](f" [LLM] Categoría: {categoria}")
# --- ESTRATEGIA PADRE-HIJO ---
items_jerarquicos = crear_chunks_jerarquicos(texto)
# Preparar batch para Chroma
textos_hijos = [item["texto_vectorizable"] for item in
items_jerarquicos]
metadatas = []
ids = []
for idx, item in enumerate(items_jerarquicos):
[Link]({
"source": nombre_archivo,
"category": categoria,
"subcategory": subcategoria,
"type": "child",
"parent_id": item["padre_id"],
"contexto_expandido": item["texto_completo_padre"]
})
[Link](f"{nombre_archivo}_child_{idx}")
# Generar Embeddings (Solo de los HIJOS)
embeddings = utils.generar_embeddings(model, textos_hijos)
# Guardar en DB
[Link](
documents=textos_hijos,
embeddings=embeddings,
metadatas=metadatas,
ids=ids
)
total_chunks += len(ids)
[Link](f" -> Insertados {len(ids)} chunks hijos (con sus
padres ocultos).")
[Link]("\n" + "="*50)
[Link](f" FINALIZADO. Total indexado: {total_chunks} vectores.")
[Link](f"Colección especial: {COLLECTION_NAME}")
[Link]("="*50)
if __name__ == "__main__":
main()
"""
=========================================================================
=======
APÉNDICE: ¿CÓMO SE HACE ESTO EN LA INDUSTRIA? (LANGCHAIN)
=========================================================================
=======
1. ¿QUÉ ES LANGCHAIN?
Es el "Framework Estándar" para construir apps con LLMs.
Imagínalo como el "jQuery" o "React" de la IA.
- Ventaja: Tiene conectores para TODO (PDFs, Excel, Notion, SQL...).
- Ventaja: Resuelve problemas complejos (como este del Auto-merging)
con una línea.
- Desventaja: Cambia muy rápido, es una "Caja Negra" y a veces es
difícil de depurar.
2. ¿QUÉ SE USA REALMENTE EN LA INDUSTRIA?
- Prototipos y Startups: LangChain es el rey (por velocidad de
desarrollo).
- Producción Crítica (Big Tech, Banca): A veces prefieren código
"Artesanal"
(como el de este script) o LlamaIndex, para tener control absoluto y
estabilidad.
LangChain es genial, pero abstracciones mágicas pueden fallar en
casos raros.
3. EJEMPLO: CÓMO SERÍA ESTE MISMO SCRIPT USANDO LANGCHAIN
(Implementación de ParentDocumentRetriever)
Observa que es mucho más corto, pero "no ves" dónde se guardan los
datos.
"""
# -----------------------------------------------------------------------
-------
# EJEMPLO DE CÓDIGO (CONCEPTUAL)
# -----------------------------------------------------------------------
-------
# from [Link] import ParentDocumentRetriever
# from [Link] import InMemoryStore # En prod usaríamos
RedisStore
# from langchain_chroma import Chroma
# from langchain_text_splitters import RecursiveCharacterTextSplitter
# from langchain_community.document_loaders import PyPDFLoader
#
# # 1. Definir los dos cortadores (Padre e Hijo)
# parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
# child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
#
# # 2. Definir la base de datos (Chroma) para los HIJOS (Búsqueda
Vectorial)
# vectorstore = Chroma(collection_name="split_parents",
embedding_function=embedding)
#
# # 3. Definir el almacén de documentos (Memoria/Redis) para los PADRES
(Texto Real)
# store = InMemoryStore()
#
# # 4. La Magia de LangChain (El Retriever lo conecta todo)
# retriever = ParentDocumentRetriever(
# vectorstore=vectorstore,
# docstore=store,
# child_splitter=child_splitter,
# parent_splitter=parent_splitter,
# )
#
# # 5. Ingesta Automática
# loader = PyPDFLoader("[Link]")
# docs = [Link]()
# retriever.add_documents(docs) # <-- AQUÍ OCURRE EL TROCEADO Y
VINCULACIÓN
# -----------------------------------------------------------------------
-------