IAIA con Clase
Acceder Empezar
Lección 28 · 10 min · Gratis

Agente de conocimiento industrial

Open In Colab

Planteamiento del problema

En entornos industriales, los ingenieros y técnicos a menudo tienen dificultades para gestionar y recuperar información completa sobre diversos equipos. Esta información está dispersa en manuales técnicos, registros de mantenimiento, protocolos de seguridad, guías de solución de problemas e inventarios de piezas. La naturaleza fragmentada de estos datos dificulta su acceso y utilización eficaz, lo que provoca ineficiencias y posibles riesgos de seguridad. Este problema requiere una solución inteligente y adaptable para proporcionar respuestas en tiempo real y conscientes del contexto a las consultas.

Solución propuesta

Para abordar estos desafíos, proponemos un flujo de trabajo agéntico que integra un sistema de generación aumentada por recuperación (RAG) con un sistema de consulta de bases de datos (FunctionCalling). Esta solución aprovecha los LLM (incluido el mecanismo de salida estructurada), los modelos de embeddings y la recuperación de datos estructurados para proporcionar información contextualmente relevante y precisa. El flujo de trabajo está orquestado por múltiples agentes, cada uno con un rol específico:

  1. RAGAgent: Utiliza LLM y modelos de embeddings para recuperar y generar información contextualmente relevante a partir de documentos técnicos.
  2. DatabaseQueryAgent: Maneja la recuperación precisa y estructurada de datos de bases de datos que contienen registros de mantenimiento, especificaciones técnicas, inventarios de piezas y registros de cumplimiento.
  3. WorkflowOrchestrator: Orquesta las interacciones entre RAGSearchAgent y DatabaseAgent, asegurando una resolución de consultas fluida y eficiente.

Detalles del conjunto de datos

Documentos PDF

Los documentos PDF contienen información detallada sobre diversos equipos industriales, categorizados en:

  1. Manuales técnicos: Guías de operación y mantenimiento.
  2. Guías de mantenimiento: Tareas de mantenimiento rutinario y preventivo.
  3. Guías de solución de problemas: Soluciones a problemas comunes.
  4. Protocolos de seguridad: Procedimientos y directrices de seguridad.

Bases de datos

Las bases de datos contienen información estructurada que complementa los documentos PDF:

  1. Base de datos de cumplimiento (compliance_db): Certificaciones de seguridad y estados de cumplimiento.
  2. Base de datos de mantenimiento (maintenance_db): Registros de actividades de mantenimiento.
  3. Base de datos de especificaciones técnicas (technical_specifications_db): Especificaciones técnicas detalladas.
  4. Base de datos de inventario y compatibilidad de piezas (parts_inventory_compatibility_db): Información sobre piezas, compatibilidad y estado del inventario.

Al integrar estos conjuntos de datos, el flujo de trabajo agéntico propuesto tiene como objetivo proporcionar un sistema completo y eficiente para gestionar y recuperar información de equipos industriales, asegurando que los ingenieros y técnicos tengan acceso a la información más relevante y actualizada.

NOTA: Ten en cuenta que todos los datos utilizados en esta demostración se han generado sintéticamente.

Arquitectura técnica:

Alt text

Instalación

Instala los paquetes de Python necesarios para el sistema IndusAgent.

!pip install mistralai==1.5.1     # Mistral AI client
!pip install qdrant-client==1.13.2 # Vector database client
!pip install gdown==5.2.0        # Google Drive download
Collecting mistralai==1.5.1
  Downloading mistralai-1.5.1-py3-none-any.whl.metadata (29 kB)
Collecting eval-type-backport>=0.2.0 (from mistralai==1.5.1)
  Downloading eval_type_backport-0.2.2-py3-none-any.whl.metadata (2.2 kB)
Requirement already satisfied: httpx>=0.27.0 in /usr/local/lib/python3.11/dist-packages (from mistralai==1.5.1) (0.28.1)
Collecting jsonpath-python>=1.0.6 (from mistralai==1.5.1)
  Downloading jsonpath_python-1.0.6-py3-none-any.whl.metadata (12 kB)
Requirement already satisfied: pydantic>=2.9.0 in /usr/local/lib/python3.11/dist-packages (from mistralai==1.5.1) (2.10.6)
Requirement already satisfied: python-dateutil>=2.8.2 in /usr/local/lib/python3.11/dist-packages (from mistralai==1.5.1) (2.8.2)
Collecting typing-inspect>=0.9.0 (from mistralai==1.5.1)
  Downloading typing_inspect-0.9.0-py3-none-any.whl.metadata (1.5 kB)
Requirement already satisfied: anyio in /usr/local/lib/python3.11/dist-packages (from httpx>=0.27.0->mistralai==1.5.1) (3.7.1)
Requirement already satisfied: certifi in /usr/local/lib/python3.11/dist-packages (from httpx>=0.27.0->mistralai==1.5.1) (2025.1.31)
Requirement already satisfied: httpcore==1.* in /usr/local/lib/python3.11/dist-packages (from httpx>=0.27.0->mistralai==1.5.1) (1.0.7)
Requirement already satisfied: idna in /usr/local/lib/python3.11/dist-packages (from httpx>=0.27.0->mistralai==1.5.1) (3.10)
Requirement already satisfied: h11<0.15,>=0.13 in /usr/local/lib/python3.11/dist-packages (from httpcore==1.*->httpx>=0.27.0->mistralai==1.5.1) (0.14.0)
Requirement already satisfied: annotated-types>=0.6.0 in /usr/local/lib/python3.11/dist-packages (from pydantic>=2.9.0->mistralai==1.5.1) (0.7.0)
Requirement already satisfied: pydantic-core==2.27.2 in /usr/local/lib/python3.11/dist-packages (from pydantic>=2.9.0->mistralai==1.5.1) (2.27.2)
Requirement already satisfied: typing-extensions>=4.12.2 in /usr/local/lib/python3.11/dist-packages (from pydantic>=2.9.0->mistralai==1.5.1) (4.12.2)
Requirement already satisfied: six>=1.5 in /usr/local/lib/python3.11/dist-packages (from python-dateutil>=2.8.2->mistralai==1.5.1) (1.17.0)
Collecting mypy-extensions>=0.3.0 (from typing-inspect>=0.9.0->mistralai==1.5.1)
  Downloading mypy_extensions-1.0.0-py3-none-any.whl.metadata (1.1 kB)
Requirement already satisfied: sniffio>=1.1 in /usr/local/lib/python3.11/dist-packages (from anyio->httpx>=0.27.0->mistralai==1.5.1) (1.3.1)
Downloading mistralai-1.5.1-py3-none-any.whl (278 kB)
   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 278.3/278.3 kB 10.1 MB/s eta 0:00:00
