apunkt/dograh
1
0
Fork 0
mirror of https://github.com/dograh-hq/dograh.git synced 2026-07-25 12:01:04 +02:00
Code Issues Projects Releases Packages Wiki Activity

chore: generate SDK and fix other realtime providers

Browse source
This commit is contained in:
Abhishek Kumar 2026-07-15 17:56:35 +05:30
parent 51525b7e24
commit d66eb8fd47
20 changed files with 535 additions and 473 deletions
Download patch file Download diff file Expand all files Collapse all files

7
api/services/configuration/options/azure.py
View file

@ -1,6 +1,10 @@
AZURE_MODELS = ["gpt-4.1-mini"]
AZURE_REALTIME_MODELS = ["gpt-4o-realtime-preview"]
AZURE_REALTIME_MODELS = [
"gpt-realtime",
"gpt-realtime-1.5",
"gpt-realtime-mini",
]
AZURE_REALTIME_VOICES = [
"alloy",
"ash",
@ -12,6 +16,7 @@ AZURE_REALTIME_VOICES = [
"verse",
]
AZURE_REALTIME_API_VERSIONS = [
"v1",
"2025-04-01-preview",
"2024-10-01-preview",
"2024-12-17",

13
api/services/configuration/registry.py
View file

@ -621,7 +621,7 @@ class OpenAIRealtimeLLMConfiguration(BaseLLMConfiguration):
GROK_REALTIME_MODELS = ["grok-voice-think-fast-1.0"]
GROK_REALTIME_VOICES = ["Ara", "Rex", "Sal", "Eve", "Leo"]
GROK_REALTIME_VOICES = ["ara", "rex", "sal", "eve", "leo"]
ULTRAVOX_REALTIME_MODELS = ["ultravox-v0.7", "fixie-ai/ultravox"]
@ -638,7 +638,7 @@ class GrokRealtimeLLMConfiguration(BaseLLMConfiguration):
},
)
voice: str = Field(
default="Ara",
default="ara",
description="Voice the model speaks in.",
json_schema_extra={
"examples": GROK_REALTIME_VOICES,
@ -756,7 +756,7 @@ class AzureRealtimeLLMConfiguration(BaseLLMConfiguration):
model_config = AZURE_REALTIME_PROVIDER_MODEL_CONFIG
provider: Literal[ServiceProviders.AZURE_REALTIME] = ServiceProviders.AZURE_REALTIME
model: str = Field(
default="gpt-4o-realtime-preview",
default="gpt-realtime",
description="Azure OpenAI realtime deployment name.",
json_schema_extra={
"examples": AZURE_REALTIME_MODELS,
@ -775,8 +775,11 @@ class AzureRealtimeLLMConfiguration(BaseLLMConfiguration):
},
)
api_version: str = Field(
default="2025-04-01-preview",
description="Azure OpenAI API version.",
default="v1",
description=(
"Azure OpenAI Realtime protocol version. Use 'v1' for the GA API; "
"date-based versions select the deprecated preview endpoint."
),
json_schema_extra={
"examples": AZURE_REALTIME_API_VERSIONS,
},

35
api/services/pipecat/realtime/azure_realtime.py
View file

@ -1,9 +1,9 @@
"""Dograh subclass of pipecat's Azure OpenAI Realtime LLM service.
Layers Dograh engine integration quirks (mute gating, TTSSpeakFrame greeting
trigger, LLMMessagesAppendFrame handling, deferred tool calls) onto pipecat's
AzureRealtimeLLMService, mirroring what DograhOpenAIRealtimeLLMService does
for the standard OpenAI Realtime endpoint.
trigger, LLMMessagesAppendFrame handling, workflow-control deferral) onto
pipecat's AzureRealtimeLLMService, mirroring what
DograhOpenAIRealtimeLLMService does for the standard OpenAI Realtime endpoint.
"""
import json
@ -40,7 +40,7 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
- User-mute audio gating
- TTSSpeakFrame as initial-response trigger
- One-off LLMMessagesAppendFrame handling
- Deferred tool calls until bot finishes speaking
- Workflow-control calls deferred until bot finishes speaking
- finalized=True on TranscriptionFrame for consistency
"""
@ -49,7 +49,7 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
self._user_is_muted: bool = False
self._handled_initial_context: bool = False
self._bot_is_speaking: bool = False
self._deferred_function_calls: list[FunctionCallFromLLM] = []
self._deferred_node_transition_function_calls: list[FunctionCallFromLLM] = []
self._pending_initial_greeting_text: str | None = None
async def process_frame(self, frame: Frame, direction: FrameDirection):
@ -81,7 +81,7 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
self._bot_is_speaking = True
elif isinstance(frame, BotStoppedSpeakingFrame):
self._bot_is_speaking = False
await self._run_pending_function_calls()
await self._run_pending_node_transition_function_calls()
await super().process_frame(frame, direction)
async def _handle_messages_append(self, frame: LLMMessagesAppendFrame):
@ -247,18 +247,19 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
)
)
async def _run_pending_function_calls(self):
if not self._deferred_function_calls:
async def _run_pending_node_transition_function_calls(self):
if not self._deferred_node_transition_function_calls:
return
function_calls = self._deferred_function_calls
self._deferred_function_calls = []
function_calls = self._deferred_node_transition_function_calls
self._deferred_node_transition_function_calls = []
logger.debug(
f"{self}: executing {len(function_calls)} deferred function call(s) "
"after bot turn ended"
f"{self}: executing {len(function_calls)} deferred workflow-control "
"call(s) after bot turn ended"
)
await self.run_function_calls(function_calls)
async def _handle_evt_function_call_arguments_done(self, evt):
"""Run ordinary tools immediately and defer workflow-control calls."""
try:
args = json.loads(evt.arguments)
@ -275,10 +276,14 @@ class DograhAzureRealtimeLLMService(AzureRealtimeLLMService):
)
]
if self._bot_is_speaking:
self._deferred_function_calls.extend(function_calls)
is_node_transition = self._function_is_node_transition(
function_call_item.name
)
if self._bot_is_speaking and is_node_transition:
self._deferred_node_transition_function_calls.extend(function_calls)
logger.debug(
f"{self}: deferring function call {function_call_item.name} "
f"{self}: deferring workflow-control call "
f"{function_call_item.name} "
"until bot stops speaking"
)
else:

30
api/services/pipecat/realtime/grok_realtime.py
View file

@ -11,8 +11,9 @@ Adds:
flow kicks off the bot's first response.
- **One-off LLMMessagesAppendFrame handling** for ephemeral realtime prompts
like user-idle checks, without mutating Dograh's local ``LLMContext``.
- **Function-call deferral** until the bot finishes speaking, to avoid racing
tool execution with the active audio turn.
- **Workflow-control deferral** so node transitions, call termination, and
transfers wait for any current bot audio to finish while ordinary tools run
immediately.
- **finalized=True on TranscriptionFrame** for parity with Dograh's other
realtime providers.
"""
@ -50,7 +51,7 @@ class DograhGrokRealtimeLLMService(GrokRealtimeLLMService):
self._user_is_muted: bool = False
self._handled_initial_context: bool = False
self._bot_is_speaking: bool = False
self._deferred_function_calls: list[FunctionCallFromLLM] = []
self._deferred_node_transition_function_calls: list[FunctionCallFromLLM] = []
self._pending_initial_greeting_text: str | None = None
async def process_frame(self, frame: Frame, direction: FrameDirection):
@ -82,7 +83,7 @@ class DograhGrokRealtimeLLMService(GrokRealtimeLLMService):
self._bot_is_speaking = True
elif isinstance(frame, BotStoppedSpeakingFrame):
self._bot_is_speaking = False
await self._run_pending_function_calls()
await self._run_pending_node_transition_function_calls()
await super().process_frame(frame, direction)
async def _handle_messages_append(self, frame: LLMMessagesAppendFrame):
@ -251,19 +252,19 @@ class DograhGrokRealtimeLLMService(GrokRealtimeLLMService):
)
)
async def _run_pending_function_calls(self):
if not self._deferred_function_calls:
async def _run_pending_node_transition_function_calls(self):
if not self._deferred_node_transition_function_calls:
return
function_calls = self._deferred_function_calls
self._deferred_function_calls = []
function_calls = self._deferred_node_transition_function_calls
self._deferred_node_transition_function_calls = []
logger.debug(
f"{self}: executing {len(function_calls)} deferred function call(s) "
"after bot turn ended"
f"{self}: executing {len(function_calls)} deferred workflow-control "
"call(s) after bot turn ended"
)
await self.run_function_calls(function_calls)
async def _handle_evt_function_call_arguments_done(self, evt):
"""Process or defer tool calls until the bot finishes speaking."""
"""Run ordinary tools immediately and defer workflow-control calls."""
try:
args = json.loads(evt.arguments)
@ -281,10 +282,11 @@ class DograhGrokRealtimeLLMService(GrokRealtimeLLMService):
)
]
if self._bot_is_speaking:
self._deferred_function_calls.extend(function_calls)
is_node_transition = self._function_is_node_transition(function_name)
if self._bot_is_speaking and is_node_transition:
self._deferred_node_transition_function_calls.extend(function_calls)
logger.debug(
f"{self}: deferring function call {function_name} "
f"{self}: deferring workflow-control call {function_name} "
"until bot stops speaking"
)
else:

39
api/services/pipecat/realtime/openai_realtime.py
View file

@ -3,8 +3,8 @@
Layers Dograh engine integration quirks onto upstream-pristine
:class:`OpenAIRealtimeLLMService`. Substantially smaller than the Gemini
subclass because OpenAI Realtime supports runtime ``session.update`` for
both ``system_instruction`` and tools — no reconnect/defer-tool-call
machinery needed.
both ``system_instruction`` and tools, so node changes do not require a
reconnect.
Adds:
@ -13,6 +13,9 @@ Adds:
flow kicks off the bot's first response.
- **One-off LLMMessagesAppendFrame handling** for ephemeral realtime prompts
like user-idle checks, without mutating Dograh's local ``LLMContext``.
- **Workflow-control deferral** so node transitions, call termination, and
transfers wait for any current bot audio to finish while ordinary tools run
immediately.
- **finalized=True on TranscriptionFrame** because every OpenAI
transcription via the ``completed`` event is final by construction.
"""
@ -53,10 +56,10 @@ class DograhOpenAIRealtimeLLMService(OpenAIRealtimeLLMService):
# LLMContextFrame arrives, so upstream's "first arrival means
# self._context is None" check no longer works.
self._handled_initial_context: bool = False
# Track bot speech locally so tool calls can be deferred until the bot
# has finished speaking, matching Dograh's Gemini Live behavior.
# Track bot speech locally so workflow-control calls can wait until the
# bot has finished speaking without delaying ordinary tools.
self._bot_is_speaking: bool = False
self._deferred_function_calls: list[FunctionCallFromLLM] = []
self._deferred_node_transition_function_calls: list[FunctionCallFromLLM] = []
self._pending_initial_greeting_text: str | None = None
# ------------------------------------------------------------------
@ -100,7 +103,7 @@ class DograhOpenAIRealtimeLLMService(OpenAIRealtimeLLMService):
self._bot_is_speaking = True
elif isinstance(frame, BotStoppedSpeakingFrame):
self._bot_is_speaking = False
await self._run_pending_function_calls()
await self._run_pending_node_transition_function_calls()
await super().process_frame(frame, direction)
async def _handle_messages_append(self, frame: LLMMessagesAppendFrame):
@ -268,19 +271,19 @@ class DograhOpenAIRealtimeLLMService(OpenAIRealtimeLLMService):
)
)
async def _run_pending_function_calls(self):
if not self._deferred_function_calls:
async def _run_pending_node_transition_function_calls(self):
if not self._deferred_node_transition_function_calls:
return
function_calls = self._deferred_function_calls
self._deferred_function_calls = []
function_calls = self._deferred_node_transition_function_calls
self._deferred_node_transition_function_calls = []
logger.debug(
f"{self}: executing {len(function_calls)} deferred function call(s) "
"after bot turn ended"
f"{self}: executing {len(function_calls)} deferred workflow-control "
"call(s) after bot turn ended"
)
await self.run_function_calls(function_calls)
async def _handle_evt_function_call_arguments_done(self, evt):
"""Process or defer tool calls until the bot finishes speaking."""
"""Run ordinary tools immediately and defer workflow-control calls."""
try:
args = json.loads(evt.arguments)
@ -297,10 +300,14 @@ class DograhOpenAIRealtimeLLMService(OpenAIRealtimeLLMService):
)
]
if self._bot_is_speaking:
self._deferred_function_calls.extend(function_calls)
is_node_transition = self._function_is_node_transition(
function_call_item.name
)
if self._bot_is_speaking and is_node_transition:
self._deferred_node_transition_function_calls.extend(function_calls)
logger.debug(
f"{self}: deferring function call {function_call_item.name} "
f"{self}: deferring workflow-control call "
f"{function_call_item.name} "
"until bot stops speaking"
)
else:

362
api/services/pipecat/realtime/ultravox_realtime.py
View file

@ -1,19 +1,16 @@
"""Dograh subclass of pipecat's Ultravox realtime LLM service.
Ultravox is audio-native and realtime, but prompt and tool configuration is
bound to call creation. Dograh therefore cannot lean on in-session updates or
Gemini-style session resumption handles. This wrapper adapts Ultravox to the
Dograh engine contract by:
Ultravox is audio-native and realtime. Its native call stages allow a client
tool result to atomically change the system prompt and tools while preserving
the call's server-side conversation history. This wrapper adapts that model to
the Dograh engine contract by:
- deferring the first call creation until the engine queues the initial node
opening via ``TTSSpeakFrame`` or ``LLMContextFrame``
- marking the call for recreation when ``system_instruction`` changes across
node transitions, then rebuilding it on the follow-up ``LLMContextFrame``
so the transition tool result is present in ``initialMessages``
- reconstructing Ultravox ``initialMessages`` from Dograh context when the
call must be recreated after a node transition
- appending a transient resumptive user nudge to recreated ``initialMessages``
after tool-result transitions, without mutating Dograh's stored context
- returning node-transition tool results with ``responseType="new-stage"`` so
the existing call keeps its complete audio-native history
- updating the next stage's system prompt and selected tools without a
disconnect/reconnect cycle
- handling Dograh-only frames such as user mute and idle append prompts
- tagging user transcripts with ``finalized=True`` for downstream parity
"""
@ -34,12 +31,7 @@ from pipecat.frames.frames import (
UserMuteStartedFrame,
UserMuteStoppedFrame,
)
from pipecat.processors.aggregators import async_tool_messages
from pipecat.processors.aggregators.llm_context import (
LLMContext,
LLMSpecificMessage,
is_given,
)
from pipecat.processors.aggregators.llm_context import LLMContext, is_given
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.llm_service import LLMService
from pipecat.services.settings import _NotGiven, assert_given
@ -58,10 +50,6 @@ class DograhUltravoxOneShotInputParams(OneShotInputParams):
_ULTRAVOX_MAX_TOOL_TIMEOUT_SECS = 40.0
_RESUMPTION_USER_MESSAGE = (
"IMPORTANT: We are resuming an existing conversation. You are given previous turns ONLY for your reference. "
"Do not use that to frame your response. Follow your ORIGINAL INSTRUCTIONS ONLY."
)
class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
@ -72,12 +60,13 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
self._context: LLMContext | None = None
self._selected_tools = None
self._user_is_muted: bool = False
self._call_system_instruction: str | None = None
self._reconnect_required: bool = False
self._call_started: bool = False
self._has_connected_once: bool = False
self._pending_reconnect_system_instruction: str | None = None
self._pending_initial_messages: list[dict[str, Any]] | None = None
self._stage_update_required: bool = False
# Ultravox applies a stage update on the matching client tool result,
# so retain the provider invocation ID until that result reaches us via
# the context aggregator. Unlike Gemini, this ID is part of the wire
# protocol needed to update the existing call without reconnecting.
self._pending_node_transition_tool_call_ids: set[str] = set()
self._pending_user_text_messages: list[str] = []
async def start(self, frame):
@ -96,9 +85,7 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
if isinstance(frame, TTSSpeakFrame):
if not self._socket:
await self._connect_call(
system_instruction=self._current_system_instruction(),
greeting_text=frame.text,
initial_messages=None,
agent_speaks_first=True,
)
else:
@ -116,18 +103,15 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
changed = await super(UltravoxRealtimeLLMService, self)._update_settings(delta)
if "output_medium" in changed:
await self._update_output_medium(assert_given(self._settings.output_medium))
if "system_instruction" in changed and self._has_connected_once:
# Mirror Gemini's "settings change means reconnect" intent, but
# defer the actual new-call creation until the subsequent
# LLMContextFrame arrives with the transition tool result. Ultravox
# cannot accept that historical tool result over a formal
# post-connect tool-response channel the way Gemini can.
self._reconnect_required = True
if "system_instruction" in changed and self._socket:
# The updated instruction is included in the native new-stage
# response when the transition tool result reaches _handle_context.
self._stage_update_required = True
handled = {"output_medium", "system_instruction"}
self._warn_unhandled_updated_settings(changed.keys() - handled)
return changed
async def _disconnect(self, preserve_completed_tool_calls: bool = True):
async def _disconnect(self):
self._disconnecting = True
await self.stop_all_metrics()
if self._socket:
@ -136,10 +120,10 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
if self._receive_task:
await self.cancel_task(self._receive_task, timeout=1.0)
self._receive_task = None
if not preserve_completed_tool_calls:
self._completed_tool_calls = set()
self._completed_tool_calls = set()
self._call_started = False
self._started_placeholder_sent = set()
self._pending_node_transition_tool_call_ids = set()
self._disconnecting = False
async def _send_user_audio(self, frame):
@ -149,39 +133,20 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
async def _handle_context(self, context: LLMContext):
self._context = context
system_instruction = self._current_system_instruction()
if self._socket and not self._reconnect_required:
await super()._handle_context(context)
if not self._socket:
await self._connect_call(
greeting_text=None,
agent_speaks_first=True,
)
return
initial_messages, history_tool_call_ids = self._build_initial_messages(context)
if history_tool_call_ids:
self._completed_tool_calls.update(history_tool_call_ids)
if self._bot_responding:
self._pending_reconnect_system_instruction = system_instruction
self._pending_initial_messages = initial_messages
return
await self._reconnect_with_context(
system_instruction=system_instruction,
initial_messages=initial_messages,
)
async def _handle_response_end(self):
await super()._handle_response_end()
if self._pending_reconnect_system_instruction is None:
return
system_instruction = self._pending_reconnect_system_instruction
initial_messages = self._pending_initial_messages
self._pending_reconnect_system_instruction = None
self._pending_initial_messages = None
await self._reconnect_with_context(
system_instruction=system_instruction,
initial_messages=initial_messages,
)
current_tools = self._current_tools_schema(context)
if self._pending_node_transition_tool_call_ids and self._tools_changed(
current_tools
):
self._stage_update_required = True
await super()._handle_context(context)
async def _handle_messages_append(self, frame: LLMMessagesAppendFrame):
texts = [
@ -199,9 +164,7 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
if not self._socket:
self._pending_user_text_messages.extend(texts)
await self._connect_call(
system_instruction=self._current_system_instruction(),
greeting_text=None,
initial_messages=None,
agent_speaks_first=False,
)
return
@ -229,17 +192,66 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
finalized=True,
)
def _requires_node_transition_context_aggregation(self) -> bool:
"""Commit any received final user transcript before changing stages.
Ultravox preserves its own audio-native history across a stage change,
but Dograh's local context still needs the final transcript before the
transition handler updates the workflow node.
"""
return True
async def _handle_tool_invocation(
self, tool_name: str, invocation_id: str, parameters: dict[str, Any]
):
if self._function_is_node_transition(tool_name):
self._pending_node_transition_tool_call_ids.add(invocation_id)
await super()._handle_tool_invocation(tool_name, invocation_id, parameters)
async def _send_tool_result(self, tool_call_id: str, result: str):
is_node_transition = tool_call_id in self._pending_node_transition_tool_call_ids
try:
if is_node_transition and self._stage_update_required:
await self._send_node_transition_stage_result(tool_call_id, result)
else:
await super()._send_tool_result(tool_call_id, result)
finally:
if is_node_transition:
self._pending_node_transition_tool_call_ids.discard(tool_call_id)
async def _send_node_transition_stage_result(self, tool_call_id: str, result: str):
"""Apply node settings using Ultravox's native call-stage protocol."""
next_tools = self._current_tools_schema(self._context)
stage = {
"systemPrompt": self._current_system_instruction(),
"selectedTools": self._selected_tools_payload(next_tools),
# Keep the workflow handler's result as the tool-result message in
# the inherited conversation history for the next generation.
"toolResultText": result,
}
logger.debug(
f"{self}: updating Ultravox call stage for tool_call_id={tool_call_id} "
f"with {len(stage['selectedTools'])} selected tool(s)"
)
await self._send(
{
"type": "client_tool_result",
"invocationId": tool_call_id,
"result": json.dumps(stage, ensure_ascii=True, default=str),
"responseType": "new-stage",
}
)
self._selected_tools = next_tools
self._stage_update_required = False
async def _connect_call(
self,
*,
system_instruction: str | None,
greeting_text: str | None,
initial_messages: list[dict[str, Any]] | None,
agent_speaks_first: bool,
):
params = self._build_one_shot_params(
greeting_text=greeting_text,
initial_messages=initial_messages,
agent_speaks_first=agent_speaks_first,
)
self._params = params
@ -265,9 +277,7 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
logger.info(f"Joining Ultravox Realtime call via URL: {join_url}")
self._socket = await websocket_client.connect(join_url)
self._receive_task = self.create_task(self._receive_messages())
self._call_system_instruction = system_instruction
self._call_started = False
self._has_connected_once = True
except Exception as e:
logger.error(
f"{self}: Ultravox call creation/join failed "
@ -365,40 +375,17 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
for pending_text in pending_texts:
await self._send_user_text(pending_text)
async def _reconnect_with_context(
self,
*,
system_instruction: str | None,
initial_messages: list[dict[str, Any]] | None,
):
call_initial_messages = self._initial_messages_for_call(initial_messages)
logger.debug(
f"{self}: reconnecting Ultravox call with initialMessages="
f"{json.dumps(call_initial_messages, ensure_ascii=True, default=str)}"
)
if self._socket:
await self._disconnect(preserve_completed_tool_calls=True)
await self._connect_call(
system_instruction=system_instruction,
greeting_text=None,
initial_messages=initial_messages,
agent_speaks_first=self._should_agent_speak_first(initial_messages),
)
self._reconnect_required = False
def _build_one_shot_params(
self,
*,
greeting_text: str | None,
initial_messages: list[dict[str, Any]] | None,
agent_speaks_first: bool,
) -> DograhUltravoxOneShotInputParams:
current_params = self._params
extra = {
key: value
for key, value in current_params.extra.items()
if key not in {"firstSpeakerSettings", "initialMessages"}
if key != "firstSpeakerSettings"
}
if greeting_text is not None:
@ -407,10 +394,6 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
extra["firstSpeakerSettings"] = {"agent": {}}
else:
extra["firstSpeakerSettings"] = {"user": {}}
call_initial_messages = self._initial_messages_for_call(initial_messages)
if call_initial_messages:
extra["initialMessages"] = call_initial_messages
output_medium = self._settings.output_medium
if isinstance(output_medium, _NotGiven):
output_medium = current_params.output_medium
@ -432,6 +415,14 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
return None
return context.tools
def _selected_tools_payload(self, tools: Any) -> list[dict[str, Any]]:
return self._to_selected_tools(tools) if tools else []
def _tools_changed(self, tools: Any) -> bool:
return self._selected_tools_payload(tools) != self._selected_tools_payload(
self._selected_tools
)
def _to_selected_tools(self, tool: Any) -> list[dict[str, Any]]:
selected_tools = super()._to_selected_tools(tool)
for selected_tool in selected_tools:
@ -462,156 +453,6 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
timeout_secs = min(float(item.timeout_secs), _ULTRAVOX_MAX_TOOL_TIMEOUT_SECS)
return f"{timeout_secs:g}s"
def _initial_messages_for_call(
self, initial_messages: list[dict[str, Any]] | None
) -> list[dict[str, Any]] | None:
if not initial_messages:
return None
if not self._should_add_resumption_user_message(initial_messages):
return initial_messages
return [
*initial_messages,
{
"role": "MESSAGE_ROLE_USER",
"text": _RESUMPTION_USER_MESSAGE,
},
]
def _build_initial_messages(
self, context: LLMContext
) -> tuple[list[dict[str, Any]] | None, set[str]]:
initial_messages: list[dict[str, Any]] = []
tool_call_id_to_name: dict[str, str] = {}
completed_tool_call_ids: set[str] = set()
for message in context.get_messages():
if isinstance(message, LLMSpecificMessage):
continue
async_payload = async_tool_messages.parse_message(message)
if async_payload is not None:
if async_payload.kind == "intermediate":
logger.error(
f"{self}: Ultravox does not support streamed async tool results; "
f"dropping intermediate result from initialMessages for "
f"tool_call_id={async_payload.tool_call_id}."
)
continue
if async_payload.kind == "final":
initial_message = self._build_ultravox_message(
role="MESSAGE_ROLE_TOOL_RESULT",
text=async_payload.result or "",
invocation_id=async_payload.tool_call_id,
tool_name=tool_call_id_to_name.get(async_payload.tool_call_id),
)
if initial_message is not None:
initial_messages.append(initial_message)
completed_tool_call_ids.add(async_payload.tool_call_id)
continue
role = message.get("role")
if role == "user":
initial_message = self._build_ultravox_message(
role="MESSAGE_ROLE_USER",
text=self._extract_text_content(message.get("content")),
)
if initial_message is not None:
initial_messages.append(initial_message)
elif role == "assistant":
text = self._extract_text_content(message.get("content"))
initial_message = self._build_ultravox_message(
role="MESSAGE_ROLE_AGENT",
text=text,
)
if initial_message is not None:
initial_messages.append(initial_message)
tool_calls = message.get("tool_calls")
if isinstance(tool_calls, list):
for tool_call in tool_calls:
if not isinstance(tool_call, dict):
continue
tool_id = tool_call.get("id")
function = tool_call.get("function")
tool_name = (
function.get("name") if isinstance(function, dict) else None
)
if isinstance(tool_id, str) and isinstance(tool_name, str):
tool_call_id_to_name[tool_id] = tool_name
initial_message = self._build_ultravox_message(
role="MESSAGE_ROLE_TOOL_CALL",
text="",
invocation_id=tool_id,
tool_name=tool_name,
)
if initial_message is not None:
initial_messages.append(initial_message)
elif (
role == "tool"
and message.get("content") != "IN_PROGRESS"
and message.get("content") != "CANCELLED"
):
tool_call_id = message.get("tool_call_id")
initial_message = self._build_ultravox_message(
role="MESSAGE_ROLE_TOOL_RESULT",
text=self._stringify_tool_result(message.get("content")),
invocation_id=tool_call_id
if isinstance(tool_call_id, str)
else None,
tool_name=(
tool_call_id_to_name.get(tool_call_id)
if isinstance(tool_call_id, str)
else None
),
)
if initial_message is not None:
initial_messages.append(initial_message)
if isinstance(tool_call_id, str):
completed_tool_call_ids.add(tool_call_id)
return (initial_messages or None), completed_tool_call_ids
@staticmethod
def _build_ultravox_message(
*,
role: str,
text: str | None,
invocation_id: str | None = None,
tool_name: str | None = None,
) -> dict[str, Any] | None:
if text is None:
return None
message: dict[str, Any] = {
"role": role,
"text": text,
}
if invocation_id is not None:
message["invocationId"] = invocation_id
if tool_name is not None:
message["toolName"] = tool_name
return message
@staticmethod
def _should_agent_speak_first(
initial_messages: list[dict[str, Any]] | None,
) -> bool:
if not initial_messages:
return True
return initial_messages[-1].get("role") in {
"MESSAGE_ROLE_USER",
"MESSAGE_ROLE_TOOL_RESULT",
}
@staticmethod
def _should_add_resumption_user_message(
initial_messages: list[dict[str, Any]] | None,
) -> bool:
if not initial_messages:
return False
return initial_messages[-1].get("role") == "MESSAGE_ROLE_TOOL_RESULT"
@staticmethod
def _is_benign_websocket_close(exc: ConnectionClosed) -> bool:
return any(
@ -636,18 +477,3 @@ class DograhUltravoxRealtimeLLMService(UltravoxRealtimeLLMService):
parts.append(text)
return "\n".join(parts) if parts else None
return None
@staticmethod
def _stringify_tool_result(content: Any) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for part in content:
if isinstance(part, dict):
text = part.get("text")
if isinstance(text, str):
parts.append(text)
if parts:
return "".join(parts)
return json.dumps(content, ensure_ascii=True, default=str)

38
api/services/pipecat/service_factory.py
View file

@ -1029,14 +1029,28 @@ def create_realtime_llm_service(user_config, audio_config: "AudioConfig"):
from api.services.pipecat.realtime.grok_realtime import (
DograhGrokRealtimeLLMService,
)
from pipecat.services.xai.realtime.events import SessionProperties
from pipecat.services.xai.realtime.events import (
AudioConfiguration,
AudioInput,
InputAudioTranscription,
SessionProperties,
)
grok_voice = voice or "ara"
if grok_voice.lower() in {"ara", "rex", "sal", "eve", "leo"}:
grok_voice = grok_voice.lower()
return DograhGrokRealtimeLLMService(
api_key=api_key,
settings=DograhGrokRealtimeLLMService.Settings(
model=model,
session_properties=SessionProperties(
voice=voice or "Ara",
voice=grok_voice,
audio=AudioConfiguration(
input=AudioInput(
transcription=InputAudioTranscription(),
),
),
),
),
)
@ -1115,19 +1129,25 @@ def create_realtime_llm_service(user_config, audio_config: "AudioConfig"):
detail="Azure Realtime requires an endpoint.",
)
_validate_runtime_service_url(endpoint, "endpoint")
api_version = (
getattr(realtime_config, "api_version", None) or "2025-04-01-preview"
)
# Construct the Azure Realtime WebSocket URL
# https://<resource>.openai.azure.com/openai/realtime?api-version=<ver>&deployment=<model>
api_version = getattr(realtime_config, "api_version", None) or "v1"
parsed_endpoint = urlparse(endpoint)
if api_version == "v1":
# Azure's GA Realtime API uses the deployment name as `model` and
# deliberately has no date-based api-version query parameter.
path = "/openai/v1/realtime"
query = urlencode({"model": model})
else:
# Preserve explicitly configured preview deployments while users
# migrate. Microsoft deprecated this protocol on April 30, 2026.
path = "/openai/realtime"
query = urlencode({"api-version": api_version, "deployment": model})
wss_url = urlunparse(
(
"wss",
parsed_endpoint.netloc,
"/openai/realtime",
path,
"",
urlencode({"api-version": api_version, "deployment": model}),
query,
"",
)
)

42
api/tests/test_azure_realtime_wrapper.py
View file

@ -86,3 +86,45 @@ async def test_tts_greeting_waits_for_session_updated_before_sending_prompt():
assert service._pending_initial_greeting_text is None
assert service._llm_needs_conversation_setup is False
service._create_response.assert_not_awaited()
@pytest.mark.asyncio
async def test_non_transition_function_call_runs_while_bot_is_speaking():
service = _make_service()
service._context = LLMContext()
service.run_function_calls = AsyncMock()
service._bot_is_speaking = True
service._pending_function_calls["call-1"] = SimpleNamespace(name="lookup_order")
await service._handle_evt_function_call_arguments_done(
SimpleNamespace(call_id="call-1", arguments='{"order_id":"123"}')
)
service.run_function_calls.assert_awaited_once()
assert service._deferred_node_transition_function_calls == []
@pytest.mark.asyncio
async def test_node_transition_function_call_waits_until_bot_stops_speaking():
service = _make_service()
service._context = LLMContext()
service.run_function_calls = AsyncMock()
service._bot_is_speaking = True
service.register_function(
"customer_support",
AsyncMock(),
is_node_transition=True,
)
service._pending_function_calls["call-1"] = SimpleNamespace(name="customer_support")
await service._handle_evt_function_call_arguments_done(
SimpleNamespace(call_id="call-1", arguments='{"department":"sales"}')
)
service.run_function_calls.assert_not_awaited()
assert len(service._deferred_node_transition_function_calls) == 1
await service._run_pending_node_transition_function_calls()
service.run_function_calls.assert_awaited_once()
assert service._deferred_node_transition_function_calls == []

40
api/tests/test_azure_speech_service_factory.py
View file

@ -170,6 +170,46 @@ def test_create_azure_realtime_blocks_private_endpoint_in_saas(monkeypatch):
assert "public IP" in exc_info.value.detail
def test_create_azure_realtime_uses_ga_websocket_url_by_default(monkeypatch):
monkeypatch.setattr("api.utils.url_security.DEPLOYMENT_MODE", "oss")
user_config = SimpleNamespace(
realtime=SimpleNamespace(
provider=ServiceProviders.AZURE_REALTIME.value,
api_key="test-key",
endpoint="https://example.openai.azure.com",
model="my-realtime-deployment",
voice="alloy",
)
)
service = create_realtime_llm_service(user_config, _audio_config())
assert service.base_url == (
"wss://example.openai.azure.com/openai/v1/realtime?model=my-realtime-deployment"
)
def test_create_azure_realtime_preserves_explicit_preview_websocket_url(monkeypatch):
monkeypatch.setattr("api.utils.url_security.DEPLOYMENT_MODE", "oss")
user_config = SimpleNamespace(
realtime=SimpleNamespace(
provider=ServiceProviders.AZURE_REALTIME.value,
api_key="test-key",
endpoint="https://example.openai.azure.com",
api_version="2025-04-01-preview",
model="my-preview-deployment",
voice="alloy",
)
)
service = create_realtime_llm_service(user_config, _audio_config())
assert service.base_url == (
"wss://example.openai.azure.com/openai/realtime?"
"api-version=2025-04-01-preview&deployment=my-preview-deployment"
)
def test_azure_embedding_service_rejects_wrong_dimension():
service = AzureOpenAIEmbeddingService(
db_client=SimpleNamespace(),

15
api/tests/test_gemini_json_schema_adapter.py
View file

@ -134,6 +134,21 @@ def test_gemini_live_service_classes_use_dograh_gemini_adapter_class():
)
def test_vertex_live_inherits_dograh_node_transition_lifecycle():
assert (
DograhGeminiLiveVertexLLMService._requires_node_transition_context_aggregation
is DograhGeminiLiveLLMService._requires_node_transition_context_aggregation
)
assert (
DograhGeminiLiveVertexLLMService._run_or_defer_function_calls
is DograhGeminiLiveLLMService._run_or_defer_function_calls
)
assert (
DograhGeminiLiveVertexLLMService._reconnect_for_node_transition
is DograhGeminiLiveLLMService._reconnect_for_node_transition
)
def test_gemini_live_config_accepts_json_schema_tools():
function_schema = FunctionSchema(
name="customer_lookup",

53
api/tests/test_grok_realtime_wrapper.py
View file

@ -130,7 +130,7 @@ async def test_messages_append_frame_sends_conversation_item():
@pytest.mark.asyncio
async def test_function_call_is_deferred_until_bot_stops_speaking():
async def test_non_transition_function_call_runs_while_bot_is_speaking():
service = _make_service()
service._context = LLMContext()
service.run_function_calls = AsyncMock()
@ -145,13 +145,38 @@ async def test_function_call_is_deferred_until_bot_stops_speaking():
)
)
service.run_function_calls.assert_not_awaited()
assert len(service._deferred_function_calls) == 1
service.run_function_calls.assert_awaited_once()
assert service._deferred_node_transition_function_calls == []
await service._run_pending_function_calls()
@pytest.mark.asyncio
async def test_node_transition_function_call_waits_until_bot_stops_speaking():
service = _make_service()
service._context = LLMContext()
service.run_function_calls = AsyncMock()
service._bot_is_speaking = True
service.register_function(
"customer_support",
AsyncMock(),
is_node_transition=True,
)
service._pending_function_calls["call-1"] = SimpleNamespace(name="customer_support")
await service._handle_evt_function_call_arguments_done(
SimpleNamespace(
call_id="call-1",
name="customer_support",
arguments='{"department":"sales"}',
)
)
service.run_function_calls.assert_not_awaited()
assert len(service._deferred_node_transition_function_calls) == 1
await service._run_pending_node_transition_function_calls()
service.run_function_calls.assert_awaited_once()
assert service._deferred_function_calls == []
assert service._deferred_node_transition_function_calls == []
@pytest.mark.asyncio
@ -190,3 +215,21 @@ def test_factory_creates_dograh_grok_realtime_service():
)
assert isinstance(service, DograhGrokRealtimeLLMService)
assert service._settings.session_properties.voice == "sal"
assert service._settings.session_properties.audio.input.transcription.model == (
"grok-transcribe"
)
def test_grok_audio_config_preserves_transcription_when_filling_sample_rates():
service = _make_service()
service._settings.session_properties.audio = events.AudioConfiguration(
input=events.AudioInput(transcription=events.InputAudioTranscription())
)
service._ensure_audio_config(input_sample_rate=16000, output_sample_rate=24000)
audio = service._settings.session_properties.audio
assert audio.input.format.rate == 16000
assert audio.input.transcription.model == "grok-transcribe"
assert audio.output.format.rate == 24000

33
api/tests/test_openai_realtime_initial_context.py
View file

@ -127,11 +127,11 @@ async def test_function_call_executes_immediately_when_bot_is_not_speaking():
)
service.run_function_calls.assert_awaited_once()
assert service._deferred_function_calls == []
assert service._deferred_node_transition_function_calls == []
@pytest.mark.asyncio
async def test_function_call_is_deferred_until_bot_stops_speaking():
async def test_non_transition_function_call_runs_while_bot_is_speaking():
service = _make_service()
service._context = LLMContext()
service.run_function_calls = AsyncMock()
@ -142,10 +142,31 @@ async def test_function_call_is_deferred_until_bot_stops_speaking():
SimpleNamespace(call_id="call-1", arguments='{"department":"sales"}')
)
service.run_function_calls.assert_not_awaited()
assert len(service._deferred_function_calls) == 1
service.run_function_calls.assert_awaited_once()
assert service._deferred_node_transition_function_calls == []
await service._run_pending_function_calls()
@pytest.mark.asyncio
async def test_node_transition_function_call_is_deferred_until_bot_stops_speaking():
service = _make_service()
service._context = LLMContext()
service.run_function_calls = AsyncMock()
service._bot_is_speaking = True
service.register_function(
"customer_support",
AsyncMock(),
is_node_transition=True,
)
service._pending_function_calls["call-1"] = SimpleNamespace(name="customer_support")
await service._handle_evt_function_call_arguments_done(
SimpleNamespace(call_id="call-1", arguments='{"department":"sales"}')
)
service.run_function_calls.assert_not_awaited()
assert len(service._deferred_node_transition_function_calls) == 1
await service._run_pending_node_transition_function_calls()
service.run_function_calls.assert_awaited_once()
assert service._deferred_function_calls == []
assert service._deferred_node_transition_function_calls == []

187
api/tests/test_ultravox_realtime_wrapper.py
View file

@ -1,3 +1,4 @@
import json
from types import SimpleNamespace
from unittest.mock import AsyncMock, call
@ -13,7 +14,6 @@ from websockets.frames import Close
from api.schemas.ai_model_configuration import EffectiveAIModelConfiguration
from api.services.configuration.registry import UltravoxRealtimeLLMConfiguration
from api.services.pipecat.realtime.ultravox_realtime import (
_RESUMPTION_USER_MESSAGE,
DograhUltravoxOneShotInputParams,
DograhUltravoxRealtimeLLMService,
)
@ -100,50 +100,35 @@ async def test_initial_context_connects_without_replay():
await service._handle_context(context)
service._connect_call.assert_awaited_once()
assert service._connect_call.await_args.kwargs["initial_messages"] is None
assert service._connect_call.await_args.kwargs["greeting_text"] is None
assert service._connect_call.await_args.kwargs["agent_speaks_first"] is True
@pytest.mark.asyncio
async def test_system_instruction_update_marks_reconnect_required():
async def test_system_instruction_update_marks_stage_update_required():
service = _make_service()
service._has_connected_once = True
service._socket = object()
changed = await service._update_settings(
DograhUltravoxRealtimeLLMService.Settings(system_instruction="new instruction")
)
assert "system_instruction" in changed
assert service._reconnect_required is True
assert service._stage_update_required is True
@pytest.mark.asyncio
async def test_system_instruction_change_reconnects_with_full_initial_messages():
async def test_node_transition_updates_native_stage_without_reconnecting():
service = _make_service()
service._socket = object()
service._has_connected_once = True
service._call_system_instruction = "old instruction"
service._reconnect_required = True
service._send = AsyncMock()
service._connect_call = AsyncMock()
service._pending_node_transition_tool_call_ids.add("call-transition")
service._stage_update_required = True
service._settings.system_instruction = "new instruction"
service._reconnect_with_context = AsyncMock()
context = LLMContext(
messages=[
{"role": "user", "content": "I want to hear the pricing."},
{
"role": "assistant",
"content": "Let me check that for you.",
"tool_calls": [
{
"id": "call-transition",
"type": "function",
"function": {
"name": "transition_to_next_node",
"arguments": '{"reason":"pricing requested"}',
},
}
],
},
{
"role": "tool",
"tool_call_id": "call-transition",
@ -155,43 +140,28 @@ async def test_system_instruction_change_reconnects_with_full_initial_messages()
await service._handle_context(context)
service._reconnect_with_context.assert_awaited_once()
initial_messages = service._reconnect_with_context.await_args.kwargs[
"initial_messages"
]
assert initial_messages == [
{
"role": "MESSAGE_ROLE_USER",
"text": "I want to hear the pricing.",
},
{
"role": "MESSAGE_ROLE_AGENT",
"text": "Let me check that for you.",
},
{
"role": "MESSAGE_ROLE_TOOL_CALL",
"text": "",
"invocationId": "call-transition",
"toolName": "transition_to_next_node",
},
{
"role": "MESSAGE_ROLE_TOOL_RESULT",
"text": '{"status":"done"}',
"invocationId": "call-transition",
"toolName": "transition_to_next_node",
},
]
service._connect_call.assert_not_awaited()
service._send.assert_awaited_once()
message = service._send.await_args.args[0]
assert message["type"] == "client_tool_result"
assert message["invocationId"] == "call-transition"
assert message["responseType"] == "new-stage"
stage = json.loads(message["result"])
assert stage["systemPrompt"] == "new instruction"
assert stage["toolResultText"] == '{"status":"done"}'
assert stage["selectedTools"][0]["temporaryTool"]["modelToolName"] == (
"transition_to_next_node"
)
assert "call-transition" in service._completed_tool_calls
assert service._pending_node_transition_tool_call_ids == set()
assert service._stage_update_required is False
@pytest.mark.asyncio
async def test_tool_context_update_does_not_reconnect_when_system_instruction_is_unchanged():
async def test_ordinary_tool_result_uses_standard_tool_response():
service = _make_service()
service._socket = object()
service._call_system_instruction = "same instruction"
service._settings.system_instruction = "same instruction"
service._reconnect_with_context = AsyncMock()
service._send_tool_result = AsyncMock()
service._send = AsyncMock()
context = LLMContext(
messages=[
@ -206,13 +176,40 @@ async def test_tool_context_update_does_not_reconnect_when_system_instruction_is
await service._handle_context(context)
service._reconnect_with_context.assert_not_awaited()
service._send_tool_result.assert_awaited_once_with(
"call-transition",
'{"status":"done"}',
service._send.assert_awaited_once_with(
{
"type": "client_tool_result",
"invocationId": "call-transition",
"result": '{"status":"done"}',
}
)
@pytest.mark.asyncio
async def test_only_registered_node_transition_invocations_are_tracked():
service = _make_service()
service.run_function_calls = AsyncMock()
service.register_function(
"transition_to_next_node",
AsyncMock(),
is_node_transition=True,
)
await service._handle_tool_invocation(
"transition_to_next_node", "call-transition", {"reason": "pricing"}
)
await service._handle_tool_invocation("lookup_price", "call-lookup", {})
assert service._pending_node_transition_tool_call_ids == {"call-transition"}
assert service.run_function_calls.await_count == 2
def test_ultravox_requires_transition_context_aggregation():
service = _make_service()
assert service._requires_node_transition_context_aggregation() is True
@pytest.mark.asyncio
async def test_messages_append_frame_sends_user_text():
service = _make_service()
@ -287,7 +284,6 @@ def test_build_one_shot_params_uses_explicit_greeting_text():
params = service._build_one_shot_params(
greeting_text="Welcome to Dograh",
initial_messages=None,
agent_speaks_first=True,
)
@ -296,85 +292,18 @@ def test_build_one_shot_params_uses_explicit_greeting_text():
}
def test_build_one_shot_params_includes_initial_messages():
def test_build_one_shot_params_uses_current_system_instruction():
service = _make_service()
service._settings.system_instruction = "Base instruction"
params = service._build_one_shot_params(
greeting_text=None,
initial_messages=[
{"role": "MESSAGE_ROLE_USER", "text": "User asked a question."},
{"role": "MESSAGE_ROLE_TOOL_RESULT", "text": '{"status":"done"}'},
],
agent_speaks_first=True,
)
assert params.extra["initialMessages"] == [
{"role": "MESSAGE_ROLE_USER", "text": "User asked a question."},
{"role": "MESSAGE_ROLE_TOOL_RESULT", "text": '{"status":"done"}'},
{"role": "MESSAGE_ROLE_USER", "text": _RESUMPTION_USER_MESSAGE},
]
assert params.system_prompt == "Base instruction"
def test_build_one_shot_params_without_tool_result_does_not_add_resumption_user_message():
service = _make_service()
service._settings.system_instruction = "Base instruction"
params = service._build_one_shot_params(
greeting_text=None,
initial_messages=[
{"role": "MESSAGE_ROLE_USER", "text": "User asked a question."},
{"role": "MESSAGE_ROLE_AGENT", "text": "Assistant replied."},
],
agent_speaks_first=False,
)
assert params.system_prompt == "Base instruction"
def test_should_agent_speak_first_when_history_ends_with_tool_result():
service = _make_service()
assert (
service._should_agent_speak_first(
[
{"role": "MESSAGE_ROLE_USER", "text": "Hello"},
{"role": "MESSAGE_ROLE_TOOL_RESULT", "text": '{"status":"done"}'},
]
)
is True
)
def test_should_not_force_agent_speaks_first_when_history_ends_with_agent():
service = _make_service()
assert (
service._should_agent_speak_first(
[{"role": "MESSAGE_ROLE_AGENT", "text": "How else can I help?"}]
)
is False
)
def test_should_add_resumption_user_message_only_when_history_ends_with_tool_result():
service = _make_service()
assert (
service._should_add_resumption_user_message(
[{"role": "MESSAGE_ROLE_TOOL_RESULT", "text": '{"status":"done"}'}]
)
is True
)
assert (
service._should_add_resumption_user_message(
[{"role": "MESSAGE_ROLE_AGENT", "text": "Assistant replied."}]
)
is False
)
def test_to_selected_tools_includes_registered_timeout():
service = _make_service()
service.register_function(