Spaces:
Sleeping
Sleeping
| 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.") | |
| 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 | |
| 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 |