From 7da1ebe68c4f86d6e68748918c635602460c48cd Mon Sep 17 00:00:00 2001 From: "raghav.mehndiratta" Date: Mon, 13 Jul 2026 20:36:58 -0700 Subject: [PATCH 1/7] alm streaming --- configs/prompts/simulation.yaml | 6 + src/eva/assistant/agentic/audio_llm_system.py | 60 ++++--- src/eva/assistant/agentic/system.py | 6 + src/eva/assistant/pipecat_server.py | 16 +- src/eva/assistant/pipeline/alm_base.py | 80 +++++++++ src/eva/assistant/pipeline/alm_gemini.py | 78 ++++++++ src/eva/assistant/pipeline/alm_vllm.py | 169 +++++++++++++++++- .../assistant/pipeline/audio_llm_processor.py | 125 ++++++++++--- src/eva/assistant/pipeline/services.py | 1 + src/eva/models/config.py | 7 + 10 files changed, 488 insertions(+), 60 deletions(-) diff --git a/configs/prompts/simulation.yaml b/configs/prompts/simulation.yaml index 45a44209..2c846a5c 100644 --- a/configs/prompts/simulation.yaml +++ b/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 diff --git a/src/eva/assistant/agentic/audio_llm_system.py b/src/eva/assistant/agentic/audio_llm_system.py index 14ae92c0..9ff227a3 100644 --- a/src/eva/assistant/agentic/audio_llm_system.py +++ b/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) diff --git a/src/eva/assistant/agentic/system.py b/src/eva/assistant/agentic/system.py index 7d197703..a937ec73 100644 --- a/src/eva/assistant/agentic/system.py +++ b/src/eva/assistant/agentic/system.py @@ -235,15 +235,21 @@ 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 + # Proof of real streaming: many raw deltas means token-by-token from the + # provider; exactly 1 means a non-streaming client fell back to a single + # chunk. Safe to ignore / silence on large evals. + 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 diff --git a/src/eva/assistant/pipecat_server.py b/src/eva/assistant/pipecat_server.py index c758f62c..feebb017 100644 --- a/src/eva/assistant/pipecat_server.py +++ b/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_full_audio_context, ) 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 @@ -485,6 +486,7 @@ async def on_assistant_response(msg: str) -> 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, ) @@ -637,6 +639,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.""" @@ -708,10 +711,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 + # 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/alm_base.py b/src/eva/assistant/pipeline/alm_base.py index d362ef6b..114f28d9 100644 --- a/src/eva/assistant/pipeline/alm_base.py +++ b/src/eva/assistant/pipeline/alm_base.py @@ -101,6 +101,68 @@ def resample_pcm16(pcm_data: bytes, from_rate: int, to_rate: int) -> bytes: return struct.pack(f"<{len(out_samples)}h", *out_samples) +class _ToolCallDump: + """Minimal tool-call-like object whose model_dump() returns the assembled dict.""" + + def __init__(self, d: dict) -> None: + self._d = d + + def model_dump(self, exclude_none: bool = False) -> dict: + return self._d + + +class _StreamedMessage: + """Minimal message-like object returned from complete_stream().""" + + def __init__(self, content: str, tool_calls: list | None) -> None: + self.content = content + self.tool_calls = tool_calls or None + + +def _assemble_stream_chunks(chunks: list) -> tuple: + """Reconstruct (content, finish_reason, usage, tool_calls) from streaming chunks. + + Returns a 4-tuple: + - full_content: str — concatenated text deltas + - finish_reason: str + - usage: usage object or None + - assembled_tool_calls: list[_ToolCallDump] or None + """ + full_content = "" + finish_reason = "unknown" + usage = None + tool_calls_by_index: dict[int, dict] = {} + + for chunk in chunks: + choices = getattr(chunk, "choices", None) or [] + if choices: + delta = getattr(choices[0], "delta", None) + if delta: + text = getattr(delta, "content", None) or "" + full_content += text + for tc in getattr(delta, "tool_calls", None) or []: + idx = getattr(tc, "index", 0) + if idx not in tool_calls_by_index: + tool_calls_by_index[idx] = { + "id": getattr(tc, "id", "") or "", + "type": "function", + "function": {"name": "", "arguments": ""}, + } + fn = getattr(tc, "function", None) + if fn: + tool_calls_by_index[idx]["function"]["name"] += getattr(fn, "name", "") or "" + tool_calls_by_index[idx]["function"]["arguments"] += getattr(fn, "arguments", "") or "" + fr = getattr(choices[0], "finish_reason", None) + if fr: + finish_reason = fr + chunk_usage = getattr(chunk, "usage", None) + if chunk_usage: + usage = chunk_usage + + assembled = [_ToolCallDump(d) for d in tool_calls_by_index.values()] if tool_calls_by_index else None + return full_content, finish_reason, usage, assembled + + class BaseALMClient(ABC): """Common interface and shared behavior for audio-LLM clients.""" @@ -189,6 +251,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, diff --git a/src/eva/assistant/pipeline/alm_gemini.py b/src/eva/assistant/pipeline/alm_gemini.py index 168a037d..39f31332 100644 --- a/src/eva/assistant/pipeline/alm_gemini.py +++ b/src/eva/assistant/pipeline/alm_gemini.py @@ -29,6 +29,8 @@ DEFAULT_SAMPLE_RATE, DEFAULT_SAMPLE_WIDTH, BaseALMClient, + _StreamedMessage, + _assemble_stream_chunks, ) from eva.utils.logging import get_logger @@ -247,6 +249,82 @@ 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, + } + 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 + + full_content, finish_reason, usage, assembled_tool_calls = _assemble_stream_chunks(chunks) + 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", (_StreamedMessage(full_content, assembled_tool_calls), 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, diff --git a/src/eva/assistant/pipeline/alm_vllm.py b/src/eva/assistant/pipeline/alm_vllm.py index 92891442..9d86faa2 100644 --- a/src/eva/assistant/pipeline/alm_vllm.py +++ b/src/eva/assistant/pipeline/alm_vllm.py @@ -15,12 +15,53 @@ DEFAULT_SAMPLE_RATE, DEFAULT_SAMPLE_WIDTH, BaseALMClient, + _StreamedMessage, + _assemble_stream_chunks, ) from eva.utils.llm_utils import approximate_reasoning_tokens from eva.utils.logging import get_logger 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 +80,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 +95,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()/complete_stream() 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,15 +114,62 @@ 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", "audio_url": {"url": f"data:audio/wav;base64,{audio_b64}"}, } + def _build_extra_body(self) -> dict[str, Any]: + """Build the extra_body payload: chat_template_kwargs + vLLM sampling params. + + The validated sampling params ride alongside chat_template_kwargs; vLLM + accepts non-OpenAI keys at the top level of extra_body. + """ + extra_body: dict[str, Any] = { + "chat_template_kwargs": {"enable_thinking": self.enable_thinking}, + } + extra_body.update(self.sampling_params) + return extra_body + async def complete( self, messages: list[dict[str, Any]], @@ -94,11 +188,7 @@ async def complete( "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens, - "extra_body": { - "chat_template_kwargs": { - "enable_thinking": self.enable_thinking, - } - }, + "extra_body": self._build_extra_body(), } if tools: kwargs["tools"] = tools @@ -162,6 +252,73 @@ 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 vLLM, 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, + "extra_body": self._build_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: + 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 + + full_content, finish_reason, usage, assembled_tool_calls = _assemble_stream_chunks(chunks) + stats = { + "prompt_tokens": usage.prompt_tokens if usage else 0, + "completion_tokens": usage.completion_tokens if usage else 0, + "reasoning_tokens": 0, + "finish_reason": finish_reason, + "model": self.model, + "cost": 0.0, + "cost_source": "self_hosted", + "latency": round(elapsed, 3), + "reasoning": None, + "reasoning_content": None, + } + yield ("final", (_StreamedMessage(full_content, assembled_tool_calls), 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"ALMvLLMClient streaming completion failed: {e}") + raise + + raise last_exception # type: ignore[misc] + async def transcribe( self, audio_bytes: bytes, diff --git a/src/eva/assistant/pipeline/audio_llm_processor.py b/src/eva/assistant/pipeline/audio_llm_processor.py index e7b23cb3..a068cca5 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 turn-stop 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 turn-stop-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 + 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. @@ -169,6 +225,8 @@ def __init__( alm_client: BaseALMClient, audio_collector: AudioLLMUserAudioCollector, output_dir: Path | None = None, + llm_streaming: bool = False, + full_audio_context: bool = False, **kwargs, ) -> None: super().__init__(**kwargs) @@ -184,6 +242,8 @@ def __init__( audit_log=audit_log, alm_client=alm_client, output_dir=output_dir, + llm_streaming=llm_streaming, + full_audio_context=full_audio_context, ) # State tracking (mirrors BenchmarkAgentProcessor) @@ -257,7 +317,8 @@ 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 +329,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 +453,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 +487,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 +515,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 +547,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 21fb0fd3..46db7f68 100644 --- a/src/eva/assistant/pipeline/services.py +++ b/src/eva/assistant/pipeline/services.py @@ -845,6 +845,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/models/config.py b/src/eva/models/config.py index ea797e8a..dd815d4c 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -196,6 +196,13 @@ class ModelConfig(BaseModel): False, description="Stream Chat Completions output to TTS sentence-by-sentence.", ) + audio_llm_full_audio_context: bool = Field( + False, + description=( + "AUDIO_LLM only: send every user turn as audio (full audio context) instead of " + "only the current turn. Removes reliance on transcriptions but grows context quickly." + ), + ) parallel_tool_calls: bool | None = Field( None, description="Forward parallel_tool_calls when tools are present; None leaves provider defaults.", From cafa332ed3a45e551b6a9c7becae276183f309a3 Mon Sep 17 00:00:00 2001 From: "raghav.mehndiratta" Date: Tue, 14 Jul 2026 20:18:22 -0700 Subject: [PATCH 2/7] bug fixes --- src/eva/assistant/pipecat_server.py | 14 ++-- src/eva/assistant/pipeline/alm_base.py | 30 ++++++- src/eva/assistant/pipeline/alm_gemini.py | 4 +- src/eva/assistant/pipeline/alm_vllm.py | 22 ++++- .../assistant/pipeline/audio_llm_processor.py | 81 ++++++------------- src/eva/models/config.py | 17 +++- 6 files changed, 88 insertions(+), 80 deletions(-) diff --git a/src/eva/assistant/pipecat_server.py b/src/eva/assistant/pipecat_server.py index feebb017..606fc92b 100644 --- a/src/eva/assistant/pipecat_server.py +++ b/src/eva/assistant/pipecat_server.py @@ -486,7 +486,6 @@ async def on_assistant_response(msg: str) -> 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, ) @@ -639,7 +638,6 @@ 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.""" @@ -711,13 +709,11 @@ 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. Finalize the turn now that the - # 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() + # 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) if message.content: diff --git a/src/eva/assistant/pipeline/alm_base.py b/src/eva/assistant/pipeline/alm_base.py index 114f28d9..e5fdbd78 100644 --- a/src/eva/assistant/pipeline/alm_base.py +++ b/src/eva/assistant/pipeline/alm_base.py @@ -101,11 +101,29 @@ def resample_pcm16(pcm_data: bytes, from_rate: int, to_rate: int) -> bytes: return struct.pack(f"<{len(out_samples)}h", *out_samples) +class _StreamedFunction: + """Function part of a reconstructed streamed tool call (attribute access).""" + + def __init__(self, name: str, arguments: str) -> None: + self.name = name + self.arguments = arguments + + class _ToolCallDump: - """Minimal tool-call-like object whose model_dump() returns the assembled dict.""" + """Reconstructed streamed tool call. + + Mirrors the shape the tool loop expects from a real OpenAI tool-call object: + attribute access (``.id`` / ``.type`` / ``.function.name`` / ``.function.arguments``) + used by ``_run_tool_loop`` to execute the call, plus ``model_dump()`` used to + serialize it back into assistant-message history. + """ def __init__(self, d: dict) -> None: self._d = d + self.id = d.get("id", "") + self.type = d.get("type", "function") + fn = d.get("function", {}) + self.function = _StreamedFunction(fn.get("name", ""), fn.get("arguments", "")) def model_dump(self, exclude_none: bool = False) -> dict: return self._d @@ -120,15 +138,17 @@ def __init__(self, content: str, tool_calls: list | None) -> None: def _assemble_stream_chunks(chunks: list) -> tuple: - """Reconstruct (content, finish_reason, usage, tool_calls) from streaming chunks. + """Reconstruct (content, reasoning, finish_reason, usage, tool_calls) from streaming chunks. - Returns a 4-tuple: + Returns a 5-tuple: - full_content: str — concatenated text deltas + - reasoning_content: str — concatenated reasoning/thinking deltas (empty if none) - finish_reason: str - usage: usage object or None - assembled_tool_calls: list[_ToolCallDump] or None """ full_content = "" + reasoning_content = "" finish_reason = "unknown" usage = None tool_calls_by_index: dict[int, dict] = {} @@ -140,6 +160,8 @@ def _assemble_stream_chunks(chunks: list) -> tuple: if delta: text = getattr(delta, "content", None) or "" full_content += text + # vLLM streams thinking either as `reasoning_content` or `reasoning` on the delta. + reasoning_content += getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None) or "" for tc in getattr(delta, "tool_calls", None) or []: idx = getattr(tc, "index", 0) if idx not in tool_calls_by_index: @@ -160,7 +182,7 @@ def _assemble_stream_chunks(chunks: list) -> tuple: usage = chunk_usage assembled = [_ToolCallDump(d) for d in tool_calls_by_index.values()] if tool_calls_by_index else None - return full_content, finish_reason, usage, assembled + return full_content, reasoning_content, finish_reason, usage, assembled class BaseALMClient(ABC): diff --git a/src/eva/assistant/pipeline/alm_gemini.py b/src/eva/assistant/pipeline/alm_gemini.py index 39f31332..9243b7ea 100644 --- a/src/eva/assistant/pipeline/alm_gemini.py +++ b/src/eva/assistant/pipeline/alm_gemini.py @@ -288,7 +288,9 @@ async def complete_stream( yield ("delta", text) elapsed = time.time() - start_time - full_content, finish_reason, usage, assembled_tool_calls = _assemble_stream_chunks(chunks) + # Gemini's OpenAI-compat endpoint does not surface reasoning content (its + # complete() returns reasoning=None), so the reasoning field is ignored here. + full_content, _reasoning, finish_reason, usage, assembled_tool_calls = _assemble_stream_chunks(chunks) reasoning_tokens = 0 if usage and hasattr(usage, "completion_tokens_details"): details = usage.completion_tokens_details diff --git a/src/eva/assistant/pipeline/alm_vllm.py b/src/eva/assistant/pipeline/alm_vllm.py index 9d86faa2..eac53096 100644 --- a/src/eva/assistant/pipeline/alm_vllm.py +++ b/src/eva/assistant/pipeline/alm_vllm.py @@ -288,18 +288,32 @@ async def complete_stream( yield ("delta", text) elapsed = time.time() - start_time - full_content, finish_reason, usage, assembled_tool_calls = _assemble_stream_chunks(chunks) + full_content, reasoning_content, finish_reason, usage, assembled_tool_calls = _assemble_stream_chunks( + chunks + ) + reasoning_content = reasoning_content or None + + # Reasoning tokens from usage if the server reported them; else approximate from + # the streamed reasoning text (mirrors the non-streaming complete() path). + 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 + if reasoning_content and reasoning_tokens == 0: + reasoning_tokens = approximate_reasoning_tokens(reasoning_content, self.model, self, logger) + stats = { "prompt_tokens": usage.prompt_tokens if usage else 0, "completion_tokens": usage.completion_tokens if usage else 0, - "reasoning_tokens": 0, + "reasoning_tokens": reasoning_tokens, "finish_reason": finish_reason, "model": self.model, "cost": 0.0, "cost_source": "self_hosted", "latency": round(elapsed, 3), - "reasoning": None, - "reasoning_content": None, + "reasoning": reasoning_content, + "reasoning_content": reasoning_content, } yield ("final", (_StreamedMessage(full_content, assembled_tool_calls), stats)) return diff --git a/src/eva/assistant/pipeline/audio_llm_processor.py b/src/eva/assistant/pipeline/audio_llm_processor.py index a068cca5..30906345 100644 --- a/src/eva/assistant/pipeline/audio_llm_processor.py +++ b/src/eva/assistant/pipeline/audio_llm_processor.py @@ -56,22 +56,19 @@ class AudioLLMUserAudioCollector(FrameProcessor): """Buffers raw audio frames during user speech for the audio-LLM pipeline. - 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 turn-stop 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. + Collects audio and pushes LLMContextFrame when the user stops speaking, which + triggers the parallel pipeline (transcription + audio-LLM processing). + + Turn boundaries are driven directly by the UserStartedSpeakingFrame / + UserStoppedSpeakingFrame frames the transport emits. In the AUDIO_LLM + pipeline the user aggregator runs the ``external`` turn-stop strategy with no + STT, so ``on_user_turn_stopped`` is transcript-gated and never fires — the + collector must finalize on the stop frame itself rather than defer to that + event. + + 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. All frames pass through unchanged. """ @@ -92,11 +89,7 @@ def __init__( self._audio_buffer = bytearray() self._pre_speech_buffer: list[bytes] = [] self._user_speaking = False - 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 turn-stop-governed turn instead of being fragmented. - self._turn_finalized = True + self._current_turn_id = 0 # Incremented on each user turn # 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 @@ -109,35 +102,21 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: 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) / (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) + 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) 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 - logger.debug( - f"VAD stop — pausing capture (buffer={len(self._audio_buffer)} bytes, " - f"turn_id={self._current_turn_id})" - ) + # 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)) elif isinstance(frame, InputAudioRawFrame): self._frame_sample_rate = frame.sample_rate @@ -154,20 +133,6 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: 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 - 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) diff --git a/src/eva/models/config.py b/src/eva/models/config.py index dd815d4c..1083ec53 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -267,7 +267,14 @@ def _autowire_self_endpointing_stt(self) -> "ModelConfig": "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). + + Only applies to the CASCADE pipeline: in AUDIO_LLM / S2S the ``stt`` field does not + instantiate an in-pipeline STT service, so there is nothing to emit the external turn + frames. Forcing 'external'/'none' there would disable the local VAD that those pipelines + rely on for turn detection, hanging the conversation (no user turn ever finalizes). """ + if self.pipeline_type != PipelineType.CASCADE: + return self if (self.stt or "").lower() not in self._SELF_ENDPOINTING_STT: return self @@ -290,11 +297,13 @@ 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: + # llm_streaming is honored by AUDIO_LLM too (via BaseALMClient.complete_stream); only + # pre_tool_speech / parallel_tool_calls remain CASCADE-only. + cascade_only_set = self.pre_tool_speech != "off" or self.parallel_tool_calls is not None + if cascade_only_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}." + "Cascade LLM flags (pre_tool_speech / parallel_tool_calls) apply only to the CASCADE " + f"pipeline; they will be ignored for pipeline_type={self.pipeline_type}." ) return self From a94343a0488bc5faed0b6fdf576a68706e40d340 Mon Sep 17 00:00:00 2001 From: "raghav.mehndiratta" Date: Wed, 15 Jul 2026 13:16:26 -0700 Subject: [PATCH 3/7] add usage counting --- src/eva/assistant/pipeline/alm_gemini.py | 3 +++ src/eva/assistant/pipeline/alm_vllm.py | 3 +++ 2 files changed, 6 insertions(+) diff --git a/src/eva/assistant/pipeline/alm_gemini.py b/src/eva/assistant/pipeline/alm_gemini.py index 9243b7ea..55300448 100644 --- a/src/eva/assistant/pipeline/alm_gemini.py +++ b/src/eva/assistant/pipeline/alm_gemini.py @@ -261,6 +261,9 @@ async def complete_stream( "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: diff --git a/src/eva/assistant/pipeline/alm_vllm.py b/src/eva/assistant/pipeline/alm_vllm.py index eac53096..6f9562ca 100644 --- a/src/eva/assistant/pipeline/alm_vllm.py +++ b/src/eva/assistant/pipeline/alm_vllm.py @@ -264,6 +264,9 @@ async def complete_stream( "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._build_extra_body(), } if tools: From 64648c6def06bd13b2ffe0162facad5946c33fe0 Mon Sep 17 00:00:00 2001 From: "raghav.mehndiratta" Date: Wed, 15 Jul 2026 15:05:04 -0700 Subject: [PATCH 4/7] cleanup --- src/eva/assistant/agentic/system.py | 5 +- src/eva/assistant/pipecat_server.py | 2 +- src/eva/assistant/pipeline/alm_base.py | 95 ++++-------------------- src/eva/assistant/pipeline/alm_gemini.py | 5 +- src/eva/assistant/pipeline/alm_vllm.py | 9 +-- src/eva/models/config.py | 14 ++-- 6 files changed, 30 insertions(+), 100 deletions(-) diff --git a/src/eva/assistant/agentic/system.py b/src/eva/assistant/agentic/system.py index a937ec73..c6ea0d96 100644 --- a/src/eva/assistant/agentic/system.py +++ b/src/eva/assistant/agentic/system.py @@ -246,9 +246,8 @@ async def _run_tool_loop( yield agg.text else: response, llm_stats = payload - # Proof of real streaming: many raw deltas means token-by-token from the - # provider; exactly 1 means a non-streaming client fell back to a single - # chunk. Safe to ignore / silence on large evals. + # 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(): diff --git a/src/eva/assistant/pipecat_server.py b/src/eva/assistant/pipecat_server.py index 606fc92b..a326a8df 100644 --- a/src/eva/assistant/pipecat_server.py +++ b/src/eva/assistant/pipecat_server.py @@ -369,7 +369,7 @@ async def _realtime_tool_handler(params) -> None: 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_full_audio_context, + 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() diff --git a/src/eva/assistant/pipeline/alm_base.py b/src/eva/assistant/pipeline/alm_base.py index e5fdbd78..ddfcbe8f 100644 --- a/src/eva/assistant/pipeline/alm_base.py +++ b/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,88 +102,24 @@ def resample_pcm16(pcm_data: bytes, from_rate: int, to_rate: int) -> bytes: return struct.pack(f"<{len(out_samples)}h", *out_samples) -class _StreamedFunction: - """Function part of a reconstructed streamed tool call (attribute access).""" +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. - def __init__(self, name: str, arguments: str) -> None: - self.name = name - self.arguments = arguments + 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. - -class _ToolCallDump: - """Reconstructed streamed tool call. - - Mirrors the shape the tool loop expects from a real OpenAI tool-call object: - attribute access (``.id`` / ``.type`` / ``.function.name`` / ``.function.arguments``) - used by ``_run_tool_loop`` to execute the call, plus ``model_dump()`` used to - serialize it back into assistant-message history. - """ - - def __init__(self, d: dict) -> None: - self._d = d - self.id = d.get("id", "") - self.type = d.get("type", "function") - fn = d.get("function", {}) - self.function = _StreamedFunction(fn.get("name", ""), fn.get("arguments", "")) - - def model_dump(self, exclude_none: bool = False) -> dict: - return self._d - - -class _StreamedMessage: - """Minimal message-like object returned from complete_stream().""" - - def __init__(self, content: str, tool_calls: list | None) -> None: - self.content = content - self.tool_calls = tool_calls or None - - -def _assemble_stream_chunks(chunks: list) -> tuple: - """Reconstruct (content, reasoning, finish_reason, usage, tool_calls) from streaming chunks. - - Returns a 5-tuple: - - full_content: str — concatenated text deltas - - reasoning_content: str — concatenated reasoning/thinking deltas (empty if none) - - finish_reason: str - - usage: usage object or None - - assembled_tool_calls: list[_ToolCallDump] or None + 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. """ - full_content = "" - reasoning_content = "" - finish_reason = "unknown" - usage = None - tool_calls_by_index: dict[int, dict] = {} - - for chunk in chunks: - choices = getattr(chunk, "choices", None) or [] - if choices: - delta = getattr(choices[0], "delta", None) - if delta: - text = getattr(delta, "content", None) or "" - full_content += text - # vLLM streams thinking either as `reasoning_content` or `reasoning` on the delta. - reasoning_content += getattr(delta, "reasoning_content", None) or getattr(delta, "reasoning", None) or "" - for tc in getattr(delta, "tool_calls", None) or []: - idx = getattr(tc, "index", 0) - if idx not in tool_calls_by_index: - tool_calls_by_index[idx] = { - "id": getattr(tc, "id", "") or "", - "type": "function", - "function": {"name": "", "arguments": ""}, - } - fn = getattr(tc, "function", None) - if fn: - tool_calls_by_index[idx]["function"]["name"] += getattr(fn, "name", "") or "" - tool_calls_by_index[idx]["function"]["arguments"] += getattr(fn, "arguments", "") or "" - fr = getattr(choices[0], "finish_reason", None) - if fr: - finish_reason = fr - chunk_usage = getattr(chunk, "usage", None) - if chunk_usage: - usage = chunk_usage - - assembled = [_ToolCallDump(d) for d in tool_calls_by_index.values()] if tool_calls_by_index else None - return full_content, reasoning_content, finish_reason, usage, assembled + 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): diff --git a/src/eva/assistant/pipeline/alm_gemini.py b/src/eva/assistant/pipeline/alm_gemini.py index 55300448..6a51eafe 100644 --- a/src/eva/assistant/pipeline/alm_gemini.py +++ b/src/eva/assistant/pipeline/alm_gemini.py @@ -29,7 +29,6 @@ DEFAULT_SAMPLE_RATE, DEFAULT_SAMPLE_WIDTH, BaseALMClient, - _StreamedMessage, _assemble_stream_chunks, ) from eva.utils.logging import get_logger @@ -293,7 +292,7 @@ async def complete_stream( # Gemini's OpenAI-compat endpoint does not surface reasoning content (its # complete() returns reasoning=None), so the reasoning field is ignored here. - full_content, _reasoning, finish_reason, usage, assembled_tool_calls = _assemble_stream_chunks(chunks) + 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 @@ -312,7 +311,7 @@ async def complete_stream( "reasoning": None, "reasoning_content": None, } - yield ("final", (_StreamedMessage(full_content, assembled_tool_calls), stats)) + yield ("final", (message, stats)) return except Exception as e: diff --git a/src/eva/assistant/pipeline/alm_vllm.py b/src/eva/assistant/pipeline/alm_vllm.py index 6f9562ca..55df75b2 100644 --- a/src/eva/assistant/pipeline/alm_vllm.py +++ b/src/eva/assistant/pipeline/alm_vllm.py @@ -15,7 +15,6 @@ DEFAULT_SAMPLE_RATE, DEFAULT_SAMPLE_WIDTH, BaseALMClient, - _StreamedMessage, _assemble_stream_chunks, ) from eva.utils.llm_utils import approximate_reasoning_tokens @@ -291,10 +290,8 @@ async def complete_stream( yield ("delta", text) elapsed = time.time() - start_time - full_content, reasoning_content, finish_reason, usage, assembled_tool_calls = _assemble_stream_chunks( - chunks - ) - reasoning_content = reasoning_content or None + message, usage, finish_reason = _assemble_stream_chunks(chunks, messages) + reasoning_content = getattr(message, "reasoning_content", None) or None # Reasoning tokens from usage if the server reported them; else approximate from # the streamed reasoning text (mirrors the non-streaming complete() path). @@ -318,7 +315,7 @@ async def complete_stream( "reasoning": reasoning_content, "reasoning_content": reasoning_content, } - yield ("final", (_StreamedMessage(full_content, assembled_tool_calls), stats)) + yield ("final", (message, stats)) return except Exception as e: diff --git a/src/eva/models/config.py b/src/eva/models/config.py index 1083ec53..5a4cf055 100644 --- a/src/eva/models/config.py +++ b/src/eva/models/config.py @@ -143,7 +143,12 @@ class ModelConfig(BaseModel): tts_params: dict[str, Any] | None = Field(None, description="Additional TTS model parameters (JSON)") s2s_params: dict[str, Any] | None = Field(None, description="Additional speech-to-speech model parameters (JSON)") audio_llm_params: dict[str, Any] | None = Field( - None, description="Audio-LLM parameters (JSON): base_url (required), api_key, model, temperature, max_tokens" + None, + description=( + "Audio-LLM parameters (JSON): base_url (required), api_key, model, temperature, " + "max_tokens, full_audio_context (send every user turn as audio instead of only the " + "current turn; removes reliance on transcriptions but grows context quickly)" + ), ) # Configurable turn start/stop strategies @@ -196,13 +201,6 @@ class ModelConfig(BaseModel): False, description="Stream Chat Completions output to TTS sentence-by-sentence.", ) - audio_llm_full_audio_context: bool = Field( - False, - description=( - "AUDIO_LLM only: send every user turn as audio (full audio context) instead of " - "only the current turn. Removes reliance on transcriptions but grows context quickly." - ), - ) parallel_tool_calls: bool | None = Field( None, description="Forward parallel_tool_calls when tools are present; None leaves provider defaults.", From faae3d1e7ef0cbd7afce140cf8c78f3ceebff250 Mon Sep 17 00:00:00 2001 From: raghavm243512 <44511569+raghavm243512@users.noreply.github.com> Date: Wed, 15 Jul 2026 22:06:59 +0000 Subject: [PATCH 5/7] Apply pre-commit --- src/eva/assistant/pipeline/audio_llm_processor.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/src/eva/assistant/pipeline/audio_llm_processor.py b/src/eva/assistant/pipeline/audio_llm_processor.py index 30906345..646598c0 100644 --- a/src/eva/assistant/pipeline/audio_llm_processor.py +++ b/src/eva/assistant/pipeline/audio_llm_processor.py @@ -459,9 +459,7 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: 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, source_sample_rate, 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()] From bc1a42b6391c82feb206c76368d8b66d84b588b9 Mon Sep 17 00:00:00 2001 From: Katrina Date: Wed, 22 Jul 2026 18:00:53 -0400 Subject: [PATCH 6/7] add pipecat logs to logs.log file --- src/eva/utils/logging.py | 54 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/src/eva/utils/logging.py b/src/eva/utils/logging.py index 71c78fd0..f5356b17 100644 --- a/src/eva/utils/logging.py +++ b/src/eva/utils/logging.py @@ -26,6 +26,50 @@ def filter(self, record: logging.LogRecord) -> bool: return current_record_id.get() == self.record_id +def _bridge_loguru_to_stdlib() -> None: + """Route pipecat's loguru logs into the stdlib ``pipecat`` logger. + + Pipecat logs through loguru (a separate logging system), so its records never + reach the stdlib handlers configured here — including the smart-turn analyzer's + ``End of Turn complete due to stop_secs`` / ``End of Turn result`` debug lines. + This installs a loguru sink that re-emits every loguru record as a stdlib log + record under the ``pipecat`` logger, preserving the original level and source + location. Called once from ``setup_logging`` so per-record file handlers (which + attach to the ``pipecat`` logger) capture these lines. + + No-op if loguru is not installed. + """ + try: + from loguru import logger as _loguru_logger + except ImportError: + return + + class _InterceptHandler: + def write(self, message) -> None: # loguru sink protocol + record = message.record + # Map loguru level name to a stdlib level number (fallback to the numeric no). + level = logging.getLevelName(record["level"].name) + if not isinstance(level, int): + level = record["level"].no + std_logger = logging.getLogger("pipecat") + log_record = std_logger.makeRecord( + "pipecat", + level, + record["file"].path, + record["line"], + record["message"], + (), + None, + func=record["function"], + ) + std_logger.handle(log_record) + + # Replace loguru's default stderr sink with our stdlib bridge at DEBUG so the + # smart-turn debug lines are captured; the stdlib handlers do the final filtering. + _loguru_logger.remove() + _loguru_logger.add(_InterceptHandler(), level="DEBUG", format="{message}") + + def get_logger(name: str) -> logging.Logger: """Get a logger under the ``eva`` hierarchy. @@ -83,6 +127,16 @@ def setup_logging( # Don't propagate to root logger root_logger.propagate = False + # Bridge pipecat's loguru logs into the stdlib ``pipecat`` logger so framework + # diagnostics (e.g. smart-turn end-of-turn reasons) land in the same sinks. + _bridge_loguru_to_stdlib() + pipecat_logger = logging.getLogger("pipecat") + pipecat_logger.setLevel(logging.DEBUG) + pipecat_logger.addHandler(console_handler) + if log_file: + pipecat_logger.addHandler(file_handler) + pipecat_logger.propagate = False + root_logger.debug(f"Logging configured: level={level}, file={log_file}") From da7d8267256780e777e60f01bac9640d2ff7bb87 Mon Sep 17 00:00:00 2001 From: Katrina Date: Wed, 22 Jul 2026 18:04:49 -0400 Subject: [PATCH 7/7] fix pre-commit --- src/eva/assistant/pipeline/audio_llm_processor.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/eva/assistant/pipeline/audio_llm_processor.py b/src/eva/assistant/pipeline/audio_llm_processor.py index 646598c0..3088438a 100644 --- a/src/eva/assistant/pipeline/audio_llm_processor.py +++ b/src/eva/assistant/pipeline/audio_llm_processor.py @@ -497,6 +497,7 @@ async def _transcribe_audio( Args: audio_data: Raw PCM audio bytes to transcribe. timestamp: ISO8601 timestamp for the transcription. + source_sample_rate: Sample rate (Hz) of the provided audio data. turn_id: Optional turn identifier for associating with audit log entry. Returns: