> For the complete documentation index, see [llms.txt](https://public-intelligence.gitbook.io/taina-agente-ia-ogtic/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://public-intelligence.gitbook.io/taina-agente-ia-ogtic/referencias-tecnicas/implementation/6.1_error_handling_and_recovery.md).

# Manejo de errores y recuperación

El sistema Taína implementa un manejo de errores robusto con múltiples capas de recuperación automática, diseñado para mantener la disponibilidad del servicio incluso ante fallos de componentes individuales. El sistema utiliza patrones de producción probados para garantizar la estabilidad y confiabilidad.

## Patrones de manejo de errores

### 1. Manejo de errores en herramientas

#### Patrón: Try-Catch con Fallback Elegante

**Implementación**: `src/tools.py` - Función `ask_knowledge_base`

```python
@function_tool()
async def ask_knowledge_base(query: str) -> str:
    """
    Busca información en la base de conocimientos gubernamentales.
    """
    if not query or len(query.strip()) < 3:
        return ERROR_RESPONSES["search_error"]
    
    try:
        # Obtener retriever de forma segura
        retriever = await _get_retriever()
        if retriever is None:
            logger.warning("Retriever no disponible para query: %s", query[:50])
            return ERROR_RESPONSES["knowledge_base_unavailable"]

        # Buscar con timeout para evitar cuelgues
        search_task = asyncio.to_thread(retriever.invoke, query)
        results = await asyncio.wait_for(search_task, timeout=10.0)

        if not results:
            logger.info("Sin resultados para query: %s", query[:50])
            return ERROR_RESPONSES["no_results_found"]

        # Procesar resultados de forma segura
        contents = []
        for doc in results:
            if hasattr(doc, 'page_content') and doc.page_content:
                content = doc.page_content.strip()
                if content:
                    contents.append(content[:2000])  # Limitar a 2000 chars por doc
        
        if not contents:
            return ERROR_RESPONSES["no_results_found"]
            
        # Combinar contenidos con límite total
        combined = "\n\n".join(contents)
        final_content = combined[:8000]  # Límite total para evitar tokens excesivos
        
        return final_content

    except asyncio.TimeoutError:
        logger.warning("Timeout en búsqueda para query: %s", query[:50])
        return ERROR_RESPONSES["search_error"]
    except Exception as e:
        # Log del error técnico para debugging
        logger.error("Error en ask_knowledge_base para '%s': %s", query[:50], str(e))
        
        # Para el usuario, respuesta amigable
        return ERROR_RESPONSES["search_error"]
```

#### Respuestas de error estandarizadas

```python
ERROR_RESPONSES = {
    "knowledge_base_unavailable": (
        "Lo siento, temporalmente no puedo acceder a la base de conocimientos. "
        "Para obtener información actualizada sobre servicios gubernamentales, "
        "puedes contactar al asterisco 4 6 2 o visitar www.gob.do"
    ),
    "no_results_found": (
        "No encontré información específica sobre tu consulta en nuestra base de datos. "
        "Te recomiendo contactar directamente a la institución correspondiente o "
        "llamar al asterisco 4 6 2 para asistencia personalizada."
    ),
    "search_error": (
        "Hubo un inconveniente técnico al buscar la información. "
        "Por favor, intenta reformular tu pregunta o contacta al asterisco 4 6 2 para ayuda inmediata."
    )
}
```

### 2. Gestión de retriever con Circuit Breaker

#### Inicialización segura con Timeout

```python
async def _get_retriever(profile: Optional[str] = None) -> Optional[VectorStoreRetriever]:
    """
    Inicializa y retorna el retriever indicado de forma segura.
    Retorna None si la inicialización falla, evitando crashes.
    """
    profile_name = _resolve_profile(profile)

    if profile_name in _initialization_failed_profiles:
        return None

    async with _retriever_lock:
        if profile_name in _retrievers:
            return _retrievers[profile_name]

        config = _RETRIEVER_CONFIG.get(profile_name)
        if config is None:
            logger.error("No hay configuración para el perfil de retriever '%s'", profile_name)
            _initialization_failed_profiles.add(profile_name)
            return None

        try:
            logger.info("Inicializando retriever '%s' desde %s", profile_name, directory)

            def _load_components() -> VectorStoreRetriever:
                if not directory.exists():
                    raise FileNotFoundError(f"Directorio ChromaDB no encontrado: {directory}")

                embeddings = get_google_embeddings()
                if embeddings is None:
                    raise RuntimeError("No se pudieron inicializar los embeddings de Google")

                retriever = get_chroma_load(
                    embeddings=embeddings,
                    directory=directory,
                    collection_name=collection_name,
                    retriever_k=retriever_k,
                )

                test_results = retriever.invoke("test")
                if not test_results:
                    logger.warning("Retriever '%s' inicializado pero sin documentos", profile_name)

                return retriever

            retriever = await asyncio.wait_for(
                asyncio.to_thread(_load_components),
                timeout=30.0,
            )
            _retrievers[profile_name] = retriever
            logger.info("Retriever '%s' inicializado exitosamente.", profile_name)
        except asyncio.TimeoutError:
            logger.error("Timeout al inicializar retriever '%s'", profile_name)
            _initialization_failed_profiles.add(profile_name)
            return None
        except Exception as e:
            logger.error("Error inicializando retriever '%s': %s", profile_name, e)
            _initialization_failed_profiles.add(profile_name)
            if profile_name in _retrievers:
                del _retrievers[profile_name]
            return None

    return _retrievers.get(profile_name)
```

### 3. Monitoreo de salud del sistema

#### Health check automático

```python
async def health_check() -> dict:
    """
    Verifica el estado de salud de las herramientas.
    Útil para monitoreo en producción.
    """
    profiles = list_available_profiles()
    health_status = {
        "knowledge_base": "unknown",
        "retriever_initialized": {name: name in _retrievers for name in profiles},
        "initialization_failed": list(_initialization_failed_profiles),
        "active_profile": _active_profile,
        "timestamp": datetime.now().isoformat()
    }
    
    try:
        # Test básico del retriever
        retriever = await _get_retriever()
        if retriever is not None:
            test_results = await asyncio.wait_for(
                asyncio.to_thread(retriever.invoke, "test salud"), 
                timeout=5.0
            )
            health_status["knowledge_base"] = "healthy" if test_results else "empty"
        else:
            health_status["knowledge_base"] = "unavailable"
    
    except Exception as e:
        health_status["knowledge_base"] = f"error: {str(e)}"
        
    return health_status
```

#### Monitor de salud en background

```python
async def periodic_health_monitor(ctx: agents.JobContext):
    """Monitor de salud del sistema en background"""
    global _SYSTEM_HEALTH
    
    if not CONFIG["enable_health_checks"]:
        return
        
    try:
        while True:
            await asyncio.sleep(CONFIG["health_check_interval"])
            
            start_time = asyncio.get_event_loop().time()
            
            try:
                # Health check de herramientas
                health_status = await health_check()
                
                # Evaluar estado general
                overall_status = "healthy"
                if health_status.get("knowledge_base") in ["error", "unavailable"]:
                    overall_status = "degraded"
                    
                    # Auto-recovery si está habilitado
                    if CONFIG["enable_auto_recovery"]:
                        logger.info("Intentando auto-recovery de knowledge base...")
                        await reset_retriever()
                
                _SYSTEM_HEALTH = {
                    "status": overall_status,
                    "last_check": asyncio.get_event_loop().time(),
                    "components": health_status
                }
                
                duration = asyncio.get_event_loop().time() - start_time
                log_system_health("overall_system", overall_status, {
                    "check_duration": duration,
                    "components": health_status
                })
                
            except Exception as e:
                log_error_with_context(e, {"operation": "health_monitor"})
                _SYSTEM_HEALTH["status"] = "unknown"
            
    except asyncio.CancelledError:
        logger.info("Health monitor cancelado")
    except Exception as e:
        logger.error(f"Error crítico en health monitor: {e}")
```

### 4. Recuperación automática

#### Reset de retriever

```python
async def reset_retriever(profile: Optional[str] = None):
    """
    Fuerza un reset del retriever en caso de problemas.
    Solo para uso en situaciones de emergencia.
    """
    targets = [_resolve_profile(profile)] if profile else list(_RETRIEVER_CONFIG.keys())
    async with _retriever_lock:
        for name in targets:
            if name in _retrievers:
                del _retrievers[name]
            _initialization_failed_profiles.discard(name)
            logger.info("Retriever '%s' reseteado manualmente", name)
```

#### Precalentamiento con validación

```python
def prewarm_process(proc: JobProcess):
    """Precalentamiento robusto con validación completa"""
    proc_id = os.getpid()
    proc.userdata["identity"] = f"worker-{proc_id}"
    logger.info(f"[Worker Identity] Proceso marcado como {proc.userdata['identity']}")
    start_time = asyncio.get_event_loop().time()
    logger.info(f"Iniciando precalentamiento del proceso {proc}")
    
    try:
        # 1. Validar entorno
        env_validation = validate_environment()
        if not env_validation["valid"]:
            logger.error(f"Validación de entorno falló: {env_validation['errors']}")
            proc.userdata["prewarmed"] = False
            proc.userdata["prewarm_error"] = "Environment validation failed"
            return
        
        if env_validation["warnings"]:
            for warning in env_validation["warnings"]:
                logger.warning(warning)
        
        # 2. Cargar VAD model con timeout
        logger.info("Cargando modelo VAD...")
        try:
            vad_model = silero.VAD.load()
            proc.userdata["vad"] = vad_model
            logger.info("Modelo VAD cargado exitosamente")
        except Exception as e:
            logger.error(f"Error cargando modelo VAD: {e}")
            proc.userdata["vad"] = None
            if not CONFIG["enable_auto_recovery"]:
                proc.userdata["prewarmed"] = False
                return
        
        # 3. Cargar credenciales GCP
        logger.info("Cargando credenciales GCP...")
        gcp_credentials = load_gcp_credentials()
        proc.userdata["gcp_credentials"] = gcp_credentials
        proc.userdata["gcp_bucket"] = os.getenv("GCP_BUCKET", "")
        
        # 4. Marcar como precalentado
        proc.userdata["prewarmed"] = True
        proc.userdata["cleanup_enabled"] = True
        proc.userdata["prewarm_time"] = start_time
        proc.userdata["config"] = CONFIG.copy()
        
        duration = asyncio.get_event_loop().time() - start_time
        log_performance("process_prewarm", duration, success=True)
        logger.info(f"Precalentamiento completado exitosamente en {duration:.2f}s")
        
    except Exception as e:
        duration = asyncio.get_event_loop().time() - start_time
        log_performance("process_prewarm", duration, success=False)
        log_error_with_context(e, {"operation": "process_prewarm"})
        
        proc.userdata["prewarmed"] = False
        proc.userdata["prewarm_error"] = str(e)
        logger.error(f"Error durante precalentamiento: {e}")
```

### 6. Logging estructurado

#### Sistema de logging de producción

```python
class ProductionLogger:
    """Sistema de logging optimizado para producción"""
    
    def __init__(self, name: str = "TainaAssistant"):
        self.name = name
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.DEBUG)
        
        # Evitar duplicación de handlers
        if not self.logger.handlers:
            self._setup_handlers()
    
    def _setup_handlers(self):
        """Configurar handlers diferenciados para producción"""
        
        # Crear directorio de logs
        log_dir = Path("logs")
        log_dir.mkdir(parents=True, exist_ok=True)
        
        # Formato optimizado para producción
        detailed_formatter = logging.Formatter(
            '{"timestamp": "%(asctime)s", "level": "%(levelname)s", '
            '"module": "%(module)s", "function": "%(funcName)s", '
            '"line": %(lineno)d, "message": "%(message)s", "pid": %(process)d}'
        )
        
        # 1. Handler para errores críticos (solo ERROR y CRITICAL)
        error_handler = RotatingFileHandler(
            log_dir / "errors.log",
            maxBytes=50*1024*1024,  # 50MB
            backupCount=5,
            encoding='utf-8'
        )
        error_handler.setLevel(logging.ERROR)
        error_handler.setFormatter(detailed_formatter)
        
        # 2. Handler para logs de aplicación (INFO+)
        app_handler = RotatingFileHandler(
            log_dir / "application.log", 
            maxBytes=100*1024*1024,  # 100MB
            backupCount=10,
            encoding='utf-8'
        )
        app_handler.setLevel(logging.INFO)
        app_handler.setFormatter(detailed_formatter)
        
        # Agregar handlers
        self.logger.addHandler(error_handler)
        self.logger.addHandler(app_handler)
```

#### Funciones de logging especializadas

```python
def log_performance(operation: str, duration: float, **kwargs):
    """Log de métricas de performance"""
    data = {
        "type": "performance",
        "operation": operation,
        "duration_ms": round(duration * 1000, 2),
        **kwargs
    }
    logger.info(f"PERFORMANCE: {json.dumps(data, ensure_ascii=False)}")

def log_error_with_context(error: Exception, context: dict):
    """Log de errores con contexto estructurado"""
    error_data = {
        "type": "error",
        "error_type": type(error).__name__,
        "error_message": str(error),
        "context": context
    }
    logger.error(f"ERROR_CONTEXT: {json.dumps(error_data, ensure_ascii=False)}")

def log_system_health(component: str, status: str, metrics: dict = None):
    """Log de salud del sistema"""
    health_data = {
        "type": "system_health",
        "component": component,
        "status": status,
        "timestamp": datetime.now().isoformat(),
        "metrics": metrics or {}
    }
    logger.info(f"HEALTH: {json.dumps(health_data, ensure_ascii=False)}")
```

## Estrategias de recuperación

### 1. Recuperación gradual

1. **Nivel 1**: Reintento inmediato con timeout
2. **Nivel 2**: Fallback a modo degradado (respuestas predefinidas)
3. **Nivel 3**: Reset de componente (retriever)
4. **Nivel 4**: Reinicio completo del agente

### 2. Circuit breaker pattern

* Detección de fallos repetidos en APIs externas
* Aislamiento temporal de servicios problemáticos
* Recuperación automática cuando el servicio se estabiliza
* Manejo específico de cuotas agotadas (error 429)

### 3. Timeout management

```python
CONFIG = {
    "stt_timeout": int(os.getenv("STT_TIMEOUT", "10")),
    "llm_timeout": int(os.getenv("LLM_TIMEOUT", "15")),
    "tts_timeout": int(os.getenv("TTS_TIMEOUT", "20")),
    "max_retry_attempts": int(os.getenv("MAX_RETRY_ATTEMPTS", "3")),
}
```

## Configuración de errores

### Variables de entorno

```ini
# Timeouts
STT_TIMEOUT=10
LLM_TIMEOUT=15
TTS_TIMEOUT=20

# Recuperación
ENABLE_AUTO_RECOVERY=true
MAX_RETRY_ATTEMPTS=3
ENABLE_HEALTH_CHECKS=true
HEALTH_CHECK_INTERVAL=300

# Memoria
MAX_MEMORY_MB=2500
WARNING_MEMORY_MB=1800
```

### Configuración de reintentos

```python
# Backoff exponencial
RETRY_ATTEMPTS = 3
INITIAL_DELAY = 1.0
MAX_DELAY = 60
BACKOFF_MULTIPLIER = 2

# Timeout por operación
OPERATION_TIMEOUTS = {
    "retriever_init": 30.0,
    "search_query": 10.0,
    "health_check": 5.0
}
```

## Casos de uso comunes

### 1. Fallo de base de conocimiento

**Síntoma**: "Retriever not initialized" **Causa**: ChromaDB no disponible o directorio no encontrado **Solución**:

* Reinicializar retriever con timeout
* Fallback a respuestas predefinidas
* Auto-recovery si está habilitado

### 2. Exceso de memoria

**Síntoma**: Uso de memoria > 2500MB **Causa**: Acumulación de objetos en memoria **Solución**:

* Limpieza automática de recursos
* Garbage collection explícito
* Reinicio del agente si es necesario

### 3. Fallo de API Externa

**Síntoma**: Timeout o error HTTP **Causa**: Servicio externo no disponible **Solución**:

* Reintento con backoff exponencial
* Circuit breaker para cuotas agotadas
* Fallback a modo degradado

### 4. Fallo de LiveKit

**Síntoma**: Conexión perdida **Causa**: Problemas de red o LiveKit Cloud **Solución**:

* Reconexión automática
* Validación de estado antes de procesar
* Logging detallado para debugging

## Mejores prácticas

### 1. Error Handling Defensivo

* Siempre manejar excepciones con try-catch
* Proporcionar fallbacks apropiados para el usuario
* Logging detallado para debugging técnico
* Respuestas amigables para el usuario final

### 2. Recuperación automática

* Monitoreo proactivo de salud del sistema
* Recuperación sin intervención manual
* Notificaciones para problemas críticos
* Circuit breaker para servicios externos

### 3. Observabilidad

* Logs estructurados en JSON
* Métricas de performance y errores
* Health checks automáticos
* Contexto detallado en logs de error

### 4. Configuración flexible

* Variables de entorno para todos los timeouts
* Configuración de reintentos
* Habilitación/deshabilitación de características
* Diferentes niveles de logging por ambiente

## Herramientas de debugging

### 1. Health Check Endpoint

```python
# Verificar estado del sistema
health_status = await health_check()
print(json.dumps(health_status, indent=2))
```

### 2. Logs de sistema

```bash
# Ver logs de errores
tail -f logs/errors.log

# Ver logs de aplicación
tail -f logs/application.log

# Ver logs del sistema
journalctl -u taina-agent -f
```

### 3. Monitoreo de memoria

```bash
# Ver uso de memoria
ps aux | grep python

# Ver logs de performance
grep "PERFORMANCE" logs/application.log
```

### 4. Debugging de retriever

```python
# Listar perfiles disponibles
profiles = list_available_profiles()
print(profiles)

# Reset manual de retriever
await reset_retriever()

# Verificar estado de inicialización
health = await health_check()
print(health["retriever_initialized"])
```

## Recursos adicionales

* [LiveKit Error Handling](https://docs.livekit.io/agents/error-handling/)
* [Python Logging Best Practices](https://docs.python.org/3/howto/logging.html)
* [Circuit Breaker Pattern](https://martinfowler.com/bliki/CircuitBreaker.html)

***

¿Necesitas ayuda? Consulta la [guía de solución de problemas](https://github.com/public-intelligence/taina_ogtic/blob/master/taina-gitbook-ogtic/taina-asistente-ia/index/how-to/troubleshoot.md) o la [documentación de configuración](/taina-agente-ia-ogtic/referencias-tecnicas/api/configuration.md).


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://public-intelligence.gitbook.io/taina-agente-ia-ogtic/referencias-tecnicas/implementation/6.1_error_handling_and_recovery.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.