LLM in Azienda: RAG Enterprise, Fine-Tuning e Guardrails

# rag_enterprise_pipeline.py
# Pipeline RAG production-ready per enterprise
# Requisiti: langchain>=0.2.0, pinecone-client>=3.0, openai>=1.0

import os
import hashlib
import logging
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_pinecone import PineconeVectorStore
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain_community.cross_encoders import HuggingFaceCrossEncoder
from pinecone import Pinecone, ServerlessSpec

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class RAGConfig:
    """Configurazione centralizzata per pipeline RAG enterprise."""
    # Model settings
    embedding_model: str = "text-embedding-3-large"
    llm_model: str = "gpt-4o"
    temperature: float = 0.1

    # Retrieval settings
    chunk_size: int = 512
    chunk_overlap: int = 64
    top_k_retrieval: int = 10
    top_k_rerank: int = 4

    # Vector store
    pinecone_index: str = "enterprise-knowledge"
    pinecone_dimension: int = 3072  # text-embedding-3-large

    # Quality settings
    min_relevance_score: float = 0.7
    max_context_tokens: int = 8000


class EnterpriseRAGPipeline:
    """
    Pipeline RAG enterprise con:
    - Chunking adattivo per documenti aziendali
    - Re-ranking semantico con cross-encoder
    - Filtraggio per rilevanza minima
    - Citazioni delle fonti
    - Cache embedding per ridurre costi API
    """

    def __init__(self, config: RAGConfig):
        self.config = config
        self._setup_components()

    def _setup_components(self):
        """Inizializza tutti i componenti della pipeline."""
        # Embeddings con cache locale
        self.embeddings = OpenAIEmbeddings(
            model=self.config.embedding_model,
            dimensions=self.config.dimension
        )

        # LLM con temperature bassa per risposte precise
        self.llm = ChatOpenAI(
            model=self.config.llm_model,
            temperature=self.config.temperature,
            max_tokens=2048
        )

        # Pinecone vector store
        pc = Pinecone(api_key=os.environ["PINECONE_API_KEY"])

        # Crea index se non esiste
        if self.config.pinecone_index not in pc.list_indexes().names():
            pc.create_index(
                name=self.config.pinecone_index,
                dimension=self.config.pinecone_dimension,
                metric="cosine",
                spec=ServerlessSpec(cloud="aws", region="us-east-1")
            )

        index = pc.Index(self.config.pinecone_index)
        self.vector_store = PineconeVectorStore(
            index=index,
            embedding=self.embeddings
        )

        # Cross-encoder per re-ranking (migliora qualità retrieval del 30-40%)
        reranker_model = HuggingFaceCrossEncoder(
            model_name="cross-encoder/ms-marco-MiniLM-L-6-v2"
        )
        self.reranker = CrossEncoderReranker(
            model=reranker_model,
            top_n=self.config.top_k_rerank
        )

        # Retriever con re-ranking
        base_retriever = self.vector_store.as_retriever(
            search_type="similarity",
            search_kwargs={"k": self.config.top_k_retrieval}
        )
        self.retriever = ContextualCompressionRetriever(
            base_compressor=self.reranker,
            base_retriever=base_retriever
        )

        # Prompt template enterprise con istruzioni precise
        self.prompt = PromptTemplate(
            template="""Sei un assistente aziendale esperto. Usa SOLO le informazioni
nel contesto seguente per rispondere alla domanda. Se la risposta non e nel contesto,
dillo esplicitamente. Non inventare mai informazioni.

CONTESTO:
{context}

DOMANDA: {question}

RISPOSTA (cita le fonti specifiche quando possibile):""",
            input_variables=["context", "question"]
        )

        # Chain QA completa
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.retriever,
            chain_type_kwargs={"prompt": self.prompt},
            return_source_documents=True
        )

    def ingest_documents(
        self,
        documents: List[Dict],
        batch_size: int = 100
    ) -> int:
        """
        Indicizza documenti aziendali nel vector store.

        Args:
            documents: Lista di dict con 'content', 'metadata', 'source'
            batch_size: Documenti per batch (ottimizza costi API)

        Returns:
            Numero di chunk indicizzati
        """
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.config.chunk_size,
            chunk_overlap=self.config.chunk_overlap,
            separators=["\n\n", "\n", ". ", " ", ""]
        )

        total_chunks = 0
        batch = []

        for doc in documents:
            # Crea hash per deduplication
            content_hash = hashlib.md5(
                doc["content"].encode()
            ).hexdigest()

            chunks = splitter.create_documents(
                [doc["content"]],
                metadatas=[{
                    **doc.get("metadata", {}),
                    "source": doc["source"],
                    "content_hash": content_hash
                }]
            )

            batch.extend(chunks)

            if len(batch) >= batch_size:
                self.vector_store.add_documents(batch)
                total_chunks += len(batch)
                logger.info(f"Indicizzati {total_chunks} chunk")
                batch = []

        # Processa batch rimanente
        if batch:
            self.vector_store.add_documents(batch)
            total_chunks += len(batch)

        return total_chunks

    def query(
        self,
        question: str,
        filters: Optional[Dict] = None
    ) -> Dict:
        """
        Esegui una query sulla knowledge base aziendale.

        Args:
            question: Domanda in linguaggio naturale
            filters: Filtri metadata (es. {"department": "legal"})

        Returns:
            Dict con answer, sources, confidence
        """
        # Applica filtri se presenti
        if filters:
            self.retriever.base_retriever.search_kwargs["filter"] = filters

        result = self.qa_chain.invoke({"query": question})

        # Estrai fonti uniche
        sources = list(set([
            doc.metadata.get("source", "unknown")
            for doc in result["source_documents"]
        ]))

        return {
            "answer": result["result"],
            "sources": sources,
            "num_docs_retrieved": len(result["source_documents"])
        }


