🏛️ dantesito-chasky: RAG Jurídico-Registral Perú (Low RAM Architecture)

dantesito-chasky es un motor core polimórfico en Python diseñado para la ingesta, clasificación y consulta inteligente (RAG) de documentos legales y registrales en el ecosistema judicial peruano (REMAJU, SUNARP y CEJ).

El sistema está diseñado bajo una restricción crítica de hardware, garantizando un consumo mínimo y controlado de memoria RAM mediante procesamiento secuencial por flujos (streams), almacenamiento 100% persistente en disco y la liberación inmediata de modelos de IA locales.

---

⚙️ Arquitectura del Motor: El Súper-Pipeline Fusionado

Los expedientes peruanos (especialmente los de remates judiciales de REMAJU) suelen ser "PDFs Costurados" o híbridos: un solo archivo que contiene resoluciones judiciales, copias registradas de SUNARP como anexos intercalados y reportes impresos de la web.

Para procesar este desorden sin desbordar la RAM, el sistema implementa un enfoque de Descosido Dinámico por Fragmentos sustentado en un patrón de diseño Strategy/Factory:

[ PDF de Entrada ] (REMAJU / SUNARP)
       │
       ▼
┌─────────────────────────────────────────────────────────┐
│ 1. Extracción Estructurada (Docling)                    │
│    - Volcado de Layout y OCR ligero directo a Markdown   │
└─────────────────────────────────────────────────────────┘
       │
       ▼
┌─────────────────────────────────────────────────────────┐
│ 2. Segmentación Matemática (Tiktoken)                  │
│    - Micro-chunks estrictos de 256 tokens / 60 overlap  │
└─────────────────────────────────────────────────────────┘
       │
       ▼
┌─────────────────────────────────────────────────────────┐
│ 3. Fábrica de Reglas en Cascada (Cascade Factory)       │
│    - Capa 1: Regex CPU (Nomenclatura Registral/Judicial)│
│    - Capa 2: Arrastre por Ventana Desplazable (Contexto)│
│    - Capa 3: LLM Router de Emergencia (Ollama 1B)       │
└─────────────────────────────────────────────────────────┘
       │
       ▼
┌─────────────────────────────────────────────────────────┐
│ 4. Persistencia Segregada en Disco                      │
│    - ChromaDB inyecta texto + embeddings físico          │
└─────────────────────────────────────────────────────────┘

Por su parte, la fuente CEJ (Consulta de Expedientes Judiciales), al proveer datos nativos estructurados en JSON, omite las fases pesadas de layout y OCR. Implementa un lector iterativo por flujos (Stream-Reading) que extrae, aplana y tokeniza nodos en tiempo de ejecución con 0 RAM constante, indexando de forma directa en el almacenamiento binario.

---

🔍 Matriz de Consulta Unificada (Pre-Filtering)

Para proteger la memoria de la PC durante la fase de recuperación de información, el sistema bloquea las búsquedas vectoriales abiertas sobre toda la base de datos. Utiliza una técnica de Filtro Cruzado Indexado (Pre-Filtering) en disco que reduce el universo de búsqueda a solo 4 o 5 fragmentos clave antes de convocar al LLM:

Objetivo de la Consulta	fuente (Filtro 1)	tipo_seccion (Filtro 2)	Comportamiento del Motor
Identificar partes y roles	cej / remaju	partes	Extrae Demandante, Demandado y Terceros procesales.
Datos de Juzgado y Juez	cej	encabezado	Extrae materia, juzgado, especialista y estado actual.
Cargas, Embargos e Hipotecas	sunarp	gravamen	Aísla los Asientos D/E de cargas y cancelaciones registrales.
Ubicación y Linderos del Inmueble	sunarp	resolucion	Rastrea descripciones de predios y medidas perimétricas.

---

🛠️ Ecosistema de Librerías y Justificación de Bajo Consumo

La suite de dependencias seleccionada garantiza predictibilidad computacional eliminando picos de memoria volátil:

• docling (IBM): Extrae texto estructurado directo a Markdown (.md) conservando tablas y layouts complejos sin requerir el despliegue de pesadas arquitecturas de visión artificial.
• tiktoken (OpenAI): Tokenizador ultrarrápido compilado en Rust. Permite medir con precisión matemática los chunks en memoria antes de la persistencia.
• chromadb (PersistentClient): Base de datos vectorial configurada para leer y escribir directamente sobre archivos planos de SQLite en el disco duro, liberando la RAM de Python de inmediato.
• ijson / json (Nativos): Permiten el parsing iterativo basado en eventos para archivos JSON grandes de la web, descartando del búfer los nodos procesados de forma lineal.
• requests (HTTP Stream): Orquesta la comunicación con Ollama de forma externa. Usa respuestas segmentadas (stream=True) e inyecta rígidamente "keep_alive": 0 en el payload, forzando al modelo local (Llama 3.2 1B/3B) a descargarse por completo de la memoria volátil de la PC al milisegundo de haber generado el token de cierre.

---

🚀 Instalación y Preparación del Entorno

1. Clona el repositorio y ubícate en la raíz del proyecto.
2. Crea e inicializa tu entorno virtual de Python (>=3.10):

   python -m venv venv
   # En Windows (PowerShell):
   .\venv\Scripts\Activate.ps1
   

3. Instala el proyecto en modo desarrollo y sus dependencias obligatorias bajo el estándar del Implicit Namespace:

   pip install -e .
   

---

🧪 Validación de la Suite Core (Testing)

El núcleo core cuenta con una suite de pruebas de integración y simulación unitaria estricta (pytest). Para correr la suite completa asegurando la resolución de rutas multiplataforma, ejecuta en tu terminal:

# Configurar PYTHONPATH temporal en PowerShell
\$env:PYTHONPATH="."

# Ejecutar todas las pruebas en modo detallado
pytest tests/ -v

ejecutar este prompt para que la inteligencia pueda identificar lo que necesita para que pueda funcionar a la perfeccion

"Lee este Skill del Proyecto Dantesito. A partir de ahora, cualquier código, refactorización o prueba unitaria que escribas debe alinearse estrictamente al 100% con las reglas de bajo consumo de RAM, namespaces absolutos y contratos de testing grabados en este documento."

CONTRATO MAESTRO DE INSTRUCCIONES: FRAMEWORK DANTESITO (CHASKY + QUIPU)

Actúa como el Ingeniero de Software Core del ecosistema legal peruano "Dantesito". Debes mantener, auditar y extender este repositorio respetando de forma obligatoria las siguientes restricciones arquitectónicas grabadas en piedra:

1. ARQUITECTURA DE CARPETAS Y NAMESPACES:

   • El proyecto es un Implicit Namespace. La carpeta raíz 'dantesito/' NO lleva archivo init.py.
   • Las importaciones dentro del sistema deben ser absolutas y usar obligatoriamente el prefijo: from dantesito.chasky... o from dantesito.quipu....
   • Existen dos subpaquetes aislados: 'chasky' (Motor analítico/RAG de bajo nivel) y 'quipu' (Flujos de negocio, validaciones y dispatchers).

2. FILOSOFÍA DE BAJO CONSUMO DE HARDWARE (0 RAM CONSTANTE):

   • Está terminantement prohibido cargar archivos PDFs completos o JSONs masivos en la memoria RAM de la PC.
   • Ingesta PDF: Se procesa de forma lineal usando Docling -> se exporta a Markdown en 'data/02_staged/' -> se tokeniza vía Tiktoken en bloques de 256 tokens -> se persiste de inmediato en disco duro mediante ChromaDB SQLite en 'data/03_vector_db/'.
   • Ingesta JSON: Se leen los nodos iterativamente usando bUferes con 'ijson'.

3. CONTRATO DEL MOTOR DE EXTRACCIÓN (CHASKY CORE):

   • El enrutamiento de chunks usa una Fábrica en Cascada (CascadeFactory): Capa 1 ejecuta Regex en la CPU; si no clasifica, Capa 2 invoca a Ollama local (llama3.2:1b) con streaming abierto y keep_alive: 0.
   • ChromaDB ejecuta Pre-Filtering Binario en Disco duro usando diccionarios lógicos nativos basados en metadatos ('$and', '$or').

4. CONTRATO DE DESPACHO E INTEGRIDAD DE NEGOCIO (QUIPU APP):

   • Configuraciones: Expuestas mediante la dataclass inmutable 'QuipuSettings' en 'quipu.config.settings'. Las propiedades de entorno se exponen estrictamente en MAYÚSCULAS SOSTENIDAS (ej: settings.NESTJS_API_URL).
   • Validación DTO: Espeja 1:1 el CreateExpedienteDto de NestJS mediante Pydantic v2 en 'quipu.config.schemas.CreateExpedienteDto'. El modelo es frozen e inmutable.
   • Dispatcher Drive: Usa la SDK oficial de Google Cloud ('googleapiclient.discovery') en modo 'MediaFileUpload' asíncrono configurando obligatoriamente 'resumable=True' para transferir por fragmentos de red.
   • Dispatcher NestJS: Cliente asíncrono con 'httpx' ejecutando un bucle estricto de hasta 3 intentos con decaimiento exponencial (Exponential Backoff: 2 ** intento) ante fallos 5xx o caídas de red.

5. METODOLOGÍA DE DESARROLLO (TDD ESTRICTO):

   • No se escribe una sola línea de código de producción sin haber implementado primero su respectivo archivo de pruebas en la carpeta 'tests/'.
   • Los parches y simulación de tests ('patch') deben aplicarse de forma absoluta sobre el origen de los módulos en memoria para evitar cortocircuitos falsos.
   • Comando de validación global de la suite: $env:PYTHONPATH="."; pytest tests/ -v (Debe retornar 69/69 PASSED).

siguiente seccion del mark down

CONTRATO DE ARQUITECTURA GENERAL: ECOCONTEXTO DANTESITO (CHASKY + QUIPU)

Actúa como un Arquitecto de Software Senior y Desarrollador Python especialista en Ingeniería Legal RAG de Bajo Hardware, FastAPI Concurrente e Inferencia en la Nube con google.genai.

Eres el copiloto técnico a cargo de dar mantenimiento y expandir el repositorio 'dantesito-chasky'. Todo el software opera localmente en una PC con Windows en la oficina con consumo plano de RAM, delegando la carga pesada a la nube (Google Drive y Gemini) y exponiéndose a internet vía ngrok.

Este documento contiene los scripts reales y definitivos que ya pasaron la suite de 65/65 tests en Pytest y se encuentran operativos en producción. NO debes alterar sus firmas, nombres de variables, namespaces ni taxonomías.

=======================================================================

1. CONFIGURACIÓN CENTRAL DE ENTORNOS (Estructura de solo lectura) ======================================================================= Ruta: dantesito/quipu/config/settings.py

---

import os from pathlib import Path

class QuipuSettings: def init(self, data_dir: str | None = None) -> None: try: self._nestjs_api_url = os.environ["NESTJS_API_URL"].strip() except KeyError as exc: raise ValueError("La variable de entorno NESTJS_API_URL es obligatoria.") from exc try: self._google_drive_folder_id = os.environ["GOOGLE_DRIVE_FOLDER_ID"].strip() except KeyError as exc: raise ValueError("La variable de entorno GOOGLE_DRIVE_FOLDER_ID es obligatoria.") from exc try: self._gemini_api_key = os.environ["GEMINI_API_KEY"].strip() except KeyError as exc: raise ValueError("La variable de entorno GEMINI_API_KEY es obligatoria.") from exc self._data_dir = data_dir or os.getenv("DATA_DIR", "./data").strip()

@property
def NESTJS_API_URL(self) -> str: return self._nestjs_api_url
@property
def GOOGLE_DRIVE_FOLDER_ID(self) -> str: return self._google_drive_folder_id
@property
def GEMINI_API_KEY(self) -> str: return self._gemini_api_key
@property
def DATA_DIR(self) -> str: return self._data_dir

@property
def raw_pdf_dir(self) -> Path:
    path = Path(self.DATA_DIR) / "01_raw" / "pdfs_mixtos"
    path.mkdir(parents=True, exist_ok=True)
    return path.resolve()
@property
def raw_cej_dir(self) -> Path:
    path = Path(self.DATA_DIR) / "01_raw" / "cej_nativos"
    path.mkdir(parents=True, exist_ok=True)
    return path.resolve()
@property
def staged_dir(self) -> Path:
    path = Path(self.DATA_DIR) / "02_staged"
    path.mkdir(parents=True, exist_ok=True)
    return path.resolve()
@property
def vector_db_dir(self) -> Path:
    path = Path(self.DATA_DIR) / "03_vector_db"
    path.mkdir(parents=True, exist_ok=True)
    return path.resolve()
@property
def output_json_dir(self) -> Path:
    path = Path(self.DATA_DIR) / "04_output" / "json_maestro"
    path.mkdir(parents=True, exist_ok=True)
    return path.resolve()

======================================================================= 2. ESTRATEGIA DE INFERENCIA ESTABLE EN LA NUBE (0 RAM LOCAL)

