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

Merge branch 'main' into filipi/websocket_transport_example_twilio

Browse Source
This commit is contained in:
Filipi Fuchter
2025-06-17 15:54:31 -03:00
parent 80ce097f90 5614e35ac4
commit cbd0529674
16 changed files with 171 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
CHANGELOG.md
View File

@@ -9,6 +9,55 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- Added Polish support to `AWSTranscribeSTTService`.
- Added new frames `FrameProcessorPauseFrame` and `FrameProcessorResumeFrame`
which allow pausing and resuming frame processing for a given frame
processor. These are control frames, so they are ordered. Pausing frame
processor will keep old frames in the internal queues until resume takes
place. Frames being pushed while a frame processor is paused will be pushed to
the queues. When frame processing is resumed all queued frames will be
processed in order. Also added `FrameProcessorPauseUrgentFrame` and
`FrameProcessorResumeUrgentFrame` which are system frames and therefore they
have high priority.
- Added a property called `has_function_calls_in_progress` in
`LLMAssistantContextAggregator` that exposes whether a function call is in
progress.
### Changed
- Upgraded `daily-python` to 0.19.3.
### Fixed
- Fixed an issue with `GroqTTSService` where it was not properly parsing the
WAV file header.
- Fixed an issue with `GoogleSTTService` where it was constantly reconnecting
before starting to receive audio from the user.
- Fixed an issue where `GoogleLLMService`'s TTFB value was incorrect.
### Other
- Rename `14e-function-calling-gemini.py` to `14e-function-calling-google.py`.
## [0.0.71] - 2025-06-10
### Added
- Adds a parameter called `additional_span_attributes` to PipelineTask that
lets you add any additional attributes you'd like to the conversation span.
### Fixed
- Fixed an issue with `CartesiaSTTService` initialization.
## [0.0.70] - 2025-06-10
### Added
- Added `ExotelFrameSerializer` to handle telephony calls via Exotel.
- Added the option `informal` to `TranslationConfig` on Gladia config.

2
README.md
View File

