Changes to Audio LLM pipeline

.env.example

@@ -69,15 +69,15 @@ 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
 
 #i STT provider parameters. Must include "api_key" and "model". Use "urls" for round-robin load balancing.
 #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

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

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.

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)
 

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,13 +116,18 @@ 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
         self.audit_log = audit_log
         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
 

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:

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)