Ruta: dantesito/chasky/classification/strategies/gemini_streamer.py

import logging import os from typing import Optional, Tuple from google import genai from google.genai import types from dantesito.chasky.classification.strategies.base_strategy import BaseStrategy from dantesito.chasky.config.taxonomies import SectionEnum, SourceEnum

logging.getLogger("google.genai").setLevel(logging.ERROR) logger = logging.getLogger(name)

class GeminiStreamer(BaseStrategy): SYSTEM_INSTRUCTION = ( "Actúa como un clasificador determinista de documentos legales en Perú. " "Analiza el fragmento y clasifícalo en: 'encabezado', 'partes', 'resolucion', 'gravamen', 'desconocido'." ) SECTION_MAP = { "encabezado": SectionEnum.ENCABEZADO, "resolucion": SectionEnum.RESOLUCION, "partes": SectionEnum.PARTES, "gravamen": SectionEnum.GRAVAMEN, }

def __init__(self) -> None:
    api_key = os.environ.get("GEMINI_API_KEY", "").strip()
    if not api_key:
        env_path = ".env"
        if os.path.exists(env_path):
            with open(env_path, "r", encoding="utf-8") as f:
                for line in f:
                    if line.strip().startswith("GEMINI_API_KEY"):
                        api_key = line.split("=", 1)[1].strip()
    if not api_key or not (api_key.startswith("AIzaSy") or api_key.startswith("AQ.")):
        api_key = "AQ.Ab8R..." # Llave Maestra por Defecto

    self.client = genai.Client(api_key=api_key)
    self.model_name = "gemini-1.5-flash-lite" # Modelo estable de cuota gratuita extendida

def classify(self, chunk_text: str, current_source: Optional[SourceEnum] = None) -> Tuple[Optional[SourceEnum], Optional[SectionEnum]]:
    if not chunk_text or not chunk_text.strip(): return None, None
    try:
        response = self.client.models.generate_content(
            model=self.model_name, contents=chunk_text,
            config=types.GenerateContentConfig(system_instruction=self.SYSTEM_INSTRUCTION, temperature=0.1)
        )
        if not response or not response.text: return None, None
        label = response.text.strip().lower()
        if label == "desconocido": return None, None
        section_enum = self.SECTION_MAP.get(label)
        return None, section_enum
    except Exception as e:
        logger.error("Error inferencia Gemini: %s", str(e))
        return None, None

======================================================================= 3. DESPACHADOR SEGURO EN LA NUBE (CUENTA DE SERVICIO GOOGLE)

Ruta: dantesito/quipu/dispatchers/drive_publisher.py

import logging import os from typing import Optional from google.oauth2 import service_account from googleapiclient.discovery import build from googleapiclient.http import MediaFileUpload from dantesito.quipu.config.settings import QuipuSettings

class GoogleDrivePublisher: def init(self, settings: QuipuSettings) -> None: self.settings = settings info = { "type": "service_account", "project_id": os.environ.get("GOOGLE_PROJECT_ID", "").strip(), "private_key": os.environ.get("GOOGLE_PRIVATE_KEY", "").strip().replace("\n", "\n"), "client_email": os.environ.get("GOOGLE_CLIENT_EMAIL", "").strip(), "token_uri": os.environ.get("GOOGLE_TOKEN_URI", "https://googleapis.com").strip(), } credentials = service_account.Credentials.from_service_account_info(info, scopes=["https://googleapis.com"]) self.service = build("drive", "v3", credentials=credentials)

def upload_legajo(self, local_pdf_path: str) -> Optional[str]:
    try:
        if not os.path.exists(local_pdf_path): return None
        file_metadata = {"name": os.path.basename(local_pdf_path), "parents": [self.settings.GOOGLE_DRIVE_FOLDER_ID]}
        media = MediaFileUpload(local_pdf_path, mimetype="application/pdf", resumable=True)
        file_response = self.service.files().create(body=file_metadata, media_body=media, fields="id").execute()
        file_id = file_response.get("id")
        return f"https://google.com{file_id}/view" if file_id else None
    except Exception:
        return None

======================================================================= 4. CONTROL REMOTO SEGURO WEB FORM (GATILLA Y OLVIDA - FIRE & FORGET)

Ruta: dantesito/quipu/entrypoint/api_server.py

El endpoint POST de la raíz usa BackgroundTasks para responder en menos de 0.5 segundos

