From a5111fc3f6b5e5a652463f4011eaffbeb4a21c6d Mon Sep 17 00:00:00 2001 From: "raghav.mehndiratta" Date: Mon, 29 Jun 2026 15:44:03 -1000 Subject: [PATCH 1/2] gemini user sim and audio quality --- .env.example | 16 +- src/eva/assistant/audio_bridge.py | 46 ++- src/eva/models/config.py | 27 +- src/eva/run_benchmark.py | 4 +- src/eva/user_simulator/audio_bridge.py | 23 +- src/eva/user_simulator/base.py | 15 + src/eva/user_simulator/factory.py | 11 +- src/eva/user_simulator/gemini_live.py | 393 ++++++++++++++++++++++ src/eva/user_simulator/openai_realtime.py | 14 +- tests/unit/user_simulator/test_factory.py | 16 +- 10 files changed, 508 insertions(+), 57 deletions(-) create mode 100644 src/eva/user_simulator/gemini_live.py diff --git a/.env.example b/.env.example index 2810e5bf..55170bdb 100644 --- a/.env.example +++ b/.env.example @@ -239,22 +239,24 @@ EVA_MODEL_LIST='[ # --- User simulator provider --- #i Caller provider. ElevenLabs is the backward-compatible default. #d enum -#e elevenlabs,openai_realtime +#e elevenlabs,openai_realtime,gemini_live EVA_USER_SIMULATOR__PROVIDER=elevenlabs -#i OpenAI Realtime caller model. Used only when provider=openai_realtime. +#i Realtime caller model. Used when provider=openai_realtime (e.g. gpt-realtime-1.5) or +#i provider=gemini_live (e.g. gemini-2.5-flash-native-audio-preview-09-2025). +#i gemini_live requires one of GEMINI_API_KEY, GOOGLE_API_KEY, GOOGLE_CLOUD_PROJECT (Vertex AI), or GOOGLE_APPLICATION_CREDENTIALS. #d string -#x EVA_USER_SIMULATOR__PROVIDER=openai_realtime +#x EVA_USER_SIMULATOR__PROVIDER=openai_realtime,gemini_live #v EVA_USER_SIMULATOR__MODEL=gpt-realtime-1.5 -#i OpenAI voice for the female caller persona. +#i Voice for the female caller persona. Provider-specific (e.g. marin for openai_realtime, Aoede for gemini_live). #d string -#x EVA_USER_SIMULATOR__PROVIDER=openai_realtime +#x EVA_USER_SIMULATOR__PROVIDER=openai_realtime,gemini_live #v EVA_USER_SIMULATOR__FEMALE_VOICE=marin -#i OpenAI voice for the male caller persona. +#i Voice for the male caller persona. Provider-specific (e.g. cedar for openai_realtime, Charon for gemini_live). #d string -#x EVA_USER_SIMULATOR__PROVIDER=openai_realtime +#x EVA_USER_SIMULATOR__PROVIDER=openai_realtime,gemini_live #v EVA_USER_SIMULATOR__MALE_VOICE=cedar # --- Language (mutually exclusive with Accent and Behavior) --- diff --git a/src/eva/assistant/audio_bridge.py b/src/eva/assistant/audio_bridge.py index af8fc9af..6efcb5ac 100644 --- a/src/eva/assistant/audio_bridge.py +++ b/src/eva/assistant/audio_bridge.py @@ -35,20 +35,42 @@ def mulaw_8k_to_pcm16_16k(mulaw_bytes: bytes) -> bytes: return pcm_16k +def pcm16_16k_to_mulaw_8k(pcm_bytes: bytes) -> bytes: + """Convert 16kHz 16-bit PCM to 8kHz mu-law. + + Uses soxr VHQ resampling for proper anti-aliasing during the 2:1 downsampling. + audioop.ratecv lacks an anti-aliasing filter, causing frequencies 4-8kHz (sibilants) + to alias back into the speech band as electrical-sounding distortion. + """ + audio_data = np.frombuffer(pcm_bytes, dtype=np.int16) + resampled = soxr.resample(audio_data, 16000, 8000, quality="VHQ") + expected_samples = round(len(audio_data) * 8000 / 16000) + if len(resampled) < expected_samples: + resampled = np.pad(resampled, (0, expected_samples - len(resampled))) + elif len(resampled) > expected_samples: + resampled = resampled[:expected_samples] + pcm_8k = np.clip(resampled, -32768, 32767).astype(np.int16).tobytes() + return audioop.lin2ulaw(pcm_8k, 2) + + def mulaw_8k_to_pcm16_24k(mulaw_bytes: bytes) -> bytes: - """Convert 8kHz mu-law audio to 24kHz 16-bit PCM.""" + """Convert 8kHz mu-law audio to 24kHz 16-bit PCM. + + Uses soxr VHQ resampling for clean 3:1 upsampling. audioop.ratecv restarts + its interpolation filter on every call (stateless) which creates audible + discontinuities at chunk boundaries for the 3:1 ratio. + """ # Decode mu-law to 16-bit PCM at 8kHz pcm_8k = audioop.ulaw2lin(mulaw_bytes, 2) - # Upsample from 8kHz to 24kHz - pcm_24k, _ = audioop.ratecv(pcm_8k, 2, 1, 8000, 24000, None) - # audioop.ratecv can produce ±2 samples; clamp to exact 3× input length - # so that the inverse conversion recovers the original sample count. - expected_bytes = len(pcm_8k) * 3 - if len(pcm_24k) < expected_bytes: - pcm_24k = pcm_24k + b"\x00" * (expected_bytes - len(pcm_24k)) - elif len(pcm_24k) > expected_bytes: - pcm_24k = pcm_24k[:expected_bytes] - return pcm_24k + # Upsample from 8kHz to 24kHz using high-quality resampler + audio_data = np.frombuffer(pcm_8k, dtype=np.int16) + resampled = soxr.resample(audio_data, 8000, 24000, quality="VHQ") + expected_samples = round(len(audio_data) * 24000 / 8000) + if len(resampled) < expected_samples: + resampled = np.pad(resampled, (0, expected_samples - len(resampled))) + elif len(resampled) > expected_samples: + resampled = resampled[:expected_samples] + return np.clip(resampled, -32768, 32767).astype(np.int16).tobytes() def pcm16_24k_to_mulaw_8k(pcm_bytes: bytes) -> bytes: @@ -67,7 +89,7 @@ def pcm16_24k_to_mulaw_8k(pcm_bytes: bytes) -> bytes: resampled = np.pad(resampled, (0, expected_samples - len(resampled))) elif len(resampled) > expected_samples: resampled = resampled[:expected_samples] - pcm_8k = resampled.astype(np.int16).tobytes() + pcm_8k = np.clip(resampled, -32768, 32767).astype(np.int16).tobytes() # Encode to mu-law return audioop.lin2ulaw(pcm_8k, 2) diff --git a/src/eva/models/config.py b/src/eva/models/config.py index 561f54c6..416be552 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -369,8 +369,17 @@ class OpenAIRealtimeSimulatorConfig(BaseModel): male_voice: str = Field("cedar", description="Voice used for male caller personas.") +class GeminiLiveSimulatorConfig(BaseModel): + """Gemini Live-specific settings for the user simulator.""" + + provider: Literal["gemini_live"] = "gemini_live" + model: str = Field("gemini-2.5-flash-native-audio-preview-09-2025", description="Gemini Live model.") + female_voice: str = Field("Aoede", description="Voice used for female caller personas.") + male_voice: str = Field("Charon", description="Voice used for male caller personas.") + + UserSimulatorConfig = Annotated[ - ElevenLabsSimulatorConfig | OpenAIRealtimeSimulatorConfig, + ElevenLabsSimulatorConfig | OpenAIRealtimeSimulatorConfig | GeminiLiveSimulatorConfig, Field(discriminator="provider"), ] @@ -720,6 +729,22 @@ def _check_openai_realtime_simulator(self) -> "RunConfig": raise ValueError("EVA_USER_SIMULATOR__PROVIDER=openai_realtime requires OPENAI_API_KEY to be set.") return self + @model_validator(mode="after") + def _check_gemini_live_simulator(self) -> "RunConfig": + """When gemini_live user simulator is selected, Gemini credentials must be present.""" + if not isinstance(self.user_simulator, GeminiLiveSimulatorConfig): + return self + has_credentials = any( + os.environ.get(key) + for key in ("GEMINI_API_KEY", "GOOGLE_API_KEY", "GOOGLE_CLOUD_PROJECT", "GOOGLE_APPLICATION_CREDENTIALS") + ) + if not has_credentials: + raise ValueError( + "EVA_USER_SIMULATOR__PROVIDER=gemini_live requires one of GEMINI_API_KEY, GOOGLE_API_KEY, " + "GOOGLE_CLOUD_PROJECT (Vertex AI), or GOOGLE_APPLICATION_CREDENTIALS to be set." + ) + return self + @model_validator(mode="before") @classmethod def _handle_all_keyword(cls, data: Any): diff --git a/src/eva/run_benchmark.py b/src/eva/run_benchmark.py index 44d7e56b..c3dba1ca 100644 --- a/src/eva/run_benchmark.py +++ b/src/eva/run_benchmark.py @@ -6,7 +6,7 @@ from dotenv import load_dotenv from eva.metrics.runner import MetricsRunner -from eva.models.config import OpenAIRealtimeSimulatorConfig, PipelineType, RunConfig +from eva.models.config import GeminiLiveSimulatorConfig, OpenAIRealtimeSimulatorConfig, PipelineType, RunConfig from eva.models.record import EvaluationRecord from eva.orchestrator.runner import BenchmarkRunner from eva.utils import router @@ -113,7 +113,7 @@ async def run_benchmark(config: RunConfig) -> int: else: logger.info(f" S2S model: {config.model.s2s}") logger.info(f" User simulator: {config.user_simulator.provider}") - if isinstance(config.user_simulator, OpenAIRealtimeSimulatorConfig): + if isinstance(config.user_simulator, (OpenAIRealtimeSimulatorConfig, GeminiLiveSimulatorConfig)): logger.info(f" User simulator model: {config.user_simulator.model}") logger.info(f" Max concurrent: {config.max_concurrent_conversations}") logger.info(f" Time limit: {config.conversation_time_limit_seconds}s") diff --git a/src/eva/user_simulator/audio_bridge.py b/src/eva/user_simulator/audio_bridge.py index d1e75415..5c86ba35 100644 --- a/src/eva/user_simulator/audio_bridge.py +++ b/src/eva/user_simulator/audio_bridge.py @@ -14,15 +14,10 @@ from collections.abc import Callable import websockets -from websockets.protocol import State as WebSocketState - -try: - import audioop -except ImportError: - import audioop_lts as audioop - from elevenlabs.conversational_ai.conversation import AudioInterface +from websockets.protocol import State as WebSocketState +from eva.assistant.audio_bridge import pcm16_16k_to_mulaw_8k from eva.user_simulator.perturbation import AudioPerturbator from eva.utils.logging import get_logger @@ -294,19 +289,7 @@ def _convert_pcm_to_mulaw(pcm_data: bytes) -> bytes: mulaw encoded audio data at 8kHz (for assistant) """ try: - # Downsample from 16kHz to 8kHz - pcm_8khz, _ = audioop.ratecv( - pcm_data, - PCM_SAMPLE_WIDTH, # 16-bit PCM - 1, # mono - ELEVENLABS_OUTPUT_RATE, # from 16kHz - ASSISTANT_SAMPLE_RATE, # to 8kHz - None, - ) - - # Convert 16-bit PCM to μ-law - mulaw_data = audioop.lin2ulaw(pcm_8khz, PCM_SAMPLE_WIDTH) - return mulaw_data + return pcm16_16k_to_mulaw_8k(pcm_data) except Exception as e: logger.warning(f"Error converting PCM to mulaw: {e}") return b"" diff --git a/src/eva/user_simulator/base.py b/src/eva/user_simulator/base.py index bef28e3a..d71f1814 100644 --- a/src/eva/user_simulator/base.py +++ b/src/eva/user_simulator/base.py @@ -21,6 +21,21 @@ _BEHAVIORS_PATH = Path(__file__).parent.parent.parent.parent / "configs" / "user_behaviors.yaml" +# Shared across realtime-style callers (OpenAI Realtime, Gemini Live) that expose +# an ``end_call`` function tool to let the simulated caller hang up. +END_CALL_DESCRIPTION = """Use this to end the phone call and hang up. + +Call this function when it is time to end the call and one of the following is true: +1. The agent has confirmed your request is resolved, all steps are completed, and you have said goodbye. +2. The agent has initiated a transfer to a live agent. +3. The agent has been unable to make progress for at least 5 consecutive turns. +4. The agent says goodbye or indicates the conversation is over. +5. The agent indicates that the remainder of your request cannot be fulfilled. +6. The assistant reports an unrecoverable processing error. + +Never call this tool in the same turn that you provide the agent with data, an identifier, +an approval to proceed, a transfer request, or any other information. Say a brief goodbye first.""" + @lru_cache(maxsize=1) def load_behavior_prompts() -> dict: diff --git a/src/eva/user_simulator/factory.py b/src/eva/user_simulator/factory.py index d2cef7e2..41852cb3 100644 --- a/src/eva/user_simulator/factory.py +++ b/src/eva/user_simulator/factory.py @@ -4,7 +4,12 @@ from typing import Any -from eva.models.config import ElevenLabsSimulatorConfig, OpenAIRealtimeSimulatorConfig, UserSimulatorConfig +from eva.models.config import ( + ElevenLabsSimulatorConfig, + GeminiLiveSimulatorConfig, + OpenAIRealtimeSimulatorConfig, + UserSimulatorConfig, +) from eva.user_simulator.base import AbstractUserSimulator @@ -21,4 +26,8 @@ def create_user_simulator( from eva.user_simulator.openai_realtime import OpenAIRealtimeUserSimulator return OpenAIRealtimeUserSimulator(simulator_config=simulator_config, **kwargs) + if isinstance(simulator_config, GeminiLiveSimulatorConfig): + from eva.user_simulator.gemini_live import GeminiLiveUserSimulator + + return GeminiLiveUserSimulator(simulator_config=simulator_config, **kwargs) raise ValueError(f"Unknown user simulator provider: {simulator_config.provider!r}") diff --git a/src/eva/user_simulator/gemini_live.py b/src/eva/user_simulator/gemini_live.py new file mode 100644 index 00000000..f9d5d233 --- /dev/null +++ b/src/eva/user_simulator/gemini_live.py @@ -0,0 +1,393 @@ +"""Gemini Live implementation of the EVA simulated caller. + +Mirrors the OpenAI Realtime caller but drives a second Gemini Live session as the +simulated user. Audio flows: + + Assistant (8 kHz mulaw, via the audio bridge) + -> 16 kHz PCM16 -> Gemini Live realtime input + Gemini Live output (24 kHz PCM16) + -> 16 kHz PCM16 -> audio bridge -> assistant + +Gemini's automatic activity detection handles turn-taking, so unlike the OpenAI +caller there is no manual response sequencing. An ``end_call`` function tool lets +the caller hang up under the same conditions as the OpenAI caller. +""" + +from __future__ import annotations + +import asyncio +import os +from contextlib import suppress +from pathlib import Path +from typing import Any + +from google import genai +from google.genai import types + +try: + import audioop +except ImportError: + import audioop_lts as audioop + +from eva.models.config import GeminiLiveSimulatorConfig, PerturbationConfig +from eva.user_simulator.audio_bridge import BotToBotAudioBridge +from eva.user_simulator.base import END_CALL_DESCRIPTION, AbstractUserSimulator +from eva.utils.audio_utils import save_pcm_as_wav +from eva.utils.logging import get_logger + +logger = get_logger(__name__) + +GEMINI_SAMPLE_RATE = 24000 +BRIDGE_SAMPLE_RATE = 16000 +ASSISTANT_SAMPLE_RATE = 8000 +_PERSONA_GENDER = {1: "F", 2: "M"} + +# Known Gemini Live prebuilt voice names. An unrecognised voice is accepted +# silently by the API and falls back to a default voice, so warn up front. +# Half-cascade models (e.g. gemini-3.1-flash-live-preview) support the first +# eight; native-audio models support the full set. +_KNOWN_GEMINI_VOICES = frozenset( + { + "Aoede", "Charon", "Fenrir", "Kore", "Leda", "Orus", "Puck", "Zephyr", + "Achernar", "Achird", "Algenib", "Algieba", "Alnilam", "Autonoe", + "Callirrhoe", "Despina", "Enceladus", "Erinome", "Gacrux", "Iapetus", + "Laomedeia", "Pulcherrima", "Rasalgethi", "Sadachbia", "Sadaltager", + "Schedar", "Sulafat", "Umbriel", "Vindemiatrix", "Zubenelgenubi", + } +) + + +class GeminiLiveUserSimulator(AbstractUserSimulator): + """Use a Gemini Live session as EVA's simulated caller.""" + + def __init__( + self, + current_date_time: str, + persona_config: dict, + goal: dict, + server_url: str, + output_dir: Path, + agent_id: str, + timeout: int = 600, + perturbation_config: PerturbationConfig | None = None, + language: str = "en", + *, + simulator_config: GeminiLiveSimulatorConfig, + ) -> None: + super().__init__( + current_date_time=current_date_time, + persona_config=persona_config, + goal=goal, + server_url=server_url, + output_dir=output_dir, + agent_id=agent_id, + timeout=timeout, + perturbation_config=perturbation_config, + language=language, + provider="gemini_live", + ) + if perturbation_config and perturbation_config.accent is not None: + raise ValueError("Gemini Live caller does not support ElevenLabs-specific accent variants") + self.simulator_config = simulator_config + self._assistant_audio_queue: asyncio.Queue[bytes] = asyncio.Queue() + self._caller_transcript_parts: list[str] = [] + self._caller_audio_seen = False + self._input_resampler_state = None + self._output_resampler_state = None + + @property + def caller_model(self) -> str: + return self.simulator_config.model + + @property + def caller_voice(self) -> str: + gender = _PERSONA_GENDER.get(self.persona_config.get("user_persona_id")) + if gender == "M": + return self.simulator_config.male_voice + return self.simulator_config.female_voice + + def _build_live_config(self) -> types.LiveConnectConfig: + voice = self.caller_voice + if voice.lower() not in {v.lower() for v in _KNOWN_GEMINI_VOICES}: + logger.warning( + f"Configured Gemini caller voice {voice!r} is not a recognised prebuilt voice; " + f"Gemini will silently fall back to a default voice. Known voices: " + f"{', '.join(sorted(_KNOWN_GEMINI_VOICES))}" + ) + self.event_logger.log_event("invalid_voice", {"voice": voice}) + return types.LiveConnectConfig( + response_modalities=[types.Modality.AUDIO], + system_instruction=self._build_prompt(), + speech_config=types.SpeechConfig( + voice_config=types.VoiceConfig( + prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=voice) + ), + ), + input_audio_transcription=types.AudioTranscriptionConfig(), + output_audio_transcription=types.AudioTranscriptionConfig(), + tools=[ + types.Tool( + function_declarations=[ + types.FunctionDeclaration( + name="end_call", + description=END_CALL_DESCRIPTION, + parameters=types.Schema(type="OBJECT", properties={}), + behavior=types.Behavior.BLOCKING, + ) + ] + ) + ], + ) + + def _create_client(self, api_key: str | None) -> genai.Client: + """Create a google-genai client. + + There are two distinct auth backends and you need exactly ONE of them — + a service-account JSON and an API key are not combined: + + 1. Vertex AI — GOOGLE_CLOUD_PROJECT (+ GOOGLE_CLOUD_LOCATION). Authenticated + via Application Default Credentials, i.e. GOOGLE_APPLICATION_CREDENTIALS + (service-account JSON), workload identity, or ``gcloud auth``. The API + key is ignored on this path. + 2. Gemini Developer API — GEMINI_API_KEY / GOOGLE_API_KEY. The + service-account JSON is ignored on this path. + + Vertex is preferred when a project is configured so service-account + deployments work; otherwise we fall back to the API key, then to the + SDK's own default resolution (e.g. GOOGLE_GENAI_USE_VERTEXAI=true + ADC). + """ + project = os.environ.get("GOOGLE_CLOUD_PROJECT") + if project: + location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1") + logger.info(f"Gemini caller using Vertex AI (project={project}, location={location})") + return genai.Client(vertexai=True, project=project, location=location) + if api_key: + logger.info("Gemini caller using Developer API key") + return genai.Client(api_key=api_key) + logger.warning("No explicit Gemini credentials; relying on google-genai default resolution") + return genai.Client() + + async def run_conversation(self) -> str: + api_key = os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY") + if not api_key and not os.environ.get("GOOGLE_CLOUD_PROJECT") and not os.environ.get( + "GOOGLE_APPLICATION_CREDENTIALS" + ): + raise ValueError( + "Gemini Live caller requires one of GEMINI_API_KEY, GOOGLE_API_KEY, GOOGLE_CLOUD_PROJECT, " + "or GOOGLE_APPLICATION_CREDENTIALS" + ) + + try: + await self._run_gemini_conversation(api_key) + except Exception as exc: + logger.error(f"Gemini caller simulation error: {exc}", exc_info=True) + self._end_reason = "error" + self.event_logger.log_error(str(exc)) + if self._audio_interface is not None: + with suppress(Exception): + await self._audio_interface.stop_async() + self.event_logger.log_connection_state("session_ended", {"reason": self._end_reason}) + finally: + self.event_logger.save() + return self._end_reason + + async def _run_gemini_conversation(self, api_key: str | None) -> None: + conversation_id = self.output_dir.name + self._audio_interface = BotToBotAudioBridge( + websocket_uri=self.server_url, + conversation_id=conversation_id, + record_callback=self._record_audio, + event_logger=self.event_logger, + conversation_done_callback=self._on_conversation_end, + perturbator=self._perturbator, + disconnect_reason="assistant_disconnect", + ) + await self._audio_interface.start_async() + self._audio_interface.start(self._on_assistant_audio) + self.event_logger.log_connection_state( + "connected", + { + "server_url": self.server_url, + "caller_provider": self.provider, + "caller_model": self.caller_model, + "caller_voice": self.caller_voice, + "caller_input_format": "audio/pcm;rate=16000", + "caller_output_format": f"audio/pcm;rate={GEMINI_SAMPLE_RATE}", + "assistant_input_transport": "audio/pcmu_8000hz", + "caller_turn_detection": "gemini_automatic_activity_detection", + }, + ) + + client = self._create_client(api_key) + forward_task: asyncio.Task | None = None + listener_task: asyncio.Task | None = None + completion_task: asyncio.Task | None = None + try: + async with client.aio.live.connect(model=self.caller_model, config=self._build_live_config()) as session: + self.event_logger.log_connection_state("session_started") + forward_task = asyncio.create_task(self._forward_assistant_audio(session)) + listener_task = asyncio.create_task(self._listen_for_caller_events(session)) + completion_task = asyncio.create_task(self._wait_for_conversation_end()) + + await self._wait_for_session_completion(completion_task, forward_task, listener_task) + + # Allow final goodbye audio and transcripts to flush before closing. + await asyncio.sleep(4.0) + finally: + for task in (completion_task, forward_task, listener_task): + if task is not None: + await self._cancel_background_task(task) + await self._audio_interface.stop_async() + self._save_user_audio() + self.event_logger.log_connection_state("session_ended", {"reason": self._end_reason}) + + @staticmethod + async def _cancel_background_task(task: asyncio.Task) -> None: + """Cancel and consume a background task without interrupting cleanup.""" + task.cancel() + with suppress(asyncio.CancelledError, Exception): + await task + + async def _wait_for_conversation_end(self) -> None: + try: + await asyncio.wait_for(self._conversation_done.wait(), timeout=self.timeout) + except TimeoutError: + self.event_logger.log_event("timeout", {"duration": self.timeout}) + self._on_conversation_end("timeout") + + async def _wait_for_session_completion( + self, + completion_task: asyncio.Task, + forward_task: asyncio.Task, + listener_task: asyncio.Task, + ) -> None: + done, _ = await asyncio.wait( + {completion_task, forward_task, listener_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + if completion_task in done: + return + + finished_task = next(iter(done)) + if self._conversation_done.is_set(): + await completion_task + return + + exception = finished_task.exception() + if exception is not None: + raise exception + task_name = "listener" if finished_task is listener_task else "audio forwarder" + raise RuntimeError(f"Gemini Live {task_name} stopped unexpectedly") + + def _on_assistant_audio(self, mulaw_audio: bytes) -> None: + # Don't echo the caller's own audio back into Gemini while it is speaking. + if mulaw_audio and not self._caller_audio_is_playing(): + self._assistant_audio_queue.put_nowait(mulaw_audio) + + def _caller_audio_is_playing(self) -> bool: + if self._audio_interface is None: + return False + return self._audio_interface.is_caller_playing() + + async def _forward_assistant_audio(self, session: Any) -> None: + while True: + mulaw_audio = await self._assistant_audio_queue.get() + if not mulaw_audio: + continue + pcm16_8k = audioop.ulaw2lin(mulaw_audio, 2) + pcm16_16k, self._input_resampler_state = audioop.ratecv( + pcm16_8k, + 2, + 1, + ASSISTANT_SAMPLE_RATE, + BRIDGE_SAMPLE_RATE, + self._input_resampler_state, + ) + with suppress(Exception): + await session.send_realtime_input( + audio=types.Blob(data=pcm16_16k, mime_type="audio/pcm;rate=16000") + ) + + async def _listen_for_caller_events(self, session: Any) -> None: + try: + while not self._conversation_done.is_set(): + try: + response = await asyncio.wait_for(session._receive(), timeout=2.0) + except TimeoutError: + continue + if response is None: + continue + await self._handle_caller_event(session, response) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.error(f"Gemini caller event loop error: {exc}", exc_info=True) + self.event_logger.log_error(str(exc)) + self._on_conversation_end("error") + + async def _handle_caller_event(self, session: Any, response: Any) -> None: + sc = getattr(response, "server_content", None) + if sc is not None: + if sc.model_turn: + for part in sc.model_turn.parts: + if part.inline_data and part.inline_data.data: + self._emit_caller_audio(bytes(part.inline_data.data)) + if sc.input_transcription and (sc.input_transcription.text or "").strip(): + # Input transcription = what Gemini heard, i.e. the assistant. + self._on_assistant_speaks(sc.input_transcription.text.strip()) + if sc.output_transcription and (sc.output_transcription.text or "").strip(): + # Output transcription = the simulated caller's own speech. + self._caller_transcript_parts.append(sc.output_transcription.text.strip()) + if sc.turn_complete: + self._flush_caller_transcript() + self._flush_caller_output() + self._output_resampler_state = None + + tool_call = getattr(response, "tool_call", None) + if tool_call: + for fc in tool_call.function_calls: + if fc.name == "end_call": + self.event_logger.log_event("tool_call", {"name": "end_call", "arguments": dict(fc.args or {})}) + with suppress(Exception): + await session.send_tool_response( + function_responses=[ + types.FunctionResponse(id=fc.id, name=fc.name, response={"status": "ended"}) + ] + ) + self._flush_caller_transcript() + self._on_conversation_end("goodbye") + + def _emit_caller_audio(self, pcm16_24k: bytes) -> None: + if self._audio_interface is None or len(pcm16_24k) < 2: + return + pcm16_16k, self._output_resampler_state = audioop.ratecv( + pcm16_24k, + 2, + 1, + GEMINI_SAMPLE_RATE, + BRIDGE_SAMPLE_RATE, + self._output_resampler_state, + ) + self._audio_interface.output(pcm16_16k) + self._caller_audio_seen = True + + def _flush_caller_transcript(self) -> None: + transcript = " ".join(part for part in self._caller_transcript_parts if part).strip() + self._caller_transcript_parts.clear() + if transcript: + self._on_user_speaks(transcript) + + def _flush_caller_output(self) -> None: + if self._caller_audio_seen and self._audio_interface is not None: + self._audio_interface.output(b"\x00\x00") + self._caller_audio_seen = False + + def _save_user_audio(self) -> None: + if not self._user_clean_audio_chunks: + return + save_pcm_as_wav( + b"".join(self._user_clean_audio_chunks), + self.output_dir / "audio_user_clean.wav", + sample_rate=BRIDGE_SAMPLE_RATE, + num_channels=1, + ) diff --git a/src/eva/user_simulator/openai_realtime.py b/src/eva/user_simulator/openai_realtime.py index e056fd89..24ce1db7 100644 --- a/src/eva/user_simulator/openai_realtime.py +++ b/src/eva/user_simulator/openai_realtime.py @@ -19,7 +19,7 @@ from eva.models.config import OpenAIRealtimeSimulatorConfig, PerturbationConfig from eva.user_simulator.audio_bridge import BotToBotAudioBridge -from eva.user_simulator.base import AbstractUserSimulator +from eva.user_simulator.base import END_CALL_DESCRIPTION, AbstractUserSimulator from eva.utils.audio_utils import save_pcm_as_wav from eva.utils.logging import get_logger @@ -43,18 +43,6 @@ CALLER_RESPONSE_SETTLE_SECONDS = 2.0 CALLER_RESPONSE_POLL_SECONDS = 0.05 CALLER_PLAYBACK_DRAIN_SECONDS = 15.0 -END_CALL_DESCRIPTION = """Use this to end the phone call and hang up. - -Call this function when it is time to end the call and one of the following is true: -1. The agent has confirmed your request is resolved, all steps are completed, and you have said goodbye. -2. The agent has initiated a transfer to a live agent. -3. The agent has been unable to make progress for at least 5 consecutive turns. -4. The agent says goodbye or indicates the conversation is over. -5. The agent indicates that the remainder of your request cannot be fulfilled. -6. The assistant reports an unrecoverable processing error. - -Never call this tool in the same turn that you provide the agent with data, an identifier, -an approval to proceed, a transfer request, or any other information. Say a brief goodbye first.""" class OpenAIRealtimeUserSimulator(AbstractUserSimulator): diff --git a/tests/unit/user_simulator/test_factory.py b/tests/unit/user_simulator/test_factory.py index d47f1abe..aa60aea9 100644 --- a/tests/unit/user_simulator/test_factory.py +++ b/tests/unit/user_simulator/test_factory.py @@ -1,8 +1,13 @@ from pathlib import Path -from eva.models.config import ElevenLabsSimulatorConfig, OpenAIRealtimeSimulatorConfig +from eva.models.config import ( + ElevenLabsSimulatorConfig, + GeminiLiveSimulatorConfig, + OpenAIRealtimeSimulatorConfig, +) from eva.user_simulator.elevenlabs import ElevenLabsUserSimulator from eva.user_simulator.factory import create_user_simulator +from eva.user_simulator.gemini_live import GeminiLiveUserSimulator from eva.user_simulator.openai_realtime import OpenAIRealtimeUserSimulator @@ -43,3 +48,12 @@ def test_factory_selects_openai_realtime(tmp_path): assert isinstance(simulator, OpenAIRealtimeUserSimulator) assert simulator.caller_model == "gpt-realtime-1.5" + + +def test_factory_selects_gemini_live(tmp_path): + config = GeminiLiveSimulatorConfig() + simulator = create_user_simulator(config, **_kwargs(tmp_path)) + + assert isinstance(simulator, GeminiLiveUserSimulator) + assert simulator.provider == "gemini_live" + assert simulator.caller_model == "gemini-2.5-flash-native-audio-preview-09-2025" From 43b70ef21277f68f35e6d7688938b7ad6443f329 Mon Sep 17 00:00:00 2001 From: raghavm243512 <44511569+raghavm243512@users.noreply.github.com> Date: Tue, 30 Jun 2026 01:52:13 +0000 Subject: [PATCH 2/2] Apply pre-commit --- src/eva/user_simulator/gemini_live.py | 49 ++++++++++++++++++++------- 1 file changed, 36 insertions(+), 13 deletions(-) diff --git a/src/eva/user_simulator/gemini_live.py b/src/eva/user_simulator/gemini_live.py index f9d5d233..d301bad0 100644 --- a/src/eva/user_simulator/gemini_live.py +++ b/src/eva/user_simulator/gemini_live.py @@ -48,11 +48,36 @@ # eight; native-audio models support the full set. _KNOWN_GEMINI_VOICES = frozenset( { - "Aoede", "Charon", "Fenrir", "Kore", "Leda", "Orus", "Puck", "Zephyr", - "Achernar", "Achird", "Algenib", "Algieba", "Alnilam", "Autonoe", - "Callirrhoe", "Despina", "Enceladus", "Erinome", "Gacrux", "Iapetus", - "Laomedeia", "Pulcherrima", "Rasalgethi", "Sadachbia", "Sadaltager", - "Schedar", "Sulafat", "Umbriel", "Vindemiatrix", "Zubenelgenubi", + "Aoede", + "Charon", + "Fenrir", + "Kore", + "Leda", + "Orus", + "Puck", + "Zephyr", + "Achernar", + "Achird", + "Algenib", + "Algieba", + "Alnilam", + "Autonoe", + "Callirrhoe", + "Despina", + "Enceladus", + "Erinome", + "Gacrux", + "Iapetus", + "Laomedeia", + "Pulcherrima", + "Rasalgethi", + "Sadachbia", + "Sadaltager", + "Schedar", + "Sulafat", + "Umbriel", + "Vindemiatrix", + "Zubenelgenubi", } ) @@ -119,9 +144,7 @@ def _build_live_config(self) -> types.LiveConnectConfig: response_modalities=[types.Modality.AUDIO], system_instruction=self._build_prompt(), speech_config=types.SpeechConfig( - voice_config=types.VoiceConfig( - prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=voice) - ), + voice_config=types.VoiceConfig(prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=voice)), ), input_audio_transcription=types.AudioTranscriptionConfig(), output_audio_transcription=types.AudioTranscriptionConfig(), @@ -169,8 +192,10 @@ def _create_client(self, api_key: str | None) -> genai.Client: async def run_conversation(self) -> str: api_key = os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY") - if not api_key and not os.environ.get("GOOGLE_CLOUD_PROJECT") and not os.environ.get( - "GOOGLE_APPLICATION_CREDENTIALS" + if ( + not api_key + and not os.environ.get("GOOGLE_CLOUD_PROJECT") + and not os.environ.get("GOOGLE_APPLICATION_CREDENTIALS") ): raise ValueError( "Gemini Live caller requires one of GEMINI_API_KEY, GOOGLE_API_KEY, GOOGLE_CLOUD_PROJECT, " @@ -304,9 +329,7 @@ async def _forward_assistant_audio(self, session: Any) -> None: self._input_resampler_state, ) with suppress(Exception): - await session.send_realtime_input( - audio=types.Blob(data=pcm16_16k, mime_type="audio/pcm;rate=16000") - ) + await session.send_realtime_input(audio=types.Blob(data=pcm16_16k, mime_type="audio/pcm;rate=16000")) async def _listen_for_caller_events(self, session: Any) -> None: try: