> 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.3_data_pipeline.md).

# Pipeline de datos

El pipeline de procesamiento de datos de Taina está diseñado para transformar información de servicios gubernamentales en documentos optimizados para RAG (Retrieval Augmented Generation) y almacenarlos en ChromaDB. Este sistema permite la ingesta, procesamiento y indexación de datos gubernamentales para el sistema de herramientas especializadas.

## Arquitectura del pipeline

### 1. Componentes principales

1. `fetch_service.sh` descarga el servicio bruto desde el catálogo gubernamental.
2. `process_service.py` limpia HTML y construye documentos listos para RAG.
3. `src/ingest.py` indexa los documentos en ChromaDB junto con la colección QA opcional.

### 2. Flujo de procesamiento

#### Paso 1: Obtención de datos

```bash
# Obtener datos de un servicio específico
./fetch_service.sh 163

# Configuración
TOKEN="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
BASE_URL="https://catalogo-staging-iv3sjd4zmq-ue.a.run.app/items/services"
FIELDS="name,description,objective,institution_id.name,..."
```

#### Paso 2: Procesamiento RAG

```bash
# Procesar datos para RAG
python3 process_service.py 163

# Salida generada
163.json                    # JSON original de la API
163_clean.json             # JSON con HTML limpio
163_rag_documents.json     # Documentos preparados para ChromaDB
163_summary.json           # Resumen del procesamiento
```

#### Paso 3: Carga a ChromaDB

```bash
# Ejecutar ingesta con Docker
docker compose exec taina-backend python -m src.ingest --json_dir data/chunks
```

## Scripts del pipeline

### 1. fetch\_service.sh - Obtención de datos

**Propósito**: Obtiene datos de servicios desde la API del catálogo gubernamental.

```bash
#!/bin/bash
# fetch_service.sh - Obtener datos de un servicio específico

SERVICE_ID=$1
TOKEN="eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..."
BASE_URL="https://catalogo-staging-iv3sjd4zmq-ue.a.run.app/items/services"
FIELDS="name,description,objective,institution_id.name,procedures,requirements,locations,digital_channels,contact_info"

echo "🔍 Obteniendo datos del servicio $SERVICE_ID..."

# Obtener datos de la API
curl -s -H "Authorization: Bearer $TOKEN" \
     "$BASE_URL/$SERVICE_ID?fields=$FIELDS" \
     | jq '.' > "${SERVICE_ID}.json"

if [ $? -eq 0 ]; then
    echo "✅ Datos obtenidos exitosamente"
else
    echo "❌ Error al obtener datos"
    exit 1
fi
```

**Configuración**:

* `TOKEN`: Token de autorización para la API
* `BASE_URL`: URL base de la API del catálogo
* `FIELDS`: Campos específicos a obtener

**Transformación**:

* Usa `jq` para transformar JSON de la API
* Convierte campos a español
* Estructura datos para procesamiento posterior

### 2. process\_service.py - Procesamiento RAG

**Propósito**: Limpia HTML y crea documentos optimizados para RAG.