# Utilizzo enterprise
if __name__ == "__main__":
    config = RAGConfig()
    pipeline = EnterpriseRAGPipeline(config)

    # Indicizza documentazione aziendale
    docs = [
        {
            "content": "La policy aziendale prevede 30 giorni di ferie annuali...",
            "source": "hr-policy-2025.pdf",
            "metadata": {"department": "HR", "version": "2025.1"}
        },
        # ... altri documenti
    ]
    n_chunks = pipeline.ingest_documents(docs)
    print(f"Indicizzati {n_chunks} chunk")

    # Query con filtro dipartimento
    result = pipeline.query(
        question="Quanti giorni di ferie ho diritto?",
        filters={"department": "HR"}
    )
    print(f"Risposta: {result['answer']}")
    print(f"Fonti: {result['sources']}")

# fine_tuning_lora.py
# Fine-tuning efficiente con LoRA per LLM enterprise
# Requisiti: transformers>=4.40, peft>=0.10, trl>=0.8

import torch
from datasets import Dataset
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    BitsAndBytesConfig
)
from peft import LoraConfig, get_peft_model, TaskType
from trl import SFTTrainer
import json


def prepare_training_data(raw_examples: list) -> Dataset:
    """
    Prepara dati di training nel formato chat per instruction tuning.

    Args:
        raw_examples: Lista di dict con 'instruction', 'input', 'output'

    Returns:
        Dataset HuggingFace pronto per training
    """
    def format_example(example: dict) -> dict:
        # Formato Alpaca/chat standard
        if example.get("input"):
            text = f"""### Istruzione:
{example['instruction']}

### Input:
{example['input']}

### Risposta:
{example['output']}"""
        else:
            text = f"""### Istruzione:
{example['instruction']}

### Risposta:
{example['output']}"""
        return {"text": text}

    formatted = [format_example(ex) for ex in raw_examples]
    return Dataset.from_list(formatted)


def create_lora_model(
    base_model_name: str = "mistralai/Mistral-7B-Instruct-v0.3",
    lora_rank: int = 16,
    lora_alpha: int = 32,
    quantize: bool = True
):
    """
    Carica modello base con configurazione LoRA.

    Parametri LoRA:
    - rank (r=16): Dimensione matrici adattamento. Più alto = più espressivita
      ma più parametri (default: 8-32 per enterprise)
    - alpha (32): Scala learning rate LoRA. Tipicamente 2x rank.
    - target_modules: Layer da addestrare (q/v attention per Mistral)
    """
    # Quantizzazione 4-bit per ridurre VRAM (da 16GB a 6GB per 7B params)
    bnb_config = None
    if quantize:
        bnb_config = BitsAndBytesConfig(
            load_in_4bit=True,
            bnb_4bit_quant_type="nf4",
            bnb_4bit_compute_dtype=torch.float16,
            bnb_4bit_use_double_quant=True
        )

    # Carica modello base
    model = AutoModelForCausalLM.from_pretrained(
        base_model_name,
        quantization_config=bnb_config,
        device_map="auto",
        torch_dtype=torch.float16
    )
    tokenizer = AutoTokenizer.from_pretrained(base_model_name)
    tokenizer.pad_token = tokenizer.eos_token

    # Configurazione LoRA
    lora_config = LoraConfig(
        task_type=TaskType.CAUSAL_LM,
        r=lora_rank,
        lora_alpha=lora_alpha,
        lora_dropout=0.1,
        # Solo questi layer: riduce parametri trainable del 95%+
        target_modules=[
            "q_proj", "v_proj", "k_proj", "o_proj",
            "gate_proj", "up_proj", "down_proj"
        ],
        bias="none"
    )

    # Applica LoRA al modello
    model = get_peft_model(model, lora_config)
    trainable, total = model.get_nb_trainable_parameters()
    print(f"Parametri trainable: {trainable:,} / {total:,} "
          f"({100 * trainable / total:.2f}%)")
    # Output tipico: "Parametri trainable: 6,815,744 / 7,248,220,160 (0.09%)"

    return model, tokenizer


def run_fine_tuning(
    model,
    tokenizer,
    dataset: Dataset,
    output_dir: str = "./fine_tuned_model"
):
    """Esegui il fine-tuning con SFTTrainer."""
    training_args = TrainingArguments(
        output_dir=output_dir,
        num_train_epochs=3,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,  # Effective batch = 16
        learning_rate=2e-4,
        warmup_ratio=0.03,
        lr_scheduler_type="cosine",
        logging_steps=10,
        save_strategy="epoch",
        evaluation_strategy="epoch",
        fp16=True,
        report_to="mlflow",  # Traccia esperimenti
        run_name="enterprise-lora-ft"
    )

    trainer = SFTTrainer(
        model=model,
        tokenizer=tokenizer,
        args=training_args,
        train_dataset=dataset,
        dataset_text_field="text",
        max_seq_length=2048,
        packing=False  # True per dataset omogenei (più veloce)
    )

    trainer.train()
    trainer.save_model(output_dir)
    print(f"Modello salvato in {output_dir}")


# Esempio utilizzo per tone-of-voice aziendale
if __name__ == "__main__":
    # Esempi di training per assistente legale in stile formale
    examples = [
        {
            "instruction": "Riassumi il contratto in linguaggio formale",
            "input": "Il fornitore deve consegnare la merce entro 30 giorni...",
            "output": "Con la presente si notifica che il fornitore e contrattualmente obbligato..."
        },
        # ... minimo 1000 esempi per risultati accettabili
    ]

    dataset = prepare_training_data(examples)
    model, tokenizer = create_lora_model()
    run_fine_tuning(model, tokenizer, dataset)

# docker-compose.yml
# Deployment vLLM enterprise con monitoring e load balancing

version: '3.8'

services:
  # vLLM API Server (replica x2 per alta disponibilità)
  vllm-primary:
    image: vllm/vllm-openai:latest
    command: >
      python -m vllm.entrypoints.openai.api_server
      --model mistralai/Mistral-7B-Instruct-v0.3
      --quantization awq
      --max-model-len 8192
      --gpu-memory-utilization 0.85
      --port 8000
      --host 0.0.0.0
      --api-key ${VLLM_API_KEY}
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    environment:
      - HUGGING_FACE_HUB_TOKEN=${HF_TOKEN}
    ports:
      - "8000:8000"
    volumes:
      - model-cache:/root/.cache/huggingface
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
    restart: unless-stopped

  vllm-secondary:
    # Replica con stessa config per load balancing
    extends:
      service: vllm-primary
    ports:
      - "8001:8000"

  # Nginx reverse proxy con load balancing
  nginx:
    image: nginx:alpine
    ports:
      - "443:443"
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - ./certs:/etc/nginx/certs:ro
    depends_on:
      - vllm-primary
      - vllm-secondary
    restart: unless-stopped

  # Prometheus per monitoring
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
      - prometheus-data:/prometheus
    ports:
      - "9090:9090"
    restart: unless-stopped

  # Grafana per dashboard
  grafana:
    image: grafana/grafana:latest
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=${GRAFANA_PASSWORD}
    volumes:
      - grafana-data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards:ro
    depends_on:
      - prometheus
    restart: unless-stopped

volumes:
  model-cache:
  prometheus-data:
  grafana-data:

---
# nginx.conf - Load balancing con health check
# upstream vllm_backend {
#     least_conn;
#     server vllm-primary:8000 max_fails=3 fail_timeout=30s;
#     server vllm-secondary:8000 max_fails=3 fail_timeout=30s;
# }

# enterprise_guardrails.py
# Sistema guardrails enterprise per LLM production
# Requisiti: nemoguardrails>=0.8, presidio-analyzer>=2.2, openai>=1.0

