from pydantic import BaseModel, Field, model_validator from datetime import date, datetime from typing import Optional, List, Dict, Any import json import openai CONTRACT_TYPES = [ "Appalto", "Concessione", "Procedura negoziata", "Manifestazione d'interesse", # or "Manifestazione di interesse", usually found along with "Procedura aperta" "Procedura aperta", "Indagine di mercato" ] class ContractEntity(BaseModel): municipality: Optional[str] = Field(None, description="Nome completo del Comune citato nel bando (e.g. 'Comune di Castelletto Monferrato').") provincia: Optional[str] = Field(None, description="Provincia del Comune di riferimento (sigle, e.g. 'TO', 'MI').") regione: Optional[str] = Field(None, description="Regione italiana cui appartiene il Comune di riferimento (e.g. 'Piemonte', 'Lombardia').") cig_code: Optional[str] = Field(None, description="Codice Identificativo Gara espresso in 10 caratteri alfanumerici (e.g. 'B2AF744338').") proposal_deadline: Optional[str] = Field(None, description="Data di scadenza per la presentazione dell'offerta.") contract_deadline: Optional[str] = Field(None, description="Data di scadenza del contratto.") contract_duration: Optional[str] = Field(None, description="Durata in anni del contratto (e.g. se il bando menziona la validità per gli anni 2024/2025, 2025/2026 e 2026/2027, la durata è '3 anni')") extension_period: Optional[str] = Field(None, description="Durata in anni, o mesi se meno di un anno, della proroga. Non confondere con rinnovo!") renewal_period: Optional[str] = Field(None, description="Durata in anni, o mesi se meno di un anno, del rinnovo. Non confondere con proroga!") active: Optional[bool] = Field(False, description="Se il bando è aperto o no (oggi <= proposal_deadline).") meals_per_year: Optional[int] = Field(None, description="Numero di pasti all'anno.") meals_per_day: Optional[int] = Field(None, description=f"Numero di pasti al giorno.") contract_type: Optional[str] = Field(None, description=f"Tipologia di affidamento del contratto. Tipologie valide: {CONTRACT_TYPES}.") current_src: Optional[str] = Field(None, description="Nome completo dell'attuale società che si occupa del servizio (e.g. 'Progetti e Soluzioni SpA').") required_sw: Optional[bool] = Field(None, description="Se è richiesto un sistema informatizzato (sì/no).") addition_points: Optional[bool] = Field(None, description="Se sono attribuiti dei punti per migliorie proposte (sì/no).") current_sw: Optional[str] = Field(None, description="Il nome del software attualmente in uso, se indicato.") summary: Optional[str] = Field(None, description="Riassunto contenente le informazioni chiave del contratto.") @model_validator(mode="after") def compute_active(cls, model): """ Dopo che tutti i campi sono validati, ricalcola `active`: True se oggi ≤ proposal_deadline, altrimenti False. """ model.active = date.today() <= datetime.strptime(model.proposal_deadline, "%d-%m-%Y").date() return model @classmethod def fields_list(cls) -> List[str]: return list(cls.model_fields.keys()) # == Preprocessing Instructions == SYSTEM_PREPROCESS = f""" Sei un assistente italiano specializzato nell'analisi di bandi di gara. Di seguito il modello dati in formato JSON Schema, dove per ogni campo è indicata la descrizione ('description'): {ContractEntity.model_json_schema()} Per ogni query fornita dall'utente: - Identifica l'intento, ossia la categoria cui la domanda si riferisce (campo "intent"), scegliendo **esattamente** uno tra i campi dello schema JSON sopra; - Estrai il nome completo del comune citato, se presente (campo "municipality"); - Riscrivi la domanda in italiano usando termini attinenti all'ambito dei bandi di gare (campo "rewritten_query"), senza ripetere il comune; - Basandoti sulla descrizione del campo identificato come intento, riscrivi la domanda in italiano (campo "rewritten_query"), senza ripetere il comune, usando il linguaggio tecnico necessario **per trovare** tramite il retriever l'informazione descritta. Restituisci **solo** un JSON con i tre campi: "municipality", "rewritten_query" e "intent". """ FEW_SHOTS_PREPROCESS = [ ("Quanto dura il contratto per Verona?", {"intent":"contract_duration", "municipality":"Comune di Verona", "rewritten_query":"Durata del contratto?"}), ("Devo sapere se richiedono una piattaforma ad-hoc in particolare.", {"intent":"required_sw", "municipality":"", "rewritten_query":"È richiesto un software specifico?"}), ("Quando finisce il contratto con Roma?", {"intent":"contract_deadline", "municipality":"Comune di Roma", "rewritten_query":"Quando scade il contratto?"}), ("Qual è la scadenza per presentare l'offerta per il bando di Bologna?", {"intent": "proposal_deadline", "municipality": "Comune di Bologna", "rewritten_query": "Qual è la data di scadenza per la presentazione dell'offerta?"}), ("Vorrei sapere quanti pasti servono ogni giorno.", {"intent": "meals_per_day", "municipality": "", "rewritten_query": "Numero di pasti al giorno?"}), ("A Firenze, chi è l'attuale gestore del servizio mensa scolastica?", {"intent": "current_src", "municipality": "Comune di Firenze", "rewritten_query": "Chi è l'attuale gestore del servizio?"}), ("Mi serve il CIG della gara a Bari.", {"intent": "cig_code", "municipality": "Comune di Bari", "rewritten_query": "Qual è il CIG?"}), ("Il contratto a Palermo prevede una proroga?", {"intent": "extension_period", "municipality": "Comune di Palermo", "rewritten_query": "È prevista una proroga?"}), ("Che tipo di affidamento è stato utilizzato nel bando di Trento?", {"intent": "contract_type", "municipality": "Comune di Trento", "rewritten_query": "Qual è il tipo di affidamento?"}), ("Cosa devo sapere di importante sul bando di Como?", {"intent":"summary", "municipality":"Comune di Como", "rewritten_query":"Riassunto delle informazioni chiave?"}), ] def preprocess_query(raw: str, feedback: Optional[str] = None) -> Dict[str, Any]: """ Calls the LLM once to extract: - intent (one of ContractEntity fields) - municipality (str) - rewritten_query (str) Optionally refines rewritten_query if feedback is provided. """ msgs = [{"role": "system", "content": SYSTEM_PREPROCESS}] # Using few shots for example_q, example_out in FEW_SHOTS_PREPROCESS: msgs.append({"role": "user", "content": example_q}) msgs.append({ "role": "assistant", "content": json.dumps(example_out, ensure_ascii=False) }) # Raw user query msgs.append({"role": "user", "content": raw}) # Query refining if we have feedback if feedback: msgs.append({ "role": "user", "content": f"Feedback: {feedback}\n" "Modifica SOLO il campo \"rewritten_query\" tenendo in considerazione questo feedback." }) resp = openai.chat.completions.create( model="gpt-4o-mini", messages=msgs, temperature=0.0 ) text = resp.choices[0].message.content.strip() try: obj = json.loads(text) except json.JSONDecodeError as e: raise ValueError(f"Non ho potuto parsare JSON dalla risposta:\n{text}") from e for key in ("intent", "municipality", "rewritten_query"): if key not in obj: raise KeyError(f"Il campo '{key}' manca nella risposta: {obj}") return obj