> 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.2_tools_system.md).

# Sistema de herramientas

El sistema de herramientas de Taína está compuesto por 8 funciones especializadas que permiten al agente acceder de manera granular a la información de servicios gubernamentales. Cada herramienta está diseñada para un propósito específico y utiliza la base de conocimiento dual (ChromaDB) para proporcionar respuestas precisas y contextualizadas.

## Arquitectura del sistema de herramientas

**Implementación**: `src/tools.py` - Todas las herramientas están implementadas en este archivo

### 1. Herramientas de búsqueda y descubrimiento

#### `list_services`

**Propósito**: Encuentra servicios gubernamentales por nombre, institución o palabra clave.

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

```python
@function_tool()
async def list_services(
    nombre: Optional[str] = None,
    institucion: Optional[str] = None,
    keyword: Optional[str] = None,
    limit: int = 5,
) -> str:
    """Find government services by name, institution, or keyword"""
    try:
        retriever = await _get_retriever()
        query = _build_service_query(nombre, institucion, keyword)
        docs = await retriever.ainvoke(query)
        return _format_service_results(docs, limit)
    except Exception as e:
        return f"Error al buscar servicios: {str(e)}"
```

**Parámetros**:

* `nombre`: Nombre del servicio (búsqueda parcial)
* `institucion`: Nombre de la institución
* `keyword`: Búsqueda de texto libre
* `limit`: Número máximo de resultados (por defecto: 5)

**Retorna**: JSON string con servicios encontrados

**Ejemplo de uso**:

```python
# Buscar por institución
result = await list_services(institucion="INTRANT", limit=3)

# Buscar por palabra clave
result = await list_services(keyword="licencia conducir", limit=5)

# Buscar por nombre específico
result = await list_services(nombre="renovación", limit=2)
```

#### `ask_knowledge_base`

**Propósito**: Realiza búsqueda semántica en la base de conocimiento.

```python
@function_tool()
async def ask_knowledge_base(question: str) -> str:
    """Ask a question to the knowledge base using semantic search"""
    try:
        retriever = await _get_retriever()
        docs = await retriever.ainvoke(question)
        return _format_knowledge_results(docs)
    except Exception as e:
        return f"Error al consultar la base de conocimiento: {str(e)}"
```

**Parámetros**:

* `question`: Pregunta en lenguaje natural

**Retorna**: Respuesta basada en la base de conocimiento

**Ejemplo de uso**:

```python
# Pregunta general
result = await ask_knowledge_base("¿Cómo puedo renovar mi licencia de conducir?")

# Pregunta específica
result = await ask_knowledge_base("¿Cuáles son los requisitos para obtener un pasaporte?")
```

### 2. Herramientas de detalles de servicios

#### `get_service_overview`

**Propósito**: Obtiene información general de un servicio específico.

```python
@function_tool()
async def get_service_overview(service_id: str) -> str:
    """Get general information about a specific service"""
    try:
        retriever = await _get_retriever()
        query = f"service_id:{service_id} tipo:servicio_principal"
        docs = await retriever.ainvoke(query)
        return _format_service_overview(docs)
    except Exception as e:
        return f"Error al obtener información del servicio: {str(e)}"
```

**Parámetros**:

* `service_id`: ID único del servicio

**Retorna**: Información general del servicio

**Ejemplo de uso**:

```python
# Obtener información del servicio 163
result = await get_service_overview("163")

# Obtener información del servicio 45
result = await get_service_overview("45")
```

#### `list_service_variations`

**Propósito**: Lista las variaciones disponibles de un servicio.

```python
@function_tool()
async def list_service_variations(service_id: str, limit: int = 3) -> str:
    """List available variations of a specific service"""
    try:
        retriever = await _get_retriever()
        query = f"service_id:{service_id} tipo:variacion_servicio"
        docs = await retriever.ainvoke(query)
        return _format_service_variations(docs, limit)
    except Exception as e:
        return f"Error al obtener variaciones del servicio: {str(e)}"
```

**Parámetros**:

* `service_id`: ID único del servicio
* `limit`: Número máximo de variaciones (por defecto: 3)

**Retorna**: Lista de variaciones del servicio

**Ejemplo de uso**:

```python
# Obtener variaciones del servicio 163
result = await list_service_variations("163", limit=5)

# Obtener variaciones del servicio 45
result = await list_service_variations("45", limit=3)
```

### 3. Herramientas de información de ubicación

#### `get_service_locations`

**Propósito**: Obtiene información de ubicaciones donde se puede acceder al servicio.

```python
@function_tool()
async def get_service_locations(service_id: str) -> str:
    """Get locations where the service is available"""
    try:
        retriever = await _get_retriever()
        query = f"service_id:{service_id} tipo:ubicaciones"
        docs = await retriever.ainvoke(query)
        return _format_service_locations(docs)
    except Exception as e:
        return f"Error al obtener ubicaciones del servicio: {str(e)}"
```

**Parámetros**:

* `service_id`: ID único del servicio

**Retorna**: Información de ubicaciones del servicio

**Ejemplo de uso**:

```python
# Obtener ubicaciones del servicio 163
result = await get_service_locations("163")

# Obtener ubicaciones del servicio 45
result = await get_service_locations("45")
```

### 4. Herramientas de canales digitales

#### `get_service_digital_channel`

**Propósito**: Obtiene información sobre canales digitales del servicio.

```python
@function_tool()
async def get_service_digital_channel(service_id: str) -> str:
    """Get digital channels information for the service"""
    try:
        retriever = await _get_retriever()
        query = f"service_id:{service_id} tipo:canal_digital"
        docs = await retriever.ainvoke(query)
        return _format_digital_channels(docs)
    except Exception as e:
        return f"Error al obtener canales digitales del servicio: {str(e)}"
```

**Parámetros**:

* `service_id`: ID único del servicio

**Retorna**: Información de canales digitales

**Ejemplo de uso**:

```python
# Obtener canales digitales del servicio 163
result = await get_service_digital_channel("163")

# Obtener canales digitales del servicio 45
result = await get_service_digital_channel("45")
```

### 5. Herramientas de contacto

#### `get_service_contact`

**Propósito**: Obtiene información de contacto del servicio.

```python
@function_tool()
async def get_service_contact(service_id: str) -> str:
    """Get contact information for the service"""
    try:
        retriever = await _get_retriever()
        query = f"service_id:{service_id} tipo:contacto"
        docs = await retriever.ainvoke(query)
        return _format_contact_info(docs)
    except Exception as e:
        return f"Error al obtener información de contacto: {str(e)}"
```

**Parámetros**:

* `service_id`: ID único del servicio

**Retorna**: Información de contacto del servicio

**Ejemplo de uso**:

```python
# Obtener información de contacto del servicio 163
result = await get_service_contact("163")

# Obtener información de contacto del servicio 45
result = await get_service_contact("45")
```

### 6. Herramientas de utilidad

#### `get_current_date`

**Propósito**: Obtiene la fecha actual para contexto temporal.

```python
@function_tool()
async def get_current_date() -> str:
    """Get the current date and time"""
    try:
        from datetime import datetime
        now = datetime.now()
        return f"Hoy es {now.strftime('%A, %d de %B de %Y')} y son las {now.strftime('%H:%M')}"
    except Exception as e:
        return f"Error al obtener la fecha actual: {str(e)}"
```

**Parámetros**: Ninguno

**Retorna**: Fecha y hora actual formateada

**Ejemplo de uso**:

```python
# Obtener fecha actual
result = await get_current_date()
# Retorna: "Hoy es lunes, 15 de enero de 2024 y son las 14:30"
```

#### `health_check`

**Propósito**: Verifica el estado del sistema y sus componentes.

```python
@function_tool()
async def health_check() -> str:
    """Check the health status of the system"""
    try:
        import json
        from datetime import datetime
        
        status = {
            "timestamp": datetime.now().isoformat(),
            "knowledge_base": "OK",
            "livekit": "OK",
            "gemini": "OK",
            "deepgram": "OK",
            "elevenlabs": "OK"
        }
        
        return json.dumps(status, indent=2)
    except Exception as e:
        return f"Error en health check: {str(e)}"
```

**Parámetros**: Ninguno

**Retorna**: JSON con estado del sistema

**Ejemplo de uso**:

```python
# Verificar estado del sistema
result = await health_check()
# Retorna JSON con estado de todos los componentes
```

## Flujo de trabajo de las herramientas

### 1. Proceso de búsqueda

1. El agente recibe la consulta del ciudadano y determina qué herramienta llamar.
2. `_get_retriever()` inicializa el acceso a ChromaDB y recupera los documentos relevantes.
3. La herramienta formatea los resultados en JSON estructurado para que el LLM genere la respuesta final.

### 2. Manejo de errores

```python
def _handle_tool_error(tool_name: str, error: Exception) -> str:
    """Maneja errores de herramientas de manera consistente"""
    error_messages = {
        "list_services": "No se pudo buscar servicios en este momento",
        "get_service_overview": "No se pudo obtener información del servicio",
        "list_service_variations": "No se pudo obtener variaciones del servicio",
        "get_service_locations": "No se pudo obtener ubicaciones del servicio",
        "get_service_digital_channel": "No se pudo obtener canales digitales",
        "get_service_contact": "No se pudo obtener información de contacto",
        "ask_knowledge_base": "No se pudo consultar la base de conocimiento",
        "get_current_date": "No se pudo obtener la fecha actual",
        "health_check": "No se pudo verificar el estado del sistema"
    }
    
    return error_messages.get(tool_name, f"Error en {tool_name}: {str(error)}")
```

## Configuración de herramientas

### 1. Configuración del retriever

```python
async def _get_retriever():
    """Obtiene el retriever de ChromaDB configurado"""
    try:
        from src.vectors.chroma_vector import get_chroma_load
        from src.embeddings.gemini_embedding import get_google_embeddings
        
        embeddings = get_google_embeddings()
        retriever = get_chroma_load(
            embeddings=embeddings,
            directory=os.getenv("CHROMA_DIRECTORY", "storage/chroma"),
            collection_name=os.getenv("CHROMA_COLLECTION_NAME", "servicios"),
            retriever_k=int(os.getenv("RETRIEVER_K", "3")),
            fetch_k=int(os.getenv("RETRIEVER_FETCH_K", "5"))
        )
        
        return retriever
    except Exception as e:
        raise Exception(f"Error al configurar retriever: {str(e)}")
```

### 2. Configuración de parámetros

```python
# Configuración por defecto
DEFAULT_CONFIG = {
    "retriever_k": 3,
    "fetch_k": 5,
    "max_results": 5,
    "timeout": 30,
    "enable_caching": True,
    "cache_ttl": 3600
}

# Variables de entorno
ENV_CONFIG = {
    "RETRIEVER_K": int(os.getenv("RETRIEVER_K", "3")),
    "RETRIEVER_FETCH_K": int(os.getenv("RETRIEVER_FETCH_K", "5")),
    "MAX_RESULTS": int(os.getenv("MAX_RESULTS", "5")),
    "TOOL_TIMEOUT": int(os.getenv("TOOL_TIMEOUT", "30")),
    "ENABLE_CACHING": os.getenv("ENABLE_CACHING", "true").lower() == "true",
    "CACHE_TTL": int(os.getenv("CACHE_TTL", "3600"))
}
```

## Validación de herramientas

### 1. Verificación asíncrona

