NSF RAG v1.0

app.py

@@ -0,0 +1,334 @@
+"""
+app.py
+──────
+Interfaz web Gradio para el sistema RAG de corrección de castellano s.XVI.
+
+Arranque:
+    python app.py
+
+Requiere:
+    - .env con OPENAI_API_KEY
+    - (opcional) corpus en ./corpus/
+"""
+
+import os
+import json
+import gradio as gr
+from dotenv import load_dotenv
+
+from knowledge_base import SAMPLE_PAIRS
+from corpus_loader import CorpusLoader
+from vector_store import VectorStore
+from rag_corrector import RAGCorrector
+from evaluator import Evaluator
+
+load_dotenv()
+
+# ── Inicialización ────────────────────────────────────────────────────────────
+
+print("🚀 Inicializando Scriptorium RAG...")
+
+vs = VectorStore()
+
+# Cargar corpus desde disco (si existe) + pares de ejemplo embebidos
+loader = CorpusLoader(os.getenv("CORPUS_PATH", "./corpus"))
+disk_pairs = loader.load()
+all_pairs  = SAMPLE_PAIRS + disk_pairs
+
+# Indexar todo en ChromaDB
+vs.index(all_pairs)
+
+corrector = RAGCorrector(vs)
+evaluator = Evaluator()
+
+print(f"✅ Sistema listo. Documentos en vector store: {vs.count()}")
+
+# ── Ejemplos de demostración ──────────────────────────────────────────────────
+
+DEMO_EXAMPLES = [
+    "q̃ fizo merçed al dho lugar de las alcaualas del anno de mill e quinientos",
+    "el escriuano del cabildo faze fe y da testimouio verdadero de todo lo sobredho",
+    "en la muy noble çibdad de burgos a veynte dias del mes de março anno dho",
+    "yo juan de la torre vezino desta uilla de toledo otorgo e conosco",
+    "sepan quantos esta carta de poder vieren como yo pero lopez vezino dela villa",
+    "fizo pareçer ante si a los testigos q̃ dixeron ser mayores de veynte annos",
+]
+
+# ── Función principal ─────────────────────────────────────────────────────────
+
+def corregir(htr_text: str, top_k: int, mostrar_prompt: bool):
+    if not htr_text.strip():
+        return "", "", "", "", "⚠ Introduce un texto HTR para corregir."
+
+    if not os.getenv("OPENAI_API_KEY"):
+        return "", "", "", "", "❌ Falta OPENAI_API_KEY en el fichero .env"
+
+    try:
+        result = corrector.correct(htr_text, top_k=int(top_k))
+    except Exception as e:
+        return "", "", "", "", f"❌ Error al llamar a la API: {e}"
+
+    corrected  = result["corrected"]
+    retrieved  = result["retrieved"]
+    htr_errors = result["htr_errors"]
+    grafia_w   = result["grafia_warns"]
+
+    # ── Panel de documentos recuperados ──────────────────────────────────────
+    docs_md = f"### Top-{len(retrieved)} documentos recuperados\n\n"
+    for i, doc in enumerate(retrieved, 1):
+        docs_md += (
+            f"**{i}. [{doc['type']} · {doc['region']} · {doc['date']}]**  "
+            f"*similitud: {doc['score']}*\n\n"
+            f"- **HTR:** `{doc['htr']}`\n"
+            f"- **GT:**  `{doc['gt']}`\n"
+        )
+        if doc["corrections"]:
+            docs_md += f"- **Correcciones:** {', '.join(doc['corrections'])}\n"
+        docs_md += "\n---\n"
+
+    # ── Panel de análisis ─────────────────────────────────────────────────────
+    analysis_md = "### Análisis del texto\n\n"
+
+    if htr_errors:
+        analysis_md += "**⚠ Posibles errores HTR detectados:**\n"
+        for e in htr_errors:
+            analysis_md += f"- `{e['htr']}` → `{e['gt']}`: {e['context']}  \n  *Ej: {e['example']}*\n"
+        analysis_md += "\n"
+
+    if grafia_w:
+        analysis_md += "**✦ Alertas de grafía (NO modernizar):**\n"
+        for g in grafia_w:
+            analysis_md += f"- `{g['modern']}` → mantener `{g['ancient']}`: {g['rule']}\n"
+        analysis_md += "\n"
+
+    if not htr_errors and not grafia_w:
+        analysis_md += "*No se detectaron patrones conocidos de error en el texto.*\n"
+
+    # Diff visual (diferencias)
+    diff_md = "### Diferencias HTR → Corregido\n\n"
+    orig_words = htr_text.split()
+    corr_words = corrected.split()
+    diff_parts = []
+    max_len = max(len(orig_words), len(corr_words))
+    changed = 0
+    for i in range(max_len):
+        o = orig_words[i] if i < len(orig_words) else "—"
+        c = corr_words[i] if i < len(corr_words) else "—"
+        if o != c:
+            diff_parts.append(f"~~{o}~~ → **{c}**")
+            changed += 1
+        else:
+            diff_parts.append(c)
+    diff_md += " ".join(diff_parts)
+    diff_md += f"\n\n*{changed} palabra(s) modificada(s) de {len(orig_words)} totales.*"
+
+    # ── Prompt (opcional) ─────────────────────────────────────────���───────────
+    prompt_md = ""
+    if mostrar_prompt:
+        prompt_md = f"```\nSYSTEM:\n{result.get('_system', '(ver rag_corrector.py)')}\n\nUSER:\n{result['prompt']}\n```"
+
+    status = f"✅ Corrección completada con **{result['model']}** · {vs.count()} docs en índice"
+
+    return corrected, docs_md, analysis_md, diff_md, status
+
+
+def evaluar_par(htr_text: str, gt_text: str):
+    if not htr_text.strip() or not gt_text.strip():
+        return "⚠ Introduce tanto el texto HTR como el groundtruth."
+    try:
+        result = corrector.correct(htr_text)
+        metrics = evaluator.evaluate_pair(htr_text, result["corrected"], gt_text)
+        m = metrics
+        mod = m["modernism"]
+        report = (
+            f"### Métricas de evaluación\n\n"
+            f"| Métrica | Antes (HTR) | Después (RAG) | Mejora |\n"
+            f"|---------|------------|---------------|--------|\n"
+            f"| **CER** | {m['cer_before']:.2%} | {m['cer_after']:.2%} | {m['cer_improvement']:+.2%} |\n"
+            f"| **WER** | {m['wer_before']:.2%} | {m['wer_after']:.2%} | {m['wer_improvement']:+.2%} |\n\n"
+            f"**Detector de modernismos:** score={mod['score']:.2f}  "
+            f"({mod['count']} problema(s) detectado(s))\n"
+        )
+        if mod["issues"]:
+            report += "\nFormas modernas introducidas incorrectamente:\n"
+            for iss in mod["issues"]:
+                report += f"- `{iss['modern']}` (debería ser `{iss['ancient']}`): {iss['rule']}\n"
+
+        report += f"\n**Texto corregido por RAG:**\n> {result['corrected']}"
+        return report
+    except Exception as e:
+        return f"❌ Error: {e}"
+
+
+def add_to_corpus(htr_text: str, gt_text: str, doc_type: str, region: str, date: str):
+    if not htr_text.strip() or not gt_text.strip():
+        return "⚠ HTR y GT son obligatorios."
+    try:
+        pair_id = f"user_{abs(hash(htr_text)) % 100000:05d}"
+        new_pair = {
+            "id":      pair_id,
+            "htr":     htr_text.strip(),
+            "gt":      gt_text.strip(),
+            "type":    doc_type or "desconocido",
+            "region":  region or "desconocida",
+            "date":    date or "",
+            "corrections": [],
+            "source":  "user_added",
+        }
+        added = vs.index([new_pair])
+        if added:
+            return f"✅ Par añadido al corpus con id `{pair_id}`. Total: {vs.count()} docs."
+        else:
+            return f"ℹ Par ya existía en el corpus (id: `{pair_id}`)."
+    except Exception as e:
+        return f"❌ Error: {e}"
+
+
+# ── Interfaz Gradio ───────────────────────────────────────────────────────────
+
+with gr.Blocks(
+    title="Scriptorium RAG",
+    theme=gr.themes.Base(
+        primary_hue="amber",
+        secondary_hue="stone",
+        neutral_hue="stone",
+        font=gr.themes.GoogleFont("IM Fell English"),
+    ),
+    css="""
+    .header { text-align: center; padding: 20px 0 10px; }
+    .header h1 { font-size: 2.2em; color: #92400e; letter-spacing: 0.15em; }
+    .header p  { color: #78716c; font-style: italic; }
+    .status-bar { font-size: 0.85em; padding: 6px 12px; border-radius: 6px; }
+    """,
+) as demo:
+
+    # ── Header ────────────────────────────────────────────────────────────────
+    gr.HTML("""
+    <div class="header">
+      <h1>RAG CODEX for NSF project</h1>
+      <p>Sistema RAG de corrección de castellano del siglo XVI</p>
+    </div>
+    """)
+
+    with gr.Tabs():
+
+        # ── Pestaña 1: Corrección ─────────────────────────────────────────────
+        with gr.TabItem(" Corrección HTR"):
+            with gr.Row():
+                with gr.Column(scale=2):
+                    htr_input = gr.Textbox(
+                        label="Texto HTR (entrada del reconocedor)",
+                        placeholder="Pega aquí el texto resultado del HTR…",
+                        lines=6,
+                    )
+                    with gr.Row():
+                        top_k_slider = gr.Slider(
+                            minimum=1, maximum=10, value=5, step=1,
+                            label="Documentos recuperados (k)",
+                        )
+                        show_prompt = gr.Checkbox(label="Mostrar prompt RAG", value=False)
+                    btn_corregir = gr.Button("✦ Corregir con RAG", variant="primary")
+
+                    gr.Examples(
+                        examples=DEMO_EXAMPLES,
+                        inputs=htr_input,
+                        label="Ejemplos de demostración",
+                    )
+
+                with gr.Column(scale=2):
+                    corrected_out = gr.Textbox(
+                        label="Texto corregido (salida RAG)",
+                        lines=6,
+                        interactive=False,
+                    )
+                    status_out = gr.Markdown(elem_classes=["status-bar"])
+
+            with gr.Row():
+                with gr.Column():
+                    docs_out = gr.Markdown(label="Documentos recuperados del corpus")
+                with gr.Column():
+                    analysis_out = gr.Markdown(label="Análisis de patrones")
+
+            diff_out = gr.Markdown(label="Diferencias palabra a palabra")
+            prompt_out = gr.Markdown(label="Prompt enviado al LLM", visible=False)
+
+            btn_corregir.click(
+                fn=corregir,
+                inputs=[htr_input, top_k_slider, show_prompt],
+                outputs=[corrected_out, docs_out, analysis_out, diff_out, status_out],
+            )
+
+        # ── Pestaña 2: Evaluación ─────────────────────────────────────────────
+        with gr.TabItem(" Evaluación con GT"):
+            gr.Markdown("Compara la corrección RAG contra el groundtruth real para medir CER/WER y detectar modernismos.")
+            with gr.Row():
+                eval_htr = gr.Textbox(label="Texto HTR", lines=4)
+                eval_gt  = gr.Textbox(label="Groundtruth (referencia)", lines=4)
+            btn_eval = gr.Button("Evaluar", variant="primary")
+            eval_out = gr.Markdown()
+            btn_eval.click(fn=evaluar_par, inputs=[eval_htr, eval_gt], outputs=eval_out)
+
+        # ── Pestaña 3: Añadir al corpus ───────────────────────────────────────
+        with gr.TabItem("➕ Añadir al corpus"):
+            gr.Markdown("Añade nuevos pares al vector store para mejorar el RAG de forma continua.")
+            with gr.Row():
+                add_htr    = gr.Textbox(label="Texto HTR", lines=4)
+                add_gt     = gr.Textbox(label="Groundtruth corregido", lines=4)
+            with gr.Row():
+                add_type   = gr.Textbox(label="Tipo documental", placeholder="notarial / judicial / eclesiastico")
+                add_region = gr.Textbox(label="Región", placeholder="Castilla, Andalucía…")
+                add_date   = gr.Textbox(label="Fecha", placeholder="1542")
+            btn_add  = gr.Button("Añadir al corpus", variant="primary")
+            add_out  = gr.Markdown()
+            btn_add.click(
+                fn=add_to_corpus,
+                inputs=[add_htr, add_gt, add_type, add_region, add_date],
+                outputs=add_out,
+            )
+
+        # ── Pestaña 4: Info del sistema ───────────────────────────────────────
+        with gr.TabItem("ℹ Sistema"):
+            gr.Markdown(f"""
+## Estado del sistema
+
+- **Modelo LLM:** {os.getenv('OPENAI_MODEL', 'gpt-4o')}
+- **Vector store:** ChromaDB (persistente en `{os.getenv('CHROMA_PATH','./chroma_db')}`)
+- **Documentos indexados:** {vs.count()}
+- **Corpus cargado desde disco:** {len(disk_pairs)} pares
+- **Pares de ejemplo embebidos:** {len(SAMPLE_PAIRS)}
+
+## Arquitectura
+
+```
+Texto HTR
+   │
+   ├─► Detector de patrones HTR (knowledge_base.py)
+   ├─► Detector de grafías modernas (knowledge_base.py)
+   │
+   ├─► Embedding (text-embedding-3-small)
+   │        │
+   │        └─► Búsqueda top-k en ChromaDB ──► Few-shot dinámico
+   │
+   └─► Prompt constructor ──► GPT-4o ──► Texto corregido
+```
+
+## Formato del corpus
+
+Para añadir tu corpus, crea `./corpus/` con ficheros JSON:
+```json
+[
+  {{"id": "doc001", "htr": "texto htr...", "gt": "groundtruth...",
+    "type": "notarial", "region": "Castilla", "date": "1542"}},
+  ...
+]
+```
+O CSV con columnas: `id, htr, gt, type, region, date`
+""")
+
+if __name__ == "__main__":
+    demo.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=False,
+        show_error=True,
+    )

