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

Merge branch 'main' into smart_turn

Browse Source
This commit is contained in:
Filipi Fuchter
2025-04-17 09:36:30 -03:00
parent 3f0688aefa 5be3e6979e
commit 230d2f80fa
103 changed files with 1021 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -6,6 +6,7 @@
import asyncio
from abc import abstractmethod
from dataclasses import dataclass
from typing import Dict, List, Literal, Set
from loguru import logger
@@ -46,6 +47,16 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.utils.time import time_now_iso8601
@dataclass
class LLMUserAggregatorParams:
aggregation_timeout: float = 1.0
@dataclass
class LLMAssistantAggregatorParams:
expect_stripped_words: bool = True
class LLMFullResponseAggregator(FrameProcessor):
"""This is an LLM aggregator that aggregates a full LLM completion. It
aggregates LLM text frames (tokens) received between
@@ -230,11 +241,23 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
def __init__(
self,
context: OpenAILLMContext,
aggregation_timeout: float = 1.0,
*,
params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
**kwargs,
):
super().__init__(context=context, role="user", **kwargs)
self._aggregation_timeout = aggregation_timeout
self._params = params
if "aggregation_timeout" in kwargs:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Parameter 'aggregation_timeout' is deprecated, use 'params' instead.",
DeprecationWarning,
)
self._params.aggregation_timeout = kwargs["aggregation_timeout"]
self._seen_interim_results = False
self._user_speaking = False
@@ -357,7 +380,9 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
async def _aggregation_task_handler(self):
while True:
try:
await asyncio.wait_for(self._aggregation_event.wait(), self._aggregation_timeout)
await asyncio.wait_for(
self._aggregation_event.wait(), self._params.aggregation_timeout
)
await self._maybe_push_bot_interruption()
except asyncio.TimeoutError:
if not self._user_speaking:
@@ -394,9 +419,27 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
"""
def __init__(self, context: OpenAILLMContext, *, expect_stripped_words: bool = True, **kwargs):
def __init__(
self,
context: OpenAILLMContext,
*,
params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
**kwargs,
):
super().__init__(context=context, role="assistant", **kwargs)
self._expect_stripped_words = expect_stripped_words
self._params = params
if "expect_stripped_words" in kwargs:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Parameter 'expect_stripped_words' is deprecated, use 'params' instead.",
DeprecationWarning,
)
self._params.expect_stripped_words = kwargs["expect_stripped_words"]
self._started = 0
self._function_calls_in_progress: Dict[str, FunctionCallInProgressFrame] = {}
@@ -558,7 +601,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
if not self._started:
return
if self._expect_stripped_words:
if self._params.expect_stripped_words:
self._aggregation += f" {frame.text}" if self._aggregation else frame.text
else:
self._aggregation += frame.text
@@ -572,8 +615,14 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
class LLMUserResponseAggregator(LLMUserContextAggregator):
def __init__(self, messages: List[dict] = [], **kwargs):
super().__init__(context=OpenAILLMContext(messages), **kwargs)
def __init__(
self,
messages: List[dict] = [],
*,
params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
**kwargs,
):
super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs)
async def push_aggregation(self):
if len(self._aggregation) > 0:
@@ -588,8 +637,14 @@ class LLMUserResponseAggregator(LLMUserContextAggregator):
class LLMAssistantResponseAggregator(LLMAssistantContextAggregator):
def __init__(self, messages: List[dict] = [], **kwargs):
super().__init__(context=OpenAILLMContext(messages), **kwargs)
def __init__(
self,
messages: List[dict] = [],
*,
params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
**kwargs,
):
super().__init__(context=OpenAILLMContext(messages), params=params, **kwargs)
async def push_aggregation(self):
if len(self._aggregation) > 0:

27
src/pipecat/services/anthropic/llm.py
View File

