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

feat: add keyterms_prompt to AssemblyAI service

Browse Source
This commit is contained in:
dan-ince-aai
2025-10-15 14:27:52 +01:00
parent 8fe3bcd484
commit d16c36c56d
2 changed files with 12 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/pipecat/services/assemblyai/models.py
View File

@@ -108,6 +108,7 @@ class AssemblyAIConnectionParams(BaseModel):
end_of_turn_confidence_threshold: Confidence threshold for end-of-turn detection. end_of_turn_confidence_threshold: Confidence threshold for end-of-turn detection.
min_end_of_turn_silence_when_confident: Minimum silence duration when confident about end-of-turn. min_end_of_turn_silence_when_confident: Minimum silence duration when confident about end-of-turn.
max_turn_silence: Maximum silence duration before forcing end-of-turn. max_turn_silence: Maximum silence duration before forcing end-of-turn.
keyterms_prompt: List of key terms to guide transcription. Will be JSON serialized before sending.
""" """
sample_rate: int = 16000 sample_rate: int = 16000
@@ -117,3 +118,4 @@ class AssemblyAIConnectionParams(BaseModel):
end_of_turn_confidence_threshold: Optional[float] = None end_of_turn_confidence_threshold: Optional[float] = None
min_end_of_turn_silence_when_confident: Optional[int] = None min_end_of_turn_silence_when_confident: Optional[int] = None
max_turn_silence: Optional[int] = None max_turn_silence: Optional[int] = None
keyterms_prompt: Optional[List[str]] = None

15
src/pipecat/services/assemblyai/stt.py
View File

@@ -174,11 +174,16 @@ class AssemblyAISTTService(STTService):
def _build_ws_url(self) -> str: def _build_ws_url(self) -> str:
"""Build WebSocket URL with query parameters using urllib.parse.urlencode.""" """Build WebSocket URL with query parameters using urllib.parse.urlencode."""
params = { params = {}
k: str(v).lower() if isinstance(v, bool) else v for k, v in self._connection_params.model_dump().items():
for k, v in self._connection_params.model_dump().items() if v is not None:
if v is not None if k == "keyterms_prompt":
} params[k] = json.dumps(v)
elif isinstance(v, bool):
params[k] = str(v).lower()
else:
params[k] = v
if params: if params:
query_string = urlencode(params) query_string = urlencode(params)
return f"{self._api_endpoint_base_url}?{query_string}" return f"{self._api_endpoint_base_url}?{query_string}"