Spaces:
Sleeping
Sleeping
| import os | |
| import pytz | |
| import gradio as gr | |
| from datetime import datetime | |
| from pinecone import Pinecone | |
| from llama_index.llms.groq import Groq | |
| from llama_index.core import VectorStoreIndex, Settings | |
| from llama_index.vector_stores.pinecone import PineconeVectorStore | |
| from llama_index.embeddings.huggingface import HuggingFaceEmbedding | |
| # --- CONFIGURAZIONE --- | |
| GROQ_API_KEY = os.getenv("GROQ_API_KEY") | |
| PINECONE_API_KEY = os.getenv("PINECONE_API_KEY") | |
| if not GROQ_API_KEY or not PINECONE_API_KEY: | |
| raise ValueError("⚠️ ERRORE CRITICO: API Keys mancanti nei Secrets!") | |
| # --- SETUP MODELLI (globali: costosi da ricaricare, non contengono stato conversazione) --- | |
| embed_model = HuggingFaceEmbedding(model_name="intfloat/multilingual-e5-large") | |
| Settings.embed_model = embed_model | |
| # max_retries=0: disabilita i retry automatici su errore 429 (limite giornaliero Groq). | |
| llm = Groq(model="llama-3.3-70b-versatile", api_key=GROQ_API_KEY, max_retries=0) | |
| Settings.llm = llm | |
| # --- INDICE PINECONE (globale: read-only, nessuno stato conversazione) --- | |
| try: | |
| pc = Pinecone(api_key=PINECONE_API_KEY) | |
| _pinecone_index = VectorStoreIndex.from_vector_store( | |
| vector_store=PineconeVectorStore( | |
| pinecone_index=pc.Index("prometheus-main"), | |
| namespace="accessibility-poc" | |
| ) | |
| ) | |
| print("✅ Connessione Pinecone stabilita.") | |
| except Exception as e: | |
| _pinecone_index = None | |
| print(f"❌ Errore connessione Pinecone: {e}") | |
| SYSTEM_PROMPT = ( | |
| "Sei Prometheus, un assistente virtuale per l'Accessibilità e il Turismo Inclusivo. " | |
| "Il tuo compito è fornire informazioni utili come se fossi un operatore esperto allo sportello informazioni." | |
| "\n\n" | |
| "🚫 DIVIETI ASSOLUTI DI LINGUAGGIO (Blacklist): " | |
| "- MAI usare parole come: 'scheda tecnica', 'json', 'database', 'record', 'file', 'documento', 'valore', 'campo', 'null'. " | |
| "- EVITA frasi come: 'In base ai dati', 'Secondo la scheda', 'Il sistema dice che'. " | |
| "\n\n" | |
| "🧠 DIZIONARIO SEMANTICO (Interpretazione): " | |
| "- 'ingresso_senza_gradini: true' = 'L ingresso è in piano / senza barriere'. " | |
| "- 'ingresso_senza_gradini: false' = 'Sono presenti gradini all ingresso'. " | |
| "- 'bagni_accessibili: true' = 'È disponibile un bagno attrezzato'. " | |
| "\n\n" | |
| "✅ REGOLE DI RISPOSTA (Stile Conversazionale): " | |
| "1. Conferme (True): Rispondi direttamente. Es: Si, confermo che la struttura è dotata di ascensore. " | |
| "2. Negazioni (False): Sii gentile. Es: Purtroppo al momento non risulta presente un servizio di audioguida. " | |
| "3. Dati Mancanti (Null/Assente): Non citare la scheda. Di semplicemente: Non ho informazioni aggiornate su questo specifico servizio. oppure Al momento non mi risulta questa informazione, ti consiglio di verificare contattando la struttura. " | |
| "4. Note (Stringhe): Integra le note nel discorso. Es: Ti segnalo pero che l ascensore è attualmente in manutenzione. " | |
| "5. FEDELTA ASSOLUTA (Anti-Allucinazione): Attieniti ESCLUSIVAMENTE ai fatti forniti nel contesto. Non aggiungere MAI commenti personali, supposizioni o frasi di circostanza." | |
| "\n\n" | |
| "OBIETTIVO: Sii neutrale, preciso e disponibile. L utente deve sentirsi aiutato da una persona competente." | |
| ) | |
| # --- DIZIONARIO GLOBALE: engine per session_hash --- | |
| # Gradio 3.x non ha gr.State accessibile facilmente nei Blocks con fn complesse. | |
| # Usiamo un dizionario server-side keyed su request.session_hash (isolamento per sessione). | |
| _session_engines = {} | |
| def get_or_create_engine(session_hash): | |
| """Restituisce l'engine della sessione, creandolo se non esiste.""" | |
| if session_hash not in _session_engines: | |
| if _pinecone_index is None: | |
| return None | |
| try: | |
| _session_engines[session_hash] = _pinecone_index.as_chat_engine( | |
| chat_mode="condense_plus_context", | |
| system_prompt=SYSTEM_PROMPT, | |
| verbose=True | |
| ) | |
| print(f"✅ Nuovo engine creato per sessione: {session_hash}") | |
| except Exception as e: | |
| print(f"❌ Errore creazione engine: {e}") | |
| return None | |
| return _session_engines[session_hash] | |
| # --- RISPOSTA --- | |
| def respond(message, history, request: gr.Request): | |
| session_hash = request.session_hash if request else "default" | |
| engine = get_or_create_engine(session_hash) | |
| if engine is None: | |
| return "⚠️ Servizio momentaneamente non disponibile. Riprova tra qualche istante." | |
| try: | |
| fuso_italia = pytz.timezone("Europe/Rome") | |
| adesso = datetime.now(fuso_italia) | |
| ora_data = adesso.strftime("%A %d %B %Y, ore %H:%M") | |
| messaggio_con_data = f"[Oggi è {ora_data}]\n{message}" | |
| return str(engine.chat(messaggio_con_data)) | |
| except Exception as e: | |
| err_type = type(e).__name__ | |
| err_str = str(e) | |
| print(f"❌ ERRORE in chat [{session_hash}]: {err_type}: {err_str}") | |
| if "429" in err_str or "rate_limit" in err_str.lower() or "RateLimitError" in err_type: | |
| return ( | |
| "⏳ Il servizio ha raggiunto il limite di utilizzo giornaliero. " | |
| "Riprova domani mattina dopo le 02:00 (ora italiana). " | |
| "Ci scusiamo per il disagio." | |
| ) | |
| return "Mi scuso, al momento non riesco a recuperare questa informazione. Riprova tra qualche istante." | |
| # --- UI --- | |
| disclaimer = """ | |
| <div style=" | |
| margin-top: 24px; | |
| padding: 12px 20px; | |
| border-top: 1px solid #d1d5db; | |
| background-color: #f9fafb; | |
| border-radius: 8px; | |
| text-align: center; | |
| font-size: 13px; | |
| color: #6b7280; | |
| line-height: 1.6; | |
| "> | |
| <strong>Versione dimostrativa</strong> — | |
| Questa applicazione è a scopo esclusivamente illustrativo. | |
| Tutte le strutture, i dati e le informazioni presenti sono fittizi | |
| e non corrispondono a realtà esistenti. | |
| </div> | |
| """ | |
| demo = gr.ChatInterface( | |
| fn=respond, | |
| title="ℹ️ Sportello Accessibilità - Prometheus", | |
| description=( | |
| "Chiedimi informazioni sui servizi per persone con disabilità di musei e luoghi pubblici.<br>" | |
| "Applicazione dimostrativa con aziende di agriturismo fittizie nella città di Siena." | |
| ), | |
| examples=["Ci sono agriturismo accessibili a Siena?", "Parcheggi per disabili?"], | |
| cache_examples=False | |
| ) | |
| with demo: | |
| gr.HTML(disclaimer) | |
| if __name__ == "__main__": | |
| demo.launch(server_name="0.0.0.0", server_port=7860) |