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..c6ea0d96 100644 --- a/src/eva/assistant/agentic/system.py +++ b/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 diff --git a/src/eva/assistant/pipecat_server.py b/src/eva/assistant/pipecat_server.py index c758f62c..a326a8df 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_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) diff --git a/src/eva/assistant/pipeline/alm_base.py b/src/eva/assistant/pipeline/alm_base.py index d362ef6b..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,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, diff --git a/src/eva/assistant/pipeline/alm_gemini.py b/src/eva/assistant/pipeline/alm_gemini.py index 168a037d..6a51eafe 100644 --- a/src/eva/assistant/pipeline/alm_gemini.py +++ b/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, diff --git a/src/eva/assistant/pipeline/alm_vllm.py b/src/eva/assistant/pipeline/alm_vllm.py index 92891442..55df75b2 100644 --- a/src/eva/assistant/pipeline/alm_vllm.py +++ b/src/eva/assistant/pipeline/alm_vllm.py @@ -15,12 +15,52 @@ DEFAULT_SAMPLE_RATE, DEFAULT_SAMPLE_WIDTH, BaseALMClient, + _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 +79,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 +94,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 +113,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 +187,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 +251,88 @@ 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, + # 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: + 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 + + 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). + 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": reasoning_tokens, + "finish_reason": finish_reason, + "model": self.model, + "cost": 0.0, + "cost_source": "self_hosted", + "latency": round(elapsed, 3), + "reasoning": reasoning_content, + "reasoning_content": reasoning_content, + } + 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"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..3088438a 100644 --- a/src/eva/assistant/pipeline/audio_llm_processor.py +++ b/src/eva/assistant/pipeline/audio_llm_processor.py @@ -56,9 +56,16 @@ 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 + 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. @@ -85,6 +92,10 @@ def __init__( 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 + # 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) @@ -93,7 +104,7 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: 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 + pre_speech_duration_ms = len(pre_speech_bytes) / (self._frame_sample_rate * 2) * 1000 logger.debug( f"Prepending {len(pre_speech_bytes)} bytes ({pre_speech_duration_ms:.0f}ms) of pre-speech audio" ) @@ -108,13 +119,14 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: await self._user_context_aggregator.push_frame(LLMContextFrame(context=self._context)) 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)) @@ -143,6 +155,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 +190,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 +207,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 +282,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 +294,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 +418,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 +452,14 @@ 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 +478,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, @@ -465,6 +497,7 @@ async def _transcribe_audio(self, audio_data: bytes, timestamp: str, turn_id: in 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: @@ -478,7 +511,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..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 @@ -260,7 +265,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 @@ -283,11 +295,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 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}")