wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Align Ultravox Realtime service with OpenAI/Gemini patterns

Browse Source 
- Add InterruptionFrame handling with stop_all_metrics()
- Add processing metrics (start/stop) at response boundaries
- Fix agent transcript handling for voice and text modalities:
  - Voice mode: push LLMTextFrame (append_to_context=False) and
    TTSTextFrame for deltas, skip duplicated final text
  - Text mode: push LLMTextFrame with proper response lifecycle,
    no TTSTextFrame (downstream TTS handles audio)
- Add output_medium parameter to AgentInputParams and OneShotInputParams
- Improve TTFB measurement using VAD speech end time
- Update example with user turn strategies and transcript events
- Add text-only output example (50a-ultravox-realtime-text.py)
This commit is contained in:
Mark Backman
2026-02-23 17:11:08 -05:00
parent 97b93ebe57
commit 907ff58d41
6 changed files with 384 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
src/pipecat/services/ultravox/llm.py
View File

@@ -31,6 +31,7 @@ from pipecat.frames.frames import (
Frame,
InputAudioRawFrame,
InputTextRawFrame,
InterruptionFrame,
LLMContextFrame,
LLMFullResponseEndFrame,
LLMFullResponseStartFrame,
@@ -42,7 +43,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
TTSTextFrame,
UserAudioRawFrame,
UserStoppedSpeakingFrame,
VADUserStoppedSpeakingFrame,
)
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response import (
@@ -90,6 +91,9 @@ class AgentInputParams(BaseModel):
template_context: Context variables to use when instantiating a call with the
agent. Defaults to an empty dict.
metadata: Metadata to attach to the call. Default to an empty dict.
output_medium: The initial output medium for the agent. Use "text" for text
responses or "voice" for audio responses. Defaults to None, which uses the
agent's default.
max_duration: The maximum duration of the call. Defaults to None, which will
use the agent's default maximum duration.
extra: Extra parameters to include in the agent call creation request. Defaults
@@ -101,6 +105,7 @@ class AgentInputParams(BaseModel):
agent_id: uuid.UUID
template_context: Dict[str, Any] = Field(default_factory=dict)
metadata: Dict[str, str] = Field(default_factory=dict)
output_medium: Optional[Literal["text", "voice"]] = None
max_duration: Optional[datetime.timedelta] = Field(
default=None, ge=datetime.timedelta(seconds=10), le=datetime.timedelta(hours=1)
)
@@ -117,6 +122,8 @@ class OneShotInputParams(BaseModel):
model: Model identifier to use. Defaults to "fixie-ai/ultravox".
voice: Voice identifier for speech generation. Defaults to None.
metadata: Metadata to attach to the call. Default to an empty dict.
output_medium: The initial output medium for the agent. Use "text" for text
responses or "voice" for audio responses. Defaults to None (voice).
max_duration: The maximum duration of the call. Defaults to one hour.
extra: Extra parameters to include in the call creation request. Defaults
to an empty dict. See the Ultravox API documentation for valid arguments:
@@ -129,6 +136,7 @@ class OneShotInputParams(BaseModel):
model: Optional[str] = None
voice: Optional[uuid.UUID] = None
metadata: Dict[str, str] = Field(default_factory=dict)
output_medium: Optional[Literal["text", "voice"]] = None
max_duration: datetime.timedelta = Field(
default=datetime.timedelta(hours=1),
ge=datetime.timedelta(seconds=10),
@@ -210,6 +218,14 @@ class UltravoxRealtimeLLMService(LLMService):
self._sample_rate = 48000
self._resampler = create_stream_resampler()
def can_generate_metrics(self) -> bool:
"""Check if the service can generate usage metrics.
Returns:
True if metrics generation is supported.
"""
return True
#
# standard AIService frame handling
#
@@ -237,6 +253,14 @@ class UltravoxRealtimeLLMService(LLMService):
except Exception as e:
await self.push_error("Failed to connect to Ultravox", e, fatal=True)
@staticmethod
def _output_medium_to_api(medium: Optional[Literal["text", "voice"]]) -> Optional[str]:
if medium == "text":
return "MESSAGE_MEDIUM_TEXT"
elif medium == "voice":
return "MESSAGE_MEDIUM_VOICE"
return None
async def _start_agent_call(self, params: AgentInputParams) -> str:
request_body = {
"templateContext": params.template_context,
@@ -247,6 +271,9 @@ class UltravoxRealtimeLLMService(LLMService):
}
},
}
initial_output_medium = self._output_medium_to_api(params.output_medium)
if initial_output_medium:
request_body["initialOutputMedium"] = initial_output_medium
if params.max_duration:
request_body["maxDuration"] = f"{params.max_duration.total_seconds():3f}s"
request_body = request_body | params.extra
@@ -277,7 +304,11 @@ class UltravoxRealtimeLLMService(LLMService):
"inputSampleRate": self._sample_rate,
}
},
} | params.extra
}
initial_output_medium = self._output_medium_to_api(params.output_medium)
if initial_output_medium:
request_body["initialOutputMedium"] = initial_output_medium
request_body = request_body | params.extra
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.ultravox.ai/api/calls",
@@ -367,18 +398,17 @@ class UltravoxRealtimeLLMService(LLMService):
else LLMContext.from_openai_context(frame.context)
)
await self._handle_context(context)
elif isinstance(frame, InterruptionFrame):
await self.stop_all_metrics()
await self.push_frame(frame, direction)
elif isinstance(frame, InputTextRawFrame):
await self._send_user_text(frame.text)
await self.push_frame(frame, direction)
elif isinstance(frame, InputAudioRawFrame):
await self._send_user_audio(frame)
await self.push_frame(frame, direction)
elif isinstance(frame, UserStoppedSpeakingFrame):
# This may or may not align with Ultravox's end of user speech detection,
# which relies on a more complex endpointing model. In particular it will
# yield a seemingly very slow TTFB in the case of endpointing false
# negatives. It will be close in the majority of cases though.
await self.start_ttfb_metrics()
elif isinstance(frame, VADUserStoppedSpeakingFrame):
await self._handle_vad_user_stopped_speaking(frame)
await self.push_frame(frame, direction)
else:
await self.push_frame(frame, direction)
@@ -399,6 +429,25 @@ class UltravoxRealtimeLLMService(LLMService):
}
await self._send(socket_message)
async def _handle_vad_user_stopped_speaking(self, frame: VADUserStoppedSpeakingFrame):
"""Handle VAD user stopped speaking frame.
Calculates the actual speech end time and starts a timeout task to wait
for the final transcription before reporting TTFB.
Args:
frame: The VAD user stopped speaking frame.
"""
# Skip TTFB measurement if stop_secs is not set
if frame.stop_secs == 0.0:
return
# Calculate the actual speech end time (current time minus VAD stop delay).
# This approximates when the last user audio was sent to the Ultravox service,
# which we use to measure against the eventual transcription response.
speech_end_time = frame.timestamp - frame.stop_secs
await self.start_ttfb_metrics(start_time=speech_end_time)
async def _send_user_audio(self, frame: InputAudioRawFrame):
"""Send user audio frame to Ultravox Realtime."""
if not self._socket:
@@ -502,6 +551,7 @@ class UltravoxRealtimeLLMService(LLMService):
if not audio:
return
if not self._bot_responding:
await self.start_processing_metrics()
await self.stop_ttfb_metrics()
await self.push_frame(LLMFullResponseStartFrame())
await self.push_frame(TTSStartedFrame())
@@ -509,6 +559,7 @@ class UltravoxRealtimeLLMService(LLMService):
await self.push_frame(TTSAudioRawFrame(audio, self._sample_rate, 1))
async def _handle_response_end(self):
await self.stop_processing_metrics()
if self._bot_responding == "voice":
await self.push_frame(TTSStoppedFrame())
await self.push_frame(LLMFullResponseEndFrame())
@@ -542,22 +593,29 @@ class UltravoxRealtimeLLMService(LLMService):
async def _handle_agent_transcript(
self, medium: str, text: Optional[str], delta: Optional[str], final: bool
):
if text or delta:
frame = LLMTextFrame(text=text or delta)
frame.skip_tts = medium == "voice"
await self.push_frame(frame)
if medium == "text":
if text:
await self.stop_ttfb_metrics()
await self.push_frame(LLMFullResponseStartFrame())
await self.push_frame(TTSStartedFrame())
await self.push_frame(TTSTextFrame(text=text, aggregated_by=AggregationType.WORD))
self._bot_responding = "text"
elif final:
if medium == "voice":
# In voice mode, audio is handled by _handle_audio(). Here we push
# text transcripts of the audio for downstream consumers.
if (text or delta) and not final:
frame = LLMTextFrame(text=text or delta)
frame.append_to_context = False
await self.push_frame(frame)
if delta:
tts_frame = TTSTextFrame(text=delta, aggregated_by=AggregationType.WORD)
tts_frame.includes_inter_frame_spaces = True
await self.push_frame(tts_frame)
elif medium == "text":
if final:
await self.stop_processing_metrics()
await self.push_frame(LLMFullResponseEndFrame())
self._bot_responding = None
elif delta:
await self.push_frame(TTSTextFrame(text=delta, aggregated_by=AggregationType.WORD))
elif text or delta:
if not self._bot_responding:
await self.start_processing_metrics()
await self.stop_ttfb_metrics()
await self.push_frame(LLMFullResponseStartFrame())
self._bot_responding = "text"
await self.push_frame(LLMTextFrame(text=text or delta))
def create_context_aggregator(
self,