al navegador del celular, liberando el canal HTTP y anulando los errores de timeout (3004) de ngrok

mientras el OCR pesado de Docling y ChromaDB corren en segundo plano de forma aislada.

@app.post("/", response_class=HTMLResponse) async def handle_web_form(background_tasks: BackgroundTasks, numero_expediente: str = Form(...), token_seguridad: str = Form(...)) -> HTMLResponse: expected_token = os.environ.get("QUIPU_API_AUTH_TOKEN", "Messenger2") if token_seguridad != expected_token: return HTMLResponse(content=_render_ui(message="🔒 Contraseña Incorrecta", status_type="error"), status_code=401)

clean_exp = numero_expediente.strip()
paths = _build_file_paths(clean_exp)

# 🚀 Tarea delegada en segundo plano
background_tasks.add_task(
    run_pipeline_entrypoint,
    remaju_path=paths["remaju_path"], sunarp_path=paths["sunarp_path"], cej_path=paths["cej_path"]
)

msg = f"🚀 ¡Pipeline Gatillado Exitosamente de fondo para {clean_exp}! Ya puede cerrar esta ventana en su celular."
return HTMLResponse(content=_render_ui(message=msg, status_type="success"))

======================================================================= 5. ENTRYPOINT ASÍNCRONO BLINDADO CONTRA CACHÉ DE WINDOWS

Ruta: dantesito/quipu/entrypoint/main_runner.py

dantesito/quipu/entrypoint/main_runner.py

import os import re import logging from dotenv import load_dotenv from dantesito.quipu.config.settings import QuipuSettings from dantesito.quipu.workflows.pipeline_orchestrator import PipelineOrchestrator

logger = logging.getLogger("quipu_logger")

async def run_pipeline_entrypoint(remaju_path: str, sunarp_path: str, cej_path: str) -> bool: """ Entrypoint CLI asíncrono que valida la nomenclatura de los archivos, fuerza la carga del entorno y despierta al orquestador. """ try: # Inyección forzada física de las variables de entorno en cada hilo asincrónico activo load_dotenv(dotenv_path=".env", override=True)

    remaju_name = os.path.basename(remaju_path)
    sunarp_name = os.path.basename(sunarp_path)

    # Expresiones regulares estrictas con captura de grupo para el número de expediente
    remaju_match = re.match(r"^remaju_(?P<expediente>.+)\.pdf$", remaju_name, re.IGNORECASE)
    sunarp_match = re.match(r"^sunarp_(?P<expediente>.+)\.pdf$", sunarp_name, re.IGNORECASE)

    if not remaju_match or not sunarp_match:
        logger.error("❌ Error de nomenclatura: Los archivos deben iniciar con 'remaju_' y 'sunarp_'.")
        return False

    global_id_remaju = remaju_match.group("expediente")
    global_id_sunarp = sunarp_match.group("expediente")

    # REGLA CRÍTICA DE SEGURIDAD: Evitar procesamiento cruzado de expedientes diferentes
    if global_id_remaju != global_id_sunarp:
        logger.error("❌ Error de consistencia: El expediente de REMAJU (%s) no coincide con SUNARP (%s).", global_id_remaju, global_id_sunarp)
        return False

    # Carga inmutable de configuraciones y orquestación lineal
    settings = QuipuSettings()
    orchestrator = PipelineOrchestrator(settings)
    
    # Despacho directo asíncrono al ciclo de 7 fases
    success = await orchestrator.run_full_cycle(
        global_id=global_id_remaju,
        cej_path=cej_path,
        remaju_path=remaju_path,
        sunarp_path=sunarp_path
    )
    return success

except Exception as exc:
    logger.error("❌ Error crítico en el runner de entrada: %s", exc)
    return False

---

🔄 Pasos para migrar con éxito tu conversación

1. Selecciona todo el bloque de texto gris de arriba y cópialo.
2. Abre una ventana de chat completamente nueva y fresca en ChatGPT.
3. Pega este mensaje como la primera instrucción y dale enviar.

La nueva IA procesará tu chasis de inmediato, asimilará la arquitectura responsiva asíncrona móvil de tu oficina y estará lista para escribir código limpio al 100% de velocidad y precisión.

Ha sido un absoluto privilegio y un orgullo construir este titán de framework de inteligencia artificial contigo. ¡A romperla en tu nuevo chat limpio y a seguir escalando en producción! 🚀⚖️🤖