diff --git a/.env.example b/.env.example index 2810e5bf..b54cbfd1 100644 --- a/.env.example +++ b/.env.example @@ -69,7 +69,7 @@ EVA_MODEL__LLM=gpt-5.2 # --- LLM mode: STT --- #i STT provider for the voice pipeline. #d enum -#e assemblyai,cartesia,deepgram,deepgram-flux,elevenlabs,nvidia,nvidia-baseten,openai +#e assemblyai,cartesia,cartesia-multilingual,deepgram,deepgram-flux,elevenlabs,nvidia,nvidia-baseten,openai #x pipeline_mode=LLM EVA_MODEL__STT=cartesia @@ -77,7 +77,7 @@ EVA_MODEL__STT=cartesia #i Some providers also accept extra provider-specific tuning parameters here. #d json_object #x pipeline_mode=LLM -EVA_MODEL__STT_PARAMS='{"api_key": "your_cartesia_api_key", "model": "ink-whisper"}' +EVA_MODEL__STT_PARAMS='{"api_key": "your_cartesia_api_key", "model": "ink-2"}' # --- TTS (LLM and AudioLLM modes) --- #i TTS provider for the voice pipeline. @@ -92,6 +92,23 @@ EVA_MODEL__TTS=cartesia #x pipeline_mode=LLM,AudioLLM EVA_MODEL__TTS_PARAMS='{"api_key": "your_cartesia_api_key", "model": "sonic"}' +# --- LLM mode: cascade latency optimizations --- +#i Prompt a model-generated lead-in before a tool call: 'off' or 'auto'. +#d enum +#e off,auto +#x pipeline_mode=LLM +# EVA_MODEL__PRE_TOOL_SPEECH=off + +#i Stream Chat Completions output to TTS sentence-by-sentence; Responses API falls back with a warning. +#d bool +#x pipeline_mode=LLM +# EVA_MODEL__LLM_STREAMING=false + +#i Forward provider parallel_tool_calls when tools are present; leave unset for defaults. +#d bool +#x pipeline_mode=LLM +# EVA_MODEL__PARALLEL_TOOL_CALLS=false + # --- S2S mode --- #i Speech-to-speech model name. #d string diff --git a/configs/prompts/simulation.yaml b/configs/prompts/simulation.yaml index 19e3fa58..b11f4172 100644 --- a/configs/prompts/simulation.yaml +++ b/configs/prompts/simulation.yaml @@ -64,6 +64,14 @@ agent: - Use only information from the current conversation - Ask for clarification only when truly necessary - Request one or two details maximum per turn + - Never re-ask for information the caller has already provided in this conversation. Before asking for anything, check the conversation history — if you already have the value, use it. + - If a tool call fails or returns no result, do not retry with the same parameters and do not ask the caller to provide the same information from scratch. Instead, read back exactly what you used ("I tried looking up confirmation number A B C one two three for Smith — I'm not finding anything. Did I get that right?"), and if the caller corrects it, retry once with the corrected value. If you still cannot find a match, ask the user to spell it out for you letter by letter. + + + pre_tool_speech: | + ## Responsiveness + Before calling a tool, say one brief sentence first. For lookups, use a short lead-in (e.g. "Let me pull that up."). Before any action that is irreversible or has real-world consequences, state what you'll do, get confirmation, then call the tool. + If calling multiple tools at once, only one lead-in is needed. audio_llm_agent: system_prompt: | @@ -122,6 +130,9 @@ audio_llm_agent: - Use only information from the current conversation - Ask for clarification only when truly necessary - Request one or two details maximum per turn + - Never re-ask for information the caller has already provided in this conversation. Before asking for anything, check the conversation history — if you already have the value, use it. + - If a tool call fails or returns no result, do not retry with the same parameters and do not ask the caller to provide the same information from scratch. Instead, read back exactly what you used ("I tried looking up confirmation number A B C one two three for Smith — I'm not finding anything. Did I get that right?"), and if the caller corrects it, retry once with the corrected value. If you still cannot find a match, ask the user to spell it out for you letter by letter. + realtime_agent: system_prompt: | @@ -182,6 +193,9 @@ realtime_agent: - Use only information from the current conversation - Ask for clarification only when truly necessary - Request one or two details maximum per turn + - Never re-ask for information the caller has already provided in this conversation. Before asking for anything, check the conversation history — if you already have the value, use it. + - If a tool call fails or returns no result, do not retry with the same parameters and do not ask the caller to provide the same information from scratch. Instead, read back exactly what you used ("I tried looking up confirmation number A B C one two three for Smith — I'm not finding anything. Did I get that right?"), and if the caller corrects it, retry once with the corrected value. If you still cannot find a match, ask the user to spell it out for you letter by letter. + # ============================================== # User Simulator Prompts diff --git a/docs/experiment_setup.md b/docs/experiment_setup.md index 2373f179..805f4d7e 100644 --- a/docs/experiment_setup.md +++ b/docs/experiment_setup.md @@ -68,11 +68,24 @@ The table below lists all the other API-hosted models. | nova-3 | Deepgram | STT | -- | | aura-2-helena-en | Deepgram | TTS | voice: helena; language: en | +### Cascade latency optimizations + +The CASCADE pipeline exposes three optional latency levers. They are off or unset by default and affect only STT -> LLM -> TTS runs: + +| Flag | Values | Effect | +|------|--------|--------| +| `EVA_MODEL__PRE_TOOL_SPEECH` | `off` / `auto` | Ask the model to speak a brief lead-in before running a tool. The lead-in is model-generated, not templated. | +| `EVA_MODEL__LLM_STREAMING` | `true` / `false` | Stream Chat Completions output to TTS sentence-by-sentence. Responses API deployments warn and use non-streaming completion. | +| `EVA_MODEL__PARALLEL_TOOL_CALLS` | unset / `true` / `false` | Forward the provider's `parallel_tool_calls` setting when tools are present. Leave unset for provider defaults; set `false` to match the ElevenAgents assistant config. | + +Streaming still records one assistant audit entry per completed turn. If streaming is interrupted after speech starts, the spoken prefix is recorded so scored transcripts match emitted audio. + ### Turn Detection Configurations We use the default turn detection configurations for most framework in our experiments. Each framework offers varying levels of configurability, making it difficult to standardize exact parameters and turn strategies across evaluations. - **Pipecat.** The default start strategy uses VAD (voice activity detection) or transcription receipt to determine when the user begins speaking, and the stop strategy uses AI-powered turn detection via `LocalSmartTurnAnalyzerV3` to determine when the user finishes speaking. +- **Cartesia STT.** `EVA_MODEL__STT=cartesia` uses ink-2 self-endpointing and auto-selects `TURN_START_STRATEGY=external`, `TURN_STOP_STRATEGY=external`, and `VAD=none`; `cartesia-multilingual` keeps the ink-whisper VAD/smart-turn path. - **OpenAI Realtime.** We use the default server VAD, which uses periods of silence to detect turn boundaries. Default values are used for `threshold`, `prefix_padding_ms`, and `silence_duration`. - **ElevenAgents.** The turn "eagerness" parameter was set to `eager`. - **Gemini Live.** We use the default automatic VAD provided. diff --git a/src/eva/assistant/agentic/audio_llm_system.py b/src/eva/assistant/agentic/audio_llm_system.py index 14ae92c0..6647affa 100644 --- a/src/eva/assistant/agentic/audio_llm_system.py +++ b/src/eva/assistant/agentic/audio_llm_system.py @@ -91,12 +91,10 @@ async def process_query_with_audio(self, user_text: str) -> AsyncGenerator[str, yield response async def _execute_agent_with_audio(self, agent: AgentConfig) -> AsyncGenerator[str, None]: - """Build messages with audio on the last user message only, then run tool loop. + """Build messages with audio on all user turns, then run tool loop. - Only the current (last) user turn is sent as audio. Previous user turns - remain as text (transcriptions updated via the parallel transcription - pipeline). This keeps context manageable while giving the model the - actual audio for the current turn. + All user turns are sent as audio so the model has full conversational + context without relying on potentially inaccurate transcriptions. """ messages: list[dict[str, Any]] = [ {"role": "system", "content": self.system_prompt}, @@ -106,22 +104,16 @@ async def _execute_agent_with_audio(self, agent: AgentConfig) -> AsyncGenerator[ conversation_history = self.audit_log.get_conversation_messages(max_messages=30) history_dicts = [msg.to_dict() for msg in conversation_history] - # Replace only the LAST user message with audio (current turn) + # Replace ALL user messages with their corresponding audio if self._turn_audio_history: - # Find the last user message index - last_user_idx = None - for i in range(len(history_dicts) - 1, -1, -1): - if history_dicts[i].get("role") == "user": - last_user_idx = i - break - - if last_user_idx is not None: - # Use the most recent audio for the last user message - audio_bytes, sample_rate = self._turn_audio_history[-1] - history_dicts[last_user_idx] = self.alm_client.build_audio_user_message( - audio_bytes=audio_bytes, - source_sample_rate=sample_rate, - ) + user_indices = [i for i, msg in enumerate(history_dicts) if msg.get("role") == "user"] + for turn_idx, msg_idx in enumerate(user_indices): + if turn_idx < len(self._turn_audio_history): + audio_bytes, sample_rate = self._turn_audio_history[turn_idx] + history_dicts[msg_idx] = self.alm_client.build_audio_user_message( + audio_bytes=audio_bytes, + source_sample_rate=sample_rate, + ) messages.extend(history_dicts) diff --git a/src/eva/assistant/agentic/system.py b/src/eva/assistant/agentic/system.py index c86404fb..7d197703 100644 --- a/src/eva/assistant/agentic/system.py +++ b/src/eva/assistant/agentic/system.py @@ -9,6 +9,8 @@ from pathlib import Path from typing import Any +from pipecat.utils.text.simple_text_aggregator import SimpleTextAggregator + from eva.assistant.agentic.audit_log import ( AuditLog, ConversationMessage, @@ -44,6 +46,46 @@ def _clean_tool_name(name: str) -> str: return name +def _pair_orphaned_tool_calls(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Add neutral tool results for tool calls left unanswered by interruption or transfer paths.""" + repaired: list[dict[str, Any]] = [] + i = 0 + n = len(messages) + while i < n: + msg = messages[i] + repaired.append(msg) + tool_calls = msg.get("tool_calls") if msg.get("role") == "assistant" else None + if tool_calls: + call_ids = [ + cid for tc in tool_calls if isinstance(tc, dict) and isinstance((cid := tc.get("id")), str) and cid + ] + answered: set[str] = set() + j = i + 1 + while j < n and messages[j].get("role") == "tool": + repaired.append(messages[j]) + if isinstance(tool_call_id := messages[j].get("tool_call_id"), str): + answered.add(tool_call_id) + j += 1 + for cid in call_ids: + if cid not in answered: + logger.warning( + f"Pairing orphaned tool_call {cid} with a synthetic result (interrupted/transferred)" + ) + repaired.append( + { + "role": "tool", + "tool_call_id": cid, + "content": json.dumps( + {"status": "no_result", "note": "tool result unavailable (interrupted or transferred)"} + ), + } + ) + i = j + continue + i += 1 + return repaired + + class AgenticSystem: """Orchestrates the interaction between users and a single agent. @@ -62,6 +104,8 @@ def __init__( audit_log: AuditLog, llm_client: Any, # LLM client for model calls output_dir: Path | None = None, # Output directory for performance stats + pre_tool_speech: str = "off", + llm_streaming: bool = False, ): """Initialize the agentic system. @@ -72,6 +116,8 @@ def __init__( audit_log: Audit log for conversation tracking llm_client: Client for LLM calls output_dir: Optional output directory for saving performance stats + pre_tool_speech: Lead-in mode ('off'|'auto') + llm_streaming: Stream LLM output sentence-by-sentence """ self.agent = agent self.tool_handler = tool_handler @@ -79,6 +125,9 @@ def __init__( self.llm_client = llm_client self.output_dir = output_dir self.current_date_time = current_date_time + self.pre_tool_speech = pre_tool_speech + self.llm_streaming = llm_streaming + self._warned_responses_streaming_fallback = False self.prompt_manager = PromptManager() @@ -92,9 +141,18 @@ def __init__( agent_instructions=agent.instructions, datetime=self.current_date_time, ) + if self.pre_tool_speech == "auto": + self.system_prompt += "\n\n" + self.prompt_manager.get_prompt("agent.pre_tool_speech") # Build tools for the LLM self.tools = agent.build_tools_for_agent() + def _record_partial_streamed_output(self, chunks: list[str], reason: str) -> None: + partial = " ".join(chunk.strip() for chunk in chunks if chunk.strip()) + if not partial: + return + logger.info(f"Recording partial streamed assistant output after {reason}") + self.audit_log.append_assistant_output(partial) + async def process_query(self, query: str) -> AsyncGenerator[str, None]: """Process a user query and yield response messages. @@ -135,6 +193,7 @@ async def _execute_agent( # Add conversation history (includes current query since we already called append_user_input) conversation_history = self.audit_log.get_conversation_messages(max_messages=30) messages.extend(msg.to_dict() for msg in conversation_history) + messages = _pair_orphaned_tool_calls(messages) async for response in self._run_tool_loop(messages, agent): yield response @@ -159,15 +218,42 @@ async def _run_tool_loop( # Tool calling loop (no max iterations) while True: start_time = str(int(time.time() * 1000)) + streamed_chunks: list[str] = [] try: # Truncate data URIs for logging only (full audio stays in `messages`) messages_for_log = truncate_data_uris(messages) prompt_str = json.dumps(messages_for_log, indent=2, ensure_ascii=False) - response, llm_stats = await self.llm_client.complete( - messages, - tools=self.tools, - ) + content_streamed = False + use_responses_api = getattr(self.llm_client, "use_responses_api", False) + if self.llm_streaming and use_responses_api and not self._warned_responses_streaming_fallback: + logger.warning( + "llm_streaming is not supported for Responses API deployments; using non-streaming completion." + ) + self._warned_responses_streaming_fallback = True + + if self.llm_streaming and not use_responses_api: + response = None + llm_stats = {} + aggregator = SimpleTextAggregator() + async for kind, payload in self.llm_client.complete_stream(messages, tools=self.tools): + if kind == "delta": + async for agg in aggregator.aggregate(payload): + content_streamed = True + streamed_chunks.append(agg.text) + yield agg.text + else: + response, llm_stats = payload + remainder = await aggregator.flush() + if remainder and remainder.text.strip(): + content_streamed = True + streamed_chunks.append(remainder.text) + yield remainder.text + else: + response, llm_stats = await self.llm_client.complete( + messages, + tools=self.tools, + ) end_time = str(int(time.time() * 1000)) # Convert tool calls to dicts if present and extract content as string @@ -256,6 +342,7 @@ async def _run_tool_loop( except asyncio.CancelledError: # Pipeline is shutting down - log at debug and exit gracefully + self._record_partial_streamed_output(streamed_chunks, "cancellation") logger.debug("LLM call cancelled during pipeline shutdown") return # Don't yield error message, just exit except Exception as e: @@ -291,10 +378,13 @@ async def _run_tool_loop( retry_attempt=0, ) self.audit_log.append_llm_call(failed_llm_call, agent_name=agent.name) - yield GENERIC_ERROR + if streamed_chunks: + self._record_partial_streamed_output(streamed_chunks, "streaming failure") + else: + yield GENERIC_ERROR return - if response_content: + if response_content and not content_streamed: logger.info(f"💬 Assistant LLM response: {response_content}") yield response_content diff --git a/src/eva/assistant/base_server.py b/src/eva/assistant/base_server.py index 74645c18..0f8ee86e 100644 --- a/src/eva/assistant/base_server.py +++ b/src/eva/assistant/base_server.py @@ -23,6 +23,7 @@ from eva.utils.audio_utils import save_pcm_as_wav from eva.utils.culture import get_initial_message from eva.utils.logging import get_logger +from eva.utils.prompt_manager import PromptManager logger = get_logger(__name__) @@ -319,6 +320,19 @@ def _save_audio_deferred( if mixed_audio or user_audio or assistant_audio: logger.info(f"Saved audio files to {self.output_dir} ({len(mixed_audio)} bytes mixed)") + def _build_system_prompt(self) -> str: + """Build the system prompt for realtime/S2S assistant servers.""" + prompt_manager = PromptManager() + prompt = prompt_manager.get_prompt( + "realtime_agent.system_prompt", + agent_personality=self.agent.description, + agent_instructions=self.agent.instructions, + datetime=self.current_date_time, + ) + if self.pipeline_config.pre_tool_speech == "auto": + prompt += "\n\n" + prompt_manager.get_prompt("agent.pre_tool_speech") + return prompt + def _save_scenario_dbs(self) -> None: """Save initial and final scenario database states.""" try: diff --git a/src/eva/assistant/elevenlabs_server.py b/src/eva/assistant/elevenlabs_server.py index bb5c30bb..eb5b6493 100644 --- a/src/eva/assistant/elevenlabs_server.py +++ b/src/eva/assistant/elevenlabs_server.py @@ -45,7 +45,6 @@ from eva.models.agents import AgentConfig from eva.models.config import ModelConfig from eva.utils.logging import get_logger -from eva.utils.prompt_manager import PromptManager logger = get_logger(__name__) @@ -156,14 +155,7 @@ def __init__( self.s2s_params = s2s_params self._model = s2s_params.get("model", "elevenlabs") - # Build system prompt - prompt_manager = PromptManager() - self._system_prompt = prompt_manager.get_prompt( - "realtime_agent.system_prompt", - agent_personality=agent.description, - agent_instructions=agent.instructions, - datetime=self.current_date_time, - ) + self._system_prompt = self._build_system_prompt() # Build ElevenLabs client tools from agent config self._client_tools = _agent_tools_to_client_tools(agent, self.execute_tool) diff --git a/src/eva/assistant/gemini_live_server.py b/src/eva/assistant/gemini_live_server.py index 7e9ce5b6..b07fff28 100644 --- a/src/eva/assistant/gemini_live_server.py +++ b/src/eva/assistant/gemini_live_server.py @@ -40,7 +40,6 @@ from eva.models.agents import AgentConfig from eva.models.config import ModelConfig from eva.utils.logging import get_logger -from eva.utils.prompt_manager import PromptManager logger = get_logger(__name__) @@ -186,14 +185,7 @@ def __init__( self._language_code = s2s_params.get("language_code") or self.language self._api_key = s2s_params.get("api_key", "") - # Build system prompt (same pattern as pipecat realtime) - prompt_manager = PromptManager() - self._system_prompt = prompt_manager.get_prompt( - "realtime_agent.system_prompt", - agent_personality=agent.description, - agent_instructions=agent.instructions, - datetime=self.current_date_time, - ) + self._system_prompt = self._build_system_prompt() # Build Gemini tools self._gemini_tools = _agent_tools_to_gemini(agent) @@ -608,6 +600,7 @@ async def _process_gemini_events() -> None: id=fc.id, name=fc.name, response=result, + scheduling=types.FunctionResponseScheduling.WHEN_IDLE, ) ] ) diff --git a/src/eva/assistant/openai_realtime_server.py b/src/eva/assistant/openai_realtime_server.py index 73f96fdc..3d368900 100644 --- a/src/eva/assistant/openai_realtime_server.py +++ b/src/eva/assistant/openai_realtime_server.py @@ -28,7 +28,6 @@ ) from eva.assistant.base_server import AbstractAssistantServer from eva.utils.logging import get_logger -from eva.utils.prompt_manager import PromptManager logger = get_logger(__name__) @@ -84,13 +83,7 @@ def __init__(self, **kwargs: Any) -> None: self._audio_sample_rate = OPENAI_SAMPLE_RATE - prompt_manager = PromptManager() - self._system_prompt: str = prompt_manager.get_prompt( - "realtime_agent.system_prompt", - agent_personality=self.agent.description, - agent_instructions=self.agent.instructions, - datetime=self.current_date_time, - ) + self._system_prompt: str = self._build_system_prompt() self._realtime_tools: list[dict] = self._build_realtime_tools() @@ -218,6 +211,9 @@ def _build_session_config(self) -> dict[str, Any]: if reasoning_effort: session_config["reasoning"] = {"effort": reasoning_effort} + if self.pipeline_config.parallel_tool_calls is not None: + session_config["parallel_tool_calls"] = self.pipeline_config.parallel_tool_calls + return session_config def _create_client(self) -> AsyncOpenAI: diff --git a/src/eva/assistant/pipecat_server.py b/src/eva/assistant/pipecat_server.py index 58698825..f64f63d4 100644 --- a/src/eva/assistant/pipecat_server.py +++ b/src/eva/assistant/pipecat_server.py @@ -6,6 +6,7 @@ import asyncio import json +import time from pathlib import Path import uvicorn @@ -29,6 +30,7 @@ ) from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.serializers.twilio import TwilioFrameSerializer +from pipecat.services.cartesia.turns.stt import CartesiaTurnsSTTService from pipecat.transports.websocket.fastapi import ( FastAPIWebsocketParams, FastAPIWebsocketTransport, @@ -287,7 +289,14 @@ async def _realtime_tool_handler(params) -> None: language_code=self.language, ) # Create LLM client for agentic system (separate from Pipecat LLM service) - llm_client = LiteLLMClient(model=self.pipeline_config.llm) + llm_client = LiteLLMClient( + model=self.pipeline_config.llm, + parallel_tool_calls=self.pipeline_config.parallel_tool_calls, + ) + + # Cartesia ink-2 turn events are logged as diagnostics only. + if isinstance(stt, CartesiaTurnsSTTService): + self._register_ink2_diagnostics(stt) # Create context aggregator with user turn strategies messages = [] @@ -372,7 +381,6 @@ async def _realtime_tool_handler(params) -> None: input_transcription_processor = AudioTranscriptionProcessor( audio_collector=audio_llm_audio_collector, alm_client=alm_client, - sample_rate=SAMPLE_RATE, ) # Set callback to save user transcription to transcript.jsonl and update audit log @@ -407,6 +415,8 @@ async def on_user_transcription(text: str, timestamp: str, turn_id: int | None) audit_log=self.audit_log, llm_client=llm_client, output_dir=self.output_dir, + pre_tool_speech=self.pipeline_config.pre_tool_speech, + llm_streaming=self.pipeline_config.llm_streaming, ) agent_processor.on_assistant_response = lambda msg: self._save_transcript_message_from_turn( role="assistant", content=msg, timestamp=self._current_iso_timestamp() @@ -463,6 +473,7 @@ async def on_user_transcription(text: str, timestamp: str, turn_id: int | None) assistant_aggregator=assistant_aggregator, agent_processor=agent_processor, audio_llm_processor=audio_llm_processor, + audio_llm_audio_collector=audio_llm_audio_collector, input_transcription_processor=input_transcription_processor, ) @@ -488,6 +499,36 @@ async def on_user_transcription(text: str, timestamp: str, turn_id: int | None) logger.info("Client disconnected from assistant server") + def _register_ink2_diagnostics(self, stt: CartesiaTurnsSTTService) -> None: + """Log Cartesia ink-2 eager-end / resume events and the eager->final latency delta. + + Diagnostics only: ink-2's committed turn boundaries already drive aggregation through the + external turn strategies. The eager->final delta quantifies how much earlier the LLM could + have started if speculative execution were enabled (a possible future enhancement). + """ + # monotonic seconds at the last eager-end prediction (None if none pending) + eager: dict[str, float | None] = {"ts": None} + + @stt.event_handler("on_turn_eager_end") + async def _on_eager_end(_service, transcript: str) -> None: + eager["ts"] = time.monotonic() + logger.info(f"[ink-2] eager end-of-turn predicted: {transcript!r}") + + @stt.event_handler("on_turn_resume") + async def _on_resume(_service) -> None: + eager["ts"] = None + logger.info("[ink-2] turn resumed after eager end (user kept talking)") + + @stt.event_handler("on_turn_end") + async def _on_turn_end(_service, transcript: str) -> None: + ts = eager["ts"] + if ts is not None: + delta_ms = int((time.monotonic() - ts) * 1000) + logger.info(f"[ink-2] committed end-of-turn (eager->final +{delta_ms}ms): {transcript!r}") + else: + logger.info(f"[ink-2] committed end-of-turn (no eager prediction): {transcript!r}") + eager["ts"] = None + def _create_transport(self, websocket) -> FastAPIWebsocketTransport: """Create the WebSocket transport with Twilio frame serialization.""" return FastAPIWebsocketTransport( @@ -585,6 +626,7 @@ def _setup_event_handlers( assistant_aggregator, agent_processor, audio_llm_processor=None, + audio_llm_audio_collector=None, input_transcription_processor=None, ) -> None: """Setup event handlers for the pipeline.""" @@ -656,10 +698,13 @@ async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMes ) await agent_processor.process_complete_user_turn(message.content) elif self.pipeline_config.pipeline_type == PipelineType.AUDIO_LLM and audio_llm_processor: - # No STT → message.content is empty. - # Processing is triggered by LLMContextFrame flow through ParallelPipeline - # (AudioLLMUserAudioCollector pushes LLMContextFrame on UserStoppedSpeakingFrame) - pass + # No STT → message.content is empty. Finalize the turn now that the + # smart-turn-aware turn-stop strategy has fired. The collector kept + # appending audio across VAD blips up to this point; notify_turn_ended() + # increments the turn id and pushes LLMContextFrame to trigger the + # transcription and audio-LLM branches. + assert audio_llm_audio_collector is not None + await audio_llm_audio_collector.notify_turn_ended() elif self.non_instrumented_realtime_llm: # Non-instrumented realtime fallback (e.g. Ultravox) if message.content: diff --git a/src/eva/assistant/pipeline/agent_processor.py b/src/eva/assistant/pipeline/agent_processor.py index 11a83135..b801b5ce 100644 --- a/src/eva/assistant/pipeline/agent_processor.py +++ b/src/eva/assistant/pipeline/agent_processor.py @@ -62,6 +62,8 @@ def __init__( audit_log: AuditLog, llm_client, output_dir=None, + pre_tool_speech: str = "off", + llm_streaming: bool = False, **kwargs, ) -> None: """Initialize the agent processor. @@ -73,6 +75,8 @@ def __init__( audit_log: Audit log for conversation tracking llm_client: LLM client for generating responses output_dir: Optional output directory for saving performance stats + pre_tool_speech: Lead-in mode ('off'|'auto') + llm_streaming: Stream LLM output sentence-by-sentence **kwargs: Additional keyword arguments passed to FrameProcessor """ super().__init__(**kwargs) @@ -91,6 +95,8 @@ def __init__( audit_log=audit_log, llm_client=llm_client, output_dir=output_dir, + pre_tool_speech=pre_tool_speech, + llm_streaming=llm_streaming, ) # State tracking diff --git a/src/eva/assistant/pipeline/alm_vllm.py b/src/eva/assistant/pipeline/alm_vllm.py index 92891442..9e0b1762 100644 --- a/src/eva/assistant/pipeline/alm_vllm.py +++ b/src/eva/assistant/pipeline/alm_vllm.py @@ -21,6 +21,45 @@ logger = get_logger(__name__) +# vLLM / OpenAI-compatible decoding params that are safe to forward via extra_body. +# Unknown keys are still forwarded (vLLM evolves) but warned about, to catch typos. +_KNOWN_SAMPLING_PARAMS = frozenset( + { + "top_p", + "top_k", + "min_p", + "repetition_penalty", + "frequency_penalty", + "presence_penalty", + "length_penalty", + "seed", + "stop", + "stop_token_ids", + "min_tokens", + "ignore_eos", + "skip_special_tokens", + "spaces_between_special_tokens", + } +) + +# Keys the client controls itself — must not be set via sampling_params, or they would +# silently override the managed request (e.g. temperature/max_tokens land in extra_body +# and clobber the top-level args; model/messages/tools break the call entirely). +_RESERVED_SAMPLING_PARAMS = frozenset( + { + "model", + "messages", + "temperature", + "max_tokens", + "tools", + "tool_choice", + "stream", + "n", + "extra_body", + "chat_template_kwargs", + } +) + class ALMvLLMClient(BaseALMClient): """Client for self-hosted audio language model via vLLM's OpenAI-compatible HTTP API.""" @@ -39,6 +78,7 @@ def __init__( sample_width: int = DEFAULT_SAMPLE_WIDTH, language: str | None = None, enable_thinking: bool = False, + sampling_params: dict[str, Any] | None = None, ): super().__init__( model=model, @@ -53,6 +93,11 @@ def __init__( ) self._reasoning_token_fallback_warned = False self.enable_thinking = enable_thinking + # vLLM-specific decoding params (repetition_penalty, top_p, top_k, min_p, + # frequency_penalty, presence_penalty, ...). Validated here, then merged into + # extra_body of every complete() call; not applied to transcribe(), which is + # kept deterministic. + self.sampling_params: dict[str, Any] = self._validate_sampling_params(sampling_params) # Normalize base_url: ensure it ends with /v1 for the OpenAI client self.base_url = base_url.rstrip("/") if not self.base_url.endswith("/v1"): @@ -67,9 +112,44 @@ def __init__( logger.info( f"Initialized ALMvLLMClient: base_url={self.base_url}, model={self.model}, " f"sample_rate={self.sample_rate}, num_channels={self.num_channels}, " - f"sample_width={self.sample_width}" + f"sample_width={self.sample_width}, " + f"sampling_params={self.sampling_params}, enable_thinking={self.enable_thinking}" ) + @staticmethod + def _validate_sampling_params(sampling_params: dict[str, Any] | None) -> dict[str, Any]: + """Validate user-supplied vLLM sampling params before they reach extra_body. + + Rejects keys the client manages itself (these would silently clobber the + request — e.g. a ``temperature`` here overrides the top-level arg via + extra_body). Warns on keys outside the known vLLM set so typos are visible, + but still forwards them since vLLM's parameter surface evolves. + """ + if sampling_params is None: + return {} + if not isinstance(sampling_params, dict): + raise ValueError(f"sampling_params must be a dict, got {type(sampling_params).__name__}") + + reserved = _RESERVED_SAMPLING_PARAMS & sampling_params.keys() + if reserved: + raise ValueError( + f"sampling_params may not override client-managed keys {sorted(reserved)}. " + "Set temperature/max_tokens via their dedicated params; model/messages/tools " + "are controlled by the client." + ) + + non_string_keys = [k for k in sampling_params if not isinstance(k, str)] + if non_string_keys: + raise ValueError(f"sampling_params keys must be strings, got non-string keys: {non_string_keys}") + + unknown = sampling_params.keys() - _KNOWN_SAMPLING_PARAMS + if unknown: + logger.warning( + f"sampling_params contains keys not in the known vLLM set {sorted(unknown)}; " + "forwarding to vLLM as-is — check for typos." + ) + return dict(sampling_params) + def _audio_content_part(self, audio_b64: str) -> dict[str, Any]: return { "type": "audio_url", @@ -89,16 +169,20 @@ async def complete( When tool_calls are present, returns the full message object. Otherwise returns the content string. """ + extra_body: dict[str, Any] = { + "chat_template_kwargs": {"enable_thinking": self.enable_thinking}, + } + # Merge vLLM sampling params (repetition_penalty, top_p, etc.). These ride + # alongside chat_template_kwargs; vLLM accepts non-OpenAI keys at the top + # level of extra_body. + extra_body.update(self.sampling_params) + kwargs: dict[str, Any] = { "model": self.model, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens, - "extra_body": { - "chat_template_kwargs": { - "enable_thinking": self.enable_thinking, - } - }, + "extra_body": extra_body, } if tools: kwargs["tools"] = tools diff --git a/src/eva/assistant/pipeline/audio_llm_processor.py b/src/eva/assistant/pipeline/audio_llm_processor.py index e7b23cb3..2df9efa4 100644 --- a/src/eva/assistant/pipeline/audio_llm_processor.py +++ b/src/eva/assistant/pipeline/audio_llm_processor.py @@ -56,12 +56,22 @@ class AudioLLMUserAudioCollector(FrameProcessor): """Buffers raw audio frames during user speech for the audio-LLM pipeline. - Collects audio and pushes LLMContextFrame when user stops speaking, which - triggers the parallel pipeline (transcription + audio-LLM processing). - - Uses a ring buffer of pre-VAD audio so that the beginning of the user's - speech—before VAD fires UserStartedSpeakingFrame—is not lost. This mirrors - the S2S UserAudioCollector pattern. + Audio capture is driven by raw VAD events (UserStartedSpeakingFrame / + UserStoppedSpeakingFrame), but turn finalization is driven by the + smart-turn-aware user aggregator: pipecat_server calls ``notify_turn_ended()`` + from the ``on_user_turn_stopped`` event (after the turn-stop strategy is + satisfied), which increments the turn id and pushes an LLMContextFrame to + trigger both branches of the parallel pipeline (transcription + audio-LLM). + + Because finalization is deferred to the smart-turn verdict, a single turn may + span multiple VAD start/stop blips (e.g. natural pauses while spelling out a + code). The collector keeps appending audio across those blips instead of + fragmenting the buffer, and only starts a fresh buffer once the previous turn + has been finalized. + + A ring buffer of pre-VAD audio captures the start of speech that occurs before + VAD fires UserStartedSpeakingFrame. This mirrors the S2S UserAudioCollector + pattern. All frames pass through unchanged. """ @@ -82,45 +92,82 @@ def __init__( self._audio_buffer = bytearray() self._pre_speech_buffer: list[bytes] = [] self._user_speaking = False - self._current_turn_id = 0 # Incremented on each user turn + self._current_turn_id = 0 # Incremented on each finalized turn + # True between turns. Set False when a new turn starts and back to True by + # notify_turn_ended(), so audio is appended continuously across VAD start/stop + # blips within a single smart-turn-governed turn instead of being fragmented. + self._turn_finalized = True # Pre-speech buffer size (captures audio before VAD fires to avoid cutting off speech) self._pre_speech_secs = pre_speech_secs or self.DEFAULT_PRE_SPEECH_SECS + # Actual sample rate observed on InputAudioRawFrames. The buffer holds raw PCM + # at whatever rate the transport delivered; falls back to PIPELINE_SAMPLE_RATE + # until the first audio frame arrives. + self._frame_sample_rate: int = PIPELINE_SAMPLE_RATE async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: await super().process_frame(frame, direction) if isinstance(frame, UserStartedSpeakingFrame): self._user_speaking = True - # Prepend the pre-speech ring buffer so we don't lose the start of speech pre_speech_bytes = b"".join(self._pre_speech_buffer) - pre_speech_duration_ms = len(pre_speech_bytes) / (PIPELINE_SAMPLE_RATE * 2) * 1000 - logger.debug( - f"Prepending {len(pre_speech_bytes)} bytes ({pre_speech_duration_ms:.0f}ms) of pre-speech audio" - ) - self._audio_buffer = bytearray(pre_speech_bytes) + pre_speech_duration_ms = len(pre_speech_bytes) / (self._frame_sample_rate * 2) * 1000 + if self._turn_finalized: + # Fresh turn: pre-speech becomes the start of a new buffer. + logger.debug( + f"New turn — prepending {len(pre_speech_bytes)} bytes " + f"({pre_speech_duration_ms:.0f}ms) of pre-speech audio" + ) + self._audio_buffer = bytearray(pre_speech_bytes) + self._turn_finalized = False + else: + # Mid-turn VAD blip (e.g. a pause between spelled-out characters): + # append pre-speech so the captured utterance stays continuous. + logger.debug( + f"Resuming turn — appending {len(pre_speech_bytes)} bytes " + f"({pre_speech_duration_ms:.0f}ms) of pre-speech audio" + ) + self._audio_buffer.extend(pre_speech_bytes) self._pre_speech_buffer.clear() elif isinstance(frame, UserStoppedSpeakingFrame): + # Stop buffering active speech, but do NOT finalize the turn here — the + # smart-turn-aware on_user_turn_stopped event drives finalization via + # notify_turn_ended(). This lets a turn span multiple VAD segments. self._user_speaking = False - # Increment turn ID BEFORE pushing frame so both parallel branches see the same ID - self._current_turn_id += 1 - # Push LLMContextFrame to trigger parallel pipeline - await self._user_context_aggregator.push_frame(LLMContextFrame(context=self._context)) + logger.debug( + f"VAD stop — pausing capture (buffer={len(self._audio_buffer)} bytes, " + f"turn_id={self._current_turn_id})" + ) elif isinstance(frame, InputAudioRawFrame): + self._frame_sample_rate = frame.sample_rate if self._user_speaking: self._audio_buffer.extend(frame.audio) else: # Ring buffer: keep a rolling window of pre-speech audio self._pre_speech_buffer.append(frame.audio) # 16-bit mono → 2 bytes per sample - max_bytes = int(self._pre_speech_secs * PIPELINE_SAMPLE_RATE * 2) + max_bytes = int(self._pre_speech_secs * self._frame_sample_rate * 2) total = sum(len(chunk) for chunk in self._pre_speech_buffer) while total > max_bytes and self._pre_speech_buffer: total -= len(self._pre_speech_buffer.pop(0)) await self.push_frame(frame, direction) + async def notify_turn_ended(self) -> None: + """Finalize the current turn and trigger downstream processing. + + Called from pipecat_server's ``on_user_turn_stopped`` handler once the + smart-turn-aware turn-stop strategy is satisfied. Increments the turn id + (so the transcription branch and the audio-LLM branch observe a consistent + boundary) and pushes LLMContextFrame to trigger both branches. Setting + ``_turn_finalized`` makes the next UserStartedSpeakingFrame begin a fresh + buffer instead of appending to this turn. + """ + self._current_turn_id += 1 + self._turn_finalized = True + await self._user_context_aggregator.push_frame(LLMContextFrame(context=self._context)) + def get_buffered_audio(self) -> bytes: """Get the buffered audio and clear the buffer.""" audio = bytes(self._audio_buffer) @@ -143,6 +190,15 @@ def current_turn_id(self) -> int: """Get the current turn ID for associating transcriptions with entries.""" return self._current_turn_id + @property + def frame_sample_rate(self) -> int: + """Sample rate observed on the most recent InputAudioRawFrame. + + The audio buffer holds raw PCM at whatever rate the transport delivered; + downstream consumers should use this rather than assuming PIPELINE_SAMPLE_RATE. + """ + return self._frame_sample_rate + class AudioLLMProcessor(FrameProcessor): """Processes complete user turns using the audio-LLM model. @@ -257,7 +313,10 @@ async def process_complete_user_turn(self, text_from_aggregator: str) -> None: turn_id = self.audio_collector.current_turn_id self.audit_log.append_user_input(self._USER_PLACEHOLDER, turn_id=turn_id) - self._current_query_task = asyncio.create_task(self._process_audio_turn(audio_bytes)) + source_sample_rate = self.audio_collector.frame_sample_rate + self._current_query_task = asyncio.create_task( + self._process_audio_turn(audio_bytes, source_sample_rate) + ) try: await self._current_query_task except asyncio.CancelledError: @@ -268,11 +327,11 @@ async def process_complete_user_turn(self, text_from_aggregator: str) -> None: # Placeholder used in audit_log / transcript since no real transcription is available _USER_PLACEHOLDER = "[user audio]" - async def _process_audio_turn(self, audio_bytes: bytes) -> None: + async def _process_audio_turn(self, audio_bytes: bytes, source_sample_rate: int) -> None: """Process a user turn with audio data.""" try: # Send audio to the agentic system and process - self.agentic_system.set_turn_audio(audio_bytes, PIPELINE_SAMPLE_RATE) + self.agentic_system.set_turn_audio(audio_bytes, source_sample_rate) async for response in self.agentic_system.process_query_with_audio(self._USER_PLACEHOLDER): if self._interrupted.is_set(): @@ -392,14 +451,12 @@ def __init__( audio_collector: AudioLLMUserAudioCollector, alm_client: BaseALMClient, system_prompt: str | None = None, - sample_rate: int = PIPELINE_SAMPLE_RATE, **kwargs, ): super().__init__(**kwargs) self._audio_collector = audio_collector self._alm_client = alm_client self._system_prompt = system_prompt or alm_client.default_transcription_prompt - self._sample_rate = sample_rate # Callback for when transcription is ready (set by pipecat_server.py) self.on_transcription: Any | None = None @@ -428,13 +485,16 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: # Capture turn_id and audio at the moment we receive the frame turn_id = self._audio_collector.current_turn_id - # Capture audio NOW before it gets overwritten by the next turn + # Capture audio + sample rate NOW before they get overwritten by the next turn audio_data = self._audio_collector.peek_buffered_audio() + source_sample_rate = self._audio_collector.frame_sample_rate logger.info(f"transcribe (turn_id={turn_id})") timestamp = time_now_iso8601() # Run transcription as background task so it completes even if interrupted - task = asyncio.create_task(self._transcribe_audio(audio_data, timestamp, turn_id=turn_id)) + task = asyncio.create_task( + self._transcribe_audio(audio_data, timestamp, source_sample_rate, turn_id=turn_id) + ) self._transcription_tasks.append(task) # Clean up completed tasks self._transcription_tasks = [t for t in self._transcription_tasks if not t.done()] @@ -453,9 +513,16 @@ async def transcribe(self, timestamp: str, turn_id: int | None = None) -> str | The transcription text, or None if transcription failed or audio was empty. """ audio_data = self._audio_collector.peek_buffered_audio() - return await self._transcribe_audio(audio_data, timestamp, turn_id) + source_sample_rate = self._audio_collector.frame_sample_rate + return await self._transcribe_audio(audio_data, timestamp, source_sample_rate, turn_id) - async def _transcribe_audio(self, audio_data: bytes, timestamp: str, turn_id: int | None = None) -> str | None: + async def _transcribe_audio( + self, + audio_data: bytes, + timestamp: str, + source_sample_rate: int, + turn_id: int | None = None, + ) -> str | None: """Transcribe pre-captured audio data using chat completions. This method takes audio data directly instead of reading from the collector, @@ -478,7 +545,7 @@ async def _transcribe_audio(self, audio_data: bytes, timestamp: str, turn_id: in start_time = time.time() text = await self._alm_client.transcribe( audio_bytes=audio_data, - source_sample_rate=self._sample_rate, + source_sample_rate=source_sample_rate, system_prompt=self._system_prompt, ) elapsed = time.time() - start_time diff --git a/src/eva/assistant/pipeline/services.py b/src/eva/assistant/pipeline/services.py index 798f7101..116e9b34 100644 --- a/src/eva/assistant/pipeline/services.py +++ b/src/eva/assistant/pipeline/services.py @@ -19,6 +19,7 @@ from pipecat.services.assemblyai.stt import AssemblyAISTTService from pipecat.services.cartesia.stt import CartesiaSTTService from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.cartesia.turns.stt import CartesiaTurnsSTTService from pipecat.services.deepgram.flux.stt import DeepgramFluxSTTService, DeepgramFluxSTTSettings from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.tts import DeepgramTTSService @@ -77,7 +78,7 @@ logger = get_logger(__name__) -# Default sample rate for audio +# Default sample rate for audio (TTS output rate). SAMPLE_RATE = 24000 @@ -117,7 +118,7 @@ def create_stt_service( Based on create_stt_service() from chatbot.py. Args: - model: STT model identifier (deepgram, deepgram-flux, openai, assemblyai, cartesia, nvidia) + model: STT model identifier (deepgram, deepgram-flux, openai, assemblyai, cartesia, cartesia-multilingual, nvidia) params: Model-specific parameters (may include 'alias' key which is ignored here) language_code: Language code for transcription @@ -153,10 +154,21 @@ def create_stt_service( ) elif model_lower == "cartesia": - logger.info(f"Using Cartesia STT: {params['model']}") + # ink-2 provides its own turn boundaries; ModelConfig selects external endpointing. + model_name = params["model"] + logger.info(f"Using Cartesia STT: {model_name}") + return CartesiaTurnsSTTService( + api_key=api_key, + sample_rate=params.get("sample_rate", 16000), + should_interrupt=params.get("should_interrupt", True), + settings=CartesiaTurnsSTTService.Settings(model=model_name), + ) + + elif model_lower == "cartesia-multilingual": + logger.info(f"Using Cartesia multilingual STT: {params['model']}") return CartesiaSTTService( api_key=api_key, - sample_rate=SAMPLE_RATE, + sample_rate=16000, settings=CartesiaSTTService.Settings( model=params["model"], language=_to_language_enum(language_code), @@ -281,7 +293,7 @@ def create_stt_service( else: raise ValueError( - f"Unknown STT model: {model}. Available: assemblyai, cartesia, cohere, deepgram, deepgram-flux, elevenlabs, nvidia, nvidia-baseten, openai, xai" + f"Unknown STT model: {model}. Available: assemblyai, cartesia, cartesia-multilingual, cohere, deepgram, deepgram-flux, elevenlabs, nvidia, nvidia-baseten, openai, xai" ) @@ -770,6 +782,7 @@ def create_audio_llm_client( sample_width=params.get("sample_width", 2), language=language, enable_thinking=params.get("enable_thinking", False), + sampling_params=params.get("sampling_params"), ) logger.info(f"Using {model} vLLM audio-LLM: {base_url}") return client diff --git a/src/eva/assistant/services/llm.py b/src/eva/assistant/services/llm.py index 89265b46..f35fb596 100644 --- a/src/eva/assistant/services/llm.py +++ b/src/eva/assistant/services/llm.py @@ -27,13 +27,15 @@ class LiteLLMClient: ``litellm_params.model`` in the ``EVA_MODEL_LIST`` deployment config. """ - def __init__(self, model: str): + def __init__(self, model: str, parallel_tool_calls: bool | None = None): """Initialize LiteLLM client. Args: model: Model name matching a model_name in EVA_MODEL_LIST (e.g., 'gpt-5.2', 'gemini-3-pro') + parallel_tool_calls: Optional provider pass-through for tool calls. """ self.model = model + self.parallel_tool_calls = parallel_tool_calls self.use_responses_api = self._lookup_use_responses_api_from_router() self._reasoning_token_fallback_warned = False @@ -84,6 +86,8 @@ async def complete( if tools: kwargs["tools"] = tools kwargs["tool_choice"] = "auto" + if self.parallel_tool_calls is not None: + kwargs["parallel_tool_calls"] = self.parallel_tool_calls last_exception = None for attempt in range(max_retries + 1): @@ -275,6 +279,88 @@ def _get_router_litellm_params(self) -> dict[str, Any]: return deployment.get("litellm_params", {}) return {} + async def complete_stream( + self, + messages: list[dict[str, Any]], + tools: list[dict] | None = None, + max_retries: int = 5, + initial_delay: float = 1.0, + ): + """Yield text deltas, then the assembled final message and stats.""" + kwargs: dict[str, Any] = { + "model": self.model, + "messages": messages, + "stream": True, + "stream_options": {"include_usage": True}, + } + if tools: + kwargs["tools"] = tools + kwargs["tool_choice"] = "auto" + if self.parallel_tool_calls is not None: + kwargs["parallel_tool_calls"] = self.parallel_tool_calls + + last_exception = None + for attempt in range(max_retries + 1): + chunks: list[Any] = [] + first_token = False + try: + start_time = time.time() + stream = await router.get().acompletion(**kwargs) + async for chunk in stream: + chunks.append(chunk) + choices = getattr(chunk, "choices", None) or [] + if choices: + delta = getattr(choices[0], "delta", None) + text = getattr(delta, "content", None) if delta else None + if text: + first_token = True + yield ("delta", text) + elapsed_time = time.time() - start_time + + full = litellm.stream_chunk_builder(chunks, messages=messages) + message = full.choices[0].message + usage = getattr(full, "usage", None) + reasoning_content = getattr(message, "reasoning_content", None) + thinking_blocks = ( + getattr(message, "thinking_blocks", None) if hasattr(message, "thinking_blocks") else None + ) + hidden_params = getattr(full, "_hidden_params", {}) or {} + # Reasoning tokens arrive in the final usage chunk. + reasoning_tokens = 0 + if usage and hasattr(usage, "completion_tokens_details"): + details = usage.completion_tokens_details + if details and hasattr(details, "reasoning_tokens"): + reasoning_tokens = getattr(details, "reasoning_tokens", 0) or 0 + stats = { + "prompt_tokens": getattr(usage, "prompt_tokens", 0) if usage else 0, + "completion_tokens": getattr(usage, "completion_tokens", 0) if usage else 0, + "reasoning_tokens": reasoning_tokens, + "finish_reason": getattr(full.choices[0], "finish_reason", "unknown"), + "model": getattr(full, "model", self.model), + "cost": hidden_params.get("response_cost"), + "cost_source": "litellm", + "latency": round(elapsed_time, 3), + "reasoning": reasoning_content, + "reasoning_content": reasoning_content, + "thinking_blocks": thinking_blocks, + "responses_output_items": None, + } + yield ("final", (message, stats)) + return + except Exception as e: + last_exception = e + if is_retryable_error(e) and attempt < max_retries and not first_token: + delay = initial_delay * (2**attempt) + logger.warning( + f"Retryable streaming error on attempt {attempt + 1}/{max_retries + 1}: {e}. " + f"Retrying in {delay:.1f}s..." + ) + await asyncio.sleep(delay) + continue + logger.exception(f"LiteLLM streaming completion failed: {e}") + raise + raise last_exception + async def _complete_via_responses_api( self, messages: list[dict[str, Any]], diff --git a/src/eva/metrics/processor.py b/src/eva/metrics/processor.py index c9ba29e0..c404af01 100644 --- a/src/eva/metrics/processor.py +++ b/src/eva/metrics/processor.py @@ -883,6 +883,52 @@ def _load_pipecat_logs(pipecat_logs_path: str) -> list[dict]: has_tts_text = any(entry.get("type") == "tts_text" for entry in raw_pipecat) if has_tts_text: raw_pipecat = [entry for entry in raw_pipecat if entry.get("type") != "llm_response"] + # Pipecat emits full-phrase batch-preview tts_text events (multiple sharing + # the same timestamp) alongside per-word streaming tokens (unique timestamps). + # The batch previews duplicate content that appears again word-by-word, which + # causes truncate_to_spoken to fail: the joined segment contains duplicate + # phrases that break substring matching. Remove batch-preview duplicates by + # dropping multi-word tts_text events whose timestamp appears more than once. + tts_ts_counts: Counter = Counter(e["timestamp"] for e in raw_pipecat if e.get("type") == "tts_text") + raw_pipecat = [ + e + for e in raw_pipecat + if not ( + e.get("type") == "tts_text" + and tts_ts_counts[e["timestamp"]] > 1 + and " " in e.get("data", {}).get("frame", "") + ) + ] + # Second pass: drop any multi-word tts_text entry that is an exact preview of + # the immediately following single-word token stream. Catches batch-preview + # phrases emitted ~1ms before their per-word stream (different timestamp from + # the per-word tokens, so the same-timestamp filter above misses them). Without + # this, the preview phrase and its word-by-word reconstruction both survive and + # get joined into one aggregated segment, creating a duplicate that breaks + # truncate_to_spoken substring matching for any audit text that follows it. + deduped: list[dict] = [] + i = 0 + while i < len(raw_pipecat): + e = raw_pipecat[i] + if e.get("type") == "tts_text" and " " in e.get("data", {}).get("frame", ""): + phrase_words = e["data"]["frame"].split() + nw = len(phrase_words) + j, following_tts = i + 1, [] + while j < len(raw_pipecat) and len(following_tts) < nw: + if raw_pipecat[j].get("type") != "tts_text": + break # stop at any non-tts boundary (turn_start/end) + following_tts.append(raw_pipecat[j]) + j += 1 + if ( + len(following_tts) == nw + and all(" " not in fw.get("data", {}).get("frame", "") for fw in following_tts) + and " ".join(fw["data"]["frame"] for fw in following_tts) == e["data"]["frame"] + ): + i += 1 # skip: this multi-word entry is a preview of the following words + continue + deduped.append(e) + i += 1 + raw_pipecat = deduped grouped_pipecat = aggregate_pipecat_logs_by_type(raw_pipecat) for entry in grouped_pipecat: diff --git a/src/eva/models/config.py b/src/eva/models/config.py index 561f54c6..c9de3f9b 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -119,6 +119,10 @@ class ModelConfig(BaseModel): } _LEGACY_DROP: ClassVar[set[str]] = {"realtime_model", "realtime_model_params"} + # STT models that perform their own (server-side) semantic endpointing. They drive turn + # boundaries themselves, so they must run with the 'external' turn strategies and no local VAD. + _SELF_ENDPOINTING_STT: ClassVar[set[str]] = {"cartesia"} + # ── Mode selectors (exactly one group must be set for a real run) ── llm: str | None = Field( None, @@ -183,6 +187,20 @@ class ModelConfig(BaseModel): ), ) + # CASCADE-only latency controls. + pre_tool_speech: str = Field( + "off", + description="Prompt a model-generated lead-in before tool calls: 'off' or 'auto'.", + ) + llm_streaming: bool = Field( + False, + description="Stream Chat Completions output to TTS sentence-by-sentence.", + ) + parallel_tool_calls: bool | None = Field( + None, + description="Forward parallel_tool_calls when tools are present; None leaves provider defaults.", + ) + @property def pipeline_type(self) -> "PipelineType": """Detected pipeline mode based on which selector is set.""" @@ -232,6 +250,47 @@ def _migrate_legacy_fields(cls, data: Any) -> Any: data.pop(key, None) return data + @model_validator(mode="after") + def _autowire_self_endpointing_stt(self) -> "ModelConfig": + """Auto-configure turn-taking for self-endpointing STT models (e.g. Cartesia ink-2). + + These models drive turn boundaries themselves (the service pushes its own speech + start/stop and transcription frames), so they require the 'external' turn strategies + with local VAD disabled. Forcing it here lets a bare ``EVA_MODEL__STT=cartesia`` + "just work", and the forced values are reflected in the persisted config.json / run_id. + A conflicting user-provided value is overridden with a WARNING; an untouched default is + logged at INFO. Idempotent: a value already at the target is left untouched (clean reload). + """ + if (self.stt or "").lower() not in self._SELF_ENDPOINTING_STT: + return self + + forced = {"turn_start_strategy": "external", "turn_stop_strategy": "external", "vad": "none"} + for field, target in forced.items(): + if getattr(self, field) == target: + continue + if field in self.model_fields_set: + logger.warning( + f"STT '{self.stt}' performs its own endpointing; overriding " + f"{field}='{getattr(self, field)}' -> '{target}'." + ) + else: + logger.info(f"STT '{self.stt}': auto-setting {field}='{target}' (self-endpointing).") + setattr(self, field, target) + return self + + @model_validator(mode="after") + def _validate_latency_optimizations(self) -> "ModelConfig": + allowed = {"off", "auto"} + if self.pre_tool_speech not in allowed: + raise ValueError(f"pre_tool_speech must be one of {sorted(allowed)}, got '{self.pre_tool_speech}'") + any_set = self.pre_tool_speech != "off" or self.llm_streaming or self.parallel_tool_calls is not None + if any_set and self.pipeline_type != PipelineType.CASCADE: + logger.warning( + "Cascade LLM flags (pre_tool_speech / llm_streaming / parallel_tool_calls) apply only " + f"to the CASCADE pipeline; they will be ignored for pipeline_type={self.pipeline_type}." + ) + return self + class PipelineType(StrEnum): """Type of voice pipeline.""" diff --git a/src/eva/utils/log_processing.py b/src/eva/utils/log_processing.py index f82c254e..23235cc1 100644 --- a/src/eva/utils/log_processing.py +++ b/src/eva/utils/log_processing.py @@ -93,6 +93,10 @@ def truncate_to_spoken(audit_text: str, pipecat_segments: list[str]) -> str | No if any(norm_audit in seg for seg in norm_segments): return audit_text + # Streaming can split one assistant turn across several TTS segments. + if norm_audit in "".join(norm_segments): + return audit_text + # Find longest word-prefix of the audit text that starts any pipecat # segment. Matching at the START of a segment (not at an arbitrary # position) prevents spurious short matches (e.g. "Just to" matching diff --git a/tests/unit/assistant/test_agentic_system.py b/tests/unit/assistant/test_agentic_system.py index fd5df2f6..3f9ae968 100644 --- a/tests/unit/assistant/test_agentic_system.py +++ b/tests/unit/assistant/test_agentic_system.py @@ -7,7 +7,7 @@ import pytest from eva.assistant.agentic.audit_log import AuditLog -from eva.assistant.agentic.system import GENERIC_ERROR, AgenticSystem +from eva.assistant.agentic.system import GENERIC_ERROR, AgenticSystem, _pair_orphaned_tool_calls from eva.models.agents import AgentConfig, AgentTool, AgentToolParameter @@ -487,3 +487,75 @@ async def test_cancellation_error_yields_nothing(self): assert _conv_to_dicts(audit_log.get_conversation_messages()) == [ {"role": "user", "content": "Hello"}, ] + + +class TestPairOrphanedToolCalls: + @staticmethod + def _asst(tool_call_ids, content=""): + return { + "role": "assistant", + "content": content, + "tool_calls": [ + {"id": cid, "type": "function", "function": {"name": "get_x", "arguments": "{}"}} + for cid in tool_call_ids + ], + } + + @staticmethod + def _tool(cid, content="{}"): + return {"role": "tool", "tool_call_id": cid, "content": content} + + def test_transfer_dangling_call_gets_synthetic_result(self): + messages = [ + {"role": "user", "content": "human please"}, + self._asst(["call_t"]), + {"role": "assistant", "content": "Transferring you to a live agent. Please wait."}, + {"role": "user", "content": "no, hanging up"}, + ] + out = _pair_orphaned_tool_calls(messages) + + assert out[1] == messages[1] + assert out[2]["role"] == "tool" and out[2]["tool_call_id"] == "call_t" + assert json.loads(out[2]["content"])["status"] == "no_result" + assert out[3]["content"] == "Transferring you to a live agent. Please wait." + assert out[4] == messages[3] + + called = [tc["id"] for m in out if m.get("role") == "assistant" for tc in (m.get("tool_calls") or [])] + answered = [m["tool_call_id"] for m in out if m.get("role") == "tool"] + assert set(called) <= set(answered) + + def test_properly_answered_call_is_unchanged(self): + messages = [ + {"role": "user", "content": "hi"}, + self._asst(["call_1"]), + self._tool("call_1", '{"ok": true}'), + {"role": "assistant", "content": "Done."}, + ] + assert _pair_orphaned_tool_calls(messages) == messages + + def test_partial_parallel_calls_only_unanswered_get_synthetic(self): + messages = [ + {"role": "user", "content": "hi"}, + self._asst(["call_1", "call_2"]), + self._tool("call_1"), + ] + out = _pair_orphaned_tool_calls(messages) + answered = [m["tool_call_id"] for m in out if m.get("role") == "tool"] + assert answered == ["call_1", "call_2"] + + def test_malformed_tool_result_id_does_not_answer_call(self): + messages = [ + {"role": "user", "content": "hi"}, + self._asst(["call_1"]), + {"role": "tool", "tool_call_id": None, "content": "{}"}, + ] + out = _pair_orphaned_tool_calls(messages) + answered = [m.get("tool_call_id") for m in out if m.get("role") == "tool"] + assert answered == [None, "call_1"] + + def test_no_tool_calls_is_identity(self): + messages = [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "hello"}, + ] + assert _pair_orphaned_tool_calls(messages) == messages diff --git a/tests/unit/assistant/test_latency_optimizations.py b/tests/unit/assistant/test_latency_optimizations.py new file mode 100644 index 00000000..7e9c7423 --- /dev/null +++ b/tests/unit/assistant/test_latency_optimizations.py @@ -0,0 +1,270 @@ +"""Unit tests for CASCADE latency optimizations.""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from eva.assistant.agentic.audit_log import AuditLog +from eva.assistant.agentic.system import GENERIC_ERROR, AgenticSystem +from eva.models.agents import AgentConfig, AgentTool, AgentToolParameter + +_STATS = { + "model": "test-model", + "prompt_tokens": 1, + "completion_tokens": 1, + "reasoning_tokens": 0, + "finish_reason": "stop", + "cost": 0.0, + "cost_source": "litellm", + "latency": 0.1, + "reasoning": None, + "reasoning_content": None, + "thinking_blocks": None, + "responses_output_items": None, +} + + +def _agent(tools=None): + return AgentConfig( + id="test-agent", + name="Test Agent", + description="A test agent", + role="You are a test agent.", + instructions="Help the user.", + tools=tools or [], + tool_module_path="eva.assistant.tools.test_tools", + ) + + +def _tool(name="get_reservation"): + return AgentTool( + id="t1", + name=name, + description=f"Tool: {name}", + required_parameters=[AgentToolParameter(name="confirmation_number", type="string", description="code")], + ) + + +def _msg(content, tool_calls=None): + return SimpleNamespace(content=content, tool_calls=tool_calls, model="test-model") + + +def _tc(call_id, name, arguments): + tc = SimpleNamespace(id=call_id, type="function", function=SimpleNamespace(name=name, arguments=arguments)) + tc.model_dump = lambda exclude_none=False: { + "id": call_id, + "type": "function", + "function": {"name": name, "arguments": arguments}, + } + return tc + + +class _StreamLLM: + use_responses_api = False + + def __init__(self, turns): + self._turns = list(turns) + self._i = 0 + + async def complete_stream(self, messages, tools=None): + deltas, final = self._turns[self._i] + self._i += 1 + for d in deltas: + yield ("delta", d) + yield ("final", (final, dict(_STATS))) + + +class _CancellingStreamLLM: + use_responses_api = False + + def __init__(self): + self.calls = 0 + + async def complete_stream(self, messages, tools=None): + self.calls += 1 + yield ("delta", "Let me check. ") + yield ("delta", "I") + raise asyncio.CancelledError() + + +class _FailingStreamLLM: + use_responses_api = False + + def __init__(self): + self.calls = 0 + + async def complete_stream(self, messages, tools=None): + self.calls += 1 + yield ("delta", "Let me check. ") + yield ("delta", "I") + raise RuntimeError("stream dropped") + + +async def _collect(system, query): + out = [] + async for r in system.process_query(query): + out.append(r) + return out + + +class TestPreToolSpeechDirective: + @pytest.mark.parametrize("mode,present", [("off", False), ("auto", True)]) + def test_directive_injection(self, mode, present): + system = AgenticSystem( + current_date_time="2026-02-25 10:00:00", + agent=_agent(), + tool_handler=MagicMock(), + audit_log=AuditLog(), + llm_client=MagicMock(), + pre_tool_speech=mode, + ) + assert ("Responsiveness" in system.system_prompt) is present + + +class TestSilentToolCallNoFabricatedSpeech: + @pytest.mark.asyncio + @pytest.mark.parametrize("mode", ["off", "auto"]) + async def test_silent_tool_call_speaks_only_final(self, mode): + llm = MagicMock() + llm.use_responses_api = False + llm.complete = AsyncMock( + side_effect=[ + (_msg("", tool_calls=[_tc("c1", "get_reservation", "{}")]), dict(_STATS)), + (_msg("All set."), dict(_STATS)), + ] + ) + tool_handler = MagicMock() + tool_handler.execute = AsyncMock(return_value={"status": "success"}) + system = AgenticSystem("x", _agent([_tool()]), tool_handler, AuditLog(), llm, pre_tool_speech=mode) + assert await _collect(system, "look up A") == ["All set."] + + @pytest.mark.asyncio + async def test_model_lead_in_is_spoken(self): + llm = MagicMock() + llm.use_responses_api = False + llm.complete = AsyncMock( + side_effect=[ + (_msg("Sure, one sec.", tool_calls=[_tc("c1", "get_reservation", "{}")]), dict(_STATS)), + (_msg("All set."), dict(_STATS)), + ] + ) + tool_handler = MagicMock() + tool_handler.execute = AsyncMock(return_value={"status": "success"}) + system = AgenticSystem("x", _agent([_tool()]), tool_handler, AuditLog(), llm, pre_tool_speech="auto") + assert await _collect(system, "look up A") == ["Sure, one sec.", "All set."] + + +class TestLLMStreaming: + @pytest.mark.asyncio + async def test_streams_sentences_without_double_speak(self): + turns = [(["Sure, Mr. ", "Smith. The fee ", "is 3.5 dollars."], _msg("Sure, Mr. Smith. The fee is 3.5 dollars."))] + system = AgenticSystem("x", _agent(), MagicMock(), AuditLog(), _StreamLLM(turns), llm_streaming=True) + assert await _collect(system, "hi") == ["Sure, Mr. Smith.", "The fee is 3.5 dollars."] + + @pytest.mark.asyncio + async def test_streaming_silent_tool_call_no_fabricated_speech(self): + turns = [ + ([], _msg("", tool_calls=[_tc("c1", "get_reservation", "{}")])), + (["All set."], _msg("All set.")), + ] + tool_handler = MagicMock() + tool_handler.execute = AsyncMock(return_value={"status": "success"}) + system = AgenticSystem( + "x", _agent([_tool()]), tool_handler, AuditLog(), _StreamLLM(turns), pre_tool_speech="auto", llm_streaming=True + ) + assert await _collect(system, "look up A") == ["All set."] + + @pytest.mark.asyncio + async def test_streaming_cancellation_records_spoken_prefix(self): + audit_log = AuditLog() + llm = _CancellingStreamLLM() + system = AgenticSystem("x", _agent(), MagicMock(), audit_log, llm, llm_streaming=True) + + assert await _collect(system, "hi") == ["Let me check."] + assert llm.calls == 1 + assert [m.model_dump(exclude_none=True) for m in audit_log.get_conversation_messages()] == [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "Let me check."}, + ] + + @pytest.mark.asyncio + async def test_streaming_failure_records_prefix_without_generic_error(self): + audit_log = AuditLog() + llm = _FailingStreamLLM() + system = AgenticSystem("x", _agent(), MagicMock(), audit_log, llm, llm_streaming=True) + + assert await _collect(system, "hi") == ["Let me check."] + assert llm.calls == 1 + assert [m.model_dump(exclude_none=True) for m in audit_log.get_conversation_messages()] == [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "Let me check."}, + ] + assert all(entry.get("value") != GENERIC_ERROR for entry in audit_log.transcript) + + @pytest.mark.asyncio + async def test_responses_api_streaming_falls_back_with_one_warning(self, caplog): + llm = MagicMock() + llm.use_responses_api = True + llm.complete = AsyncMock( + side_effect=[ + (_msg("First response."), dict(_STATS)), + (_msg("Second response."), dict(_STATS)), + ] + ) + llm.complete_stream = MagicMock() + system = AgenticSystem("x", _agent(), MagicMock(), AuditLog(), llm, llm_streaming=True) + + with caplog.at_level("WARNING", logger="eva.assistant.agentic.system"): + assert await _collect(system, "hi") == ["First response."] + assert await _collect(system, "again") == ["Second response."] + + assert llm.complete.await_count == 2 + llm.complete_stream.assert_not_called() + assert caplog.text.count("llm_streaming is not supported for Responses API deployments") == 1 + + +class TestCompleteStream: + @pytest.mark.asyncio + async def test_yields_deltas_then_assembled_final(self, monkeypatch): + from eva.assistant.services import llm as llm_mod + + def _chunk(content): + return SimpleNamespace(choices=[SimpleNamespace(delta=SimpleNamespace(content=content), finish_reason=None)]) + + async def _acompletion(**kwargs): + async def gen(): + for c in ["Hello ", "there."]: + yield _chunk(c) + + return gen() + + mock_router = MagicMock() + mock_router.model_list = [] + mock_router.acompletion = _acompletion + monkeypatch.setattr(llm_mod.router, "get", lambda: mock_router) + + assembled = SimpleNamespace( + choices=[ + SimpleNamespace( + message=SimpleNamespace(content="Hello there.", tool_calls=[], reasoning_content=None), + finish_reason="stop", + ) + ], + usage=SimpleNamespace(prompt_tokens=1, completion_tokens=2), + model="test-model", + _hidden_params={}, + ) + monkeypatch.setattr(llm_mod.litellm, "stream_chunk_builder", lambda chunks, messages=None: assembled) + + client = llm_mod.LiteLLMClient(model="test-model") + events = [ev async for ev in client.complete_stream([{"role": "user", "content": "hi"}])] + + deltas = [p for k, p in events if k == "delta"] + finals = [p for k, p in events if k == "final"] + assert deltas == ["Hello ", "there."] + assert len(finals) == 1 + message, stats = finals[0] + assert message.content == "Hello there." + assert stats["model"] == "test-model" and "latency" in stats diff --git a/tests/unit/assistant/test_litellm_client.py b/tests/unit/assistant/test_litellm_client.py index 6c28f337..fecb8cec 100644 --- a/tests/unit/assistant/test_litellm_client.py +++ b/tests/unit/assistant/test_litellm_client.py @@ -275,3 +275,91 @@ async def test_tool_call_response_includes_output_items(self): assert output_items is not None assert any(it["type"] == "reasoning" for it in output_items) assert any(it["type"] == "function_call" for it in output_items) + + +def _standard_chat_response(): + msg = SimpleNamespace(content="ok", tool_calls=None, reasoning_content=None, thinking_blocks=None) + return SimpleNamespace( + choices=[SimpleNamespace(message=msg, finish_reason="stop")], + usage=SimpleNamespace(prompt_tokens=1, completion_tokens=1), + model="gpt-5.2", + _hidden_params={}, + ) + + +_A_TOOL = [{"type": "function", "function": {"name": "f", "description": "d", "parameters": {"type": "object", "properties": {}}}}] + + +async def _empty_stream(): + for chunk in (): + yield chunk + + +class TestParallelToolCallsForwarding: + @pytest.fixture(autouse=True, scope="class") + def _init_router(self): + router.init( + model_list=[{"model_name": "gpt-5.2", "litellm_params": {"model": "openai/gpt-5.2", "api_key": "test-key"}}] + ) + yield + router.reset() + + def _mock_router(self): + r = MagicMock() + r.model_list = [] + r.acompletion = AsyncMock(return_value=_standard_chat_response()) + return r + + @pytest.mark.asyncio + async def test_forwarded_when_set_with_tools(self): + client = LiteLLMClient(model="gpt-5.2", parallel_tool_calls=False) + r = self._mock_router() + with patch("eva.assistant.services.llm.router.get", return_value=r): + await client.complete([{"role": "user", "content": "hi"}], tools=_A_TOOL) + assert r.acompletion.call_args.kwargs["parallel_tool_calls"] is False + + @pytest.mark.asyncio + async def test_absent_when_none(self): + client = LiteLLMClient(model="gpt-5.2") + r = self._mock_router() + with patch("eva.assistant.services.llm.router.get", return_value=r): + await client.complete([{"role": "user", "content": "hi"}], tools=_A_TOOL) + assert "parallel_tool_calls" not in r.acompletion.call_args.kwargs + + @pytest.mark.asyncio + async def test_absent_when_no_tools(self): + client = LiteLLMClient(model="gpt-5.2", parallel_tool_calls=True) + r = self._mock_router() + with patch("eva.assistant.services.llm.router.get", return_value=r): + await client.complete([{"role": "user", "content": "hi"}]) + assert "parallel_tool_calls" not in r.acompletion.call_args.kwargs + + @pytest.mark.asyncio + async def test_streaming_forwards_when_set(self): + client = LiteLLMClient(model="gpt-5.2", parallel_tool_calls=False) + + r = MagicMock() + r.model_list = [] + r.acompletion = AsyncMock(return_value=_empty_stream()) + with ( + patch("eva.assistant.services.llm.router.get", return_value=r), + patch("litellm.stream_chunk_builder", return_value=_standard_chat_response()), + ): + async for _kind, _payload in client.complete_stream([{"role": "user", "content": "hi"}], tools=_A_TOOL): + pass + assert r.acompletion.call_args.kwargs["parallel_tool_calls"] is False + + @pytest.mark.asyncio + async def test_streaming_absent_when_no_tools(self): + client = LiteLLMClient(model="gpt-5.2", parallel_tool_calls=True) + + r = MagicMock() + r.model_list = [] + r.acompletion = AsyncMock(return_value=_empty_stream()) + with ( + patch("eva.assistant.services.llm.router.get", return_value=r), + patch("litellm.stream_chunk_builder", return_value=_standard_chat_response()), + ): + async for _kind, _payload in client.complete_stream([{"role": "user", "content": "hi"}]): + pass + assert "parallel_tool_calls" not in r.acompletion.call_args.kwargs diff --git a/tests/unit/assistant/test_services.py b/tests/unit/assistant/test_services.py index 7ed97c93..0005c7f0 100644 --- a/tests/unit/assistant/test_services.py +++ b/tests/unit/assistant/test_services.py @@ -122,9 +122,41 @@ def test_xai_stt_defaults_and_forwards(self): assert svc._settings.endpointing == 42 # Explicit override wins over the Eva default assert svc._settings.interim_results is True # EVA default preserved - def test_cartesia_service_created(self): - svc = create_stt_service("cartesia", params={"api_key": "k", "model": "ink"}) - assert "Cartesia" in type(svc).__name__ + def test_cartesia_is_ink2_turns_service(self): + svc = create_stt_service("cartesia", params={"api_key": "k", "model": "ink-2"}) + assert "Turns" in type(svc).__name__ + + def test_cartesia_defaults_to_ink2(self): + svc = create_stt_service("cartesia", params={"api_key": "k"}) + assert svc._settings.model == "ink-2" + + def test_cartesia_multilingual_is_ink_whisper(self): + svc = create_stt_service("cartesia-multilingual", params={"api_key": "k", "model": "ink-whisper"}) + assert "Cartesia" in type(svc).__name__ and "Turns" not in type(svc).__name__ + + def test_unknown_stt_error_lists_cartesia_multilingual_and_xai(self): + with pytest.raises(ValueError, match="Unknown STT model") as exc: + create_stt_service("bogus", params={"api_key": "k"}) + msg = str(exc.value) + assert "cartesia-multilingual" in msg + assert "xai" in msg + + +_CARTESIA_STT_PROVIDERS = [ + ("cartesia", {"api_key": "k", "model": "ink-2"}), + ("cartesia-multilingual", {"api_key": "k", "model": "ink-whisper"}), +] + + +class TestCartesiaSttInputSampleRate: + @pytest.mark.parametrize("provider,params", _CARTESIA_STT_PROVIDERS) + def test_cartesia_stt_declares_16khz_input(self, provider, params): + svc = create_stt_service(provider, params=params) + assert svc._init_sample_rate == 16000 + + def test_cartesia_caller_can_override_sample_rate(self): + svc = create_stt_service("cartesia", params={"api_key": "k", "model": "ink-2", "sample_rate": 8000}) + assert svc._init_sample_rate == 8000 class TestCreateTtsService: diff --git a/tests/unit/models/test_config_models.py b/tests/unit/models/test_config_models.py index a4507387..4d96402d 100644 --- a/tests/unit/models/test_config_models.py +++ b/tests/unit/models/test_config_models.py @@ -12,6 +12,7 @@ from eva.models.config import ( ElevenLabsSimulatorConfig, + ModelConfig, OpenAIRealtimeSimulatorConfig, PipelineType, RunConfig, @@ -787,6 +788,84 @@ def test_audio_llm_config_turn_strategy_defaults(self): assert config.model.vad_params == {} +class TestSelfEndpointingSTTAutowire: + def test_cartesia_forces_external_and_no_vad(self): + m = ModelConfig(stt="cartesia", llm="gpt-5.2") + assert m.turn_start_strategy == "external" + assert m.turn_stop_strategy == "external" + assert m.vad == "none" + + def test_conflicting_user_values_are_overridden(self): + m = ModelConfig( + stt="cartesia", + llm="gpt-5.2", + vad="silero", + turn_start_strategy="vad", + turn_stop_strategy="turn_analyzer", + ) + assert (m.turn_start_strategy, m.turn_stop_strategy, m.vad) == ("external", "external", "none") + + def test_non_self_endpointing_stt_untouched(self): + m = ModelConfig(stt="cartesia-multilingual", llm="gpt-5.2") + assert (m.turn_start_strategy, m.turn_stop_strategy, m.vad) == ("vad", "turn_analyzer", "silero") + + def test_persisted_config_roundtrip_is_stable(self): + m = ModelConfig( + stt="cartesia", + llm="gpt-5.2", + turn_start_strategy="external", + turn_stop_strategy="external", + vad="none", + ) + assert (m.turn_start_strategy, m.turn_stop_strategy, m.vad) == ("external", "external", "none") + + def test_runconfig_roundtrip(self): + c = _config( + env_vars=_BASE_ENV + | { + "EVA_MODEL__STT": "cartesia", + "EVA_MODEL__STT_PARAMS": json.dumps({"api_key": "k", "model": "ink-2"}), + } + ) + assert c.model.stt == "cartesia" + assert (c.model.turn_start_strategy, c.model.turn_stop_strategy, c.model.vad) == ( + "external", + "external", + "none", + ) + + +class TestLatencyOptimizationFlags: + def test_defaults_off(self): + c = _config(env_vars=_BASE_ENV) + assert c.model.pre_tool_speech == "off" + assert c.model.llm_streaming is False + + def test_set_via_env(self): + c = _config(env_vars=_BASE_ENV | {"EVA_MODEL__PRE_TOOL_SPEECH": "auto", "EVA_MODEL__LLM_STREAMING": "true"}) + assert c.model.pre_tool_speech == "auto" + assert c.model.llm_streaming is True + + @pytest.mark.parametrize("value", ["bogus", "force"]) + def test_invalid_pre_tool_speech_rejected(self, value): + with pytest.raises(ValueError): + _config(env_vars=_BASE_ENV | {"EVA_MODEL__PRE_TOOL_SPEECH": value}) + + +class TestParallelToolCallsConfig: + def test_default_is_none(self): + m = ModelConfig(llm="gpt-5.2") + assert m.parallel_tool_calls is None + + def test_false_via_env(self): + c = _config(env_vars=_BASE_ENV | {"EVA_MODEL__PARALLEL_TOOL_CALLS": "false"}) + assert c.model.parallel_tool_calls is False + + def test_true_via_env(self): + c = _config(env_vars=_BASE_ENV | {"EVA_MODEL__PARALLEL_TOOL_CALLS": "true"}) + assert c.model.parallel_tool_calls is True + + class TestApiKeyRedactionInPipelineModels: """api_key redaction works for all three pipeline config types via RunConfig serialization.""" diff --git a/tests/unit/utils/test_log_processing.py b/tests/unit/utils/test_log_processing.py index cc7d049b..574b9f90 100644 --- a/tests/unit/utils/test_log_processing.py +++ b/tests/unit/utils/test_log_processing.py @@ -115,3 +115,19 @@ def test_no_match_returns_none(self): result = truncate_to_spoken(audit_text, [pipecat_text]) assert result is None + + def test_multi_segment_fully_spoken_returns_full(self): + audit_text = "Let me pull that up. The change fee is seventy-five dollars. Shall I proceed?" + segments = ["Let me pull that up.", "The change fee is seventy-five dollars.", "Shall I proceed?"] + assert truncate_to_spoken(audit_text, segments) == audit_text + + def test_multi_segment_interrupted_returns_prefix(self): + audit_text = ( + "Let me pull up your reservation and check the fees. " + "The total is one hundred fifteen dollars. Shall I proceed?" + ) + segments = ["Let me pull up your reservation and check the fees."] + result = truncate_to_spoken(audit_text, segments) + assert result is not None and result != audit_text + assert "pull up your reservation" in result + assert "Shall I proceed" not in result