Merge branch 'main' into sarvam/tts-v3

CHANGELOG.md

@@ -11,6 +11,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Added support for `bulbul:v3` model in `SarvamTTSService` and `SarvamHttpTTSService`.
 
+- Added `keyterms_prompt` parameter to `AssemblyAIConnectionParams`.
+
+- Added `speech_model` parameter to `AssemblyAIConnectionParams` to access the multilingual model.
+- 
+- Added support for trickle ICE to the `SmallWebRTCTransport`.
+
 - Added support for updating `OpenAITTSService` settings (`instructions` and
   `speed`) at runtime via `TTSUpdateSettingsFrame`.
 
@@ -33,11 +39,17 @@ 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 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
   task that it's executing a `PipelineRunner` cleanly. Also,
@@ -59,6 +71,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

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 = []

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,10 @@ 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,
         )
@@ -211,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)
 
@@ -256,6 +270,74 @@ 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"}
+
+    @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."""
@@ -495,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:
@@ -584,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.")

src/pipecat/services/assemblyai/models.py

@@ -108,6 +108,8 @@ 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.
+        speech_model: Select between English and multilingual models. Defaults to "universal-streaming-english".
     """
 
     sample_rate: int = 16000
@@ -117,3 +119,7 @@ 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
+    speech_model: Literal["universal-streaming-english", "universal-streaming-multilingual"] = (
+        "universal-streaming-english"
+    )

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 = {
+        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:
             query_string = urlencode(params)
             return f"{self._api_endpoint_base_url}?{query_string}"

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, 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
 

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
             ]

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)

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()]