```python
#!/usr/bin/env python3
# process_service.py - Procesar datos para RAG

import json
import sys
from bs4 import BeautifulSoup
import re

def clean_html_for_rag(html_content):
    """Limpia HTML optimizado para RAG/ChromaDB"""
    if not html_content:
        return ""
    
    # Decodifica entidades HTML
    html_content = html_content.replace(' ', ' ')
    html_content = html_content.replace('&', '&')
    html_content = html_content.replace('&lt;', '<')
    html_content = html_content.replace('&gt;', '>')
    
    # Parse HTML
    soup = BeautifulSoup(html_content, 'html.parser')
    
    # Convierte listas a texto numerado
    for ul in soup.find_all('ul'):
        for i, li in enumerate(ul.find_all('li'), 1):
            li.string = f"{i}. {li.get_text()}"
        ul.unwrap()
    
    # Normaliza espacios y saltos de línea
    text = soup.get_text()
    text = re.sub(r'\s+', ' ', text)
    text = text.strip()
    
    return text

def process_json_for_chromadb(data):
    """Procesa recursivamente un JSON para ChromaDB"""
    if isinstance(data, dict):
        result = {}
        for key, value in data.items():
            if isinstance(value, str) and '<' in value and '>' in value:
                # Campo HTML detectado
                result[key] = clean_html_for_rag(value)
            else:
                result[key] = process_json_for_chromadb(value)
        return result
    elif isinstance(data, list):
        return [process_json_for_chromadb(item) for item in data]
    else:
        return data

def create_rag_documents(json_data, service_id):
    """Crea documentos optimizados para RAG"""
    documents = []
    
    # Documento principal del servicio
    main_doc = {
        "id": f"servicio_{service_id}",
        "content": f"Servicio: {json_data.get('name', 'Sin nombre')}\n"
                  f"Descripción: {json_data.get('description', 'Sin descripción')}\n"
                  f"Objetivo: {json_data.get('objective', 'Sin objetivo')}\n"
                  f"Institución: {json_data.get('institution_id', {}).get('name', 'Sin institución')}",
        "metadata": {
            "service_id": service_id,
            "tipo": "servicio_principal",
            "institucion": json_data.get('institution_id', {}).get('name', 'Sin institución'),
            "nombre_servicio": json_data.get('name', 'Sin nombre')
        }
    }
    documents.append(main_doc)
    
    # Documentos por variación
    procedures = json_data.get('procedures', [])
    for i, procedure in enumerate(procedures):
        if procedure.get('name'):
            variation_doc = {
                "id": f"variacion_{service_id}_{i}",
                "content": f"Procedimiento: {procedure.get('name', 'Sin nombre')}\n"
                          f"Descripción: {procedure.get('description', 'Sin descripción')}\n"
                          f"Precio: {procedure.get('price', 'Sin precio')}",
                "metadata": {
                    "service_id": service_id,
                    "tipo": "variacion_servicio",
                    "categoria": procedure.get('name', 'Sin categoría'),
                    "precio": procedure.get('price', 0)
                }
            }
            documents.append(variation_doc)
    
    # Documento de ubicaciones
    locations = json_data.get('locations', [])
    if locations:
        locations_content = "Oficinas disponibles:\n"
        for location in locations:
            locations_content += f"• {location.get('name', 'Sin nombre')}\n"
            if location.get('address'):
                locations_content += f"  Dirección: {location.get('address')}\n"
            if location.get('phone'):
                locations_content += f"  Teléfono: {location.get('phone')}\n"
        
        locations_doc = {
            "id": f"ubicaciones_{service_id}",
            "content": locations_content,
            "metadata": {
                "service_id": service_id,
                "tipo": "ubicaciones",
                "total_oficinas": len(locations)
            }
        }
        documents.append(locations_doc)
    
    # Documento de canales digitales
    digital_channels = json_data.get('digital_channels', [])
    if digital_channels:
        digital_content = "Canales digitales disponibles:\n"
        for channel in digital_channels:
            digital_content += f"• {channel.get('name', 'Sin nombre')}\n"
            if channel.get('url'):
                digital_content += f"  URL: {channel.get('url')}\n"
        
        digital_doc = {
            "id": f"digital_{service_id}",
            "content": digital_content,
            "metadata": {
                "service_id": service_id,
                "tipo": "canal_digital",
                "total_canales": len(digital_channels)
            }
        }
        documents.append(digital_doc)
    
    return documents

def main():
    if len(sys.argv) != 2:
        print("Uso: python3 process_service.py <service_id>")
        sys.exit(1)
    
    service_id = sys.argv[1]
    input_file = f"{service_id}.json"
    
    try:
        # Cargar JSON original
        with open(input_file, 'r', encoding='utf-8') as f:
            raw_data = json.load(f)
        
        # Procesar para RAG
        clean_data = process_json_for_chromadb(raw_data)
        
        # Guardar JSON limpio
        with open(f"{service_id}_clean.json", 'w', encoding='utf-8') as f:
            json.dump(clean_data, f, ensure_ascii=False, indent=2)
        
        # Crear documentos RAG
        rag_documents = create_rag_documents(clean_data, service_id)
        
        # Guardar documentos RAG
        with open(f"{service_id}_rag_documents.json", 'w', encoding='utf-8') as f:
            json.dump(rag_documents, f, ensure_ascii=False, indent=2)
        
        # Crear resumen
        summary = {
            "service_id": service_id,
            "total_documents": len(rag_documents),
            "document_types": list(set(doc["metadata"]["tipo"] for doc in rag_documents)),
            "processed_at": "2025-01-08T00:00:00Z"
        }
        
        with open(f"{service_id}_summary.json", 'w', encoding='utf-8') as f:
            json.dump(summary, f, ensure_ascii=False, indent=2)
        
        print(f"✅ Servicio {service_id} procesado exitosamente")
        print(f"📄 Documentos generados: {len(rag_documents)}")
        print(f"📋 Tipos: {', '.join(summary['document_types'])}")
        
    except Exception as e:
        print(f"❌ Error procesando servicio {service_id}: {str(e)}")
        sys.exit(1)

if __name__ == "__main__":
    main()
```

### 3. run\_pipeline.sh - Orquestación Completa

**Propósito**: Orquesta el pipeline completo para un servicio.

```bash
#!/bin/bash
# run_pipeline.sh - Pipeline completo para un servicio

SERVICE_ID=$1

if [ -z "$SERVICE_ID" ]; then
    echo "❌ Error: Se requiere un ID de servicio"
    echo "Uso: ./run_pipeline.sh <service_id>"
    exit 1
fi

echo "🚀 Iniciando pipeline para servicio $SERVICE_ID..."

# Paso 1: Obtener datos
echo "📥 Paso 1: Obteniendo datos..."
./fetch_service.sh $SERVICE_ID

if [ $? -ne 0 ]; then
    echo "❌ Error en obtención de datos"
    exit 1
fi

# Paso 2: Procesar datos
echo "🔧 Paso 2: Procesando datos..."
python3 process_service.py $SERVICE_ID

if [ $? -ne 0 ]; then
    echo "❌ Error en procesamiento"
    exit 1
fi

# Paso 3: Verificar resultados
echo "✅ Paso 3: Verificando resultados..."
if [ -f "${SERVICE_ID}_rag_documents.json" ]; then
    echo "✅ Pipeline completado exitosamente"
    echo "📄 Archivos generados:"
    ls -la ${SERVICE_ID}_*.json
else
    echo "❌ Error: No se generaron documentos RAG"
    exit 1
fi
```

### 4. Ingesta en ChromaDB

**Propósito**: Carga los documentos procesados en ChromaDB.

```bash
# Ingesta con Docker Compose
docker compose exec taina-backend python -m src.ingest \
    --json_dir data/chunks

# Verificar resultado
docker compose exec taina-backend python3 -c "
import chromadb
client = chromadb.PersistentClient(path='storage/chroma')
col = client.get_collection('servicios')
print(f'Documentos: {col.count()}')
"
```

## Pipeline de QA (Preguntas/Respuestas)

### 1. Generación de QA

```bash
# Generar preguntas/respuestas por servicio
docker compose exec taina-backend python -m src.qa.generator \
  --input_dir data/chunks \
  --output_dir data/QA
```

### 2. Construcción de Índice QA

```bash
# Construir índice QA dedicado
docker compose exec taina-backend python -m src.qa.build_qa_index \
  --qa_dir data/QA \
  --persist_path storage/chroma_qa \
  --collection_name servicios_qa \
  --reset
```

### 3. Estructura de QA

```json
{
  "service_id": "163",
  "qa_pairs": [
    {
      "question": "¿Cómo puedo renovar mi licencia de conducir?",
      "answer": "Para renovar tu licencia de conducir, necesitas...",
      "metadata": {
        "intent": "renovacion_licencia",
        "source_chunk_id": "servicio_163",
        "placeholder": false
      }
    }
  ]
}
```

## Tipos de documentos generados

### 1. Documento principal

```json
{
  "id": "servicio_163",
  "content": "Servicio: Renovación Licencia de Conducir\nDescripción: Servicio para renovar licencias...",
  "metadata": {
    "service_id": "163",
    "tipo": "servicio_principal",
    "institucion": "INTRANT",
    "nombre_servicio": "Renovación Licencia de Conducir"
  }
}
```

### 2. Documentos de variaciones

```json
{
  "id": "variacion_163_0",
  "content": "Procedimiento: Renovación Licencia Categoría 02\nDescripción: Procedimiento para renovar...\nPrecio: 1900.0",
  "metadata": {
    "service_id": "163",
    "tipo": "variacion_servicio",
    "categoria": "Renovación Licencia Categoría 02",
    "precio": 1900.0
  }
}
```

### 3. Documento de ubicaciones

```json
{
  "id": "ubicaciones_163",
  "content": "Oficinas disponibles:\n• Sede Principal INTRANT\n  Dirección: Av. 27 de Febrero...",
  "metadata": {
    "service_id": "163",
    "tipo": "ubicaciones",
    "total_oficinas": 15
  }
}
```

### 4. Documento de canales digitales

```json
{
  "id": "digital_163",
  "content": "Canales digitales disponibles:\n• Portal Web\n  URL: https://...",
  "metadata": {
    "service_id": "163",
    "tipo": "canal_digital",
    "total_canales": 3
  }
}
```

## Integración con ChromaDB

### 1. Carga de documentos

```python
import chromadb
import json

# Cargar documentos RAG
with open('163_rag_documents.json', 'r') as f:
    documents = json.load(f)

# Configurar ChromaDB
client = chromadb.Client()
collection = client.create_collection("servicios_gobierno")

# Insertar documentos
for doc in documents:
    collection.add(
        documents=[doc['content']],
        metadatas=[doc['metadata']],
        ids=[doc['id']]
    )
```

### 2. Consultas RAG

