DEV Community: Jesus Oviedo Riquelme

LLMZ25-6 Review : Propuesta de Integración con LangChain para lus-laboris-py

Jesus Oviedo Riquelme — Sun, 26 Oct 2025 18:42:23 +0000

lus-laboris-py es un sistema de investigación legal impulsado por IA que actualmente utiliza una arquitectura basada en FastAPI con integración de OpenAI, Qdrant para búsqueda vectorial, y Arize Phoenix para monitoreo. Este post propone una modernización arquitectónica integrando LangChain para mejorar la modularidad, escalabilidad y mantenibilidad del sistema.

Arquitectura Actual vs Propuesta

Arquitectura Actual de lus-laboris-py

Según el README del proyecto github.com/jesusoviedo/lus-laboris-py, el sistema actual implementa:

Usuario → FastAPI → OpenAI LLM → Qdrant Vector Search → Respuesta Legal

Stack Tecnológico Actual:

FastAPI para la API
OpenAI para procesamiento de LLM
Qdrant para búsqueda vectorial
Arize Phoenix para monitoreo
Docker para containerización
UV como gestor de paquetes

Características Actuales:

Sistema de investigación legal automatizado
Integración con Qdrant para búsqueda de documentos legales
Monitoreo con Phoenix
Documentación con Guías en docs/README.md
Deploy a GCP

Limitaciones detectadas:

Arquitectura sin framework especializado (sin LangChain)
Flujo de trabajo custom sin estándares LCEL
Prompts probablemente embebidos en código
No aprovecha retriever patterns de LangChain
Limitada modularidad y reutilización

Arquitectura Propuesta (Con LangChain)

Usuario → Query Analysis → Multi-Search → Re-ranking → Legal Analysis → Response

Ventajas de la Propuesta:

Modular con RunnableLambda para orquestación flexible
Prompts separados con PromptTemplate para mejor mantenibilidad
Recuperación avanzada con ParentDocumentRetriever para jerarquía de documentos
Re-ranking con ContextualCompressionRetriever y Flashrank
Observabilidad mejorada con Phoenix integrado a LangChain
Escalable con LCEL (LangChain Expression Language)
Patterns estándar de la industria
Fácil integración con nuevos LLMs (Claude, Gemini, etc.)

Implementación Propuesta

1. Estructura de Prompts

from langchain.prompts import PromptTemplate

QUERY_ANALYSIS_PROMPT = PromptTemplate(
    input_variables=["user_query", "country", "legal_area"],
    template="""
Eres un experto en análisis de consultas legales.

Consulta del usuario: {user_query}
País: {country}
Área legal: {legal_area}

Instrucciones:
1. Identifica el tipo de consulta legal
2. Extrae conceptos clave y términos legales relevantes
3. Genera entre 3-5 consultas de búsqueda optimizadas

Devuelve JSON con: query_type, key_concepts, search_queries, related_areas
"""
)

LEGAL_ANALYSIS_PROMPT = PromptTemplate(
    input_variables=["query", "legal_documents", "jurisdiction"],
    template="""
Eres un abogado experto analizando documentos legales.

Consulta: {query}
Jurisdicción: {jurisdiction}

Documentos legales:
{legal_documents}

Analiza y proporciona: resumen ejecutivo, análisis legal, leyes relevantes, recomendaciones
"""
)

2. Pipeline con RunnableLambda

from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
from langchain_qdrant import QdrantVectorStore

llm = ChatOpenAI(model="gpt-4o", temperature=0.3)

def _analyze_query(inputs):
    """Etapa 1: Analizar consulta"""
    chain = QUERY_ANALYSIS_PROMPT | llm
    result = chain.invoke({
        "user_query": inputs["query"],
        "country": inputs["country"],
        "legal_area": inputs["legal_area"]
    })
    return {**inputs, "query_analysis": parse_json(result.content)}

def _retrieve_docs(inputs):
    """Etapa 2: Recuperar documentos"""
    all_docs = []
    for query in inputs["search_queries"]:
        docs = compressed_retriever.invoke(query)
        all_docs.extend(docs)
    return {**inputs, "documents": deduplicate(all_docs)}

def _analyze_legal(inputs):
    """Etapa 3: Análisis legal con LLM"""
    chain = LEGAL_ANALYSIS_PROMPT | llm
    result = chain.invoke({
        "query": inputs["query"],
        "legal_documents": format_docs(inputs["documents"]),
        "jurisdiction": inputs["country"]
    })
    return {**inputs, "legal_analysis": result.content}

# Pipeline completo
legal_workflow = (
    RunnableLambda(_analyze_query)
    | RunnableLambda(_retrieve_docs)
    | RunnableLambda(_analyze_legal)
)

def process_legal_query(query: str, country: str = "Spain"):
    """Procesar consulta legal"""
    return legal_workflow.invoke({"query": query, "country": country})

Comparación Antes vs Después

Aspecto	Sin LangChain	Con LangChain
Prompts	Embebidos en código	`PromptTemplate` separado
Workflow	Hardcodeado	Pipeline modular
Retrieval	Manual	`ParentDocumentRetriever` + re-ranking
Observabilidad	Limitada	Phoenix integrado
Debugging	Difícil	Traces completos

Ventajas de la Nueva Arquitectura

✅ Modularidad: Componentes intercambiables
✅ Observabilidad: Traces con Phoenix
✅ Escalabilidad: Fácil agregar features
✅ Mantenibilidad: Código más limpio
✅ Optimización: A/B testing de prompts
✅ Performance: Retrieval avanzado + re-ranking

Plan de Migración (10 semanas)

Fase 1-2: Preparación (Semanas 1-2)

Instalar dependencias de LangChain: uv add langchain langchain-openai langchain-qdrant
Configurar integración de Phoenix con LangChain
Revisar estructura actual del proyecto
Estudiar README y documentación existente

Fase 3-4: Migración de Prompts (Semanas 3-4)

Extraer prompts del código actual
Convertir a PromptTemplate de LangChain
Crear archivo legal_prompts.py con templates especializados
Testing A/B de prompts

Fase 5-6: Implementación de Retrievers (Semanas 5-6)

Configurar ParentDocumentRetriever con Qdrant existente
Implementar ContextualCompressionRetriever con Flashrank
Optimizar chunking para documentos legales (500-800 chars)
Implementar deduplicación de resultados

Fase 7-8: Pipeline con RunnableLambda (Semanas 7-8)

Convertir funciones existentes a RunnableLambda
Implementar workflow_chain con LCEL
Migrar lógica de búsqueda a retriever chain
Testing end-to-end

Fase 9-10: Optimización y Deployment (Semanas 9-10)

Configurar observabilidad mejorada con Phoenix
Monitorear métricas (latency, cost, quality)
Optimizar prompts y parámetros
Actualizar documentación y guías
Deploy a GCP con nueva arquitectura

Conclusión

La integración de LangChain en lus-laboris-py transformará el sistema en una arquitectura moderna, observable y escalable, siguiendo las mejores prácticas demostradas en el proyecto ziweidoushu.

Beneficios Esperados

Modularidad Mejorada
- Componentes intercambiables (Qdrant, Pinecone, etc.)
- Fácil cambiar de GPT-4 a Claude u otros LLMs
- Prompts reutilizables y versionados
Observabilidad Profesional
- Traces automáticos con Phoenix
- Métricas detalladas de latency, tokens, costos
- Debug simplificado con spans y context
Escalabilidad Futura
- Agregar nuevas fuentes legales es trivial
- Soporte multi-idioma con translate chains
- Agentes para búsqueda dinámica
Mantenibilidad
- Código más limpio y organizado
- Separación clara de concerns
- Testing más fácil de unit e integration

Repositorios:

ziweidoushu (referencia): github.com/clementlwm94/ziweidoushu
lus-laboris-py: github.com/jesusoviedo/lus-laboris-py

Referencias:

LLMZ25-5 Review : Arquitectura RAG del Proyecto Ziweidoushu

Jesus Oviedo Riquelme — Sun, 26 Oct 2025 18:33:42 +0000

En este post, exploramos cómo LangChain se integra en un sistema RAG (Retrieval-Augmented Generation) real para crear un asistente de astrología china (紫微斗数). El proyecto ziweidoushu es un ejemplo de arquitectura de producción que utiliza LangChain para orquestar un flujo complejo de procesamiento de documentos, recuperación híbrida, y generación de respuestas personalizadas.

Repositorio del Proyecto: github.com/clementlwm94/ziweidoushu

Ziweidoushu es un sistema que:

Genera cartas natales chinas desde datos de nacimiento
Recupera conocimiento relevante de una base de datos vectorial (Qdrant)
Combina información específica del usuario con conocimiento del dominio
Genera respuestas personalizadas usando LLMs (GPT-4)

¿Qué es LangChain y por qué usarlo?

LangChain es un framework de Python diseñado para construir aplicaciones basadas en LLMs (Large Language Models). Proporciona:

Chains (Cadenas): Flujos de trabajo pre-construidos que conectan múltiples componentes
Agents (Agentes): Componentes que pueden tomar decisiones y usar herramientas
Memory (Memoria): Gestión de estado conversacional
Prompts: Plantillas reutilizables para interacciones con LLMs
Retrievers: Integraciones con bases de datos vectoriales y sistemas de búsqueda

Ventajas de usar LangChain:

✅ Modularidad: Componentes intercambiables
✅ Escalabilidad: Fácil agregar nuevas funcionalidades
✅ Observabilidad: Integración nativa con Phoenix/OpenTelemetry
✅ Estándares: Protocolo LCEL (LangChain Expression Language)

Arquitectura del Proyecto Ziweidoushu

El flujo de trabajo de ziweidoushu implementa un pipeline RAG de 9 etapas usando LangChain:

Usuario → Generación de Carta → Generación de Consultas → Recuperación → Resumen → Generación de Respuesta → Respuesta Final

Componentes LangChain Utilizados

1. PromptTemplate: Gestión de Prompts Especializados

LangChain PromptTemplate permite crear prompts reutilizables con variables dinámicas. En el proyecto, se definen múltiples prompts especializados:

from langchain.prompts import PromptTemplate

query_prompt_text = """
角色: 你是一个专业的紫微斗数检索查询生成器

目标: 基于用户的命盘关键信息与具体问题，生成若干高质量、可直接用于检索的查询字符串

约束:
- 仅使用用户的命盘中出现的宫位、主星、辅星、煞星；不要编造
- 使用标准中文术语与命名
- 生成 1-3 条查询，按相关性从高到低排序

输入:
- 用户命盘: {user_chart}
- 用户问题: {user_question}

输出: 仅输出 JSON 字符串数组
"""

query_prompt = PromptTemplate(
    input_variables=["user_question", "user_chart"],
    template=query_prompt_text,
)

Ventajas:

Separación de lógica y prompts
Reutilización en múltiples contextos
Fácil mantenimiento y actualización

2. RunnableLambda: Orquestación del Pipeline

RunnableLambda permite encadenar funciones Python en un flujo de trabajo secuencial. El proyecto implementa un pipeline de 9 etapas:

from langchain_core.runnables import RunnableLambda

def _with_chart(inputs: Dict[str, Any]) -> Dict[str, Any]:
    """Etapa 1: Genera la carta natal"""
    chart = full_chart_generation(inputs["birth_date"], inputs["birth_hour"], inputs["gender"])
    return {**inputs, "user_chart": chart}

def _with_queries(inputs: Dict[str, Any]) -> Dict[str, Any]:
    """Etapa 4: Genera consultas de búsqueda"""
    queries = generate_queries(
        inputs["translated_question"],
        inputs["user_chart"],
        top_n=inputs.get("top_n_queries", 3),
        llm=inputs.get("llm"),
    )
    return {**inputs, "queries": queries}

# Pipeline completo con RunnableLambda
workflow_chain = (
    RunnableLambda(_with_chart)          # 1. Generar carta
    | RunnableLambda(_with_llm)          # 2. Inicializar LLM
    | RunnableLambda(_with_translated_question)  # 3. Traducir pregunta
    | RunnableLambda(_with_queries)      # 4. Generar consultas
    | RunnableLambda(_with_documents)     # 5. Recuperar documentos
    | RunnableLambda(_with_documents_text) # 6. Formatear documentos
    | RunnableLambda(_with_summary)       # 7. Resumir pasajes
    | RunnableLambda(_with_answer)         # 8. Generar respuesta
    | RunnableLambda(_with_localized_answer) # 9. Localizar respuesta
    | RunnableLambda(_finalize)           # Finalizar
)

¿Por qué RunnableLambda?

Expresa flujo secuencial claramente
Permite paso de estado entre etapas
Integración con herramientas de observabilidad (Phoenix)
Composición flexible con operador |

3. ParentDocumentRetriever: Recuperación Jerárquica

ParentDocumentRetriever implementa un patrón de recuperación jerárquico que combina chunks pequeños (para búsqueda precisa) con documentos padres completos (para contexto):

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import LocalFileStore
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Almacén de documentos padres
fs = LocalFileStore(STORE_PATH)
doc_store = create_kv_docstore(fs)

# Splitter para chunks hijos
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

# Retriever jerárquico
retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,           # Busca chunks pequeños
    docstore=doc_store,                 # Trae documentos padres completos
    child_splitter=child_splitter,
    search_kwargs={"k": 10},           # Top-10 chunks
    id_key="source_id"
)

¿Cómo funciona?

Busca en vector DB usando chunks pequeños (400 chars)
Recupera documentos padres completos desde docstore
Proporciona contexto más rico que solo chunks

Ventajas:

✅ Precisión de búsqueda (chunks pequeños)
✅ Contexto completo (documentos padres)
✅ Mejor calidad de RAG

4. ContextualCompressionRetriever: Re-ranking con Flashrank

ContextualCompressionRetriever mejora la calidad de documentos recuperados mediante re-ranking contextual:

from langchain.retrievers.contextual_compression import ContextualCompressionRetriever
from langchain_community.document_compressors import FlashrankRerank

# Compresor de contexto
compressor = FlashrankRerank()

# Retriever con compresión
compression_retriever = ContextualCompressionRetriever(
    base_retriever=retriever,
    base_compressor=compressor,
)

¿Qué hace FlashrankRerank?

Usa un modelo de re-ranking (MRM3) entrenado para relevancia
Reordena documentos por relevancia contextual
Mejora la precisión del Top-K

Ventajas:

✅ Reduce ruido de recuperación inicial
✅ Mejora precisión en Top-5/Top-10
✅ Mejor calidad de contexto para LLM

5. ChatOpenAI: Integración con LLMs

ChatOpenAI proporciona una interfaz unificada para interactuar con modelos OpenAI:

from langchain_openai import ChatOpenAI

def _create_llm(api_key: Optional[str], llm_model: Optional[str]) -> ChatOpenAI:
    target_model = llm_model or LLM_MODEL  # "gpt-4o-mini"
    return ChatOpenAI(
        model=target_model,
        temperature=LLM_TEMPERATURE,  # 0.3
        api_key=api_key
    )

Ventajas:

API unificada (no importa si usas GPT-4, Claude, etc.)
Integración con LCEL (LangChain Expression Language)
Manejo automático de rate limiting
Streaming support

6. QdrantVectorStore: Búsqueda Híbrida

QdrantVectorStore integra Qdrant con LangChain para búsqueda híbrida (dense + sparse):

from langchain_qdrant import QdrantVectorStore, FastEmbedSparse, RetrievalMode
from langchain_community.embeddings.fastembed import FastEmbedEmbeddings

# Embeddings densos (semánticos)
embeddings = FastEmbedEmbeddings(model_name="jinaai/jina-embeddings-v2-base-zh")

# Embeddings sparse (keyword-based)
sparse_embeddings = FastEmbedSparse(model_name="Qdrant/BM25")

# Vector store con modo híbrido
vectorstore = QdrantVectorStore(
    embedding=embeddings,                 # Dense embeddings
    client=client,
    collection_name=COLLECTION_NAME,
    sparse_embedding=sparse_embeddings,   # Sparse embeddings
    retrieval_mode=RetrievalMode.HYBRID,  # 🎯 Modo híbrido
)

¿Qué es búsqueda híbrida?

Dense: Captura similitud semántica
Sparse (BM25): Captura coincidencias de keywords
Híbrida: Combina ambos para mejor recall

Ventajas:

✅ Mejor recall que solo dense o solo sparse
✅ Optimizado para preguntas específicas de dominio
✅ Embeddings optimizados para chino (jina-v2-base-zh)

Flujo de Trabajo Completo

El pipeline completo de ziweidoushu con LangChain:

graph TD
    A[User Question + Birth Data] --> B[Generate Chart]
    B --> C[Translate Question]
    C --> D[Generate Queries]
    D --> E[Hybrid Retrieval]
    E --> F[Flashrank Rerank]
    F --> G[Summarize Passages]
    G --> H[Generate Answer]
    H --> I[Translate Answer]
    I --> J[Final Response]

Implementación con LangChain:

def run_qdrant_rag_workflow(
    question: str,
    birth_date: str,
    birth_hour: int,
    gender: str,
    top_n_queries: int = 5,
    api_key: Optional[str] = None,
    llm_model: Optional[str] = None,
) -> Dict[str, Any]:
    """Ejecutar el workflow RAG completo de Qdrant."""

    # Crear payload inicial
    payload = {
        "question": question,
        "birth_date": birth_date,
        "birth_hour": birth_hour,
        "gender": gender,
        "top_n_queries": top_n_queries,
        "api_key": api_key,
        "llm_model": llm_model,
    }

    # Ejecutar pipeline completo
    result = workflow_chain.invoke(payload)

    return result

Ventajas de Usar LangChain

Modularidad Extrema
- Cada componente es intercambiable
- Fácil cambiar de Qdrant a Pinecone
- Fácil cambiar de GPT-4 a Claude
Observabilidad Integrada
- Traces automáticos con Phoenix
- Mide latencia, tokens, costos
- Debug más fácil con spans detallados
LCEL (LangChain Expression Language)
- Composición fluida con |
- Streaming nativo
- Batching automático
Retrievers Reutilizables
- ParentDocumentRetriever con docstore
- ContextualCompressionRetriever con re-ranking
- Fácil cambiar estrategias de recuperación
Ecosistema Maduro
- Integración con 100+ herramientas
- Documentación excelente
- Comunidad activa

Lecciones Aprendidas

Usa RunnableLambda para pipelines secuenciales complejos
- Más flexibilidad que SequentialChain
- Permite paso de estado complejo
- Integración mejor con observabilidad
Separa prompts del código
- PromptTemplate hace código más limpio
- Fácil iterar en prompts sin tocar lógica
- Reutilización en múltiples contextos
Invierte en retrieval quality
- ParentDocumentRetriever + ContextualCompressionRetriever = mejor RAG
- Búsqueda híbrida mejora recall
- Re-ranking mejora precisión
Monitorea todo
- Integra Phoenix desde el inicio
- Mide métricas clave (latency, quality)
- Debug con traces
Optimiza para tu dominio
- Embeddings específicos (jina-v2-base-zh para chino)
- Prompts especializados por tarea
- Chunking adaptado a tu contenido

Conclusión

El proyecto ziweidoushu demuestra cómo LangChain puede usarse para construir un sistema RAG de producción robusto y escalable. Los componentes clave que hacen esto posible son:

RunnableLambda: Orquestación flexible del pipeline
PromptTemplate: Gestión de prompts especializados
ParentDocumentRetriever: Recuperación jerárquica
ContextualCompressionRetriever: Re-ranking contextual
QdrantVectorStore: Búsqueda híbrida
ChatOpenAI: Integración LLM

La arquitectura resultante es modular, observable y mantenible, lo que permite:

Iteración rápida en prompts
Cambio fácil de componentes
Debug eficiente con Phoenix
Escalabilidad hacia más funcionalidades

Referencias

LLMZ25-4 Review : Mejorando React + TailwindCSS

Jesus Oviedo Riquelme — Sun, 26 Oct 2025 14:27:48 +0000

Siguiendo nuestra discusión previa sobre React y TailwindCSS en QuizMate, este post explora cómo mejorar las implementaciones de estas tecnologías, ya sea que estés empezando desde el enfoque de QuizMate o integrándolas en proyectos como lus-laboris-py.

Cubriremos mejoras listas para producción, optimizaciones de rendimiento y patrones profesionales que elevan tus aplicaciones React + TailwindCSS.

Análisis del Estado Actual: QuizMate

Lo que QuizMate Hace Bien ✅

Configuración simple - Enfoque basado en CDN elimina la complejidad del build
Arquitectura basada en componentes - Separación limpia de responsabilidades
Clases de utilidad de TailwindCSS - Estilizado rápido sin CSS personalizado
Manejo de estado reactivo - useEffect y useState para flujo de datos
Buenos patrones de UX - Estados de carga, manejo de errores, renderizado condicional

Áreas para Mejorar 📈

Optimizaciones de rendimiento
Organización de código y reutilización
Proceso de build para producción
Seguridad de tipos y límites de errores
Estrategias de testing

Mejora 1: Migrar a una Configuración con Build

Enfoque Actual (CDN)

<!-- Simple pero limitado -->
<script src="https://unpkg.com/react@18/umd/react.production.min.js"></script>

Enfoque Mejorado (Con Build)

Paso 1: Inicializar un Proyecto React Moderno

# Usando Vite (Recomendado - Rápido y Moderno)
npm create vite@latest quizmate-frontend -- --template react

# O usando Create React App
npx create-react-app quizmate-frontend

Paso 2: Instalar TailwindCSS

cd quizmate-frontend
npm install -D tailwindcss postcss autoprefixer
npx tailwindcss init -p

Paso 3: Configurar TailwindCSS

tailwind.config.js:

/** @type {import('tailwindcss').Config} */
export default {
  content: [
    "./index.html",
    "./src/**/*.{js,ts,jsx,tsx}",
  ],
  theme: {
    extend: {
      colors: {
        'quiz-blue': '#3B82F6',
        'quiz-green': '#10B981',
      },
    },
  },
  plugins: [],
}

Paso 4: Agregar Directivas de TailwindCSS

src/index.css:

@tailwind base;
@tailwind components;
@tailwind utilities;

@layer components {
  .btn-primary {
    @apply bg-blue-600 text-white px-4 py-2 rounded-md 
           font-medium hover:bg-blue-700 transition-colors 
           disabled:bg-gray-300 disabled:cursor-not-allowed;
  }

  .card {
    @apply bg-white rounded-lg shadow-md p-6;
  }
}

Beneficios:

✅ Tree-shaking elimina CSS no usado
✅ Carga de páginas más rápida
✅ Mejor experiencia de desarrollo (hot reload)
✅ Puedes usar TypeScript
✅ Tamaños de bundle más pequeños

Mejora 2: Extracción de Componentes y Reutilización

Código Actual (Todo en un Archivo)

function App() {
    // 500+ líneas de código en un archivo
    return <div>...</div>;
}

Estructura Mejorada

src/
├── components/
│   ├── common/
│   │   ├── Button.jsx
│   │   ├── Card.jsx
│   │   ├── Input.jsx
│   │   └── LoadingSpinner.jsx
│   ├── quiz/
│   │   ├── FileUpload.jsx
│   │   ├── QuizCreator.jsx
│   │   └── QuizTaker.jsx
│   └── layout/
│       ├── Header.jsx
│       └── Footer.jsx
├── hooks/
│   ├── useQuizGeneration.js
│   ├── useFileUpload.js
│   └── useSources.js
├── services/
│   └── api.js
└── utils/
    └── validation.js

Ejemplo: Componente Botón Reutilizable

src/components/common/Button.jsx:

import PropTypes from 'prop-types';

export const Button = ({ 
    children, 
    onClick, 
    disabled = false, 
    variant = 'primary',
    size = 'md',
    className = '',
    ...props 
}) => {
    const clasesBase = 'font-medium rounded-md transition-colors';

    const clasesVariante = {
        primary: 'bg-blue-600 text-white hover:bg-blue-700',
        secondary: 'bg-gray-200 text-gray-900 hover:bg-gray-300',
        success: 'bg-green-600 text-white hover:bg-green-700',
        danger: 'bg-red-600 text-white hover:bg-red-700',
    };

    const clasesTamaño = {
        sm: 'px-3 py-1.5 text-sm',
        md: 'px-4 py-2 text-base',
        lg: 'px-6 py-3 text-lg',
    };

    const clasesDeshabilitado = 'disabled:bg-gray-300 disabled:cursor-not-allowed';

    return (
        <button
            onClick={onClick}
            disabled={disabled}
            className={`${clasesBase} ${clasesVariante[variant]} 
                       ${clasesTamaño[size]} ${clasesDeshabilitado} ${className}`}
            {...props}
        >
            {children}
        </button>
    );
};

Button.propTypes = {
    children: PropTypes.node.isRequired,
    onClick: PropTypes.func,
    disabled: PropTypes.bool,
    variant: PropTypes.oneOf(['primary', 'secondary', 'success', 'danger']),
    size: PropTypes.oneOf(['sm', 'md', 'lg']),
    className: PropTypes.string,
};

Uso:

import { Button } from './components/common/Button';

<Button 
    onClick={handleClick}
    variant="primary"
    size="md"
    disabled={isLoading}
>
    Generar Quiz
</Button>

Mejora 3: Hooks Personalizados para Separación de Lógica

Código Actual (Lógica Mezclada con UI)

function QuizCreator() {
    const [sources, setSources] = useState([]);
    const [generating, setGenerating] = useState(false);

    // Lógica de API mezclada con UI
    const fetchSources = async () => {
        // ...
    };

    return <div>...</div>;
}

Mejorado: Extraer a Hook Personalizado

src/hooks/useQuizGeneration.js:

import { useState, useEffect } from 'react';

export const useQuizGeneration = () => {
    const [sources, setSources] = useState([]);
    const [loading, setLoading] = useState(true);
    const [error, setError] = useState(null);

    useEffect(() => {
        fetchSources();
    }, []);

    const fetchSources = async () => {
        try {
            setLoading(true);
            const response = await fetch('/ingestion/sources');
            if (!response.ok) throw new Error('Falló al obtener fuentes');
            const data = await response.json();
            setSources(data);
        } catch (err) {
            setError(err.message);
        } finally {
            setLoading(false);
        }
    };

    const generateQuiz = async (config) => {
        try {
            setLoading(true);
            const response = await fetch('/api/quiz', {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify(config),
            });

            if (!response.ok) throw new Error('Falló al generar quiz');

            const data = await response.json();
            return data.quiz;
        } catch (err) {
            setError(err.message);
            throw err;
        } finally {
            setLoading(false);
        }
    };

    return {
        sources,
        loading,
        error,
        generateQuiz,
        refetch: fetchSources,
    };
};

Uso:

import { useQuizGeneration } from './hooks/useQuizGeneration';

function QuizCreator() {
    const { sources, loading, error, generateQuiz } = useQuizGeneration();
    const [query, setQuery] = useState('');

    const handleGenerate = async () => {
        try {
            const quiz = await generateQuiz({
                query: query.trim(),
                numberOfQuestions: 5,
            });
            onQuizGenerated(quiz);
        } catch (err) {
            console.error('Falló al generar quiz:', err);
        }
    };

    return (
        <div>
            {loading && <LoadingSpinner />}
            {error && <ErrorMessage>{error}</ErrorMessage>}
            {/* Componentes UI */}
        </div>
    );
}

Beneficios:

✅ Lógica reutilizable
✅ Testeable de forma aislada
✅ Código de componente más limpio
✅ Mejor manejo de errores

Mejora 4: Capa de Servicio API

Código Actual (Llamadas Fetch Inline)

const response = await fetch('/api/quiz', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({...})
});
const data = await response.json();

Mejorado: Servicio API Centralizado

src/services/api.js:

const API_BASE_URL = import.meta.env.VITE_API_URL || 'http://localhost:8080';

class ApiService {
    async request(endpoint, options = {}) {
        const url = `${API_BASE_URL}${endpoint}`;
        const config = {
            headers: {
                'Content-Type': 'application/json',
                ...options.headers,
            },
            ...options,
        };

        try {
            const response = await fetch(url, config);

            if (!response.ok) {
                const error = await response.json().catch(() => ({}));
                throw new Error(error.message || `HTTP ${response.status}`);
            }

            const data = await response.json();
            return data;
        } catch (error) {
            console.error(`Error API [${endpoint}]:`, error);
            throw error;
        }
    }

    async post(endpoint, body) {
        return this.request(endpoint, {
            method: 'POST',
            body: JSON.stringify(body),
        });
    }

    async get(endpoint) {
        return this.request(endpoint, { method: 'GET' });
    }

    async uploadFile(endpoint, file, additionalData) {
        const formData = new FormData();
        formData.append('file', file);

        if (additionalData) {
            Object.entries(additionalData).forEach(([key, value]) => {
                formData.append(key, value);
            });
        }

        return this.request(endpoint, {
            method: 'POST',
            body: formData,
            headers: {}, // Dejar que el navegador establezca Content-Type con boundary
        });
    }
}

export const apiService = new ApiService();

// Métodos API específicos
export const quizApi = {
    generate: (config) => apiService.post('/api/quiz', config),
    evaluate: (request) => apiService.post('/api/evaluation/evaluate-query', request),
};

export const ingestionApi = {
    upload: (file, sourceName) => apiService.uploadFile('/ingestion/upload', file, { sourceName }),
    getSources: () => apiService.get('/ingestion/sources'),
};

Uso:

import { quizApi, ingestionApi } from './services/api';

// Limpio y simple
const quiz = await quizApi.generate({ query, source, numberOfQuestions });
const sources = await ingestionApi.getSources();
await ingestionApi.upload(file, sourceName);

Mejora 5: Mejores Prácticas de TailwindCSS

1. Usar @apply para Patrones Repetitivos

En lugar de:

<div className="bg-white rounded-lg shadow-md p-6">...</div>

Crear componentes reutilizables:

/* tailwind.css */
@layer components {
  .card {
    @apply bg-white rounded-lg shadow-md p-6;
  }

  .card-hover {
    @apply card hover:shadow-lg transition-shadow;
  }

  .input-field {
    @apply block w-full px-3 py-2 border border-gray-300 
           rounded-md shadow-sm focus:outline-none 
           focus:ring-blue-500 focus:border-blue-500;
  }
}

Uso:

<div className="card">...</div>
<div className="card-hover">...</div>
<input className="input-field" />

2. Usar Variables CSS para Temas

tailwind.config.js:

export default {
  theme: {
    extend: {
      colors: {
        primary: {
          50: 'var(--color-primary-50)',
          500: 'var(--color-primary-500)',
          900: 'var(--color-primary-900)',
        },
      },
    },
  },
}

3. Organizar con Valores Arbitrarios

// Tamaños de texto responsivos
<div className="text-[clamp(1rem,4vw,2rem)]">
  Texto Adaptativo
</div>

// Cálculos complejos
<div className="w-[calc(100%-2rem)]">
  Ancho Calculado
</div>

// Mejor que hardcoded
<div className="h-[600px]">
  Altura Fija
</div>

4. Usar Modo JIT

Asegurar que JIT está habilitado en tailwind.config.js:

export default {
  mode: 'jit', // Compilación Just-in-Time
  content: ['./src/**/*.{js,jsx,ts,tsx}'],
}

Mejora 6: Límites de Errores y Estados de Carga

Componente Error Boundary

src/components/common/ErrorBoundary.jsx:

import React from 'react';

class ErrorBoundary extends React.Component {
    constructor(props) {
        super(props);
        this.state = { hasError: false, error: null };
    }

    static getDerivedStateFromError(error) {
        return { hasError: true, error };
    }

    componentDidCatch(error, errorInfo) {
        console.error('Error capturado por boundary:', error, errorInfo);
    }

    render() {
        if (this.state.hasError) {
            return (
                <div className="min-h-screen flex items-center justify-center bg-gray-50">
                    <div className="text-center">
                        <h2 className="text-2xl font-bold text-red-600 mb-4">
                            Algo salió mal
                        </h2>
                        <p className="text-gray-600 mb-4">
                            {this.state.error?.message || 'Ocurrió un error inesperado'}
                        </p>
                        <button
                            onClick={() => window.location.reload()}
                            className="bg-blue-600 text-white px-4 py-2 rounded-md hover:bg-blue-700"
                        >
                            Recargar Página
                        </button>
                    </div>
                </div>
            );
        }

        return this.props.children;
    }
}

export default ErrorBoundary;

Uso:

import ErrorBoundary from './components/common/ErrorBoundary';

function App() {
    return (
        <ErrorBoundary>
            <TusComponentes />
        </ErrorBoundary>
    );
}

Componente de Carga

src/components/common/LoadingSpinner.jsx:

export const LoadingSpinner = ({ size = 'md', className = '' }) => {
    const sizeClasses = {
        sm: 'w-4 h-4',
        md: 'w-8 h-8',
        lg: 'w-12 h-12',
    };

    return (
        <div className={`flex justify-center items-center ${className}`}>
            <div className={`${sizeClasses[size]} border-4 border-blue-200 border-t-blue-600 rounded-full animate-spin`} />
        </div>
    );
};

Mejora 7: Testing

Configurar Testing

npm install -D vitest @testing-library/react @testing-library/jest-dom

src/__tests__/Button.test.jsx:

import { render, screen, fireEvent } from '@testing-library/react';
import { Button } from '../components/common/Button';

describe('Componente Button', () => {
    it('renderiza con texto', () => {
        render(<Button>Haz Clic</Button>);
        expect(screen.getByText('Haz Clic')).toBeInTheDocument();
    });

    it('llama onClick cuando se hace clic', () => {
        const handleClick = vi.fn();
        render(<Button onClick={handleClick}>Clic</Button>);

        fireEvent.click(screen.getByText('Clic'));
        expect(handleClick).toHaveBeenCalledTimes(1);
    });

    it('está deshabilitado cuando disabled prop es true', () => {
        render(<Button disabled>Clic</Button>);
        expect(screen.getByText('Clic')).toBeDisabled();
    });
});

Aplicar Estas Mejoras a lus-laboris-py

Camino de Migración

Fase 1: Configurar Build Moderno

npm create vite@latest lus-laboris-frontend -- --template react
cd lus-laboris-frontend
npm install -D tailwindcss postcss autoprefixer
npx tailwindcss init -p

Fase 2: Crear Integración API

npm install axios # o usar fetch

Fase 3: Estructura de Componentes

// api/lusLaborisApi.js
import axios from 'axios';

const api = axios.create({
    baseURL: 'http://localhost:5000', // Tu backend Python
});

export const lusLaborisApi = {
    getJobs: () => api.get('/api/trabajos'),
    getJobById: (id) => api.get(`/api/trabajos/${id}`),
    aplicarTrabajo: (jobId, datosAplicacion) => 
        api.post(`/api/trabajos/${jobId}/aplicar`, datosAplicacion),
};

Fase 4: Migración Gradual

Comenzar con nuevas características en React
Mantener funcionalidad existente funcionando
Migrar página por página
Mantener compatibilidad de API

Despliegue en Producción

Build para Producción

npm run build
# Salida: dist/

Servir con tu Backend

Python (Flask):

from flask import Flask, send_from_directory

app = Flask(__name__, static_folder='../lus-laboris-frontend/dist')

@app.route('/')
def index():
    return send_from_directory('../lus-laboris-frontend/dist', 'index.html')

Python (FastAPI):

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse

app = FastAPI()

app.mount("/static", StaticFiles(directory="../lus-laboris-frontend/dist"), name="static")

@app.get("/")
async def index():
    return FileResponse('../lus-laboris-frontend/dist/index.html')

Puntos Clave

Configuración con build proporciona mejor DX y rendimiento
Extracción de componentes mejora mantenibilidad
Hooks personalizados separan lógica de presentación
Capa de servicio API centraliza comunicación
Mejores prácticas de TailwindCSS optimizan estilizado
Límites de errores mejoran resiliencia
Testing asegura calidad

Comparación: Antes vs Después

Aspecto	Enfoque CDN (QuizMate)	Enfoque con Build (Mejorado)
Tiempo de Setup	< 1 minuto	~5 minutos
Velocidad de Desarrollo	Rápido inicialmente	Rápido con HMR
Tamaño de Bundle	~500KB (todo React)	Optimizado, tree-shaken
Tiempo de Build	0 segundos	~1-2 segundos
Listo para Producción	Bueno para prototipos	Mejor para producción
Experiencia de Desarrollador	Buena	Excelente

Conclusión

Mientras el enfoque de CDN de QuizMate es excelente para comenzar rápidamente, migrar a una configuración con build y organización adecuada produce beneficios significativos:

Mejor experiencia de desarrollador
Rendimiento mejorado
Mantenibilidad aumentada
Código listo para producción
Opciones de seguridad de tipos

Estas mejoras aplican ya sea que estés mejorando QuizMate o construyendo un nuevo frontend para proyectos como lus-laboris-py.

LLMZ25-3 Review : Construyendo Interfaces Modernas con React y TailwindCSS

Jesus Oviedo Riquelme — Sun, 26 Oct 2025 14:17:11 +0000

En el panorama actual del desarrollo web, crear interfaces de usuario hermosas y responsivas de manera eficiente es crucial. QuizMate demuestra cómo React y TailwindCSS pueden trabajar juntos para crear frontends poderosos sin configuraciones de build complejas. Este post explora cómo QuizMate implementa estas tecnologías y cómo enfoques similares pueden mejorar otros proyectos como lus-laboris-py.

Cómo QuizMate Usa React y TailwindCSS

1. Configuración Sin Herramientas de Build Complejas

QuizMate toma un enfoque pragmático usando React vía CDN y TailwindCSS a través de su CDN:

<!-- Enfoque de un solo archivo HTML -->
<!DOCTYPE html>
<html lang="es">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>QuizMate - Generador de Quiz IA</title>

    <!-- Cargar React desde CDN -->
    <script crossorigin src="https://unpkg.com/react@18/umd/react.production.min.js"></script>
    <script crossorigin src="https://unpkg.com/react-dom@18/umd/react-dom.production.min.js"></script>

    <!-- Cargar Babel para transformación JSX -->
    <script src="https://unpkg.com/@babel/standalone/babel.min.js"></script>

    <!-- Cargar TailwindCSS desde CDN -->
    <script src="https://cdn.tailwindcss.com"></script>
</head>

Ventajas:

✅ No se requiere paso de build
✅ Rápido para prototipar y desplegar
✅ Perfecto para aplicaciones de una sola página
✅ Funciona con cualquier backend (Java, Python, etc.)
✅ Fácil de integrar con proyectos existentes

2. Arquitectura Basada en Componentes

La aplicación está organizada en componentes React reutilizables:

// Componente Principal App
function App() {
    const [quiz, setQuiz] = useState(null);

    return (
        <div className="min-h-screen p-8 bg-gray-50">
            <div className="max-w-4xl mx-auto">
                <h1 className="text-4xl font-bold text-gray-900 mb-2">
                    QuizMate
                </h1>
                <p className="text-gray-600 mb-8">
                    Generación de quiz potenciada por IA desde tus documentos
                </p>

                <div className="space-y-6">
                    {!quiz ? (
                        <>
                            <FileUpload onUploadSuccess={handleUploadSuccess} />
                            <QuizCreator onQuizGenerated={handleQuizGenerated} />
                        </>
                    ) : (
                        <QuizTaker quiz={quiz} onRestart={handleRestartQuiz} />
                    )}
                </div>
            </div>
        </div>
    );
}

Tres Componentes Principales:

Componente FileUpload - Maneja la carga de documentos
Componente QuizCreator - Permite a los usuarios configurar y generar quizzes
Componente QuizTaker - Interfaz interactiva de quiz

3. Clases de Utilidad de TailwindCSS en Acción

QuizMate usa extensivamente TailwindCSS para el estilo:

function FileUpload({ onUploadSuccess }) {
    const [file, setFile] = useState(null);
    const [uploading, setUploading] = useState(false);

    return (
        <div className="bg-white rounded-lg shadow-md p-6">
            <h2 className="text-2xl font-bold mb-4 text-gray-800">
                Subir Documento
            </h2>

            <button
                onClick={handleUpload}
                disabled={uploading}
                className={`w-full py-2 px-4 rounded-md font-medium transition-colors ${
                    uploading
                        ? 'bg-gray-300 text-gray-500 cursor-not-allowed'
                        : 'bg-blue-600 text-white hover:bg-blue-700'
                }`}
            >
                {uploading ? 'Subiendo...' : 'Subir e Ingesar'}
            </button>
        </div>
    );
}

Características Clave de TailwindCSS Usadas:

Diseño responsive (max-w-4xl mx-auto)
Sistema de colores (bg-gray-50, text-gray-900)
Utilidades de espaciado (p-6, mb-4, space-y-6)
Utilidades de sombra (shadow-md, rounded-lg)
Estados hover (hover:bg-blue-700)
Transiciones (transition-colors)
Clases condicionales (estados dinámicos de botones)

4. Manejo de Estado con React Hooks

La aplicación usa React hooks para el manejo de estado:

function QuizCreator({ onQuizGenerated }) {
    const [sources, setSources] = useState([]);
    const [selectedSource, setSelectedSource] = useState('');
    const [query, setQuery] = useState('');
    const [numberOfQuestions, setNumberOfQuestions] = useState(5);
    const [generating, setGenerating] = useState(false);
    const [message, setMessage] = useState({ type: '', text: '' });

    useEffect(() => {
        fetchSources();
    }, []);

    const fetchSources = async () => {
        try {
            const response = await fetch('/ingestion/sources');
            const data = await response.json();
            setSources(data);
            if (data.length > 0) {
                setSelectedSource(data[0]);
            }
        } catch (error) {
            setMessage({ type: 'error', text: 'Falló al cargar fuentes' });
        }
    };

    const handleGenerateQuiz = async () => {
        setGenerating(true);
        try {
            const response = await fetch('/api/quiz', {
                method: 'POST',
                headers: { 'Content-Type': 'application/json' },
                body: JSON.stringify({
                    query: query.trim(),
                    source: selectedSource,
                    numberOfQuestions: numberOfQuestions
                })
            });

            const data = await response.json();
            if (response.ok && data.quiz) {
                onQuizGenerated(data.quiz);
            }
        } finally {
            setGenerating(false);
        }
    };
}

Beneficios de Este Enfoque

1. Simplicidad

Sin configuración de webpack
Sin complejidad de pipeline de build
Fácil de entender y mantener

2. Experiencia de Desarrollador

No se necesita hot reloading para el desarrollo inicial
Todo el código en un solo lugar
Fácil de depurar

3. Rendimiento

TailwindCSS vía CDN está optimizado
React vía CDN está minificado y listo para producción
Sin preocupaciones de tamaño de bundle para apps pequeñas

4. Flexibilidad

Funciona con cualquier backend
Puede servirse como archivos estáticos
Fácil de desplegar en cualquier lugar

Cómo Aplicar Esto a lus-laboris-py

El proyecto lus-laboris-py podría beneficiarse de este enfoque:

Estado Actual (Probablemente)

Backend Python (FastAPI/Flask)
Puede usar plantillas Jinja2 o HTML básico
Interactividad limitada
CSS personalizado o estilizado mínimo

Implementación Recomendada

Paso 1: Crear un Directorio Frontend

mkdir -p lus-laboris-py/frontend
cd lus-laboris-py/frontend

Paso 2: Agregar un Archivo HTML Simple con React + TailwindCSS

Crear frontend/index.html:

<!DOCTYPE html>
<html lang="es">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>LUS Laboris</title>

    <script crossorigin src="https://unpkg.com/react@18/umd/react.production.min.js"></script>
    <script crossorigin src="https://unpkg.com/react-dom@18/umd/react-dom.production.min.js"></script>
    <script src="https://unpkg.com/@babel/standalone/babel.min.js"></script>
    <script src="https://cdn.tailwindcss.com"></script>
</head>
<body class="bg-gray-50">
    <div id="root"></div>

    <script type="text/babel">
        const { useState, useEffect } = React;

        function ListaTrabajos() {
            const [trabajos, setTrabajos] = useState([]);
            const [cargando, setCargando] = useState(true);

            useEffect(() => {
                obtenerTrabajos();
            }, []);

            const obtenerTrabajos = async () => {
                try {
                    const response = await fetch('/api/trabajos'); // Tu endpoint del backend
                    const data = await response.json();
                    setTrabajos(data);
                } catch (error) {
                    console.error('Error al obtener trabajos:', error);
                } finally {
                    setCargando(false);
                }
            };

            if (cargando) {
                return (
                    <div className="flex justify-center items-center h-64">
                        <div className="text-gray-600">Cargando...</div>
                    </div>
                );
            }

            return (
                <div className="max-w-6xl mx-auto p-6">
                    <h1 className="text-3xl font-bold text-gray-900 mb-6">
                        Ofertas de Empleo
                    </h1>

                    <div className="grid grid-cols-1 md:grid-cols-2 lg:grid-cols-3 gap-6">
                        {trabajos.map(trabajo => (
                            <div key={trabajo.id} className="bg-white rounded-lg shadow-md p-6 hover:shadow-lg transition-shadow">
                                <h3 className="text-xl font-semibold text-gray-900 mb-2">
                                    {trabajo.titulo}
                                </h3>
                                <p className="text-gray-600 mb-4">{trabajo.empresa}</p>
                                <div className="flex flex-wrap gap-2">
                                    <span className="px-3 py-1 bg-blue-100 text-blue-800 rounded-full text-sm">
                                        {trabajo.ubicacion}
                                    </span>
                                    <span className="px-3 py-1 bg-green-100 text-green-800 rounded-full text-sm">
                                        {trabajo.tipo}
                                    </span>
                                </div>
                            </div>
                        ))}
                    </div>
                </div>
            );
        }

        function App() {
            return (
                <div className="min-h-screen">
                    <nav className="bg-white shadow-md p-4">
                        <div className="max-w-7xl mx-auto">
                            <h1 className="text-2xl font-bold text-gray-900">
                                LUS Laboris
                            </h1>
                        </div>
                    </nav>
                    <main>
                        <ListaTrabajos />
                    </main>
                </div>
            );
        }

        ReactDOM.render(<App />, document.getElementById('root'));
    </script>
</body>
</html>

Paso 3: Servir desde tu Backend Python

# app.py o main.py
from flask import Flask, render_template, jsonify

app = Flask(__name__, static_folder='frontend')

@app.route('/')
def index():
    return app.send_static_file('index.html')

@app.route('/api/trabajos')
def obtener_trabajos():
    # Tu lógica del backend aquí
    trabajos = [
        {
            'id': 1,
            'titulo': 'Desarrollador Python',
            'empresa': 'Tech Corp',
            'ubicacion': 'Madrid',
            'tipo': 'Remoto'
        },
        # ... más trabajos
    ]
    return jsonify(trabajos)

if __name__ == '__main__':
    app.run(debug=True)

Puntos Clave

React + TailwindCSS vía CDN funciona bien para desarrollo rápido
No se requiere paso de build hace que sea perfecto para integrar con backends existentes
Arquitectura basada en componentes mejora la organización del código
Clases de utilidad de TailwindCSS permiten estilizado rápido sin CSS personalizado
Este enfoque encaja bien con backends Python como lus-laboris-py

Cuándo Usar Este Enfoque

✅ Bueno para:

Prototipado rápido
Aplicaciones pequeñas a medianas
Integrar con backends existentes
Proyectos sin desarrolladores frontend dedicados

❌ Considera alternativas cuando:

Necesites renderizado del lado del servidor (SSR)
Construyas aplicaciones de gran escala
Requieras manejo de estado complejo (Redux)
El equipo tenga experiencia sólida en herramientas de build modernas

Conclusión

QuizMate demuestra que se pueden construir UIs poderosas y modernas usando React y TailwindCSS sin herramientas complejas. Este enfoque es particularmente adecuado para proyectos como lus-laboris-py donde quieres agregar un frontend pulido a una aplicación Python existente.

La simplicidad de este enfoque, combinada con el poder del modelo de componentes de React y las clases de utilidad de TailwindCSS, crea una solución convincente para muchas aplicaciones web.

LLMZ25-2 Review : Construyendo Interfaces LLM con Streamlit

Jesus Oviedo Riquelme — Sun, 26 Oct 2025 02:18:41 +0000

Construir interfaces de usuario para aplicaciones de Modelos de Lenguaje Grande (LLM) puede ser desafiante, especialmente cuando quieres enfocarte en la lógica de IA en lugar del desarrollo frontend. Streamlit resuelve este problema proporcionando un framework simple basado en Python que transforma scripts de datos en aplicaciones web interactivas en minutos.

Esta guía te llevará a través del proceso de crear interfaces LLM con Streamlit, comparándolo con alternativas y demostrando patrones de implementación práctica usando ejemplos del mundo real.

Guía Paso a Paso para Integración con Streamlit

Paso 1: Instalación y Configuración

# Instalar Streamlit
pip install streamlit

# Crear tu primera app
streamlit hello

Paso 2: Estructura Básica de Interfaz LLM

import streamlit as st
import openai
from datetime import datetime

# Configuración de página
st.set_page_config(
    page_title="LLM Assistant",
    page_icon="🤖",
    layout="wide"
)

# Inicializar estado de sesión
if "conversation" not in st.session_state:
    st.session_state.conversation = []

# Barra lateral para configuración
with st.sidebar:
    st.header("Configuration")
    api_key = st.text_input("OpenAI API Key", type="password")
    model = st.selectbox("Model", ["gpt-4o-mini", "gpt-4o", "gpt-3.5-turbo"])
    temperature = st.slider("Temperature", 0.0, 2.0, 0.7)

Paso 3: Implementación de Interfaz de Chat

# Interfaz principal de chat
st.title("🤖 AI Legal Assistant")

# Mostrar historial de conversación
for message in st.session_state.conversation:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# Entrada de chat
if prompt := st.chat_input("Ask a legal question..."):
    # Agregar mensaje de usuario
    st.session_state.conversation.append({
        "role": "user", 
        "content": prompt,
        "timestamp": datetime.now()
    })

    # Mostrar mensaje de usuario
    with st.chat_message("user"):
        st.markdown(prompt)

    # Generar respuesta de IA
    with st.chat_message("assistant"):
        with st.spinner("Analyzing your question..."):
            response = generate_legal_response(prompt, api_key, model, temperature)
            st.markdown(response)

    # Agregar respuesta de IA a la conversación
    st.session_state.conversation.append({
        "role": "assistant", 
        "content": response,
        "timestamp": datetime.now()
    })

Comparación con Alternativas

Streamlit vs Gradio

Característica	Streamlit	Gradio
Curva de Aprendizaje	Suave, enfocado en Python	Empinada, requiere conocimiento ML
Personalización	Alta flexibilidad	Limitada a componentes predefinidos
Despliegue	Múltiples opciones	Principalmente Hugging Face
Caso de Uso	Apps web generales	Demos y prototipos ML
Complejidad de Código	Python simple	Basado en componentes

Streamlit vs Flask/FastAPI

Característica	Streamlit	Flask/FastAPI
Velocidad de Desarrollo	Muy rápida	Moderada a lenta
Conocimiento Frontend	No requerido	HTML/CSS/JS necesario
Desarrollo de API	Limitado	Excelente
Escalabilidad	Buena para prototipos	Excelente para producción
Curva de Aprendizaje	Mínima	Empinada

Caso de Estudio: Proyecto lus-laboris-py

El proyecto lus-laboris-py es un excelente ejemplo de cómo Streamlit puede complementar una aplicación existente. Actualmente, este proyecto cuenta con una API REST robusta implementada con FastAPI, pero le falta una interfaz de usuario accesible. Aquí es donde Streamlit puede agregar un valor significativo.

Estado Actual del Proyecto

lus-laboris-py actualmente tiene:

✅ API REST completa con FastAPI
✅ Sistema RAG funcional para investigación legal
✅ Base de datos con documentos legales
✅ Endpoints bien definidos para consultas
❌ Interfaz de usuario (ausente)

Recomendación: Agregar Streamlit como Frontend

¿Por qué Streamlit es la solución perfecta para lus-laboris-py?

Complementa FastAPI: Streamlit puede consumir la API existente sin modificar el backend
Desarrollo rápido: Se puede crear una UI completa en horas, no días
Sin conocimientos frontend: El equipo puede enfocarse en la lógica legal, no en HTML/CSS
Integración perfecta: Python nativo se integra naturalmente con FastAPI

Implementación Propuesta

# app_streamlit.py - Frontend para lus-laboris-py
import streamlit as st
import requests
import json

# Configuración
API_BASE_URL = "http://localhost:8000"  # URL de la API FastAPI

st.set_page_config(
    page_title="Lus Laboris - Asistente Legal",
    page_icon="⚖️",
    layout="wide"
)

st.title("⚖️ Lus Laboris - Asistente Legal")

# Sidebar para configuración
with st.sidebar:
    st.header("Configuración")
    api_url = st.text_input("URL de la API", value=API_BASE_URL)

# Interfaz principal de consulta legal
st.subheader("Consulta Legal")

query = st.text_area(
    "Describe tu consulta legal:",
    placeholder="Ejemplo: ¿Cuáles son los requisitos para un contrato de trabajo?",
    height=100
)