import re
import json
import logging
from typing import Optional, Dict, List, Tuple
from dataclasses import dataclass
from enum import Enum

from presidio_analyzer import AnalyzerEngine, RecognizerResult
from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import OperatorConfig

logger = logging.getLogger(__name__)


class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class GuardrailResult:
    """Risultato della validazione guardrails."""
    passed: bool
    risk_level: RiskLevel
    violations: List[str]
    anonymized_text: Optional[str] = None
    reason: Optional[str] = None


class InputGuardrails:
    """
    Guardrails per input utente:
    - Rilevamento PII (GDPR compliance)
    - Prompt injection detection
    - Topic restriction (domande fuori scope)
    - Rate limiting per utente
    """

    def __init__(self, allowed_topics: List[str] = None):
        # Presidio per rilevamento PII
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

        # Pattern prompt injection comuni
        self.injection_patterns = [
            r"ignora\s+le\s+istruzioni\s+precedenti",
            r"ignore\s+previous\s+instructions",
            r"you\s+are\s+now\s+(DAN|GPT|jailbreak)",
            r"pretend\s+you\s+(are|have no)",
            r"act\s+as\s+if\s+you",
            r"from\s+now\s+on\s+you\s+are",
            r"disregard\s+all\s+previous",
            r"system\s*:\s*you\s+are",  # Fake system prompt
            r"[INST].*[/INST]",  # Llama format injection
        ]

        # Keyword pericolose specifiche per dominio
        self.blocked_keywords = [
            "ssn", "social security", "password", "api_key",
            "private key", "seed phrase", "mnemonic"
        ]

        self.allowed_topics = allowed_topics or []

    def check_pii(self, text: str) -> Tuple[bool, str, str]:
        """
        Rileva e anonimizza PII nel testo input.

        Returns:
            (has_pii, anonymized_text, pii_types_found)
        """
        results: List[RecognizerResult] = self.analyzer.analyze(
            text=text,
            language="it",
            entities=[
                "PERSON", "EMAIL_ADDRESS", "PHONE_NUMBER",
                "CREDIT_CARD", "IBAN_CODE", "IT_FISCAL_CODE",
                "IP_ADDRESS", "URL", "MEDICAL_LICENSE"
            ]
        )

        if not results:
            return False, text, ""

        # Anonimizza con operatori specifici per tipo
        operators = {
            "PERSON": OperatorConfig("replace", {"new_value": "[NOME]"}),
            "EMAIL_ADDRESS": OperatorConfig("replace", {"new_value": "[EMAIL]"}),
            "PHONE_NUMBER": OperatorConfig("replace", {"new_value": "[TELEFONO]"}),
            "CREDIT_CARD": OperatorConfig("mask", {"chars_to_mask": 12, "from_end": False}),
            "IBAN_CODE": OperatorConfig("replace", {"new_value": "[IBAN]"}),
            "IT_FISCAL_CODE": OperatorConfig("replace", {"new_value": "[CF]"})
        }

        anonymized = self.anonymizer.anonymize(
            text=text,
            analyzer_results=results,
            operators=operators
        )

        pii_types = list(set([r.entity_type for r in results]))
        logger.warning(f"PII rilevato: {pii_types} nell'input utente")

        return True, anonymized.text, ", ".join(pii_types)

    def check_prompt_injection(self, text: str) -> Tuple[bool, str]:
        """Rileva tentativi di prompt injection."""
        text_lower = text.lower()

        for pattern in self.injection_patterns:
            if re.search(pattern, text_lower, re.IGNORECASE):
                return True, f"Pattern injection rilevato: {pattern}"

        # Check keywords pericolose
        for keyword in self.blocked_keywords:
            if keyword in text_lower:
                return True, f"Keyword bloccata: {keyword}"

        return False, ""

    def validate(self, user_input: str, user_id: str) -> GuardrailResult:
        """
        Validazione completa dell'input con tutti i guardrails.

        Returns:
            GuardrailResult con esito e dettagli violazioni
        """
        violations = []
        anonymized_text = user_input

        # 1. Check prompt injection
        is_injection, injection_reason = self.check_prompt_injection(user_input)
        if is_injection:
            return GuardrailResult(
                passed=False,
                risk_level=RiskLevel.CRITICAL,
                violations=["prompt_injection"],
                reason=injection_reason
            )

        # 2. Check e anonimizzazione PII
        has_pii, anonymized_text, pii_types = self.check_pii(user_input)
        if has_pii:
            violations.append(f"pii_detected:{pii_types}")
            logger.info(f"PII anonimizzato per utente {user_id}")

        # Input valido (PII anonimizzato se presente)
        risk = RiskLevel.LOW if not violations else RiskLevel.MEDIUM
        return GuardrailResult(
            passed=True,
            risk_level=risk,
            violations=violations,
            anonymized_text=anonymized_text
        )


