''' Sessioni simultanee Per impostazione predefinita, puoi avere fino a 10 sessioni simultanee per progetto. https://ai.google.dev/gemini-api/docs/live?hl=it https://cloud.google.com/vertex-ai/generative-ai/docs/live-api?hl=it https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/multimodal-live?hl=it#limitations Estensione della durata della sessione https://ai.google.dev/gemini-api/docs/live?hl=it#extend-session-duration Conteggio token https://ai.google.dev/gemini-api/docs/live?hl=it#token-count Rilevamento attività vocale (VAD) https://ai.google.dev/gemini-api/docs/live?hl=it#voice-activity-detection Gestione delle interruzioni https://ai.google.dev/gemini-api/docs/live?hl=it#interruptions ------------------------------------------------------------------------------------------------------------- conda activate avatar cd /home/cirillo/Desktop/GEMINI/GEMINI_RETAIL9 uvicorn main:app --reload --host 0.0.0.0 --port 9102 APRI LA PAGINA index.html ed imposta la porta identica a quella che vedi qui http://localhost:9102/static/index.html ------------------------------------------------------------------------------------------------------------- AGGIORNARE I FILE SUL SERVER il conda è gemini /root/miniconda3/envs/gemini A2HOSTING export LD_LIBRARY_PATH=/usr/lib/x86_64-linux-gnu/:$LD_LIBRARY_PATH ssh -p 7822 root@68.66.251.32 synap.si77 sudo systemctl restart uvicorn.service sudo journalctl -u uvicorn.service -f sudo systemctl daemon-reload sudo systemctl restart fastapi-gemini sudo systemctl status fastapi-gemini ''' import asyncio import datetime import base64 import json import numpy as np import scipy.signal import httpx from google import genai from google.genai.types import ( LiveConnectConfig, PrebuiltVoiceConfig, SpeechConfig, VoiceConfig, AudioTranscriptionConfig, Tool, FunctionResponse, FunctionDeclaration, StartSensitivity, EndSensitivity, CreateCachedContentConfig, ContextWindowCompressionConfig, GenerateContentConfig, SafetySetting, ThinkingConfig, SlidingWindow, Part, Schema, Content, Type) from enum import Enum from typing import Dict from contextlib import asynccontextmanager import pathlib import re from collections import deque # —————— (Opzionale: registro per Strategia M) —————— active_handlers: Dict[str, "GeminiHandler"] = {} FUNCTIONS_CATEGORIES_PATH = "CONTESTO/categorie.json" structured_cache_name: str GEMINI_model="gemini-2.5-flash-preview-04-17" GEMINI_LIVEAUDIO_model="models/gemini-2.5-flash-preview-native-audio-dialog" GOOGLE_SHEET = "https://script.google.com/macros/s/AKfycbzFsHcVfZztV3zq4guPqAs_Gz-R2jYYVv0RfB4SsOjmsyCQFZxZTyegbAOKIWrdfmdyng/exec" class SessionState(Enum): DISCONNECTED = "disconnected" LISTENING = "listening" THINKING = "thinking" SPEAKING = "speaking" def map_to_phonemes(word: str) -> list[str]: # (mantieni la tua funzione di prima senza modifiche) phoneme_rules = { "CH": ["ce", "ci", "che", "chi"], "SH": ["scia", "scio", "sci", "sce"], "kk": ["c", "ch", "g", "gh", "q"], "SS": ["s", "ss", "z"], "PP": ["p", "b", "m"], "DD": ["d", "t"], "FF": ["f", "v"], "RR": ["r"], "L": ["l", "gl"], "NN": ["n", "gn"], "E": ["e", "è", "é"], "aa": ["a", "à"], "ih": ["i", "ì"], "oh": ["o", "ò"], "ou": ["u", "ù"] } normalized = ( word.lower() .replace("à", "a").replace("è", "e").replace("é", "e") .replace("ì", "i").replace("ò", "o").replace("ù", "u") .replace("’", "'").strip() ) phonemes = [] i = 0 while i < len(normalized): matched = False for viseme, patterns in phoneme_rules.items(): for pat in sorted(patterns, key=lambda x: -len(x)): if normalized[i:i + len(pat)] == pat: phonemes.append(viseme) i += len(pat) matched = True break if matched: break if not matched: phonemes.append(normalized[i]) i += 1 return phonemes def load_function_declarations(json_path: str): """ Carica dal file JSON le specifiche di ogni funzione e ritorna una lista di FunctionDeclaration. """ with open(json_path, 'r', encoding='utf-8') as f: specs = json.load(f) declarations = [] for name, info in specs.items(): declarations.append( FunctionDeclaration( name=name, description=info['description'], parameters=Schema( type=Type.OBJECT, properties=info.get('parameters', {}) ), ) ) return declarations # def get_exhibit_description(): # pass # def get_ticketing_and_hours(): # pass class GeminiHandler: def __init__(self, client: genai.Client, voice_name: str = "Aoede", sample_rate: int = 24000): self.client = client self.voice_name = voice_name self.output_sample_rate = sample_rate # —————————————————————————————————————————— # 1) Coda per i chunk raw (48 kHz) ricevuti dal WebSocket self._raw_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=100) # 2) Coda per i chunk risampionati a 16 kHz (pronti per run()) self._chunk_queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=100) # —————————————————————————————————————————— # Lista esplicita dei task di run() e feed_audio() self._tasks: list[asyncio.Task] = [] self._closed = False # Buffer interni (testo/audio accumulati) self._total_audio_bytes: int = 0 self.accum_text = "" self.question_text = "" self._structured_json: str | None = None self.quit_event = asyncio.Event() # Stato e mute self.session_state = SessionState.DISCONNECTED self.speaker_muted = False # Strategia M: timestamp last_seen per TTL self.last_seen: datetime.datetime = datetime.datetime.utcnow() # Dizionario dei WebSocket registrati (per broadcast) self._clients: dict[str, "WebSocket"] = {} # ► Nuovo: queue dedicata all’invio dei chunk di audio al frontend # ogni elemento sarà una tupla (chunk_index, raw_bytes) self._outgoing_audio_queue: asyncio.Queue[tuple[int, bytes]] = asyncio.Queue() # Buffer per memorizzare (index, length) di ciascun chunk self.audio_chunks_buffer: list[tuple[int, int]] = [] # URL Google Sheet (stesso di prima) self.endpoint = GOOGLE_SHEET with open(FUNCTIONS_CATEGORIES_PATH, "r", encoding="utf-8") as f: self.categorie_data = json.load(f) self.last_categoria = None self.chat_history: deque[tuple[str, str]] = deque(maxlen=10) # Ultimi 10 turni async def receive(self, frame: tuple[int, bytes]): """ (abbiamo ricevuto) AUDIO FRONTEND -> BACKEND (dove siamo) -> (dove lo inviamo) GEMINI """ sample_rate, raw = frame # === LOG DI DEBUG === #print(f"[GeminiHandler.receive] ricevo chunk raw: sample_rate={sample_rate}, byte={len(raw)}") # ==================== await self._raw_queue.put(raw) async def _stream_generator(self): """ Generator asincrono che yielda i chunk a 16 kHz prelevandoli da _chunk_queue. """ while True: chunk = await self._chunk_queue.get() yield chunk async def _audio_sender(self): """ INVIO IMMEDIATO DELL'AUDIO RICEVUTO DA GEMINI AL FRONTEND Ogni volta che qualcuno mette (index, raw_audio) in _outgoing_audio_queue, estraggolo e costruisco il JSON come: { "type":"audio_chunk", "index":idx, "length":chunk_len, "audio":base64(raw_audio) } e lo invio via ws.send_text(...) a tutti i WebSocket attivi. """ while True: try: # Attendo un nuovo chunk pronto per l’invio idx, raw_audio = await self._outgoing_audio_queue.get() # Se il handler è già stato chiuso, esco subito if self._closed: return # 1) calcolo la lunghezza in byte chunk_len = len(raw_audio) # 2) codifico in base64 # raw_audio è già bytes (PCM16), quindi basta: b64chunk = base64.b64encode(raw_audio).decode("utf-8") # 3) costruisco il dizionario JSON audio_msg = { "type": "audio_chunk", "index": idx, "length": chunk_len, "audio": b64chunk } # 4) serializzo e mando a tutti i WebSocket registrati data = json.dumps(audio_msg) for ws in list(self._clients.values()): try: await ws.send_text(data) except Exception: # Se qualche WS è caduta, la ignoriamo e continuiamo pass except asyncio.CancelledError: # Se il task viene cancellato, esco pulito return except Exception: # In caso di errore generico, ignoro e continuo il loop continue async def run(self): """ Si connette a Gemini Live, accumula testo/audio e fa broadcast. Restituisce correttamente al loop e tiene aperto l'async for. default_api.get_exhibit_description() """ index = 0 try: # # 1) dichiarazioni delle funzioni, una per ogni “capitolo approfondito” # # 1) Definizione delle FunctionDeclaration con schema PARAMETERS # get_exhibit_description = FunctionDeclaration( # name="get_exhibit_description", # description="Restituisce la descrizione dettagliata della mostra", # parameters=Schema( # type=Type.OBJECT, # properties={}, # nessun argomento # ), # ) # get_ticketing_and_hours = FunctionDeclaration( # name="get_ticketing_and_hours", # description="Restituisce le politiche di biglietteria e le tariffe per la mostra.", # parameters=Schema( # type=Type.OBJECT, # properties={}, # nessun argomento # ), # ) # get_additional_services = FunctionDeclaration( # name="get_additional_services", # description="Restituisce informazioni sui servizi aggiuntivi e consigli per la visita.", # parameters=Schema( # type=Type.OBJECT, # properties={}, # nessun argomento # ), # ) # # 2) Raggruppamento delle funzioni in un Tool # tools = [ # Tool( # function_declarations=[ # get_exhibit_description, # get_ticketing_and_hours, # get_additional_services, # ] # ), # ] # 1) dichiarazioni delle funzioni, una per ogni “capitolo approfondito” # function_declarations = load_function_declarations(FUNCTIONS_CATEGORIES_PATH) # print('FUNCTION DECLARATIONS') # print(function_declarations) # # Raggruppo tutto nel tool # tools = [ # Tool( # function_declarations=function_declarations # ), # ] # PROMPT = f""" # Sei EVA, la guida virtuale dell’autosalone. Gestisci la conversazione audio con i visitatori. Il tuo compito è rispondere in modo coerente alle domande, riconoscendo il contesto della conversazione e adattando sempre il tuo comportamento alle seguenti regole: # ──────────────────────────── # GESTIONE DEGLI SCENARI # ──────────────────────────── # 1. **Domande personali rivolte all’avatar** # - Se ricevi domande personali su di te (es. età, stato civile, preferenze, inviti, ecc.), rispondi subito in modo educato e molto breve, evitando di fornire dettagli, e riporta la conversazione sull’autosalone. # - Esempi: # - “Quanti anni hai?” → “Non si cheide l'età ad una signora.” # - “Sei single?” → “Preferisco parlare di automobili.” # - “Cosa fai nel tempo libero?” → “Mi occupo solo di informazioni sull’autosalone.” # 2. **Domande specifiche su autosalone, modelli, orari, servizi** # - Puoi rispondere (solo dopo aver ricevuto i dati dal sistema) a domande che riguardano esclusivamente: # • I modelli Tesla disponibili nello showroom (Model 3, Model Y, Model S), prove su strada, allestimenti, accessori, novità di gamma, personalizzazioni e servizi per privati o aziende. # • Gli orari di apertura e chiusura dell’autosalone, compresi orari festivi e dell’assistenza clienti. # • I contatti: numeri di telefono, email, modulo di contatto, chat, prenotazioni e richieste di appuntamento o test drive. # • Le caratteristiche tecniche dei modelli Tesla (autonomia, batteria, prestazioni, display, sistemi di guida assistita, ricarica, dotazioni di sicurezza, connettività, ecc.). # • I pagamenti e finanziamenti: modalità accettate, leasing, anticipo, permuta dell’usato, promozioni, bonus rottamazione, simulazione rate. # • I servizi post-vendita: manutenzione programmata, assistenza stradale, aggiornamenti software, supporto tecnico, installazione ricarica domestica, gestione incentivi, veicolo sostitutivo, rinnovo garanzia, configurazione app Tesla, piani assicurativi personalizzati, eventi o corsi clienti. # - Se la domanda riguarda uno di questi argomenti, ma NON hai ancora ricevuto i dati dal sistema, prendi tempo con una frase tipo: # - “Controllo subito i modelli disponibili.” # - “Verifico subito gli orari.” # - “Sto controllando i metodi di pagamento.” # - “Verifico le opzioni di assicurazione.” # - NON puoi invece rispondere a domande che riguardano: # • Veicoli di altre marche (BMW, Fiat, ecc.) # • Manutenzione, assicurazione, revisioni o ricambi di auto non Tesla # • Servizi non presenti nello showroom Tesla # • Domande troppo generiche o personali non relative ai servizi Tesla # • Argomenti che non rientrano nei servizi o prodotti dello showroom # - In questi casi, rispondi gentilmente che ti occupi solo di veicoli e servizi Tesla disponibili nello showroom, e invita l’utente a chiedere informazioni sui modelli e servizi Tesla. # 3. **Domande di approfondimento o follow-up** # - Se l’utente fa una domanda di approfondimento sulla risposta precedente e hai già ricevuto informazioni aggiuntive dal sistema, prova a rispondere usando quelle informazioni, senza attendere altri dati. # - Esempio: # - Dopo aver chiesto “Che Tesla avete?” e aver ricevuto la lista, l’utente chiede “Che autonomia hanno?” → Usa subito le informazioni che già possiedi. # 4. **Arrivo di informazioni aggiuntive (dati dal secondo LLM)** # - Quando ricevi nuove informazioni dettagliate o schede informative (ad esempio su un modello specifico, sugli orari, sui pagamenti, ecc.), usa questi dati per fornire una risposta completa e precisa all’utente. # - La risposta deve essere naturale, utile e chiara, e può includere consigli o dettagli pratici in base alle informazioni ricevute. # - Esempio: # - Ricevi informazioni sui metodi di pagamento → “Accettiamo pagamenti con bonifico, carta di credito e finanziamento direttamente in sede.” # 5. **Domande fuori contesto (politica, sport, viaggi, cucina, ecc.)** # - Se ricevi domande che non riguardano l’autosalone, le auto, i modelli, i servizi o lo spazio dell’autosalone, rispondi educatamente che puoi parlare solo dell’autosalone. # - Esempi: # - “Cosa ne pensi della politica?” → “Posso aiutarti solo con informazioni sull’autosalone.” # ──────────────────────────── # ISTRUZIONI DI STILE # ──────────────────────────── # - Chiamati sempre EVA. # - Hai occhi azzurri e capelli rosa. # - Indossi una tuta da ginnastica semplice. # - Non fornire mai dettagli personali diversi da quelli elencati sopra. # - Se ricevi complimenti, rispondi gentilmente e brevemente (es. “Grazie mille!”). # - Le tue risposte devono essere sempre sintetiche, chiare e in tono accogliente. # - Quando prendi tempo (“Sto controllando…”, “Verifico subito…”), non dare MAI la risposta completa finché non ricevi le informazioni aggiuntive. # - Quando rispondi con informazioni ricevute dal sistema, sii esaustiva ma evita di dilungarti inutilmente. # - Se la domanda è fuori contesto, chiudi sempre gentilmente riportando il focus sulle informazioni relative all’autosalone. # ──────────────────────────── # ESEMPI PRATICI (da seguire come modello) # ──────────────────────────── # Utente: “Che orari fate?” # EVA: “Verifico subito gli orari.” # [Quando ricevi i dettagli sugli orari dal sistema:] # EVA: “L’autosalone è aperto dal lunedì al sabato, dalle 9 alle 19.” # Utente: “Avete modelli Tesla disponibili?” # EVA: “Controllo subito i modelli Tesla.” # [Quando ricevi l’elenco modelli:] # EVA: “Al momento abbiamo Tesla Model 3 e Tesla Model Y disponibili per la vendita.” # Utente: “Come posso pagare?” # EVA: “Sto controllando i metodi di pagamento.” # [Quando ricevi i dettagli:] # EVA: “Accettiamo pagamenti tramite bonifico, carta di credito e finanziamento.” # Utente: “Cosa fai nel tempo libero?” # EVA: “Preferisco non parlare di me. Posso aiutarti sulle auto?” # ──────────────────────────── # GESTIONE DEL CONTESTO # - Ricorda sempre lo scenario della conversazione e scegli la strategia più adatta tra quelle sopra. # - Se hai già risposto a una domanda simile e possiedi già tutte le informazioni, rispondi subito usando quelle. # - Se ricevi nuove informazioni dal sistema, fornisci la risposta dettagliata appena possibile. # """ # Costruisci la sezione di Knowledge Base da inserire nel prompt kb_sections = [] for key, data in self.categorie_data.items(): descrizione = data.get("descrizione", "").strip() # Titolo opzionale per aiutare il modello a navigare i contenuti kb_sections.append(f"## {key}\n{descrizione}\n") # Unisci tutte le descrizioni full_kb_text = "\n".join(kb_sections) # --- 2. Prompt con inclusione della KB --- PROMPT = f""" Sei EVA, la guida virtuale dell’autosalone Tesla. Rispondi alle domande degli utenti **solo** in base alle informazioni presenti nella seguente Knowledge Base, adattando sempre lo stile e le istruzioni qui sotto. ──────────────────────────── KNOWLEDGE BASE AUTOSALONE ──────────────────────────── {full_kb_text} ──────────────────────────── ISTRUZIONI E STILE DI RISPOSTA ──────────────────────────── - Usa **esclusivamente** le informazioni contenute nella Knowledge Base qui sopra per rispondere. - Se la domanda non trova risposta nella Knowledge Base, spiega gentilmente che non puoi aiutare su quell’argomento. - Se una domanda è personale (es. età, interessi), rispondi brevemente e riporta il discorso sui servizi dell’autosalone. - Le risposte devono essere sempre chiare, sintetiche, al massimo una frase non troppo lunga. - Evita dettagli inutili o argomenti non pertinenti. - Chiamati EVA se ti chiedono il nome, ma non aggiungere dettagli personali diversi da quelli consentiti. - Rispondi in modo accogliente e gentile. - Quando non hai informazioni, suggerisci di parlare con un consulente dello showroom. ──────────────────────────── ESEMPI ──────────────────────────── Utente: “Che orari fate?” EVA: “Siamo aperti dal lunedì al venerdì dalle 9 alle 19, il sabato dalle 9 alle 13.” Utente: “Avete la Model Y?” EVA: “Sì, la Model Y è disponibile in vari allestimenti presso il nostro showroom.” Utente: “Posso prenotare un test drive?” EVA: “Certo! Puoi prenotare un test drive tramite telefono, email o sito.” Utente: “Vendete BMW?” EVA: “Trattiamo solo veicoli Tesla.” Utente: “Cosa fai nel tempo libero?” EVA: “Preferisco non parlare di me. Vuoi sapere qualcosa sui nostri servizi Tesla?” ──────────────────────────── **Rispondi sempre attenendoti alle informazioni della Knowledge Base.** """ print(PROMPT) # print("StartSensitivity options:") # for s in StartSensitivity: # print(f"{s.name} = {s}") # print("\nEndSensitivity options:") # for e in EndSensitivity: # print(f"{e.name} = {e}") config = LiveConnectConfig( response_modalities=["AUDIO"], speech_config=SpeechConfig( language_code="it-IT", voice_config=VoiceConfig( prebuilt_voice_config=PrebuiltVoiceConfig( voice_name=self.voice_name ) ), ), output_audio_transcription=AudioTranscriptionConfig(), input_audio_transcription=AudioTranscriptionConfig(), #tools=tools, system_instruction=Content( parts=[Part.from_text(text=PROMPT)], role="user" ), # cached_content=self._cached_content_name, context_window_compression=ContextWindowCompressionConfig( trigger_tokens=4000, sliding_window=SlidingWindow(target_tokens=2000), ), realtime_input_config = { "automatic_activity_detection": { "disabled": False, # default "start_of_speech_sensitivity": StartSensitivity.START_SENSITIVITY_LOW, # ALTERNATIVA UNICA: .START_SENSITIVITY_HIGH "end_of_speech_sensitivity": EndSensitivity.END_SENSITIVITY_LOW, # ALTERNATIVA UNICA: .END_SENSITIVITY_HIGH "prefix_padding_ms": 20, "silence_duration_ms": 100, } } ) async with self.client.aio.live.connect( # models/gemini-2.5-flash-preview-native-audio-dialog models/gemini-2.0-flash-live-001 model=GEMINI_LIVEAUDIO_model, config=config,) as session: send = False # Usiamo UN SOLO async for per tutta la durata della sessione async for response in session.start_stream(stream=self._stream_generator(), mime_type="audio/pcm"): # All’interno di questo blocco: # - definisco i flag 'no_audio' e 'no_transcript' # - provo a leggere input_transcription, response.data e output_transcription # - se no_audio && no_transcript && send → faccio il flush no_audio = False no_transcript = False ## QUESTO E' IL CODICE CHE RICEVIAMO quando il sistema identica una CHIAMATA A FUNZIONE # setup_complete=None server_content=None tool_call=LiveServerToolCall(function_calls=[FunctionCall(id='function-call-1205134339995659609', args={}, name='get_exhibit_description')]) tool_call_cancellation=None usage_metadata=None go_away=None session_resumption_update=None # try: # print('+++ FUNCTION CALL +++ ', response.tool_call) # print('--------------------------------------------------------------------------------------------------------') # except: # print('--- NESSUNA FUNCTION CALL ---') # ————— 0) Conteggio dei tokens ————— # https://ai.google.dev/api/live?hl=it#usagemetadata if response.server_content and response.server_content.turn_complete: try: prompt_tokens = response.usage_metadata.prompt_token_count or 0 total_tokens = response.usage_metadata.total_token_count or 0 response_audio_tokens = 0 response_text_tokens = 0 if response.usage_metadata.response_tokens_details: for m in response.usage_metadata.response_tokens_details: if m.modality.name == "AUDIO": response_audio_tokens = m.token_count elif m.modality.name == "TEXT": response_text_tokens = m.token_count asyncio.create_task( self.send_to_google_sheet( endpoint_url=self.endpoint, question=question, response=response_text, prompt_tokens=prompt_tokens, response_audio_tokens=response_audio_tokens, response_text_tokens=response_text_tokens, total_tokens=total_tokens, sheet_name="Foglio1" ) ) except Exception as e: print("❌ 22 - Errore durante l'invio al Google Sheet:", e) # ————— 1) Trascrizione dell'audio che abbiamo mandato a Gemini ————— try: text_in = response.server_content.input_transcription.text print('TEXT INPUT: ', response.server_content) self.question_text += text_in ai2 = True except Exception: pass # ————— 2) Audio di risposta creato da Gemini ————— try: # arr = np.frombuffer(response.data, dtype=np.int16) # self.accum_audio += arr.tobytes() raw_audio = response.data # bytes PCM16 if response.server_content.output_transcription is not None: transcription_obj = response.server_content.output_transcription print('+ AUDIO OUTPUT: ', transcription_obj) chunk_len = len(raw_audio) #print("INDEX:", index, "LEN AUDIO sample:", chunk_len // 2) # 1) incremento solo il contatore totale dei byte ricevuti self._total_audio_bytes += chunk_len # Memorizzo l’(index, length) nel buffer self.audio_chunks_buffer.append((index, chunk_len)) # Mando in coda al task _audio_sender await self._outgoing_audio_queue.put((index, raw_audio)) index += 1 no_audio = False except Exception: no_audio = True # 3) ACCUMULO TRASCRIZIONE IN USCITA (risposta di Gemini), # ma *senza* inviare ancora alcun audio: i chunk sono già stati spediti via _audio_sender. try: transcript = response.server_content.output_transcription.text print('- TEXT OUTPUT: ', response.server_content.output_transcription.text) if ai2: print("Lancio background_structured_and_speak") self.chat_history.append(("user", self.question_text.strip())) asyncio.create_task(self._call_structured_and_speak(self.chat_history, session)) ai2 = False if transcript: self.accum_text += transcript no_transcript = False send = True else: no_transcript = True except Exception: no_transcript = True # 4) condizione di “flush”: nessun audio e nessuna trascrizione in arrivo, # ma avevamo appena messo send=True → significa che la frase è completa: if no_audio and no_transcript and send: print('+++++++++ ALL DONE! *******************') print("DOMANDA: ", self.question_text) # 🔥 flush: invio payload ai client print("RISPOSTA: ", self.accum_text) # Calcolo durata audio (in secondi) usando il contatore di byte audio_duration = (self._total_audio_bytes / 2) / self.output_sample_rate # Reset contatore subito, perché non serve più accumulare i byte self._total_audio_bytes = 0 payload = self.assemble_payload( self.accum_text, audio_duration ) print('PAYLOAD') #print(payload) await self._safe_broadcast(payload) # invio asincrono su Google Sheet question = self.question_text.strip() response_text = self.accum_text.strip() self.chat_history.append(("agent", self.accum_text.strip())) # reset dei buffer per la prossima frase # azzeriamo i campi testuali self.accum_text = "" self.question_text = "" self.audio_chunks_buffer = [] send = False # **Reset di index**: index = 0 # Se la sessione si chiude (async for esce), veniamo qui. return except asyncio.CancelledError: # se il Task viene cancellato, lascio che il context async with chiuda lo stream raise except Exception as e: print(f"[GeminiHandler.run] Errore inatteso: {e}") raise ###################################################################################### ####### VERY IMPORTANT ############################################################### ######## https://googleapis.github.io/python-genai/#caches ###################################################################################### async def _call_structured_and_speak(self, history: deque[tuple[str, str]], session): # 1) Chiama il modello strutturato in un executor loop = asyncio.get_running_loop() raw = await loop.run_in_executor( None, lambda: self._sync_generate_structured(history) ) # 2) Parsing del JSON (stringa → dict) try: raw = json.loads(raw) except Exception as e: print("❌ Errore nel parsing JSON:", e) raw = None # 3) Estrazione sicura della prima risposta # Trova l'ultima domanda dell'utente nella history ultima_domanda_utente = None for role, text in reversed(history): if role == "user": ultima_domanda_utente = text break if raw is None or "risposte" not in raw or not raw["risposte"]: categoria = "none" self._structured_json = { "domanda": ultima_domanda_utente or "", "categoria": categoria } else: categoria = raw["risposte"][0].get("categoria", "categoria_mancante") self._structured_json = { "domanda": ultima_domanda_utente or "", "categoria": categoria } # 1. Recupera l'ultima risposta dell'agent ultima_risposta_agent = await self.wait_for_last_agent(history) print('----------------------------') print(history) print('----------------------------') print('CATEGORIA: ', categoria) print('PREV CATEGORIA: ', self.last_categoria) # 4) Se categoria valida, pronuncia suddividendo in frasi # if categoria and categoria != "varie" and ultima_risposta_agent: # and categoria != self.last_categoria: # KB = self.categorie_data.get(categoria, {}) # await session.send_client_content( # turns={"role": "user", "parts": [ # { # "text": f''' # **Analizza con attenzione la tua risposta precedente all’utente:** # "{ultima_risposta_agent}" # **Ora valuta la seguente Knowledge Base aggiuntiva:** # "{KB["descrizione"]}" # **ISTRUZIONI:** # 1. Confronta la tua risposta precedente con le informazioni contenute nella KB. # 2. Se la tua risposta precedente era di attesa (ad esempio: hai detto "Controllo...", "Verifico...", oppure hai comunicato che avresti consultato informazioni aggiuntive), ora DEVI rispondere nuovamente all’utente: # - Se nella KB trovi le informazioni richieste, fornisci una risposta aggiornata, completa e precisa integrando questi nuovi dettagli. # - Se la KB NON contiene le informazioni richieste, informa chiaramente l’utente che non hai trovato i dati desiderati e consiglia di parlare con un consulente per ulteriori dettagli. # 3. Se la tua risposta precedente era già completa e la domanda dell’utente ha già ricevuto una risposta esaustiva, allora NON aggiungere nulla, # a meno che nella KB non siano presenti informazioni nuove, aggiuntive e rilevanti che possano migliorare la risposta già fornita. # In questo caso, integra solo i nuovi dettagli utili. # 4. NON ripetere informazioni già date. NON rispondere se la KB non aggiunge nulla di rilevante rispetto alla risposta precedente. # 5. Se la risposta precedente si era fermata a metà, continua da dove era arrivata completandola! # Rispondi in modo chiaro, sintetico e pertinente. # ''' # } # ]}, # turn_complete=True # Solo true sull’ultima frase! # ) # print("--- KB INVIATA ---") # print(KB["descrizione"]) # self.last_categoria = categoria # else: # print('NESSUN INVIO DI KB NECESSARIO') # 5) Pulisci lo stato self._structured_json = None async def wait_for_last_agent(self, history, check_interval=0.1, timeout=2.0): """ Attende che l'ultimo turno della history sia dell'agente. """ waited = 0 while waited < timeout: if history and history[-1][0] == "agent": return history[-1][1] await asyncio.sleep(check_interval) waited += check_interval # Timeout raggiunto return True def _sync_generate_structured(self, history: deque[tuple[str, str]]) -> str: """ Prende gli ultimi N turni della conversazione e li invia al modello strutturato. """ # 1. Assembla la history come mini-conversazione conversation_lines = ["Conversazione:"] for role, text in history: if role == "user": conversation_lines.append(f"Utente: {text}") else: conversation_lines.append(f"Agente: {text}") conversation_str = "\n".join(conversation_lines) print('CONVERSATION') print(conversation_str) # 2. Prepara la richiesta a Gemini (come contenuto "user" con la conversazione completa) contents = [ Content(role="user", parts=[Part.from_text(text=conversation_str)]) ] cfg = GenerateContentConfig( cached_content=structured_cache_name, # nota: lista! # ——— Impostazioni extra dal codice di esempio ——— thinking_config=ThinkingConfig(thinking_budget=0), # stop_sequences=[""], safety_settings=[ SafetySetting(category="HARM_CATEGORY_HARASSMENT", threshold="BLOCK_MEDIUM_AND_ABOVE"), SafetySetting(category="HARM_CATEGORY_HATE_SPEECH", threshold="BLOCK_MEDIUM_AND_ABOVE"), SafetySetting(category="HARM_CATEGORY_SEXUALLY_EXPLICIT", threshold="BLOCK_MEDIUM_AND_ABOVE"), SafetySetting(category="HARM_CATEGORY_DANGEROUS_CONTENT", threshold="BLOCK_MEDIUM_AND_ABOVE"), ], response_mime_type="application/json", response_schema=Schema( type=Type.OBJECT, required=["risposte"], properties={ "risposte": Schema( type=Type.ARRAY, items=Schema( type=Type.OBJECT, required=["categoria"], #, "risposta"], properties={ "categoria": Schema(type=Type.STRING), #"risposta": Schema(type=Type.STRING), }, ), ), }, ), ) # qui non c’è loop: restituisce subito un oggetto con .text response = self.client.models.generate_content( model=GEMINI_model, contents=contents, config=cfg, ) return response.text.strip() ###### SERVE PER CREARE DEGLI AUDIO, MA NON LO UTILIZZEREMO # async def _tts_via_live(self, text: str): # from google.genai.types import LiveConnectConfig, SpeechConfig, VoiceConfig, PrebuiltVoiceConfig, Content, Part # import uuid # cfg = LiveConnectConfig( # response_modalities=["AUDIO"], # speech_config=SpeechConfig( # language_code="it-IT", # voice_config=VoiceConfig(prebuilt_voice_config=PrebuiltVoiceConfig(voice_name=self.voice_name)) # ), # system_instruction=Content(parts=[Part.from_text(text=text)], role="user"), # ) # async with self.client.aio.live.connect( # model="models/gemini-2.0-flash-live-001", # config=cfg # ) as session: # # non mandiamo audio in ingresso (stream vuoto) # async for resp in session.start_stream(stream=iter(()), mime_type="audio/pcm"): # # reinietta i chunk audio al front-end # await self._outgoing_audio_queue.put((uuid.uuid4().int, resp.data)) async def feed_audio(self): """ (abbiamo ricevuto) AUDIO FRONTEND -> BACKEND (dove siamo) -> (dove lo inviamo) GEMINI Preleva i buffer raw (48 kHz) da _raw_queue, li converte a 16 kHz, li spezza in chunk da 960 sample, e li mette in _chunk_queue. """ try: chunk_size = 960 # 60 ms a 16 kHz residual = np.array([], dtype=np.int16) while True: # 1) prendo un buffer raw (48 kHz) raw = await self._raw_queue.get() #print(f"[GeminiHandler.feed_audio] estraggo raw 48k: {len(raw)} byte") # 2) converto in array int16 arr16 = np.frombuffer(raw, dtype=np.int16) ########################################################################### # 3) SE BISOGNA RICAMPIONARE da 48KHz A 16KHZ DECOMMENTA LE 2 RIGHE QUI SOTTO ########################################################################### # arr16 = scipy.signal.resample_poly(arr48, up=1, down=3) # arr16 = np.int16(np.clip(arr16, -32768, 32767)) # 4) concateno con l’eventuale residuo arr16 = np.concatenate([residual, arr16]) total = len(arr16) n_chunks = total // chunk_size residual = arr16[n_chunks * chunk_size :] # il residuo non processato # 5) ogni chunk di 960 sample → metto in _chunk_queue for i in range(n_chunks): chunk = arr16[i * chunk_size : (i + 1) * chunk_size] raw_chunk = chunk.tobytes() #print(f"[GeminiHandler.feed_audio] invio chunk 16k #{i}: {len(raw_chunk)} byte") await self._chunk_queue.put(raw_chunk) except asyncio.CancelledError: raise except Exception as e: #print(f"[GeminiHandler.feed_audio] Errore: {e}") raise def assemble_payload(self, text: str, duration: float) -> dict: """ Crea il JSON con parole, fonemi, segmenti e (eventualmente) audio base64. Se audio_bytes è vuoto (b''), omette la chiave "audio". """ words = text.strip().split() if not words: return {} word_count = len(words) avg_dur = duration / word_count if word_count else 0 word_timings = [] segments = [] phonemes = [] cursor = 0.0 for w in words: start = cursor end = cursor + avg_dur word_timings.append( {"word": w, "start": round(start, 2), "end": round(end, 2)} ) segments.append( { "id": len(segments), "seek": 0, "start": round(start, 2), "end": round(end, 2), "text": w, } ) mapped = map_to_phonemes(w) per_ph = avg_dur / len(mapped) if mapped else 0 for i, ph in enumerate(mapped): ph_start = start + i * per_ph ph_end = ph_start + per_ph phonemes.append( {"phoneme": ph, "start": round(ph_start, 2), "end": round(ph_end, 2)} ) cursor = end # Se audio_bytes non è vuoto, codificalo; altrimenti ometti la chiave "audio". payload = { "type": "viseme_phrase", "task": "transcribe", "language": "italian", "duration": round(duration, 2), "text": text.strip(), "segments": segments, "words": word_timings, "phonemes": phonemes, } if self.last_categoria: dati_cat = self.categorie_data.get(self.last_categoria, {}) # Media: scegli tra video/photo (se presenti) show_media = dati_cat.get("show_media") if show_media == "video" and dati_cat.get("video"): payload["media_type"] = "video" payload["media_url"] = dati_cat["video"] payload["media_volume"] = dati_cat["volume"] elif show_media == "photo" and dati_cat.get("photo"): payload["media_type"] = "photo" payload["media_url"] = dati_cat["photo"] else: payload["media_type"] = None payload["media_url"] = None # QR code sempre, se presente payload["qr_url"] = dati_cat.get("qr_url", None) else: payload["media_type"] = None payload["media_url"] = None payload["qr_url"] = None return payload async def _safe_broadcast(self, payload: dict): """ Invia il payload (JSON) a tutti i WebSocket registrati in self._clients. (resta identico a prima) """ data = json.dumps(payload) for ws in list(self._clients.values()): try: await ws.send_text(data) except Exception: pass async def send_to_google_sheet(self, endpoint_url: str,question: str, response: str, prompt_tokens: int, response_audio_tokens: int, response_text_tokens: int, total_tokens: int, sheet_name: str = "Foglio1"): """ Invia un record al Google Sheet via Apps Script: Data, Tokens_prompt, Tokens_response_audio, Tokens_response_text, Tokens_total, Domanda, Risposta """ now_str = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") # Sanifica i campi evitando virgole che rompono il CSV clean = lambda s: s.replace(",", " ").replace("\n", " ").strip() csv_payload = ",".join([ now_str, str(prompt_tokens), str(response_audio_tokens), str(response_text_tokens), str(total_tokens), clean(question), clean(response) ]) url = f"{endpoint_url}?sheet={sheet_name}" try: async with httpx.AsyncClient() as client: res = await client.post( url=url, content=csv_payload, headers={"Content-Type": "text/plain"} ) if res.status_code == 200: print("✅ Riga inviata al Google Sheet (async).") except Exception as e: print("❌ Errore durante l'invio al Google Sheet:", str(e)) def reset_history(self): self.chat_history.clear() async def aclose(self): """ Cancella i singoli Task creati, poi svuota le due code (_raw_queue e _chunk_queue). """ self.reset_history() if self._closed: return self._closed = True # Cancello ciascun Task esplicitamente (TaskGroup non ha cancel()) for t in self._tasks: t.cancel() # Attendo che terminino tutti await asyncio.gather(*self._tasks, return_exceptions=True) # Svuoto la coda raw (_raw_queue) while not self._raw_queue.empty(): try: self._raw_queue.get_nowait() except Exception: break # Svuoto la coda dei chunk a 16 kHz (_chunk_queue) while not self._chunk_queue.empty(): try: self._chunk_queue.get_nowait() except Exception: break ####### NON RIMUOVERE QUESTE 2 FUNZIONI CHE SONO CHIAMATE NEL MAIN ######### def register_ws(self, client_id: str, ws: "WebSocket"): self._clients[client_id] = ws def unregister_ws(self, client_id: str): self._clients.pop(client_id, None) @asynccontextmanager async def connect_gemini(client: genai.Client, session_id: str | None = None): handler = GeminiHandler(client) tg = asyncio.TaskGroup() # Nota: non registro ancora in active_handlers async with tg: try: # Creo i task t1 = tg.create_task(handler.run()) t2 = tg.create_task(handler.feed_audio()) t3 = tg.create_task(handler._audio_sender()) handler._tasks = [t1, t2, t3] # Ora posso registrare il handler if session_id is not None: active_handlers[session_id] = handler handler.last_seen = datetime.datetime.utcnow() # Yield all’interno del try garantisce che il finally sotto venga eseguito yield handler finally: # Pulizia sicura, anche se fallisce qualche tX _prima_ del yield await handler.aclose() if session_id and session_id in active_handlers: active_handlers.pop(session_id)