> 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/arquitectura-y-conceptos/architecture/knowledge-base.md).

# Base de conocimiento

Esta guía explica la arquitectura dual de ChromaDB en Taína, incluyendo la estructura de datos, estrategias de recuperación y optimizaciones.

Taína utiliza una arquitectura dual de ChromaDB para optimizar la recuperación de información:

* **Base Principal (Chunks)**: Documentos granulares de servicios gubernamentales
* **Base QA**: Pares pregunta-respuesta generados automáticamente
* **Metadatos Enriquecidos**: Información estructurada para filtrado preciso
* **Recuperación Híbrida**: Combinación de búsqueda semántica y filtrado por metadatos

## Arquitectura dual

**Implementación**: `src/vectors/chroma_vector.py` - Gestión de ChromaDB dual

### Estructura de directorios

```
storage/
├── chroma/                    # Base principal (chunks)
│   ├── chroma.sqlite3        # Base de datos SQLite
│   ├── chroma.sqlite3-wal    # Write-ahead log
│   └── chroma.sqlite3-shm    # Shared memory
└── chroma_qa/                # Base QA
    ├── chroma.sqlite3        # Base de datos SQLite
    ├── chroma.sqlite3-wal    # Write-ahead log
    └── chroma.sqlite3-shm    # Shared memory
```

### Configuración de colecciones

**Implementación**: `src/vectors/chroma_vector.py` - Configuración dual

```python
# src/vectors/chroma_vector.py - Configuración dual
_RETRIEVER_CONFIG = {
    "chunks": {
        "directory": Path("storage/chroma"),
        "collection": "servicios",
        "retriever_k": 3,
    },
    "qa": {
        "directory": Path("storage/chroma_qa"),
        "collection": "servicios_qa",
        "retriever_k": 3,
    },
}
```

### Gestión de retrievers

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

```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 = profile or "chunks"
    
    # Verificar si ya está inicializado
    if profile_name in _retrievers:
        return _retrievers[profile_name]
    
    # Verificar si falló anteriormente
    if profile_name in _initialization_failed_profiles:
        return None
    
    async with _retriever_lock:
        # Doble verificación
        if profile_name in _retrievers:
            return _retrievers[profile_name]
        
        try:
            # Inicializar retriever
            retriever = await _initialize_retriever(profile_name)
            _retrievers[profile_name] = retriever
            return retriever
        except Exception as e:
            logger.error("Error inicializando retriever '%s': %s", profile_name, e)
            _initialization_failed_profiles.add(profile_name)
            return None
```

## Base principal (Chunks)

### Tipos de documentos

La base principal contiene 5 tipos de documentos especializados:

#### 1. `servicio_principal` - Información general

```json
{
  "id": "servicio_163",
  "content": "Servicio: Renovación Licencia de Conducir\nInstitución: INTRANT\nDescripción: Es la renovación de licencias de conducir vencidas...\nObjetivo: Realizar renovación de licencias de conducir...\nCanales disponibles: Presencial, Digital",
  "metadata": {
    "service_id": "163",
    "tipo": "servicio_principal",
    "institucion": "INTRANT",
    "nombre_servicio": "Renovación Licencia de Conducir",
    "placeholder": false
  }
}
```

#### 2. `variacion_servicio` - Categorías específicas

```json
{
  "id": "variacion_163_0",
  "content": "Servicio: Renovación Licencia de Conducir\nVariación: Renovación Licencia Categoría 01\nInstitución: INTRANT\nPrecio: RD$ 1900.00\nProcedimiento:\n1. Presentarse en oficinas autorizadas\n2. Pasar por Registro\n3. Realizar Evaluación Médica",
  "metadata": {
    "service_id": "163",
    "tipo": "variacion_servicio",
    "categoria": "Renovación Licencia Categoría 01",
    "precio": 1900.0,
    "precio_minimo": 1000.0,
    "precio_maximo": 2500.0,
    "moneda": "DOP",
    "concepto_de_pago": "Licencia de Conducir Categoría 01",
    "placeholder": false
  }
}
```

#### 3. `ubicaciones` - Oficinas y horarios