[?25hDownloading eval_type_backport-0.2.2-py3-none-any.whl (5.8 kB)
Downloading jsonpath_python-1.0.6-py3-none-any.whl (7.6 kB)
Downloading typing_inspect-0.9.0-py3-none-any.whl (8.8 kB)
Downloading mypy_extensions-1.0.0-py3-none-any.whl (4.7 kB)
Installing collected packages: mypy-extensions, jsonpath-python, eval-type-backport, typing-inspect, mistralai
Successfully installed eval-type-backport-0.2.2 jsonpath-python-1.0.6 mistralai-1.5.1 mypy-extensions-1.0.0 typing-inspect-0.9.0
Collecting qdrant-client==1.13.2
  Downloading qdrant_client-1.13.2-py3-none-any.whl.metadata (10 kB)
Requirement already satisfied: grpcio>=1.41.0 in /usr/local/lib/python3.11/dist-packages (from qdrant-client==1.13.2) (1.70.0)
Collecting grpcio-tools>=1.41.0 (from qdrant-client==1.13.2)
  Downloading grpcio_tools-1.70.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl.metadata (5.3 kB)
Requirement already satisfied: httpx>=0.20.0 in /usr/local/lib/python3.11/dist-packages (from httpx[http2]>=0.20.0->qdrant-client==1.13.2) (0.28.1)
Requirement already satisfied: numpy>=1.21 in /usr/local/lib/python3.11/dist-packages (from qdrant-client==1.13.2) (1.26.4)
Collecting portalocker<3.0.0,>=2.7.0 (from qdrant-client==1.13.2)
  Downloading portalocker-2.10.1-py3-none-any.whl.metadata (8.5 kB)
Requirement already satisfied: pydantic>=1.10.8 in /usr/local/lib/python3.11/dist-packages (from qdrant-client==1.13.2) (2.10.6)
Requirement already satisfied: urllib3<3,>=1.26.14 in /usr/local/lib/python3.11/dist-packages (from qdrant-client==1.13.2) (2.3.0)
Collecting protobuf<6.0dev,>=5.26.1 (from grpcio-tools>=1.41.0->qdrant-client==1.13.2)
  Downloading protobuf-5.29.3-cp38-abi3-manylinux2014_x86_64.whl.metadata (592 bytes)
Requirement already satisfied: setuptools in /usr/local/lib/python3.11/dist-packages (from grpcio-tools>=1.41.0->qdrant-client==1.13.2) (75.1.0)
Requirement already satisfied: anyio in /usr/local/lib/python3.11/dist-packages (from httpx>=0.20.0->httpx[http2]>=0.20.0->qdrant-client==1.13.2) (3.7.1)
Requirement already satisfied: certifi in /usr/local/lib/python3.11/dist-packages (from httpx>=0.20.0->httpx[http2]>=0.20.0->qdrant-client==1.13.2) (2025.1.31)
Requirement already satisfied: httpcore==1.* in /usr/local/lib/python3.11/dist-packages (from httpx>=0.20.0->httpx[http2]>=0.20.0->qdrant-client==1.13.2) (1.0.7)
Requirement already satisfied: idna in /usr/local/lib/python3.11/dist-packages (from httpx>=0.20.0->httpx[http2]>=0.20.0->qdrant-client==1.13.2) (3.10)
Requirement already satisfied: h11<0.15,>=0.13 in /usr/local/lib/python3.11/dist-packages (from httpcore==1.*->httpx>=0.20.0->httpx[http2]>=0.20.0->qdrant-client==1.13.2) (0.14.0)
Requirement already satisfied: h2<5,>=3 in /usr/local/lib/python3.11/dist-packages (from httpx[http2]>=0.20.0->qdrant-client==1.13.2) (4.2.0)
Requirement already satisfied: annotated-types>=0.6.0 in /usr/local/lib/python3.11/dist-packages (from pydantic>=1.10.8->qdrant-client==1.13.2) (0.7.0)
Requirement already satisfied: pydantic-core==2.27.2 in /usr/local/lib/python3.11/dist-packages (from pydantic>=1.10.8->qdrant-client==1.13.2) (2.27.2)
Requirement already satisfied: typing-extensions>=4.12.2 in /usr/local/lib/python3.11/dist-packages (from pydantic>=1.10.8->qdrant-client==1.13.2) (4.12.2)
Requirement already satisfied: hyperframe<7,>=6.1 in /usr/local/lib/python3.11/dist-packages (from h2<5,>=3->httpx[http2]>=0.20.0->qdrant-client==1.13.2) (6.1.0)
Requirement already satisfied: hpack<5,>=4.1 in /usr/local/lib/python3.11/dist-packages (from h2<5,>=3->httpx[http2]>=0.20.0->qdrant-client==1.13.2) (4.1.0)
Requirement already satisfied: sniffio>=1.1 in /usr/local/lib/python3.11/dist-packages (from anyio->httpx>=0.20.0->httpx[http2]>=0.20.0->qdrant-client==1.13.2) (1.3.1)
Downloading qdrant_client-1.13.2-py3-none-any.whl (306 kB)
   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 306.6/306.6 kB 8.1 MB/s eta 0:00:00
[?25hDownloading grpcio_tools-1.70.0-cp311-cp311-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (2.5 MB)
   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 2.5/2.5 MB 68.2 MB/s eta 0:00:00
[?25hDownloading portalocker-2.10.1-py3-none-any.whl (18 kB)
Downloading protobuf-5.29.3-cp38-abi3-manylinux2014_x86_64.whl (319 kB)
   ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 319.7/319.7 kB 28.9 MB/s eta 0:00:00
[?25hInstalling collected packages: protobuf, portalocker, grpcio-tools, qdrant-client
  Attempting uninstall: protobuf
    Found existing installation: protobuf 4.25.6
    Uninstalling protobuf-4.25.6:
      Successfully uninstalled protobuf-4.25.6
Successfully installed grpcio-tools-1.70.0 portalocker-2.10.1 protobuf-5.29.3 qdrant-client-1.13.2
Requirement already satisfied: gdown==5.2.0 in /usr/local/lib/python3.11/dist-packages (5.2.0)
Requirement already satisfied: beautifulsoup4 in /usr/local/lib/python3.11/dist-packages (from gdown==5.2.0) (4.13.3)
Requirement already satisfied: filelock in /usr/local/lib/python3.11/dist-packages (from gdown==5.2.0) (3.17.0)
Requirement already satisfied: requests[socks] in /usr/local/lib/python3.11/dist-packages (from gdown==5.2.0) (2.32.3)
Requirement already satisfied: tqdm in /usr/local/lib/python3.11/dist-packages (from gdown==5.2.0) (4.67.1)
Requirement already satisfied: soupsieve>1.2 in /usr/local/lib/python3.11/dist-packages (from beautifulsoup4->gdown==5.2.0) (2.6)
Requirement already satisfied: typing-extensions>=4.0.0 in /usr/local/lib/python3.11/dist-packages (from beautifulsoup4->gdown==5.2.0) (4.12.2)
Requirement already satisfied: charset-normalizer<4,>=2 in /usr/local/lib/python3.11/dist-packages (from requests[socks]->gdown==5.2.0) (3.4.1)
Requirement already satisfied: idna<4,>=2.5 in /usr/local/lib/python3.11/dist-packages (from requests[socks]->gdown==5.2.0) (3.10)
Requirement already satisfied: urllib3<3,>=1.21.1 in /usr/local/lib/python3.11/dist-packages (from requests[socks]->gdown==5.2.0) (2.3.0)
Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.11/dist-packages (from requests[socks]->gdown==5.2.0) (2025.1.31)
Requirement already satisfied: PySocks!=1.5.7,>=1.5.6 in /usr/local/lib/python3.11/dist-packages (from requests[socks]->gdown==5.2.0) (1.7.1)

Importaciones

Importa las bibliotecas necesarias para operaciones de LLM, procesamiento de datos, gestión de bases de datos vectoriales y funciones de utilidad.

# Core libraries
import os
import json
import functools
import warnings
from typing import List, Dict, Any, Tuple

# LLM and Data Processing
from mistralai.client import Mistral
from pydantic import BaseModel
import pandas as pd
import sqlite3
from tqdm import tqdm

# Vector Database
from qdrant_client import QdrantClient
from qdrant_client.models import (
   PointStruct, VectorParams, Distance,
   Filter, FieldCondition, MatchValue
)

# Data Download
import gdown
import zipfile

# Suppress warnings
warnings.filterwarnings('ignore', category=DeprecationWarning)

Descargar datos

Descarga el conjunto de datos de Google Drive, lo extrae a un directorio de datos y configura el entorno de trabajo. El conjunto de datos contiene archivos CSV para operaciones de base de datos y PDF para procesamiento de documentos.

Al final del proceso, deberías poder ver los datos descargados, como se muestra en la imagen a continuación.

Descargar datos de Google Drive

file_id = "1lwYSN6ry3JOA7pw3WAx72a_IXGqqmR8y"
output_file = "data.zip"  # Change this if your file is not a ZIP file

# Google Drive direct download URL
gdrive_url = f"https://drive.google.com/uc?id={file_id}"

# Download the file
gdown.download(gdrive_url, output_file, quiet=False)

print(f"✅ File downloaded: {output_file}")
Downloading...
From: https://drive.google.com/uc?id=1lwYSN6ry3JOA7pw3WAx72a_IXGqqmR8y
To: /content/data.zip
100%|██████████| 396k/396k [00:00<00:00, 6.10MB/s]
✅ File downloaded: data.zip

Extraer y configurar el directorio de datos

# Unzip the file into the current directory
with zipfile.ZipFile(output_file, 'r') as zip_ref:
    zip_ref.extractall(".")  # Extracts directly to the current directory

print(f"✅ Files extracted to: {os.getcwd()}")  # Confirm extraction path


output_dir = "data"

# Change working directory to the extracted folder
os.chdir(output_dir)

# Verify the new working directory
print(f"📂 Current directory: {os.getcwd()}")
✅ Files extracted to: /content
📂 Current directory: /content/data
# List files in the extracted folder
print("📜 Extracted files:", os.listdir())
📜 Extracted files: ['csv_data', 'pdf_data']

Configurar variables de entorno

Configura la clave de API de Mistral como una variable de entorno para la autenticación.

os.environ["MISTRAL_API_KEY"] = "<YOUR MISTRAL API KEY>" # Get your Mistral API key from https://console.mistral.ai/api-keys/

Inicializar Mistral LLM y la base de datos vectorial Qdrant

Inicializa el cliente Mistral LLM para la generación de texto y el cliente de la base de datos vectorial Qdrant para operaciones de búsqueda de similitud.

Nota:

  1. Usaremos nuestro modelo más reciente, Mistral Small 3 para la demostración.
  2. Necesitas configurar Qdrant Cloud o una configuración de Docker antes de continuar. Puedes consultar la documentación para las instrucciones de configuración.
model = "mistral-small-latest"
mistral_client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
qdrant_client = QdrantClient(
    url= "<URL>",
    api_key= "<API KEY>",
) # Replace with your Qdrant API key and URL if you are using Qdrant Cloud - https://cloud.qdrant.io/

Prompts del sistema

El sistema utiliza tres tipos diferentes de prompts para guiar a los LLM en la generación de respuestas:

  1. Prompt de resumen de PDF: summarization_prompt se utiliza para crear resúmenes concisos de documentos PDF.
  2. Prompt de generación de respuestas: response_generation_prompt se utiliza para generar respuestas basadas en el contexto recuperado.
  3. Prompt de integración de respuesta final: final_response_generation_prompt se utiliza para resumir respuestas de múltiples fuentes: PDF y diferentes bases de datos.
# Define the prompt for generating a response
response_generation_prompt = '''Based on the following context answer the query:\n\n Context: {context}\n\n Query: {query}'''

# Prompt for summarizing the PDF text
summarization_prompt = '''Your task is to summarize the following text focusing on the core essence of the text in maximum of 2-3 sentences.'''

# Prompt for final response summarization
final_response_summarization_prompt = """You are an expert technical assistant. Your task is to create a comprehensive,
coherent response by combining information from multiple sources: database records and documentation.

Consider the following guidelines:
1. Integrate information from both sources seamlessly
2. Resolve any conflicts between sources, if they exist
3. Present information in a logical, step-by-step manner when applicable
4. Include specific technical details, measurements, and procedures when available
5. Prioritize safety-related information when present
6. Add relevant maintenance intervals or schedules if mentioned
7. Reference specific part numbers or specifications when provided

The user's query is: {query}

Based on the following responses from different sources, create a unified, clear answer:
{responses}

Remember to:
- Focus on accuracy and completeness
- Maintain technical precision
- Use clear, professional language
- Address all aspects of the query
- Highlight any important warnings or precautions"""

DataProcessor

La clase DataProcessor es un componente integral que maneja todas las operaciones de procesamiento de datos en el sistema. Gestiona datos no estructurados (PDF) y estructurados (CSV), junto con la generación y el almacenamiento de embeddings.

  • Procesamiento de documentos PDF y extracción de texto utilizando Mistral OCR.
  • Ingesta de CSV a la base de datos.
  • Generación de embeddings y almacenamiento vectorial.
  • Procesamiento por lotes de documentos y datos.

Componentes principales

1. Procesamiento de documentos

  • get_categorized_filepaths: Recorre la estructura de directorios para obtener rutas de archivos PDF categorizadas.
  • parse_pdf: Extrae texto de todas las páginas de un archivo PDF utilizando Mistral OCR.
  • process_single_pdf: Procesa PDF individuales a través de la tubería completa.
  • process_documents: Maneja el procesamiento secuencial de múltiples documentos.

2. Resumen y Embeddings

  • summarize: Genera resúmenes concisos de texto utilizando el modelo Mistral.
  • get_text_embedding: Crea embeddings de texto utilizando el modelo de embeddings de Mistral.
  • qdrant_insert_embeddings: Almacena embeddings con metadatos en la base de datos vectorial Qdrant.
  • process_and_store_embeddings: Maneja el procesamiento por lotes de embeddings.

3. Operaciones de base de datos

  • insert_csv_to_table: Carga un solo archivo CSV en una tabla de base de datos especificada.
  • insert_data_database: Maneja la inserción de múltiples archivos CSV en sus respectivas tablas.
class DataProcessor:
    """
    Handles all data processing operations including:
    - PDF parsing and text extraction
    - CSV to database ingestion
    - Embedding generation and storage
    - Batch processing of documents and data
    """
    def __init__(self, mistral_client: Mistral, qdrant_client: QdrantClient):
        self.mistral_client = mistral_client
        self.qdrant_client = qdrant_client

    def get_categorized_filepaths(self, root_dir: str) -> List[Dict[str, str]]:
        """
        Walk through the directory structure and get file paths with their categories.
        """
        categorized_files = []

        for category in os.listdir(root_dir):
            category_path = os.path.join(root_dir, category)

            if not os.path.isdir(category_path):
                continue

            for root, _, files in os.walk(category_path):
                for file in files:
                    if file.lower().endswith('.pdf'):
                        filepath = os.path.join(root, file)
                        categorized_files.append({
                            'filepath': filepath,
                            'category': category
                        })

        return categorized_files

    def parse_pdf(self, file_path: str) -> str:
        """Parse a PDF file and extract text from all pages using Mistral OCR."""

        # Upload a file

        uploaded_pdf = self.mistral_client.files.upload(
            file={
                "file_name": file_path,
                "content": open(file_path, "rb"),
            },
            purpose="ocr"
        )

        # Get a signed URL for the uploaded file

        signed_url = self.mistral_client.files.get_signed_url(file_id=uploaded_pdf.id)

        # Get OCR results

        ocr_response = self.mistral_client.ocr.process(
            model="mistral-ocr-latest",
            document={
                "type": "document_url",
                "document_url": signed_url.url,
            }
        )

        # Extract text from the OCR response

        text = "\n".join([x.markdown for x in (ocr_response.pages)])

        return text

    def summarize(self, text: str, summarization_prompt: str = summarization_prompt) -> str:
        """Summarize the given text using the Mistral model."""
        chat_response = self.mistral_client.chat.complete(
            model=model,
            messages=[
                {
                    "role": "system",
                    "content": summarization_prompt
                },
                {
                    "role": "user",
                    "content": text
                },
            ],
            temperature=0
        )
        return chat_response.choices[0].message.content

    def get_text_embedding(self, inputs: List[str]) -> List[float]:
        """Get the text embedding for the given inputs."""
        embeddings_batch_response = self.mistral_client.embeddings.create(
            model="mistral-embed",
            inputs=inputs
        )
        return embeddings_batch_response.data[0].embedding

    def qdrant_insert_embeddings(self, summaries: List[str], texts: List[str],
                               filepaths: List[str], categories: List[str]):
        """Insert embeddings into Qdrant with metadata."""
        embeddings = [self.get_text_embedding([t]) for t in summaries]

        if not self.qdrant_client.collection_exists("embeddings"):
            self.qdrant_client.create_collection(
                collection_name="embeddings",
                vectors_config=VectorParams(size=1024, distance=Distance.COSINE),
            )

        self.qdrant_client.upsert(
            collection_name="embeddings",
            points=[
                PointStruct(
                    id=idx,
                    vector=embedding,
                    payload={
                        "filepath": filepaths[idx],
                        "category": categories[idx],
                        "text": texts[idx]
                    }
                ) for idx, embedding in enumerate(embeddings)
            ]
        )

    def process_single_pdf(self, file_info: Dict[str, str]) -> Dict[str, any]:
        """Process a single PDF file through the pipeline."""
        filepath = file_info['filepath']
        category = file_info['category']

        pdf_text = self.parse_pdf(filepath)
        summary = self.summarize(pdf_text)

        return {
            'filepath': filepath,
            'category': category,
            'full_text': pdf_text,
            'summary': summary
        }

    def process_documents(self, file_list: List[Dict[str, str]]) -> List[Dict[str, any]]:
        """Process documents sequentially."""
        processed_docs = []

        for file_info in tqdm(file_list, desc="Processing PDFs"):
            try:
                processed_doc = self.process_single_pdf(file_info)
                processed_docs.append(processed_doc)
            except Exception as e:
                print(f"Error processing {file_info['filepath']}: {str(e)}")
                continue

        return processed_docs

    def insert_csv_to_table(self, file_path: str, db_path: str, table_name: str):
        """
        Insert CSV data into a table of SQLite database.

        Args:
            file_path (str): Path to the CSV file
            db_path (str): Path to the SQLite database
            table_name (str): Name of the table to create/update
        """
        df = pd.read_csv(file_path)
        conn = sqlite3.connect(db_path)
        df.to_sql(table_name, conn, if_exists='replace', index=False)
        conn.close()

    def insert_data_database(self, db_path: str, file_mappings: Dict[str, str]):
        """
        Bulk insert multiple CSV files into their respective database tables.

        Args:
            db_path (str): Path to the SQLite database
            file_mappings (Dict[str, str]): Dictionary mapping table names to CSV file paths
        """
        for table_name, file_path in file_mappings.items():
            try:
                self.insert_csv_to_table(file_path, db_path, table_name)
                print(f"Successfully inserted data into {table_name}")
            except Exception as e:
                print(f"Error inserting data into {table_name}: {str(e)}")

    def process_and_store_embeddings(self, docs: List[Dict[str, any]], batch_size: int = 10):
        """Generate embeddings and store them in Qdrant in batches."""
        for i in range(0, len(docs), batch_size):
            batch = docs[i:i + batch_size]

            texts = [doc['full_text'] for doc in batch]
            summaries = [doc['summary'] for doc in batch]
            filepaths = [doc['filepath'] for doc in batch]
            categories = [doc['category'] for doc in batch]

            try:
                self.qdrant_insert_embeddings(summaries, texts, filepaths, categories)
                print(f"Processed batch {i//batch_size + 1}/{(len(docs) + batch_size - 1)//batch_size}")
            except Exception as e:
                print(f"Error processing batch starting at index {i}: {str(e)}")
                continue

RAGAgent

La clase RAGAgent implementa la Generación Aumentada por Recuperación (RAG) para proporcionar búsqueda inteligente y generación de respuestas. Combina las capacidades de búsqueda vectorial con el LLM para dar respuestas contextualmente relevantes.

  • Categorización y clasificación de consultas.
  • Búsqueda de similitud vectorial en Qdrant.
  • Generación de respuestas conscientes del contexto.
  • Manejo de citas de documentos.

Componentes principales

1. Procesamiento de consultas

  • query_categorization: Clasifica las consultas en categorías predefinidas (manual técnico, protocolo de seguridad, etc.).
  • query: Orquesta la tubería RAG completa desde la consulta hasta la respuesta final.

2. Búsqueda y recuperación

  • qdrant_search: Realiza una búsqueda semántica utilizando embeddings de consulta, filtra los resultados por categoría de documento y devuelve los k documentos más relevantes.

3. Generación de respuestas

  • generate_response: Crea respuestas en lenguaje natural utilizando el contexto recuperado, utiliza LLM con prompts especializados, proporciona citas a los documentos fuente.

Modelo de categoría de consulta

Un modelo Pydantic que define la estructura para la categorización de consultas, utilizado por RAGAgent para clasificar las consultas en categorías relevantes (technical_manual, safety_protocol, etc.).

# Define category model for query classification
class Category(BaseModel):
   category: str
class RAGAgent:
    """
    Agent responsible for Retrieval-Augmented Generation (RAG) operations.
    """
    def __init__(self, mistral_client: Mistral, qdrant_client: QdrantClient):
        self.mistral_client = mistral_client
        self.qdrant_client = qdrant_client

    def generate_response(self, context: str, query: str) -> str:
        """Generate a response based on the given context and query."""
        chat_response = self.mistral_client.chat.complete(
            model=model,
            messages=[
                {
                    "role": "user",
                    "content": response_generation_prompt.format(context=context, query=query)
                },
            ]
        )
        return chat_response.choices[0].message.content

    def query_categorization(self, query: str) -> str:
        """Categorize the query into predefined categories."""
        chat_response = self.mistral_client.chat.parse(
            model=model,
            messages=[
                {
                    "role": "system",
                    "content": "Classify the query into one or more categories of the following list: ['technical_manual', 'safety_protocol', 'maintenance_guide', 'troubleshooting_guide']"
                },
                {
                    "role": "user",
                    "content": query,
                },
            ],
            response_format=Category,
            max_tokens=256,
            temperature=0
        )
        return json.loads(chat_response.choices[0].message.content)

    def qdrant_search(self, query: str, category: str = None, top_k: int = 5) -> List[Dict[str, Any]]:
        """Search for similar texts in Qdrant based on the query and category."""
        query_vector = DataProcessor(self.mistral_client, self.qdrant_client).get_text_embedding([query])

        retrieval_results = self.qdrant_client.search(
            collection_name="embeddings",
            query_vector=query_vector,
            query_filter=Filter(
                must=[
                    FieldCondition(
                        key='category',
                        match=MatchValue(value=category)
                    )
                ]
            ),
            limit=top_k
        )
        return retrieval_results

    def query(self, query_text: str, top_k: int = 3) -> Tuple[str, str]:
        """Process a natural language query using RAG."""
        category = self.query_categorization(query_text)["category"]

        results = self.qdrant_search(query_text, category, top_k=top_k)

        file_paths = [result.payload["filepath"] for result in results]

        retrieved_text = "\n".join([result.payload["text"] for result in results])
        citations = ",".join([result.payload["filepath"] for result in results])

        return self.generate_response(retrieved_text, query_text), citations

Herramientas de consulta de base de datos

Define herramientas de función de consulta de base de datos. Estas herramientas definen la interfaz de llamada a funciones para DatabaseQueryAgent, lo que permite la consulta estructurada de diferentes tablas de bases de datos.

tools = [
    {
        "type": "function",
        "function": {
            "name": "query_compliance",
            "description": '''Query compliance records with filters. \n\n A sample example of columns and corresponding values from db are:\n\n EquipmentID,EquipmentName,Manufacturer,Model,ComplianceType,Certification,IssueDate,ExpiryDate,ComplianceStatus,ResponsiblePerson
1,CNC Machine,ABC Corp,Model X,Safety,ISO 9001,2020-01-15,2025-01-15,Active,John Doe''',
            "parameters": {
                "type": "object",
                "properties": {
                    "filters": {
                        "type": "object",
                        "description": '''Dictionary of column names and values to filter by.''',
                        "additionalProperties": {
                            "type": "string"
                        }
                    }
                },
                "required": ["filters"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "query_maintenance",
            "description": '''Query maintenance records with filters. \n\n A sample example of columns and corresponding values from db are:\n\n EquipmentID,EquipmentName,Manufacturer,Model,InstallationDate,LastMaintenanceDate,NextMaintenanceDate,MaintenanceType,MaintenanceDetails,MaintenanceStatus,ResponsibleTechnician
1,CNC Machine,ABC Corp,Model X,2020-01-15,2023-09-01,2023-12-01,Preventive,Oil change,Completed,John Doe''',
            "parameters": {
                "type": "object",
                "properties": {
                    "filters": {
                        "type": "object",
                        "description": "Dictionary of column names and values to filter by",
                        "additionalProperties": {
                            "type": "string"
                        }
                    }
                },
                "required": ["filters"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "query_technical_specs",
            "description": '''Query technical specifications with filters.\n\n A sample example of columns and corresponding values from db are:\n\n EquipmentID,EquipmentName,Manufacturer,Model,SpecificationType,SpecificationDetail,Unit,Value,DateMeasured,MeasuredBy
1,CNC Machine,ABC Corp,Model X,Power,Motor Power,kW,15,2023-01-15,John Doe''',
            "parameters": {
                "type": "object",
                "properties": {
                    "filters": {
                        "type": "object",
                        "description": "Dictionary of column names and values to filter by",
                        "additionalProperties": {
                            "type": "string"
                        }
                    }
                },
                "required": ["filters"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "query_parts_inventory_compatibility",
            "description": '''Query parts, inventory and compatibility with filters.\n\n A sample example of columns and corresponding values from db are:\n\n PartID,PartName,EquipmentID,EquipmentName,Manufacturer,Model,PartType,Quantity,Compatibility,Supplier,LastOrderDate,NextOrderDate,PartStatus
1,Oil Filter,1,CNC Machine,ABC Corp,Model X,Filter,50,Compatible,Supplier A,2023-01-15,2023-12-01,In Stock''',
            "parameters": {
                "type": "object",
                "properties": {
                    "filters": {
                        "type": "object",
                        "description": "Dictionary of column names and values to filter by",
                        "additionalProperties": {
                            "type": "string"
                        }
                    }
                },
                "required": ["filters"],
            },
        },
    }
]

DatabaseQueryAgent

La clase DatabaseQueryAgent gestiona las interacciones con la base de datos SQLite, manejando consultas de datos estructurados a través de llamadas a funciones en varias bases de datos que contienen registros de mantenimiento, especificaciones técnicas, inventarios de piezas y registros de cumplimiento. Proporciona capacidades de consulta especializadas para diferentes tablas de bases de datos.

  • Procesamiento de consultas en lenguaje natural.
  • Consulta estructurada de bases de datos.
  • Llamada a funciones para la ejecución de consultas.
  • Formato de respuesta JSON.

Componentes principales

1. Consultas específicas de la tabla

  • query_compliance: Recupera registros de cumplimiento filtrados.
  • query_maintenance: Accede a información relacionada con el mantenimiento.
  • query_technical_specs: Obtiene especificaciones técnicas.
  • query_parts_inventory_compatibility: Recupera datos de piezas y compatibilidad.

2. Procesamiento de consultas

  • query: Procesa consultas en lenguaje natural utilizando llamadas a funciones, maneja llamadas a herramientas para operaciones de base de datos apropiadas, rastrea citas de herramientas de base de datos, formatea respuestas con resultados de consultas.
class DatabaseQueryAgent:
    """
    Agent responsible for interacting with the SQLite database.
    """

    def __init__(self, db_path: str, mistral_client: Mistral):
        self.db_path = db_path
        self.tools = tools
        self.names_to_functions = {
            'query_compliance': functools.partial(self.query_compliance),
            'query_maintenance': functools.partial(self.query_maintenance),
            'query_technical_specs': functools.partial(self.query_technical_specs),
            'query_parts_inventory_compatibility': functools.partial(self.query_parts_inventory_compatibility)
        }
        self.mistral_client = mistral_client

    def query_compliance(self, filters: Dict[str, str]) -> str:
        """
        Query compliance table with filters.

        Args:
            filters (Dict[str, str]): Dictionary of column names and values to filter by.

        Returns:
            str: The query result in JSON format.
        """
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            where_conditions = []
            params = []
            for column, value in filters.items():
                where_conditions.append(f"{column} = ?")
                params.append(value)
            where_clause = " AND ".join(where_conditions)
            query = f"SELECT * FROM compliance WHERE {where_clause}"
            cursor.execute(query, params)
            columns = [description[0] for description in cursor.description]
            result = cursor.fetchone()
            if result:
                record = dict(zip(columns, result))
                return json.dumps({'result': record})
            return json.dumps({'error': 'No matching records found'})
        except Exception as e:
            return json.dumps({'error': str(e)})
        finally:
            conn.close()

    def query_maintenance(self, filters: Dict[str, str]) -> str:
        """
        Query maintenance table with filters.

        Args:
            filters (Dict[str, str]): Dictionary of column names and values to filter by.

        Returns:
            str: The query result in JSON format.
        """
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            where_conditions = []
            params = []
            for column, value in filters.items():
                where_conditions.append(f"{column} = ?")
                params.append(value)
            where_clause = " AND ".join(where_conditions)
            query = f"SELECT * FROM maintenance WHERE {where_clause}"
            cursor.execute(query, params)
            columns = [description[0] for description in cursor.description]
            result = cursor.fetchone()
            if result:
                record = dict(zip(columns, result))
                return json.dumps({'result': record})
            return json.dumps({'error': 'No matching records found'})
        except Exception as e:
            return json.dumps({'error': str(e)})
        finally:
            conn.close()

    def query_technical_specs(self, filters: Dict[str, str]) -> str:
        """
        Query technical specifications table with filters.

        Args:
            filters (Dict[str, str]): Dictionary of column names and values to filter by.

        Returns:
            str: The query result in JSON format.
        """
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            where_conditions = []
            params = []
            for column, value in filters.items():
                where_conditions.append(f"{column} = ?")
                params.append(value)
            where_clause = " AND ".join(where_conditions)
            query = f"SELECT * FROM technical_specifications WHERE {where_clause}"
            cursor.execute(query, params)
            columns = [description[0] for description in cursor.description]
            result = cursor.fetchone()
            if result:
                record = dict(zip(columns, result))
                return json.dumps({'result': record})
            return json.dumps({'error': 'No matching records found'})
        except Exception as e:
            return json.dumps({'error': str(e)})
        finally:
            conn.close()

    def query_parts_inventory_compatibility(self, filters: Dict[str, str]) -> str:
        """
        Query parts inventory and compatibility table with filters.

        Args:
            filters (Dict[str, str]): Dictionary of column names and values to filter by.

        Returns:
            str: The query result in JSON format.
        """
        try:
            conn = sqlite3.connect(self.db_path)
            cursor = conn.cursor()
            where_conditions = []
            params = []
            for column, value in filters.items():
                where_conditions.append(f"{column} = ?")
                params.append(value)
            where_clause = " AND ".join(where_conditions)
            query = f"SELECT * FROM parts_inventory_compatibility WHERE {where_clause}"
            cursor.execute(query, params)
            columns = [description[0] for description in cursor.description]
            result = cursor.fetchone()
            if result:
                record = dict(zip(columns, result))
                return json.dumps({'result': record})
            return json.dumps({'error': 'No matching records found'})
        except Exception as e:
            return json.dumps({'error': str(e)})
        finally:
            conn.close()

    def query(self, query_text: str) -> str:
      """
      Process a natural language query using the database tools.

      Args:
          query_text (str): Natural language query

      Returns:
          str: Response from the database query
      """
      messages = [{"role": "user", "content": query_text}]

      # Get initial response with potential tool calls
      response = self.mistral_client.chat.complete(
          model=model,
          messages=messages,
          tools=self.tools,
          tool_choice="any",
      )
      messages.append(response.choices[0].message)

      citations = set()

      # Handle any tool calls
      if hasattr(response.choices[0].message, 'tool_calls') and response.choices[0].message.tool_calls:
          for tool_call in response.choices[0].message.tool_calls:
              function_name = tool_call.function.name
              print(f"Tool call: {function_name}")
              citations.add(function_name)
              function_params = json.loads(tool_call.function.arguments)
              print(f"Tool call parameters: {function_params}")
              function_result = self.names_to_functions[function_name](**function_params)
              messages.append({
                  "role": "tool",
                  "name": function_name,
                  "content": function_result,
                  "tool_call_id": tool_call.id
              })

      # Get final response
      final_response = self.mistral_client.chat.complete(
          model="mistral-small-latest",
          messages=messages
      )
      return final_response.choices[0].message.content, ",".join(list((citations)))

WorkflowOrchestrator

La clase WorkflowOrchestrator orquesta la interacción entre RAGAgent y DatabaseQueryAgent para proporcionar respuestas completas combinando información de fuentes de datos estructuradas y no estructuradas.

  • Orquestación y coordinación del flujo de trabajo.
  • Combinación e integración de respuestas.
  • Resumen de la respuesta final.
  • Gestión de citas de fuentes.

Componentes principales

1. Ejecución del flujo de trabajo

  • workflow: Gestiona la tubería completa de procesamiento de consultas, coordina las respuestas de ambos agentes, genera la respuesta unificada final, mantiene la trazabilidad a través de las citas.

2. Resumen de la respuesta

  • combine_and_summarize_responses: Fusiona y resume las respuestas de ambos agentes, aplica formato estructurado a las respuestas combinadas, utiliza prompts de resumen para una salida coherente.
class WorkflowOrchestrator:
    """
    WorkflowOrchestrator is responsible for orchestrating the workflow between RAGSearchAgent and DatabaseQueryAgent.
    Handles query processing, response combination, and final summarization.
    """
    def __init__(self,
                 rag_agent: RAGAgent,
                 db_query_agent: DatabaseQueryAgent,
                 client: Mistral):
        """
        Initialize WorkflowOrchestrator with necessary components.

        Args:
            rag_agent: RAGSearchAgent for document retrieval and generation
            db_query_agent: DatabaseQueryAgent for structured data queries
            client: Mistral client for text generation
        """
        self.rag_agent = rag_agent
        self.db_query_agent = db_query_agent
        self.client = client

    def combine_and_summarize_responses(self,
                                      responses: Dict[str, str],
                                      query: str,
                                      summarization_prompt: str = summarization_prompt) -> str:
        """
        Combine and summarize multiple responses into a coherent final response.

        Args:
            responses: Dictionary of response types and their content
            query: Original user query
            summarization_prompt: Template for summarization

        Returns:
            str: Summarized and combined response
        """
        # Format responses into a structured text
        combined_text = "\n\n".join([
            f"{source}: {content}"
            for source, content in responses.items()
        ])

        # Generate summarized response
        chat_response = self.client.chat.complete(
            model=model,
            messages=[
                {
                    "role": "system",
                    "content": summarization_prompt
                },
                {
                    "role": "user",
                    "content": f"Query: {query}\n\nResponses:\n{combined_text}"
                },
            ],
            temperature=0
        )
        return chat_response.choices[0].message.content

    def workflow(self, query: str) -> str:
        """
        Execute the workflow for processing a query.

        Args:
            query: User query

        Returns:
            str: Final response with citations
        """
        # Get responses from both agents
        db_response, tools_citations = self.db_query_agent.query(query)
        rag_response, rag_citations = self.rag_agent.query(query)

        # Combine responses into a dictionary
        responses = {
            "Database Response": db_response,
            "RAG Response": rag_response
        }

        # Generate final summarized response
        final_response = self.combine_and_summarize_responses(
            responses=responses,
            query=query,
            summarization_prompt=final_response_summarization_prompt
        )

        # Add citations
        citations = (
            f"\n\nSources:\n"
            f"- Database Tools: {((tools_citations))}\n"
            f"- PDF Sources: {rag_citations}"
        )

        return final_response + citations

Inicializar y procesar documentos

Inicializa DataProcessor y procesa documentos PDF a través de la tubería completa, desde la ingesta de archivos hasta el almacenamiento de embeddings.

# Initialize the processor
doc_processor = DataProcessor(mistral_client, qdrant_client)

# Process documents
file_list = doc_processor.get_categorized_filepaths(root_dir='./pdf_data')
processed_docs = doc_processor.process_documents(file_list)
doc_processor.process_and_store_embeddings(processed_docs)
Processing PDFs: 100%|██████████| 12/12 [00:24<00:00,  2.07s/it]
Processed batch 1/2
Processed batch 2/2

Insertar datos en tablas de la base de datos.

Carga múltiples archivos CSV en sus tablas de base de datos correspondientes en SQLite.

# Insert data into tables

db_path = "./database.db"

file_mappings = {
    "compliance": "./csv_data/compliance_db.csv",
    "maintenance": "./csv_data/maintenance_db.csv",
    "technical_specifications": "./csv_data/technical_specifications_db.csv",
    "parts_inventory_compatibility": "./csv_data/parts_inventory_compatibility_db.csv"
}

doc_processor.insert_data_database(db_path, file_mappings)
Successfully inserted data into compliance
Successfully inserted data into maintenance
Successfully inserted data into technical_specifications
Successfully inserted data into parts_inventory_compatibility

Inicializar los agentes

Inicializa los tres agentes principales:

  • RAGAgent para búsqueda de documentos y respuesta.
  • DatabaseQueryAgent para consultas de datos estructurados.
  • WorkflowAgent para orquestar respuestas.
rag_agent = RAGAgent(mistral_client, qdrant_client)
db_query_agent = DatabaseQueryAgent(db_path, mistral_client)
workflow_orchestrator = WorkflowOrchestrator(rag_agent, db_query_agent, mistral_client)

Consultas de ejemplo

query = "What are the troubleshooting steps for inaccurate machining in CNC Machine (Model X) and when was its last maintenance performed?"

print(f"Query: {query}")
print("----------------------")

answer = workflow_orchestrator.workflow(query)

print("------------Answer----------")
print(answer)
Query: What are the troubleshooting steps for inaccurate machining in CNC Machine (Model X) and when was its last maintenance performed?
----------------------
Tool call: query_maintenance
Tool call parameters: {'filters': {'EquipmentName': 'CNC Machine', 'Model': 'Model X'}}
Tool call: query_technical_specs
Tool call parameters: {'filters': {'EquipmentName': 'CNC Machine', 'Model': 'Model X', 'SpecificationType': 'Accuracy'}}
------------Answer----------
### Troubleshooting Steps for Inaccurate Machining in CNC Machine (Model X)

To address inaccurate machining in the CNC Machine (Model X), follow these comprehensive troubleshooting steps:

1. **Check the Program**:
   - Ensure that the CNC program is correct and free of errors. Even minor errors in the code can lead to significant inaccuracies in machining.

2. **Inspect the Tooling**:
   - Verify that the correct tools are being used and that they are in good condition. Worn or damaged tools can cause inaccuracies. Replace tools as needed.

3. **Verify the Setup**:
   - Check the workpiece setup to ensure it is secure and correctly positioned. Any movement or misalignment can lead to machining inaccuracies.

4. **Calibrate the Machine**:
   - Regular calibration of the machine's axes and spindles can help maintain accuracy. If the machine hasn't been calibrated recently, it may be time to do so.

5. **Check for Wear and Tear**:
   - Inspect the machine for any signs of wear and tear, such as worn bearings or guides. These components can affect the machine's accuracy over time.

6. **Environmental Factors**:
   - Ensure that the machining environment is stable. Factors such as temperature, humidity, and vibration can affect the machine's performance.

7. **Machine Maintenance**:
   - Regular maintenance, including lubrication and cleaning, can help prevent inaccuracies. Ensure that the machine is well-maintained and that all scheduled maintenance tasks are completed on time.

8. **Software and Firmware**:
   - Ensure that the machine's software and firmware are up-to-date. Outdated software can sometimes cause inaccuracies.

9. **Backlash Compensation**:
   - Check the backlash compensation settings. Incorrect settings can lead to inaccuracies, especially in high-precision machining.

10. **Consult the Manual**:
    - Refer to the machine's manual for any model-specific troubleshooting steps or recommendations.

### Last Maintenance Performed

The last maintenance for the CNC Machine (Model X) was performed on September 1, 2023. This maintenance was preventive and included an oil change. The next scheduled maintenance is set for December 1, 2023. For more detailed information, refer to the Maintenance Log in the Appendices section of the machine's documentation.

### Important Safety Precautions

- Always ensure the machine is turned off and locked out before performing any maintenance or inspection tasks.
- Wear appropriate personal protective equipment (PPE) when handling tools and machinery.
- Follow the manufacturer's guidelines for tool replacement and machine calibration to avoid any potential hazards.

By following these steps and maintaining a regular maintenance schedule, you can help ensure the accuracy and reliability of the CNC Machine (Model X).

Sources:
- Database Tools: query_technical_specs,query_maintenance
- PDF Sources: ./pdf_data/troubleshooting_guide/CNC_Machine_(Model X)_Troubleshooting_Guide.pdf,./pdf_data/troubleshooting_guide/Robotic_Arm_(Unit 7)_Troubleshooting_Guide.pdf,./pdf_data/troubleshooting_guide/Cooling_System_(Model Y)_Troubleshooting_Guide.pdf
query = "What are the safety protocols for the Cooling System (Model Y), and when is its next scheduled maintenance?"

print(f"Query: {query}")
print("----------------------")

answer = workflow_orchestrator.workflow(query)

print("------------Answer----------")
print(answer)
Query: What are the safety protocols for the Cooling System (Model Y), and when is its next scheduled maintenance?
----------------------
Tool call: query_compliance
Tool call parameters: {'filters': {'EquipmentName': 'Cooling System', 'Model': 'Model Y', 'ComplianceType': 'Safety'}}
Tool call: query_maintenance
Tool call parameters: {'filters': {'EquipmentName': 'Cooling System', 'Model': 'Model Y'}}
------------Answer----------
### Safety Protocols for Cooling System (Model Y)

The Cooling System (Model Y) is compliant with OSHA safety standards, issued on March 10, 2021, and is set to expire on March 10, 2026. The system is currently active and Alice Johnson is responsible for its compliance. Below are the detailed safety protocols and maintenance schedule for the Cooling System (Model Y).

#### Personal Protective Equipment (PPE)
- **Required PPE:**
  - Safety Glasses
  - Gloves
  - Ear Protection
  - Safety Shoes
- **PPE Usage:**
  - Always wear the required PPE when operating or maintaining the system.
  - Ensure PPE is in good condition and fits properly.

#### System Operation Safety
- **Pre-Operation Checks:**
  - Inspect System: Check for any visible damage or issues.
  - Check Coolant Levels: Ensure coolant levels are adequate.
  - Check Fans: Ensure fans are functioning properly.
- **During Operation:**
  - Monitor System: Keep a close eye on the system during operation.
  - Listen for Unusual Noises: Stop the system if you hear any unusual noises.
  - Check Cooling Performance: Ensure the system is cooling effectively.
- **Post-Operation Checks:**
  - Clean System: Clean the system and work area.
  - Inspect Components: Check components for wear and replace as needed.
  - Record Issues: Document any issues or anomalies observed during operation.

#### Emergency Procedures
- **System Failure:**
  1. Stop Operation: Immediately stop the system.
  2. Notify Supervisor: Inform your supervisor or maintenance team.
  3. Document Issue: Record the details of the issue for further investigation.
- **Injury:**
  1. Provide First Aid: Administer first aid as needed.
  2. Notify Safety Team: Inform the safety team or emergency services.
  3. Document Incident: Record the details of the incident for further investigation.
- **Fire:**
  1. Evacuate Area: Clear the area around the system.
  2. Use Fire Extinguisher: Use the appropriate fire extinguisher to put out the fire.
  3. Notify Safety Team: Inform the safety team or emergency services.

#### Hazardous Materials
- **Coolant:**
  - Handling: Wear gloves and safety glasses when handling coolant.
  - Disposal: Dispose of used coolant according to local regulations.
- **Lubricants:**
  - Handling: Wear gloves and safety glasses when handling lubricants.
  - Disposal: Dispose of used lubricants according to local regulations.

#### Environmental Safety
- **Ventilation:**
  - Ensure the work area has adequate ventilation to prevent the buildup of fumes and heat.
- **Noise Levels:**
  - Use ear protection to protect hearing from loud noises.
- **Waste Management:**
  - Dispose of waste materials according to local regulations.

### Next Scheduled Maintenance
The next scheduled maintenance for the Cooling System (Model Y) is on October 10, 2023. This maintenance is preventive in nature and involves filter cleaning. The maintenance is currently scheduled, and Alice Johnson is the responsible technician.

Sources:
- Database Tools: query_compliance,query_maintenance
- PDF Sources: ./pdf_data/safety_protocol/Cooling_System_(Model Y)_Safety_Protocol.pdf,./pdf_data/safety_protocol/CNC_Machine_(Model X)_Safety_Protocol.pdf,./pdf_data/safety_protocol/Robotic_Arm_(Unit 7)_Safety_Protocol.pdf
Lección del curso «Mistral Cookbook» de Mistral AI, publicado con licencia MIT. Traducción y adaptación al español de IA con Clase. IA con Clase no está afiliado a Mistral AI. Ver el original · Licencia
Esta lección es gratuita. El resto del curso se abre con la Membresía de IA con Clase, que incluye todos los cursos del catálogo. Ver precios
← AnteriorSiguiente: Flujo de trabajo multiagente para contratación →