01c563a3e7
- Introduce a new `turnConfig` field in `AssistantConfig` and `Assistant` models to manage user interaction settings. - Implement `TurnConfig`, `BargeInConfig`, `VadConfig`, and `TurnDetectionConfig` schemas to define turn management strategies. - Update the backend to handle turn configuration in the database and during assistant operations. - Enhance frontend components with a `TurnConfigEditor` for configuring turn settings, including VAD and barge-in strategies. - Modify existing pages to integrate turn configuration, improving user experience and interaction capabilities.
90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
"""把稳定的产品配置映射为 Pipecat 用户轮次策略。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3
|
|
from pipecat.audio.vad.silero import SileroVADAnalyzer
|
|
from pipecat.audio.vad.vad_analyzer import VADParams
|
|
from pipecat.turns.user_start import (
|
|
TranscriptionUserTurnStartStrategy,
|
|
VADUserTurnStartStrategy,
|
|
)
|
|
from pipecat.turns.user_stop import (
|
|
SpeechTimeoutUserTurnStopStrategy,
|
|
TurnAnalyzerUserTurnStopStrategy,
|
|
)
|
|
from pipecat.turns.user_turn_strategies import UserTurnStrategies
|
|
|
|
|
|
DEFAULT_VAD = {
|
|
"confidence": 0.7,
|
|
"start_secs": 0.2,
|
|
"stop_secs": 0.2,
|
|
"min_volume": 0.6,
|
|
}
|
|
DEFAULT_TURN_DETECTION = {
|
|
"strategy": "silence",
|
|
"silence_timeout_secs": 0.6,
|
|
}
|
|
|
|
|
|
def _section(config: dict[str, Any], snake: str, camel: str) -> dict[str, Any]:
|
|
value = config.get(snake, config.get(camel, {}))
|
|
return value if isinstance(value, dict) else {}
|
|
|
|
|
|
def _value(config: dict[str, Any], snake: str, camel: str, default: Any) -> Any:
|
|
return config.get(snake, config.get(camel, default))
|
|
|
|
|
|
def create_vad_analyzer(turn_config: dict[str, Any]) -> SileroVADAnalyzer:
|
|
vad = _section(turn_config, "vad", "vad")
|
|
return SileroVADAnalyzer(
|
|
params=VADParams(
|
|
confidence=float(vad.get("confidence", DEFAULT_VAD["confidence"])),
|
|
start_secs=float(_value(vad, "start_secs", "startSecs", 0.2)),
|
|
stop_secs=float(_value(vad, "stop_secs", "stopSecs", 0.2)),
|
|
min_volume=float(_value(vad, "min_volume", "minVolume", 0.6)),
|
|
)
|
|
)
|
|
|
|
|
|
def create_user_turn_strategies(
|
|
turn_config: dict[str, Any], *, enable_interruptions: bool
|
|
) -> UserTurnStrategies:
|
|
barge_in = _section(turn_config, "barge_in", "bargeIn")
|
|
start = []
|
|
strategy = barge_in.get("strategy", "vad")
|
|
if strategy == "vad":
|
|
start.append(VADUserTurnStartStrategy(enable_interruptions=enable_interruptions))
|
|
else:
|
|
start.append(
|
|
TranscriptionUserTurnStartStrategy(enable_interruptions=enable_interruptions)
|
|
)
|
|
|
|
detection = _section(turn_config, "turn_detection", "turnDetection")
|
|
if detection.get("strategy", DEFAULT_TURN_DETECTION["strategy"]) == "smart_turn":
|
|
stop = [
|
|
TurnAnalyzerUserTurnStopStrategy(
|
|
turn_analyzer=LocalSmartTurnAnalyzerV3(),
|
|
wait_for_transcript=True,
|
|
)
|
|
]
|
|
else:
|
|
stop = [
|
|
SpeechTimeoutUserTurnStopStrategy(
|
|
user_speech_timeout=float(
|
|
_value(
|
|
detection,
|
|
"silence_timeout_secs",
|
|
"silenceTimeoutSecs",
|
|
0.6,
|
|
)
|
|
),
|
|
wait_for_transcript=True,
|
|
)
|
|
]
|
|
return UserTurnStrategies(start=start, stop=stop)
|