Build Multi-Agent Quality Inspection Workflows with CrewAI and Semantic Kernel

                      
                     
"""
Production implementation for building multi-agent quality inspection workflows using CrewAI and Semantic Kernel.
Provides secure, scalable operations and integrates with external APIs for data processing.
"""

from typing import Dict, Any, List, Tuple
import os
import logging
import httpx
from pydantic import BaseModel, ValidationError
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
import asyncio

# Setting up logging for monitoring workflow execution
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class for environment variables
class Config:
    database_url: str = os.getenv('DATABASE_URL', 'sqlite:///quality_inspection.db')

# Database setup using SQLAlchemy
Base = declarative_base()

class InspectionRecord(Base):
    __tablename__ = 'inspection_records'
    id = Column(Integer, primary_key=True)
    status = Column(String)

# Create a session factory for database operations
engine = create_engine(Config.database_url)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(bind=engine)

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate the incoming data for inspections.
    
    Args:
        data: Input data to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'id' not in data:
        raise ValueError('Missing id in the data')
    if 'status' not in data:
        raise ValueError('Missing status in the data')
    return True

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields to prevent SQL injection or other security issues.
    
    Args:
        data: Data to sanitize
    Returns:
        Sanitized data
    Raises:
        ValueError: If sanitization fails
    """
    try:
        sanitized_data = {k: str(v).strip() for k, v in data.items()}
    except Exception as e:
        logger.error(f'Sanitization failed: {str(e)}')
        raise ValueError('Sanitization error')
    return sanitized_data

async def fetch_data(api_url: str) -> List[Dict[str, Any]]:
    """Fetch data from external API.
    
    Args:
        api_url: URL to fetch data from
    Returns:
        List of fetched records
    Raises:
        HTTPError: If the request to the API fails
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(api_url)
        response.raise_for_status()  # Raises an error for bad responses
        return response.json()

async def save_to_db(data: Dict[str, Any], db: Session) -> None:
    """Save a record to the database.
    
    Args:
        data: Data to save
        db: Database session
    Raises:
        Exception: If saving fails
    """
    try:
        record = InspectionRecord(**data)
        db.add(record)
        db.commit()
    except Exception as e:
        db.rollback()  # Rollback on error
        logger.error(f'Error saving to DB: {str(e)}')
        raise

async def process_batch(records: List[Dict[str, Any]]) -> None:
    """Process a batch of records for inspection.
    
    Args:
        records: List of records to process
    """
    db: Session = SessionLocal()
    for record in records:
        try:
            await validate_input(record)  # Validate the record
            sanitized_record = await sanitize_fields(record)  # Sanitize the record
            await save_to_db(sanitized_record, db)  # Save to DB
        except (ValueError, Exception) as err:
            logger.warning(f'Processing failed for record {record}: {str(err)}')
    db.close()

async def aggregate_metrics() -> Dict[str, int]:
    """Aggregate inspection metrics from the database.
    
    Returns:
        Aggregated metrics
    """
    db: Session = SessionLocal()
    count = db.query(InspectionRecord).count()
    db.close()
    return {'total_records': count}

class QualityInspectionWorkflow:
    """Main orchestrator class for the quality inspection workflow.
    
    This class ties together all the helper functions and manages the workflow.
    """

    def __init__(self, api_url: str):
        self.api_url = api_url

    async def run(self) -> None:
        """Run the entire quality inspection workflow.
        
        This method facilitates the fetching, processing, and storing of inspection records.
        """
        try:
            records = await fetch_data(self.api_url)  # Fetch data from API
            await process_batch(records)  # Process the fetched records
            metrics = await aggregate_metrics()  # Aggregate metrics
            logger.info(f'Workflow completed with metrics: {metrics}')
        except Exception as e:
            logger.error(f'Error in workflow execution: {str(e)}')

if __name__ == '__main__':
    # Example usage of the workflow
    workflow = QualityInspectionWorkflow(api_url='https://example.com/api/inspections')
    asyncio.run(workflow.run())