```json
{
  "id": "ubicaciones_163",
  "content": "Oficinas y ubicaciones disponibles:\n• Licencias de Conducir Sede Principal\n  Dirección: Av. Tiradentes esq. Héctor Homero Hernandez Vargas #7\n  Santo Domingo de Guzmán, Distrito Nacional\n  Horario: 08:00:00 - 16:00:00",
  "metadata": {
    "service_id": "163",
    "tipo": "ubicaciones",
    "total_oficinas": 15,
    "placeholder": false
  }
}
```

#### 4. `canal_digital` - Canales online

```json
{
  "id": "digital_163",
  "content": "Canal Digital: Portal WEB INTRANT\nEnlace: https://intrant.gob.do/servicios/licencias",
  "metadata": {
    "service_id": "163",
    "tipo": "canal_digital",
    "url": "https://intrant.gob.do/servicios/licencias",
    "placeholder": false
  }
}
```

#### 5. `contacto` - Información de contacto

```json
{
  "id": "contacto_163",
  "content": "Portal: https://www.intrant.gob.do\nTeléfonos: 8093386134\nCorreos: info@intrant.gob.do",
  "metadata": {
    "service_id": "163",
    "tipo": "contacto",
    "telefonos": ["8093386134"],
    "correos": ["info@intrant.gob.do"],
    "placeholder": false
  }
}
```

### Metadatos estructurados

```python
# Estructura de metadatos para filtrado
metadata_schema = {
    "service_id": str,           # ID único del servicio
    "tipo": str,                 # Tipo de documento
    "institucion": str,          # Institución responsable
    "nombre_servicio": str,      # Nombre del servicio
    "placeholder": bool,        # Si es placeholder
    "precio": float,            # Precio (opcional)
    "moneda": str,              # Moneda (opcional)
    "url": str,                 # URL (opcional)
    "telefonos": list,          # Teléfonos (opcional)
    "correos": list             # Correos (opcional)
}
```

## Base QA (Pregunta-Respuesta)

### Estructura de QA

```json
{
  "id": "qa_163_0",
  "content": "¿En qué consiste el servicio Renovación Licencia de Conducir?",
  "metadata": {
    "service_id": "163",
    "tipo": "servicio_principal",
    "intent": "descripcion_general",
    "source_chunk_id": "servicio_163",
    "placeholder": false
  }
}
```

### Tipos de intenciones

```python
# Tipos de intenciones en QA
intent_types = {
    "descripcion_general": "Descripción general del servicio",
    "institucion_responsable": "Institución que maneja el servicio",
    "costo_servicio": "Información sobre costos",
    "requisitos_servicio": "Requisitos y procedimientos",
    "ubicaciones_servicio": "Ubicaciones y horarios",
    "canal_digital": "Canales digitales disponibles",
    "contacto_servicio": "Información de contacto"
}
```

## Estrategias de recuperación

### Recuperación híbrida

```python
# src/vectors/chroma_vector.py - Recuperación híbrida
def get_hybrid_retriever(embeddings, directory, collection_name, retriever_k=3):
    """Crea retriever híbrido con filtrado por metadatos"""
    
    # Cargar base de datos vectorial
    vectorstore = Chroma(
        collection_name=collection_name,
        embedding_function=embeddings,
        persist_directory=directory
    )
    
    # Crear retriever con filtros
    retriever = VectorStoreRetriever(
        vectorstore=vectorstore,
        search_type="similarity",
        search_kwargs={
            "k": retriever_k,
            "filter": {
                "placeholder": False  # Excluir placeholders
            }
        }
    )
    
    return retriever
```

### Filtrado por metadatos

```python
# Ejemplos de filtrado por metadatos
filters = {
    # Filtrar por institución
    "institucion": "INTRANT",
    
    # Filtrar por tipo de documento
    "tipo": "variacion_servicio",
    
    # Filtrar por rango de precio
    "precio": {"$gte": 1000, "$lte": 2000},
    
    # Filtrar por moneda
    "moneda": "DOP",
    
    # Excluir placeholders
    "placeholder": False
}
```

### Búsqueda semántica

```python
# Búsqueda semántica con embeddings
async def semantic_search(query, retriever, k=3):
    """Búsqueda semántica en la base de conocimiento"""
    try:
        results = await retriever.ainvoke(query)
        
        # Procesar resultados
        processed_results = []
        for doc in results:
            processed_results.append({
                "content": doc.page_content,
                "metadata": doc.metadata,
                "score": doc.score if hasattr(doc, 'score') else None
            })
        
        return processed_results
        
    except Exception as e:
        print(f"Error en búsqueda semántica: {e}")
        return []
```