corpus_loader.py

@@ -0,0 +1,103 @@
+"""
+corpus_loader.py
+────────────────
+Carga pares (HTR, groundtruth) desde disco.
+
+Formatos soportados:
+  1. JSON  → lista de objetos {"id","htr","gt","type","region","date"}
+  2. CSV   → columnas: id, htr, gt, type, region, date
+  3. TXT   → carpeta con ficheros *.htr.txt  y  *.gt.txt  (mismo nombre base)
+
+Uso:
+    from corpus_loader import CorpusLoader
+    loader = CorpusLoader("./corpus")
+    pairs = loader.load()          # lista de dicts
+"""
+
+import json
+import csv
+import os
+from pathlib import Path
+from typing import List, Dict
+
+
+class CorpusLoader:
+    def __init__(self, corpus_path: str):
+        self.corpus_path = Path(corpus_path)
+
+    def load(self) -> List[Dict]:
+        """Detecta formato y carga todos los pares disponibles."""
+        pairs = []
+
+        if not self.corpus_path.exists():
+            print(f"⚠  Corpus path '{self.corpus_path}' no existe. Usando pares de ejemplo.")
+            return []
+
+        # ── JSON ──────────────────────────────────────────────────────────────
+        for f in sorted(self.corpus_path.glob("*.json")):
+            try:
+                with open(f, encoding="utf-8") as fh:
+                    data = json.load(fh)
+                if isinstance(data, list):
+                    pairs.extend(self._normalize(data, source=f.stem))
+                elif isinstance(data, dict):          # un solo documento
+                    pairs.append(self._normalize_one(data, source=f.stem))
+                print(f"✅ JSON cargado: {f.name}  ({len(data)} pares)")
+            except Exception as e:
+                print(f"❌ Error leyendo {f.name}: {e}")
+
+        # ── CSV ───────────────────────────────────────────────────────────────
+        for f in sorted(self.corpus_path.glob("*.csv")):
+            try:
+                with open(f, encoding="utf-8", newline="") as fh:
+                    reader = csv.DictReader(fh)
+                    rows = list(reader)
+                pairs.extend(self._normalize(rows, source=f.stem))
+                print(f"✅ CSV cargado:  {f.name}  ({len(rows)} pares)")
+            except Exception as e:
+                print(f"❌ Error leyendo {f.name}: {e}")
+
+        # ── TXT pareado ───────────────────────────────────────────────────────
+        htr_files = sorted(self.corpus_path.glob("*.htr.txt"))
+        for htr_file in htr_files:
+            gt_file = htr_file.with_suffix("").with_suffix(".gt.txt")
+            if not gt_file.exists():
+                print(f"⚠  Sin GT para {htr_file.name}, omitido.")
+                continue
+            try:
+                htr_text = htr_file.read_text(encoding="utf-8").strip()
+                gt_text  = gt_file.read_text(encoding="utf-8").strip()
+                pairs.append({
+                    "id":      htr_file.stem.replace(".htr", ""),
+                    "htr":     htr_text,
+                    "gt":      gt_text,
+                    "type":    "desconocido",
+                    "region":  "desconocida",
+                    "date":    "",
+                    "source":  "txt",
+                })
+            except Exception as e:
+                print(f"❌ Error leyendo {htr_file.name}: {e}")
+
+        if htr_files:
+            print(f"✅ TXT cargados: {len(htr_files)} pares")
+
+        print(f"\n📚 Total pares cargados desde disco: {len(pairs)}")
+        return pairs
+
+    # ── helpers ───────────────────────────────────────────────────────────────
+
+    def _normalize(self, rows: List[Dict], source: str) -> List[Dict]:
+        return [self._normalize_one(r, source) for r in rows if r.get("htr") and r.get("gt")]
+
+    def _normalize_one(self, row: Dict, source: str) -> Dict:
+        return {
+            "id":      str(row.get("id", source)),
+            "htr":     str(row.get("htr", "")).strip(),
+            "gt":      str(row.get("gt", "")).strip(),
+            "type":    str(row.get("type", "desconocido")),
+            "region":  str(row.get("region", "desconocida")),
+            "date":    str(row.get("date", "")),
+            "source":  source,
+            "corrections": row.get("corrections", []),
+        }

