alm streaming

configs/prompts/simulation.yaml

@@ -64,6 +64,8 @@ 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
@@ -127,6 +129,8 @@ 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: |
@@ -187,6 +191,8 @@ 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

src/eva/assistant/agentic/audio_llm_system.py

@@ -39,6 +39,8 @@ def __init__(
         audit_log: AuditLog,
         alm_client: BaseALMClient,
         output_dir: Path | None = None,
+        llm_streaming: bool = False,
+        full_audio_context: bool = False,
     ):
         super().__init__(
             current_date_time=current_date_time,
@@ -47,8 +49,13 @@ def __init__(
             audit_log=audit_log,
             llm_client=alm_client,
             output_dir=output_dir,
+            llm_streaming=llm_streaming,
         )
         self.alm_client: BaseALMClient = alm_client
+        # When True, every user turn is sent as audio (full audio context). When False
+        # (default), only the current turn carries audio and prior turns rely on their
+        # text transcriptions, keeping context small.
+        self.full_audio_context = full_audio_context
 
         # Override system prompt with audio-LLM specific version
         self.system_prompt = self.prompt_manager.get_prompt(
@@ -91,12 +98,14 @@ 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.
-
-        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.
+        """Build messages with audio on user turns, then run tool loop.
+
+        By default only the current (last) user turn is sent as audio; previous
+        user turns remain as text (transcriptions updated via the parallel
+        transcription pipeline), keeping context manageable. When
+        ``full_audio_context`` is enabled, every user turn is sent as audio so the
+        model has full conversational context without relying on transcriptions —
+        at the cost of a much larger context.
         """
         messages: list[dict[str, Any]] = [
             {"role": "system", "content": self.system_prompt},
@@ -106,22 +115,31 @@ 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)
         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,
-                )
+            if self.full_audio_context:
+                # Replace ALL user messages with their corresponding audio
+                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,
+                        )
+            else:
+                # Replace only the LAST user message with audio (current turn)
+                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:
+                    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,
+                    )
 
         messages.extend(history_dicts)
 

src/eva/assistant/agentic/system.py

@@ -235,15 +235,20 @@ async def _run_tool_loop(
                 if self.llm_streaming and not use_responses_api:
                     response = None
                     llm_stats = {}
+                    delta_count = 0
                     aggregator = SimpleTextAggregator()
                     async for kind, payload in self.llm_client.complete_stream(messages, tools=self.tools):
                         if kind == "delta":
+                            delta_count += 1
                             async for agg in aggregator.aggregate(payload):
                                 content_streamed = True
                                 streamed_chunks.append(agg.text)
                                 yield agg.text
                         else:
                             response, llm_stats = payload
+                    # delta_count > 1 confirms the provider streamed token-by-token rather
+                    # than a client falling back to a single non-streaming chunk.
+                    logger.debug(f"llm_streaming: received {delta_count} raw delta(s) from complete_stream")
                     remainder = await aggregator.flush()
                     if remainder and remainder.text.strip():
                         content_streamed = True

src/eva/assistant/pipecat_server.py

@@ -368,6 +368,8 @@ async def _realtime_tool_handler(params) -> None:
                     alm_client=alm_client,
                     audio_collector=audio_llm_audio_collector,
                     output_dir=self.output_dir,
+                    llm_streaming=self.pipeline_config.llm_streaming,
+                    full_audio_context=self.pipeline_config.audio_llm_params.get("full_audio_context", False),
                 )
                 audio_llm_processor.on_assistant_response = lambda msg: self._save_transcript_message_from_turn(
                     role="assistant", content=msg, timestamp=self._current_iso_timestamp()
@@ -382,7 +384,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
@@ -708,9 +709,10 @@ 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)
+                # No STT → message.content is empty, and under the forced 'external'
+                # turn-stop strategy this event is transcript-gated and won't fire anyway.
+                # Turn finalization is driven by the AudioLLMUserAudioCollector, which
+                # pushes LLMContextFrame through the ParallelPipeline on UserStoppedSpeakingFrame.
                 pass
             elif self.non_instrumented_realtime_llm:
                 # Non-instrumented realtime fallback (e.g. Ultravox)

src/eva/assistant/pipeline/alm_base.py

@@ -24,6 +24,7 @@
 from abc import ABC, abstractmethod
 from typing import Any
 
+import litellm
 from pipecat.transcriptions.language import Language
 
 from eva.models.config import LANGUAGE_DISPLAY_NAMES
@@ -101,6 +102,26 @@ def resample_pcm16(pcm_data: bytes, from_rate: int, to_rate: int) -> bytes:
     return struct.pack(f"<{len(out_samples)}h", *out_samples)
 
 
+def _assemble_stream_chunks(chunks: list, messages: list[dict[str, Any]]) -> tuple[Any, Any, str]:
+    """Reconstruct the final message from raw OpenAI-SDK stream chunks.
+
+    Delegates to litellm.stream_chunk_builder — the same assembler CASCADE's
+    LiteLLMClient.complete_stream uses (see services/llm.py) — so both pipelines
+    share one chunk-assembly implementation. It expects dict-shaped chunks, so
+    pydantic chunks from the raw AsyncOpenAI client are dumped first.
+
+    Returns (message, usage, finish_reason). ``message`` has real
+    ``.content`` / ``.tool_calls[i].function.name/arguments`` / ``model_dump()``
+    attributes, matching what AgenticSystem._run_tool_loop expects.
+    """
+    dict_chunks = [c.model_dump() if hasattr(c, "model_dump") else c for c in chunks]
+    full = litellm.stream_chunk_builder(dict_chunks, messages=messages)
+    message = full.choices[0].message
+    usage = getattr(full, "usage", None)
+    finish_reason = getattr(full.choices[0], "finish_reason", None) or "unknown"
+    return message, usage, finish_reason
+
+
 class BaseALMClient(ABC):
     """Common interface and shared behavior for audio-LLM clients."""
 
@@ -189,6 +210,24 @@ async def complete(
         the content string.
         """
 
+    async def complete_stream(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict] | None = None,
+    ):
+        """Yield text deltas then the assembled final message and stats.
+
+        Default implementation falls back to complete() for providers that
+        don't support streaming. Subclasses should override for true streaming.
+
+        Yields tuples of ("delta", text_chunk) for each text delta, then
+        ("final", (message, stats)) once when the response is complete.
+        """
+        message_or_content, stats = await self.complete(messages, tools=tools)
+        if isinstance(message_or_content, str) and message_or_content:
+            yield ("delta", message_or_content)
+        yield ("final", (message_or_content, stats))
+
     @abstractmethod
     async def transcribe(
         self,

src/eva/assistant/pipeline/alm_gemini.py

@@ -29,6 +29,7 @@
     DEFAULT_SAMPLE_RATE,
     DEFAULT_SAMPLE_WIDTH,
     BaseALMClient,
+    _assemble_stream_chunks,
 )
 from eva.utils.logging import get_logger
 
@@ -247,6 +248,87 @@ async def complete(
 
         raise last_exception  # type: ignore[misc]
 
+    async def complete_stream(
+        self,
+        messages: list[dict[str, Any]],
+        tools: list[dict] | None = None,
+    ):
+        """Stream chat completion from Gemini, yielding text deltas then the final message/stats."""
+        kwargs: dict[str, Any] = {
+            "model": self.model,
+            "messages": messages,
+            "temperature": self.temperature,
+            "max_tokens": self.max_tokens,
+            "stream": True,
+            # Ask the server to emit a final usage chunk; without this, streamed
+            # responses carry no usage and completion/prompt tokens come back as 0.
+            "stream_options": {"include_usage": True},
+        }
+        extra_body = self._gemini_extra_body()
+        if extra_body:
+            kwargs["extra_body"] = extra_body
+        if tools:
+            kwargs["tools"] = tools
+            kwargs["tool_choice"] = "auto"
+
+        last_exception: Exception | None = None
+        for attempt in range(self.max_retries + 1):
+            chunks: list[Any] = []
+            first_token = False
+            try:
+                self._maybe_refresh_token()
+                start_time = time.time()
+                stream = await self._client.chat.completions.create(**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() - start_time
+
+                # Gemini's OpenAI-compat endpoint does not surface reasoning content (its
+                # complete() returns reasoning=None), so the reasoning field is ignored here.
+                message, usage, finish_reason = _assemble_stream_chunks(chunks, messages)
+                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": usage.prompt_tokens if usage else 0,
+                    "completion_tokens": usage.completion_tokens if usage else 0,
+                    "reasoning_tokens": reasoning_tokens,
+                    "finish_reason": finish_reason,
+                    "model": self.model,
+                    "cost": 0.0,
+                    "cost_source": "gemini_openai_compat",
+                    "latency": round(elapsed, 3),
+                    "reasoning": None,
+                    "reasoning_content": None,
+                }
+                yield ("final", (message, stats))
+                return
+
+            except Exception as e:
+                last_exception = e
+                if self._is_retryable(e) and attempt < self.max_retries and not first_token:
+                    delay = self.initial_delay * (2**attempt)
+                    logger.warning(
+                        f"Retryable streaming error (attempt {attempt + 1}/{self.max_retries + 1}): {e}. "
+                        f"Retrying in {delay:.1f}s..."
+                    )
+                    await asyncio.sleep(delay)
+                    continue
+                logger.error(f"ALMGeminiClient streaming completion failed: {e}")
+                raise
+
+        raise last_exception  # type: ignore[misc]
+
     async def transcribe(
         self,
         audio_bytes: bytes,