## Optimizaciones de rendimiento

### Configuración de índices

```python
# Configuración optimizada de ChromaDB
chroma_config = {
    "collection_name": "servicios",
    "embedding_function": embeddings,
    "persist_directory": "storage/chroma",
    "collection_metadata": {
        "hnsw:space": "cosine",        # Espacio de similitud
        "hnsw:construction_ef": 200,   # Parámetro de construcción
        "hnsw:search_ef": 50,         # Parámetro de búsqueda
        "hnsw:M": 16,                 # Conectividad
        "hnsw:ef_construction": 200   # Construcción EF
    }
}
```

### Configuración de recuperación

```python
# Parámetros de recuperación optimizados
retrieval_config = {
    "chunks": {
        "retriever_k": 3,        # Documentos a recuperar
        "fetch_k": 9,           # Documentos a buscar
        "score_threshold": 0.7,  # Umbral de similitud
        "max_tokens": 2000      # Máximo de tokens
    },
    "qa": {
        "retriever_k": 3,
        "fetch_k": 9,
        "score_threshold": 0.8,
        "max_tokens": 1000
    }
}
```

## Gestión de datos

### Ingesta de datos

```python
# src/ingest.py - Proceso de ingesta
async def ingest_service_data(service_id, documents):
    """Ingiere datos de un servicio en ChromaDB"""
    
    # Configurar embeddings
    embeddings = get_google_embeddings()
    
    # Cargar base de datos
    vectorstore = Chroma(
        collection_name="servicios",
        embedding_function=embeddings,
        persist_directory="storage/chroma"
    )
    
    # Procesar documentos
    texts = [doc["content"] for doc in documents]
    metadatas = [doc["metadata"] for doc in documents]
    ids = [doc["id"] for doc in documents]
    
    # Agregar a ChromaDB
    vectorstore.add_texts(
        texts=texts,
        metadatas=metadatas,
        ids=ids
    )
    
    print(f"✅ Servicio {service_id} ingerido exitosamente")
```

### Actualización incremental

```python
# Actualización incremental de datos
async def update_service_data(service_id, new_documents):
    """Actualiza datos de un servicio existente"""
    
    # Eliminar documentos antiguos
    vectorstore.delete(where={"service_id": service_id})
    
    # Agregar documentos nuevos
    await ingest_service_data(service_id, new_documents)
    
    print(f"✅ Servicio {service_id} actualizado exitosamente")
```

## Monitoreo y métricas

### Métricas de base de datos

```python
# src/utils/kb_metrics.py
class KnowledgeBaseMetrics:
    def __init__(self):
        self.total_documents = 0
        self.total_qa_pairs = 0
        self.search_queries = 0
        self.successful_searches = 0
        self.average_search_time = 0
        self.cache_hits = 0
        self.cache_misses = 0
    
    def record_search(self, query, results, search_time, success=True):
        self.search_queries += 1
        self.average_search_time = (
            (self.average_search_time * (self.search_queries - 1) + search_time) 
            / self.search_queries
        )
        
        if success:
            self.successful_searches += 1
    
    def get_stats(self):
        return {
            "total_documents": self.total_documents,
            "total_qa_pairs": self.total_qa_pairs,
            "search_queries": self.search_queries,
            "success_rate": self.successful_searches / max(self.search_queries, 1),
            "average_search_time": self.average_search_time,
            "cache_hit_rate": self.cache_hits / max(self.cache_hits + self.cache_misses, 1)
        }
```

### Health check de base de datos

```python
# src/tools.py - Health check de base de conocimiento
@function_tool()
async def health_check() -> str:
    """Verifica el estado de la base de conocimiento"""
    
    status = {
        "timestamp": datetime.now().isoformat(),
        "knowledge_base": "unknown",
        "chromadb_chunks": "unknown",
        "chromadb_qa": "unknown",
        "total_documents": 0,
        "total_qa_pairs": 0
    }
    
    try:
        # Verificar base principal
        embeddings = get_google_embeddings()
        retriever = get_chroma_load(embeddings, "storage/chroma", "servicios")
        results = retriever.invoke("test")
        
        status["chromadb_chunks"] = "healthy"
        status["total_documents"] = len(results)
        
        # Verificar base QA
        qa_retriever = get_chroma_load(embeddings, "storage/chroma_qa", "servicios_qa")
        qa_results = qa_retriever.invoke("test")
        
        status["chromadb_qa"] = "healthy"
        status["total_qa_pairs"] = len(qa_results)
        
        if status["chromadb_chunks"] == "healthy" and status["chromadb_qa"] == "healthy":
            status["knowledge_base"] = "healthy"
        else:
            status["knowledge_base"] = "unhealthy"
            
    except Exception as e:
        status["knowledge_base"] = f"error: {str(e)}"
    
    return json.dumps(status)
```

## Backup y recuperación

### Backup de base de datos

```bash
#!/bin/bash
# backup_kb.sh - Backup de base de conocimiento

BACKUP_DIR="/home/taina/backups/kb"
DATE=$(date +%Y%m%d_%H%M%S)

# Crear directorio de backup
mkdir -p $BACKUP_DIR

# Backup de base principal
cp -r storage/chroma $BACKUP_DIR/chroma_$DATE

# Backup de base QA
cp -r storage/chroma_qa $BACKUP_DIR/chroma_qa_$DATE

# Comprimir backup
tar -czf $BACKUP_DIR/kb_backup_$DATE.tar.gz $BACKUP_DIR/chroma_$DATE $BACKUP_DIR/chroma_qa_$DATE

# Limpiar archivos temporales
rm -rf $BACKUP_DIR/chroma_$DATE $BACKUP_DIR/chroma_qa_$DATE

echo "✅ Backup completado: $BACKUP_DIR/kb_backup_$DATE.tar.gz"
```

### Restauración de base de datos

```bash
#!/bin/bash
# restore_kb.sh - Restaurar base de conocimiento

BACKUP_FILE=$1

if [ -z "$BACKUP_FILE" ]; then
    echo "Uso: ./restore_kb.sh <backup_file>"
    exit 1
fi

# Detener servicios
docker compose down

# Hacer backup actual
cp -r storage/chroma storage/chroma_backup_$(date +%Y%m%d_%H%M%S)
cp -r storage/chroma_qa storage/chroma_qa_backup_$(date +%Y%m%d_%H%M%S)

# Extraer backup
tar -xzf $BACKUP_FILE -C /tmp/

# Restaurar bases de datos
cp -r /tmp/chroma_* storage/chroma
cp -r /tmp/chroma_qa_* storage/chroma_qa

# Iniciar servicios
docker compose up -d

echo "✅ Base de conocimiento restaurada"
```

## Troubleshooting

### Problemas comunes

#### Error: "Collection not found"

```bash
# Verificar colecciones
python3 -c "
import chromadb
client = chromadb.PersistentClient(path='storage/chroma')
collections = client.list_collections()
print('Colecciones:', [c.name for c in collections])
"
```

#### Error: "Embedding dimension mismatch"

```bash
# Verificar dimensiones de embeddings
python3 -c "
from src.embeddings.gemini_embedding import get_google_embeddings
embeddings = get_google_embeddings()
print('Dimensiones:', embeddings.dimension)
"
```

#### Error: "Database locked"

```bash
# Verificar procesos que usan la base de datos
lsof storage/chroma/chroma.sqlite3

# Matar procesos si es necesario
pkill -f "src.main"
```

## Recursos adicionales

* [ChromaDB Documentation](https://docs.trychroma.com/)
* [LangChain Vector Stores](https://python.langchain.com/docs/modules/data_connection/vectorstores/)
* [Google Gemini Embeddings](https://ai.google.dev/docs/embeddings_guide)
* [Vector Database Best Practices](https://docs.trychroma.com/usage-guide)

## Próximos pasos

1. **Catálogo de Herramientas**: [Catálogo de Herramientas](https://github.com/public-intelligence/taina_ogtic/blob/master/taina-gitbook-ogtic/taina-asistente-ia/index/scripts/service-catalog.md)
2. **Pipeline de Datos**: [Pipeline de Datos](https://github.com/public-intelligence/taina_ogtic/blob/master/taina-gitbook-ogtic/general/development/data-pipeline.md)
3. **Referencia de Configuración**: [Referencia de Configuración](/taina-agente-ia-ogtic/referencias-tecnicas/api/configuration.md)

***

¿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/arquitectura-y-conceptos/architecture/knowledge-base.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.