@@ -11,7 +11,7 @@ import io
import json
import re
from dataclasses import dataclass
from typing import Any, Dict, List, Mapping, Optional, Union
from typing import Any, Dict, List, Optional, Union
import httpx
from loguru import logger
@@ -35,7 +35,9 @@ from pipecat.frames.frames import (
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMAssistantContextAggregator,
LLMUserAggregatorParams,
LLMUserContextAggregator,
)
from pipecat.processors.aggregators.openai_llm_context import (
@@ -49,10 +51,7 @@ try:
from anthropic import NOT_GIVEN, AsyncAnthropic, NotGiven
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`. "
+ "Also, set `ANTHROPIC_API_KEY` environment variable."
)
logger.error("In order to use Anthropic, you need to `pip install pipecat-ai[anthropic]`.")
raise Exception(f"Missing module: {e}")
@@ -120,8 +119,8 @@ class AnthropicLLMService(LLMService):
self,
context: OpenAILLMContext,
*,
user_kwargs: Mapping[str, Any] = {},
assistant_kwargs: Mapping[str, Any] = {},
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> AnthropicContextAggregatorPair:
"""Create an instance of AnthropicContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
@@ -129,12 +128,10 @@ class AnthropicLLMService(LLMService):
Args:
context (OpenAILLMContext): The LLM context.
user_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the user context aggregator constructor. Defaults
to an empty mapping.
assistant_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the assistant context aggregator
constructor. Defaults to an empty mapping.
user_params (LLMUserAggregatorParams, optional): User aggregator
parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User
aggregator parameters.
Returns:
AnthropicContextAggregatorPair: A pair of context aggregators, one
@@ -146,8 +143,8 @@ class AnthropicLLMService(LLMService):
if isinstance(context, OpenAILLMContext):
context = AnthropicLLMContext.from_openai_context(context)
user = AnthropicUserContextAggregator(context, **user_kwargs)
assistant = AnthropicAssistantContextAggregator(context, **assistant_kwargs)
user = AnthropicUserContextAggregator(context, params=user_params)
assistant = AnthropicAssistantContextAggregator(context, params=assistant_params)
return AnthropicContextAggregatorPair(_user=user, _assistant=assistant)
async def _process_context(self, context: OpenAILLMContext):

6
src/pipecat/services/aws/tts.py
View File

@@ -231,9 +231,9 @@ class PollyTTSService(TTSService):
yield TTSStartedFrame()
chunk_size = 8192
for i in range(0, len(audio_data), chunk_size):
chunk = audio_data[i : i + chunk_size]
CHUNK_SIZE = 1024
for i in range(0, len(audio_data), CHUNK_SIZE):
chunk = audio_data[i : i + CHUNK_SIZE]
if len(chunk) > 0:
await self.stop_ttfb_metrics()
frame = TTSAudioRawFrame(chunk, self.sample_rate, 1)

14
src/pipecat/services/deepgram/stt.py
View File

@@ -45,6 +45,7 @@ class DeepgramSTTService(STTService):
*,
api_key: str,
url: str = "",
base_url: str = "",
sample_rate: Optional[int] = None,
live_options: Optional[LiveOptions] = None,
addons: Optional[Dict] = None,
@@ -53,6 +54,17 @@ class DeepgramSTTService(STTService):
sample_rate = sample_rate or (live_options.sample_rate if live_options else None)
super().__init__(sample_rate=sample_rate, **kwargs)
if url:
import warnings
with warnings.catch_warnings():
warnings.simplefilter("always")
warnings.warn(
"Parameter 'url' is deprecated, use 'base_url' instead.",
DeprecationWarning,
)
base_url = url
default_options = LiveOptions(
encoding="linear16",
language=Language.EN,
@@ -81,7 +93,7 @@ class DeepgramSTTService(STTService):
self._client = DeepgramClient(
api_key,
config=DeepgramClientOptions(
url=url,
url=base_url,
options={"keepalive": "true"}, # verbose=logging.DEBUG
),
)

12
src/pipecat/services/deepgram/tts.py
View File

@@ -4,7 +4,6 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -19,7 +18,7 @@ from pipecat.frames.frames import (
from pipecat.services.tts_service import TTSService
try:
from deepgram import DeepgramClient, SpeakOptions
from deepgram import DeepgramClient, DeepgramClientOptions, SpeakOptions
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use Deepgram, you need to `pip install pipecat-ai[deepgram]`.")
@@ -32,6 +31,7 @@ class DeepgramTTSService(TTSService):
*,
api_key: str,
voice: str = "aura-helios-en",
base_url: str = "",
sample_rate: Optional[int] = None,
encoding: str = "linear16",
**kwargs,
@@ -42,7 +42,9 @@ class DeepgramTTSService(TTSService):
"encoding": encoding,
}
self.set_voice(voice)
self._deepgram_client = DeepgramClient(api_key=api_key)
client_options = DeepgramClientOptions(url=base_url)
self._deepgram_client = DeepgramClient(api_key, config=client_options)
def can_generate_metrics(self) -> bool:
return True
@@ -60,8 +62,8 @@ class DeepgramTTSService(TTSService):
try:
await self.start_ttfb_metrics()
response = await asyncio.to_thread(
self._deepgram_client.speak.v("1").stream, {"text": text}, options
response = await self._deepgram_client.speak.asyncrest.v("1").stream_memory(
{"text": text}, options
)
await self.start_tts_usage_metrics(text)

194
src/pipecat/services/elevenlabs/tts.py
View File

@@ -18,6 +18,7 @@ from pipecat.frames.frames import (
EndFrame,
ErrorFrame,
Frame,
LLMFullResponseEndFrame,
StartFrame,
StartInterruptionFrame,
TTSAudioRawFrame,
@@ -25,7 +26,7 @@ from pipecat.frames.frames import (
TTSStoppedFrame,
)
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.tts_service import InterruptibleWordTTSService, TTSService
from pipecat.services.tts_service import InterruptibleWordTTSService, WordTTSService
from pipecat.transcriptions.language import Language
# See .env.example for ElevenLabs configuration needed
@@ -441,8 +442,8 @@ class ElevenLabsTTSService(InterruptibleWordTTSService):
logger.error(f"{self} exception: {e}")
class ElevenLabsHttpTTSService(TTSService):
"""ElevenLabs Text-to-Speech service using HTTP streaming.
class ElevenLabsHttpTTSService(WordTTSService):
"""ElevenLabs Text-to-Speech service using HTTP streaming with word timestamps.
Args:
api_key: ElevenLabs API key
@@ -475,7 +476,13 @@ class ElevenLabsHttpTTSService(TTSService):
params: InputParams = InputParams(),
**kwargs,
):
super().__init__(sample_rate=sample_rate, **kwargs)
super().__init__(
aggregate_sentences=True,
push_text_frames=False,
push_stop_frames=True,
sample_rate=sample_rate,
**kwargs,
)
self._api_key = api_key
self._base_url = base_url
@@ -498,34 +505,136 @@ class ElevenLabsHttpTTSService(TTSService):
self._output_format = "" # initialized in start()
self._voice_settings = self._set_voice_settings()
# Track cumulative time to properly sequence word timestamps across utterances
self._cumulative_time = 0
self._started = False
# Store previous text for context within a turn
self._previous_text = ""
def language_to_service_language(self, language: Language) -> Optional[str]:
"""Convert pipecat Language to ElevenLabs language code."""
return language_to_elevenlabs_language(language)
def can_generate_metrics(self) -> bool:
"""Indicate that this service can generate usage metrics."""
return True
def _set_voice_settings(self):
return build_elevenlabs_voice_settings(self._settings)
def _reset_state(self):
"""Reset internal state variables."""
self._cumulative_time = 0
self._started = False
self._previous_text = ""
logger.debug(f"{self}: Reset internal state")
async def start(self, frame: StartFrame):
"""Initialize the service upon receiving a StartFrame."""
await super().start(frame)
self._output_format = output_format_from_sample_rate(self.sample_rate)
self._reset_state()
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using ElevenLabs streaming API.
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
await super().push_frame(frame, direction)
if isinstance(frame, (StartInterruptionFrame, TTSStoppedFrame)):
# Reset timing on interruption or stop
self._reset_state()
if isinstance(frame, TTSStoppedFrame):
await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)])
elif isinstance(frame, LLMFullResponseEndFrame):
# End of turn - reset previous text
self._previous_text = ""
def calculate_word_times(self, alignment_info: Mapping[str, Any]) -> List[Tuple[str, float]]:
"""Calculate word timing from character alignment data.
Example input data:
{
"characters": [" ", "H", "e", "l", "l", "o", " ", "w", "o", "r", "l", "d"],
"character_start_times_seconds": [0.0, 0.1, 0.15, 0.2, 0.25, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9],
"character_end_times_seconds": [0.1, 0.15, 0.2, 0.25, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
}
Would produce word times (with cumulative_time=0):
[("Hello", 0.1), ("world", 0.5)]
Args:
text: The text to convert to speech
alignment_info: Character timing data from ElevenLabs
Returns:
List of (word, timestamp) pairs
"""
chars = alignment_info.get("characters", [])
char_start_times = alignment_info.get("character_start_times_seconds", [])
if not chars or not char_start_times or len(chars) != len(char_start_times):
logger.warning(
f"Invalid alignment data: chars={len(chars)}, times={len(char_start_times)}"
)
return []
# Build the words and find their start times
words = []
word_start_times = []
current_word = ""
first_char_idx = -1
for i, char in enumerate(chars):
if char == " ":
if current_word: # Only add non-empty words
words.append(current_word)
# Use time of the first character of the word, offset by cumulative time
word_start_times.append(
self._cumulative_time + char_start_times[first_char_idx]
)
current_word = ""
first_char_idx = -1
else:
if not current_word: # This is the first character of a new word
first_char_idx = i
current_word += char
# Don't forget the last word if there's no trailing space
if current_word and first_char_idx >= 0:
words.append(current_word)
word_start_times.append(self._cumulative_time + char_start_times[first_char_idx])
# Create word-time pairs
word_times = list(zip(words, word_start_times))
return word_times
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using ElevenLabs streaming API with timestamps.
Makes a request to the ElevenLabs API to generate audio and timing data.
Tracks the duration of each utterance to ensure correct sequencing.
Includes previous text as context for better prosody continuity.
Args:
text: Text to convert to speech
Yields:
Frames containing audio data and status information
Audio and control frames
"""
logger.debug(f"{self}: Generating TTS [{text}]")
url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream"
# Use the with-timestamps endpoint
url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream/with-timestamps"
payload: Dict[str, Union[str, Dict[str, Union[float, bool]]]] = {
"text": text,
"model_id": self._model_name,
}
# Include previous text as context if available
if self._previous_text:
payload["previous_text"] = self._previous_text
if self._voice_settings:
payload["voice_settings"] = self._voice_settings
@@ -550,8 +659,6 @@ class ElevenLabsHttpTTSService(TTSService):
if self._settings["optimize_streaming_latency"] is not None:
params["optimize_streaming_latency"] = self._settings["optimize_streaming_latency"]
logger.debug(f"ElevenLabs request - payload: {payload}, params: {params}")
try:
await self.start_ttfb_metrics()
@@ -566,17 +673,66 @@ class ElevenLabsHttpTTSService(TTSService):
await self.start_tts_usage_metrics(text)
# Process the streaming response
CHUNK_SIZE = 1024
# Start TTS sequence if not already started
if not self._started:
self.start_word_timestamps()
yield TTSStartedFrame()
self._started = True
# Track the duration of this utterance based on the last character's end time
utterance_duration = 0
async for line in response.content:
line_str = line.decode("utf-8").strip()
if not line_str:
continue
try:
# Parse the JSON object
data = json.loads(line_str)
# Process audio if present
if data and "audio_base64" in data:
await self.stop_ttfb_metrics()
audio = base64.b64decode(data["audio_base64"])
yield TTSAudioRawFrame(audio, self.sample_rate, 1)
# Process alignment if present
if data and "alignment" in data:
alignment = data["alignment"]
if alignment: # Ensure alignment is not None
# Get end time of the last character in this chunk
char_end_times = alignment.get("character_end_times_seconds", [])
if char_end_times:
chunk_end_time = char_end_times[-1]
# Update to the longest end time seen so far
utterance_duration = max(utterance_duration, chunk_end_time)
# Calculate word timestamps
word_times = self.calculate_word_times(alignment)
if word_times:
await self.add_word_timestamps(word_times)
except json.JSONDecodeError as e:
logger.warning(f"Failed to parse JSON from stream: {e}")
continue
except Exception as e:
logger.error(f"Error processing response: {e}", exc_info=True)
continue
# After processing all chunks, add the total utterance duration
# to the cumulative time to ensure next utterance starts after this one
if utterance_duration > 0:
self._cumulative_time += utterance_duration
# Append the current text to previous_text for context continuity
# Only add a space if there's already text
if self._previous_text:
self._previous_text += " " + text
else:
self._previous_text = text
yield TTSStartedFrame()
async for chunk in response.content.iter_chunked(CHUNK_SIZE):
if len(chunk) > 0:
await self.stop_ttfb_metrics()
yield TTSAudioRawFrame(chunk, self.sample_rate, 1)
except Exception as e:
logger.error(f"Error in run_tts: {e}")
yield ErrorFrame(error=str(e))
finally:
await self.stop_ttfb_metrics()
yield TTSStoppedFrame()
# Let the parent class handle TTSStoppedFrame

37
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

@@ -10,9 +10,8 @@ import json
import time
from dataclasses import dataclass
from enum import Enum
from typing import Any, Dict, List, Mapping, Optional, Union
from typing import Any, Dict, List, Optional, Union
import websockets
from loguru import logger
from pydantic import BaseModel, Field
@@ -45,6 +44,10 @@ from pipecat.frames.frames import (
UserStoppedSpeakingFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
@@ -61,6 +64,13 @@ from pipecat.utils.time import time_now_iso8601
from . import events
from .audio_transcriber import AudioTranscriber
try:
import websockets
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.")
raise Exception(f"Missing module: {e}")
def language_to_gemini_language(language: Language) -> Optional[str]:
"""Maps a Language enum value to a Gemini Live supported language code.
@@ -871,8 +881,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
self,
context: OpenAILLMContext,
*,
user_kwargs: Mapping[str, Any] = {},
assistant_kwargs: Mapping[str, Any] = {},
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> GeminiMultimodalLiveContextAggregatorPair:
"""Create an instance of GeminiMultimodalLiveContextAggregatorPair from
an OpenAILLMContext. Constructor keyword arguments for both the user and
@@ -880,12 +890,10 @@ class GeminiMultimodalLiveLLMService(LLMService):
Args:
context (OpenAILLMContext): The LLM context.
user_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the user context aggregator constructor. Defaults
to an empty mapping.
assistant_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the assistant context aggregator
constructor. Defaults to an empty mapping.
user_params (LLMUserAggregatorParams, optional): User aggregator
parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User
aggregator parameters.
Returns:
GeminiMultimodalLiveContextAggregatorPair: A pair of context
@@ -896,11 +904,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
context.set_llm_adapter(self.get_llm_adapter())
GeminiMultimodalLiveContext.upgrade(context)
user = GeminiMultimodalLiveUserContextAggregator(context, **user_kwargs)
user = GeminiMultimodalLiveUserContextAggregator(context, params=user_params)
default_assistant_kwargs = {"expect_stripped_words": True}
default_assistant_kwargs.update(assistant_kwargs)
assistant = GeminiMultimodalLiveAssistantContextAggregator(
context, **default_assistant_kwargs
)
assistant_params.expect_stripped_words = True
assistant = GeminiMultimodalLiveAssistantContextAggregator(context, params=assistant_params)
return GeminiMultimodalLiveContextAggregatorPair(_user=user, _assistant=assistant)

38
src/pipecat/services/google/llm.py
View File

@@ -9,21 +9,14 @@ import io
import json
import os
import uuid
from google.api_core.exceptions import DeadlineExceeded
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
from dataclasses import dataclass
from typing import Any, Dict, List, Mapping, Optional, Union
from typing import Any, Dict, List, Optional
from loguru import logger
from PIL import Image
from pydantic import BaseModel, Field
from pipecat.adapters.services.gemini_adapter import GeminiLLMAdapter
from pipecat.frames.frames import (
AudioRawFrame,
Frame,
@@ -39,6 +32,10 @@ from pipecat.frames.frames import (
VisionImageRawFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
@@ -51,11 +48,14 @@ from pipecat.services.openai.llm import (
OpenAIUserContextAggregator,
)
# Suppress gRPC fork warnings
os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false"
try:
import google.ai.generativelanguage as glm
import google.generativeai as gai
from google.api_core.exceptions import DeadlineExceeded
from google.generativeai.types import GenerationConfig
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use Google AI, you need to `pip install pipecat-ai[google]`.")
@@ -686,8 +686,8 @@ class GoogleLLMService(LLMService):
self,
context: OpenAILLMContext,
*,
user_kwargs: Mapping[str, Any] = {},
assistant_kwargs: Mapping[str, Any] = {},
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> GoogleContextAggregatorPair:
"""Create an instance of GoogleContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
@@ -695,12 +695,10 @@ class GoogleLLMService(LLMService):
Args:
context (OpenAILLMContext): The LLM context.
user_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the user context aggregator constructor. Defaults
to an empty mapping.
assistant_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the assistant context aggregator
constructor. Defaults to an empty mapping.
user_params (LLMUserAggregatorParams, optional): User aggregator
parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User
aggregator parameters.
Returns:
GoogleContextAggregatorPair: A pair of context aggregators, one for
@@ -712,6 +710,6 @@ class GoogleLLMService(LLMService):
if isinstance(context, OpenAILLMContext):
context = GoogleLLMContext.upgrade_to_google(context)
user = GoogleUserContextAggregator(context, **user_kwargs)
assistant = GoogleAssistantContextAggregator(context, **assistant_kwargs)
user = GoogleUserContextAggregator(context, params=user_params)
assistant = GoogleAssistantContextAggregator(context, params=assistant_params)
return GoogleContextAggregatorPair(_user=user, _assistant=assistant)

4
src/pipecat/services/google/llm_vertex.py
View File

@@ -65,7 +65,9 @@ class GoogleVertexLLMService(OpenAILLMService):
base_url = self._get_base_url(params)
self._api_key = self._get_api_token(credentials, credentials_path)
super().__init__(api_key=self._api_key, base_url=base_url, model=model, **kwargs)
super().__init__(
api_key=self._api_key, base_url=base_url, model=model, params=params, **kwargs
)
@staticmethod
def _get_base_url(params: InputParams) -> str:

6
src/pipecat/services/google/tts.py
View File

@@ -346,9 +346,9 @@ class GoogleTTSService(TTSService):
audio_content = response.audio_content[44:]
# Read and yield audio data in chunks
chunk_size = 8192
for i in range(0, len(audio_content), chunk_size):
chunk = audio_content[i : i + chunk_size]
CHUNK_SIZE = 1024
for i in range(0, len(audio_content), CHUNK_SIZE):
chunk = audio_content[i : i + CHUNK_SIZE]
if not chunk:
break
await self.stop_ttfb_metrics()

23
src/pipecat/services/grok/llm.py
View File

@@ -5,11 +5,14 @@
#
from dataclasses import dataclass
from typing import Any, Mapping
from loguru import logger
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.services.openai.llm import (
OpenAIAssistantContextAggregator,
@@ -124,8 +127,8 @@ class GrokLLMService(OpenAILLMService):
self,
context: OpenAILLMContext,
*,
user_kwargs: Mapping[str, Any] = {},
assistant_kwargs: Mapping[str, Any] = {},
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> GrokContextAggregatorPair:
"""Create an instance of GrokContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
@@ -133,12 +136,10 @@ class GrokLLMService(OpenAILLMService):
Args:
context (OpenAILLMContext): The LLM context.
user_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the user context aggregator constructor. Defaults
to an empty mapping.
assistant_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the assistant context aggregator
constructor. Defaults to an empty mapping.
user_params (LLMUserAggregatorParams, optional): User aggregator
parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User
aggregator parameters.
Returns:
GrokContextAggregatorPair: A pair of context aggregators, one for
@@ -148,6 +149,6 @@ class GrokLLMService(OpenAILLMService):
"""
context.set_llm_adapter(self.get_llm_adapter())
user = OpenAIUserContextAggregator(context, **user_kwargs)
assistant = OpenAIAssistantContextAggregator(context, **assistant_kwargs)
user = OpenAIUserContextAggregator(context, params=user_params)
assistant = OpenAIAssistantContextAggregator(context, params=assistant_params)
return GrokContextAggregatorPair(_user=user, _assistant=assistant)

10
src/pipecat/services/llm_service.py
View File

@@ -6,7 +6,7 @@
import asyncio
from dataclasses import dataclass
from typing import Any, Mapping, Optional, Set, Tuple, Type
from typing import Any, Optional, Set, Tuple, Type
from loguru import logger
@@ -20,6 +20,10 @@ from pipecat.frames.frames import (
StartInterruptionFrame,
UserImageRequestFrame,
)
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
from pipecat.processors.frame_processor import FrameDirection
from pipecat.services.ai_service import AIService
@@ -55,8 +59,8 @@ class LLMService(AIService):
self,
context: OpenAILLMContext,
*,
user_kwargs: Mapping[str, Any] = {},
assistant_kwargs: Mapping[str, Any] = {},
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> Any:
pass

22
src/pipecat/services/openai/llm.py
View File

@@ -6,7 +6,7 @@
import json
from dataclasses import dataclass
from typing import Any, Mapping
from typing import Any
from pipecat.frames.frames import (
FunctionCallCancelFrame,
@@ -15,7 +15,9 @@ from pipecat.frames.frames import (
UserImageRawFrame,
)
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMAssistantContextAggregator,
LLMUserAggregatorParams,
LLMUserContextAggregator,
)
from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
@@ -38,7 +40,7 @@ class OpenAILLMService(BaseOpenAILLMService):
def __init__(
self,
*,
model: str = "gpt-4o",
model: str = "gpt-4.1",
params: BaseOpenAILLMService.InputParams = BaseOpenAILLMService.InputParams(),
**kwargs,
):
@@ -48,8 +50,8 @@ class OpenAILLMService(BaseOpenAILLMService):
self,
context: OpenAILLMContext,
*,
user_kwargs: Mapping[str, Any] = {},
assistant_kwargs: Mapping[str, Any] = {},
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> OpenAIContextAggregatorPair:
"""Create an instance of OpenAIContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
@@ -57,12 +59,8 @@ class OpenAILLMService(BaseOpenAILLMService):
Args:
context (OpenAILLMContext): The LLM context.
user_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the user context aggregator constructor. Defaults
to an empty mapping.
assistant_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the assistant context aggregator
constructor. Defaults to an empty mapping.
user_params (LLMUserAggregatorParams, optional): User aggregator parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User aggregator parameters.
Returns:
OpenAIContextAggregatorPair: A pair of context aggregators, one for
@@ -71,8 +69,8 @@ class OpenAILLMService(BaseOpenAILLMService):
"""
context.set_llm_adapter(self.get_llm_adapter())
user = OpenAIUserContextAggregator(context, **user_kwargs)
assistant = OpenAIAssistantContextAggregator(context, **assistant_kwargs)
user = OpenAIUserContextAggregator(context, params=user_params)
assistant = OpenAIAssistantContextAggregator(context, params=assistant_params)
return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)

42
src/pipecat/services/openai_realtime_beta/openai.py
View File

@@ -8,19 +8,9 @@ import base64
import json
import time
from dataclasses import dataclass
from typing import Any, Mapping
from loguru import logger
try:
import websockets
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use OpenAI, you need to `pip install pipecat-ai[openai]`. Also, set `OPENAI_API_KEY` environment variable."
)
raise Exception(f"Missing module: {e}")
from pipecat.adapters.services.open_ai_realtime_adapter import OpenAIRealtimeLLMAdapter
from pipecat.frames.frames import (
BotStoppedSpeakingFrame,
@@ -48,6 +38,10 @@ from pipecat.frames.frames import (
UserStoppedSpeakingFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage
from pipecat.processors.aggregators.llm_response import (
LLMAssistantAggregatorParams,
LLMUserAggregatorParams,
)
from pipecat.processors.aggregators.openai_llm_context import (
OpenAILLMContext,
OpenAILLMContextFrame,
@@ -65,6 +59,13 @@ from .context import (
)
from .frames import RealtimeFunctionCallResultFrame, RealtimeMessagesUpdateFrame
try:
import websockets
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use OpenAI, you need to `pip install pipecat-ai[openai]`.")
raise Exception(f"Missing module: {e}")
@dataclass
class CurrentAudioResponse:
@@ -650,8 +651,8 @@ class OpenAIRealtimeBetaLLMService(LLMService):
self,
context: OpenAILLMContext,
*,
user_kwargs: Mapping[str, Any] = {},
assistant_kwargs: Mapping[str, Any] = {},
user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(),
assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(),
) -> OpenAIContextAggregatorPair:
"""Create an instance of OpenAIContextAggregatorPair from an
OpenAILLMContext. Constructor keyword arguments for both the user and
@@ -659,12 +660,10 @@ class OpenAIRealtimeBetaLLMService(LLMService):
Args:
context (OpenAILLMContext): The LLM context.
user_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the user context aggregator constructor. Defaults
to an empty mapping.
assistant_kwargs (Mapping[str, Any], optional): Additional keyword
arguments for the assistant context aggregator
constructor. Defaults to an empty mapping.
user_params (LLMUserAggregatorParams, optional): User aggregator
parameters.
assistant_params (LLMAssistantAggregatorParams, optional): User
aggregator parameters.
Returns:
OpenAIContextAggregatorPair: A pair of context aggregators, one for
@@ -675,9 +674,8 @@ class OpenAIRealtimeBetaLLMService(LLMService):
context.set_llm_adapter(self.get_llm_adapter())
OpenAIRealtimeLLMContext.upgrade_to_realtime(context)
user = OpenAIRealtimeUserContextAggregator(context, **user_kwargs)
user = OpenAIRealtimeUserContextAggregator(context, params=user_params)
default_assistant_kwargs = {"expect_stripped_words": False}
default_assistant_kwargs.update(assistant_kwargs)
assistant = OpenAIRealtimeAssistantContextAggregator(context, **default_assistant_kwargs)
assistant_params.expect_stripped_words = False
assistant = OpenAIRealtimeAssistantContextAggregator(context, params=assistant_params)
return OpenAIContextAggregatorPair(_user=user, _assistant=assistant)

2
src/pipecat/services/openpipe/llm.py
View File

@@ -25,7 +25,7 @@ class OpenPipeLLMService(OpenAILLMService):
def __init__(
self,
*,
model: str = "gpt-4o",
model: str = "gpt-4.1",
api_key: Optional[str] = None,
base_url: Optional[str] = None,
openpipe_api_key: Optional[str] = None,

104
src/pipecat/services/tavus/video.py
View File

@@ -6,7 +6,9 @@
"""This module implements Tavus as a sink transport layer"""
import asyncio
import base64
from typing import Optional
import aiohttp
from loguru import logger
@@ -16,6 +18,7 @@ from pipecat.frames.frames import (
CancelFrame,
EndFrame,
Frame,
StartFrame,
StartInterruptionFrame,
TransportMessageUrgentFrame,
TTSAudioRawFrame,
@@ -50,6 +53,10 @@ class TavusVideoService(AIService):
self._resampler = create_default_resampler()
self._audio_buffer = bytearray()
self._queue = asyncio.Queue()
self._send_task: Optional[asyncio.Task] = None
async def initialize(self) -> str:
url = "https://tavusapi.com/v2/conversations"
headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
@@ -78,45 +85,98 @@ class TavusVideoService(AIService):
logger.debug(f"TavusVideoService persona grabbed {response_json}")
return response_json["persona_name"]
async def start(self, frame: StartFrame):
await super().start(frame)
await self._create_send_task()
async def stop(self, frame: EndFrame):
await super().stop(frame)
await self._end_conversation()
await self._cancel_send_task()
async def cancel(self, frame: CancelFrame):
await super().cancel(frame)
await self._end_conversation()
await self._cancel_send_task()
async def _end_conversation(self) -> None:
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, StartInterruptionFrame):
await self._handle_interruptions()
await self.push_frame(frame, direction)
elif isinstance(frame, TTSStartedFrame):
await self.start_processing_metrics()
await self.start_ttfb_metrics()
self._current_idx_str = str(frame.id)
elif isinstance(frame, TTSAudioRawFrame):
await self._queue_audio(frame.audio, frame.sample_rate, done=False)
elif isinstance(frame, TTSStoppedFrame):
await self._queue_audio(b"\x00\x00", self._sample_rate, done=True)
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
else:
await self.push_frame(frame, direction)
async def _handle_interruptions(self):
await self._cancel_send_task()
await self._create_send_task()
await self._send_interrupt_message()
async def _end_conversation(self):
url = f"https://tavusapi.com/v2/conversations/{self._conversation_id}/end"
headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
async with self._session.post(url, headers=headers) as r:
r.raise_for_status()
async def _encode_audio_and_send(self, audio: bytes, in_rate: int, done: bool) -> None:
async def _queue_audio(self, audio: bytes, in_rate: int, done: bool):
await self._queue.put((audio, in_rate, done))
async def _create_send_task(self):
if not self._send_task:
self._queue = asyncio.Queue()
self._send_task = self.create_task(self._send_task_handler())
async def _cancel_send_task(self):
if self._send_task:
await self.cancel_task(self._send_task)
self._send_task = None
async def _send_task_handler(self):
# Daily app-messages have a 4kb limit and also a rate limit of 20
# messages per second. Below, we only consider the rate limit because 1
# second of a 24000 sample rate would be 48000 bytes (16-bit samples and
# 1 channel). So, that is 48000 / 20 = 2400, which is below the 4kb
# limit (even including base64 encoding). For a sample rate of 16000,
# that would be 32000 / 20 = 1600.
MAX_CHUNK_SIZE = int((self._sample_rate * 2) / 20)
SLEEP_TIME = 1 / 20
audio_buffer = bytearray()
while True:
(audio, in_rate, done) = await self._queue.get()
if done:
# Send any remaining audio.
if len(audio_buffer) > 0:
await self._encode_audio_and_send(bytes(audio_buffer), done)
await self._encode_audio_and_send(audio, done)
audio_buffer.clear()
else:
audio = await self._resampler.resample(audio, in_rate, self._sample_rate)
audio_buffer.extend(audio)
while len(audio_buffer) >= MAX_CHUNK_SIZE:
chunk = audio_buffer[:MAX_CHUNK_SIZE]
audio_buffer = audio_buffer[MAX_CHUNK_SIZE:]
await self._encode_audio_and_send(bytes(chunk), done)
await asyncio.sleep(SLEEP_TIME)
async def _encode_audio_and_send(self, audio: bytes, done: bool):
"""Encodes audio to base64 and sends it to Tavus"""
if not done:
audio = await self._resampler.resample(audio, in_rate, self._sample_rate)
audio_base64 = base64.b64encode(audio).decode("utf-8")
logger.trace(f"{self}: sending {len(audio)} bytes")
await self._send_audio_message(audio_base64, done=done)
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)
if isinstance(frame, TTSStartedFrame):
await self.start_processing_metrics()
await self.start_ttfb_metrics()
self._current_idx_str = str(frame.id)
elif isinstance(frame, TTSAudioRawFrame):
await self._encode_audio_and_send(frame.audio, frame.sample_rate, done=False)
elif isinstance(frame, TTSStoppedFrame):
await self._encode_audio_and_send(b"\x00", self._sample_rate, done=True)
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
elif isinstance(frame, StartInterruptionFrame):
await self._send_interrupt_message()
else:
await self.push_frame(frame, direction)
async def _send_interrupt_message(self) -> None:
transport_frame = TransportMessageUrgentFrame(
message={
@@ -127,7 +187,7 @@ class TavusVideoService(AIService):
)
await self.push_frame(transport_frame)
async def _send_audio_message(self, audio_base64: str, done: bool) -> None:
async def _send_audio_message(self, audio_base64: str, done: bool):
transport_frame = TransportMessageUrgentFrame(
message={
"message_type": "conversation",

5
src/pipecat/transports/base_output.py
View File

@@ -386,10 +386,13 @@ class BaseOutputTransport(FrameProcessor):
async def _draw_image(self, frame: OutputImageRawFrame):
desired_size = (self._params.camera_out_width, self._params.camera_out_height)
# TODO: we should refactor in the future to support dynamic resolutions
# which is kind of what happens in P2P connections.
# We need to add support for that inside the DailyTransport
if frame.size != desired_size:
image = Image.frombytes(frame.format, frame.size, frame.image)
resized_image = image.resize(desired_size)
logger.warning(f"{frame} does not have the expected size {desired_size}, resizing")
# logger.warning(f"{frame} does not have the expected size {desired_size}, resizing")
frame = OutputImageRawFrame(
resized_image.tobytes(), resized_image.size, resized_image.format
)

11
src/pipecat/transports/services/helpers/daily_rest.py
View File

@@ -68,9 +68,9 @@ class DailyRoomProperties(BaseModel, extra="allow"):
exp: Optional[float] = None
enable_chat: bool = False
enable_prejoin_ui: bool = True
enable_prejoin_ui: bool = False
enable_emoji_reactions: bool = False
eject_at_room_exp: bool = True
eject_at_room_exp: bool = False
enable_dialout: Optional[bool] = None
enable_recording: Optional[Literal["cloud", "local", "raw-tracks"]] = None
geo: Optional[str] = None
@@ -291,6 +291,7 @@ class DailyRESTHelper:
self,
room_url: str,
expiry_time: float = 60 * 60,
eject_at_token_exp: bool = False,
owner: bool = True,
params: Optional[DailyMeetingTokenParams] = None,
) -> str:
@@ -324,12 +325,16 @@ class DailyRESTHelper:
if params is None:
params = DailyMeetingTokenParams(
properties=DailyMeetingTokenProperties(
room_name=room_name, is_owner=owner, exp=expiration
room_name=room_name,
is_owner=owner,
exp=expiration,
eject_at_token_exp=eject_at_token_exp,
)
)
else:
params.properties.room_name = room_name
params.properties.exp = expiration
params.properties.eject_at_token_exp = eject_at_token_exp
params.properties.is_owner = owner
json = params.model_dump(exclude_none=True)