```python
# Buscar por procedimiento
results = collection.query(
    query_texts=["pasos para renovar licencia"],
    where={"tipo": "variacion_servicio"}
)

# Buscar por institución
results = collection.query(
    query_texts=["servicios INTRANT"],
    where={"institucion": "INTRANT"}
)

# Buscar oficinas en Santo Domingo
results = collection.query(
    query_texts=["oficinas Santo Domingo"],
    where={"tipo": "ubicaciones"}
)
```

## Configuración y dependencias

### 1. setup\_pipeline.sh

```bash
#!/bin/bash
# setup_pipeline.sh - Instalar dependencias del pipeline

echo "🔧 Configurando pipeline de datos..."

# Verificar dependencias del sistema
echo "📋 Verificando dependencias del sistema..."

# Python3
if ! command -v python3 &> /dev/null; then
    echo "❌ Python3 no encontrado"
    exit 1
fi

# pip3
if ! command -v pip3 &> /dev/null; then
    echo "❌ pip3 no encontrado"
    exit 1
fi

# jq
if ! command -v jq &> /dev/null; then
    echo "❌ jq no encontrado. Instalando..."
    sudo apt install jq
fi

# curl
if ! command -v curl &> /dev/null; then
    echo "❌ curl no encontrado. Instalando..."
    sudo apt install curl
fi

# Instalar dependencias de Python
echo "📦 Instalando dependencias de Python..."
pip3 install beautifulsoup4 lxml

echo "✅ Pipeline configurado exitosamente"
```

### 2. Estructura de archivos

```
data/
├── chunks/          # JSONs procesados
│   ├── 163.json
│   ├── 163_clean.json
│   ├── 163_rag_documents.json
│   └── 163_summary.json
├── QA/                        # Preguntas/Respuestas
│   ├── 163_qa.json
│   └── manifest.json
└── storage/
    ├── chroma/                 # Base de conocimiento principal
    └── chroma_qa/             # Base de conocimiento QA
```

## Monitoreo y logging

### 1. Logs del pipeline

```bash
# Ver archivos generados recientemente
find . -name "*_rag_documents.json" -mtime -1

# Contar total de documentos procesados
find . -name "*_summary.json" -exec jq '.total_documents' {} \; | paste -sd+ | bc

# Ver servicios procesados hoy
find . -name "*_summary.json" -mtime -1 -exec jq -r '.service_id' {} \;
```

### 2. Validación de datos

```bash
# Validar estructura JSON
jq '.' 163.json

# Verificar campos requeridos
jq '.datos.nombre' 163.json
jq '.datos.institucion_responsable' 163.json

# Contar documentos generados
jq '. | length' 163_rag_documents.json

# Ver tipos de documentos
jq '[.[].metadata.tipo] | unique' 163_rag_documents.json
```

## Troubleshooting

### 1. Errores comunes

#### "jq command not found"

```bash
# Ubuntu/Debian
sudo apt install jq

# CentOS/RHEL
sudo yum install jq

# macOS
brew install jq
```

#### "beautifulsoup4 not found"

```bash
pip3 install beautifulsoup4 lxml
```

#### "Permission denied"

```bash
chmod +x *.sh
```

#### "Service not found"

* Verificar que el número de servicio existe en la API
* Revisar el token de autorización en `fetch_service.sh`

### 2. Validación de datos

```bash
# Verificar JSON de entrada
jq '.' 163.json

# Verificar documentos RAG
jq '. | length' 163_rag_documents.json
jq '[.[].metadata.tipo] | unique' 163_rag_documents.json
```

## Workflow recomendado

### 1. Setup inicial (una vez)

```bash
cd /home/taina/taina_ogtic/backend/data/chunks
./setup_pipeline.sh
```

### 2. Procesar servicios nuevos

```bash
# Servicio individual
./run_pipeline.sh 163

# Múltiples servicios
./run_pipeline.sh 163
./run_pipeline.sh 164
./run_pipeline.sh 165
```

### 3. Actualizar servicios existentes

```bash
# Re-procesar si hay cambios
./run_pipeline.sh 163
```

### 4. Cargar a ChromaDB

```bash
# Ejecutar ingesta con Docker
cd /home/taina/taina_ogtic
docker compose exec taina-backend python -m src.ingest --json_dir data/chunks
```

## Recursos adicionales

* [ChromaDB Documentation](https://docs.trychroma.com/)
* [BeautifulSoup Documentation](https://www.crummy.com/software/BeautifulSoup/bs4/doc/)
* [jq Documentation](https://stedolan.github.io/jq/)

***

¿Necesitas ayuda? Consulta la [guía de solución de problemas](/taina-agente-ia-ogtic/operacion-y-monitoreo/troubleshooting.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.3_data_pipeline.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.