diff --git a/CHANGELOG.md b/CHANGELOG.md index 2fc5abde4..59efa9e3a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/README.md b/README.md index f906cb8bb..7ec4c6000 100644 --- a/README.md +++ b/README.md @@ -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) | diff --git a/dev-requirements.txt b/dev-requirements.txt index af1c35721..2bab71684 100644 --- a/dev-requirements.txt +++ b/dev-requirements.txt @@ -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 diff --git a/examples/foundational/14e-function-calling-gemini.py b/examples/foundational/14e-function-calling-google.py similarity index 100% rename from examples/foundational/14e-function-calling-gemini.py rename to examples/foundational/14e-function-calling-google.py diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 1b03f341e..b76bfd274 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -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 diff --git a/pyproject.toml b/pyproject.toml index 4652b684a..b5c874919 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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" ] diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 63caf9a09..ec368789d 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -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 diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 6bf5dd688..736f98244 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -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 diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 479199550..e9aebb2a0 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -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}) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 3b66973dd..1d2f066ed 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -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 # diff --git a/src/pipecat/services/aws/stt.py b/src/pipecat/services/aws/stt.py index 4490f3795..d6aa1fa5d 100644 --- a/src/pipecat/services/aws/stt.py +++ b/src/pipecat/services/aws/stt.py @@ -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) diff --git a/src/pipecat/services/cartesia/stt.py b/src/pipecat/services/cartesia/stt.py index 618908ed9..104e4b2c5 100644 --- a/src/pipecat/services/cartesia/stt.py +++ b/src/pipecat/services/cartesia/stt.py @@ -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 diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 8960bda31..f983b7342 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -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 diff --git a/src/pipecat/services/google/stt.py b/src/pipecat/services/google/stt.py index 4fd129af3..bf60541f5 100644 --- a/src/pipecat/services/google/stt.py +++ b/src/pipecat/services/google/stt.py @@ -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() diff --git a/src/pipecat/services/groq/tts.py b/src/pipecat/services/groq/tts.py index 6f73b1629..33fd3ce34 100644 --- a/src/pipecat/services/groq/tts.py +++ b/src/pipecat/services/groq/tts.py @@ -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() diff --git a/src/pipecat/utils/tracing/turn_trace_observer.py b/src/pipecat/utils/tracing/turn_trace_observer.py index 26fc5b38a..f67ca3b28 100644 --- a/src/pipecat/utils/tracing/turn_trace_observer.py +++ b/src/pipecat/utils/tracing/turn_trace_observer.py @@ -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(