From 9ddec0f8b491cdbe4513a494449eaaa9734dde57 Mon Sep 17 00:00:00 2001 From: nbyers-altira Date: Mon, 13 Oct 2025 10:44:25 -0400 Subject: [PATCH 01/10] is_final is not part of the segmented _handle_transcription function signature --- src/pipecat/services/riva/stt.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index d00eb4f42..f077355f4 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -647,7 +647,7 @@ class RivaSegmentedSTTService(SegmentedSTTService): ) transcription_found = True - await self._handle_transcription(text, True, self._language_enum) + await self._handle_transcription(text, self._language_enum) if not transcription_found: logger.debug("No transcription results found in Riva response") From cc66ac14f1699357bc2e39be9adac376721bbcac Mon Sep 17 00:00:00 2001 From: nbyers-altira Date: Mon, 13 Oct 2025 10:48:41 -0400 Subject: [PATCH 02/10] add is_final to segmented func. sig. instead so tracing is consistent --- src/pipecat/services/riva/stt.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index f077355f4..ccd166536 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -583,7 +583,7 @@ class RivaSegmentedSTTService(SegmentedSTTService): self._config.language_code = self._language @traced_stt - async def _handle_transcription(self, transcript: str, language: Optional[Language] = None): + async def _handle_transcription(self, transcript: str, is_final: bool, language: Optional[Language] = None): """Handle a transcription result with tracing.""" pass @@ -647,7 +647,7 @@ class RivaSegmentedSTTService(SegmentedSTTService): ) transcription_found = True - await self._handle_transcription(text, self._language_enum) + await self._handle_transcription(text, True, self._language_enum) if not transcription_found: logger.debug("No transcription results found in Riva response") From d16c36c56de1e3289594c4b2081b65c3425c0d5a Mon Sep 17 00:00:00 2001 From: dan-ince-aai Date: Wed, 15 Oct 2025 14:27:52 +0100 Subject: [PATCH 03/10] feat: add keyterms_prompt to AssemblyAI service --- src/pipecat/services/assemblyai/models.py | 2 ++ src/pipecat/services/assemblyai/stt.py | 15 ++++++++++----- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/src/pipecat/services/assemblyai/models.py b/src/pipecat/services/assemblyai/models.py index b34ec554d..d263d113d 100644 --- a/src/pipecat/services/assemblyai/models.py +++ b/src/pipecat/services/assemblyai/models.py @@ -108,6 +108,7 @@ class AssemblyAIConnectionParams(BaseModel): 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. 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 @@ -117,3 +118,4 @@ class AssemblyAIConnectionParams(BaseModel): end_of_turn_confidence_threshold: Optional[float] = None min_end_of_turn_silence_when_confident: Optional[int] = None max_turn_silence: Optional[int] = None + keyterms_prompt: Optional[List[str]] = None diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index aa2fc36bc..d87261f86 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -174,11 +174,16 @@ class AssemblyAISTTService(STTService): def _build_ws_url(self) -> str: """Build WebSocket URL with query parameters using urllib.parse.urlencode.""" - params = { - k: str(v).lower() if isinstance(v, bool) else v - for k, v in self._connection_params.model_dump().items() - if v is not None - } + params = {} + for k, v in self._connection_params.model_dump().items(): + 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: query_string = urlencode(params) return f"{self._api_endpoint_base_url}?{query_string}" From d5f2dcfac0a020d8077988de71be47b210d90211 Mon Sep 17 00:00:00 2001 From: dan-ince-aai Date: Fri, 17 Oct 2025 11:32:06 +0100 Subject: [PATCH 04/10] lint --- src/pipecat/services/assemblyai/stt.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index d87261f86..c255cd4c4 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -183,7 +183,7 @@ class AssemblyAISTTService(STTService): params[k] = str(v).lower() else: params[k] = v - + if params: query_string = urlencode(params) return f"{self._api_endpoint_base_url}?{query_string}" From 3ba6b5565949023f0f31beac883e114e1b0df7a4 Mon Sep 17 00:00:00 2001 From: dan-ince-aai Date: Fri, 17 Oct 2025 11:38:03 +0100 Subject: [PATCH 05/10] feat: multilingual + changelog updates --- CHANGELOG.md | 4 ++++ src/pipecat/services/assemblyai/models.py | 4 ++++ 2 files changed, 8 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 424d6a420..45131e331 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `keyterms_prompt` parameter to `AssemblyAIConnectionParams`. + +- Added `speech_model` parameter to `AssemblyAIConnectionParams` to access the multilingual model. + - The runner `--folder` argument now supports downloading files from subdirectories. diff --git a/src/pipecat/services/assemblyai/models.py b/src/pipecat/services/assemblyai/models.py index d263d113d..52ea87d87 100644 --- a/src/pipecat/services/assemblyai/models.py +++ b/src/pipecat/services/assemblyai/models.py @@ -109,6 +109,7 @@ class AssemblyAIConnectionParams(BaseModel): 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. keyterms_prompt: List of key terms to guide transcription. Will be JSON serialized before sending. + speech_model: Select between English and multilingual models. Defaults to "universal-streaming-english". """ sample_rate: int = 16000 @@ -119,3 +120,6 @@ class AssemblyAIConnectionParams(BaseModel): min_end_of_turn_silence_when_confident: Optional[int] = None max_turn_silence: Optional[int] = None keyterms_prompt: Optional[List[str]] = None + speech_model: Literal["universal-streaming-english", "universal-streaming-multilingual"] = ( + "universal-streaming-english" + ) From 3e8a7cc2544b98b995917e9bddb11b5fdcdf6c7b Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 17 Oct 2025 08:57:45 -0300 Subject: [PATCH 06/10] Adding support for trickle ICE to the SmallWebRTCTransport. --- CHANGELOG.md | 2 + src/pipecat/runner/run.py | 8 ++++ .../transports/smallwebrtc/connection.py | 5 +++ .../transports/smallwebrtc/request_handler.py | 42 +++++++++++++++++++ 4 files changed, 57 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 938b175c1..6d581ccce 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added support for trickle ICE to the `SmallWebRTCTransport`. + - Added support for updating `OpenAITTSService` settings (`instructions` and `speed`) at runtime via `TTSUpdateSettingsFrame`. diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index a3e2984e8..9b1d233f6 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -204,6 +204,7 @@ def _setup_webrtc_routes( from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection from pipecat.transports.smallwebrtc.request_handler import ( + SmallWebRTCPatchRequest, SmallWebRTCRequest, SmallWebRTCRequestHandler, ) @@ -256,6 +257,13 @@ def _setup_webrtc_routes( ) return answer + @app.patch("/api/offer") + async def ice_candidate(request: SmallWebRTCPatchRequest): + """Handle WebRTC new ice candidate requests.""" + logger.debug(f"Received patch request: {request}") + await small_webrtc_handler.handle_patch_request(request) + return {"status": "success"} + @asynccontextmanager async def smallwebrtc_lifespan(app: FastAPI): """Manage FastAPI application lifecycle and cleanup connections.""" diff --git a/src/pipecat/transports/smallwebrtc/connection.py b/src/pipecat/transports/smallwebrtc/connection.py index c77f4e77e..60dd7798c 100644 --- a/src/pipecat/transports/smallwebrtc/connection.py +++ b/src/pipecat/transports/smallwebrtc/connection.py @@ -689,3 +689,8 @@ class SmallWebRTCConnection(BaseObject): )() if track: track.set_enabled(signalling_message.enabled) + + async def add_ice_candidate(self, candidate): + """Handle incoming ICE candidates.""" + logger.debug(f"Adding remote candidate: {candidate}") + await self.pc.addIceCandidate(candidate) diff --git a/src/pipecat/transports/smallwebrtc/request_handler.py b/src/pipecat/transports/smallwebrtc/request_handler.py index 00d9ebb6a..b2c02a03e 100644 --- a/src/pipecat/transports/smallwebrtc/request_handler.py +++ b/src/pipecat/transports/smallwebrtc/request_handler.py @@ -14,6 +14,7 @@ from dataclasses import dataclass from enum import Enum from typing import Any, Awaitable, Callable, Dict, List, Optional +from aiortc.sdp import candidate_from_sdp from fastapi import HTTPException from loguru import logger @@ -39,6 +40,34 @@ class SmallWebRTCRequest: request_data: Optional[Any] = None +@dataclass +class IceCandidate: + """The remote ice candidate object received from the peer connection. + + Parameters: + candidate: The ice candidate patch SDP string (Session Description Protocol). + sdp_mid: The SDP mid for the candidate patch. + sdp_mline_index: The SDP mline index for the candidate patch. + """ + + candidate: str + sdp_mid: str + sdp_mline_index: int + + +@dataclass +class SmallWebRTCPatchRequest: + """Small WebRTC transport session arguments for the runner. + + Parameters: + pc_id: Identifier for the peer connection. + candidates: A list of ICE candidate patches. + """ + + pc_id: str + candidates: List[IceCandidate] + + class ConnectionMode(Enum): """Enum defining the connection handling modes.""" @@ -197,6 +226,19 @@ class SmallWebRTCRequestHandler: logger.debug(f"SmallWebRTC request details: {request}") raise + async def handle_patch_request(self, request: SmallWebRTCPatchRequest): + """Handle a SmallWebRTC patch candidate request.""" + peer_connection = self._pcs_map.get(request.pc_id) + + if not peer_connection: + raise HTTPException(status_code=404, detail="Peer connection not found") + + for c in request.candidates: + candidate = candidate_from_sdp(c.candidate) + candidate.sdpMid = c.sdp_mid + candidate.sdpMLineIndex = c.sdp_mline_index + await peer_connection.add_ice_candidate(candidate) + async def close(self): """Clear the connection map.""" coros = [pc.disconnect() for pc in self._pcs_map.values()] From 855f4842dd9f156835db322a07cd3ec1d8f57031 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 17 Oct 2025 10:10:19 -0300 Subject: [PATCH 07/10] Creating the WebRTC routes that mimic the ones provided by Pipecat Cloud. --- src/pipecat/runner/run.py | 84 +++++++++++++++++++++++++++++++++++---- 1 file changed, 77 insertions(+), 7 deletions(-) diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 9b1d233f6..a90c96ac5 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -70,12 +70,14 @@ import asyncio import mimetypes import os import sys +import uuid from contextlib import asynccontextmanager +from http import HTTPMethod from pathlib import Path -from typing import Optional +from typing import Any, Dict, List, Optional, TypedDict import aiohttp -from fastapi.responses import FileResponse +from fastapi.responses import FileResponse, Response from loguru import logger from pipecat.runner.types import ( @@ -202,8 +204,9 @@ def _setup_webrtc_routes( try: from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI - from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection + from pipecat.transports.smallwebrtc.connection import IceServer, SmallWebRTCConnection from pipecat.transports.smallwebrtc.request_handler import ( + IceCandidate, SmallWebRTCPatchRequest, SmallWebRTCRequest, SmallWebRTCRequestHandler, @@ -212,6 +215,16 @@ def _setup_webrtc_routes( logger.error(f"WebRTC transport dependencies not installed: {e}") return + class IceConfig(TypedDict): + iceServers: List[IceServer] + + class StartBotResult(TypedDict, total=False): + sessionId: str + iceConfig: Optional[IceConfig] + + # In-memory store of active sessions: session_id -> session info + active_sessions: Dict[str, Dict[str, Any]] = {} + # Mount the frontend app.mount("/client", SmallWebRTCPrebuiltUI) @@ -264,6 +277,67 @@ def _setup_webrtc_routes( await small_webrtc_handler.handle_patch_request(request) return {"status": "success"} + @app.post("/start") + async def rtvi_start(request: Request): + """Mimic Pipecat Cloud's /start endpoint.""" + # Parse the request body + try: + request_data = await request.json() + logger.debug(f"Received request: {request_data}") + except Exception as e: + logger.error(f"Failed to parse request body: {e}") + request_data = {} + + # Store session info immediately in memory, replicate the behavior expected on Pipecat Cloud + session_id = str(uuid.uuid4()) + active_sessions[session_id] = request_data + + result: StartBotResult = {"sessionId": session_id} + if request_data.get("enableDefaultIceServers"): + result["iceConfig"] = IceConfig( + iceServers=[IceServer(urls="stun:stun.l.google.com:19302")] + ) + + return result + + @app.api_route( + "/sessions/{session_id}/{path:path}", + methods=["GET", "POST", "PUT", "PATCH", "DELETE"], + ) + async def proxy_request( + session_id: str, path: str, request: Request, background_tasks: BackgroundTasks + ): + """Mimic Pipecat Cloud's proxy.""" + active_session = active_sessions.get(session_id) + if not active_session: + return Response(content="Invalid or not-yet-ready session_id", status_code=404) + + if path.endswith("api/offer"): + # Parse the request body and convert to SmallWebRTCRequest + try: + request_data = await request.json() + if request.method == HTTPMethod.POST.value: + webrtc_request = SmallWebRTCRequest( + sdp=request_data["sdp"], + type=request_data["type"], + pc_id=request_data.get("pc_id"), + restart_pc=request_data.get("restart_pc"), + request_data=request_data, + ) + return await offer(webrtc_request, background_tasks) + elif request.method == HTTPMethod.PATCH.value: + patch_request = SmallWebRTCPatchRequest( + pc_id=request_data["pc_id"], + candidates=[IceCandidate(**c) for c in request_data.get("candidates", [])], + ) + return await ice_candidate(patch_request) + except Exception as e: + logger.error(f"Failed to parse WebRTC request: {e}") + return Response(content="Invalid WebRTC request", status_code=400) + + logger.info(f"Received request for path: {path}") + return Response(status_code=200) + @asynccontextmanager async def smallwebrtc_lifespan(app: FastAPI): """Manage FastAPI application lifecycle and cleanup connections.""" @@ -503,8 +577,6 @@ def _setup_daily_routes(app: FastAPI): else: logger.debug("No body data provided in request") - import aiohttp - from pipecat.runner.daily import configure async with aiohttp.ClientSession() as session: @@ -592,8 +664,6 @@ def _setup_telephony_routes(app: FastAPI, *, transport_type: str, proxy: str): async def _run_daily_direct(): """Run Daily bot with direct connection (no FastAPI server).""" try: - import aiohttp - from pipecat.runner.daily import configure except ImportError as e: logger.error("Daily transport dependencies not installed.") From f2c61ac9fd7aa4e626c03a56f19b4cc7ab411a13 Mon Sep 17 00:00:00 2001 From: Sam Sykes Date: Fri, 17 Oct 2025 14:34:37 +0100 Subject: [PATCH 08/10] Fix for AdditionVocabEntry without sounds_like items. --- CHANGELOG.md | 6 ++++++ pyproject.toml | 2 +- src/pipecat/services/speechmatics/stt.py | 2 +- 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d581ccce..209c9cb11 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,9 +33,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `CartesiaSTTService` now inherits from `WebsocketSTTService`. - Package upgrades: + - `openai` upgraded to support up to 2.x.x. - `openpipe` upgraded to support up to 5.x.x. +- `SpeechmaticsSTTService` updated dependencies for `speechmatics-rt>=0.5.0`. + ### Fixed - Fixed multiple pipeline task cancellation issues. `asyncio.CancelledError` is @@ -59,6 +62,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 incorrectly 16-bit aligned audio frames, potentially leading to internal errors or static audio. +- Fixed an issue in `SpeechmaticsSTTService` where `AdditionalVocabEntry` items + needed to have `sounds_like` for the session to start. + ### Other - Added foundational example `47-sentry-metrics.py`, demonstrating how to use the diff --git a/pyproject.toml b/pyproject.toml index 0d94d0fa3..65546311c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -102,7 +102,7 @@ silero = [ "onnxruntime>=1.20.1,<2" ] simli = [ "simli-ai~=0.1.10"] soniox = [ "pipecat-ai[websockets-base]" ] soundfile = [ "soundfile~=0.13.0" ] -speechmatics = [ "speechmatics-rt>=0.4.0" ] +speechmatics = [ "speechmatics-rt>=0.5.0" ] strands = [ "strands-agents>=1.9.1,<2" ] tavus=[] together = [] diff --git a/src/pipecat/services/speechmatics/stt.py b/src/pipecat/services/speechmatics/stt.py index 2c1db2a15..901edb0e8 100644 --- a/src/pipecat/services/speechmatics/stt.py +++ b/src/pipecat/services/speechmatics/stt.py @@ -620,7 +620,7 @@ class SpeechmaticsSTTService(STTService): transcription_config.additional_vocab = [ { "content": e.content, - "sounds_like": e.sounds_like, + **({"sounds_like": e.sounds_like} if e.sounds_like else {}), } for e in self._params.additional_vocab ] From a67a765783dccbf4893d535722d8d15f6c7894e9 Mon Sep 17 00:00:00 2001 From: nbyers-altira Date: Fri, 17 Oct 2025 13:49:52 -0400 Subject: [PATCH 09/10] add changelog, run linter --- CHANGELOG.md | 3 +++ src/pipecat/services/riva/stt.py | 4 +++- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6d581ccce..96406e52f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,6 +38,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue in `RivaSTTService` where a runtime error occurred due to a + mismatch in the _handle_transcription method's signature. + - Fixed multiple pipeline task cancellation issues. `asyncio.CancelledError` is now handled properly in `PipelineTask` making it possible to cancel an asyncio task that it's executing a `PipelineRunner` cleanly. Also, diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index ccd166536..eddd3da9e 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -583,7 +583,9 @@ class RivaSegmentedSTTService(SegmentedSTTService): self._config.language_code = self._language @traced_stt - async def _handle_transcription(self, transcript: str, is_final: bool, language: Optional[Language] = None): + async def _handle_transcription( + self, transcript: str, is_final: bool, language: Optional[Language] = None + ): """Handle a transcription result with tracing.""" pass From a4867f61aa51497c78e2043e55f037519051c9f7 Mon Sep 17 00:00:00 2001 From: nbyers-altira Date: Fri, 17 Oct 2025 13:51:49 -0400 Subject: [PATCH 10/10] be a tad more precise in changelog --- CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 96406e52f..5d8d44887 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -38,8 +38,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Fixed an issue in `RivaSTTService` where a runtime error occurred due to a - mismatch in the _handle_transcription method's signature. +- Fixed an issue in `RivaSegmentedSTTService` where a runtime error occurred due + to a mismatch in the _handle_transcription method's signature. - Fixed multiple pipeline task cancellation issues. `asyncio.CancelledError` is now handled properly in `PipelineTask` making it possible to cancel an asyncio