06 - Prompt Engineering per LLM in Produzione

# STANDARD PROMPTING - scarsa accuratezza su ragionamento complesso
standard_prompt = """
Questo cliente deve pagare una fattura di 1200 euro.
Ha già pagato il 30%. Quanto deve ancora pagare?
"""
# Risposta tipica: "840 euro" (spesso corretto ma senza garanzie)

# CHAIN-OF-THOUGHT - alta accuratezza
cot_prompt = """
Questo cliente deve pagare una fattura di 1200 euro.
Ha già pagato il 30%. Quanto deve ancora pagare?

Ragiona passo per passo:
1. Prima calcola quanto ha già pagato
2. Poi calcola il rimanente
3. Fornisci la risposta finale
"""
# Risposta:
# "1. Ha pagato: 1200 * 30/100 = 360 euro
#  2. Rimanente: 1200 - 360 = 840 euro
#  3. Il cliente deve ancora pagare 840 euro."

# ZERO-SHOT CoT - basta aggiungere "Let's think step by step"
zero_shot_cot = """
Questo cliente deve pagare una fattura di 1200 euro.
Ha già pagato il 30%. Quanto deve ancora pagare?

Pensiamo passo per passo:
"""
# Il modello genera autonomamente il ragionamento step-by-step

# Implementazione in Python
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

cot_template = ChatPromptTemplate.from_template("""
{context}

Domanda: {question}

Ragiona passo per passo prima di rispondere. Struttura la risposta come:
RAGIONAMENTO:
[il tuo ragionamento dettagliato]

RISPOSTA FINALE:
[risposta concisa]
""")

chain = cot_template | llm
response = chain.invoke({
    "context": "Il prezzo base è 1000 euro, con IVA 22% e sconto fedele del 10%.",
    "question": "Qual è il prezzo finale da pagare?"
})
      

from langchain_core.prompts import FewShotChatMessagePromptTemplate
from langchain_core.example_selectors import SemanticSimilarityExampleSelector
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

# Libreria di esempi (task di classificazione sentiment)
examples = [
    {
        "input": "Il prodotto è arrivato rotto e il supporto non risponde.",
        "output": "NEGATIVO - Problema prodotto e supporto clienti"
    },
    {
        "input": "Consegna rapidissima e prodotto esattamente come descritto!",
        "output": "POSITIVO - Soddisfazione consegna e prodotto"
    },
    {
        "input": "Il prezzo è nella media, niente di speciale.",
        "output": "NEUTRO - Valutazione prezzo"
    },
    {
        "input": "qualità eccellente, lo riacquistero sicuramente.",
        "output": "POSITIVO - Alta qualità, fidelizzazione"
    },
    {
        "input": "Spedizione lenta, prodotto ok ma poteva andare meglio.",
        "output": "MISTO - Problema spedizione, prodotto accettabile"
    },
    # ... altri esempi
]

# Selettore semantico: scegli i 3 esempi più simili alla query
example_selector = SemanticSimilarityExampleSelector.from_examples(
    examples,
    OpenAIEmbeddings(),
    FAISS,
    k=3
)

# Template per gli esempi
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}")
])

# Few-shot prompt con selezione dinamica
few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_selector=example_selector,
    example_prompt=example_prompt,
)

# Prompt finale
final_prompt = ChatPromptTemplate.from_messages([
    ("system", """Sei un analista di sentiment per recensioni e-commerce.
Classifica il sentiment come: POSITIVO, NEGATIVO, NEUTRO, o MISTO.
Includi sempre la categoria principale del problema/punto di forza."""),
    few_shot_prompt,
    ("human", "{input}")
])

chain = final_prompt | llm
result = chain.invoke({"input": "Prodotto ottimo ma imballaggio pessimo."})
# Output: "MISTO - qualità prodotto vs problema imballaggio"
      

from pydantic import BaseModel, Field
from typing import List, Optional, Literal
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate


# Definisci il schema dell'output
class ProductAnalysis(BaseModel):
    """Analisi strutturata di un prodotto"""
    sentiment: Literal["positivo", "negativo", "neutro", "misto"] = Field(
        description="Il sentiment generale della recensione"
    )
    score: int = Field(
        description="Score da 1 a 10", ge=1, le=10
    )
    punti_forza: List[str] = Field(
        description="Lista dei punti di forza menzionati",
        default_factory=list
    )
    punti_deboli: List[str] = Field(
        description="Lista dei punti deboli menzionati",
        default_factory=list
    )
    categoria: str = Field(
        description="Categoria principale (es. 'qualità', 'spedizione', 'supporto')"
    )
    risposta_suggerita: Optional[str] = Field(
        description="Risposta suggerita per il team supporto (se sentiment negativo)",
        default=None
    )


class ReviewBatchResult(BaseModel):
    """Risultato dell'analisi di un batch di recensioni"""
    total_reviews: int
    sentiment_distribution: dict
    top_issues: List[str]
    recommendations: List[str]


# LLM con structured output
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(ProductAnalysis)

prompt = ChatPromptTemplate.from_template("""
Analizza questa recensione di prodotto e classifica il sentiment.

Recensione: {review}

Estrai tutte le informazioni richieste in modo preciso.""")

chain = prompt | structured_llm

# L'output è già un oggetto Pydantic validato
result: ProductAnalysis = chain.invoke({
    "review": "Prodotto di ottima qualità, spedizione lenta. Supporto ha risposto velocemente."
})

print(f"Sentiment: {result.sentiment}")
print(f"Score: {result.score}/10")
print(f"Punti forza: {result.punti_forza}")
print(f"Punti deboli: {result.punti_deboli}")

# Output garantito:
# Sentiment: misto
# Score: 6/10
# Punti forza: ['qualità prodotto', 'supporto reattivo']
# Punti deboli: ['spedizione lenta']
      

from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from datetime import datetime
import hashlib
import json
import yaml
from pathlib import Path


@dataclass
class PromptVersion:
    """Una versione specifica di un prompt"""
    version: str
    template: str
    variables: List[str]
    description: str
    created_at: datetime = field(default_factory=datetime.now)
    created_by: str = ""
    tags: List[str] = field(default_factory=list)
    performance_metrics: Dict[str, float] = field(default_factory=dict)
    is_active: bool = True

    @property
    def template_hash(self) -> str:
        """Hash del template per deduplication"""
        return hashlib.md5(self.template.encode()).hexdigest()[:8]

    def render(self, **kwargs) -> str:
        """Renderizza il template con le variabili fornite"""
        try:
            return self.template.format(**kwargs)
        except KeyError as e:
            raise ValueError(f"Variabile mancante nel template: {e}")

    def validate_variables(self, provided: Dict) -> List[str]:
        """Verifica che tutte le variabili richieste siano fornite"""
        missing = [v for v in self.variables if v not in provided]
        return missing


class PromptRegistry:
    """
    Registry centralizzato per la gestione dei prompt in produzione.
    Supporta versioning, A/B testing e rollback.
    """

    def __init__(self, storage_path: str = "./prompts"):
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(exist_ok=True)
        self.prompts: Dict[str, List[PromptVersion]] = {}
        self._load_from_disk()

    def register(
        self,
        name: str,
        template: str,
        variables: List[str],
        description: str = "",
        version: Optional[str] = None,
        tags: List[str] = None
    ) -> PromptVersion:
        """Registra una nuova versione di un prompt"""
        if name not in self.prompts:
            self.prompts[name] = []

        # Auto-versioning se non specificato
        if version is None:
            existing = len(self.prompts[name])
            version = f"v{existing + 1:03d}"

        prompt_version = PromptVersion(
            version=version,
            template=template,
            variables=variables,
            description=description,
            tags=tags or []
        )

        # Disattiva versione precedente attiva
        for existing_v in self.prompts[name]:
            existing_v.is_active = False

        self.prompts[name].append(prompt_version)
        self._save_to_disk(name, prompt_version)

        return prompt_version

    def get(self, name: str, version: Optional[str] = None) -> PromptVersion:
        """Ottieni una versione del prompt"""
        if name not in self.prompts:
            raise KeyError(f"Prompt '{name}' non trovato nel registry")

        if version is None:
            # Versione attiva più recente
            active = [p for p in self.prompts[name] if p.is_active]
            if not active:
                raise ValueError(f"Nessuna versione attiva per '{name}'")
            return active[-1]

        for p in self.prompts[name]:
            if p.version == version:
                return p

        raise KeyError(f"Versione '{version}' non trovata per '{name}'")

    def rollback(self, name: str, version: str) -> PromptVersion:
        """Rollback a una versione precedente"""
        target = self.get(name, version)

        # Disattiva tutto
        for p in self.prompts[name]:
            p.is_active = False

        # Attiva la versione target
        target.is_active = True

        return target

    def update_metrics(self, name: str, version: str, metrics: Dict[str, float]):
        """Aggiorna le metriche di performance di una versione"""
        prompt = self.get(name, version)
        prompt.performance_metrics.update(metrics)
        self._save_to_disk(name, prompt)

    def get_history(self, name: str) -> List[Dict]:
        """Ottieni la storia delle versioni"""
        if name not in self.prompts:
            return []
        return [
            {
                "version": p.version,
                "created_at": p.created_at.isoformat(),
                "is_active": p.is_active,
                "metrics": p.performance_metrics,
                "hash": p.template_hash
            }
            for p in self.prompts[name]
        ]

    def _save_to_disk(self, name: str, prompt: PromptVersion):
        """Persisti il prompt su disco"""
        file_path = self.storage_path / f"{name}_{prompt.version}.yaml"
        data = {
            "version": prompt.version,
            "template": prompt.template,
            "variables": prompt.variables,
            "description": prompt.description,
            "tags": prompt.tags,
            "is_active": prompt.is_active,
            "performance_metrics": prompt.performance_metrics
        }
        with open(file_path, 'w') as f:
            yaml.dump(data, f, allow_unicode=True)

    def _load_from_disk(self):
        """Carica i prompt da disco all'avvio"""
        for file_path in self.storage_path.glob("*.yaml"):
            try:
                with open(file_path) as f:
                    data = yaml.safe_load(f)

                name = "_".join(file_path.stem.split("_")[:-1])
                if name not in self.prompts:
                    self.prompts[name] = []

                self.prompts[name].append(PromptVersion(**data))
            except Exception:
                pass


