From a6a64e76457aeb69f46d92145bc4025598142111 Mon Sep 17 00:00:00 2001 From: Katrina Date: Fri, 19 Jun 2026 16:55:56 -0400 Subject: [PATCH 1/3] fix faithfulness prompt to use full transcript --- src/eva/metrics/processor.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/src/eva/metrics/processor.py b/src/eva/metrics/processor.py index 30cd909b..c404af01 100644 --- a/src/eva/metrics/processor.py +++ b/src/eva/metrics/processor.py @@ -899,6 +899,36 @@ def _load_pipecat_logs(pipecat_logs_path: str) -> list[dict]: and " " in e.get("data", {}).get("frame", "") ) ] + # Second pass: drop any multi-word tts_text entry that is an exact preview of + # the immediately following single-word token stream. Catches batch-preview + # phrases emitted ~1ms before their per-word stream (different timestamp from + # the per-word tokens, so the same-timestamp filter above misses them). Without + # this, the preview phrase and its word-by-word reconstruction both survive and + # get joined into one aggregated segment, creating a duplicate that breaks + # truncate_to_spoken substring matching for any audit text that follows it. + deduped: list[dict] = [] + i = 0 + while i < len(raw_pipecat): + e = raw_pipecat[i] + if e.get("type") == "tts_text" and " " in e.get("data", {}).get("frame", ""): + phrase_words = e["data"]["frame"].split() + nw = len(phrase_words) + j, following_tts = i + 1, [] + while j < len(raw_pipecat) and len(following_tts) < nw: + if raw_pipecat[j].get("type") != "tts_text": + break # stop at any non-tts boundary (turn_start/end) + following_tts.append(raw_pipecat[j]) + j += 1 + if ( + len(following_tts) == nw + and all(" " not in fw.get("data", {}).get("frame", "") for fw in following_tts) + and " ".join(fw["data"]["frame"] for fw in following_tts) == e["data"]["frame"] + ): + i += 1 # skip: this multi-word entry is a preview of the following words + continue + deduped.append(e) + i += 1 + raw_pipecat = deduped grouped_pipecat = aggregate_pipecat_logs_by_type(raw_pipecat) for entry in grouped_pipecat: From 932776a68a6825401adeb95232840a70f332e46d Mon Sep 17 00:00:00 2001 From: Katrina Date: Fri, 19 Jun 2026 17:27:46 -0400 Subject: [PATCH 2/3] add to realtime models as well --- src/eva/assistant/base_server.py | 14 ++++++++++++++ src/eva/assistant/elevenlabs_server.py | 10 +--------- src/eva/assistant/gemini_live_server.py | 11 ++--------- src/eva/assistant/openai_realtime_server.py | 12 ++++-------- 4 files changed, 21 insertions(+), 26 deletions(-) diff --git a/src/eva/assistant/base_server.py b/src/eva/assistant/base_server.py index 74645c18..0f8ee86e 100644 --- a/src/eva/assistant/base_server.py +++ b/src/eva/assistant/base_server.py @@ -23,6 +23,7 @@ from eva.utils.audio_utils import save_pcm_as_wav from eva.utils.culture import get_initial_message from eva.utils.logging import get_logger +from eva.utils.prompt_manager import PromptManager logger = get_logger(__name__) @@ -319,6 +320,19 @@ def _save_audio_deferred( if mixed_audio or user_audio or assistant_audio: logger.info(f"Saved audio files to {self.output_dir} ({len(mixed_audio)} bytes mixed)") + def _build_system_prompt(self) -> str: + """Build the system prompt for realtime/S2S assistant servers.""" + prompt_manager = PromptManager() + prompt = prompt_manager.get_prompt( + "realtime_agent.system_prompt", + agent_personality=self.agent.description, + agent_instructions=self.agent.instructions, + datetime=self.current_date_time, + ) + if self.pipeline_config.pre_tool_speech == "auto": + prompt += "\n\n" + prompt_manager.get_prompt("agent.pre_tool_speech") + return prompt + def _save_scenario_dbs(self) -> None: """Save initial and final scenario database states.""" try: diff --git a/src/eva/assistant/elevenlabs_server.py b/src/eva/assistant/elevenlabs_server.py index bb5c30bb..eb5b6493 100644 --- a/src/eva/assistant/elevenlabs_server.py +++ b/src/eva/assistant/elevenlabs_server.py @@ -45,7 +45,6 @@ from eva.models.agents import AgentConfig from eva.models.config import ModelConfig from eva.utils.logging import get_logger -from eva.utils.prompt_manager import PromptManager logger = get_logger(__name__) @@ -156,14 +155,7 @@ def __init__( self.s2s_params = s2s_params self._model = s2s_params.get("model", "elevenlabs") - # Build system prompt - prompt_manager = PromptManager() - self._system_prompt = prompt_manager.get_prompt( - "realtime_agent.system_prompt", - agent_personality=agent.description, - agent_instructions=agent.instructions, - datetime=self.current_date_time, - ) + self._system_prompt = self._build_system_prompt() # Build ElevenLabs client tools from agent config self._client_tools = _agent_tools_to_client_tools(agent, self.execute_tool) diff --git a/src/eva/assistant/gemini_live_server.py b/src/eva/assistant/gemini_live_server.py index 7e9ce5b6..b07fff28 100644 --- a/src/eva/assistant/gemini_live_server.py +++ b/src/eva/assistant/gemini_live_server.py @@ -40,7 +40,6 @@ from eva.models.agents import AgentConfig from eva.models.config import ModelConfig from eva.utils.logging import get_logger -from eva.utils.prompt_manager import PromptManager logger = get_logger(__name__) @@ -186,14 +185,7 @@ def __init__( self._language_code = s2s_params.get("language_code") or self.language self._api_key = s2s_params.get("api_key", "") - # Build system prompt (same pattern as pipecat realtime) - prompt_manager = PromptManager() - self._system_prompt = prompt_manager.get_prompt( - "realtime_agent.system_prompt", - agent_personality=agent.description, - agent_instructions=agent.instructions, - datetime=self.current_date_time, - ) + self._system_prompt = self._build_system_prompt() # Build Gemini tools self._gemini_tools = _agent_tools_to_gemini(agent) @@ -608,6 +600,7 @@ async def _process_gemini_events() -> None: id=fc.id, name=fc.name, response=result, + scheduling=types.FunctionResponseScheduling.WHEN_IDLE, ) ] ) diff --git a/src/eva/assistant/openai_realtime_server.py b/src/eva/assistant/openai_realtime_server.py index 73f96fdc..3d368900 100644 --- a/src/eva/assistant/openai_realtime_server.py +++ b/src/eva/assistant/openai_realtime_server.py @@ -28,7 +28,6 @@ ) from eva.assistant.base_server import AbstractAssistantServer from eva.utils.logging import get_logger -from eva.utils.prompt_manager import PromptManager logger = get_logger(__name__) @@ -84,13 +83,7 @@ def __init__(self, **kwargs: Any) -> None: self._audio_sample_rate = OPENAI_SAMPLE_RATE - prompt_manager = PromptManager() - self._system_prompt: str = prompt_manager.get_prompt( - "realtime_agent.system_prompt", - agent_personality=self.agent.description, - agent_instructions=self.agent.instructions, - datetime=self.current_date_time, - ) + self._system_prompt: str = self._build_system_prompt() self._realtime_tools: list[dict] = self._build_realtime_tools() @@ -218,6 +211,9 @@ def _build_session_config(self) -> dict[str, Any]: if reasoning_effort: session_config["reasoning"] = {"effort": reasoning_effort} + if self.pipeline_config.parallel_tool_calls is not None: + session_config["parallel_tool_calls"] = self.pipeline_config.parallel_tool_calls + return session_config def _create_client(self) -> AsyncOpenAI: From 03e984dba9750f190b9eb254489b50e52427a7c2 Mon Sep 17 00:00:00 2001 From: akshaykalkunte Date: Wed, 1 Jul 2026 03:04:22 +0000 Subject: [PATCH 3/3] Changes for audio-native --- configs/prompts/simulation.yaml | 9 ++ src/eva/assistant/agentic/audio_llm_system.py | 32 ++--- src/eva/assistant/pipecat_server.py | 14 +- src/eva/assistant/pipeline/alm_vllm.py | 96 +++++++++++++- .../assistant/pipeline/audio_llm_processor.py | 123 ++++++++++++++---- src/eva/assistant/pipeline/services.py | 1 + 6 files changed, 216 insertions(+), 59 deletions(-) diff --git a/configs/prompts/simulation.yaml b/configs/prompts/simulation.yaml index 45a44209..b11f4172 100644 --- a/configs/prompts/simulation.yaml +++ b/configs/prompts/simulation.yaml @@ -64,6 +64,9 @@ 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 +130,9 @@ audio_llm_agent: - Use only information from the current conversation - Ask for clarification only when truly necessary - Request one or two details maximum per turn + - Never re-ask for information the caller has already provided in this conversation. Before asking for anything, check the conversation history — if you already have the value, use it. + - If a tool call fails or returns no result, do not retry with the same parameters and do not ask the caller to provide the same information from scratch. Instead, read back exactly what you used ("I tried looking up confirmation number A B C one two three for Smith — I'm not finding anything. Did I get that right?"), and if the caller corrects it, retry once with the corrected value. If you still cannot find a match, ask the user to spell it out for you letter by letter. + realtime_agent: system_prompt: | @@ -187,6 +193,9 @@ realtime_agent: - Use only information from the current conversation - Ask for clarification only when truly necessary - Request one or two details maximum per turn + - Never re-ask for information the caller has already provided in this conversation. Before asking for anything, check the conversation history — if you already have the value, use it. + - If a tool call fails or returns no result, do not retry with the same parameters and do not ask the caller to provide the same information from scratch. Instead, read back exactly what you used ("I tried looking up confirmation number A B C one two three for Smith — I'm not finding anything. Did I get that right?"), and if the caller corrects it, retry once with the corrected value. If you still cannot find a match, ask the user to spell it out for you letter by letter. + # ============================================== # User Simulator Prompts diff --git a/src/eva/assistant/agentic/audio_llm_system.py b/src/eva/assistant/agentic/audio_llm_system.py index 14ae92c0..6647affa 100644 --- a/src/eva/assistant/agentic/audio_llm_system.py +++ b/src/eva/assistant/agentic/audio_llm_system.py @@ -91,12 +91,10 @@ async def process_query_with_audio(self, user_text: str) -> AsyncGenerator[str, yield response async def _execute_agent_with_audio(self, agent: AgentConfig) -> AsyncGenerator[str, None]: - """Build messages with audio on the last user message only, then run tool loop. + """Build messages with audio on all user turns, then run tool loop. - Only the current (last) user turn is sent as audio. Previous user turns - remain as text (transcriptions updated via the parallel transcription - pipeline). This keeps context manageable while giving the model the - actual audio for the current turn. + All user turns are sent as audio so the model has full conversational + context without relying on potentially inaccurate transcriptions. """ messages: list[dict[str, Any]] = [ {"role": "system", "content": self.system_prompt}, @@ -106,22 +104,16 @@ async def _execute_agent_with_audio(self, agent: AgentConfig) -> AsyncGenerator[ conversation_history = self.audit_log.get_conversation_messages(max_messages=30) history_dicts = [msg.to_dict() for msg in conversation_history] - # Replace only the LAST user message with audio (current turn) + # Replace ALL user messages with their corresponding audio if self._turn_audio_history: - # Find the last user message index - last_user_idx = None - for i in range(len(history_dicts) - 1, -1, -1): - if history_dicts[i].get("role") == "user": - last_user_idx = i - break - - if last_user_idx is not None: - # Use the most recent audio for the last user message - audio_bytes, sample_rate = self._turn_audio_history[-1] - history_dicts[last_user_idx] = self.alm_client.build_audio_user_message( - audio_bytes=audio_bytes, - source_sample_rate=sample_rate, - ) + user_indices = [i for i, msg in enumerate(history_dicts) if msg.get("role") == "user"] + for turn_idx, msg_idx in enumerate(user_indices): + if turn_idx < len(self._turn_audio_history): + audio_bytes, sample_rate = self._turn_audio_history[turn_idx] + history_dicts[msg_idx] = self.alm_client.build_audio_user_message( + audio_bytes=audio_bytes, + source_sample_rate=sample_rate, + ) messages.extend(history_dicts) diff --git a/src/eva/assistant/pipecat_server.py b/src/eva/assistant/pipecat_server.py index 31608761..f64f63d4 100644 --- a/src/eva/assistant/pipecat_server.py +++ b/src/eva/assistant/pipecat_server.py @@ -381,7 +381,6 @@ async def _realtime_tool_handler(params) -> None: input_transcription_processor = AudioTranscriptionProcessor( audio_collector=audio_llm_audio_collector, alm_client=alm_client, - sample_rate=SAMPLE_RATE, ) # Set callback to save user transcription to transcript.jsonl and update audit log @@ -474,6 +473,7 @@ async def on_user_transcription(text: str, timestamp: str, turn_id: int | None) assistant_aggregator=assistant_aggregator, agent_processor=agent_processor, audio_llm_processor=audio_llm_processor, + audio_llm_audio_collector=audio_llm_audio_collector, input_transcription_processor=input_transcription_processor, ) @@ -626,6 +626,7 @@ def _setup_event_handlers( assistant_aggregator, agent_processor, audio_llm_processor=None, + audio_llm_audio_collector=None, input_transcription_processor=None, ) -> None: """Setup event handlers for the pipeline.""" @@ -697,10 +698,13 @@ async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMes ) await agent_processor.process_complete_user_turn(message.content) elif self.pipeline_config.pipeline_type == PipelineType.AUDIO_LLM and audio_llm_processor: - # No STT → message.content is empty. - # Processing is triggered by LLMContextFrame flow through ParallelPipeline - # (AudioLLMUserAudioCollector pushes LLMContextFrame on UserStoppedSpeakingFrame) - pass + # No STT → message.content is empty. Finalize the turn now that the + # smart-turn-aware turn-stop strategy has fired. The collector kept + # appending audio across VAD blips up to this point; notify_turn_ended() + # increments the turn id and pushes LLMContextFrame to trigger the + # transcription and audio-LLM branches. + assert audio_llm_audio_collector is not None + await audio_llm_audio_collector.notify_turn_ended() elif self.non_instrumented_realtime_llm: # Non-instrumented realtime fallback (e.g. Ultravox) if message.content: diff --git a/src/eva/assistant/pipeline/alm_vllm.py b/src/eva/assistant/pipeline/alm_vllm.py index 92891442..9e0b1762 100644 --- a/src/eva/assistant/pipeline/alm_vllm.py +++ b/src/eva/assistant/pipeline/alm_vllm.py @@ -21,6 +21,45 @@ logger = get_logger(__name__) +# vLLM / OpenAI-compatible decoding params that are safe to forward via extra_body. +# Unknown keys are still forwarded (vLLM evolves) but warned about, to catch typos. +_KNOWN_SAMPLING_PARAMS = frozenset( + { + "top_p", + "top_k", + "min_p", + "repetition_penalty", + "frequency_penalty", + "presence_penalty", + "length_penalty", + "seed", + "stop", + "stop_token_ids", + "min_tokens", + "ignore_eos", + "skip_special_tokens", + "spaces_between_special_tokens", + } +) + +# Keys the client controls itself — must not be set via sampling_params, or they would +# silently override the managed request (e.g. temperature/max_tokens land in extra_body +# and clobber the top-level args; model/messages/tools break the call entirely). +_RESERVED_SAMPLING_PARAMS = frozenset( + { + "model", + "messages", + "temperature", + "max_tokens", + "tools", + "tool_choice", + "stream", + "n", + "extra_body", + "chat_template_kwargs", + } +) + class ALMvLLMClient(BaseALMClient): """Client for self-hosted audio language model via vLLM's OpenAI-compatible HTTP API.""" @@ -39,6 +78,7 @@ def __init__( sample_width: int = DEFAULT_SAMPLE_WIDTH, language: str | None = None, enable_thinking: bool = False, + sampling_params: dict[str, Any] | None = None, ): super().__init__( model=model, @@ -53,6 +93,11 @@ def __init__( ) self._reasoning_token_fallback_warned = False self.enable_thinking = enable_thinking + # vLLM-specific decoding params (repetition_penalty, top_p, top_k, min_p, + # frequency_penalty, presence_penalty, ...). Validated here, then merged into + # extra_body of every complete() call; not applied to transcribe(), which is + # kept deterministic. + self.sampling_params: dict[str, Any] = self._validate_sampling_params(sampling_params) # Normalize base_url: ensure it ends with /v1 for the OpenAI client self.base_url = base_url.rstrip("/") if not self.base_url.endswith("/v1"): @@ -67,9 +112,44 @@ def __init__( logger.info( f"Initialized ALMvLLMClient: base_url={self.base_url}, model={self.model}, " f"sample_rate={self.sample_rate}, num_channels={self.num_channels}, " - f"sample_width={self.sample_width}" + f"sample_width={self.sample_width}, " + f"sampling_params={self.sampling_params}, enable_thinking={self.enable_thinking}" ) + @staticmethod + def _validate_sampling_params(sampling_params: dict[str, Any] | None) -> dict[str, Any]: + """Validate user-supplied vLLM sampling params before they reach extra_body. + + Rejects keys the client manages itself (these would silently clobber the + request — e.g. a ``temperature`` here overrides the top-level arg via + extra_body). Warns on keys outside the known vLLM set so typos are visible, + but still forwards them since vLLM's parameter surface evolves. + """ + if sampling_params is None: + return {} + if not isinstance(sampling_params, dict): + raise ValueError(f"sampling_params must be a dict, got {type(sampling_params).__name__}") + + reserved = _RESERVED_SAMPLING_PARAMS & sampling_params.keys() + if reserved: + raise ValueError( + f"sampling_params may not override client-managed keys {sorted(reserved)}. " + "Set temperature/max_tokens via their dedicated params; model/messages/tools " + "are controlled by the client." + ) + + non_string_keys = [k for k in sampling_params if not isinstance(k, str)] + if non_string_keys: + raise ValueError(f"sampling_params keys must be strings, got non-string keys: {non_string_keys}") + + unknown = sampling_params.keys() - _KNOWN_SAMPLING_PARAMS + if unknown: + logger.warning( + f"sampling_params contains keys not in the known vLLM set {sorted(unknown)}; " + "forwarding to vLLM as-is — check for typos." + ) + return dict(sampling_params) + def _audio_content_part(self, audio_b64: str) -> dict[str, Any]: return { "type": "audio_url", @@ -89,16 +169,20 @@ async def complete( When tool_calls are present, returns the full message object. Otherwise returns the content string. """ + extra_body: dict[str, Any] = { + "chat_template_kwargs": {"enable_thinking": self.enable_thinking}, + } + # Merge vLLM sampling params (repetition_penalty, top_p, etc.). These ride + # alongside chat_template_kwargs; vLLM accepts non-OpenAI keys at the top + # level of extra_body. + extra_body.update(self.sampling_params) + kwargs: dict[str, Any] = { "model": self.model, "messages": messages, "temperature": self.temperature, "max_tokens": self.max_tokens, - "extra_body": { - "chat_template_kwargs": { - "enable_thinking": self.enable_thinking, - } - }, + "extra_body": extra_body, } if tools: kwargs["tools"] = tools diff --git a/src/eva/assistant/pipeline/audio_llm_processor.py b/src/eva/assistant/pipeline/audio_llm_processor.py index e7b23cb3..2df9efa4 100644 --- a/src/eva/assistant/pipeline/audio_llm_processor.py +++ b/src/eva/assistant/pipeline/audio_llm_processor.py @@ -56,12 +56,22 @@ class AudioLLMUserAudioCollector(FrameProcessor): """Buffers raw audio frames during user speech for the audio-LLM pipeline. - Collects audio and pushes LLMContextFrame when user stops speaking, which - triggers the parallel pipeline (transcription + audio-LLM processing). - - Uses a ring buffer of pre-VAD audio so that the beginning of the user's - speech—before VAD fires UserStartedSpeakingFrame—is not lost. This mirrors - the S2S UserAudioCollector pattern. + Audio capture is driven by raw VAD events (UserStartedSpeakingFrame / + UserStoppedSpeakingFrame), but turn finalization is driven by the + smart-turn-aware user aggregator: pipecat_server calls ``notify_turn_ended()`` + from the ``on_user_turn_stopped`` event (after the turn-stop strategy is + satisfied), which increments the turn id and pushes an LLMContextFrame to + trigger both branches of the parallel pipeline (transcription + audio-LLM). + + Because finalization is deferred to the smart-turn verdict, a single turn may + span multiple VAD start/stop blips (e.g. natural pauses while spelling out a + code). The collector keeps appending audio across those blips instead of + fragmenting the buffer, and only starts a fresh buffer once the previous turn + has been finalized. + + A ring buffer of pre-VAD audio captures the start of speech that occurs before + VAD fires UserStartedSpeakingFrame. This mirrors the S2S UserAudioCollector + pattern. All frames pass through unchanged. """ @@ -82,45 +92,82 @@ def __init__( self._audio_buffer = bytearray() self._pre_speech_buffer: list[bytes] = [] self._user_speaking = False - self._current_turn_id = 0 # Incremented on each user turn + self._current_turn_id = 0 # Incremented on each finalized turn + # True between turns. Set False when a new turn starts and back to True by + # notify_turn_ended(), so audio is appended continuously across VAD start/stop + # blips within a single smart-turn-governed turn instead of being fragmented. + self._turn_finalized = True # Pre-speech buffer size (captures audio before VAD fires to avoid cutting off speech) self._pre_speech_secs = pre_speech_secs or self.DEFAULT_PRE_SPEECH_SECS + # Actual sample rate observed on InputAudioRawFrames. The buffer holds raw PCM + # at whatever rate the transport delivered; falls back to PIPELINE_SAMPLE_RATE + # until the first audio frame arrives. + self._frame_sample_rate: int = PIPELINE_SAMPLE_RATE async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: await super().process_frame(frame, direction) if isinstance(frame, UserStartedSpeakingFrame): self._user_speaking = True - # Prepend the pre-speech ring buffer so we don't lose the start of speech pre_speech_bytes = b"".join(self._pre_speech_buffer) - pre_speech_duration_ms = len(pre_speech_bytes) / (PIPELINE_SAMPLE_RATE * 2) * 1000 - logger.debug( - f"Prepending {len(pre_speech_bytes)} bytes ({pre_speech_duration_ms:.0f}ms) of pre-speech audio" - ) - self._audio_buffer = bytearray(pre_speech_bytes) + pre_speech_duration_ms = len(pre_speech_bytes) / (self._frame_sample_rate * 2) * 1000 + if self._turn_finalized: + # Fresh turn: pre-speech becomes the start of a new buffer. + logger.debug( + f"New turn — prepending {len(pre_speech_bytes)} bytes " + f"({pre_speech_duration_ms:.0f}ms) of pre-speech audio" + ) + self._audio_buffer = bytearray(pre_speech_bytes) + self._turn_finalized = False + else: + # Mid-turn VAD blip (e.g. a pause between spelled-out characters): + # append pre-speech so the captured utterance stays continuous. + logger.debug( + f"Resuming turn — appending {len(pre_speech_bytes)} bytes " + f"({pre_speech_duration_ms:.0f}ms) of pre-speech audio" + ) + self._audio_buffer.extend(pre_speech_bytes) self._pre_speech_buffer.clear() elif isinstance(frame, UserStoppedSpeakingFrame): + # Stop buffering active speech, but do NOT finalize the turn here — the + # smart-turn-aware on_user_turn_stopped event drives finalization via + # notify_turn_ended(). This lets a turn span multiple VAD segments. self._user_speaking = False - # Increment turn ID BEFORE pushing frame so both parallel branches see the same ID - self._current_turn_id += 1 - # Push LLMContextFrame to trigger parallel pipeline - await self._user_context_aggregator.push_frame(LLMContextFrame(context=self._context)) + logger.debug( + f"VAD stop — pausing capture (buffer={len(self._audio_buffer)} bytes, " + f"turn_id={self._current_turn_id})" + ) elif isinstance(frame, InputAudioRawFrame): + self._frame_sample_rate = frame.sample_rate if self._user_speaking: self._audio_buffer.extend(frame.audio) else: # Ring buffer: keep a rolling window of pre-speech audio self._pre_speech_buffer.append(frame.audio) # 16-bit mono → 2 bytes per sample - max_bytes = int(self._pre_speech_secs * PIPELINE_SAMPLE_RATE * 2) + max_bytes = int(self._pre_speech_secs * self._frame_sample_rate * 2) total = sum(len(chunk) for chunk in self._pre_speech_buffer) while total > max_bytes and self._pre_speech_buffer: total -= len(self._pre_speech_buffer.pop(0)) await self.push_frame(frame, direction) + async def notify_turn_ended(self) -> None: + """Finalize the current turn and trigger downstream processing. + + Called from pipecat_server's ``on_user_turn_stopped`` handler once the + smart-turn-aware turn-stop strategy is satisfied. Increments the turn id + (so the transcription branch and the audio-LLM branch observe a consistent + boundary) and pushes LLMContextFrame to trigger both branches. Setting + ``_turn_finalized`` makes the next UserStartedSpeakingFrame begin a fresh + buffer instead of appending to this turn. + """ + self._current_turn_id += 1 + self._turn_finalized = True + await self._user_context_aggregator.push_frame(LLMContextFrame(context=self._context)) + def get_buffered_audio(self) -> bytes: """Get the buffered audio and clear the buffer.""" audio = bytes(self._audio_buffer) @@ -143,6 +190,15 @@ def current_turn_id(self) -> int: """Get the current turn ID for associating transcriptions with entries.""" return self._current_turn_id + @property + def frame_sample_rate(self) -> int: + """Sample rate observed on the most recent InputAudioRawFrame. + + The audio buffer holds raw PCM at whatever rate the transport delivered; + downstream consumers should use this rather than assuming PIPELINE_SAMPLE_RATE. + """ + return self._frame_sample_rate + class AudioLLMProcessor(FrameProcessor): """Processes complete user turns using the audio-LLM model. @@ -257,7 +313,10 @@ async def process_complete_user_turn(self, text_from_aggregator: str) -> None: turn_id = self.audio_collector.current_turn_id self.audit_log.append_user_input(self._USER_PLACEHOLDER, turn_id=turn_id) - self._current_query_task = asyncio.create_task(self._process_audio_turn(audio_bytes)) + source_sample_rate = self.audio_collector.frame_sample_rate + self._current_query_task = asyncio.create_task( + self._process_audio_turn(audio_bytes, source_sample_rate) + ) try: await self._current_query_task except asyncio.CancelledError: @@ -268,11 +327,11 @@ async def process_complete_user_turn(self, text_from_aggregator: str) -> None: # Placeholder used in audit_log / transcript since no real transcription is available _USER_PLACEHOLDER = "[user audio]" - async def _process_audio_turn(self, audio_bytes: bytes) -> None: + async def _process_audio_turn(self, audio_bytes: bytes, source_sample_rate: int) -> None: """Process a user turn with audio data.""" try: # Send audio to the agentic system and process - self.agentic_system.set_turn_audio(audio_bytes, PIPELINE_SAMPLE_RATE) + self.agentic_system.set_turn_audio(audio_bytes, source_sample_rate) async for response in self.agentic_system.process_query_with_audio(self._USER_PLACEHOLDER): if self._interrupted.is_set(): @@ -392,14 +451,12 @@ def __init__( audio_collector: AudioLLMUserAudioCollector, alm_client: BaseALMClient, system_prompt: str | None = None, - sample_rate: int = PIPELINE_SAMPLE_RATE, **kwargs, ): super().__init__(**kwargs) self._audio_collector = audio_collector self._alm_client = alm_client self._system_prompt = system_prompt or alm_client.default_transcription_prompt - self._sample_rate = sample_rate # Callback for when transcription is ready (set by pipecat_server.py) self.on_transcription: Any | None = None @@ -428,13 +485,16 @@ async def process_frame(self, frame: Frame, direction: FrameDirection) -> None: # Capture turn_id and audio at the moment we receive the frame turn_id = self._audio_collector.current_turn_id - # Capture audio NOW before it gets overwritten by the next turn + # Capture audio + sample rate NOW before they get overwritten by the next turn audio_data = self._audio_collector.peek_buffered_audio() + source_sample_rate = self._audio_collector.frame_sample_rate logger.info(f"transcribe (turn_id={turn_id})") timestamp = time_now_iso8601() # Run transcription as background task so it completes even if interrupted - task = asyncio.create_task(self._transcribe_audio(audio_data, timestamp, turn_id=turn_id)) + task = asyncio.create_task( + self._transcribe_audio(audio_data, timestamp, source_sample_rate, turn_id=turn_id) + ) self._transcription_tasks.append(task) # Clean up completed tasks self._transcription_tasks = [t for t in self._transcription_tasks if not t.done()] @@ -453,9 +513,16 @@ async def transcribe(self, timestamp: str, turn_id: int | None = None) -> str | The transcription text, or None if transcription failed or audio was empty. """ audio_data = self._audio_collector.peek_buffered_audio() - return await self._transcribe_audio(audio_data, timestamp, turn_id) + source_sample_rate = self._audio_collector.frame_sample_rate + return await self._transcribe_audio(audio_data, timestamp, source_sample_rate, turn_id) - async def _transcribe_audio(self, audio_data: bytes, timestamp: str, turn_id: int | None = None) -> str | None: + async def _transcribe_audio( + self, + audio_data: bytes, + timestamp: str, + source_sample_rate: int, + turn_id: int | None = None, + ) -> str | None: """Transcribe pre-captured audio data using chat completions. This method takes audio data directly instead of reading from the collector, @@ -478,7 +545,7 @@ async def _transcribe_audio(self, audio_data: bytes, timestamp: str, turn_id: in start_time = time.time() text = await self._alm_client.transcribe( audio_bytes=audio_data, - source_sample_rate=self._sample_rate, + source_sample_rate=source_sample_rate, system_prompt=self._system_prompt, ) elapsed = time.time() - start_time diff --git a/src/eva/assistant/pipeline/services.py b/src/eva/assistant/pipeline/services.py index 120cbbe0..116e9b34 100644 --- a/src/eva/assistant/pipeline/services.py +++ b/src/eva/assistant/pipeline/services.py @@ -782,6 +782,7 @@ def create_audio_llm_client( sample_width=params.get("sample_width", 2), language=language, enable_thinking=params.get("enable_thinking", False), + sampling_params=params.get("sampling_params"), ) logger.info(f"Using {model} vLLM audio-LLM: {base_url}") return client