if st.button("Consultar", type="primary"):
    if query.strip():
        with st.spinner("Procesando consulta legal..."):
            try:
                # Llamada a la API FastAPI existente
                response = requests.post(
                    f"{api_url}/query",
                    json={"query": query},
                    headers={"Content-Type": "application/json"}
                )

                if response.status_code == 200:
                    result = response.json()

                    # Mostrar respuesta
                    st.success("Consulta procesada exitosamente")

                    # Respuesta principal
                    st.subheader("Respuesta:")
                    st.markdown(result.get("answer", "No se encontró respuesta"))

                    # Documentos fuente
                    if "sources" in result:
                        st.subheader("Documentos Fuente:")
                        for i, source in enumerate(result["sources"], 1):
                            with st.expander(f"Documento {i}: {source.get('title', 'Sin título')}"):
                                st.write(source.get("content", ""))
                                st.caption(f"Relevancia: {source.get('score', 'N/A')}")

                    # Métricas de rendimiento
                    if "metrics" in result:
                        col1, col2, col3 = st.columns(3)
                        with col1:
                            st.metric("Tiempo de Respuesta", f"{result['metrics'].get('response_time', 0):.2f}s")
                        with col2:
                            st.metric("Documentos Recuperados", result['metrics'].get('documents_found', 0))
                        with col3:
                            st.metric("Confianza", f"{result['metrics'].get('confidence', 0):.2%}")

                else:
                    st.error(f"Error en la API: {response.status_code}")

            except requests.exceptions.RequestException as e:
                st.error(f"Error de conexión: {str(e)}")
    else:
        st.warning("Por favor, ingresa una consulta legal")

# Sección de gestión de documentos
st.subheader("Gestión de Documentos")

tab1, tab2, tab3 = st.tabs(["Subir Documentos", "Buscar Documentos", "Estadísticas"])

with tab1:
    st.write("Cargar nuevos documentos legales al sistema")
    uploaded_file = st.file_uploader(
        "Seleccionar archivo legal",
        type=['pdf', 'docx', 'txt'],
        help="Formatos soportados: PDF, DOCX, TXT"
    )

    if uploaded_file:
        if st.button("Procesar Documento"):
            with st.spinner("Procesando documento..."):
                files = {"file": uploaded_file.getvalue()}
                response = requests.post(f"{api_url}/upload", files=files)

                if response.status_code == 200:
                    st.success("Documento procesado exitosamente")
                else:
                    st.error("Error al procesar el documento")

with tab2:
    st.write("Buscar documentos en la base de datos")
    search_query = st.text_input("Término de búsqueda")

    if st.button("Buscar"):
        if search_query:
            with st.spinner("Buscando..."):
                response = requests.get(f"{api_url}/search", params={"q": search_query})

                if response.status_code == 200:
                    results = response.json()
                    for doc in results.get("documents", []):
                        with st.expander(f"{doc.get('title', 'Sin título')}"):
                            st.write(doc.get("summary", ""))
                            st.caption(f"Fecha: {doc.get('date', 'N/A')}")

with tab3:
    st.write("Estadísticas del sistema")

    # Obtener estadísticas de la API
    try:
        stats_response = requests.get(f"{api_url}/stats")
        if stats_response.status_code == 200:
            stats = stats_response.json()

            col1, col2, col3, col4 = st.columns(4)

            with col1:
                st.metric("Total Documentos", stats.get("total_documents", 0))
            with col2:
                st.metric("Consultas Hoy", stats.get("queries_today", 0))
            with col3:
                st.metric("Tiempo Promedio", f"{stats.get('avg_response_time', 0):.2f}s")
            with col4:
                st.metric("Tasa de Éxito", f"{stats.get('success_rate', 0):.1%}")

            # Gráfico de consultas por día
            if "daily_queries" in stats:
                st.subheader("Consultas por Día")
                st.line_chart(stats["daily_queries"])

    except requests.exceptions.RequestException:
        st.error("No se pudieron cargar las estadísticas")

# Footer
st.markdown("---")
st.markdown("**Lus Laboris** - Sistema de Investigación Legal Asistida por IA")
st.caption("Powered by FastAPI + Streamlit + RAG")

Beneficios de esta Implementación

Reutilización del Backend: No se modifica la API FastAPI existente
UI Completa: Interfaz intuitiva para usuarios no técnicos
Desarrollo Rápido: Implementación en horas, no semanas
Mantenimiento Simple: Un solo archivo Python para toda la UI
Escalabilidad: Fácil despliegue con Streamlit Cloud

Arquitectura Propuesta

lus-laboris-py/
├── api/                    # Backend FastAPI existente
│   ├── main.py
│   ├── models/
│   └── routes/
├── frontend/               # Nuevo frontend Streamlit
│   ├── app_streamlit.py   # Aplicación principal
│   ├── components/        # Componentes reutilizables
│   └── utils/             # Utilidades
├── data/                  # Documentos legales
└── requirements.txt       # Dependencias actualizadas

Patrones de Diseño Comunes

Patrón 1: Aplicaciones Multi-Página

# Navegación de páginas
pages = {
    "Chat": chat_page,
    "Documents": documents_page,
    "Analytics": analytics_page,
    "Settings": settings_page
}

selected_page = st.sidebar.selectbox("Navigate", list(pages.keys()))
pages[selected_page]()

Patrón 2: Actualizaciones en Tiempo Real

# Auto-refresh para datos en tiempo real
if st.button("Enable Real-time Updates"):
    placeholder = st.empty()

    while True:
        latest_data = fetch_latest_data()
        placeholder.json(latest_data)
        time.sleep(5)  # Actualizar cada 5 segundos

Patrón 3: Pipeline de Procesamiento de Archivos

# Procesamiento de archivos con progreso
def process_files(uploaded_files):
    progress_bar = st.progress(0)
    status_text = st.empty()

    for i, file in enumerate(uploaded_files):
        status_text.text(f"Processing {file.name}...")
        process_single_file(file)
        progress_bar.progress((i + 1) / len(uploaded_files))

    st.success("All files processed!")

Tips de Despliegue

1. Streamlit Community Cloud

# .streamlit/config.toml
[server]
port = 8501
headless = true

[browser]
gatherUsageStats = false

2. Despliegue con Docker

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .
EXPOSE 8501

CMD ["streamlit", "run", "app_streamlit.py", "--server.port=8501", "--server.address=0.0.0.0"]

3. Configuración de Ambiente

# Variables de ambiente
import os
from dotenv import load_dotenv

load_dotenv()

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
DATABASE_URL = os.getenv("DATABASE_URL")
API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:8000")
DEBUG_MODE = os.getenv("DEBUG_MODE", "False").lower() == "true"

Optimización de Rendimiento

1. Caché

@st.cache_data
def load_legal_documents():
    return load_documents_from_database()

@st.cache_resource
def initialize_llm_model():
    return load_llm_model()

2. Operaciones Asíncronas

import asyncio
import aiohttp

async def fetch_legal_data_async(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

3. Connection Pooling de Base de Datos

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20
)

Mejores Prácticas

Gestión de Estado de Sesión: Usa st.session_state para mantener datos de usuario
Manejo de Errores: Implementa manejo comprensivo de errores para llamadas API
Estados de Carga: Siempre muestra indicadores de carga para operaciones largas
Diseño Responsivo: Usa columnas y contenedores para mejor layout
Seguridad: Nunca expongas API keys en código del lado del cliente
Testing: Escribe pruebas unitarias para tus funciones de lógica central

Recursos:

LLMZ25-1 Review : Streamlit La Herramienta Perfecta para Interfaces de Proyectos LLM

Jesus Oviedo Riquelme — Sun, 26 Oct 2025 02:16:10 +0000

Streamlit ha revolucionado la forma en que los desarrolladores crean aplicaciones web interactivas para proyectos de ciencia de datos y machine learning. Con su enfoque basado en Python simple, se ha convertido en el framework de referencia para construir interfaces de LLM (Modelos de Lenguaje Grande), haciendo que las aplicaciones de IA complejas sean accesibles para usuarios sin requerir conocimientos extensos de desarrollo frontend.

Por qué Streamlit es Ideal para Proyectos LLM

Streamlit ofrece varias ventajas clave que lo hacen perfecto para aplicaciones LLM:

Prototipado Rápido: Transforma scripts de Python en aplicaciones web en minutos
Sin Conocimientos de Frontend: Desarrollo puro en Python
Ecosistema Rico de Widgets: Componentes integrados para entrada de texto, interfaces de chat y visualización de datos
Gestión de Estado de Sesión: Perfecto para mantener historial de conversaciones
Actualizaciones en Tiempo Real: Actualización automática de UI cuando cambian los datos
Despliegue Fácil: Despliegue con un clic a Streamlit Community Cloud

Caso de Estudio: LegalMind - Un Asistente Legal Basado en RAG

Examinemos cómo se implementó Streamlit en LegalMind, un asistente legal sofisticado para el sistema de justicia criminal de India.

Características Clave Implementadas

1. Interfaz de Chat Interactiva

with st.form("chat_form", clear_on_submit=True):
    query = st.text_input(
        "Ask a legal question:",
        placeholder="Type your question here...",
        disabled=st.session_state.is_generating
    )
    submitted = st.form_submit_button("Send")

2. Streaming de Respuesta en Tiempo Real

# Visualización incremental de respuesta
final_answer = ""
for char in answer:
    final_answer += char
    placeholder.markdown(f"""
        <div style='padding:15px; margin:10px 0; border-radius:12px; background-color:#1e1e1e;'>
            <p><strong>🧑‍💼 Q:</strong> {query}</p>
            <div style='margin-top:5px;'>
                <strong>🤖 A:</strong> {final_answer} ▌
            </div>
        </div>
    """, unsafe_allow_html=True)
    time.sleep(0.01)  # Streaming carácter por carácter

3. Recolección de Feedback de Usuario

def save_feedback(idx, chat):
    def task():
        with engine.connect() as conn:
            conn.execute(text(
                "INSERT INTO feedback (query,response,feedback,timestamp,latency,answer_length) "
                "VALUES (:q,:r,:f,:t,:l,:al)"
            ), {
                "q": chat['query'],
                "r": chat['answer'],
                "f": chat['feedback'],
                "t": chat['timestamp'],
                "l": chat['latency'],
                "al": len(chat['answer'])
            })
            conn.commit()
    threading.Thread(target=task).start()

4. Dashboard de Analíticas Completo

# Dashboard con 6+ visualizaciones
st.bar_chart(df_feedback['feedback'].value_counts())  # Distribución de feedback
st.bar_chart(df_feedback['latency'])                  # Tiempo de respuesta
st.bar_chart(df_feedback['answer_length'])            # Longitud de respuesta
st.line_chart(avg_length)                             # Tendencias en el tiempo
st.image(wordcloud.to_array())                        # Nube de palabras
st.bar_chart(top_words.set_index('Keyword'))         # Palabras clave principales

Arquitectura Técnica

La aplicación LegalMind demuestra un sistema RAG (Retrieval-Augmented Generation) completo:

Frontend: UI Streamlit con interfaz de chat y dashboard
Backend: Pipeline RAG con búsqueda híbrida (Elasticsearch + Qdrant)
Base de Datos: PostgreSQL para almacenamiento de feedback
Integración LLM: OpenAI GPT-4o-mini vía API

Mejores Prácticas Demostradas

Gestión de Estado de Sesión: Manejo apropiado del historial de chat y estado del usuario
Integración de Base de Datos: PostgreSQL para almacenamiento persistente de datos
Manejo de Errores: Manejo elegante de fallos de API y casos edge
Optimización de Rendimiento: Threading para operaciones de base de datos no bloqueantes
Experiencia de Usuario: Feedback en tiempo real, estados de carga y diseño responsivo

Ejemplo de Código: Interfaz de Chat LLM Básica

Aquí hay una versión simplificada de cómo crear una interfaz de chat LLM con Streamlit:

import streamlit as st
from openai import OpenAI

# Inicializar cliente OpenAI
client = OpenAI(api_key="your-api-key")

# Configuración de página
st.set_page_config(page_title="LLM Chat Assistant", layout="wide")
st.title("🤖 AI Assistant")

# Inicializar estado de sesión
if "messages" not in st.session_state:
    st.session_state.messages = []

# Mostrar historial de chat
for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

# Entrada de chat
if prompt := st.chat_input("What would you like to know?"):
    # Agregar mensaje de usuario
    st.session_state.messages.append({"role": "user", "content": prompt})

    # Mostrar mensaje de usuario
    with st.chat_message("user"):
        st.markdown(prompt)

    # Generar respuesta de IA
    with st.chat_message("assistant"):
        with st.spinner("Thinking..."):
            response = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "m", "content": m["content"]} for m in st.session_state.messages]
            )
            answer = response.choices[0].message.content

        st.markdown(answer)

    # Agregar respuesta de IA al historial
    st.session_state.messages.append({"role": "assistant", "content": answer})

Despliegue Hecho Fácil

Streamlit ofrece múltiples opciones de despliegue:

Streamlit Community Cloud: Hosting gratuito para aplicaciones públicas
Docker: Containerizar tu aplicación
Plataformas Cloud: Desplegar en AWS, GCP o Azure
Desarrollo Local: Ejecutar con streamlit run app.py

Conclusión

Streamlit ha demostrado ser una excelente opción para interfaces de proyectos LLM, como lo demuestra el proyecto LegalMind. Su simplicidad, características poderosas y ecosistema extenso lo convierten en la herramienta perfecta para desarrolladores que quieren enfocarse en la lógica de IA en lugar del desarrollo frontend.

La combinación de la facilidad de uso de Streamlit con aplicaciones LLM sofisticadas crea interfaces poderosas y amigables para el usuario que pueden ser desplegadas rápidamente y mantenidas fácilmente.

Recursos:

LLPY-14: Evaluación y Métricas de Calidad - Midiendo el Éxito del RAG

Jesus Oviedo Riquelme — Fri, 17 Oct 2025 18:30:05 +0000

🎯 El Desafío de Medir Calidad en RAG

Imagina que tu sistema RAG está en producción:

✅ API responde en 1-3 segundos
✅ 100+ requests por día
✅ Usuarios parecen contentos

Pero hay preguntas críticas sin respuesta:

¿Las respuestas son correctas?
¿El LLM está inventando información (hallucinations)?
¿Los documentos recuperados son relevantes?
¿Qué % de queries tienen buena calidad?
¿Cómo identificar y fix problemas sistemáticos?

Sin evaluación: Volando a ciegas, optimizando por intuición, descubriendo problemas cuando usuarios se quejan.

📊 La Magnitud del Problema

Dimensiones de Calidad en RAG

Un sistema RAG tiene múltiples puntos de falla:

Query → Embedding → Qdrant Search → Reranking → LLM Generation → Answer

Cada paso puede fallar:
├─ Embedding: Query mal entendido
├─ Search: Documentos irrelevantes recuperados
├─ Reranking: Orden incorrecto
└─ LLM: Hallucination, respuesta incompleta, tono inadecuado

Tipos de Métricas Necesarias

📊 Retrieval Metrics: ¿Recuperamos los documentos correctos?
- Precision@k
- Recall@k
- MRR (Mean Reciprocal Rank)
- nDCG (Normalized Discounted Cumulative Gain)
🤖 Generation Metrics: ¿El LLM genera buenas respuestas?
- Relevance (pregunta ↔ respuesta)
- Correctness (contexto ↔ respuesta)
- Completeness
- Hallucination rate
- Grounding (respuesta basada en contexto)
⚡ System Metrics: ¿El sistema es rápido y confiable?
- Latency (p50, p95, p99)
- Throughput (requests/s)
- Error rate
- Availability (uptime)
💰 Cost Metrics: ¿Es sostenible económicamente?
- Tokens consumidos por query
- Costo por query
- Costo mensual proyectado

💡 La Solución: Evaluación Multi-Nivel

Arquitectura de Evaluación

┌───────────────────────────────────────────────────────┐
│              User Query (Production)                  │
└─────────────────┬─────────────────────────────────────┘
                  │
         ┌────────▼────────┐
         │  RAG Pipeline   │
         │  (Fast path)    │
         └────────┬────────┘
                  │
    ┌─────────────┴──────────────┐
    │                            │
Response to User         Enqueue Evaluation
(1-3 seconds)           (Non-blocking, async)
                                 │
                   ┌─────────────▼──────────────┐
                   │   Evaluation Worker        │
                   │   (Background thread)      │
                   └─────────────┬──────────────┘
                                 │
                   ┌─────────────▼──────────────┐
                   │   Phoenix Evals            │
                   │   (LLM-as-a-Judge)        │
                   ├────────────────────────────┤
                   │  - Relevance               │
                   │  - Hallucination           │
                   │  - Toxicity                │
                   │  - Grounding               │
                   └─────────────┬──────────────┘
                                 │
                   ┌─────────────▼──────────────┐
                   │   Phoenix UI               │
                   │   - Dashboards             │
                   │   - Metrics                │
                   │   - Drill-down             │
                   └────────────────────────────┘

Key insight: Evaluación asíncrona no bloquea respuesta al usuario.

🤖 LLM-as-a-Judge Evaluation

EvaluationService

Archivo src/lus_laboris_api/api/services/evaluation_service.py:

"""
Asynchronous evaluation service using Phoenix Evals
"""

import asyncio
import logging
import queue
from concurrent.futures import ThreadPoolExecutor
from typing import Any

from phoenix.evals import (
    HALLUCINATION_PROMPT_TEMPLATE,
    RAG_RELEVANCY_PROMPT_TEMPLATE,
    OpenAIModel,
)

from ..config import settings
from .phoenix_service import phoenix_service

logger = logging.getLogger(__name__)


class EvaluationService:
    """Service for asynchronous RAG response evaluation"""

    def __init__(self):
        self.enabled = settings.api_phoenix_enabled
        self.evaluation_queue = queue.Queue()
        self.executor = ThreadPoolExecutor(
            max_workers=2,
            thread_name_prefix="eval-worker"
        )

        if self.enabled:
            self._initialize_evaluators()
            self._start_evaluation_worker()
            logger.info("Evaluation service initialized")
        else:
            logger.warning("Evaluation service disabled")

    def _initialize_evaluators(self):
        """Initialize Phoenix evaluators with LLM"""
        try:
            # Use cost-effective model for evaluations
            eval_model = "gpt-4o-mini"  # Barato y bueno para evals

            self.eval_model = OpenAIModel(
                model=eval_model,
                api_key=settings.openai_api_key
            )

            logger.info(f"Evaluators initialized with model: {eval_model}")

        except Exception as e:
            logger.exception("Failed to initialize evaluators")
            self.enabled = False

    def _start_evaluation_worker(self):
        """Start worker that processes evaluations in background"""

        def worker():
            logger.info("Evaluation worker started")
            while True:
                try:
                    # Get task from queue
                    eval_task = self.evaluation_queue.get(timeout=1.0)

                    if eval_task is None:  # Shutdown signal
                        break

                    # Execute evaluation
                    self._run_evaluation(eval_task)

                    self.evaluation_queue.task_done()

                except queue.Empty:
                    continue
                except Exception as e:
                    logger.exception("Error in evaluation worker")

        # Start worker in separate thread
        self.executor.submit(worker)

    def enqueue_evaluation(
        self,
        session_id: str,
        question: str,
        context: str,
        answer: str,
        documents: list[dict],
        metadata: dict | None = None,
    ):
        """Enqueue evaluation for async processing (non-blocking)"""
        if not self.enabled:
            return

        eval_task = {
            "session_id": session_id,
            "question": question,
            "context": context,
            "answer": answer,
            "documents": documents,
            "metadata": metadata or {},
            "timestamp": datetime.now().isoformat(),
        }

        self.evaluation_queue.put(eval_task)
        logger.debug(f"Evaluation enqueued for session {session_id}")

    def _run_evaluation(self, eval_task: dict):
        """Execute evaluations (runs async loop)"""
        try:
            # Create new event loop for this thread
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            loop.run_until_complete(self._run_evaluation_async(eval_task))
        finally:
            loop.close()

    async def _run_evaluation_async(self, eval_task: dict):
        """Execute Phoenix Evals in parallel (optimized)"""
        session_id = eval_task["session_id"]
        question = eval_task["question"]
        context = eval_task["context"]
        answer = eval_task["answer"]

        try:
            logger.info(f"Running parallel evaluations for session {session_id}")

            start_time = time.time()

            # ✅ Run all 3 evaluations in parallel
            evaluation_results = await asyncio.gather(
                asyncio.to_thread(
                    self._evaluate_relevance,
                    question, context, answer
                ),
                asyncio.to_thread(
                    self._evaluate_hallucination,
                    question, context, answer
                ),
                asyncio.to_thread(
                    self._evaluate_toxicity,
                    answer
                ),
                return_exceptions=True,  # Don't fail if one fails
            )

            eval_time = time.time() - start_time

            # Extract results
            relevance_score = (
                evaluation_results[0] 
                if not isinstance(evaluation_results[0], Exception) 
                else None
            )
            hallucination_score = (
                evaluation_results[1]
                if not isinstance(evaluation_results[1], Exception)
                else None
            )
            toxicity_score = (
                evaluation_results[2]
                if not isinstance(evaluation_results[2], Exception)
                else None
            )

            # Track results in Phoenix
            phoenix_service.track_evaluation(
                session_id=session_id,
                metrics={
                    "relevance": relevance_score,
                    "hallucination": hallucination_score,
                    "toxicity": toxicity_score,
                    "grounding": 1 - hallucination_score if hallucination_score else None,
                },
                metadata={
                    **eval_task["metadata"],
                    "evaluation_time": eval_time,
                }
            )

            logger.info(
                f"Evaluation completed for session {session_id} in {eval_time:.2f}s"
            )

        except Exception as e:
            logger.exception(f"Evaluation failed for session {session_id}")

    def _evaluate_relevance(self, question: str, context: str, answer: str) -> float:
        """Evaluate if answer is relevant to question"""
        try:
            # Format prompt with Phoenix RAG_RELEVANCY_PROMPT_TEMPLATE
            # Template expects: input, reference
            variable_values = {
                "input": question,     # Spanish question
                "reference": context,   # Spanish context
            }
            prompt = RAG_RELEVANCY_PROMPT_TEMPLATE.format(variable_values)

            # Call model directly (it's callable)
            eval_response = self.eval_model(prompt)

            # Parse response (Phoenix returns "relevant" or "irrelevant")
            if eval_response and "relevant" in eval_response.lower():
                return 1.0  # Highly relevant
            if eval_response and "irrelevant" in eval_response.lower():
                return 0.0  # Not relevant
            return 0.5  # Indeterminate

        except Exception as e:
            logger.exception("Failed to evaluate relevance")
            return None

    def _evaluate_hallucination(self, question: str, context: str, answer: str) -> float:
        """Evaluate if answer contains hallucinations"""
        try:
            # Format prompt with Phoenix HALLUCINATION_PROMPT_TEMPLATE
            # Template expects: input, reference, output
            variable_values = {
                "input": question,   # Spanish question
                "reference": context,  # Spanish context
                "output": answer,    # Spanish answer
            }
            prompt = HALLUCINATION_PROMPT_TEMPLATE.format(variable_values)

            # Call model directly (it's callable)
            eval_response = self.eval_model(prompt)

            # Parse response (Phoenix returns "factual" or "hallucinated")
            if eval_response and "factual" in eval_response.lower():
                return 0.0  # No hallucinations
            if eval_response and ("hallucinated" in eval_response.lower() or 
                                  "hallucination" in eval_response.lower()):
                return 1.0  # High hallucination
            return 0.5  # Indeterminate

        except Exception as e:
            logger.exception("Failed to evaluate hallucination")
            return None

    def _evaluate_toxicity(self, answer: str) -> float:
        """Evaluate if answer contains toxic content"""
        try:
            # Use LLM to evaluate toxicity (Spanish prompt)
            toxicity_prompt = f"""
            Evalúa si la siguiente respuesta contiene contenido tóxico, ofensivo o inapropiado.

            RESPUESTA: {answer}

            Responde únicamente con "tóxico" o "no-tóxico".
            """

            # Call model directly
            eval_response = self.eval_model(toxicity_prompt)

            # Parse response
            if eval_response and ("no-tóxico" in eval_response.lower() or 
                                  "no toxico" in eval_response.lower()):
                return 0.0  # Non-toxic
            if eval_response and ("tóxico" in eval_response.lower() or 
                                  "toxico" in eval_response.lower()):
                return 1.0  # Toxic
            return 0.0  # Default: non-toxic (legal domain)

        except Exception as e:
            logger.exception("Failed to evaluate toxicity")
            return None


# Global service instance
evaluation_service = EvaluationService()

Características del código REAL (simplificado):

✅ Async/Non-blocking: No retrasa respuesta al usuario
✅ Queue-based: ThreadPoolExecutor procesa en background
✅ Parallel evaluations: 3 evals en paralelo con asyncio.gather
✅ Phoenix templates: HALLUCINATION_PROMPT_TEMPLATE, RAG_RELEVANCY_PROMPT_TEMPLATE
✅ Direct model calls: self.eval_model(prompt) - más simple que run_evals() con pandas
✅ Cost-effective: usa gpt-4o-mini ($0.15/1M tokens vs $30/1M GPT-4)
✅ Binary scoring: Retorna 1.0 (positivo), 0.0 (negativo), 0.5 (indeterminado)

Evaluaciones Implementadas

1. Relevance (Relevancia):

Question: "¿Cuántos días de vacaciones?"
Answer: "Según el Artículo 218, son 12 días hábiles."

Eval Prompt:
"Is the answer relevant to the question?
 Question: {question}
 Answer: {answer}

 Score 0-1 where 1 = highly relevant"

Score: 0.95 ✅ (muy relevante)

2. Hallucination (Invención):

Context: "Artículo 218: todo trabajador que cumpla un año... 12 días hábiles"
Answer: "Son 12 días hábiles según Artículo 218"

Eval Prompt:
"Does the response contain information NOT in the reference?
 Reference: {context}
 Response: {answer}

 Score 0-1 where 1 = full hallucination"

Score: 0.05 ✅ (casi sin hallucination)

3. Toxicity (Toxicidad):

Answer: "Según el Artículo 218..."

Eval Prompt:
"Does the text contain toxic, offensive, or harmful content?
 Text: {answer}

 Score 0-1 where 1 = very toxic"

Score: 0.0 ✅ (sin toxicidad)

📊 Métricas de Retrieval (Conceptos clave)

Nota: El proyecto actual NO implementa estas métricas automáticamente, pero son conceptos importantes para entender cómo evaluar sistemas RAG.

Métricas Tradicionales de Retrieval

Para evaluar si Qdrant está recuperando los documentos correctos, se usan métricas estándar:

1. Precision@k: ¿Cuántos de los top-k resultados son relevantes?

Precision@k = (Documentos relevantes en top-k) / k

Ejemplo:
Retrieved: [218, 42, 219, 81, 220]  # Top-5
Relevant:  [218, 219, 220]           # Ground truth

Precision@1 = 1/1 = 1.0   # Doc #1 es relevante ✅
Precision@3 = 2/3 = 0.67  # 2 de 3 son relevantes
Precision@5 = 3/5 = 0.60  # 3 de 5 son relevantes

2. Recall@k: ¿Qué % de documentos relevantes recuperamos?

Recall@k = (Documentos relevantes en top-k) / Total relevantes

Ejemplo:
Recall@5 = 3/3 = 1.0  # ✅ Recuperamos TODOS los relevantes

3. MRR (Mean Reciprocal Rank): Posición promedio del primer resultado relevante

MRR = 1 / (Posición del primer relevante)

Query 1: Relevante en #1 → MRR = 1/1 = 1.0
Query 2: Relevante en #3 → MRR = 1/3 = 0.33
Promedio: 0.67

Cómo implementarlas:

Crear un ground truth dataset: 50-100 queries con documentos relevantes anotados manualmente
Ejecutar queries contra Qdrant
Comparar resultados vs ground truth
Calcular métricas

📊 Evaluación pre-producción del proyecto: Antes de implementar el sistema en producción, se realizaron evaluaciones exhaustivas documentadas en:

02_vectorstore_embedding_exploration.ipynb
- Evaluación de 11 modelos de embeddings
- Métricas de performance y calidad
- Justificación del modelo seleccionado (multilingual-e5-small)
03_rag_pipeline_evaluation.ipynb
- Ground truth dataset de 50+ queries legales
- Métricas de retrieval: Precision@k, Recall@k, MRR, nDCG@k, Hit Rate@k
- Evaluación de reranking: Comparación antes/después con cross-encoder
- Evaluación LLM: LLM-as-a-Judge para medir calidad de respuestas
- Análisis comparativo: Múltiples configuraciones del pipeline

Estos notebooks representan la evidencia empírica que respaldó las decisiones técnicas del sistema RAG.

Herramientas recomendadas:

RAGAS: Framework de evaluación de RAG (github.com/explodinggradients/ragas)
Phoenix Evals: Incluye retrieval metrics (docs.arize.com/phoenix)

🎯 Integración en el RAG Pipeline

En el RAGService

async def answer_question(self, question: str, session_id: str):
    """Complete RAG pipeline with evaluation"""

    # 1. Retrieve documents
    documents, _ = self._retrieve_documents(question, session_id)

    # 2. Generate answer
    answer = await self._generate_response(question, documents, session_id)

    # 3. Build context
    context = self._build_context(documents)

    # 4. Enqueue evaluation (NON-BLOCKING)
    evaluation_service.enqueue_evaluation(
        session_id=session_id,
        question=question,
        context=context,
        answer=answer,
        documents=documents,
        metadata={
            "llm_provider": self.llm_provider,
            "llm_model": self.llm_model,
            "processing_time": processing_time,
        }
    )

    # 5. Return response immediately (evaluation runs in background)
    return {
        "success": True,
        "answer": answer,
        "session_id": session_id,
        ...
    }

Timeline:

t=0ms:    User request arrives
t=100ms:  Documents retrieved
t=2000ms: LLM generates answer
t=2010ms: Response returned to user ← USER HAPPY
t=2050ms: Evaluation enqueued (async)
t=4500ms: Evaluations complete (background) ← METRICS TRACKED

User experience: 2 seconds (no ve los 2.5s de evaluación)

📊 Visualización en Phoenix UI

Dashboard de Métricas

Acceso: http://localhost:6006 (local) o Phoenix Cloud

Vista de Evaluaciones:

┌─────────────────────────────────────────────────────┐
│ Evaluation Results (Last 7 days)                   │
│                                                     │
│ Total Queries Evaluated: 1,234                     │
│                                                     │
│ Relevance (avg):      0.89 ████████████████░░ 89%  │
│ Hallucination (avg):  0.05 ░░░░░░░░░░░░░░░░░░  5%  │
│ Toxicity (avg):       0.01 ░░░░░░░░░░░░░░░░░░  1%  │
│ Grounding (avg):      0.95 █████████████████░ 95%  │
│                                                     │
│ ┌─────────────────────────────────────────────┐   │
│ │  Relevance Distribution                     │   │
│ │  ┌────────────────────────────────────┐    │   │
│ │  │ [0.0-0.2]:  ▓ 2%                    │    │   │
│ │  │ [0.2-0.4]:  ▓ 3%                    │    │   │
│ │  │ [0.4-0.6]:  ▓▓ 5%                   │    │   │
│ │  │ [0.6-0.8]:  ▓▓▓▓▓ 12%               │    │   │
│ │  │ [0.8-1.0]:  ▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓ 78%     │    │   │
│ │  └────────────────────────────────────┘    │   │
│ └─────────────────────────────────────────────┘   │
│                                                     │
│ Top Issues:                                         │
│ - 12 queries with low relevance (<0.6)             │
│ - 6 potential hallucinations detected (>0.3)       │
│ - 0 toxic responses                                │
└─────────────────────────────────────────────────────┘

Drill-Down Individual

┌─────────────────────────────────────────────────────┐
│ Session: abc-123-def                                │
│ Timestamp: 2024-10-16 14:30:22                      │
│                                                     │
│ Question:                                           │
│ "¿Cuántos días de vacaciones corresponden?"        │
│                                                     │
│ Answer:                                             │
│ "Según el Artículo 218 del Código del Trabajo..."  │
│                                                     │
│ Evaluations:                                        │
│ ├─ Relevance:      0.92 ✅ (Highly relevant)       │
│ ├─ Hallucination:  0.03 ✅ (Minimal)               │
│ ├─ Toxicity:       0.00 ✅ (None)                  │
│ └─ Grounding:      0.97 ✅ (Well grounded)         │
│                                                     │
│ Retrieved Documents:                                │
│ 1. Art. 218 (score: 0.912, rerank: 0.987)         │
│ 2. Art. 219 (score: 0.876, rerank: 0.921)         │
│ 3. Art. 220 (score: 0.845, rerank: 0.889)         │
│                                                     │
│ Metadata:                                           │
│ - LLM: OpenAI GPT-3.5-turbo                        │
│ - Processing time: 1.234s                           │
│ - Reranking: Enabled                                │
└─────────────────────────────────────────────────────┘

🎯 Casos de Uso Reales

Para Identificar Problemas de Calidad:

"Algunas respuestas parecen incorrectas, pero no sé cuáles"

Solución:

Abrir Phoenix UI
Filtrar por relevance < 0.6
Ver queries problemáticas:
- "¿Qué es un contrato eventual?" → Relevance: 0.4
- Contexto recuperado: artículos sobre "contratos colectivos"
- Problema: Embeddings confunden "eventual" con "colectivo"

Fix: Mejorar pre-processing de queries o usar modelo de embeddings más grande

Para Detectar Hallucinations:

"¿El LLM está inventando información?"

Solución:

Phoenix UI → Filter hallucination > 0.3
Ver casos:
- Query: "¿Cuál es el salario mínimo?"
- Context: (vacío - no hay artículos sobre salario)
- Answer: "El salario mínimo es $2,500,000..." ⚠️
- Hallucination score: 0.9

Fix: Agregar guardrail: Si no hay contexto relevante → "No tengo información suficiente"

Para Optimizar Performance:

"¿Qué configuración da mejores resultados?"

Solución: A/B Testing

# Configuración A: Sin reranking
results_a = run_eval_suite(use_reranking=False)
# Relevance: 0.85

# Configuración B: Con reranking
results_b = run_eval_suite(use_reranking=True)
# Relevance: 0.92 (+7%)

# Conclusión: Reranking vale la pena!

Para Monitoreo Continuo:

"Quiero alertas cuando calidad baja"

Solución:

# En Phoenix o monitoring system
if avg_relevance_last_24h < 0.80:
    send_alert("⚠️ RAG quality degraded! Avg relevance: 0.78")

if hallucination_rate_last_24h > 0.10:
    send_alert("🚨 High hallucination rate: 12%")

🚀 El Impacto Transformador

Antes de Evaluación Automática:

🤷 Quality unknown: "Parece que funciona bien..."
🐛 Reactive debugging: Usuarios reportan problemas
📊 No metrics: Sin datos para optimizar
🎲 Blind optimization: Cambios sin medir impacto
⏱️ Manual testing: Probar 10-20 queries manualmente

Después de Evaluación Automática:

📊 Quality measured: "89% relevance, 5% hallucination"
🔍 Proactive debugging: Identificar problemas antes que usuarios
📈 Data-driven: Optimizar basado en métricas reales
🎯 A/B testing: Comparar configuraciones objetivamente
⚡ Continuous evaluation: 100% de queries evaluadas automáticamente

Métricas de Mejora:

Aspecto	Sin Evaluación	Con Evaluación	Mejora
Quality visibility	0%	100%	+∞
Problem detection	Días (user reports)	Minutos (automated)	-99%
Evaluation coverage	1-5% (manual)	100% (automated)	+2000%
Optimization confidence	Baja (guessing)	Alta (data-driven)	N/A
Time to fix issues	Días	Horas	-90%

💡 Lecciones Aprendidas

1. Async Evaluation es Crítico

Evaluación no debe agregar latencia a la respuesta del usuario. Queue + ThreadPoolExecutor = perfecto.

2. gpt-4o-mini es Ideal para Evals

GPT-4: Muy caro para evaluar cada query ($30/1M tokens)
GPT-3.5-turbo: Calidad inconsistente
gpt-4o-mini: Balance perfecto ($0.15/1M tokens, buena calidad)

3. Phoenix Templates > Custom

Los templates de Phoenix (HALLUCINATION, RAG_RELEVANCY) están battle-tested. Úsalos.

4. Parallel Evaluations = 3x Faster

asyncio.gather ejecuta 3 evals en paralelo → 2.5s en lugar de 7.5s

5. Ground Truth es Gold

Tener un dataset de 50-100 query-answer pairs validados permite comparaciones objetivas.

6. Monitorea Tendencias, No Puntos

Un query malo no importa. 10% de queries malos sí importa. Phoenix dashboards muestran trends.

🎯 El Propósito Más Grande

Evaluación y métricas no son "nice to have" - son el sistema de calidad que garantiza que el RAG entrega valor real. Sin evaluación:

No sabes si está funcionando
No puedes optimizar
No detectas regresiones
No justificas inversión

Con evaluación automática continua:

✅ Visibility: Calidad medida en cada query
✅ Confidence: Cambios basados en datos
✅ Quality assurance: Detección temprana de problemas
✅ Continuous improvement: Optimización basada en métricas
✅ Stakeholder reporting: Dashboards para el negocio
✅ Cost optimization: Identificar oportunidades de ahorro

Estamos convirtiendo el RAG de una "caja negra" a un sistema medido, monitoreado y optimizable, donde cada respuesta contribuye a mejorar el siguiente millón de respuestas.

🔗 Recursos y Enlaces

🎊 Fin de la Serie LLPY

Hemos completado un viaje de 14 posts construyendo un sistema RAG de clase mundial:

Serie Completa:

LLPY-01: Introducción al Sistema RAG
LLPY-02: Configuración con UV
LLPY-03: Extracción de Datos Legales
LLPY-04: Vectorización y Embeddings
LLPY-05: Qdrant Base de Datos Vectorial
LLPY-06: FastAPI API REST Robusta
LLPY-07: Integración LLMs (OpenAI + Gemini)
LLPY-08: Reranking para Precisión
LLPY-09: Phoenix y OpenTelemetry
LLPY-10: Autenticación JWT con RSA
LLPY-11: Terraform - Infraestructura como Código
LLPY-12: Docker y Containerización
LLPY-13: CI/CD con GitHub Actions
LLPY-14: Evaluación y Métricas de Calidad ← ESTE POST

Lo que Construimos:

✅ Sistema RAG completo end-to-end

✅ 413 artículos legales vectorizados

✅ API REST producción-ready

✅ Multi-provider LLM (OpenAI + Gemini)

✅ Observabilidad completa (Phoenix + OpenTelemetry)

✅ Infraestructura como código (Terraform)

✅ CI/CD automatizado (GitHub Actions)

✅ Evaluación continua (LLM-as-a-judge)

Métricas del Sistema (Observables con Phoenix):

Métrica	Objetivo	Medición
Latency (p50)	<2s	✅ Phoenix tracking
Latency (p95)	<5s	✅ Phoenix tracking
Relevance (avg)	>0.8	✅ LLM-as-a-Judge
Hallucination (avg)	<0.1	✅ LLM-as-a-Judge
Toxicity (avg)	<0.05	✅ LLM-as-a-Judge
Overall Quality	>0.85	✅ Weighted average
Cost per query	<$0.01	✅ Token tracking

Nota: Las métricas se recopilan automáticamente en Phoenix con cada query evaluada. Los valores "Objetivo" representan benchmarks típicos de RAG de producción, no mediciones actuales.

Impacto Transformador Final:

De 0 a sistema RAG de producción en 14 posts:

🚀 Performance: Retrieval optimizado (gRPC), generación async
🎯 Quality: LLM-as-a-Judge evaluando relevance, hallucination, toxicity
💰 Cost: gpt-4o-mini para evals ($0.15/1M tokens), escalable
🔒 Security: JWT RSA auth, GCP Secret Manager
📊 Observability: Phoenix + OpenTelemetry tracking completo
🔄 Deployment: CI/CD automático con GitHub Actions
📈 Evaluation: 100% de queries evaluadas asíncronamente (non-blocking)

Gracias por acompañarnos en este viaje! 🎉

Si tienes preguntas o quieres contribuir al proyecto, visita el repositorio en GitHub.

LLPY-13: CI/CD con GitHub Actions - Automatización Completa

Jesus Oviedo Riquelme — Fri, 17 Oct 2025 18:29:09 +0000

🎯 El Desafío de los Deployments Manuales

Imagina el proceso tradicional de deployment:

1. Developer hace cambios → 10 min
2. Run tests localmente → 5 min
3. Fix issues → 15 min
4. Build Docker image → 5 min
5. Push a Docker Hub → 3 min
6. SSH a servidor → 2 min
7. Pull nueva imagen → 2 min
8. Restart servicio → 1 min
9. Verificar deployment → 3 min
10. Rollback si falla → 10 min

TOTAL: 56 minutos + posibilidad de error humano

Problemas:

⏱️ Tiempo desperdiciado: 1 hora por deploy
🐛 Errores humanos: Olvidar un paso, typo en comando
📝 Sin tracking: ¿Quién deployó qué, cuándo?
🔄 Inconsistencia: Diferentes procesos por developer
🚨 Sin rollback fácil: Revertir requiere proceso manual

📊 La Magnitud del Problema

Requisitos de CI/CD Moderno

✅ Quality Gates: Tests, linting, security scan automáticos
🏗️ Build Automation: Docker images con cada cambio
📦 Artifact Management: Versionado y storage de images
🚀 Auto Deployment: Deploy a staging/prod automático
🔍 Validation: Health checks post-deployment
📊 Notifications: Alertas de successes/failures
🔄 Rollback: Fácil volver a versión anterior
🔐 Secrets Management: API keys, credentials seguros

7 Workflows del Proyecto

#	Workflow	Trigger	Propósito	Duración
1	code-quality.yml	Push/PR	Tests, linting, security	3-5 min
2	docker-processing-build-publish.yml	Push/Manual	Build processing image	4-6 min
3	docker-api-build-publish.yml	Push/Manual	Build API image	5-7 min
4	terraform-apply-on-tf-change.yml	Push .tf files	Apply infra changes	3-8 min
5	deploy-qdrant.yml	Manual/Push	Deploy Qdrant to VM	5-10 min
6	update-batch-job-image.yml	Manual	Update batch job image	2-3 min
7	update-api-secrets-deploy.yml	Manual	Update secrets + deploy API	3-5 min

💡 La Solución: GitHub Actions

¿Qué es GitHub Actions?

GitHub Actions es una plataforma de CI/CD integrada en GitHub que permite:

🤖 Automatización: Workflows como código (YAML)
⚡ Trigger flexible: Push, PR, schedule, manual
🔧 Actions marketplace: Miles de actions reutilizables
🔐 Secrets management: Secrets encriptados integrados
📊 Logging: Logs detallados de cada ejecución
💰 Free tier: 2000 minutos/mes para repos públicos

Arquitectura de un Workflow

name: Mi Workflow               # Nombre del workflow
on: [push, pull_request]        # Triggers
jobs:                           # Jobs (paralelos por defecto)
  build:                        # Job 1
    runs-on: ubuntu-latest      # Runner
    steps:                      # Steps (secuenciales)
      - uses: actions/checkout@v4    # Action del marketplace
      - run: echo "Hello"            # Comando shell
      - name: Build
        run: docker build -t app .   # Step con nombre

🚀 Workflow 1: Code Quality & Style

Archivo .github/workflows/code-quality.yml:

name: Code Quality & Style

on:
  push:
    branches: ['**']
    paths:
      - 'src/**/*.py'
      - 'utils/**/*.py'
      - 'tests/**/*.py'
      - '.pre-commit-config.yaml'
      - 'pyproject.toml'
      - '**/*.md'
      - '**/*.yml'
      - '**/*.yaml'

  pull_request:
    branches: ['**']
    paths:
      - 'src/**/*.py'
      - 'utils/**/*.py'
      - 'tests/**/*.py'
      - '.pre-commit-config.yaml'
      - 'pyproject.toml'
      - '**/*.md'
      - '**/*.yml'
      - '**/*.yaml'

jobs:
  # Job 1: Pre-commit hooks
  pre-commit:
    name: Pre-Commit Hooks
    runs-on: ubuntu-latest

    steps:
    - name: Checkout code
      uses: actions/checkout@v5
      with:
        fetch-depth: 0

    - name: Set up Python
      uses: actions/setup-python@v6
      with:
        python-version: '3.13'

    - name: Install UV
      uses: astral-sh/setup-uv@v6
      with:
        version: latest

    - name: Install pre-commit
      run: uv tool install pre-commit

    - name: Cache pre-commit hooks
      uses: actions/cache@v4
      with:
        path: ~/.cache/pre-commit
        key: pre-commit-${{ runner.os }}-${{ hashFiles('.pre-commit-config.yaml') }}

    - name: Run pre-commit hooks
      run: uv tool run pre-commit run --all-files --show-diff-on-failure

  # Job 2: Lint API code
  lint-api:
    name: Lint API Code
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v5

    - name: Set up Python
      uses: actions/setup-python@v6
      with:
        python-version: '3.13'

    - name: Install UV
      uses: astral-sh/setup-uv@v6

    - name: Install ruff
      run: uv tool install ruff

    - name: Run ruff linter
      run: uv tool run ruff check src/lus_laboris_api --output-format=github

  # Job 3: Security scan
  security-scan:
    name: Security Scan (Bandit)
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v5

    - name: Set up Python
      uses: actions/setup-python@v6
      with:
        python-version: '3.13'

    - name: Install UV
      uses: astral-sh/setup-uv@v6

    - name: Install bandit
      run: uv tool install bandit

    - name: Run bandit security scan
      run: uv tool run bandit -r src/ utils/ -f github

  # Job 4: Type checking
  type-check:
    name: Type Checking (MyPy)
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v5

    - name: Set up Python
      uses: actions/setup-python@v6
      with:
        python-version: '3.13'

    - name: Install UV
      uses: astral-sh/setup-uv@v6

    - name: Install dependencies
      run: |
        cd src/lus_laboris_api
        uv sync

    - name: Run mypy
      run: |
        cd src/lus_laboris_api
        uv run mypy api/ --ignore-missing-imports

  # Job 5: Tests
  test-api:
    name: API Tests (PyTest)
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v5

    - name: Set up Python
      uses: actions/setup-python@v6
      with:
        python-version: '3.13'

    - name: Install UV
      uses: astral-sh/setup-uv@v6

    - name: Install dependencies
      run: |
        cd src/lus_laboris_api
        uv sync

    - name: Run tests with coverage
      run: |
        cd src/lus_laboris_api
        uv run pytest tests/ --cov=api --cov-report=term --cov-report=html

    - name: Upload coverage reports
      uses: actions/upload-artifact@v4
      with:
        name: coverage-report
        path: src/lus_laboris_api/htmlcov/

Quality Gates:

✅ Pre-commit: Formatting, trailing whitespace, YAML syntax
✅ Linting: Ruff (10-100x más rápido que flake8)
✅ Security: Bandit detecta vulnerabilidades
✅ Type safety: MyPy valida type hints
✅ Tests: PyTest con coverage report

Triggers:

Push a cualquier branch que toque Python files
Pull requests

Resultado: Si algún job falla → PR bloqueado → Fix required

🐳 Workflows 2-3: Docker Build & Publish

Workflow 2: Processing Image

Archivo .github/workflows/docker-processing-build-publish.yml:

name: Build & Publish Docker Image (Processing)

on:
  workflow_dispatch:
  push:
    paths:
      - 'src/processing/Dockerfile'
      - 'src/processing/extract_law_text.py'
      - 'src/processing/pyproject.toml'
      - 'src/processing/.python-version'

jobs:
  build-and-push:
    runs-on: ubuntu-latest

    steps:
    - name: Checkout code
      uses: actions/checkout@v5

    - name: Set up Docker Buildx
      uses: docker/setup-buildx-action@v3

    - name: Login to Docker Hub
      uses: docker/login-action@v3
      with:
        username: ${{ secrets.DOCKER_HUB_USERNAME }}
        password: ${{ secrets.DOCKER_HUB_PASSWORD }}

    - name: Get date tag
      id: date_tag
      run: echo "tag=$(date +'%Y%m%d')" >> $GITHUB_OUTPUT

    - name: Build and push
      uses: docker/build-push-action@v6
      with:
        context: ./src/processing
        push: true
        tags: |
          ${{ secrets.DOCKER_HUB_USERNAME }}/lus-laboris-processing:latest
          ${{ secrets.DOCKER_HUB_USERNAME }}/lus-laboris-processing:${{ steps.date_tag.outputs.tag }}

Workflow 3: API Image

Similar a workflow 2, pero para src/lus_laboris_api/

Ventajas:

✅ Automated tagging: latest + fecha (20241016)
✅ Trigger smart: Solo rebuild si Dockerfile o código cambia
✅ Buildx: Soporte multi-platform (AMD64, ARM64)
✅ Registry caching: Reutiliza layers de builds anteriores

🏗️ Workflow 4: Terraform Apply

Archivo .github/workflows/terraform-apply-on-tf-change.yml:

name: Terraform Apply on TF Change

on:
  push:
    paths:
      - 'terraform/**/*.tf'
      - 'terraform/**/*.tfvars'
  workflow_dispatch:

jobs:
  terraform:
    runs-on: ubuntu-latest
    env:
      GOOGLE_APPLICATION_CREDENTIALS: gcp-key.json

    steps:
    - name: Checkout code
      uses: actions/checkout@v4

    - name: Setup Terraform
      uses: hashicorp/setup-terraform@v3
      with:
        terraform_version: 1.9.0

    - name: Create GCP credentials file
      run: |
        echo '${{ secrets.GSA_KEY }}' > gcp-key.json

    - name: Create terraform.tfvars
      working-directory: ./terraform
      run: |
        # Generate tfvars from GitHub secrets/variables
        cat > terraform.tfvars <<EOF
        project_id = "${{ secrets.GCP_PROJECT_ID }}"
        region = "${{ secrets.GCP_REGION }}"
        bucket_name = "${{ vars.GCP_BUCKET_NAME }}"
        # ... (más variables)
        EOF

    - name: Terraform Init
      working-directory: ./terraform
      run: terraform init

    - name: Terraform Validate
      working-directory: ./terraform
      run: terraform validate

    - name: Terraform Plan
      working-directory: ./terraform
      run: terraform plan -out=tfplan

    - name: Terraform Apply
      working-directory: ./terraform
      run: terraform apply -auto-approve tfplan

    - name: Show Outputs
      working-directory: ./terraform
      run: terraform output -json

Nota sobre el workflow real: El proyecto simplifica este paso usando el script tf_menu.sh:

El workflow genera un .env file con todas las variables de GitHub
Luego ejecuta bash tf_menu.sh 2 que auto-genera terraform.tfvars desde .env
Este enfoque reutiliza la misma lógica que el desarrollo local

Triggers:

Cambios en cualquier archivo .tf (no .tfvars)
Manual via workflow_dispatch

Proceso:

✅ Checkout código
✅ Authenticate to GCP
✅ Setup Terraform
✅ Generate .env file con variables de GitHub secrets/vars
✅ Run tf_menu.sh 2 para crear terraform.tfvars
✅ Init → Plan → Apply (auto-approve)

Resultado: Infraestructura actualizada automáticamente en 3-8 minutos

🗄️ Workflow 5: Deploy Qdrant to VM

Archivo .github/workflows/deploy-qdrant.yml:

name: Deploy Qdrant to GCP VM

on:
  workflow_dispatch:
  push:
    paths:
      - 'services/vectordb/**'
      - 'terraform/modules/compute_engine/**'

jobs:
  deploy:
    runs-on: ubuntu-latest

    steps:
    - name: Checkout code
      uses: actions/checkout@v4

    - name: Authenticate to GCP
      uses: google-github-actions/auth@v2
      with:
        credentials_json: ${{ secrets.GSA_KEY }}

    - name: Set up Cloud SDK
      uses: google-github-actions/setup-gcloud@v2

    - name: Get VM details
      id: vm_info
      run: |
        VM_EXTERNAL_IP=$(gcloud compute instances describe ${{ vars.GCP_COMPUTE_ENGINE_VM_NAME }} \
          --zone=${{ vars.GCP_COMPUTE_ENGINE_VM_ZONE }} \
          --format='get(networkInterfaces[0].accessConfigs[0].natIP)')

        echo "vm_ip=${VM_EXTERNAL_IP}" >> $GITHUB_OUTPUT

    - name: Install Docker on VM
      run: |
        gcloud compute ssh ${{ vars.GCP_COMPUTE_ENGINE_VM_NAME }} \
          --zone=${{ vars.GCP_COMPUTE_ENGINE_VM_ZONE }} \
          --command="
            # Install Docker if not exists
            if ! command -v docker &> /dev/null; then
              curl -fsSL https://get.docker.com -o get-docker.sh
              sudo sh get-docker.sh
              sudo usermod -aG docker \$USER
            fi

            # Install Docker Compose
            sudo curl -L https://github.com/docker/compose/releases/latest/download/docker-compose-\$(uname -s)-\$(uname -m) \
              -o /usr/local/bin/docker-compose
            sudo chmod +x /usr/local/bin/docker-compose
          "

    - name: Deploy Qdrant
      run: |
        # Copy docker-compose.yml
        gcloud compute scp services/vectordb/docker-compose.yml \
          ${{ vars.GCP_COMPUTE_ENGINE_VM_NAME }}:~/docker-compose.yml \
          --zone=${{ vars.GCP_COMPUTE_ENGINE_VM_ZONE }}

        # Create .env with API key
        gcloud compute ssh ${{ vars.GCP_COMPUTE_ENGINE_VM_NAME }} \
          --zone=${{ vars.GCP_COMPUTE_ENGINE_VM_ZONE }} \
          --command="
            echo 'QDRANT_API_KEY=${{ secrets.QDRANT_API_KEY }}' > ~/.env

            # Start Qdrant
            docker-compose up -d

            # Wait and verify
            sleep 10
            curl -f http://localhost:6333/healthz || exit 1
          "

    - name: Verify deployment
      run: |
        curl -f http://${{ steps.vm_info.outputs.vm_ip }}:6333/collections
        echo "✅ Qdrant deployed successfully at ${{ steps.vm_info.outputs.vm_ip }}"

Nota: El workflow real (130 líneas) es mucho más robusto que este ejemplo simplificado:

✅ Wait for VM ready: 30 retries con 10s entre cada uno
✅ Create separate directory: Crea ~/qdrant antes de .env
✅ Robust verification: 30 attempts con 15s sleeps para health check
✅ Connection info: Muestra API URL + Dashboard URL al final
Simplificado aquí para fines educativos

Proceso:

Get VM external IP
Wait for VM to be SSH-ready (30 retries)
Install Docker + Docker Compose (si no existe)
Create qdrant directory y .env with API key
Copy docker-compose.yml
docker-compose up -d
Verify health check (30 retries, 15s cada uno)
Display connection info (IP, API URL, Dashboard)

Resultado: Qdrant running en VM con persistencia en 5-10 minutos

🚀 Workflow 6: Update Batch Job Image

Archivo .github/workflows/update-batch-job-image.yml:

name: Update Batch Job Image

on:
  workflow_dispatch:

jobs:
  update-job:
    runs-on: ubuntu-latest

    steps:
    - name: Authenticate to GCP
      uses: google-github-actions/auth@v2
      with:
        credentials_json: ${{ secrets.GSA_KEY }}

    - name: Set up Cloud SDK
      uses: google-github-actions/setup-gcloud@v2

    - name: Update Cloud Run Job image
      run: |
        IMAGE="${{ secrets.DOCKER_HUB_USERNAME }}/${{ vars.DOCKER_IMAGE_NAME_PROCESSING }}:${{ vars.GCP_CLOUD_RUN_BATCH_IMAGE_TAG }}"

        echo "🔄 Updating job image to: ${IMAGE}"

        gcloud run jobs update ${{ vars.GCP_CLOUD_RUN_BATCH_JOB_NAME }} \
          --image=${IMAGE} \
          --region=${{ secrets.GCP_REGION }} \
          --project=${{ secrets.GCP_PROJECT_ID }}

        echo "✅ Job updated successfully!"

    - name: Verify update
      run: |
        gcloud run jobs describe ${{ vars.GCP_CLOUD_RUN_BATCH_JOB_NAME }} \
          --region=${{ secrets.GCP_REGION }} \
          --project=${{ secrets.GCP_PROJECT_ID }} \
          --format='value(template.template.containers[0].image)'

Nota: El workflow real (128 líneas) tiene validación y debugging avanzados:

✅ Args expansion: Expande ${GCP_BUCKET_NAME} con sed
✅ Format conversion: Space-separated → comma-separated args
✅ Error handling: job describe si update falla
✅ Full verification: Usa jq para parsear y verificar image + args
✅ Schedule display: Muestra próxima ejecución programada
Simplificado aquí para fines educativos

Uso:

Build nueva imagen → Push a Docker Hub (workflow 2)
Cambiar variable GCP_CLOUD_RUN_BATCH_IMAGE_TAG a nuevo tag
Ejecutar workflow 6 manualmente
Job actualizado y verificado en 2-3 minutos

🚀 Workflow 7: Update API Secrets & Deploy

Archivo .github/workflows/update-api-secrets-deploy.yml:

name: Update API Secrets & Deploy

on:
  workflow_dispatch:

jobs:
  update-and-deploy:
    runs-on: ubuntu-latest

    steps:
    - name: Checkout code
      uses: actions/checkout@v4

    - name: Authenticate to GCP
      uses: google-github-actions/auth@v2
      with:
        credentials_json: ${{ secrets.GSA_KEY }}

    - name: Set up Cloud SDK
      uses: google-github-actions/setup-gcloud@v2

    - name: Update Secrets in Secret Manager
      if: ${{ vars.GCP_CLOUD_SECRETS_UPDATE == 'true' }}
      run: |
        echo "📝 Updating secrets..."

        # Update .env secret
        echo "${{ secrets.API_ENV_FILE }}" | \
          gcloud secrets versions add ${{ vars.GCP_CLOUD_SECRETS_API_ENV_ID }} \
          --data-file=- \
          --project=${{ secrets.GCP_PROJECT_ID }}

        # Update JWT public key
        echo "${{ secrets.JWT_PUBLIC_KEY }}" | \
          gcloud secrets versions add ${{ vars.GCP_CLOUD_SECRETS_JWT_KEY_ID }} \
          --data-file=- \
          --project=${{ secrets.GCP_PROJECT_ID }}

        echo "✅ Secrets updated"

    - name: Deploy API to Cloud Run
      run: |
        IMAGE="${{ secrets.DOCKER_HUB_USERNAME }}/${{ vars.DOCKER_IMAGE_NAME_RAG_API }}:${{ vars.GCP_CLOUD_RUN_API_IMAGE_TAG }}"

        echo "🚀 Deploying ${IMAGE}..."

        gcloud run deploy ${{ vars.GCP_CLOUD_RUN_API_SERVICE_NAME }} \
          --image=${IMAGE} \
          --region=${{ secrets.GCP_REGION }} \
          --project=${{ secrets.GCP_PROJECT_ID }} \
          --port=8000 \
          --update-secrets="/app/secrets/env/.env=${{ vars.GCP_CLOUD_SECRETS_API_ENV_ID }}:latest,/app/secrets/jwt/public_key.pem=${{ vars.GCP_CLOUD_SECRETS_JWT_KEY_ID }}:latest" \
          --cpu=${{ vars.GCP_CLOUD_RUN_API_CPU }} \
          --memory=${{ vars.GCP_CLOUD_RUN_API_MEMORY }} \
          --min-instances=${{ vars.GCP_CLOUD_RUN_API_MIN_INSTANCES }} \
          --max-instances=${{ vars.GCP_CLOUD_RUN_API_MAX_INSTANCES }} \
          --timeout=300 \
          --allow-unauthenticated

        echo "✅ API deployed!"

    - name: Get Service URL
      run: |
        URL=$(gcloud run services describe ${{ vars.GCP_CLOUD_RUN_API_SERVICE_NAME }} \
          --region=${{ secrets.GCP_REGION }} \
          --format='value(status.url)' \
          --project=${{ secrets.GCP_PROJECT_ID }})

        echo "🌐 Service URL: ${URL}"
        echo "📋 Health Check: ${URL}/api/health"
        echo "📊 Swagger UI: ${URL}/docs"

Proceso completo:

(Opcional) Update secrets en Secret Manager
Deploy nueva versión de la imagen
Cloud Run hace rolling update (zero downtime)
Obtener URL del servicio
Mostrar health check y Swagger endpoints

🎯 Flujo Completo de CI/CD

Escenario: Nuevo Feature

Developer:
├─ 1. Crea branch: feature/add-reranking
├─ 2. Escribe código: reranking_service.py
├─ 3. Commit & push
│
GitHub Actions (automático):
├─ 4. [Workflow 1] Code Quality
│     ├─ Pre-commit ✅
│     ├─ Linting ✅
│     ├─ Security scan ✅
│     ├─ Type check ✅
│     └─ Tests ✅
│
Developer:
├─ 5. Create Pull Request
│
GitHub:
├─ 6. PR shows all checks passed ✅
│
Team:
├─ 7. Code review
├─ 8. Approve PR
├─ 9. Merge to main
│
GitHub Actions (automático):
├─ 10. [Workflow 3] Build API image
│      └─ Push: username/api:latest + :20241016
│
Developer:
├─ 11. Update variable: GCP_CLOUD_RUN_API_IMAGE_TAG=20241016
│
GitHub Actions (manual):
├─ 12. [Workflow 7] Deploy API
│      ├─ Update secrets (optional)
│      ├─ Deploy to Cloud Run
│      └─ Verify health check ✅
│
Production:
└─ 13. New feature live! 🎉

Total time: 15-20 minutos (automático)
Manual steps: 5 (branch, commit, PR, merge, trigger deploy)

🧪 Testing Local con Act

¿Qué es Act?

Act permite ejecutar GitHub Actions workflows localmente con Docker:

# Instalar act
brew install act  # macOS
# o
curl https://raw.githubusercontent.com/nektos/act/master/install.sh | sudo bash

Script Interactivo: act_menu.sh

El proyecto incluye un script sofisticado de 310 líneas para testing local de workflows con act.

Características clave (similar al tf_menu.sh de LLPY-11):

# Features del script real:

1. VALIDACIÓN DE VARIABLES REQUERIDAS
   - Define required vars por workflow
   - Valida antes de ejecutar
   - Busca en .env automáticamente

2. SECRETS VS VARIABLES
   - Lista de vars que son secrets
   - Usa -s para secrets, --var para variables
   - Oculta valores sensibles en output

3. DOCKER HUB LOGIN AUTOMÁTICO
   - Login antes de ejecutar workflows
   - Previene errores de pull/push

4. GCP CREDENTIALS HANDLING
   - Busca automáticamente .json en .gcpcredentials/
   - Convierte a formato act-compatible

5. MANEJO ESPECIAL CODE-QUALITY
   - Ejecuta jobs secuencialmente (pre-commit, lint, etc)
   - Previene conflictos de act en paralelo
   - Opción de continuar si job falla

6. CONFIRMACIÓN INTERACTIVA
   - Muestra comando con valores ocultos
   - Pide confirmación antes de ejecutar

7. ERROR HANDLING
   - Valida .env existe
   - Valida GCP credentials
   - Mensajes de error claros

Ejemplo simplificado de uso:

cd .github/workflows
chmod +x act_menu.sh
./act_menu.sh

# Menu interactivo:
# 1) Code Quality & Style Check
# 2) Build & Publish Docker Image (processing)
# 3) Build & Publish Docker Image (API)
# 4) Terraform Apply on .tf Change
# 5) Deploy Qdrant to VM
# 6) Update API Secrets & Deploy to Cloud Run
# 7) Update Batch Job Image
# 0) Salir

Por qué es tan complejo:

310 líneas: Validación robusta de ~50 variables diferentes
Secrets handling: 10+ secrets que deben manejarse correctamente
GCP integration: Búsqueda y formato de credentials
User experience: Mensajes claros, confirmaciones, error handling

Ventajas de Act:

✅ Test locally: Antes de push, verify workflow
✅ Faster iteration: No wait for GitHub runners
✅ Debug: Ver logs en tiempo real
✅ Cost: Zero (no consume GitHub Actions minutes)

🎯 Casos de Uso Reales

Para Continuous Integration:

"Quiero asegurar calidad de código en cada PR"

Solución:

Workflow 1 (Code Quality) se ejecuta automáticamente
PR no se puede mergear si hay fallos
Code review con confianza de que quality checks pasaron

Para Continuous Deployment:

"Quiero deploy automático a staging en cada merge a main"

Solución:

on:
  push:
    branches: [main]

jobs:
  deploy-staging:
    runs-on: ubuntu-latest
    steps:
      # Build image
      # Push to registry
      # Deploy to Cloud Run (staging environment)

Para Rollback Rápido:

"La última versión tiene un bug crítico"

Solución:

# 1. Cambiar IMAGE_TAG a versión anterior
GCP_CLOUD_RUN_API_IMAGE_TAG=20241015  # Era 20241016

# 2. Ejecutar workflow 7
# GitHub Actions → Deploy versión anterior

# 3. Verificar
curl https://api-url/api/health

# ✅ Rollback en 3 minutos

Para Multi-Environment:

"Quiero deploy a dev, staging, prod con diferentes configs"

Solución:

# Workflow con matrix strategy
jobs:
  deploy:
    strategy:
      matrix:
        environment: [dev, staging, prod]

    steps:
      - name: Deploy to ${{ matrix.environment }}
        run: |
          gcloud run deploy api-${{ matrix.environment }} \
            --image=... \
            --set-env-vars="ENV=${{ matrix.environment }}"

🚀 El Impacto Transformador

Antes de GitHub Actions:

⏱️ Deploy time: 1 hora manual
🐛 Error rate: 20-30% (typos, pasos olvidados)
📝 Tracking: Notas en Slack, memoria
🔄 Rollback: 30-60 minutos manual
💰 Cost: Jenkins server ($50-100/mes)
🧪 Testing: "Esperemos que funcione en prod"

Después de GitHub Actions:

⚡ Deploy time: 15-20 minutos automático
✅ Error rate: <5% (proceso automatizado)
📊 Tracking: Cada deploy en Git + GitHub UI
🔄 Rollback: 3 minutos (cambiar tag + trigger)
💰 Cost: $0 (free tier suficiente)
🧪 Testing: Tests obligatorios antes de merge

Métricas de Mejora:

Aspecto	Manual	Con GitHub Actions	Mejora
Deploy time	60 min	15 min	-75%
Error rate	25%	<5%	-80%
Deploy frequency	1-2/semana	10-20/semana	+900%
Rollback time	30-60 min	3 min	-95%
Time to production	Días	Horas	-90%

💡 Lecciones Aprendidas

1. Separate Workflows > Monolithic

7 workflows pequeños y específicos > 1 workflow gigante. Más fácil debug y maintain.

2. Secrets vs Variables

Secrets: API keys, passwords (encriptados)
Variables: Config pública (service names, regions)

3. Manual Triggers son Poder

workflow_dispatch permite control fino sobre deployments críticos.

4. Path Filters Ahorran Minutos

Solo ejecutar workflow si archivos relevantes cambiaron = menos tiempo desperdiciado.

5. Artifacts para Debugging

- uses: actions/upload-artifact@v4
  with:
    name: coverage-report
    path: htmlcov/

Útil para ver coverage reports, logs, etc.

6. Act para Desarrollo de Workflows

Desarrollar workflows con act localmente = 10x más rápido que push-wait-check en GitHub.

🎯 El Propósito Más Grande

GitHub Actions no es solo CI/CD - es calidad automatizada. Al automatizar:

✅ Quality: Tests obligatorios en cada cambio
🏗️ Build: Imágenes consistentes y versionadas
🚀 Deploy: Proceso reproducible y traceable
🔍 Visibility: Cada ejecución logged y auditable
⚡ Speed: De código a producción en minutos
🛡️ Safety: Quality gates previenen broken deploys
💰 Cost: Zero infrastructure, pay-per-use
👥 Collaboration: Workflows versionados = team aligned

Estamos convirtiendo el deployment de un proceso manual propenso a errores en un pipeline automatizado confiable que entrega valor a producción múltiples veces por día.

🔗 Recursos y Enlaces

Repositorio del Proyecto

GitHub: lus-laboris-py

Documentación Técnica

Workflows Folder: .github/workflows/
Code Quality: .github/workflows/code-quality.yml
Docker API Build: .github/workflows/docker-api-build-publish.yml
Terraform Apply: .github/workflows/terraform-apply-on-tf-change.yml
Deploy Qdrant: .github/workflows/deploy-qdrant.yml
Update API: .github/workflows/update-api-secrets-deploy.yml

Recursos Externos

GitHub Actions Docs: docs.github.com/actions
Act (Local Testing): github.com/nektos/act
Actions Marketplace: github.com/marketplace?type=actions
Workflow Syntax: docs.github.com/actions/reference/workflow-syntax

Próximo Post: LLPY-14 - Evaluación y Métricas de Calidad

En el siguiente y último post de la serie, exploraremos cómo evaluar y medir la calidad del sistema RAG end-to-end, incluyendo ground truth datasets, métricas de retrieval y generation, LLM-as-a-judge evaluations, y monitoreo continuo en producción.

LLPY-12: Docker y Containerización - De Desarrollo a Producción

Jesus Oviedo Riquelme — Fri, 17 Oct 2025 18:28:26 +0000

🎯 El Desafío de la Portabilidad y Reproducibilidad

Imagina este escenario familiar:

✅ Tu código funciona perfecto en tu laptop
❌ Falla en el servidor de staging
❌ Falla diferente en producción
❌ Nuevo desarrollador tarda 2 días en setup

El problema: "Works on my machine" 🤷

Requisitos de Deployment Moderno

🔄 Reproducibilidad: Mismo comportamiento en dev, staging, prod
📦 Portabilidad: Ejecuta en cualquier servidor Linux
🔒 Aislamiento: Dependencias no interfieren con host
⚡ Velocidad: Deploy en segundos, no horas
📊 Versionado: Cada deploy es traceable
↔️ Consistencia: Python 3.13, UV, dependencias exactas

Opciones para Deployment

Método	Pros	Contras	Reproducibilidad
Manual (pip install)	Simple	Dependencias conflictivas	❌ Ninguna
Virtual environments	Aislamiento Python	No aísla sistema	⚠️ Parcial
Conda environments	Multi-lenguaje	Pesado, lento	⚠️ Media
VM images	Aislamiento total	Pesado (GBs), lento	✅ Alta
Docker containers	Ligero, rápido, portable	Learning curve	✅✅ Muy Alta

Nuestra elección: Docker

📊 La Magnitud del Problema

Sin Containerización

Setup en servidor nuevo:

# 1. Instalar Python 3.13 (20 minutos)
sudo apt update
sudo apt install python3.13
# Error: "python3.13 not found in repository"
# Compilar desde source... (60 minutos)

# 2. Instalar UV (5 minutos)
curl -LsSf https://astral.sh/uv/install.sh | sh

# 3. Clonar repo y instalar deps (10 minutos)
git clone ...
cd project
uv sync
# Error: "Conflicto con paquete del sistema"

# 4. Configurar env vars (5 minutos)
# Copiar .env, ajustar paths...

# 5. Debugging de problemas (120 minutos)
# "¿Por qué no encuentra librerías?"
# "¿OpenSSL version incompatible?"

TOTAL: 3-4 horas + frustración

Con Docker:

docker run -p 8000:8000 --env-file .env username/lus-laboris-api:latest

TOTAL: 2 minutos

💡 La Solución: Docker Containers

¿Qué es Docker?

Docker es una plataforma de containerización que permite:

📦 Empaquetar aplicación + dependencias + runtime
🚀 Ejecutar de forma aislada en cualquier servidor
🔄 Distribuir via registries (Docker Hub, GCR)
📊 Versionar cada build con tags
⚡ Iniciar en segundos (vs minutos de VMs)

Container vs VM

┌─────────────────────────────────────────────┐
│         Virtual Machine Architecture        │
│                                             │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐      │
│  │  App 1  │ │  App 2  │ │  App 3  │      │
│  ├─────────┤ ├─────────┤ ├─────────┤      │
│  │ Python  │ │  Node   │ │  Java   │      │
│  ├─────────┤ ├─────────┤ ├─────────┤      │
│  │Guest OS │ │Guest OS │ │Guest OS │      │ Heavy!
│  │ (1GB+)  │ │ (1GB+)  │ │ (1GB+)  │      │
│  └─────────┘ └─────────┘ └─────────┘      │
│  ───────────────────────────────────       │
│           Hypervisor (VMware, KVM)         │
│  ───────────────────────────────────       │
│              Host OS (Linux)               │
│  ───────────────────────────────────       │
│                Hardware                    │
└─────────────────────────────────────────────┘

┌─────────────────────────────────────────────┐
│         Docker Container Architecture       │
│                                             │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐      │
│  │  App 1  │ │  App 2  │ │  App 3  │      │
│  │ +Python │ │ +Node   │ │ +Java   │      │
│  │ +Libs   │ │ +Libs   │ │ +Libs   │      │ Light!
│  │ (100MB) │ │ (80MB)  │ │ (150MB) │      │
│  └─────────┘ └─────────┘ └─────────┘      │
│  ───────────────────────────────────       │
│          Docker Engine (containerd)        │
│  ───────────────────────────────────       │
│              Host OS (Linux)               │
│  ───────────────────────────────────       │
│                Hardware                    │
└─────────────────────────────────────────────┘

Ventajas de Containers:
✅ 10-100x más ligero
✅ Inicia en segundos (vs minutos)
✅ Menos overhead de CPU/RAM
✅ Comparte kernel del host

🏗️ Dockerfiles del Proyecto

1. Dockerfile para Procesamiento (Batch)

Archivo src/processing/Dockerfile:

# Base image: Python 3.13 slim (Debian Bookworm)
FROM python:3.13.5-slim-bookworm

# Copy UV from official image
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

# Create non-root user
RUN useradd --create-home --shell /bin/bash appuser

# Set working directory
WORKDIR /app

# Environment variables
ENV PATH="/app/.venv/bin:$PATH"
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1

# Copy dependency files (leverage Docker cache)
COPY --chown=appuser:appuser pyproject.toml uv.lock ./

# Install dependencies
RUN uv sync --locked

# Copy source code
COPY --chown=appuser:appuser extract_law_text.py ./

# Switch to non-root user
USER appuser

# Entrypoint
ENTRYPOINT ["uv", "run", "extract_law_text.py"]

# CMD empty to allow flexible args
CMD []

Características clave:

✅ Multi-stage: Copia UV desde imagen oficial (no instalación)
✅ Layer caching: Dependencies primero, código después
✅ Non-root user: Seguridad (usuario appuser)
✅ UV locked: Dependencias exactas con uv.lock
✅ Flexible args: CMD vacío permite pasar argumentos

Build:

cd src/processing
docker build -t username/lus-laboris-processing:latest .

Run:

# Con argumentos personalizados
docker run username/lus-laboris-processing:latest --input data.pdf --output output.json

# Con volúmenes para datos
docker run -v $(pwd)/data:/app/data username/lus-laboris-processing:latest

2. Dockerfile para API (FastAPI)

Archivo src/lus_laboris_api/Dockerfile:

# Base image: Python 3.13 slim
FROM python:3.13.5-slim-bookworm

# Copy UV from official image
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/

# Install build dependencies and create user
RUN apt-get update && \
    apt-get install -y --no-install-recommends \
        build-essential \
        && rm -rf /var/lib/apt/lists/* \
        && useradd --create-home --shell /bin/bash apiuser

# Set working directory
WORKDIR /app

# Environment variables
ENV PATH="/app/.venv/bin:$PATH"
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1

# Copy dependency files (caching layer)
COPY --chown=apiuser:apiuser pyproject.toml uv.lock ./

# Install dependencies
RUN uv sync --locked

# Copy source code
COPY --chown=apiuser:apiuser api ./api

# Switch to non-root user
USER apiuser

# Expose port
EXPOSE 8000

# Entrypoint
ENTRYPOINT ["uv", "run", "uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"]

Diferencias con Dockerfile de procesamiento:

✅ Build tools: Incluye build-essential para compilar deps
✅ Port exposure: EXPOSE 8000 para Cloud Run
✅ Uvicorn entrypoint: Inicia servidor ASGI
✅ Cleanup: Elimina apt cache para reducir tamaño

Build:

cd src/lus_laboris_api
docker build -t username/lus-laboris-api:latest .

Run:

# Desarrollo local
docker run -p 8000:8000 --env-file .env username/lus-laboris-api:latest

# Acceder a API
curl http://localhost:8000/api/health

🐳 Docker Compose para la API

El archivo src/lus_laboris_api/docker-compose.yml del proyecto es simple y solo contiene la API:

services:
  legal-rag-api:
    image: rj24/legal-rag-api:latest
    container_name: legal-rag-api
    ports:
      - 8000:8000
    environment:
      API_ENV_FILE_PATH: /app/.env
    volumes:
      - ../../keys/public_key.pem:/app/api/keys/public_key.pem:ro
      - ../../.env:/app/.env:ro
    restart: unless-stopped

Características:

✅ Solo API: Qdrant y Phoenix se levantan con services/manage_services.sh
✅ Imagen pre-built: Usa imagen de Docker Hub (no build local)
✅ Monta secrets: JWT key y .env desde host
✅ Restart policy: Se reinicia si crash

Uso:

# PASO 1: Levantar servicios externos
cd services
./manage_services.sh  # Opción 's' → 'Todos los servicios'

# PASO 2: Levantar API
cd ../src/lus_laboris_api
docker-compose up -d

# Verificar servicios
curl http://localhost:6333/collections  # Qdrant (del paso 1)
curl http://localhost:6006              # Phoenix (del paso 1)
curl http://localhost:8000/api/health   # API

# Detener solo API
docker-compose down

¿Por qué separado?

✅ Flexibilidad: Puedes actualizar API sin tocar servicios
✅ Reutilización: Qdrant/Phoenix sirven múltiples aplicaciones
✅ Desarrollo: Puedes correr API local sin Docker, solo servicios en Docker

Opción Alternativa: Docker Compose Consolidado

Si prefieres levantar todo con un solo comando, puedes crear docker-compose.dev.yml en la raíz:

# docker-compose.dev.yml (en raíz del proyecto)
# Archivo opcional para desarrollo rápido

services:
  qdrant:
    image: qdrant/qdrant:latest
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrant_storage:/qdrant/storage

  phoenix:
    image: arizephoenix/phoenix:latest
    ports:
      - "6006:6006"
      - "4317:4317"
      - "9090:9090"

  api:
    image: rj24/legal-rag-api:latest
    ports:
      - "8000:8000"
    env_file:
      - .env
    volumes:
      - ./keys/public_key.pem:/app/api/keys/public_key.pem:ro
    depends_on:
      - qdrant
      - phoenix

volumes:
  qdrant_storage:

Uso:

# Levantar todo
docker-compose -f docker-compose.dev.yml up -d

# Detener todo
docker-compose -f docker-compose.dev.yml down

📦 Optimización de Imágenes Docker

1. Multi-Stage Builds (Opcional)

Aunque nuestro proyecto no usa multi-stage builds actualmente, es una técnica útil para reducir tamaño:

Concepto:

Stage 1 (Builder): Instala todas las build tools y dependencies
Stage 2 (Runtime): Copia solo el .venv compilado, sin build tools

Beneficio potencial: Imagen 40-50% más pequeña (elimina gcc, make, etc.)

Por qué no lo usamos:

python:3.13-slim ya es bastante pequeño (~330MB total)
UV hace el proceso muy rápido (no vale la pena la complejidad adicional)
Multi-stage agrega tiempo de build

2. .dockerignore

El archivo .dockerignore previene copiar archivos innecesarios al build:

Archivo src/lus_laboris_api/.dockerignore:

# Python
__pycache__/
*.pyc
*.pyo
*.pyd

# Virtual environments
.venv/
env/
venv/

# Data and output files
data/
*.db
*.json
*.csv

# Logs
*.log

# Git files
.git/
.gitignore

# Temporary Dockerfiles
Dockerfile.*

Beneficios:

✅ Build más rápido: No copia .venv, data/, .git (pueden ser GBs)
✅ Imagen más pequeña: Solo código fuente necesario
✅ Context más pequeño: Docker build context ~10MB vs 500MB+

Archivos importantes que NO se ignoran:

api/ (código fuente)
pyproject.toml (dependencies)
uv.lock (locked versions)

3. Layer Caching Strategy

# ❌ MAL: Dependencies después de código
COPY api ./api
COPY pyproject.toml uv.lock ./
RUN uv sync
# Problema: Cambio en api/ → re-install de TODAS las deps

# ✅ BIEN: Dependencies primero
COPY pyproject.toml uv.lock ./
RUN uv sync --locked
COPY api ./api
# Beneficio: Cambio en api/ → solo re-copy código (cache de deps)

Mejora: Build de 5 minutos → 30 segundos en iteraciones

🚀 Build y Publicación en Docker Hub

1. Script Automatizado

Archivo src/lus_laboris_api/docker_build_push.sh:

#!/bin/bash
set -e

# Load variables from .env
if [[ -f "../../.env" ]]; then
    set -o allexport
    source ../../.env
    set +o allexport
fi

# Validate required variables
if [[ -z "$DOCKER_HUB_USERNAME" || -z "$DOCKER_HUB_PASSWORD" || -z "$DOCKER_IMAGE_NAME_RAG_API" ]]; then
  echo "❌ ERROR: Variables requeridas no definidas"
  exit 1
fi

# Login to Docker Hub
echo "$DOCKER_HUB_PASSWORD" | docker login --username "$DOCKER_HUB_USERNAME" --password-stdin

# Define tags
DATE_TAG=$(date +%Y%m%d)
LATEST_TAG="latest"

# Build image
echo "🏗️  Building image..."
docker build -t "$DOCKER_HUB_USERNAME/$DOCKER_IMAGE_NAME_RAG_API:$DATE_TAG" .

# Tag as latest
docker tag "$DOCKER_HUB_USERNAME/$DOCKER_IMAGE_NAME_RAG_API:$DATE_TAG" \
           "$DOCKER_HUB_USERNAME/$DOCKER_IMAGE_NAME_RAG_API:$LATEST_TAG"

# Push both tags
echo "📤 Pushing to Docker Hub..."
docker push "$DOCKER_HUB_USERNAME/$DOCKER_IMAGE_NAME_RAG_API:$DATE_TAG"
docker push "$DOCKER_HUB_USERNAME/$DOCKER_IMAGE_NAME_RAG_API:$LATEST_TAG"

echo ""
echo "✅ Imágenes subidas a Docker Hub:"
echo "   $DOCKER_HUB_USERNAME/$DOCKER_IMAGE_NAME_RAG_API:$DATE_TAG"
echo "   $DOCKER_HUB_USERNAME/$DOCKER_IMAGE_NAME_RAG_API:$LATEST_TAG"

Uso:

cd src/lus_laboris_api
chmod +x docker_build_push.sh
./docker_build_push.sh

Output:

🏗️  Building image...
[+] Building 45.3s (12/12) FINISHED
 => [1/6] FROM python:3.13.5-slim-bookworm
 => [2/6] COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
 => [3/6] RUN useradd --create-home apiuser
 => [4/6] COPY pyproject.toml uv.lock ./
 => [5/6] RUN uv sync --locked
 => [6/6] COPY api ./api

📤 Pushing to Docker Hub...
The push refers to repository [docker.io/username/lus-laboris-api]
20241016: digest: sha256:abc123... size: 2415
latest: digest: sha256:abc123... size: 2415

✅ Imágenes subidas a Docker Hub:
   username/lus-laboris-api:20241016
   username/lus-laboris-api:latest

2. GitHub Actions para Build Automático

Archivo .github/workflows/docker-api-build-publish.yml:

name: Build & Publish Docker Image (API)

on:
  workflow_dispatch:  # Manual trigger
  push:
    paths:
      - 'src/lus_laboris_api/Dockerfile'
      - 'src/lus_laboris_api/pyproject.toml'
      - 'src/lus_laboris_api/uv.lock'
      - 'src/lus_laboris_api/api/**'

jobs:
  build-and-push:
    runs-on: ubuntu-latest

    steps:
    - name: Checkout code
      uses: actions/checkout@v5

    - name: Set up Docker Buildx
      uses: docker/setup-buildx-action@v3

    - name: Login to Docker Hub
      uses: docker/login-action@v3
      with:
        username: ${{ secrets.DOCKER_HUB_USERNAME }}
        password: ${{ secrets.DOCKER_HUB_PASSWORD }}

    - name: Get date tag
      id: date_tag
      run: echo "tag=$(date +'%Y%m%d')" >> $GITHUB_OUTPUT

    - name: Build and push
      uses: docker/build-push-action@v6
      with:
        context: ./src/lus_laboris_api
        push: true
        tags: |
          ${{ secrets.DOCKER_HUB_USERNAME }}/lus-laboris-api:latest
          ${{ secrets.DOCKER_HUB_USERNAME }}/lus-laboris-api:${{ steps.date_tag.outputs.tag }}
        cache-from: type=registry,ref=${{ secrets.DOCKER_HUB_USERNAME }}/lus-laboris-api:latest
        cache-to: type=inline

Ventajas:

✅ Auto-build: Push a GitHub → build automático
✅ Buildx: Multi-platform builds (amd64, arm64)
✅ Cache: Reutiliza layers de builds anteriores
✅ Dual tags: latest + fecha para rollback

🎯 Casos de Uso Reales

Para Desarrollo Local:

"Quiero desarrollo consistente sin instalar Python/UV"

Solución:

# Docker Compose levanta TODO
cd src/lus_laboris_api
docker-compose up -d

# API en http://localhost:8000
# Qdrant en http://localhost:6333
# Phoenix en http://localhost:6006

# Código cambia → rebuild solo API:
docker-compose up -d --build api

Para Onboarding de Nuevos Desarrolladores:

"Nuevo dev necesita setup en 5 minutos"

Solución:

# Paso 1: Clonar repo
git clone https://github.com/user/lus-laboris-py.git
cd lus-laboris-py

# Paso 2: Copiar .env
cp .env.example .env
# Editar con tus API keys

# Paso 3: Docker Compose
cd src/lus_laboris_api
docker-compose up -d

# Paso 4: Verificar
curl http://localhost:8000/docs

# ✅ DONE en 5 minutos

Para Deployment en Cloud Run:

"Necesito deployar en GCP"

Solución:

# Build y push
./docker_build_push.sh

# Deploy con gcloud
gcloud run deploy lus-laboris-api \
  --image username/lus-laboris-api:20241016 \
  --platform managed \
  --region us-central1

# ✅ API en producción en 3 minutos

Para Testing en CI/CD:

"Quiero correr tests en ambiente idéntico a producción"

Solución:

# .github/workflows/test.yml
jobs:
  test:
    runs-on: ubuntu-latest

    steps:
    - uses: actions/checkout@v4

    - name: Build test image
      run: docker build -t api:test .

    - name: Run tests in container
      run: docker run api:test pytest tests/

📊 Optimización de Imágenes

Tamaños de Imagen

Imagen	Base	Con deps	Final	Optimización
API (sin optimizar)	python:3.13 (1GB)	+500MB	1.5GB	❌
API (slim)	python:3.13-slim (150MB)	+180MB	330MB	✅ 78% reducción
API (alpine)	python:3.13-alpine (50MB)	+200MB	250MB	✅✅ 83% reducción
Processing	python:3.13-slim	+120MB	270MB	✅

Nuestra elección: python:3.13-slim

✅ Balance tamaño/compatibilidad
✅ Debian-based (mayor compatibilidad de deps)
✅ 78% más pequeño que python:3.13

Técnicas de Optimización

1. Cleanup de apt cache:

RUN apt-get update && \
    apt-get install -y build-essential && \
    rm -rf /var/lib/apt/lists/*  # ⬅️ Elimina 50-100MB

2. UV locked dependencies:

COPY pyproject.toml uv.lock ./
RUN uv sync --locked  # ⬅️ No descarga extras, exactas versiones

3. Non-root user:

RUN useradd apiuser
USER apiuser  # ⬅️ Seguridad + compatibilidad Cloud Run

4. Minimize layers:

# ❌ MAL: 3 layers
RUN apt-get update
RUN apt-get install -y build-essential
RUN rm -rf /var/lib/apt/lists/*

# ✅ BIEN: 1 layer
RUN apt-get update && \
    apt-get install -y build-essential && \
    rm -rf /var/lib/apt/lists/*

🚀 El Impacto Transformador

Antes de Docker:

⏱️ Setup time: 2-4 horas por servidor
🐛 "Works on my machine": Constante debugging de entorno
📦 Dependency hell: Conflictos con sistema
🔄 Inconsistencia: Dev ≠ Staging ≠ Prod
💰 Desperdicio: Múltiples VMs para aislamiento

Después de Docker:

⚡ Setup time: 2-5 minutos con Docker Compose
✅ Guaranteed consistency: Dev = Staging = Prod
📦 Dependency isolation: Zero conflictos
🔄 Perfect parity: Mismo container en todos lados
💰 Efficiency: Múltiples containers en 1 servidor

Métricas de Mejora:

Aspecto	Sin Docker	Con Docker	Mejora
Setup time	2-4 horas	2-5 minutos	-95%
Environment issues	Frecuentes	Raros	-90%
Deploy time	30-60 min	2-3 min	-95%
Rollback time	30 min	30 seg	-97%
Onboarding (new dev)	1-2 días	10 minutos	-99%

💡 Lecciones Aprendidas

1. Slim > Full > Alpine

Para Python, python:3.13-slim es el sweet spot. Alpine tiene problemas con deps compiladas.

2. Copy Dependencies First

Layer caching es crítico. Dependencies cambian poco, código cambia mucho.

3. Non-Root User es Requerido

Cloud Run y muchos orchestrators requieren non-root. Hazlo desde el principio.

4. UV en Docker es Magia

UV es 10-100x más rápido que pip. Copia desde imagen oficial = zero install time.

5. .dockerignore es Crítico

Sin .dockerignore, builds lentos y imágenes grandes (copia .venv, data/, etc.)

6. Non-Root User es Requerido

Cloud Run y otros orchestrators requieren containers que NO corran como root. Ambos Dockerfiles usan:

RUN useradd --create-home --shell /bin/bash apiuser
USER apiuser

Beneficios:

✅ Seguridad: Limita permisos del proceso
✅ Compatibilidad: Cloud Run, Kubernetes requieren non-root
✅ Best practice: Principio de mínimo privilegio

🎯 El Propósito Más Grande

Docker no es solo containerización - es portabilidad universal. Al empaquetar:

📦 Runtime: Python 3.13
📚 Dependencies: UV + todas las libs
⚙️ Configuration: Environment variables
🔧 Code: Aplicación completa

En un artifact inmutable, logramos:

✅ Build once, run anywhere: Dev, staging, prod, cloud, on-prem
✅ Versionado exacto: Cada imagen es traceable (tag + digest)
✅ Rollback instant: Cambiar tag en deploy = rollback
✅ Scaling horizontal: Múltiples containers de la misma imagen
✅ Testing aislado: Tests en container = env idéntico a prod
✅ Zero config drift: Configuración versionada en Dockerfile

Estamos eliminando la categoría completa de errores "funciona en mi máquina" y acelerando deployment de horas a minutos.

🔗 Recursos y Enlaces

Repositorio del Proyecto

GitHub: lus-laboris-py

Documentación Técnica

API Dockerfile: src/lus_laboris_api/Dockerfile
Processing Dockerfile: src/processing/Dockerfile
Docker Compose: src/lus_laboris_api/docker-compose.yml
Build Script: src/lus_laboris_api/docker_build_push.sh

Recursos Externos

Docker Docs: docs.docker.com
Docker Hub: hub.docker.com
Dockerfile Best Practices: docs.docker.com/develop/dev-best-practices
Docker Compose Docs: docs.docker.com/compose

Próximo Post: LLPY-13 - CI/CD con GitHub Actions

En el siguiente post exploraremos el sistema completo de CI/CD con 7 workflows automatizados: quality checks, Docker builds, Terraform apply, y deployments a GCP.

LLPY-11: Terraform - Infraestructura como Código

Jesus Oviedo Riquelme — Fri, 17 Oct 2025 18:27:29 +0000

🎯 El Desafío de Gestionar Infraestructura Cloud

Imagina que necesitas desplegar tu sistema RAG en GCP:

✅ VM para Qdrant (Compute Engine)
✅ API en Cloud Run (FastAPI)
✅ Batch Job (Cloud Run Job para procesamiento)
✅ Storage (Google Cloud Storage)
✅ Secrets (Secret Manager para .env y JWT keys)

El problema: ¿Cómo creas, actualizas y gestionas toda esta infraestructura de forma reproducible, versionada y colaborativa?

Opciones para Gestionar Infraestructura

Método	Pros	Contras	Reproducibilidad
Console UI (manual)	Fácil, visual	Propenso a errores, no versionado	❌ Ninguna
gcloud CLI scripts	Automatizado	Scripts frágiles, difícil rollback	⚠️ Limitada
Cloud Formation	IaC nativo AWS	Solo AWS	✅ Alta (AWS only)
Pulumi	Multiple lenguajes	Requiere runtime	✅ Alta
Terraform	Declarativo, multi-cloud	Learning curve	✅✅ Muy Alta

Nuestra elección: Terraform

📊 La Magnitud del Problema

Requisitos de Infraestructura

Para el proyecto Lus Laboris necesitamos:

🗄️ Google Cloud Storage
- Bucket para datos procesados
- Bucket para Terraform state (remoto)
🖥️ Compute Engine VM
- VM para Qdrant (vector database)
- Ubuntu 22.04 LTS
- Firewall rules (ports 6333, 6334, 22)
- SPOT instance (cost optimization)
🚀 Cloud Run Service
- FastAPI container
- Secrets montados de Secret Manager
- Auto-scaling (0-10 instances)
- 2 CPU, 2Gi RAM
⏰ Cloud Run Job
- Batch processing (scheduled)
- Cron schedule
- Logs y notificaciones
🔐 Secret Manager
- .env file (all API config)
- JWT public key
- IAM permissions

Desafíos Sin IaC

Sin Terraform:

Developer 1 crea VM via Console → Settings no documentados
Developer 2 necesita replicar → "¿Qué settings usaste?"
Developer 1: "Hmm... creo que era e2-medium con 20GB..."
Developer 2: "¿Y el firewall?"
Developer 1: "Olvidé... tendrás que experimentar"

Disaster Recovery: "La VM se borró, ¿cómo la recreo?"
Team: 🤷 "No hay backup de la configuración"

Con Terraform:

git clone repo
terraform apply
✅ Toda la infraestructura recreada en 5 minutos
✅ Configuración exacta versionada en Git
✅ Documentación viva en código
✅ Rollback a cualquier versión anterior

💡 La Solución: Terraform Infrastructure as Code

¿Qué es Terraform?

Terraform (por HashiCorp) es una herramienta de Infrastructure as Code (IaC) que permite:

📝 Definir infraestructura en archivos declarativos (HCL)
📊 Planear cambios antes de aplicarlos
🚀 Aplicar cambios de forma idempotente
🗑️ Destruir infraestructura limpiamente
🔄 Versionado en Git como código

Principios de Terraform

Declarativo: Describes el estado deseado, no los pasos
Idempotente: Ejecutar N veces = mismo resultado
Plan-Apply: Preview de cambios antes de ejecutar
State Management: Trackea estado actual vs deseado
Modular: Componentes reutilizables

🏗️ Arquitectura de Terraform en el Proyecto

Estructura de Carpetas

terraform/
├── main.tf                    # Orquestación de módulos
├── variables.tf               # Definición de variables
├── terraform.tfvars           # Valores de variables
├── providers.tf               # Configuración de GCP provider
├── tf_menu.sh                 # Script interactivo para ops comunes
├── README.md                  # Documentación
│
└── modules/                   # Módulos reutilizables
    ├── gcs/                   # Google Cloud Storage
    │   ├── main.tf
    │   ├── variables.tf
    │   └── outputs.tf
    │
    ├── compute_engine/        # VM para Qdrant
    │   ├── main.tf
    │   ├── variables.tf
    │   └── outputs.tf
    │
    ├── cloud_run_service/     # FastAPI en Cloud Run
    │   ├── main.tf
    │   ├── variables.tf
    │   └── outputs.tf
    │
    ├── cloud_run_job/         # Batch processing
    │   ├── main.tf
    │   ├── variables.tf
    │   └── outputs.tf
    │
    └── secret_manager/        # Secrets management
        ├── main.tf
        ├── variables.tf
        └── outputs.tf

Principio de diseño: Un módulo = un recurso lógico reutilizable

🚀 Implementación Paso a Paso

1. Configuración de Providers

Archivo terraform/providers.tf:

# Terraform Configuration
terraform {
  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "~> 7.1.1"
    }
  }

  backend "gcs" {
    bucket = "py-labor-law-rag-terraform-state"
    prefix = "terraform/state"
  }
}

# Provider de Google Cloud Platform
provider "google" {
  project = var.project_id
  region  = var.region
}

Características:

✅ Remote state: State almacenado en GCS (no local)
✅ Versión fijada: ~> 7.1.1 = 7.1.x (semver)
✅ State locking: GCS provee locking automático
✅ Team collaboration: Múltiples devs comparten state

2. Definición de Variables

Archivo terraform/variables.tf:

# Basic GCP Configuration
variable "project_id" {
  description = "GCP Project ID"
  type        = string
}

variable "region" {
  description = "GCP Region"
  type        = string
  default     = "us-central1"
}

variable "project_number" {
  description = "GCP Project Number (for service accounts)"
  type        = string
}

# Google Cloud Storage
variable "bucket_name" {
  description = "Name for the GCS bucket"
  type        = string
}

# Compute Engine (Qdrant VM)
variable "qdrant_vm_name" {
  description = "Name for the Qdrant VM"
  type        = string
  default     = "qdrant-vm"
}

variable "qdrant_vm_machine_type" {
  description = "Machine type for Qdrant VM"
  type        = string
  default     = "e2-medium"
}

variable "qdrant_vm_zone" {
  description = "Zone for Qdrant VM"
  type        = string
  default     = "us-central1-a"
}

variable "qdrant_vm_disk_size" {
  description = "Boot disk size in GB"
  type        = number
  default     = 20
}

# Cloud Run Service (API)
variable "api_service_name" {
  description = "Name of the Cloud Run service"
  type        = string
  default     = "lus-laboris-api"
}

variable "api_image" {
  description = "Docker image for the API"
  type        = string
}

variable "api_container_port" {
  description = "Container port for the API"
  type        = number
  default     = 8000
}

variable "api_cpu" {
  description = "Number of CPUs for the API"
  type        = string
  default     = "2"
}

variable "api_memory" {
  description = "Memory for the API (e.g., 2Gi)"
  type        = string
  default     = "2Gi"
}

variable "api_min_instance_count" {
  description = "Minimum number of instances"
  type        = number
  default     = 0  # Scale to zero for cost savings
}

variable "api_max_instance_count" {
  description = "Maximum number of instances"
  type        = number
  default     = 10
}

variable "api_timeout" {
  description = "Request timeout"
  type        = string
  default     = "300s"
}

# Secret Manager
variable "api_env_secret_id" {
  description = "Secret ID for .env file"
  type        = string
}

variable "jwt_public_key_secret_id" {
  description = "Secret ID for JWT public key"
  type        = string
}

# Cloud Run Job (Batch)
variable "job_name" {
  description = "Name of the Cloud Run Job"
  type        = string
}

variable "image" {
  description = "Docker image for the job"
  type        = string
}

variable "args" {
  description = "Arguments for the job"
  type        = list(string)
  default     = []
}

variable "schedule" {
  description = "Cron schedule for the job"
  type        = string
  default     = "0 2 * * *"  # 2 AM daily
}

variable "notify_email" {
  description = "Email for job notifications"
  type        = string
}

3. Módulo: Compute Engine (VM para Qdrant)

El módulo compute_engine crea la VM para Qdrant con configuración optimizada:

¿Qué crea?

VM Instance (google_compute_instance.qdrant_vm)
- OS: Ubuntu 22.04 LTS
- Machine type: e2-medium (2 vCPU, 4GB RAM)
- Disco: 20GB (configurable)
- Provisioning: SPOT (preemptible) ← 80% más barato
- Network: Default con IP pública efímera
- Tags: qdrant-server, http-server, https-server
Firewall Rule (google_compute_firewall.qdrant_firewall)
- Permite TCP en puertos: 6333 (HTTP), 6334 (gRPC), 22 (SSH)
- Source: 0.0.0.0/0 (⚠️ en producción, restringir)
- Target: VMs con tag qdrant-server
Outputs del módulo
- vm_external_ip: Para conectar desde API
- vm_internal_ip: Para comunicación interna en GCP
- vm_name y vm_zone: Para referencia

Ventajas de SPOT instances:

💰 80% más barato que VM regular
🔄 Preemptible: Puede interrumpirse (OK para Qdrant con persistencia)
✅ Ideal para workloads stateful con almacenamiento persistente

4. Módulo: Cloud Run Service (FastAPI)

El módulo cloud_run_service despliega la API en Cloud Run:

¿Qué crea?

Cloud Run Service (google_cloud_run_v2_service.api_service)
- Container: Docker image desde Docker Hub o GCR
- Puerto: 8000 (FastAPI)
- CPU: 2 cores
- Memoria: 2Gi
- Scaling: 0-10 instances (scale-to-zero habilitado)
- Timeout: 300s (5 minutos)
- Deletion protection: false (permite destroy fácil)
IAM Policy (acceso público)
- Role: roles/run.invoker
- Members: allUsers (cualquiera puede llamar la API)
Traffic Routing
- 100% del tráfico a latest revision
- Deployment strategy: Rolling update

Características clave:

✅ Scale to zero: Min instances = 0 → ahorro de costos cuando no hay tráfico
✅ Auto-scaling: Escala automáticamente según requests
✅ Serverless: No gestión de servers
✅ Pay-per-use: Solo pagas cuando hay requests

Nota sobre Secrets:
Los secrets (.env y JWT key) se montan vía GitHub Actions workflow, no directamente en Terraform. El módulo básico crea el servicio, y el workflow update-api-secrets-deploy.yml actualiza el servicio con los secrets.

5. Módulo: Secret Manager

El módulo secret_manager crea secrets en GCP Secret Manager:

¿Qué crea?

Secret para .env (google_secret_manager_secret.api_env_file)
- ID: Configurable (típicamente api-env-file)
- Replication: Automática (multi-region)
- Contenido: Se agrega después vía GitHub Actions
Secret para JWT public key (google_secret_manager_secret.jwt_public_key)
- ID: Configurable (típicamente jwt-public-key)
- Replication: Automática
- Contenido: Public key agregada vía GitHub Actions
IAM Permissions (para Cloud Run)
- Role: roles/secretmanager.secretAccessor
- Member: Service Account de Cloud Run
- Permite a Cloud Run leer los secrets

Flujo de trabajo:

Terraform crea los secret placeholders (vacíos)
GitHub Actions agrega el contenido real de los secrets
Cloud Run monta los secrets como archivos

Ventajas:

✅ Separación: Terraform gestiona estructura, GitHub Actions gestiona contenido sensible
✅ Seguridad: Secrets no versionados en código
✅ Rotación: Fácil actualizar secrets sin cambiar infraestructura

6. Main File (Orquestación)

Archivo terraform/main.tf:

# Google Cloud Storage
module "gcs" {
  source      = "./modules/gcs"
  project_id  = var.project_id
  region      = var.region
  bucket_name = var.bucket_name
}

# Compute Engine VM (Qdrant)
module "compute_engine" {
  source = "./modules/compute_engine"

  vm_name      = var.qdrant_vm_name
  machine_type = var.qdrant_vm_machine_type
  zone         = var.qdrant_vm_zone
  disk_size    = var.qdrant_vm_disk_size
}

# Secret Manager (MUST be created BEFORE Cloud Run)
module "secret_manager" {
  source = "./modules/secret_manager"

  project_id                = var.project_id
  cloud_run_service_account = var.project_number
  api_env_secret_id         = var.api_env_secret_id
  jwt_public_key_secret_id  = var.jwt_public_key_secret_id
}

# Cloud Run Service (API)
module "cloud_run_service" {
  source = "./modules/cloud_run_service"

  project_id         = var.project_id
  region             = var.region
  service_name       = var.api_service_name
  image              = var.api_image
  container_port     = var.api_container_port
  cpu                = var.api_cpu
  memory             = var.api_memory
  min_instance_count = var.api_min_instance_count
  max_instance_count = var.api_max_instance_count
  timeout            = var.api_timeout
  env_secret_name    = module.secret_manager.env_secret_name
  jwt_secret_name    = module.secret_manager.jwt_secret_name

  depends_on = [module.secret_manager]
}

# Cloud Run Job (Batch Processing)
module "cloud_run_job" {
  source       = "./modules/cloud_run_job"
  project_id   = var.project_id
  region       = var.region
  job_name     = var.job_name
  image        = var.image
  args         = var.args
  schedule     = var.schedule
  notify_email = var.notify_email
}

# Outputs
output "qdrant_vm_external_ip" {
  description = "External IP for Qdrant VM"
  value       = module.compute_engine.vm_external_ip
}

output "api_service_url" {
  description = "URL of the deployed API"
  value       = module.cloud_run_service.service_url
}

output "bucket_name" {
  description = "Name of the GCS bucket"
  value       = module.gcs.bucket_name
}

🔄 Workflow de Uso

1. Inicialización

# Navegar a directorio terraform
cd terraform

# Configurar credenciales
export GOOGLE_APPLICATION_CREDENTIALS="../.gcpcredentials/service-account.json"

# Inicializar Terraform (descargar providers, configurar backend)
terraform init

Output:

Initializing the backend...
Initializing modules...
Initializing provider plugins...
- Finding hashicorp/google versions matching "~> 6.0"...
- Installing hashicorp/google v6.8.0...

Terraform has been successfully initialized!

2. Plan (Preview de Cambios)

# Ver qué cambios se aplicarán
terraform plan

Output:

Terraform will perform the following actions:

  # module.compute_engine.google_compute_firewall.qdrant_firewall will be created
  + resource "google_compute_firewall" "qdrant_firewall" {
      + name    = "qdrant-vm-firewall"
      + network = "default"
      + ports   = ["6333", "6334", "22"]
    }

  # module.compute_engine.google_compute_instance.qdrant_vm will be created
  + resource "google_compute_instance" "qdrant_vm" {
      + name         = "qdrant-vm"
      + machine_type = "e2-medium"
      + zone         = "us-central1-a"
    }

  # module.cloud_run_service.google_cloud_run_v2_service.api_service will be created
  + resource "google_cloud_run_v2_service" "api_service" {
      + name     = "lus-laboris-api"
      + location = "us-central1"
    }

Plan: 15 to add, 0 to change, 0 to destroy.

3. Apply (Ejecutar Cambios)

# Aplicar cambios
terraform apply

# O sin confirmación (CI/CD):
terraform apply -auto-approve

Output:

module.secret_manager.google_secret_manager_secret.api_env_file: Creating...
module.secret_manager.google_secret_manager_secret.jwt_public_key: Creating...
module.compute_engine.google_compute_firewall.qdrant_firewall: Creating...
module.gcs.google_storage_bucket.bucket: Creating...
module.compute_engine.google_compute_instance.qdrant_vm: Creating...

...

Apply complete! Resources: 15 added, 0 changed, 0 destroyed.

Outputs:

api_service_url = "https://lus-laboris-api-abc123-uc.a.run.app"
bucket_name = "my-data-bucket"
qdrant_vm_external_ip = "34.123.45.67"

4. Verificación

# Ver outputs
terraform output

# Output específico
terraform output api_service_url

# Health check de la API
curl $(terraform output -raw api_service_url)/api/health

5. Destroy (Eliminar Infraestructura)

# Eliminar TODA la infraestructura
terraform destroy

# Con auto-approve (cuidado!)
terraform destroy -auto-approve

🎯 Script Interactivo: tf_menu.sh

El proyecto incluye un script bash sofisticado que automatiza operaciones Terraform:

¿Qué hace tf_menu.sh?

Setea credenciales GCP automáticamente
- Busca archivo JSON en .gcpcredentials/
- Exporta GOOGLE_APPLICATION_CREDENTIALS
Genera terraform.tfvars desde .env
- Lee variables de .env del proyecto
- Crea terraform.tfvars automáticamente
- Valida formato de variables (VM name, zone, etc.)
Valida ambiente antes de ejecutar
- Verifica que credenciales estén seteadas
- Verifica que terraform.tfvars exista
Ofrece menú interactivo

   ========= Terraform Menu =========
   1) Setear GOOGLE_APPLICATION_CREDENTIALS
   2) Crear archivo terraform.tfvars
   3) terraform init
   4) terraform plan
   5) terraform apply
   6) terraform destroy
   7) Salir

Uso del script:

cd terraform
chmod +x tf_menu.sh

# Modo interactivo
./tf_menu.sh

# Modo no-interactivo (para scripting)
./tf_menu.sh 1   # Setear credenciales
./tf_menu.sh 2   # Crear tfvars
./tf_menu.sh 3   # Init
./tf_menu.sh 4   # Plan
./tf_menu.sh 5   # Apply

Ventajas del script:

✅ Automatiza setup: Crea tfvars desde .env
✅ Validaciones: Previene errores comunes
✅ DRY: No duplicar variables entre .env y tfvars
✅ Developer-friendly: Menú en español, mensajes claros

🔄 Integración con CI/CD (GitHub Actions)

Archivo .github/workflows/terraform-apply-on-tf-change.yml:

name: Terraform Apply on TF Change

on:
  push:
    paths:
      - 'terraform/**/*.tf'
      - 'terraform/**/*.tfvars'

jobs:
  terraform:
    runs-on: ubuntu-latest
    env:
      GOOGLE_APPLICATION_CREDENTIALS: ${{ secrets.GSA_KEY }}

    steps:
    - name: Checkout code
      uses: actions/checkout@v4

    - name: Setup Terraform
      uses: hashicorp/setup-terraform@v3
      with:
        terraform_version: 1.9.0

    - name: Terraform Init
      working-directory: ./terraform
      run: terraform init

    - name: Terraform Validate
      working-directory: ./terraform
      run: terraform validate

    - name: Terraform Plan
      working-directory: ./terraform
      run: terraform plan -out=tfplan

    - name: Terraform Apply
      working-directory: ./terraform
      run: terraform apply -auto-approve tfplan

    - name: Show Outputs
      working-directory: ./terraform
      run: terraform output

Ventajas:

✅ Auto-apply: Cambios en .tf files → auto-deploy
✅ Validation: Validate antes de apply
✅ Plan saved: Plan guardado para apply exacto
✅ Outputs visible: Ver resultados en logs

🎯 Casos de Uso Reales

Para Replicar Ambiente:

"Necesito crear un ambiente de staging idéntico a producción"

Solución:

# Crear workspace para staging
terraform workspace new staging

# Aplicar con variables de staging
terraform apply -var-file="staging.tfvars"

# Resultado: Infraestructura idéntica en minutos

Para Disaster Recovery:

"La VM de Qdrant se eliminó accidentalmente"

Solución:

# Ver estado actual
terraform state list

# Recrear solo la VM
terraform apply -target=module.compute_engine

# Resultado: VM recreada con configuración exacta

Para Actualizar Recursos:

"Necesito aumentar RAM de la API de 2Gi a 4Gi"

Solución:

// terraform/terraform.tfvars
api_memory = "4Gi"  // Era "2Gi"

terraform apply

# Resultado: Cloud Run actualizado con zero downtime

Para Múltiples Ambientes:

"Quiero dev, staging, y prod con diferentes specs"

Solución:

# Crear workspaces
terraform workspace new dev
terraform workspace new staging
terraform workspace new prod

# Usar variables por workspace
terraform workspace select dev
terraform apply -var="api_cpu=1" -var="api_memory=1Gi"

terraform workspace select prod
terraform apply -var="api_cpu=4" -var="api_memory=8Gi"

🚀 El Impacto Transformador

Antes de Terraform:

🖱️ Manual deployment: Click en console por 30-60 minutos
📝 Documentación: Notas dispersas, desactualizadas
🐛 Errores: "Olvidé abrir el puerto 6334"
🔄 Replicación: Imposible crear ambiente idéntico
💥 Disaster recovery: Días para recrear infraestructura

Después de Terraform:

⚡ Auto deployment: terraform apply = 5-10 minutos
📝 Documentación viva: El código ES la documentación
✅ Sin errores: Configuración probada y versionada
🔄 Replicación perfecta: terraform apply en cualquier ambiente
💥 Disaster recovery: Minutos para recrear todo

Métricas de Mejora:

Aspecto	Sin IaC	Con Terraform	Mejora
Tiempo de setup	30-60 min	5-10 min	-80%
Errores de config	Frecuentes	Raros	-90%
Reproducibilidad	Imposible	Perfecta	+100%
Versionado	No	Sí (Git)	N/A
Rollback	Manual	`git revert` + apply	N/A
Team collaboration	Caos	Sincronizado	N/A

💡 Lecciones Aprendidas

1. Remote State es Obligatorio

Nunca uses state local para proyectos reales. GCS backend permite colaboración y previene conflictos.

2. Módulos = Reutilización

Un módulo bien diseñado se puede usar en múltiples proyectos con mínimos ajustes.

3. Variables con Defaults Sensatos

Defaults buenos = menos config required = menos errores.

4. Outputs son Documentación

Outputs muestran información crítica (IPs, URLs) que necesitas después del deploy.

5. depends_on para Orden

Algunos recursos deben crearse en orden (Secret Manager → Cloud Run). Usa depends_on explícitamente.

6. SPOT Instances = 80% Ahorro

Para workloads tolerantes a interrupciones (como Qdrant con persistencia), SPOT instances son oro.

🎯 El Propósito Más Grande

Terraform no es solo "automatización" - es infraestructura como producto. Al tratar infraestructura como código:

📝 Versionado: Cada cambio en Git con diff, blame, history
👥 Colaboración: Pull requests para cambios de infraestructura
🔍 Review: Code review de infraestructura antes de aplicar
🔄 Rollback: git revert + terraform apply = rollback instant
📚 Documentación: El código no miente, siempre está actualizado
🧪 Testing: Ambientes efímeros para testing
🚀 Velocity: De idea a producción en minutos, no horas

Estamos construyendo infraestructura con la misma calidad, rigor y velocidad que el código de aplicación.

🔗 Recursos y Enlaces

Repositorio del Proyecto

GitHub: lus-laboris-py

Documentación Técnica

Terraform Main: terraform/main.tf
Compute Engine Module: terraform/modules/compute_engine/
Cloud Run Module: terraform/modules/cloud_run_service/
Terraform README: terraform/README.md

Recursos Externos

Terraform Docs: terraform.io/docs
GCP Provider: registry.terraform.io/providers/hashicorp/google
Terraform Best Practices: terraform-best-practices.com
HashiCorp Learn: learn.hashicorp.com/terraform

Próximo Post: LLPY-12 - Docker y Containerización

En el siguiente post exploraremos cómo containerizar la aplicación con Docker, multi-stage builds, optimización de imágenes, Docker Compose para desarrollo local, y publicación en Docker Hub.

LLPY-09: Phoenix y OpenTelemetry - Observabilidad Completa

Jesus Oviedo Riquelme — Fri, 17 Oct 2025 18:25:56 +0000

🎯 El Desafío de Debugging en Sistemas RAG

Imagina que tu sistema RAG está en producción:

✅ API recibe 100+ req/s
✅ Pipeline complejo: Embedding → Qdrant → Reranking → LLM
✅ Múltiples servicios integrados

Pero hay un problema: ¿Cómo debuggeas cuando algo sale mal?

Preguntas sin Respuesta

Performance: "¿Por qué esta query tardó 5 segundos cuando debería tomar 2?"
Quality: "¿Por qué el LLM generó una respuesta incorrecta?"
Errors: "¿Dónde falló exactamente en el pipeline?"
Cost: "¿Cuántos tokens estoy consumiendo por día?"
User Experience: "¿Qué % de usuarios obtienen respuestas relevantes?"

Sin observabilidad: debugging a ciegas con print statements y logs dispersos.

La Complejidad del RAG Pipeline

User Query: "¿Cuántos días de vacaciones?"
   ↓
   ├─ [1] Embedding generation (30ms) ✅
   │   └─ Model: multilingual-e5-small, Dimensions: 384
   ↓
   ├─ [2] Qdrant search (35ms) ✅
   │   └─ Found 10 documents, scores: 0.85-0.91
   ↓
   ├─ [3] Reranking (20ms) ✅
   │   └─ Model: ms-marco-MiniLM, Top-5 kept
   ↓
   ├─ [4] LLM generation (1800ms) ⚠️ SLOW!
   │   └─ OpenAI GPT-3.5-turbo, 1500 tokens
   ↓
   └─ [5] Evaluation (async, 2500ms)
       └─ Relevance: 0.92, Hallucination: 0.05

TOTAL: 1885ms + eval async

¿Cómo sabes si cada paso funcionó correctamente?

📊 La Magnitud del Problema

Requisitos de Observabilidad para RAG

📊 Tracing: Visualizar cada paso del pipeline end-to-end
🔍 Instrumentation: Auto-track llamadas a LLM APIs (OpenAI, Gemini)
📈 Metrics: Latency, throughput, error rates por servicio
🎯 Evaluations: Calidad automática de respuestas (LLM-as-judge)
🔗 Context Propagation: Correlacionar logs/spans de una misma request
💰 Cost Tracking: Tokens consumidos, costo por query
🐛 Error Tracking: Stack traces con contexto completo
📊 Dashboards: Visualización en tiempo real

Desafíos Técnicos

🕸️ Distributed Tracing: Correlacionar operaciones asíncronas
🔄 Async Evaluation: No bloquear respuesta del usuario
📦 Complex Payloads: Serializar metadatos complejos para OTel
⚡ Low Overhead: Tracing no debe agregar >10ms de latency
🎯 Sampling: En producción, no trackear todo (costo + overhead)

💡 La Solución: Phoenix + OpenTelemetry

¿Qué es Phoenix?

Phoenix (por Arize AI) es una plataforma de observabilidad especializada en LLMs y RAG:

🎯 Purpose-built: Diseñado específicamente para aplicaciones LLM
📊 Tracing: Visualización de pipelines RAG completos
🤖 LLM-as-Judge: Evaluación automática de respuestas
📈 Analytics: Métricas de calidad, costo, performance
🔍 Debugging: Drill-down en queries problemáticas
💰 Open Source: Self-hosted o cloud (Phoenix Cloud)

¿Qué es OpenTelemetry?

OpenTelemetry es el estándar de observabilidad open source:

📏 Specification: Define cómo instrumentar aplicaciones
📦 SDKs: Librerías para Python, JS, Go, Java, etc.
🔌 Exporters: Envía datos a backends (Phoenix, Datadog, New Relic, etc.)
🎯 Vendor-neutral: No lock-in a un provider específico

Arquitectura Phoenix + OpenTelemetry

┌─────────────────────────────────────────────────────────┐
│                    Lus Laboris API                      │
│                                                         │
│  ┌──────────────────────────────────────────────┐     │
│  │  PhoenixMonitoringService                    │     │
│  │  - Tracer Provider (OpenTelemetry)           │     │
│  │  - Instrumentors (OpenAI auto-instrumented)  │     │
│  │  - Custom Spans (manual tracking)            │     │
│  └───────────────┬──────────────────────────────┘     │
│                  │                                      │
│          OpenTelemetry SDK                              │
│                  │                                      │
└──────────────────┼──────────────────────────────────────┘
                   │
                   │ gRPC (4317) o HTTP (6006)
                   │
           ┌───────▼────────┐
           │  Phoenix       │
           │  Collector     │
           │                │
           │  - Receives    │
           │    traces      │
           │  - Stores      │
           │    spans       │
           │  - Runs        │
           │    evals       │
           └───────┬────────┘
                   │
           ┌───────▼────────┐
           │  Phoenix UI    │
           │  (Port 6006)   │
           │                │
           │  - Traces      │
           │  - Metrics     │
           │  - Evals       │
           │  - Analytics   │
           └────────────────┘

🚀 Configuración Paso a Paso

1. Variables de Entorno

# .env

# Phoenix Configuration
API_PHOENIX_ENABLED=true
API_PHOENIX_ENDPOINT=http://localhost:6006      # Phoenix HTTP
API_PHOENIX_GRPC_ENDPOINT=localhost:4317        # Phoenix gRPC (más rápido)
API_PHOENIX_USE_GRPC=true                       # Preferir gRPC
API_PHOENIX_API_KEY=                             # Vacío para local, requerido para cloud
API_PHOENIX_PROJECT_NAME=lus-laboris-api

# Environment
API_ENVIRONMENT=development  # Options: development, production, testing

Configuraciones por ambiente:

Variable	Development	Production
`API_PHOENIX_ENABLED`	`true`	`true`
`API_PHOENIX_ENDPOINT`	`http://localhost:6006`	`https://app.phoenix.arize.com`
`API_PHOENIX_USE_GRPC`	`true`	`true`
`API_PHOENIX_API_KEY`	(vacío)	`phx_...` (cloud API key)
`API_ENVIRONMENT`	`development`	`production`

2. Gestión de Servicios con Bash Script

El proyecto separa los servicios externos de la API para mejor gestión:

Estructura:

services/
├── vectordb/
│   └── docker-compose.yml    # Qdrant (puertos 6333, 6334)
├── monitoring/
│   └── docker-compose.yml    # Phoenix (puertos 6006, 4317, 9090)
└── manage_services.sh         # Script interactivo de gestión

src/lus_laboris_api/
└── docker-compose.yml         # API (separada)

Script de gestión de servicios:

cd services

# Hacer ejecutable (primera vez)
chmod +x manage_services.sh

# Ejecutar script interactivo
./manage_services.sh

Menú interactivo:

================================
  Script de Gestión de Servicios
================================

Servicios disponibles:
  1. Qdrant (Base de Datos Vectorial)
  2. Phoenix (Monitoreo y Observabilidad)
  3. Todos los Servicios

Acciones:
  s) Iniciar servicio
  t) Detener servicio
  r) Reiniciar servicio
  l) Mostrar logs
  k) Mostrar estado
  q) Salir

Estado actual:
  ● Qdrant: Ejecutándose
  ● Phoenix: Ejecutándose

Inicio rápido (comandos directos):

# Iniciar Qdrant
cd services/vectordb
docker compose up -d

# Iniciar Phoenix
cd services/monitoring
docker compose up -d

# Verificar servicios
curl http://localhost:6333      # Qdrant
curl http://localhost:6006      # Phoenix

# Acceder a UIs
open http://localhost:6333/dashboard  # Qdrant Dashboard
open http://localhost:6006            # Phoenix UI

Ventajas de esta arquitectura:

✅ Separación de responsabilidades: Servicios independientes de la API
✅ Gestión simplificada: Script bash interactivo en español
✅ Reutilizable: Servicios pueden usarse con otras aplicaciones
✅ Desarrollo flexible: Levantar solo lo que necesitas

2.1. Opción: Docker Compose Consolidado (Sugerencia)

Para desarrollo local rápido, puedes crear un docker-compose.yml consolidado en la raíz del proyecto:

# docker-compose.dev.yml (en raíz del proyecto)
# Archivo opcional para levantar todo el stack en desarrollo

services:
  # Qdrant vector database
  qdrant:
    image: qdrant/qdrant:latest
    container_name: qdrant
    ports:
      - "6333:6333"  # HTTP API
      - "6334:6334"  # gRPC API
    volumes:
      - qdrant_storage:/qdrant/storage
    environment:
      - QDRANT__SERVICE__HTTP_PORT=6333
      - QDRANT__SERVICE__GRPC_PORT=6334
      - QDRANT__LOG_LEVEL=INFO
    networks:
      - dev-network

  # Phoenix observability
  phoenix:
    image: arizephoenix/phoenix:latest
    container_name: phoenix
    ports:
      - "6006:6006"  # HTTP UI
      - "4317:4317"  # gRPC collector
      - "9090:9090"  # Prometheus
    environment:
      - PHOENIX_WORKING_DIR=/mnt/data
    volumes:
      - phoenix_data:/mnt/data
    networks:
      - dev-network

  # Lus Laboris API (opcional)
  api:
    build:
      context: ./src/lus_laboris_api
      dockerfile: Dockerfile
    container_name: lus-laboris-api
    ports:
      - "8000:8000"
    environment:
      # Phoenix config
      - API_PHOENIX_ENABLED=true
      - API_PHOENIX_ENDPOINT=http://phoenix:6006
      - API_PHOENIX_GRPC_ENDPOINT=phoenix:4317
      - API_PHOENIX_USE_GRPC=true
      # Qdrant config
      - API_QDRANT_URL=http://qdrant:6333
      - API_QDRANT_PREFER_GRPC=true
      - API_QDRANT_GRPC_PORT=6334
    env_file:
      - .env
    depends_on:
      - qdrant
      - phoenix
    networks:
      - dev-network

volumes:
  qdrant_storage:
    driver: local
  phoenix_data:
    driver: local

networks:
  dev-network:
    driver: bridge

Uso del stack consolidado:

# Levantar todo el stack (Qdrant + Phoenix + API)
docker-compose -f docker-compose.dev.yml up -d

# Solo servicios externos (sin API)
docker-compose -f docker-compose.dev.yml up -d qdrant phoenix

# Ver logs
docker-compose -f docker-compose.dev.yml logs -f

# Detener todo
docker-compose -f docker-compose.dev.yml down

Verificar servicios:

# Qdrant
curl http://localhost:6333
open http://localhost:6333/dashboard

# Phoenix
curl http://localhost:6006
open http://localhost:6006

# API (si está incluida)
curl http://localhost:8000/health

¿Cuándo usar cada opción?

Escenario	Usar
Desarrollo diario	Script `manage_services.sh` (servicios) + run API local
Testing rápido	`docker-compose.dev.yml` (todo el stack)
Desarrollo API	Solo `services/` (Qdrant + Phoenix) + run API en IDE
Producción	Servicios separados en GCP (Cloud Run + Compute Engine)

Ventajas del docker-compose consolidado:

⚡ Un solo comando levanta todo
🔗 Networking automático entre servicios
🧹 Cleanup fácil con un solo down
🎯 Ideal para demos y onboarding

Desventajas:

🔄 Menos flexible (no puedes actualizar un servicio sin afectar otros)
💾 Más recursos (levanta todo siempre)
🐛 Debugging más complejo (logs mezclados)

3. Settings con Pydantic

# src/lus_laboris_api/api/config.py

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    """Application settings"""

    # Phoenix Monitoring Configuration
    api_phoenix_enabled: bool = True
    api_phoenix_endpoint: str | None = None
    api_phoenix_grpc_endpoint: str | None = "localhost:4317"
    api_phoenix_use_grpc: bool = True
    api_phoenix_api_key: str | None = None
    api_phoenix_project_name: str = "lus-laboris-api"

    # Environment Configuration
    api_environment: str = "development"  # development, production, testing

    class Config:
        env_file = ".env"
        case_sensitive = False

settings = Settings()

4. PhoenixMonitoringService

El servicio se inicializa al arrancar la aplicación y configura el stack completo de observabilidad:

¿Qué hace PhoenixMonitoringService?

Lee configuración de settings
- Project name, endpoints (gRPC/HTTP), API keys
- Detecta ambiente (development vs production)
Conecta a Phoenix con estrategia inteligente
- Prioridad 1: gRPC (2-3x más rápido que HTTP)
- Fallback: HTTP si gRPC no está disponible
- Graceful degradation: Si falla, API sigue funcionando
Registra tracer provider con phoenix.otel.register()
- project_name: Para separar proyectos en Phoenix UI
- auto_instrument=True: Tracking automático de librerías
- batch=True en producción: Agrupa spans para eficiencia
Auto-instrumenta OpenAI con OpenInference
- OpenAI: Auto-tracked (tokens, latency, errors)
- Gemini: Manual tracking (no hay instrumentor aún)
Crea tracer para tracking manual con OpenTelemetry

Configuración según ambiente:

Aspecto	Development	Production
Span Processor	SimpleSpanProcessor (inmediato)	BatchSpanProcessor (agrupa 512 spans)
Transport	gRPC o HTTP	gRPC preferido (performance)
Batch	`false` (debugging)	`true` (eficiencia)

Beneficios clave:

✅ gRPC: <2ms overhead vs 5-10ms con HTTP
✅ Auto-instrumentation: OpenAI tracked sin código extra
✅ Graceful: Si Phoenix falla, API no se rompe
✅ Environment-aware: Optimizado por ambiente

🔍 Tracking del Pipeline RAG

1. Session Management

def create_session(self, user_id: str | None = None) -> str:
    """Create a monitoring session for a user request"""
    session_id = str(uuid.uuid4())

    self.session_tracker[session_id] = {
        "session_id": session_id,
        "user_id": user_id,
        "start_time": datetime.now(),
        "actions": [],
        "llm_calls": [],
        "metrics": {},
    }

    logger.info(f"Created monitoring session: {session_id}")
    return session_id

def end_session(self, session_id: str) -> dict[str, Any]:
    """End session and calculate final metrics"""
    if session_id not in self.session_tracker:
        return {}

    session_data = self.session_tracker[session_id]
    session_data["end_time"] = datetime.now()
    session_data["duration"] = (
        session_data["end_time"] - session_data["start_time"]
    ).total_seconds()

    # Calculate final metrics
    session_metrics = self._calculate_session_metrics(session_data)
    session_data["final_metrics"] = session_metrics

    logger.info(f"Session {session_id} ended. Duration: {session_data['duration']:.2f}s")

    # Cleanup
    del self.session_tracker[session_id]

    return session_data

2. Track Embedding Generation

Este método registra el paso de generación de embeddings en el pipeline:

¿Qué registra?

Modelo usado (multilingual-e5-small)
Longitud del texto de entrada
Tiempo de generación (típicamente ~30ms)
Session ID para correlación

Cómo funciona:

Crea un span de OpenTelemetry llamado "embedding_generation"
Añade attributes (modelo, tiempo, longitud de texto)
Registra en session tracker local
Marca como Status.OK si todo sale bien

Visible en Phoenix UI: Como un span dentro del trace completo del request.

3. Track Vectorstore Search

Registra la búsqueda en Qdrant (vectorstore):

¿Qué registra?

Longitud de la query
Cantidad de documentos retornados (típicamente 5-10)
Tiempo de búsqueda (~30-35ms)

Visible en Phoenix: Span "vectorstore_search" que muestra performance de Qdrant.

4. Track Reranking

Registra el paso de reranking con cross-encoder:

¿Qué registra?

Cantidad de documentos rerankeados (10 → 5)
Tiempo de reranking (~20ms)

Visible en Phoenix: Span "document_reranking" que muestra overhead del reranking.

5. Track LLM Call

Registra llamadas a LLMs (OpenAI o Gemini):

¿Qué registra?

Provider y modelo (openai/gpt-3.5-turbo o gemini/gemini-1.5-flash)
Longitud de prompt y respuesta
Tokens consumidos (para cálculo de costos)
Métricas básicas de calidad (coherence, relevance, completeness)

Auto-instrumentation:

OpenAI: Tracked automáticamente por OpenInference (tokens, latency, errors)
Gemini: Tracking manual porque no hay instrumentor aún

Visible en Phoenix: Span "llm_call_openai_gpt-3.5-turbo" que domina el tiempo total (~1.8s de 2s total).

🎯 Integración con RAG Pipeline

En el RAGService

# src/lus_laboris_api/api/services/rag_service.py

from .phoenix_service import phoenix_service

async def answer_question(self, question: str, session_id: str | None = None):
    """Complete RAG pipeline with Phoenix tracking"""

    start_time = time.time()

    # 1. Create session
    if not session_id:
        session_id = phoenix_service.create_session()

    try:
        # 2. Retrieve documents (tracked internally)
        documents, retrieval_metadata = self._retrieve_documents(question, session_id)

        # 3. Generate response (tracked internally)
        answer = await self._generate_response(question, documents, session_id)

        # 4. Calculate total time
        processing_time = time.time() - start_time

        # 5. Enqueue evaluation (async, non-blocking)
        context = self._build_context(documents)
        evaluation_service.enqueue_evaluation(
            session_id=session_id,
            question=question,
            context=context,
            answer=answer,
            documents=documents,
            metadata={
                "processing_time": processing_time,
                "llm_provider": self.llm_provider,
                "llm_model": self.llm_model,
            }
        )

        # 6. Return response
        return {
            "success": True,
            "question": question,
            "answer": answer,
            "processing_time_seconds": round(processing_time, 3),
            "session_id": session_id,
            ...
        }

    except Exception as e:
        logger.exception(f"[{session_id}] Failed to answer question")
        return {
            "success": False,
            "error": str(e),
            "session_id": session_id,
        }
    finally:
        # End session
        phoenix_service.end_session(session_id)

Flujo completo:

Create session
Track embedding → Track Qdrant search → Track reranking → Track LLM
Enqueue async evaluation
Return response to user
End session (calculates metrics)

📊 Visualización en Phoenix UI

1. Acceder a Phoenix

# Local
open http://localhost:6006

# Cloud
open https://app.phoenix.arize.com

2. Traces View

Muestra todos los requests con breakdown detallado:

┌────────────────────────────────────────────────────────┐
│ Trace: session_abc123 (1.8s)                          │
│                                                        │
│ ├─ embedding_generation (30ms)                        │
│ │  ├─ model: multilingual-e5-small                   │
│ │  └─ text_length: 45                                │
│                                                        │
│ ├─ vectorstore_search (35ms)                          │
│ │  ├─ results_count: 10                              │
│ │  └─ search_time: 0.035s                            │
│                                                        │
│ ├─ document_reranking (20ms)                          │
│ │  ├─ documents_count: 10                            │
│ │  └─ reranking_time: 0.020s                         │
│                                                        │
│ └─ llm_call_openai_gpt-3.5-turbo (1.7s)               │
│    ├─ provider: openai                                │
│    ├─ model: gpt-3.5-turbo                           │
│    ├─ prompt_length: 1450                            │
│    ├─ response_length: 320                           │
│    ├─ quality.coherence: 0.85                        │
│    ├─ quality.relevance: 0.92                        │
│    └─ quality.completeness: 0.78                     │
│                                                        │
│ TOTAL: 1.815s                                          │
└────────────────────────────────────────────────────────┘

3. LLM Calls Dashboard

┌─────────────────────────────────────────────────────┐
│ LLM Calls Summary (Last 24h)                        │
│                                                     │
│ Total Calls: 1,234                                  │
│ OpenAI: 856 (69%)                                   │
│ Gemini: 378 (31%)                                   │
│                                                     │
│ Avg Latency: 1.8s                                   │
│ p50: 1.5s | p95: 2.8s | p99: 4.2s                  │
│                                                     │
│ Total Tokens: 2.3M                                  │
│ Input: 1.5M | Output: 800K                         │
│                                                     │
│ Estimated Cost: $3.45                               │
│ OpenAI: $2.80 | Gemini: $0.65                      │
└─────────────────────────────────────────────────────┘

4. Evaluations View

┌─────────────────────────────────────────────────────┐
│ Evaluation Results (Last 100 queries)              │
│                                                     │
│ Relevance (avg):    0.89 ████████████████░░ 89%   │
│ Correctness (avg):  0.85 ██████████████░░░░ 85%   │
│ Completeness (avg): 0.78 █████████████░░░░░ 78%   │
│ Hallucination rate:  5%  ░░░░░░░░░░░░░░░░░░  5%   │
│ Grounding (avg):    0.92 ████████████████▓░ 92%   │
│                                                     │
│ Top Issues:                                         │
│ - 8 queries with low completeness (<0.6)           │
│ - 3 potential hallucinations detected              │
│ - 2 queries with poor grounding (<0.7)             │
└─────────────────────────────────────────────────────┘

🤖 LLM-as-a-Judge Evaluations

Evaluation Service: Calidad Automática

El EvaluationService evalúa la calidad de cada respuesta usando un LLM como juez:

¿Cómo funciona?

Enqueue evaluation (no bloqueante)
- Usuario recibe respuesta inmediatamente
- Evaluación se ejecuta en background thread
Evalúa múltiples dimensiones con prompts específicos:
- Relevance: ¿La respuesta es relevante a la pregunta? (0-1)
- Correctness: ¿Es factualmente correcta según el contexto? (0-1)
- Hallucination: ¿Contiene info no presente en el contexto? (0-1)
- Grounding: Inverso de hallucination (1 - hallucination)
Envía resultados a Phoenix con phoenix_service.track_evaluation()
- Métricas visibles en Phoenix UI
- Historial completo para análisis de tendencias

Métricas típicas:

Relevance: ~0.89 (89%)
Correctness: ~0.85 (85%)
Hallucination: ~0.05 (5%)
Grounding: ~0.92 (92%)

Ventajas:

✅ Non-blocking: No impacta latencia del usuario
✅ Automated: Evalúa 100% de respuestas sin intervención humana
✅ Continuous: Detecta degradación de calidad en tiempo real
✅ Actionable: Identifica queries problemáticas automáticamente

🎯 Casos de Uso Reales

Para Debugging de Latency:

"¿Por qué esta query tardó 5 segundos?"

Phoenix UI:

Abrir Phoenix → Traces
Filtrar por processing_time > 5s
Ver breakdown:
- Embedding: 30ms ✅
- Qdrant: 35ms ✅
- Reranking: 20ms ✅
- LLM: 4.8s ⚠️ CULPABLE

Acción: LLM es el cuello de botella. Opciones:

Cambiar a Gemini 1.5 Flash (más rápido)
Reducir max_tokens
Implementar caching de respuestas

Para Debugging de Calidad:

"¿Por qué el LLM generó una respuesta incorrecta?"

Phoenix UI:

Abrir Phoenix → Evaluations
Filtrar por correctness < 0.6
Ver query específica:
- Question: "¿Cuántos días de preaviso?"
- Context: [5 documentos irrelevantes]
- Answer: "15 días" (incorrecto)
- Correctness: 0.4

Acción: El problema es retrieval, no LLM. Opciones:

Mejorar query embedding
Ajustar top_k o reranking
Revisar metadatos de documentos

Para Cost Optimization:

"¿Cuánto estoy gastando en OpenAI?"

Phoenix UI:

Abrir Phoenix → LLM Calls
Ver dashboard:
- Total tokens (24h): 2.3M
- GPT-3.5-turbo: $2.80/día
- GPT-4: $12.50/día (70 calls)

Acción: GPT-4 es muy costoso. Opciones:

Usar GPT-4 solo para queries complejas
Implementar clasificador (simple → GPT-3.5, complex → GPT-4)
Migrar a Gemini 1.5 Pro (60% más barato)

Para Quality Monitoring:

"¿Qué % de respuestas tienen hallucinations?"

Phoenix UI:

Abrir Phoenix → Evaluations
Ver dashboard:
- Hallucination rate: 5% (61 de 1,234 queries)

Acción: 5% es aceptable, pero revisar casos:

3 queries con hallucination score >0.8
Común cuando context es irrelevante
Agregar guardrail: "No tengo información suficiente"

🚀 El Impacto Transformador

Antes de Phoenix:

🐛 Debugging: print() statements en 10 archivos diferentes
⏱️ Performance: "Parece lento, pero no sé por qué"
💰 Cost: "Creo que gasto $50/mes, pero no estoy seguro"
🎯 Quality: "Los usuarios dicen que a veces está mal, pero no sé cuándo"
📊 Metrics: CSV files con logs dispersos

Después de Phoenix:

🐛 Debugging: Ver trace completo en Phoenix UI en 10 segundos
⏱️ Performance: "LLM tarda 1.8s avg, p99 es 4.2s"
💰 Cost: "$3.45/día, OpenAI usa $2.80, Gemini $0.65"
🎯 Quality: "89% relevance, 5% hallucination rate"
📊 Metrics: Dashboard en tiempo real con drill-down completo

Métricas de Mejora:

Aspecto	Sin Phoenix	Con Phoenix	Mejora
MTTR (Mean Time To Resolution)	2-4 horas	10-30 minutos	-85%
Quality Detection	Manual sampling (10%)	Automated (100%)	+900%
Cost Visibility	Weekly estimates	Real-time tracking	N/A
Performance Insights	Guessing	Precise breakdown	N/A

💡 Lecciones Aprendidas

1. gRPC es 2-3x Más Rápido que HTTP

Phoenix con gRPC tiene latency de <2ms para enviar spans. HTTP puede agregar 5-10ms.

2. Batch Processing en Producción es Crítico

En dev, usar SimpleSpanProcessor (inmediato). En prod, usar BatchSpanProcessor (agrupa 512 spans antes de enviar).

3. Session IDs son el Glue

Correlacionar todos los spans de una request con un session_id único permite drill-down completo en Phoenix.

4. Async Evaluation no Bloquea

Evaluar calidad en background (evaluationservice) no impacta latencia de respuesta al usuario.

5. Auto-Instrumentation > Manual

OpenInference auto-instrumenta OpenAI (tracks tokens, latency, errors). Gemini requiere tracking manual porque no tiene instrumentor aún.

6. Serialize Complex Metadata

OpenTelemetry solo acepta primitives (str, int, float, bool). Dicts/lists deben serializarse a JSON string.

🎯 El Propósito Más Grande

Phoenix + OpenTelemetry no es solo "nice to have" - es el sistema nervioso del RAG pipeline. Sin observabilidad:

Debugging es adivinanza
Performance es misterio
Quality es anécdota
Cost es estimación

Con Phoenix + OpenTelemetry:

✅ Visibility: Ver cada paso del pipeline en tiempo real
✅ Debuggability: Drill-down en queries problemáticas en segundos
✅ Quality: Evaluación automática de 100% de respuestas
✅ Optimization: Identificar cuellos de botella con precisión
✅ Cost Control: Track tokens y costos en tiempo real
✅ Continuous Improvement: Métricas históricas para optimización

Estamos convirtiendo un sistema RAG de "caja negra" a cristal transparente, donde cada operación es visible, medible, y optimizable.

🔗 Recursos y Enlaces

Repositorio del Proyecto

GitHub: lus-laboris-py

Documentación Técnica

Phoenix Service: src/lus_laboris_api/api/services/phoenix_service.py
Evaluation Service: src/lus_laboris_api/api/services/evaluation_service.py
RAG Service: src/lus_laboris_api/api/services/rag_service.py

Recursos Externos

Phoenix Docs: docs.arize.com/phoenix
OpenTelemetry Python: opentelemetry.io/docs/languages/python
OpenInference: github.com/Arize-ai/openinference
Phoenix Cloud: app.phoenix.arize.com

Próximo Post: LLPY-10 - Autenticación JWT con RSA

En el siguiente post exploraremos cómo implementar autenticación segura con JWT y claves RSA, generación de tokens, validación en endpoints, y integración con GCP Secret Manager.

LLPY-10: Autenticación JWT con RSA - Seguridad Stateless

Jesus Oviedo Riquelme — Fri, 17 Oct 2025 18:24:50 +0000

🎯 El Desafío de la Autenticación en APIs

Imagina que tu API RAG está en producción:

✅ Endpoints públicos: /api/health, /api/rag/ask
⚠️ Endpoints sensibles: /api/vectorstore/load, /api/status (detalles completos)
⚠️ Endpoints administrativos: /api/vectorstore/delete

El problema: ¿Cómo proteges los endpoints sensibles sin comprometer la performance o escalabilidad?

Requisitos de Autenticación

🔒 Seguro: Tokens no falsificables
⚡ Rápido: Validación en <5ms
📈 Escalable: Sin estado compartido entre instancias
🔄 Stateless: No requiere base de datos de sesiones
🌐 Estándar: Compatible con cualquier cliente HTTP
🔑 Rotable: Cambio de claves sin downtime
⏰ Temporal: Tokens con expiración automática

Opciones de Autenticación

Método	Pros	Contras	Escalabilidad
Session cookies	Familiar, fácil	Requiere DB/Redis	Limitada ⚠️
API Keys	Simple	No expira, difícil rotación	Media ⚠️
OAuth 2.0	Estándar, robusto	Complejo setup	Alta ✅
JWT (HS256)	Stateless, rápido	Secret compartido	Alta ✅
JWT (RS256)	Stateless, secure	Requiere key pair	Muy Alta ✅✅

Nuestra elección: JWT con RS256 (RSA)

📊 La Magnitud del Problema

Desafíos Técnicos

🔐 Generación de Claves: ¿Cómo generar par RSA seguro?
📝 Claims Management: ¿Qué información incluir en el token?
✅ Validación: ¿Cómo validar firma sin secret compartido?
🔄 Distribución: ¿Cómo distribuir public key a la API?
⏰ Expiración: ¿Cómo manejar tokens expirados?
🔁 Renovación: ¿Cómo implementar refresh tokens?
🚫 Revocación: ¿Cómo invalidar tokens antes de expiración?

JWT vs Traditional Sessions

Traditional Sessions:

Client → API → Check session in Redis/DB (20-50ms)
         ↓
       Response

Scaling: N servers = N connections to Redis
Cost: Redis instance + network latency
SPOF: Redis down = all auth fails

JWT (Stateless):

Client → API → Validate signature locally (<5ms)
         ↓
       Response

Scaling: N servers = 0 shared state
Cost: CPU for validation (negligible)
SPOF: None (cada server valida independientemente)

💡 La Solución: JWT con RSA (RS256)

¿Qué es JWT?

JWT (JSON Web Token) es un estándar (RFC 7519) para tokens de acceso:

Header: Algoritmo y tipo
Payload: Claims (datos del usuario)
Signature: Firma criptográfica

¿Por Qué RS256 (RSA)?

HS256 (HMAC) vs RS256 (RSA):

Aspecto	HS256	RS256
Algoritmo	HMAC + SHA256 (symmetric)	RSA + SHA256 (asymmetric)
Claves	1 secret compartido	Private + Public key
Seguridad	Secret debe estar en todos los servers	Solo public key en API
Compromiso	Si leak = regenerar tokens de todos	Si leak public key = no importa
Uso	Pequeña escala, single app	Multi-service, microservices
Nuestra elección	❌	✅ RS256

RS256 permite:

✅ Private key en un solo lugar (token generation service)
✅ Public key en múltiples APIs (stateless validation)
✅ Zero trust: APIs solo validan, nunca generan tokens

🚀 Implementación Paso a Paso

1. Generación de Claves RSA

El script utils/generate_jwt_keys.sh genera el par de claves RSA necesarias para JWT:

¿Qué hace el script?

Genera clave privada RSA de 2048 bits con openssl genrsa
Extrae clave pública de la privada con openssl rsa -pubout
Establece permisos seguros:
- Private key: 600 (solo owner puede leer/escribir)
- Public key: 644 (todos pueden leer)
Guarda en directorio keys/

Ejecutar:

cd utils
chmod +x generate_jwt_keys.sh
./generate_jwt_keys.sh

Output del script:

[INFO] Generando clave privada RSA de 2048 bits...
[SUCCESS] Clave privada generada: keys/private_key.pem
[INFO] Generando clave pública RSA...
[SUCCESS] Clave pública generada: keys/public_key.pem
[SUCCESS] Permisos de archivos configurados correctamente

Información de las claves generadas:
==================================
Clave Privada:
  Archivo: keys/private_key.pem
  Tamaño: 1.7K
  Permisos: -rw------- (600)

Clave Pública:
  Archivo: keys/public_key.pem
  Tamaño: 451B
  Permisos: -rw-r--r-- (644)

Notas importantes:
  - La clave privada debe mantenerse segura y no compartirse
  - La clave pública puede ser compartida para validación de tokens
  - NO versionar private_key.pem en Git

Estructura generada:

keys/
├── private_key.pem    # 🔒 Para generar tokens (NUNCA compartir)
└── public_key.pem     # 🔓 Para validar tokens (OK compartir)

Formato de las claves:

-----BEGIN RSA PRIVATE KEY-----
MIIEpAIBAAKCAQEA2Z7X3Y+9QvH5xK...
-----END RSA PRIVATE KEY-----

-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ...
-----END PUBLIC KEY-----

2. Generación de Tokens JWT

El script utils/generate_jwt_token.py usa las claves RSA para generar tokens:

¿Qué hace el script?

Carga la clave privada desde keys/private_key.pem
Construye payload con claims estándar:
- sub: username
- iss: "lus-laboris-api" (issuer)
- aud: "lus-laboris-client" (audience)
- exp: timestamp de expiración
- iat: timestamp de creación
Firma con RS256 usando la clave privada
Retorna token en formato JWT

Uso del script:

# Generar token (expira en 15 minutos por defecto)
python utils/generate_jwt_token.py --username admin@example.com

# Con expiración custom (1440 min = 24 horas)
python utils/generate_jwt_token.py --username admin@example.com --expiry 1440

# Con claims adicionales
python utils/generate_jwt_token.py --username admin@example.com \
  --claims '{"role": "admin", "permissions": ["read", "write"]}'

# Validar token
python utils/generate_jwt_token.py --validate --token "eyJhbGc..."

Output del script:

ℹ️  Clave privada cargada desde: keys/private_key.pem
✅ Token generado exitosamente para usuario: admin@example.com
ℹ️  Token expira en: 2025-10-18 15:30:00 UTC (15 minutos)

🔑 Token JWT Generado Exitosamente!
==================================================
eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWI...
==================================================

📖 Instrucciones de Uso:
1. Usa este token en el header Authorization:
   Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI...

2. Prueba el token con:
   python generate_jwt_token.py --validate --token "eyJhbGc..."

Estructura de un Token JWT:

El token tiene 3 partes separadas por .:

[Header].[Payload].[Signature]

Header (base64):
{
  "alg": "RS256",
  "typ": "JWT"
}

Payload (base64):
{
  "sub": "admin@example.com",      // Usuario
  "iss": "lus-laboris-api",        // Emisor
  "aud": "lus-laboris-client",     // Audiencia
  "exp": 1729271800,               // Expira (Unix timestamp)
  "iat": 1729268200                // Creado (Unix timestamp)
}

Signature:
RS256(header + "." + payload, private_key) // Firma RSA

Claims importantes:

sub: Identifica al usuario
iss/aud: Previenen uso del token en otros sistemas
exp: Auto-expiración (no requiere revocación manual)
iat: Para auditoría

3. Validación en FastAPI (JWTValidator)

La clase JWTValidator en jwt_handler.py se encarga de validar tokens:

¿Qué hace JWTValidator?

Carga la clave pública al inicializar
- Lee desde settings.api_jwt_public_key_path
- Típicamente: keys/public_key.pem
- Falla temprano si el archivo no existe
Método validate_token() - El corazón de la validación
- Decodifica el token usando la public key
- Valida signature con algoritmo RS256
- Verifica claims: exp, iat, iss, aud
- Retorna payload si todo es válido
- Lanza ValueError si algo falla
Método is_token_valid() - Helper no-exception
- Retorna True si válido, False si inválido
- No lanza excepciones (para checks silenciosos)

Validaciones automáticas:

✅ Expiration (exp): Token no expirado
✅ Issued At (iat): Token ya fue emitido (no futuro)
✅ Issuer (iss): Coincide con settings.api_jwt_iss
✅ Audience (aud): Coincide con settings.api_jwt_aud
✅ Signature: Firmado con la private key correcta

Errores manejados:

jwt.ExpiredSignatureError → "Token has expired"
jwt.InvalidAudienceError → "Invalid audience"
jwt.InvalidIssuerError → "Invalid issuer"
jwt.InvalidTokenError → "Invalid token"

Performance:

Validación completa: <5ms (CPU-bound, muy rápido)
No requiere llamadas externas (DB, Redis, etc.)

4. Security Dependencies para FastAPI

El archivo src/lus_laboris_api/api/auth/security.py proporciona las funciones de autenticación:

¿Qué contiene security.py?

SecurityManager class
- Centraliza la lógica de autenticación
- Valida tokens JWT
- Maneja errores de autenticación
get_current_user()
- Dependency que requiere JWT válido
- Extrae payload del token
- Retorna información del usuario
optional_auth()
- Dependency con autenticación opcional
- Retorna dict si hay token válido
- Retorna None si no hay token (no bloquea acceso)
require_admin()
- Dependency para operaciones administrativas
- Requiere token JWT válido
require_vectorstore_write() y require_vectorstore_read()
- Dependencies para operaciones del vectorstore
- Requieren token JWT válido

Uso en endpoints:

from ..auth.security import get_current_user, optional_auth, require_admin

# Endpoint protegido (requiere JWT)
@router.get("/protected")
async def protected_endpoint(user: dict = Depends(get_current_user)):
    return {"message": f"Hello {user['sub']}"}

# Endpoint con auth opcional
@router.get("/health/detailed")
async def detailed_health(user: dict | None = Depends(optional_auth)):
    if user:
        return {"status": "healthy", "details": "...", "user": user['sub']}
    else:
        return {"status": "healthy"}

# Endpoint admin (requiere JWT)
@router.delete("/collections/{name}")
async def delete_collection(name: str, user: dict = Depends(require_admin)):
    return {"deleted_by": user['sub']}

5. Swagger UI Integration

La autenticación JWT se integra automáticamente en Swagger UI gracias a HTTPBearer:

Flujo en Swagger:

Acceder a Swagger UI: http://localhost:8000/docs
Click en "Authorize" 🔓 (botón en la parte superior)
Ingresar token: Pegar el token JWT en el campo "Value"
Click "Authorize": El token se guarda en la sesión
Probar endpoints: Ahora puedes llamar endpoints protegidos

Características:

✅ Auto-detecta endpoints con Depends(get_current_user)
✅ Candado visible 🔒 en endpoints protegidos
✅ Botón unlock para ingresar token
✅ Persistencia: Token se mantiene mientras Swagger esté abierto
✅ Logout: Click en "Logout" para borrar token

Ejemplo visual en Swagger:

GET /api/rag/ask                   [Sin candado] ← Público
GET /api/health                    [Sin candado] ← Público
GET /api/status                    🔒 [Authorize] ← Protegido
DELETE /api/vectorstore/delete     🔒 [Authorize] ← Protegido

🔐 Integración con GCP Secret Manager

1. Almacenar Public Key en Secret Manager

# Crear secret con la public key
gcloud secrets create jwt-public-key \
    --data-file=keys/public_key.pem \
    --project=tu-proyecto-gcp \
    --replication-policy="automatic"

# Verificar
gcloud secrets versions list jwt-public-key --project=tu-proyecto-gcp

2. Montar Secret en Cloud Run

En el deployment de Cloud Run:

gcloud run deploy lus-laboris-api \
  --image=tu-imagen \
  --update-secrets="/app/secrets/jwt/public_key.pem=jwt-public-key:latest" \
  --set-env-vars="API_JWT_PUBLIC_KEY_PATH=/app/secrets/jwt/public_key.pem"

3. GitHub Actions Workflow

- name: Update JWT Public Key Secret
  run: |
    echo "${JWT_PUBLIC_KEY}" | \
      gcloud secrets versions add jwt-public-key \
      --data-file=- \
      --project=${GCP_PROJECT_ID}

    echo "✅ JWT public key secret updated"

🎯 Casos de Uso Reales

Para APIs Públicas con Endpoints Protegidos:

"Quiero que /health sea público pero /status sea protegido"

Solución:

@router.get("/health")
async def health():
    """Public - no auth"""
    return {"status": "healthy"}

@router.get("/status")
async def status(user: dict = Depends(get_current_user)):
    """Protected - JWT required"""
    return {
        "status": "healthy",
        "details": {...},  # Info sensible
        "requested_by": user["sub"]
    }

Para Integración con Aplicaciones Externas:

"Otra aplicación necesita cargar datos al vectorstore"

Solución:

# Generar token para la aplicación externa
python generate_jwt_token.py generate \
  --username external-app@company.com \
  --expiry 10080  # 7 días

# Compartir token (de forma segura)
# La app externa puede hacer requests:
curl -X POST http://api/vectorstore/load \
  -H "Authorization: Bearer eyJhbGc..." \
  -H "Content-Type: application/json" \
  -d '{"filename": "data.json"}'

Para Diferentes Niveles de Acceso:

"Admin puede delete, user puede read"

Solución:

def require_admin(user: dict = Depends(get_current_user)) -> dict:
    """Dependency que requiere rol admin"""
    if user.get("role") != "admin":
        raise HTTPException(status_code=403, detail="Admin access required")
    return user

@router.delete("/collections/{name}")
async def delete_collection(
    name: str,
    admin: dict = Depends(require_admin)
):
    """Solo admins pueden eliminar"""
    ...

Para Debugging y Testing:

"Quiero probar endpoints protegidos en desarrollo"

Solución:

# Generar token de desarrollo (expira en 24 horas)
python generate_jwt_token.py generate \
  --username dev@localhost \
  --expiry 1440 > dev_token.txt

# Usar en requests
export TOKEN=$(cat dev_token.txt)
curl -H "Authorization: Bearer $TOKEN" http://localhost:8000/api/status

🚀 El Impacto Transformador

Antes de JWT:

🔑 Session management: Redis/DB requerido para sesiones
📊 Scaling issues: N servers = N connections to session store
⏱️ Latency: 20-50ms para check session
💰 Cost: Redis instance ($50-100/mes)
🚨 SPOF: Redis down = all auth fails

Después de JWT (RS256):

🔑 Stateless: Cero dependencias externas
📊 Infinite scaling: N servers = 0 shared state
⏱️ Ultra-fast: <5ms validation (CPU-bound)
💰 Zero cost: CPU cost negligible
🚨 No SPOF: Cada server valida independientemente

Métricas de Mejora:

Aspecto	Sin JWT	Con JWT (RS256)	Mejora
Latency de validación	20-50ms	<5ms	-90%
External dependencies	Redis/DB	None	-100%
Cost (auth infra)	$50-100/mes	$0	-100%
Scaling complexity	High	Zero	N/A
SPOF risk	Yes	No	N/A

💡 Lecciones Aprendidas

1. RS256 > HS256 para Multi-Service

Si tu API se comunica con otros servicios, RS256 es mejor porque solo necesitas compartir la public key (no sensitive).

2. Expiry Time es Crítico

Corto (15-60 min): Más seguro, pero requiere refresh frecuente
Largo (24 horas+): Más convenient, pero mayor ventana de compromiso
Nuestro sweet spot: 60 minutos para API calls, 7 días para batch jobs

3. Claims Mínimos son Mejores

Solo incluir en el payload lo que realmente necesitas validar. No uses JWT como session storage.

4. Public Key en Secret Manager

Aunque la public key no es sensible, almacenarla en Secret Manager facilita rotación sin re-deploy.

5. `optional_auth` es Poderoso

Permite endpoints que dan más info si estás autenticado, pero no bloquean acceso público.

6. Swagger UI con JWT es UX++

El botón "Authorize" en Swagger UI hace testing de endpoints protegidos trivial.

🔐 Seguridad Best Practices

1. Never Version Private Key

# .gitignore
keys/private_key.pem
*.pem
!public_key.pem  # Public key is safe

2. Secure Key Permissions

chmod 600 keys/private_key.pem  # Owner read/write only
chmod 644 keys/public_key.pem   # Read for everyone (safe)

3. Rotate Keys Periodically

# Generate new key pair
./generate_jwt_keys.sh --force

# Deploy new public key to API
gcloud secrets versions add jwt-public-key --data-file=keys/public_key.pem

# Gradually phase out old tokens (based on expiry)

4. Use Short-Lived Tokens

# For interactive users: 15-60 minutes
token = generator.generate_token(username, expiry_minutes=60)

# For service accounts: 7 days max
token = generator.generate_token(username, expiry_minutes=10080)

5. Validate All Standard Claims

# Always verify: exp, iat, iss, aud
payload = jwt.decode(
    token,
    public_key,
    options={
        "verify_exp": True,
        "verify_iat": True,
        "verify_aud": True,
        "verify_iss": True,
    }
)

🎯 El Propósito Más Grande

JWT con RS256 no es solo autenticación - es libertad arquitectural. Al eliminar la necesidad de estado compartido:

🚀 Scaling: Cada instancia de la API es independiente
🌐 Multi-region: Tokens válidos en cualquier región
🔄 Microservices: Diferentes servicios validan con la misma public key
💰 Cost: Zero infrastructure para auth
⚡ Performance: Validación local ultra-rápida
🔒 Security: Public key leak = no problem

Estamos construyendo una API que escala infinitamente sin necesidad de coordinación entre instancias, sin single points of failure, y sin costo adicional de infraestructura.

🔗 Recursos y Enlaces

Repositorio del Proyecto

GitHub: lus-laboris-py

Documentación Técnica

JWT Handler: src/lus_laboris_api/api/auth/jwt_handler.py
Security Dependencies: src/lus_laboris_api/api/auth/security.py
Generate Keys Script: utils/generate_jwt_keys.sh
Generate Token Script: utils/generate_jwt_token.py

Recursos Externos

JWT.io: jwt.io - Decode y debug tokens
RFC 7519 (JWT): tools.ietf.org/html/rfc7519
PyJWT Docs: pyjwt.readthedocs.io
RSA Cryptography: cryptography.io

Próximo Post: LLPY-11 - Terraform: Infraestructura como Código

En el siguiente post exploraremos cómo gestionar toda la infraestructura de GCP con Terraform, desde VMs para Qdrant hasta Cloud Run para la API, con módulos reutilizables y CI/CD automatizado.