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

Update optional dependency version ranges and remove SDK dependencies

Browse Source 
Widen version ranges for stable packages (anthropic, azure, deepgram,
groq, livekit, nvidia-riva-client, fastapi, ormsgpack, opentelemetry,
faster-whisper) and add upper bounds to previously uncapped packages
(hume, pyjwt, livekit-api, camb).

Replace CartesiaHttpTTSService's internal use of the Cartesia SDK with
direct aiohttp calls, accepting an optional aiohttp_session parameter.

Replace fal-client SDK calls in FalSTTService and FalImageGenService
with direct HTTP to bypass the SDK's aggressive retry/backoff logic
that caused significant latency regressions.
This commit is contained in:
Mark Backman
2026-03-05 15:06:54 -05:00
parent 05fa727c22
commit 3f97c91983
8 changed files with 246 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/pipecat/services/cartesia/tts.py
View File

@@ -13,6 +13,7 @@ from dataclasses import dataclass, field
from enum import Enum
from typing import Any, AsyncGenerator, List, Literal, Mapping, Optional
import aiohttp
from loguru import logger
from pydantic import BaseModel, Field
@@ -33,13 +34,8 @@ from pipecat.utils.text.base_text_aggregator import BaseTextAggregator
from pipecat.utils.text.skip_tags_aggregator import SkipTagsAggregator
from pipecat.utils.tracing.service_decorators import traced_tts
# Suppress regex warnings from pydub (used by cartesia)
warnings.filterwarnings("ignore", message="invalid escape sequence", category=SyntaxWarning)
# See .env.example for Cartesia configuration needed
try:
from cartesia import AsyncCartesia
from websockets.asyncio.client import connect as websocket_connect
from websockets.protocol import State
except ModuleNotFoundError as e:
@@ -721,6 +717,7 @@ class CartesiaHttpTTSService(TTSService):
model: str = "sonic-3",
base_url: str = "https://api.cartesia.ai",
cartesia_version: str = "2024-11-13",
aiohttp_session: Optional[aiohttp.ClientSession] = None,
sample_rate: Optional[int] = None,
encoding: str = "pcm_s16le",
container: str = "raw",
@@ -735,6 +732,8 @@ class CartesiaHttpTTSService(TTSService):
model: TTS model to use (e.g., "sonic-3").
base_url: Base URL for Cartesia HTTP API.
cartesia_version: API version string for Cartesia service.
aiohttp_session: Optional aiohttp ClientSession for HTTP requests.
If not provided, a session will be created and managed internally.
sample_rate: Audio sample rate. If None, uses default.
encoding: Audio encoding format.
container: Audio container format.
@@ -766,10 +765,8 @@ class CartesiaHttpTTSService(TTSService):
self._base_url = base_url
self._cartesia_version = cartesia_version
self._client = AsyncCartesia(
api_key=api_key,
base_url=base_url,
)
self._session: aiohttp.ClientSession | None = aiohttp_session
self._owns_session = aiohttp_session is None
def can_generate_metrics(self) -> bool:
"""Check if this service can generate processing metrics.
@@ -798,6 +795,14 @@ class CartesiaHttpTTSService(TTSService):
"""
await super().start(frame)
self._settings.output_sample_rate = self.sample_rate
if self._owns_session:
self._session = aiohttp.ClientSession()
async def _close_session(self):
"""Close the HTTP session if we own it."""
if self._owns_session and self._session:
await self._session.close()
self._session = None
async def stop(self, frame: EndFrame):
"""Stop the Cartesia HTTP TTS service.
@@ -806,7 +811,7 @@ class CartesiaHttpTTSService(TTSService):
frame: The end frame.
"""
await super().stop(frame)
await self._client.close()
await self._close_session()
async def cancel(self, frame: CancelFrame):
"""Cancel the Cartesia HTTP TTS service.
@@ -815,7 +820,7 @@ class CartesiaHttpTTSService(TTSService):
frame: The cancel frame.
"""
await super().cancel(frame)
await self._client.close()
await self._close_session()
@traced_tts
async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
@@ -874,8 +879,6 @@ class CartesiaHttpTTSService(TTSService):
yield TTSStartedFrame(context_id=context_id)
session = await self._client._get_session()
headers = {
"Cartesia-Version": self._cartesia_version,
"X-API-Key": self._api_key,
@@ -884,7 +887,7 @@ class CartesiaHttpTTSService(TTSService):
url = f"{self._base_url}/tts/bytes"
async with session.post(url, json=payload, headers=headers) as response:
async with self._session.post(url, json=payload, headers=headers) as response:
if response.status != 200:
error_text = await response.text()
yield ErrorFrame(error=f"Cartesia API error: {error_text}")

28
src/pipecat/services/fal/image.py
View File

@@ -25,13 +25,6 @@ from pipecat.frames.frames import ErrorFrame, Frame, URLImageRawFrame
from pipecat.services.image_service import ImageGenService
from pipecat.services.settings import ImageGenSettings
try:
import fal_client
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use Fal, you need to `pip install pipecat-ai[fal]`.")
raise Exception(f"Missing module: {e}")
@dataclass
class FalImageGenSettings(ImageGenSettings):
@@ -91,6 +84,7 @@ class FalImageGenService(ImageGenService):
super().__init__(settings=FalImageGenSettings(model=model), **kwargs)
self._params = params
self._aiohttp_session = aiohttp_session
self._api_key = key or os.getenv("FAL_KEY", "")
if key:
os.environ["FAL_KEY"] = key
@@ -112,10 +106,22 @@ class FalImageGenService(ImageGenService):
logger.debug(f"Generating image from prompt: {prompt}")
response = await fal_client.run_async(
self._settings.model,
arguments={"prompt": prompt, **self._params.model_dump(exclude_none=True)},
)
headers = {
"Authorization": f"Key {self._api_key}",
"Content-Type": "application/json",
}
payload = {"prompt": prompt, **self._params.model_dump(exclude_none=True)}
async with self._aiohttp_session.post(
f"https://fal.run/{self._settings.model}",
json=payload,
headers=headers,
) as resp:
if resp.status != 200:
error_text = await resp.text()
yield ErrorFrame(error=f"Fal API error ({resp.status}): {error_text}")
return
response = await resp.json()
image_url = response["images"][0]["url"] if response else None

50
src/pipecat/services/fal/stt.py
View File

@@ -10,10 +10,12 @@ This module provides integration with Fal's Wizper API for speech-to-text
transcription using segmented audio processing.
"""
import base64
import os
from dataclasses import dataclass, field
from typing import Any, AsyncGenerator, Optional
from typing import AsyncGenerator, Optional
import aiohttp
from loguru import logger
from pydantic import BaseModel
@@ -25,15 +27,6 @@ from pipecat.transcriptions.language import Language, resolve_language
from pipecat.utils.time import time_now_iso8601
from pipecat.utils.tracing.service_decorators import traced_stt
try:
import fal_client
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use Fal, you need to `pip install pipecat-ai[fal]`. Also, set `FAL_KEY` environment variable."
)
raise Exception(f"Missing module: {e}")
def language_to_fal_language(language: Language) -> Optional[str]:
"""Convert a Language enum to Fal's Wizper language code.
@@ -192,6 +185,7 @@ class FalSTTService(SegmentedSTTService):
self,
*,
api_key: Optional[str] = None,
aiohttp_session: Optional[aiohttp.ClientSession] = None,
sample_rate: Optional[int] = None,
params: Optional[InputParams] = None,
ttfs_p99_latency: Optional[float] = FAL_TTFS_P99,
@@ -201,6 +195,8 @@ class FalSTTService(SegmentedSTTService):
Args:
api_key: Fal API key. If not provided, will check FAL_KEY environment variable.
aiohttp_session: Optional aiohttp ClientSession for HTTP requests.
If not provided, a session will be created and managed internally.
sample_rate: Audio sample rate in Hz. If not provided, uses the pipeline's rate.
params: Configuration parameters for the Wizper API.
ttfs_p99_latency: P99 latency from speech end to final transcript in seconds.
@@ -224,14 +220,14 @@ class FalSTTService(SegmentedSTTService):
**kwargs,
)
if api_key:
os.environ["FAL_KEY"] = api_key
elif "FAL_KEY" not in os.environ:
self._api_key = api_key or os.getenv("FAL_KEY", "")
if not self._api_key:
raise ValueError(
"FAL_KEY must be provided either through api_key parameter or environment variable"
)
self._fal_client = fal_client.AsyncClient(key=api_key or os.getenv("FAL_KEY"))
self._session: aiohttp.ClientSession | None = aiohttp_session
self._owns_session = aiohttp_session is None
def can_generate_metrics(self) -> bool:
"""Check if the service can generate processing metrics.
@@ -275,12 +271,26 @@ class FalSTTService(SegmentedSTTService):
try:
await self.start_processing_metrics()
# Send to Fal directly (audio is already in WAV format from base class)
data_uri = fal_client.encode(audio, "audio/x-wav")
response = await self._fal_client.run(
"fal-ai/wizper",
arguments={"audio_url": data_uri, **self._settings.given_fields()},
)
if not self._session:
self._session = aiohttp.ClientSession()
data_uri = f"data:audio/x-wav;base64,{base64.b64encode(audio).decode()}"
payload = {"audio_url": data_uri, **self._settings.given_fields()}
headers = {
"Authorization": f"Key {self._api_key}",
"Content-Type": "application/json",
}
async with self._session.post(
"https://fal.run/fal-ai/wizper",
json=payload,
headers=headers,
) as resp:
if resp.status != 200:
error_text = await resp.text()
yield ErrorFrame(error=f"Fal API error ({resp.status}): {error_text}")
return
response = await resp.json()
if response and "text" in response:
text = response["text"].strip()