evaluator.py

@@ -0,0 +1,163 @@
+"""
+evaluator.py
+────────────
+Métricas de calidad para las correcciones del sistema RAG.
+
+  - CER  (Character Error Rate)   : nivel de carácter
+  - WER  (Word Error Rate)        : nivel de palabra
+  - Modernism score               : penalización por formas modernas introducidas
+  - Batch evaluation              : evalúa el sistema sobre un conjunto de pares con GT
+
+Uso:
+    from evaluator import Evaluator
+    ev = Evaluator()
+    metrics = ev.evaluate_pair(htr="...", corrected="...", gt="...")
+    report  = ev.batch_evaluate(corrector, pairs[:50])
+"""
+
+import re
+from typing import List, Dict, Tuple
+from knowledge_base import GRAFIA_PATTERNS
+
+
+class Evaluator:
+
+    # ── Métricas de edición ──────────────────────────────────────────────────
+
+    @staticmethod
+    def cer(reference: str, hypothesis: str) -> float:
+        """Character Error Rate (Levenshtein a nivel carácter)."""
+        r, h = list(reference), list(hypothesis)
+        return Evaluator._levenshtein(r, h) / max(len(r), 1)
+
+    @staticmethod
+    def wer(reference: str, hypothesis: str) -> float:
+        """Word Error Rate (Levenshtein a nivel palabra)."""
+        r = reference.split()
+        h = hypothesis.split()
+        return Evaluator._levenshtein(r, h) / max(len(r), 1)
+
+    @staticmethod
+    def _levenshtein(seq1: list, seq2: list) -> int:
+        m, n = len(seq1), len(seq2)
+        dp = list(range(n + 1))
+        for i in range(1, m + 1):
+            prev = dp[:]
+            dp[0] = i
+            for j in range(1, n + 1):
+                if seq1[i - 1] == seq2[j - 1]:
+                    dp[j] = prev[j - 1]
+                else:
+                    dp[j] = 1 + min(prev[j], dp[j - 1], prev[j - 1])
+        return dp[n]
+
+    # ── Detector de modernismos ──────────────────────────────────────────────
+
+    @staticmethod
+    def modernism_penalty(original_htr: str, corrected: str) -> Dict:
+        """
+        Detecta formas modernas introducidas por el LLM que no estaban
+        en el HTR original. Retorna lista de problemas detectados.
+        """
+        issues = []
+        orig_lower = original_htr.lower()
+        corr_lower = corrected.lower()
+
+        for p in GRAFIA_PATTERNS:
+            modern = p["modern"].lower()
+            ancient_forms = [f.strip().lower() for f in p["ancient"].split("/")]
+
+            # Si el corrected contiene la forma moderna Y el original no la tenía
+            if modern in corr_lower and modern not in orig_lower:
+                # Verificar que tampoco era una forma antigua válida
+                if not any(af in orig_lower for af in ancient_forms):
+                    issues.append({
+                        "modern":  p["modern"],
+                        "ancient": p["ancient"],
+                        "rule":    p["rule"],
+                    })
+
+        return {
+            "count":  len(issues),
+            "issues": issues,
+            "score":  max(0.0, 1.0 - len(issues) * 0.1),  # 0.0–1.0
+        }
+
+    # ── Evaluación de un par ─────────────────────────────────────────────────
+
+    def evaluate_pair(
+        self, htr: str, corrected: str, gt: str
+    ) -> Dict:
+        """
+        Evalúa una sola corrección comparando con el groundtruth.
+        """
+        cer_htr  = self.cer(gt, htr)        # CER antes de corregir
+        cer_corr = self.cer(gt, corrected)   # CER después de corregir
+        wer_htr  = self.wer(gt, htr)
+        wer_corr = self.wer(gt, corrected)
+
+        modernism = self.modernism_penalty(htr, corrected)
+
+        return {
+            "cer_before":       round(cer_htr, 4),
+            "cer_after":        round(cer_corr, 4),
+            "cer_improvement":  round(cer_htr - cer_corr, 4),
+            "wer_before":       round(wer_htr, 4),
+            "wer_after":        round(wer_corr, 4),
+            "wer_improvement":  round(wer_htr - wer_corr, 4),
+            "modernism":        modernism,
+        }
+
+    # ── Evaluación en batch ──────────────────────────────────────────────────
+
+    def batch_evaluate(
+        self, corrector, pairs: List[Dict], verbose: bool = True
+    ) -> Dict:
+        """
+        Evalúa el sistema sobre una lista de pares con groundtruth.
+        Retorna métricas agregadas + detalle por par.
+        """
+        results = []
+        for i, pair in enumerate(pairs):
+            if verbose:
+                print(f"  Evaluando {i+1}/{len(pairs)}: {pair['id']}")
+            try:
+                out = corrector.correct(pair["htr"])
+                metrics = self.evaluate_pair(
+                    htr=pair["htr"],
+                    corrected=out["corrected"],
+                    gt=pair["gt"],
+                )
+                metrics["id"] = pair["id"]
+                metrics["htr"] = pair["htr"]
+                metrics["corrected"] = out["corrected"]
+                metrics["gt"] = pair["gt"]
+                results.append(metrics)
+            except Exception as e:
+                print(f"  ❌ Error en {pair['id']}: {e}")
+
+        if not results:
+            return {"error": "Sin resultados"}
+
+        avg = lambda key: round(sum(r[key] for r in results) / len(results), 4)
+
+        summary = {
+            "n_evaluated":          len(results),
+            "avg_cer_before":       avg("cer_before"),
+            "avg_cer_after":        avg("cer_after"),
+            "avg_cer_improvement":  avg("cer_improvement"),
+            "avg_wer_before":       avg("wer_before"),
+            "avg_wer_after":        avg("wer_after"),
+            "avg_wer_improvement":  avg("wer_improvement"),
+            "avg_modernism_score":  avg("modernism"),   # via nested
+            "detail":               results,
+        }
+
+        if verbose:
+            print(f"\n📊 RESUMEN EVALUACIÓN ({len(results)} pares)")
+            print(f"   CER: {summary['avg_cer_before']:.2%} → {summary['avg_cer_after']:.2%}  "
+                  f"(mejora: {summary['avg_cer_improvement']:+.2%})")
+            print(f"   WER: {summary['avg_wer_before']:.2%} → {summary['avg_wer_after']:.2%}  "
+                  f"(mejora: {summary['avg_wer_improvement']:+.2%})")
+
+        return summary

knowledge_base.py

@@ -0,0 +1,124 @@
+"""
+knowledge_base.py
+─────────────────
+Patrones de error HTR y grafías del castellano s.XVI.
+Edita estas listas para ampliar la base de conocimiento estática.
+"""
+
+# ── Confusiones visuales habituales en HTR sobre letra procesal / gótica ──────
+HTR_ERROR_PATTERNS = [
+    {"htr": "u",  "gt": "n",  "context": "confusión visual u/n en minúscula gótica",         "example": "uuestro → nuestro"},
+    {"htr": "n",  "gt": "u",  "context": "confusión visual n/u en procesal",                  "example": "sno → suo (raro)"},
+    {"htr": "c",  "gt": "e",  "context": "confusión c/e en cursiva procesal",                 "example": "merçcd → merçed"},
+    {"htr": "rn", "gt": "m",  "context": "rn interpretado como m por el modelo",              "example": "cornprar → comprar"},
+    {"htr": "ii", "gt": "u",  "context": "doble i confundida con u",                          "example": "diios → duos"},
+    {"htr": "cl", "gt": "d",  "context": "cl confundido con d en procesal",                   "example": "clado → dado"},
+    {"htr": "f",  "gt": "s",  "context": "s larga (ſ) leída como f por el modelo",            "example": "feñor → señor"},
+    {"htr": "1",  "gt": "l",  "context": "1 numérico confundido con l minúscula",              "example": "1ugar → lugar"},
+    {"htr": "i",  "gt": "j",  "context": "i/j sin distinción gráfica en s.XVI",               "example": "iusticia → justicia"},
+    {"htr": "ó",  "gt": "a",  "context": "a cerrada interpretada como o",                     "example": "cóso → caso"},
+    {"htr": "ll", "gt": "h",  "context": "ll confundida con h en algunas manos",               "example": "llonra → honra"},
+    {"htr": "vn", "gt": "un", "context": "v usada como u consonántica al inicio de sílaba",   "example": "vno → uno"},
+    {"htr": "e",  "gt": "c",  "context": "c abierta leída como e",                            "example": "etro → otro"},
+    {"htr": "o",  "gt": "a",  "context": "a redondeada confundida con o",                     "example": "los → las"},
+]
+
+# ── Grafías propias del castellano s.XVI que NO deben modernizarse ────────────
+GRAFIA_PATTERNS = [
+    {"modern": "hizo",      "ancient": "fizo",          "rule": "f- inicial latina ante diptongo ie/ue"},
+    {"modern": "hijo",      "ancient": "fijo",          "rule": "f- inicial conservada"},
+    {"modern": "hacer",     "ancient": "fazer / hazer", "rule": "variación f/h en infinitivos"},
+    {"modern": "dicho",     "ancient": "dho / dicho",   "rule": "abreviatura notarial dho"},
+    {"modern": "merced",    "ancient": "merçed",        "rule": "cedilla ante e/i"},
+    {"modern": "que",       "ancient": "q̃ / que",       "rule": "abreviatura con tilde volada"},
+    {"modern": "porque",    "ancient": "porq̃ / porque", "rule": "abreviatura frecuente"},
+    {"modern": "vecino",    "ancient": "vezino",        "rule": "z/c ante vocal anterior"},
+    {"modern": "precio",    "ancient": "presçio",       "rule": "variación s/c+cedilla"},
+    {"modern": "ciudad",    "ancient": "çibdad / cibdad","rule": "forma medieval con b/v"},
+    {"modern": "mucho",     "ancient": "muncho",        "rule": "nasalización muncho/mucho"},
+    {"modern": "mismo",     "ancient": "mesmo",         "rule": "mesmo forma habitual s.XVI"},
+    {"modern": "también",   "ancient": "tanbién",       "rule": "asimilación nasal"},
+    {"modern": "escribano", "ancient": "escriuano",     "rule": "v/u gráfica"},
+    {"modern": "nuestro",   "ancient": "nuestro / nro", "rule": "abreviatura nro en notarial"},
+    {"modern": "señor",     "ancient": "señor / sr",    "rule": "abreviatura sr"},
+    {"modern": "dicho",     "ancient": "dcho / dho",    "rule": "doble abreviatura notarial"},
+    {"modern": "tienen",    "ancient": "tienen / tienẽ","rule": "tilde abreviativa nasal final"},
+    {"modern": "lugar",     "ancient": "lugar / lug̃r",  "rule": "abreviatura con tilde"},
+    {"modern": "presente",  "ancient": "prezente / presente", "rule": "alternancia s/z"},
+]
+
+# ── Documentos de ejemplo embebidos (se amplían con el corpus real) ───────────
+SAMPLE_PAIRS = [
+    {
+        "id": "sample_001",
+        "type": "notarial",
+        "region": "Castilla",
+        "date": "1542",
+        "htr": "q̃ fizo merçed al dho lugar de las alcaualas",
+        "gt":  "que fizo merçed al dicho lugar de las alcaualas",
+        "corrections": ["dho → dicho (abreviatura expandida)"],
+    },
+    {
+        "id": "sample_002",
+        "type": "notarial",
+        "region": "Andalucía",
+        "date": "1561",
+        "htr": "el escriuano del cabildo faze fe y da testimouio verdadero",
+        "gt":  "el escriuano del cabildo faze fe y da testimonio verdadero",
+        "corrections": ["testimouio → testimonio (u/n confusión HTR)"],
+    },
+    {
+        "id": "sample_003",
+        "type": "judicial",
+        "region": "Castilla",
+        "date": "1534",
+        "htr": "en la muy noble çibdad de burgos a veynte dias del mes de março",
+        "gt":  "en la muy noble çibdad de burgos a veynte días del mes de março",
+        "corrections": ["días: acento restituido"],
+    },
+    {
+        "id": "sample_004",
+        "type": "eclesiastico",
+        "region": "Toledo",
+        "date": "1578",
+        "htr": "el vezino dela villa de alcala prezento ante nos vna peticion",
+        "gt":  "el vezino de la villa de alcalá presentó ante nos una petición",
+        "corrections": ["dela → de la", "alcala → alcalá", "prezento → presentó", "vna → una"],
+    },
+    {
+        "id": "sample_005",
+        "type": "notarial",
+        "region": "Extremadura",
+        "date": "1549",
+        "htr": "yo el sobredho escriuano doy fe q̃ conosco al otorgante",
+        "gt":  "yo el sobredicho escriuano doy fe que conosco al otorgante",
+        "corrections": ["sobredho → sobredicho", "q̃ → que"],
+    },
+    {
+        "id": "sample_006",
+        "type": "judicial",
+        "region": "Castilla",
+        "date": "1556",
+        "htr": "fizo pareçer ante si a juan de la torre vezino desta villa",
+        "gt":  "fizo pareçer ante sí a juan de la torre vezino desta villa",
+        "corrections": ["si → sí (acento diacrítico)"],
+    },
+    {
+        "id": "sample_007",
+        "type": "notarial",
+        "region": "Aragón",
+        "date": "1523",
+        "htr": "sepan quantos esta carta de poder vieren como yo pero lopez",
+        "gt":  "sepan quantos esta carta de poder vieren como yo pero lópez",
+        "corrections": ["lopez → lópez (acento)"],
+    },
+    {
+        "id": "sample_008",
+        "type": "eclesiastico",
+        "region": "Sevilla",
+        "date": "1587",
+        "htr": "en el nonbre de dios e de la virgen maria su madre amen",
+        "gt":  "en el nonbre de dios e de la virgen maría su madre amén",
+        "corrections": ["maría, amén: acentos restituidos"],
+    },
+]

rag_corrector.py

@@ -0,0 +1,171 @@
+"""
+rag_corrector.py
+────────────────
+Núcleo del sistema RAG.
+
+  1. Detecta posibles errores HTR y grafías modernas en el texto de entrada.
+  2. Recupera ejemplos similares del vector store (few-shot dinámico).
+  3. Construye el prompt con reglas, ejemplos y alertas.
+  4. Llama a GPT-4o y devuelve el texto corregido + trazabilidad.
+
+Uso:
+    from rag_corrector import RAGCorrector
+    corrector = RAGCorrector(vector_store)
+    result = corrector.correct("texto htr aqui")
+    print(result["corrected"])
+    print(result["prompt"])          # para depuración
+"""
+
+import os
+from typing import List, Dict, Tuple
+from openai import OpenAI
+from dotenv import load_dotenv
+
+from knowledge_base import HTR_ERROR_PATTERNS, GRAFIA_PATTERNS
+
+load_dotenv()
+
+MODEL   = os.getenv("OPENAI_MODEL", "gpt-4o")
+TOP_K   = int(os.getenv("TOP_K", 5))
+
+SYSTEM_PROMPT = """Eres un corrector especializado en documentos notariales y judiciales \
+españoles del siglo XVI (castellano antiguo).
+
+Tu ÚNICA tarea es corregir los errores introducidos por el proceso automático de \
+reconocimiento de texto manuscrito (HTR). NO debes modernizar el texto bajo ninguna \
+circunstancia.
+
+REGLAS ABSOLUTAS — incumplirlas invalida la corrección:
+1. Conserva SIEMPRE las grafías propias del s.XVI:
+   fizo, fazer, hazer, merçed, vezino, mesmo, çibdad, escriuano,
+   dho (=dicho), q̃ (=que), nro (=nuestro), vn/vna, etc.
+2. NO conviertas f→h inicial latina (fizo ≠ hizo, fazer ≠ hacer).
+3. Conserva abreviaturas y tildes voladas (q̃, nro, dho, sr).
+4. Corrige SOLO lo que claramente sea un error HTR (grafema confundido visualmente).
+5. Si no estás seguro de si algo es error HTR o forma s.XVI válida → conserva el original.
+6. Responde ÚNICAMENTE con el texto corregido. Sin explicaciones, sin comillas."""
+
+
+class RAGCorrector:
+    def __init__(self, vector_store):
+        self.vs = vector_store
+        #self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL", "https://api.x.ai/v1"),)
+    # ── API pública ──────────────────────────────────────────────────────────
+
+    def correct(self, htr_text: str, top_k: int = TOP_K) -> Dict:
+        """
+        Corrige un texto HTR usando RAG.
+
+        Retorna dict con:
+          corrected     : str   — texto corregido
+          prompt        : str   — prompt completo enviado al LLM
+          retrieved     : list  — documentos recuperados del vector store
+          htr_errors    : list  — patrones HTR detectados
+          grafia_warns  : list  — grafías modernas detectadas (alertas)
+          model         : str   — modelo usado
+        """
+        retrieved    = self.vs.retrieve(htr_text, k=top_k)
+        htr_errors   = self._detect_htr_errors(htr_text)
+        grafia_warns = self._detect_grafias(htr_text)
+
+        prompt = self._build_prompt(htr_text, retrieved, htr_errors, grafia_warns)
+
+        corrected = self._call_llm(prompt)
+
+        return {
+            "corrected":    corrected,
+            "prompt":       prompt,
+            "retrieved":    retrieved,
+            "htr_errors":   htr_errors,
+            "grafia_warns": grafia_warns,
+            "model":        MODEL,
+        }
+
+    # ── Detección de patrones ────────────────────────────────────────────────
+
+    def _detect_htr_errors(self, text: str) -> List[Dict]:
+        found = []
+        for p in HTR_ERROR_PATTERNS:
+            if p["htr"] in text:
+                found.append(p)
+        return found
+
+    def _detect_grafias(self, text: str) -> List[Dict]:
+        """Detecta formas modernas que NO deberían modernizarse."""
+        found = []
+        lower = text.lower()
+        for p in GRAFIA_PATTERNS:
+            if p["modern"].lower() in lower:
+                found.append(p)
+        return found
+
+    # ── Constructor de prompt ────────────────────────────────────────────────
+
+    def _build_prompt(
+        self,
+        htr_text: str,
+        retrieved: List[Dict],
+        htr_errors: List[Dict],
+        grafia_warns: List[Dict],
+    ) -> str:
+
+        sections = []
+
+        # Few-shot dinámico: ejemplos recuperados
+        if retrieved:
+            examples = []
+            for i, doc in enumerate(retrieved, 1):
+                corr = "; ".join(doc["corrections"]) if doc["corrections"] else "—"
+                examples.append(
+                    f"Ejemplo {i} [{doc['type']}, {doc['region']}, {doc['date']}]"
+                    f" (similitud={doc['score']}):\n"
+                    f"  HTR:  \"{doc['htr']}\"\n"
+                    f"  GT:   \"{doc['gt']}\"\n"
+                    f"  Correcciones aplicadas: {corr}"
+                )
+            sections.append(
+                "EJEMPLOS DEL CORPUS (similares al texto a corregir):\n"
+                + "\n\n".join(examples)
+            )
+
+        # Alertas de patrones HTR detectados
+        if htr_errors:
+            hints = "\n".join(
+                f"  • '{p['htr']}' puede ser '{p['gt']}': {p['context']} (ej: {p['example']})"
+                for p in htr_errors
+            )
+            sections.append(f"POSIBLES ERRORES HTR DETECTADOS EN ESTE TEXTO:\n{hints}")
+
+        # Alertas de grafías modernas
+        if grafia_warns:
+            warns = "\n".join(
+                f"  • '{p['modern']}' → mantener como '{p['ancient']}': {p['rule']}"
+                for p in grafia_warns
+            )
+            sections.append(
+                f"ALERTA — GRAFÍAS QUE NO DEBEN MODERNIZARSE:\n{warns}"
+            )
+
+        context_block = "\n\n".join(sections)
+
+        return (
+            f"{context_block}\n\n"
+            f"TEXTO HTR A CORREGIR:\n\"{htr_text}\""
+            if context_block
+            else f"TEXTO HTR A CORREGIR:\n\"{htr_text}\""
+        )
+
+    # ── Llamada al LLM ───────────────────────────────────────────────────────
+
+    def _call_llm(self, user_prompt: str) -> str:
+        response = self.client.chat.completions.create(
+            model=MODEL,
+            temperature=0.1,        # baja temperatura: reproducibilidad
+            max_tokens=1024,
+            messages=[
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "user",   "content": user_prompt},
+            ],
+        )
+        return response.choices[0].message.content.strip().strip('"').strip("'")

