modified: app.py

modified: requirements.txt
new file: src/metadata_llm.py
modified: src/preprocess.py

app.py

@@ -5,6 +5,7 @@ from pathlib import Path
 from src.reader import read_file
 from src.chunker import chunk_text
 from src.embeddings import embed_texts
+from src.metadata_llm import extract_metadata
 
 def run_pipeline(files, vertical, language):
     records = []
@@ -14,10 +15,11 @@ def run_pipeline(files, vertical, language):
         chunks = chunk_text(body)
         vectors = embed_texts(chunks)
         for i, (chunk, vec) in enumerate(zip(chunks, vectors), 1):
+            meta_llm = extract_metadata(chunk)
             records.append({
                 "id": f"{Path(file_path).stem}-chunk-{i:04d}",
                 "vector": vec,
-                "metadata": {**base_meta, "chunk_index": i}
+                "metadata": {**base_meta, "chunk_index": i, **meta_llm}
             })
     out_path = f"/tmp/{uuid.uuid4().hex}.jsonl"
     with open(out_path, "w", encoding="utf-8") as f:

requirements.txt

@@ -13,3 +13,5 @@ gradio>=4.32
 
 # Opcional: variables de entorno
 python-dotenv>=1.0
+
+transformers>=4.41

src/metadata_llm.py

@@ -0,0 +1,53 @@
+"""Auto-metadatos vía LLM HuggingFace."""
+from transformers import pipeline
+import os
+
+# Por defecto usa Gemma 2B instruct, puedes cambiarlo.
+LLM_MODEL = os.getenv("LLM_METADATA_MODEL", "google/gemma-1.1-2b-it")
+
+# Sólo cargar el pipeline una vez (lazy).
+_llm = None
+
+def get_llm():
+    global _llm
+    if _llm is None:
+        _llm = pipeline(
+            "text-generation",
+            model=LLM_MODEL,
+            device_map="auto",
+            max_new_tokens=256,
+            trust_remote_code=True
+        )
+    return _llm
+
+def build_prompt(text: str) -> str:
+    # Puedes personalizar el prompt aquí
+    return (
+        "Analiza el siguiente texto y responde en JSON con las claves:\n"
+        "\"summary\": resumen en una frase;\n"
+        "\"topics\": lista de hasta 5 palabras clave o temas principales;\n"
+        "\"language\": idioma detectado ('es', 'en', etc.);\n"
+        "\"vertical\": vertical temática (SEO-LLM, eCommerce, etc).\n\n"
+        "TEXTO:\n" + text.strip() + "\n\nJSON:"
+    )
+
+def extract_metadata(text: str) -> dict:
+    llm = get_llm()
+    prompt = build_prompt(text)
+    output = llm(prompt)[0]['generated_text']
+    # Extraer sólo el bloque JSON del resultado
+    import re, json
+    m = re.search(r'\\{.*\\}', output, re.DOTALL)
+    if not m:
+        return {}
+    try:
+        result = json.loads(m.group(0))
+        return result
+    except Exception:
+        return {}
+
+# --- CLI de prueba rápida
+if __name__ == "__main__":
+    chunk = "El SEO para LLMs consiste en optimizar contenido pensando en cómo los grandes modelos de lenguaje recuperan información relevante, mejorando la estructura, claridad y contexto del texto para aumentar la calidad de las respuestas generadas por IA."
+    meta = extract_metadata(chunk)
+    print(meta)

src/preprocess.py

@@ -1,4 +1,4 @@
-"""CLI: lee → chunkea → embed → JSONL."""
+"""CLI entry‑point: read files → chunk → embed → enrich metadata → JSONL."""
 from pathlib import Path
 from typing import Dict, List
 import json
@@ -6,36 +6,47 @@ import click
 from .reader import read_file
 from .chunker import chunk_text
 from .embeddings import embed_texts
+from .metadata_llm import extract_metadata
 
 @click.command()
-@click.option("--input-dir", type=click.Path(exists=True, file_okay=False),
-              required=True, help="Carpeta con documentos.")
-@click.option("--output", type=click.Path(), required=True,
-              help="Ruta del JSONL de salida.")
-@click.option("--vertical", default="general", help="Vertical.")
-@click.option("--language", default="es", help="Idioma.")
+@click.option('--input-dir', type=click.Path(exists=True, file_okay=False), required=True, help='Directory with docs.')
+@click.option('--output', type=click.Path(), required=True, help='JSONL output path.')
+@click.option('--vertical', default='general', help='Vertical tag.')
+@click.option('--language', default='es', help='Language tag.')
 def main(input_dir: str, output: str, vertical: str, language: str):
-    recs: List[Dict] = []
-    for p in Path(input_dir).iterdir():
-        if not p.is_file():
+    records: List[Dict] = []
+    for path in Path(input_dir).iterdir():
+        if not path.is_file():
             continue
-        meta, body = read_file(p)
-        base_meta = {"vertical": vertical, "language": language, "source": p.name, **meta}
+        meta, body = read_file(path)
+        merged_meta = {
+            'vertical': vertical,
+            'language': language,
+            'source': path.name,
+            **meta
+        }
         chunks = chunk_text(body)
-        vecs = embed_texts(chunks)
-        for i, (chunk, vec) in enumerate(zip(chunks, vecs), 1):
-            recs.append({
-                "id": f"{p.stem}-chunk-{i:04d}",
-                "vector": vec,
-                "metadata": {**base_meta, "chunk_index": i}
-            })
-    out = Path(output)
-    out.parent.mkdir(parents=True, exist_ok=True)
-    with out.open("w", encoding="utf-8") as f:
-        for r in recs:
+        embeddings = embed_texts(chunks)
+        for i, (chunk, vec) in enumerate(zip(chunks, embeddings), 1):
+            meta_llm = extract_metadata(chunk)
+            rec = {
+                'id': f"{path.stem}-chunk-{i:04d}",
+                'vector': vec,
+                'metadata': {
+                    **merged_meta,
+                    'chunk_index': i,
+                    **meta_llm   # summary, topics, vertical, language
+                }
+            }
+            records.append(rec)
+
+    out_path = Path(output)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    with out_path.open('w', encoding='utf-8') as f:
+        for r in records:
             json.dump(r, f, ensure_ascii=False)
-            f.write("\n")
-    click.echo(f"Wrote {len(recs)} records → {out}")
+            f.write('\n')
+    click.echo(f"Wrote {len(records)} records → {out_path}")
 
-if __name__ == "__main__":
+if __name__ == '__main__':
     main()