# Utilizzo
registry = PromptRegistry()

# Registra prompt RAG v1
registry.register(
    name="rag_answer",
    template="""Sei un assistente tecnico. Rispondi basandoti SOLO sul contesto.

Contesto: {context}
Domanda: {question}
Risposta:""",
    variables=["context", "question"],
    description="Prompt RAG base v1"
)

# Registra prompt RAG v2 (migliorato)
registry.register(
    name="rag_answer",
    template="""Sei un assistente tecnico preciso. Rispondi basandoti ESCLUSIVAMENTE
sul contesto fornito. Se il contesto non contiene la risposta, dillo esplicitamente.

Contesto:
{context}

Domanda: {question}

Fornisci una risposta concisa e accurata:""",
    variables=["context", "question"],
    description="Prompt RAG v2 - più chiaro sul fallback"
)

# Usa la versione attiva
prompt = registry.get("rag_answer")
rendered = prompt.render(context="...", question="...")
      

import random
from scipy import stats
from typing import Callable, Tuple
import numpy as np


class PromptABTest:
    """Framework per A/B testing di prompt con significativita statistica"""

    def __init__(
        self,
        prompt_a: PromptVersion,
        prompt_b: PromptVersion,
        traffic_split: float = 0.5,  # 50% traffico a B
        min_samples: int = 100        # Campioni minimi per significativita
    ):
        self.prompt_a = prompt_a
        self.prompt_b = prompt_b
        self.traffic_split = traffic_split
        self.min_samples = min_samples

        self.results_a: List[float] = []  # Scores per prompt A
        self.results_b: List[float] = []  # Scores per prompt B

    def assign_variant(self, user_id: str = None) -> Tuple[str, PromptVersion]:
        """
        Assegna una variante all'utente.
        Deterministica se user_id e fornito (stesso utente vede sempre la stessa variante).
        """
        if user_id:
            # Hashing deterministico per consistenza per utente
            h = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
            use_b = (h % 100) < (self.traffic_split * 100)
        else:
            use_b = random.random() < self.traffic_split

        if use_b:
            return "B", self.prompt_b
        return "A", self.prompt_a

    def record_result(self, variant: str, score: float):
        """Registra il risultato per una variante"""
        if variant == "A":
            self.results_a.append(score)
        else:
            self.results_b.append(score)

    def is_statistically_significant(self, alpha: float = 0.05) -> bool:
        """Verifica significativita statistica con t-test"""
        if len(self.results_a) < self.min_samples or len(self.results_b) < self.min_samples:
            return False

        _, p_value = stats.ttest_ind(self.results_a, self.results_b)
        return p_value < alpha

    def get_winner(self) -> Optional[str]:
        """Determina il vincitore se statisticamente significativo"""
        if not self.is_statistically_significant():
            return None  # Non abbastanza dati

        mean_a = np.mean(self.results_a)
        mean_b = np.mean(self.results_b)

        return "B" if mean_b > mean_a else "A"

    def report(self) -> dict:
        """Report completo del test"""
        mean_a = np.mean(self.results_a) if self.results_a else 0
        mean_b = np.mean(self.results_b) if self.results_b else 0

        p_value = None
        if len(self.results_a) > 1 and len(self.results_b) > 1:
            _, p_value = stats.ttest_ind(self.results_a, self.results_b)

        return {
            "samples_a": len(self.results_a),
            "samples_b": len(self.results_b),
            "mean_score_a": mean_a,
            "mean_score_b": mean_b,
            "improvement_pct": ((mean_b - mean_a) / mean_a * 100) if mean_a > 0 else 0,
            "p_value": p_value,
            "is_significant": self.is_statistically_significant(),
            "winner": self.get_winner(),
            "recommendation": "Deploy B" if self.get_winner() == "B" else
                              "Keep A" if self.get_winner() == "A" else
                              "Collect more data"
        }
      