vector_store.py

@@ -0,0 +1,153 @@
+"""
+vector_store.py
+───────────────
+Indexa pares HTR/GT en ChromaDB usando embeddings de OpenAI.
+Recupera los top-k documentos más similares a un texto de consulta.
+
+Uso:
+    from vector_store import VectorStore
+    vs = VectorStore()
+    vs.index(pairs)                        # indexar corpus
+    results = vs.retrieve("texto htr...", k=5)
+"""
+
+import json
+import os
+from typing import List, Dict
+from tqdm import tqdm
+
+import chromadb
+from chromadb.utils import embedding_functions
+from dotenv import load_dotenv
+
+load_dotenv()
+
+CHROMA_PATH  = os.getenv("CHROMA_PATH", "./chroma_db")
+OPENAI_KEY   = os.getenv("OPENAI_API_KEY", "")
+EMBED_MODEL  = "text-embedding-3-small"
+COLLECTION   = "scriptorium_corpus"
+
+
+class VectorStore:
+    def __init__(self):
+        self.client = chromadb.PersistentClient(path=CHROMA_PATH)
+
+        # Función de embedding: OpenAI si hay clave, sino local (sentencetransformers)
+        #if OPENAI_KEY:
+        #    self.ef = embedding_functions.OpenAIEmbeddingFunction(
+        #        api_key=OPENAI_KEY,
+        #        model_name=EMBED_MODEL,
+        #    )
+        #else:
+        #    # Fallback: modelo local multilingüe
+        #    self.ef = embedding_functions.SentenceTransformerEmbeddingFunction(
+        #        model_name="intfloat/multilingual-e5-small"
+        #    )
+
+        self.ef = embedding_functions.SentenceTransformerEmbeddingFunction(
+            model_name="intfloat/multilingual-e5-small")
+
+        self.collection = self.client.get_or_create_collection(
+            name=COLLECTION,
+            embedding_function=self.ef,
+            metadata={"hnsw:space": "cosine"},
+        )
+
+    # ── Indexación ────────────────────────────────────────────────────────────
+
+    def index(self, pairs: List[Dict], batch_size: int = 50) -> int:
+        """
+        Indexa los pares HTR/GT. Cada fragmento se almacena con:
+          - document : texto que se embebe  (htr + ' [SEP] ' + gt)
+          - metadata : tipo, región, fecha, htr, gt originales
+          - id       : identificador único del par
+        Retorna el número de documentos nuevos añadidos.
+        """
+        existing_ids = set(self.collection.get(include=[])["ids"])
+        to_add = [p for p in pairs if p["id"] not in existing_ids]
+
+        if not to_add:
+            print(f"ℹ  Vector store ya actualizado ({len(existing_ids)} documentos).")
+            return 0
+
+        print(f"🔄 Indexando {len(to_add)} documentos nuevos...")
+
+        for i in tqdm(range(0, len(to_add), batch_size), desc="Indexando"):
+            batch = to_add[i : i + batch_size]
+
+            # El texto que se embebe combina HTR + GT para capturar
+            # tanto los errores como las formas correctas
+            documents = [
+                f"HTR: {p['htr']} [SEP] GT: {p['gt']}"
+                for p in batch
+            ]
+            metadatas = [
+                {
+                    "htr":     p["htr"],
+                    "gt":      p["gt"],
+                    "type":    p.get("type", ""),
+                    "region":  p.get("region", ""),
+                    "date":    p.get("date", ""),
+                    "corrections": json.dumps(
+                        p.get("corrections", []), ensure_ascii=False
+                    ),
+                }
+                for p in batch
+            ]
+            ids = [p["id"] for p in batch]
+
+            self.collection.add(
+                documents=documents,
+                metadatas=metadatas,
+                ids=ids,
+            )
+
+        print(f"✅ Indexación completa. Total en store: {self.collection.count()}")
+        return len(to_add)
+
+    # ── Recuperación ─────────────────────────────────────────────────────────
+
+    def retrieve(self, query: str, k: int = 5) -> List[Dict]:
+        """
+        Recupera los k pares más similares al texto HTR de consulta.
+        Retorna lista de dicts con htr, gt, type, region, date, corrections, score.
+        """
+        if self.collection.count() == 0:
+            return []
+
+        results = self.collection.query(
+            query_texts=[query],
+            n_results=min(k, self.collection.count()),
+            include=["metadatas", "distances"],
+        )
+
+        retrieved = []
+        for meta, dist in zip(
+            results["metadatas"][0], results["distances"][0]
+        ):
+            retrieved.append({
+                "htr":         meta["htr"],
+                "gt":          meta["gt"],
+                "type":        meta.get("type", ""),
+                "region":      meta.get("region", ""),
+                "date":        meta.get("date", ""),
+                "corrections": json.loads(meta.get("corrections", "[]")),
+                "score":       round(1 - dist, 4),   # cosine similarity
+            })
+
+        return retrieved
+
+    # ── Utilidades ────────────────────────────────────────────────────────────
+
+    def count(self) -> int:
+        return self.collection.count()
+
+    def reset(self):
+        """Elimina y recrea la colección (útil para re-indexar desde cero)."""
+        self.client.delete_collection(COLLECTION)
+        self.collection = self.client.get_or_create_collection(
+            name=COLLECTION,
+            embedding_function=self.ef,
+            metadata={"hnsw:space": "cosine"},
+        )
+        print("🗑  Vector store reseteado.")