@@ -53,7 +53,7 @@ You can connect to Pipecat from any platform using our official SDKs:
| Category | Services |
| ------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), Cartesia, [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Parakeet (NVIDIA)](https://docs.pipecat.ai/server/services/stt/parakeet), [Ultravox](https://docs.pipecat.ai/server/services/stt/ultravox), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) |
| Speech-to-Text | [AssemblyAI](https://docs.pipecat.ai/server/services/stt/assemblyai), [AWS](https://docs.pipecat.ai/server/services/stt/aws), [Azure](https://docs.pipecat.ai/server/services/stt/azure), [Cartesia](https://docs.pipecat.ai/server/services/stt/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/stt/deepgram), [Fal Wizper](https://docs.pipecat.ai/server/services/stt/fal), [Gladia](https://docs.pipecat.ai/server/services/stt/gladia), [Google](https://docs.pipecat.ai/server/services/stt/google), [Groq (Whisper)](https://docs.pipecat.ai/server/services/stt/groq), [OpenAI (Whisper)](https://docs.pipecat.ai/server/services/stt/openai), [Parakeet (NVIDIA)](https://docs.pipecat.ai/server/services/stt/parakeet), [Ultravox](https://docs.pipecat.ai/server/services/stt/ultravox), [Whisper](https://docs.pipecat.ai/server/services/stt/whisper) |
| LLMs | [Anthropic](https://docs.pipecat.ai/server/services/llm/anthropic), [AWS](https://docs.pipecat.ai/server/services/llm/aws), [Azure](https://docs.pipecat.ai/server/services/llm/azure), [Cerebras](https://docs.pipecat.ai/server/services/llm/cerebras), [DeepSeek](https://docs.pipecat.ai/server/services/llm/deepseek), [Fireworks AI](https://docs.pipecat.ai/server/services/llm/fireworks), [Gemini](https://docs.pipecat.ai/server/services/llm/gemini), [Grok](https://docs.pipecat.ai/server/services/llm/grok), [Groq](https://docs.pipecat.ai/server/services/llm/groq), [NVIDIA NIM](https://docs.pipecat.ai/server/services/llm/nim), [Ollama](https://docs.pipecat.ai/server/services/llm/ollama), [OpenAI](https://docs.pipecat.ai/server/services/llm/openai), [OpenRouter](https://docs.pipecat.ai/server/services/llm/openrouter), [Perplexity](https://docs.pipecat.ai/server/services/llm/perplexity), [Qwen](https://docs.pipecat.ai/server/services/llm/qwen), [Together AI](https://docs.pipecat.ai/server/services/llm/together) |
| Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [FastPitch (NVIDIA)](https://docs.pipecat.ai/server/services/tts/fastpitch), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [Sarvam](https://docs.pipecat.ai/server/services/tts/sarvam), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) |
| Speech-to-Speech | [AWS Nova Sonic](https://docs.pipecat.ai/server/services/s2s/aws), [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) |

4
dev-requirements.txt
View File

@@ -3,11 +3,11 @@ coverage~=7.6.12
grpcio-tools~=1.67.1
pip-tools~=7.4.1
pre-commit~=4.0.1
pyright~=1.1.397
pyright~=1.1.400
pytest~=8.3.4
pytest-asyncio~=0.25.3
pytest-aiohttp==1.1.0
ruff~=0.11.1
ruff~=0.11.13
setuptools~=70.0.0
setuptools_scm~=8.1.0
python-dotenv~=1.0.1

0
examples/foundational/14e-function-calling-gemini.py → examples/foundational/14e-function-calling-google.py
View File

8
examples/foundational/22b-natural-conversation-proposal.py
View File

@@ -312,9 +312,6 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si
# to start the conversation.
bot_output_gate = OutputGate(notifier=notifier, start_open=True)
async def block_user_stopped_speaking(frame):
return not isinstance(frame, UserStoppedSpeakingFrame)
async def pass_only_llm_trigger_frames(frame):
return (
isinstance(frame, OpenAILLMContextFrame)
@@ -331,11 +328,6 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si
stt,
context_aggregator.user(),
ParallelPipeline(
[
# Pass everything except UserStoppedSpeaking to the elements after
# this ParallelPipeline
FunctionFilter(filter=block_user_stopped_speaking),
],
[
# Ignore everything except an OpenAILLMContextFrame. Pass a specially constructed
# LLMMessagesFrame to the statement classifier LLM. The only frame this

4
pyproject.toml
View File

@@ -47,7 +47,7 @@ azure = [ "azure-cognitiveservices-speech~=1.42.0"]
cartesia = [ "cartesia~=2.0.3", "websockets~=13.1" ]
cerebras = []
deepseek = []
daily = [ "daily-python~=0.19.2" ]
daily = [ "daily-python~=0.19.3" ]
deepgram = [ "deepgram-sdk~=4.1.0" ]
elevenlabs = [ "websockets~=13.1" ]
fal = [ "fal-client~=0.5.9" ]
@@ -58,7 +58,7 @@ google = [ "google-cloud-speech~=2.32.0", "google-cloud-texttospeech~=2.26.0", "
grok = []
groq = [ "groq~=0.23.0" ]
gstreamer = [ "pygobject~=3.50.0" ]
krisp = [ "pipecat-ai-krisp~=0.3.0" ]
krisp = [ "pipecat-ai-krisp~=0.4.0" ]
koala = [ "pvkoala~=2.0.3" ]
langchain = [ "langchain~=0.3.20", "langchain-community~=0.3.20", "langchain-openai~=0.3.9" ]
livekit = [ "livekit~=0.22.0", "livekit-api~=0.8.2", "tenacity~=9.0.0" ]

44
src/pipecat/frames/frames.py
View File

@@ -527,6 +527,29 @@ class StopTaskFrame(SystemFrame):
pass
@dataclass
class FrameProcessorPauseUrgentFrame(SystemFrame):
"""This processor is used to pause frame processing for the given processor
as fast as possible. Pausing frame processing will keep frames in the
internal queue which will then be processed when frame processing is resumed
with `FrameProcessorResumeFrame`.
"""
processor: str
@dataclass
class FrameProcessorResumeUrgentFrame(SystemFrame):
"""This processor is used to resume frame processing for the given processor
if it was previously paused as fast as possible. After resuming frame
processing all queued frames will be processed in the order received.
"""
processor: str
@dataclass
class StartInterruptionFrame(SystemFrame):
"""Emitted by VAD to indicate that a user has started speaking (i.e. is
@@ -854,6 +877,27 @@ class StopFrame(ControlFrame):
pass
@dataclass
class FrameProcessorPauseFrame(ControlFrame):
"""This processor is used to pause frame processing for the given
processor. Pausing frame processing will keep frames in the internal queue
which will then be processed when frame processing is resumed with
`FrameProcessorResumeFrame`."""
processor: str
@dataclass
class FrameProcessorResumeFrame(ControlFrame):
"""This processor is used to resume frame processing for the given processor
if it was previously paused. After resuming frame processing all queued
frames will be processed in the order received.
"""
processor: str
@dataclass
class LLMFullResponseStartFrame(ControlFrame):
"""Used to indicate the beginning of an LLM response. Following by one or

10
src/pipecat/pipeline/task.py
View File

@@ -184,7 +184,9 @@ class PipelineTask(BaseTask):
the idle timeout is reached.
enable_turn_tracking: Whether to enable turn tracking.
enable_turn_tracing: Whether to enable turn tracing.
conversation_id: Optional custom ID for the conversation.
additional_span_attributes: Optional dictionary of attributes to propagate as
OpenTelemetry conversation span attributes.
"""
def __init__(
@@ -205,6 +207,7 @@ class PipelineTask(BaseTask):
enable_turn_tracking: bool = True,
enable_tracing: bool = False,
conversation_id: Optional[str] = None,
additional_span_attributes: Optional[dict] = None,
):
super().__init__()
self._pipeline = pipeline
@@ -217,6 +220,7 @@ class PipelineTask(BaseTask):
self._enable_turn_tracking = enable_turn_tracking
self._enable_tracing = enable_tracing and is_tracing_available()
self._conversation_id = conversation_id
self._additional_span_attributes = additional_span_attributes or {}
if self._params.observers:
import warnings
@@ -235,7 +239,9 @@ class PipelineTask(BaseTask):
observers.append(self._turn_tracking_observer)
if self._enable_tracing and self._turn_tracking_observer:
self._turn_trace_observer = TurnTraceObserver(
self._turn_tracking_observer, conversation_id=self._conversation_id
self._turn_tracking_observer,
conversation_id=self._conversation_id,
additional_span_attributes=self._additional_span_attributes,
)
observers.append(self._turn_trace_observer)
self._finished = False

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

@@ -504,6 +504,15 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator):
self._function_calls_in_progress: Dict[str, Optional[FunctionCallInProgressFrame]] = {}
self._context_updated_tasks: Set[asyncio.Task] = set()
@property
def has_function_calls_in_progress(self) -> bool:
"""Check if there are any function calls currently in progress.
Returns:
bool: True if function calls are in progress, False otherwise
"""
return bool(self._function_calls_in_progress)
async def handle_aggregation(self, aggregation: str):
self._context.add_message({"role": "assistant", "content": aggregation})

16
src/pipecat/processors/frame_processor.py
View File

@@ -17,6 +17,10 @@ from pipecat.frames.frames import (
CancelFrame,
ErrorFrame,
Frame,
FrameProcessorPauseFrame,
FrameProcessorPauseUrgentFrame,
FrameProcessorResumeFrame,
FrameProcessorResumeUrgentFrame,
StartFrame,
StartInterruptionFrame,
StopInterruptionFrame,
@@ -259,6 +263,10 @@ class FrameProcessor(BaseObject):
self._should_report_ttfb = True
elif isinstance(frame, CancelFrame):
await self.__cancel(frame)
elif isinstance(frame, (FrameProcessorPauseFrame, FrameProcessorPauseUrgentFrame)):
await self.__pause(frame)
elif isinstance(frame, (FrameProcessorResumeFrame, FrameProcessorResumeUrgentFrame)):
await self.__resume(frame)
async def push_error(self, error: ErrorFrame):
await self.push_frame(error, FrameDirection.UPSTREAM)
@@ -287,6 +295,14 @@ class FrameProcessor(BaseObject):
await self.__cancel_input_task()
await self.__cancel_push_task()
async def __pause(self, frame: FrameProcessorPauseFrame | FrameProcessorPauseUrgentFrame):
if frame.name == self.name:
await self.pause_processing_frames()
async def __resume(self, frame: FrameProcessorResumeFrame | FrameProcessorResumeUrgentFrame):
if frame.name == self.name:
await self.resume_processing_frames()
#
# Handle interruptions
#

1
src/pipecat/services/aws/stt.py
View File

@@ -266,6 +266,7 @@ class AWSTranscribeSTTService(STTService):
Language.JA: "ja-JP",
Language.KO: "ko-KR",
Language.ZH: "zh-CN",
Language.PL: "pl-PL",
}
return language_map.get(language)

2
src/pipecat/services/cartesia/stt.py
View File

@@ -101,7 +101,7 @@ class CartesiaSTTService(STTService):
)
self._settings = merged_options
self.set_model_name(merged_options["model"])
self.set_model_name(merged_options.model)
self._api_key = api_key
self._base_url = base_url or "api.cartesia.ai"
self._connection = None

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

@@ -555,10 +555,11 @@ class GoogleLLMService(LLMService):
contents=messages,
config=generation_config,
)
await self.stop_ttfb_metrics()
function_calls = []
async for chunk in response:
# Stop TTFB metrics after the first chunk
await self.stop_ttfb_metrics()
if chunk.usage_metadata:
prompt_tokens += chunk.usage_metadata.prompt_token_count or 0
completion_tokens += chunk.usage_metadata.candidates_token_count or 0

5
src/pipecat/services/google/stt.py
View File

@@ -747,6 +747,11 @@ class GoogleSTTService(STTService):
try:
while True:
try:
if self._request_queue.empty():
# wait for 10ms in case we don't have audio
await asyncio.sleep(0.01)
continue
# Start bi-directional streaming
streaming_recognize = await self._client.streaming_recognize(
requests=self._request_generator()

38
src/pipecat/services/groq/tts.py
View File

@@ -4,6 +4,8 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
import io
import wave
from typing import AsyncGenerator, Optional
from loguru import logger
@@ -78,22 +80,26 @@ class GroqTTSService(TTSService):
await self.start_ttfb_metrics()
yield TTSStartedFrame()
response = await self._client.audio.speech.create(
model=self._model_name,
voice=self._voice_id,
response_format=self._output_format,
input=text,
)
try:
response = await self._client.audio.speech.create(
model=self._model_name,
voice=self._voice_id,
response_format=self._output_format,
input=text,
)
async for data in response.iter_bytes():
if measuring_ttfb:
await self.stop_ttfb_metrics()
measuring_ttfb = False
# remove wav header if present
if data.startswith(b"RIFF"):
data = data[44:]
if len(data) == 0:
continue
yield TTSAudioRawFrame(data, self.sample_rate, 1)
async for data in response.iter_bytes():
if measuring_ttfb:
await self.stop_ttfb_metrics()
measuring_ttfb = False
with wave.open(io.BytesIO(data)) as w:
channels = w.getnchannels()
frame_rate = w.getframerate()
num_frames = w.getnframes()
bytes = w.readframes(num_frames)
yield TTSAudioRawFrame(bytes, frame_rate, channels)
except Exception as e:
logger.error(f"{self} exception: {e}")
yield TTSStoppedFrame()

10
src/pipecat/utils/tracing/turn_trace_observer.py
View File

@@ -35,7 +35,11 @@ class TurnTraceObserver(BaseObserver):
"""
def __init__(
self, turn_tracker: TurnTrackingObserver, conversation_id: Optional[str] = None, **kwargs
self,
turn_tracker: TurnTrackingObserver,
conversation_id: Optional[str] = None,
additional_span_attributes: Optional[dict] = None,
**kwargs,
):
super().__init__(**kwargs)
self._turn_tracker = turn_tracker
@@ -47,6 +51,7 @@ class TurnTraceObserver(BaseObserver):
# Conversation tracking properties
self._conversation_span: Optional["Span"] = None
self._conversation_id = conversation_id
self._additional_span_attributes = additional_span_attributes or {}
if turn_tracker:
@@ -89,6 +94,9 @@ class TurnTraceObserver(BaseObserver):
# Set span attributes
self._conversation_span.set_attribute("conversation.id", conversation_id)
self._conversation_span.set_attribute("conversation.type", "voice")
# Set custom otel attributes if provided
for k, v in (self._additional_span_attributes or {}).items():
self._conversation_span.set_attribute(k, v)
# Update the conversation context provider
context_provider.set_current_conversation_context(