class OutputGuardrails:
    """
    Guardrails per output del modello:
    - Rilevamento allucinazioni (confidence scoring)
    - Filtraggio contenuti tossici
    - Leak di dati sensibili nell'output
    - Validazione format per output strutturati
    """

    TOXIC_PATTERNS = [
        r"\b(odio|kill|violenza|terrorismo)\b",
        r"come\s+(creare|costruire|produrre)\s+(armi|esplosivi|veleni)",
    ]

    def __init__(self):
        self.analyzer = AnalyzerEngine()

    def check_output_pii(self, output: str) -> Tuple[bool, List[str]]:
        """Verifica che l'output non contenga PII non intenzionale."""
        results = self.analyzer.analyze(
            text=output,
            language="it",
            entities=["CREDIT_CARD", "IBAN_CODE", "IT_FISCAL_CODE"]
        )
        if results:
            pii_types = [r.entity_type for r in results]
            return True, pii_types
        return False, []

    def check_toxicity(self, output: str) -> Tuple[bool, str]:
        """Rilevamento contenuti tossici nell'output."""
        for pattern in self.TOXIC_PATTERNS:
            if re.search(pattern, output, re.IGNORECASE):
                return True, f"Contenuto tossico: {pattern}"
        return False, ""

    def validate(self, output: str) -> GuardrailResult:
        """Validazione completa dell'output LLM."""
        violations = []

        # Check PII nell'output
        has_pii, pii_types = self.check_output_pii(output)
        if has_pii:
            violations.append(f"output_pii:{pii_types}")
            return GuardrailResult(
                passed=False,
                risk_level=RiskLevel.HIGH,
                violations=violations,
                reason="Output contiene dati sensibili"
            )

        # Check tossicita
        is_toxic, toxic_reason = self.check_toxicity(output)
        if is_toxic:
            return GuardrailResult(
                passed=False,
                risk_level=RiskLevel.CRITICAL,
                violations=["toxic_output"],
                reason=toxic_reason
            )

        return GuardrailResult(
            passed=True,
            risk_level=RiskLevel.LOW,
            violations=[]
        )


class LLMGateway:
    """
    Gateway enterprise che integra LLM + guardrails.
    Punto centrale per tutte le chiamate LLM in azienda.
    """

    def __init__(self, llm_client, input_guardrails: InputGuardrails,
                 output_guardrails: OutputGuardrails):
        self.llm = llm_client
        self.input_guard = input_guardrails
        self.output_guard = output_guardrails

    def complete(
        self,
        user_message: str,
        user_id: str,
        system_prompt: str = "",
        max_retries: int = 1
    ) -> Dict:
        """
        Chiamata LLM con guardrails completi.

        Returns:
            {'response': str, 'input_risk': str, 'output_risk': str, 'blocked': bool}
        """
        # 1. Valida input
        input_result = self.input_guard.validate(user_message, user_id)
        if not input_result.passed:
            logger.warning(
                f"Input bloccato per {user_id}: {input_result.violations}"
            )
            return {
                "response": "Non posso elaborare questa richiesta.",
                "input_risk": input_result.risk_level.value,
                "blocked": True,
                "reason": input_result.reason
            }

        # Usa testo anonimizzato se PII trovato
        safe_input = input_result.anonymized_text or user_message

        # 2. Chiamata LLM
        messages = []
        if system_prompt:
            messages.append({"role": "system", "content": system_prompt})
        messages.append({"role": "user", "content": safe_input})

        llm_response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            max_tokens=1024,
            temperature=0.1
        )
        output_text = llm_response.choices[0].message.content

        # 3. Valida output
        output_result = self.output_guard.validate(output_text)
        if not output_result.passed:
            logger.error(
                f"Output bloccato: {output_result.violations}"
            )
            return {
                "response": "Impossibile fornire una risposta per questa richiesta.",
                "output_risk": output_result.risk_level.value,
                "blocked": True
            }

        return {
            "response": output_text,
            "input_risk": input_result.risk_level.value,
            "output_risk": output_result.risk_level.value,
            "blocked": False,
            "input_violations": input_result.violations
        }