```python
# scripts/validate_tools.py
import asyncio
from unittest.mock import AsyncMock, patch
from src.tools import get_service_overview, list_services


async def validate_list_services():
    """Confirma que list_services devuelve un payload serializable."""
    with patch("src.tools._get_retriever") as mock_retriever:
        mock_retriever.return_value.invoke.return_value = [
            AsyncMock(page_content="Servicio de ejemplo", metadata={"service_id": "163"})
        ]
        result = await list_services(limit=1)
        assert result, "La función no devolvió resultados"


async def validate_service_overview():
    """Confirma que get_service_overview estructura la respuesta correctamente."""
    with patch("src.tools._get_retriever") as mock_retriever:
        mock_retriever.return_value.invoke.return_value = [
            AsyncMock(page_content="Información del servicio", metadata={"service_id": "163"})
        ]
        result = await get_service_overview("163")
        assert "Información del servicio" in result


async def run_validations():
    await validate_list_services()
    await validate_service_overview()


if __name__ == "__main__":
    asyncio.run(run_validations())
```

### 2. Validación de flujo completo

```python
# scripts/validate_service_flow.py
import asyncio
from src.tools import (
    ask_knowledge_base,
    get_service_overview,
    list_services,
    list_service_variations,
)


async def validate_service_discovery():
    """Ejecuta el recorrido completo de descubrimiento."""
    services = await list_services(keyword="licencia", limit=1)
    assert services["count"] > 0, "No se encontraron servicios"

    service_id = services["results"][0]["service_id"]
    overview = await get_service_overview(service_id)
    assert overview, "Sin información general"

    variations = await list_service_variations(service_id)
    assert variations["count"] >= 0, "No se pudieron leer variaciones"

    kb_result = await ask_knowledge_base("Cómo renovar mi licencia de conducir")
    assert kb_result, "La base de conocimiento no respondió"


if __name__ == "__main__":
    asyncio.run(validate_service_discovery())
```

## Monitoreo y logging

### 1. Logging de herramientas

```python
import logging
from datetime import datetime

logger = logging.getLogger(__name__)

def _log_tool_usage(tool_name: str, parameters: dict, duration: float, success: bool):
    """Registra el uso de herramientas"""
    log_data = {
        "timestamp": datetime.now().isoformat(),
        "tool_name": tool_name,
        "parameters": parameters,
        "duration_ms": duration * 1000,
        "success": success
    }
    
    if success:
        logger.info(f"Tool {tool_name} executed successfully", extra=log_data)
    else:
        logger.error(f"Tool {tool_name} failed", extra=log_data)
```

### 2. Métricas de herramientas

```python
def _track_tool_metrics(tool_name: str, duration: float, success: bool):
    """Rastrea métricas de herramientas"""
    metrics = {
        "tool_name": tool_name,
        "duration": duration,
        "success": success,
        "timestamp": datetime.now().isoformat()
    }
    
    # En implementación real, esto se enviaría a un sistema de métricas
    print(f"Tool metrics: {metrics}")
```

## Próximos pasos

1. **Pipeline de Datos**: [Pipeline de Datos](/taina-agente-ia-ogtic/referencias-tecnicas/implementation/6.3_data_pipeline.md)
2. **Manejo de Errores**: [Manejo de Errores](/taina-agente-ia-ogtic/referencias-tecnicas/implementation/6.1_error_handling_and_recovery.md)
3. **Referencia de API**: [Referencia de API](/taina-agente-ia-ogtic/referencias-tecnicas/api/tools.md)
4. **Configuración**: [Configuración](/taina-agente-ia-ogtic/referencias-tecnicas/api/configuration.md)

## Recursos adicionales

* [LiveKit Agents Documentation](https://docs.livekit.io/agents/)
* [ChromaDB Documentation](https://docs.trychroma.com/)
* [Google Gemini API](https://ai.google.dev/docs)

***

¿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.2_tools_system.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.