from langchain_openai import ChatOpenAI
from pydantic import BaseModel


class JudgeScore(BaseModel):
    """Score strutturato del giudice LLM"""
    accuracy: int        # 1-5: precisione fattuale
    relevance: int       # 1-5: rilevanza alla domanda
    clarity: int         # 1-5: chiarezza della risposta
    completeness: int    # 1-5: completezza della risposta
    overall: int         # 1-5: valutazione complessiva
    reasoning: str       # Spiegazione del punteggio
    issues: list         # Lista dei problemi riscontrati


class PromptTester:
    """Framework di testing automatico per prompt LLM"""

    def __init__(
        self,
        model_under_test: ChatOpenAI,
        judge_model: ChatOpenAI = None
    ):
        self.model = model_under_test
        self.judge = judge_model or ChatOpenAI(model="gpt-4o-mini", temperature=0)
        self.structured_judge = self.judge.with_structured_output(JudgeScore)

    def evaluate_single(
        self,
        prompt: str,
        input_vars: Dict,
        expected_output: str = None
    ) -> JudgeScore:
        """Valuta un singolo output del prompt"""
        # Genera output con il modello testato
        rendered_prompt = prompt.format(**input_vars)
        actual_output = self.model.invoke(rendered_prompt).content

        # Valuta con il giudice LLM
        judge_prompt = f"""Valuta questa risposta AI su scale 1-5 per ogni dimensione.

Domanda/Input: {input_vars.get('question', rendered_prompt[:200])}

Risposta generata: {actual_output}

{f"Risposta attesa: {expected_output}" if expected_output else ""}

Valuta obiettivamente e fornisci esempi specifici per ogni punto."""

        return self.structured_judge.invoke(judge_prompt)

    def run_test_suite(
        self,
        prompt: str,
        test_cases: List[Dict]
    ) -> dict:
        """
        Esegui una suite di test su un prompt.
        test_cases: lista di {"input": {vars}, "expected": "output atteso"}
        """
        scores = []
        failed_cases = []

        for i, test_case in enumerate(test_cases):
            try:
                score = self.evaluate_single(
                    prompt=prompt,
                    input_vars=test_case["input"],
                    expected_output=test_case.get("expected")
                )
                scores.append(score)

                # Flag test case con score basso
                if score.overall < 3:
                    failed_cases.append({
                        "index": i,
                        "input": test_case["input"],
                        "score": score.overall,
                        "issues": score.issues,
                        "reasoning": score.reasoning
                    })

            except Exception as e:
                failed_cases.append({"index": i, "error": str(e)})

        if not scores:
            return {"error": "Nessun test completato"}

        return {
            "total_tests": len(test_cases),
            "completed": len(scores),
            "avg_overall": np.mean([s.overall for s in scores]),
            "avg_accuracy": np.mean([s.accuracy for s in scores]),
            "avg_relevance": np.mean([s.relevance for s in scores]),
            "avg_clarity": np.mean([s.clarity for s in scores]),
            "pass_rate": len([s for s in scores if s.overall >= 3]) / len(scores),
            "failed_cases": failed_cases,
            "recommendation": "DEPLOY" if np.mean([s.overall for s in scores]) >= 4.0
                              else "REVIEW" if np.mean([s.overall for s in scores]) >= 3.0
                              else "REJECT"
        }


# Test suite di esempio per un prompt RAG
test_suite = [
    {
        "input": {
            "context": "LangChain è un framework Python per costruire applicazioni LLM.",
            "question": "Cos'è LangChain?"
        },
        "expected": "LangChain è un framework per costruire applicazioni basate su LLM"
    },
    {
        "input": {
            "context": "Il prezzo di abbonamento base è 29 euro al mese.",
            "question": "Quanto costa l'abbonamento premium?"
        },
        "expected": "Il contesto non specifica il prezzo dell'abbonamento premium"
    },
]