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

Merge remote-tracking branch 'origin/main' into feat/provisional-vad

Browse source 
# Conflicts:
#	api/routes/main.py
#	api/tests/test_active_calls.py
#	pipecat
#	scripts/rolling_update.sh
This commit is contained in:
Abhishek Kumar 2026-06-29 13:43:57 +05:30
commit cc658131aa
28 changed files with 1101 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

117
api/services/pipecat/run_pipeline.py
View file

@ -50,7 +50,7 @@ from api.services.pipecat.service_factory import (
create_realtime_llm_service,
create_stt_service,
create_tts_service,
stt_uses_flux_turns,
stt_uses_external_turns,
)
from api.services.pipecat.tracing_config import (
ensure_tracing,
@ -97,6 +97,19 @@ from pipecat.utils.run_context import set_current_org_id, set_current_run_id
# Setup tracing if enabled
ensure_tracing()
DEFAULT_USER_TURN_STOP_TIMEOUT = 5.0
EXTERNAL_TURN_USER_STOP_TIMEOUT = 30.0
def _resolve_user_turn_stop_timeout(
run_configs: dict, *, uses_external_turns: bool
) -> float:
if "user_turn_stop_timeout" in run_configs:
return float(run_configs["user_turn_stop_timeout"])
if uses_external_turns:
return EXTERNAL_TURN_USER_STOP_TIMEOUT
return DEFAULT_USER_TURN_STOP_TIMEOUT
def _create_realtime_user_turn_config(provider: str):
"""Return user turn strategies and optional local VAD for realtime providers."""
@ -154,6 +167,34 @@ async def run_pipeline_telephony(
user_id: int,
call_id: str,
transport_kwargs: dict,
) -> None:
"""Run a pipeline for any telephony provider."""
# Register before any async setup so deploy drains see calls that are still
# resolving DB/config/transport state.
register_active_call(workflow_run_id)
try:
await _run_pipeline_telephony_impl(
websocket,
provider_name=provider_name,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
user_id=user_id,
call_id=call_id,
transport_kwargs=transport_kwargs,
)
finally:
unregister_active_call(workflow_run_id)
async def _run_pipeline_telephony_impl(
websocket,
*,
provider_name: str,
workflow_id: int,
workflow_run_id: int,
user_id: int,
call_id: str,
transport_kwargs: dict,
) -> None:
"""Run a pipeline for any telephony provider.
@ -227,7 +268,7 @@ async def run_pipeline_telephony(
)
try:
await _run_pipeline(
await _run_pipeline_impl(
transport,
workflow_id,
workflow_run_id,
@ -251,6 +292,31 @@ async def run_pipeline_smallwebrtc(
user_id: int,
call_context_vars: dict = {},
user_provider_id: str | None = None,
) -> None:
"""Run pipeline for WebRTC connections."""
# Register before any async setup so deploy drains see calls that are still
# resolving DB/config/transport state.
register_active_call(workflow_run_id)
try:
await _run_pipeline_smallwebrtc_impl(
webrtc_connection,
workflow_id,
workflow_run_id,
user_id,
call_context_vars=call_context_vars,
user_provider_id=user_provider_id,
)
finally:
unregister_active_call(workflow_run_id)
async def _run_pipeline_smallwebrtc_impl(
webrtc_connection: SmallWebRTCConnection,
workflow_id: int,
workflow_run_id: int,
user_id: int,
call_context_vars: dict = {},
user_provider_id: str | None = None,
) -> None:
"""Run pipeline for WebRTC connections"""
logger.debug(
@ -300,7 +366,7 @@ async def run_pipeline_smallwebrtc(
ambient_noise_config,
is_realtime=is_realtime,
)
await _run_pipeline(
await _run_pipeline_impl(
transport,
workflow_id,
workflow_run_id,
@ -323,6 +389,35 @@ async def _run_pipeline(
user_provider_id: str | None = None,
workflow_run=None,
resolved_user_config=None,
) -> None:
"""Run the pipeline with active-call drain accounting."""
register_active_call(workflow_run_id)
try:
await _run_pipeline_impl(
transport,
workflow_id,
workflow_run_id,
user_id,
call_context_vars=call_context_vars,
audio_config=audio_config,
user_provider_id=user_provider_id,
workflow_run=workflow_run,
resolved_user_config=resolved_user_config,
)
finally:
unregister_active_call(workflow_run_id)
async def _run_pipeline_impl(
transport,
workflow_id: int,
workflow_run_id: int,
user_id: int,
call_context_vars: dict = {},
audio_config: AudioConfig = None,
user_provider_id: str | None = None,
workflow_run=None,
resolved_user_config=None,
) -> None:
"""
Run the pipeline with the given transport and configuration
@ -624,16 +719,18 @@ async def _run_pipeline(
# Configure turn strategies based on STT provider, model, and workflow configuration
if is_realtime:
uses_external_turns = False
# Realtime services still need user-turn tracking even when the model
# itself owns speech generation and interruption behavior.
user_turn_strategies, user_vad_analyzer = _create_realtime_user_turn_config(
user_config.realtime.provider
)
else:
# Deepgram Flux and supported Dograh managed Flux languages emit their
# own turn boundaries, so the aggregator follows those external signals.
# Other models use configurable turn detection.
if stt_uses_flux_turns(user_config):
# Some STT services emit their own turn boundaries, so the aggregator
# follows those external signals. Other models use configurable turn
# detection.
uses_external_turns = stt_uses_external_turns(user_config)
if uses_external_turns:
user_turn_strategies = UserTurnStrategies(
start=[
VADUserTurnStartStrategy(),
@ -663,9 +760,15 @@ async def _run_pipeline(
stop=[SpeechTimeoutUserTurnStopStrategy()],
)
user_turn_stop_timeout = _resolve_user_turn_stop_timeout(
run_configs,
uses_external_turns=uses_external_turns,
)
user_params = LLMUserAggregatorParams(
user_turn_strategies=user_turn_strategies,
user_mute_strategies=user_mute_strategies,
user_turn_stop_timeout=user_turn_stop_timeout,
user_idle_timeout=max_user_idle_timeout,
vad_analyzer=user_vad_analyzer,
)

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

@ -21,12 +21,13 @@ from pipecat.services.aws.llm import AWSBedrockLLMService, AWSBedrockLLMSettings
from pipecat.services.azure.llm import AzureLLMService, AzureLLMSettings
from pipecat.services.azure.stt import AzureSTTService, AzureSTTSettings
from pipecat.services.azure.tts import AzureTTSService, AzureTTSSettings
from pipecat.services.cartesia.stt import CartesiaSTTService
from pipecat.services.cartesia.stt import CartesiaSTTService, CartesiaSTTSettings
from pipecat.services.cartesia.tts import (
CartesiaTTSService,
CartesiaTTSSettings,
GenerationConfig,
)
from pipecat.services.cartesia.turns.stt import CartesiaTurnsSTTService
from pipecat.services.deepgram.flux.stt import (
DeepgramFluxSTTService,
DeepgramFluxSTTSettings,
@ -106,11 +107,13 @@ def dograh_stt_uses_flux_language(language: str | None) -> bool:
return language in DEEPGRAM_FLUX_MULTILINGUAL_LANGUAGE_OPTIONS
def stt_uses_flux_turns(user_config) -> bool:
def stt_uses_external_turns(user_config) -> bool:
if user_config.stt.provider == ServiceProviders.DEEPGRAM.value:
return user_config.stt.model in DEEPGRAM_FLUX_MODELS
if user_config.stt.provider == ServiceProviders.DOGRAH.value:
return dograh_stt_uses_flux_language(getattr(user_config.stt, "language", None))
if user_config.stt.provider == ServiceProviders.CARTESIA.value:
return user_config.stt.model == "ink-2"
return False
@ -215,8 +218,20 @@ def create_stt_service(
sample_rate=audio_config.transport_in_sample_rate,
)
elif user_config.stt.provider == ServiceProviders.CARTESIA.value:
if user_config.stt.model == "ink-2":
return CartesiaTurnsSTTService(
api_key=user_config.stt.api_key,
should_interrupt=False, # Let UserAggregator emit interruption frames.
sample_rate=audio_config.transport_in_sample_rate,
)
language = getattr(user_config.stt, "language", None) or "en"
return CartesiaSTTService(
api_key=user_config.stt.api_key,
settings=CartesiaSTTSettings(
model=user_config.stt.model,
language=language,
),
sample_rate=audio_config.transport_in_sample_rate,
)
elif user_config.stt.provider == ServiceProviders.DOGRAH.value: