diff --git a/CHANGELOG.md b/CHANGELOG.md index e4ef1e019..dd6079689 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `HeartbeatFrame`s are now control frames. This will make it easier to detect + pipeline freezes. Previously, heartbeat frames were system frames which meant + they were not get queued with other frames, making it difficult to detect + pipeline stalls. + - Updated `OpenAIRealtimeBetaLLMService` to accept `language` in the `InputAudioTranscription` class for all models. @@ -50,6 +55,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue with `ElevenLabsTTSService` where the context was not being + closed. + - Fixed function calling in `AWSNovaSonicLLMService`. - Fixed an issue that would cause multiple `PipelineTask.on_idle_timeout` diff --git a/dot-env.template b/dot-env.template index 20d73b3ad..4cf716137 100644 --- a/dot-env.template +++ b/dot-env.template @@ -107,4 +107,7 @@ MINIMAX_API_KEY=... MINIMAX_GROUP_ID=... # Sarvam AI -SARVAM_API_KEY=... \ No newline at end of file +SARVAM_API_KEY=... + +# Sentry +SENTRY_DSN=... \ No newline at end of file diff --git a/examples/foundational/39a-mcp-run-sse.py b/examples/foundational/39a-mcp-run-sse.py index 8b43b3022..fbbcbcc6a 100644 --- a/examples/foundational/39a-mcp-run-sse.py +++ b/examples/foundational/39a-mcp-run-sse.py @@ -9,6 +9,7 @@ import os from dotenv import load_dotenv from loguru import logger +from mcp.client.session_group import SseServerParameters from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.pipeline.pipeline import Pipeline @@ -63,7 +64,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si try: # https://docs.mcp.run/integrating/tutorials/mcp-run-sse-openai-agents/ - mcp = MCPClient(server_params=os.getenv("MCP_RUN_SSE_URL")) + mcp = MCPClient(server_params=SseServerParameters(url=os.getenv("MCP_RUN_SSE_URL"))) except Exception as e: logger.error(f"error setting up mcp") logger.exception("error trace:") diff --git a/examples/foundational/39b-multiple-mcp.py b/examples/foundational/39b-multiple-mcp.py index 3ba90d580..7d1396834 100644 --- a/examples/foundational/39b-multiple-mcp.py +++ b/examples/foundational/39b-multiple-mcp.py @@ -15,6 +15,7 @@ import aiohttp from dotenv import load_dotenv from loguru import logger from mcp import StdioServerParameters +from mcp.client.session_group import SseServerParameters from PIL import Image from pipecat.adapters.schemas.tools_schema import ToolsSchema @@ -149,7 +150,7 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si # https://docs.mcp.run/integrating/tutorials/mcp-run-sse-openai-agents/ # ie. "https://www.mcp.run/api/mcp/sse?..." # ensure the profile has a tool or few installed - mcp_run = MCPClient(server_params=os.getenv("MCP_RUN_SSE_URL")) + mcp_run = MCPClient(server_params=SseServerParameters(url=os.getenv("MCP_RUN_SSE_URL"))) except Exception as e: logger.error(f"error setting up mcp.run") logger.exception("error trace:") diff --git a/examples/sentry-metrics/bot.py b/examples/sentry-metrics/bot.py index 8ff412bb7..44f9a0daa 100644 --- a/examples/sentry-metrics/bot.py +++ b/examples/sentry-metrics/bot.py @@ -49,7 +49,7 @@ async def main(): # Initialize Sentry sentry_sdk.init( - dsn="your-project-dsn", + dsn=os.getenv("SENTRY_DSN"), traces_sample_rate=1.0, ) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index ec368789d..4b2d934ab 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -7,6 +7,7 @@ from dataclasses import dataclass, field from enum import Enum from typing import ( + TYPE_CHECKING, Any, Awaitable, Callable, @@ -26,6 +27,9 @@ from pipecat.transcriptions.language import Language from pipecat.utils.time import nanoseconds_to_str from pipecat.utils.utils import obj_count, obj_id +if TYPE_CHECKING: + from pipecat.processors.frame_processor import FrameProcessor + class KeypadEntry(str, Enum): """DTMF entries.""" @@ -485,16 +489,6 @@ class FatalErrorFrame(ErrorFrame): fatal: bool = field(default=True, init=False) -@dataclass -class HeartbeatFrame(SystemFrame): - """This frame is used by the pipeline task as a mechanism to know if the - pipeline is running properly. - - """ - - timestamp: int - - @dataclass class EndTaskFrame(SystemFrame): """This is used to notify the pipeline task that the pipeline should be @@ -529,25 +523,25 @@ class StopTaskFrame(SystemFrame): @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`. + """This frame 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 + processor: "FrameProcessor" @dataclass class FrameProcessorResumeUrgentFrame(SystemFrame): - """This processor is used to resume frame processing for the given processor + """This frame 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 + processor: "FrameProcessor" @dataclass @@ -877,25 +871,37 @@ class StopFrame(ControlFrame): pass +@dataclass +class HeartbeatFrame(ControlFrame): + """This frame is used by the pipeline task as a mechanism to know if the + pipeline is running properly. + + """ + + timestamp: int + + @dataclass class FrameProcessorPauseFrame(ControlFrame): - """This processor is used to pause frame processing for the given + """This frame 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`.""" + `FrameProcessorResumeFrame`. - processor: str + """ + + processor: "FrameProcessor" @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. + """This frame 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 + processor: "FrameProcessor" @dataclass diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 3166735d2..1f318c411 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -43,7 +43,7 @@ from pipecat.utils.tracing.setup import is_tracing_available from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver HEARTBEAT_SECONDS = 1.0 -HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5 +HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 10 class PipelineParams(BaseModel): diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index b73fb0e9f..7696e8e65 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -296,11 +296,11 @@ class FrameProcessor(BaseObject): await self.__cancel_push_task() async def __pause(self, frame: FrameProcessorPauseFrame | FrameProcessorPauseUrgentFrame): - if frame.name == self.name: + if frame.processor.name == self.name: await self.pause_processing_frames() async def __resume(self, frame: FrameProcessorResumeFrame | FrameProcessorResumeUrgentFrame): - if frame.name == self.name: + if frame.processor.name == self.name: await self.resume_processing_frames() # diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index e0301360e..ccd9b5b3f 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -284,7 +284,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService): logger.trace(f"{self}: flushing audio") msg = {"context_id": self._context_id, "flush": True} await self._websocket.send(json.dumps(msg)) - self._context_id = None async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): await super().push_frame(frame, direction) @@ -380,6 +379,12 @@ class ElevenLabsTTSService(AudioContextWordTTSService): if self._context_id and self._websocket: logger.trace(f"Closing context {self._context_id} due to interruption") try: + # ElevenLabs requires that Pipecat manages the contexts and closes them + # when they're not longer in use. Since a StartInterruptionFrame is pushed + # every time the user speaks, we'll use this as a trigger to close the context + # and reset the state. + # Note: We do not need to call remove_audio_context here, as the context is + # automatically reset when super ()._handle_interruption is called. await self._websocket.send( json.dumps({"context_id": self._context_id, "close_context": True}) ) @@ -391,10 +396,20 @@ class ElevenLabsTTSService(AudioContextWordTTSService): async def _receive_messages(self): async for message in self._get_websocket(): msg = json.loads(message) - # Check if this message belongs to the current context + received_ctx_id = msg.get("contextId") + + # Handle final messages first, regardless of context availability + # At the moment, this message is received AFTER the close_context message is + # sent, so it doesn't serve any functional purpose. For now, we'll just log it. + if msg.get("isFinal") is True: + logger.trace(f"Received final message for context {received_ctx_id}") + continue + + # Check if this message belongs to the current context. + # This should never happen, so warn about it. if not self.audio_context_available(received_ctx_id): - logger.trace(f"Ignoring message from unavailable context: {received_ctx_id}") + logger.warning(f"Ignoring message from unavailable context: {received_ctx_id}") continue if msg.get("audio"): @@ -408,21 +423,26 @@ class ElevenLabsTTSService(AudioContextWordTTSService): word_times = calculate_word_times(msg["alignment"], self._cumulative_time) await self.add_word_timestamps(word_times) self._cumulative_time = word_times[-1][1] - if msg.get("isFinal"): - logger.trace(f"Received final message for context {received_ctx_id}") - await self.remove_audio_context(received_ctx_id) - # Reset context tracking if this was our active context - if self._context_id == received_ctx_id: - self._context_id = None - self._started = False async def _keepalive_task_handler(self): while True: await asyncio.sleep(10) try: - # Send an empty message to keep the connection alive if self._websocket and self._websocket.open: - await self._websocket.send(json.dumps({})) + if self._context_id: + # Send keepalive with context ID to keep the connection alive + keepalive_message = { + "text": "", + "context_id": self._context_id, + } + logger.trace(f"Sending keepalive for context {self._context_id}") + else: + # It's possible to have a user interruption which clears the context + # without generating a new TTS response. In this case, we'll just send + # an empty message to keep the connection alive. + keepalive_message = {"text": ""} + logger.trace("Sending keepalive without context") + await self._websocket.send(json.dumps(keepalive_message)) except websockets.ConnectionClosed as e: logger.warning(f"{self} keepalive error: {e}") break @@ -441,14 +461,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService): await self._connect() try: - # Close previous context if there was one - if self._context_id and not self._started: - await self._websocket.send( - json.dumps({"context_id": self._context_id, "close_context": True}) - ) - await self.remove_audio_context(self._context_id) - self._context_id = None - if not self._started: await self.start_ttfb_metrics() yield TTSStartedFrame() @@ -473,9 +485,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService): logger.error(f"{self} error sending message: {e}") yield TTSStoppedFrame() self._started = False - if self._context_id: - await self.remove_audio_context(self._context_id) - self._context_id = None return yield None except Exception as e: diff --git a/src/pipecat/services/mcp_service.py b/src/pipecat/services/mcp_service.py index ae54b84ff..a644d8f1b 100644 --- a/src/pipecat/services/mcp_service.py +++ b/src/pipecat/services/mcp_service.py @@ -8,8 +8,8 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema from pipecat.utils.base_object import BaseObject try: - from mcp import ClientSession, StdioServerParameters, types - from mcp.client.session import ClientSession + from mcp import ClientSession, StdioServerParameters + from mcp.client.session_group import SseServerParameters from mcp.client.sse import sse_client from mcp.client.stdio import stdio_client except ModuleNotFoundError as e: @@ -21,7 +21,7 @@ except ModuleNotFoundError as e: class MCPClient(BaseObject): def __init__( self, - server_params: Union[StdioServerParameters, str], + server_params: Union[StdioServerParameters, SseServerParameters], **kwargs, ): super().__init__(**kwargs) @@ -30,12 +30,12 @@ class MCPClient(BaseObject): if isinstance(server_params, StdioServerParameters): self._client = stdio_client self._register_tools = self._stdio_register_tools - elif isinstance(server_params, str): + elif isinstance(server_params, SseServerParameters): self._client = sse_client self._register_tools = self._sse_register_tools else: raise TypeError( - f"{self} invalid argument type: `server_params` must be either StdioServerParameters or an SSE server url string." + f"{self} invalid argument type: `server_params` must be either StdioServerParameters or SseServerParameters." ) async def register_tools(self, llm) -> ToolsSchema: @@ -90,7 +90,12 @@ class MCPClient(BaseObject): logger.debug(f"Executing tool '{function_name}' with call ID: {tool_call_id}") logger.trace(f"Tool arguments: {json.dumps(arguments, indent=2)}") try: - async with self._client(self._server_params) as (read, write): + async with self._client( + url=self._server_params.url, + headers=self._server_params.headers, + timeout=self._server_params.timeout, + sse_read_timeout=self._server_params.sse_read_timeout, + ) as (read, write): async with self._session(read, write) as session: await session.initialize() await self._call_tool(session, function_name, arguments, result_callback) @@ -100,10 +105,14 @@ class MCPClient(BaseObject): logger.exception("Full exception details:") await result_callback(error_msg) - logger.debug("Starting registration of mcp.run tools") - tool_schemas: List[FunctionSchema] = [] + logger.debug(f"SSE server parameters: {self._server_params}") - async with self._client(self._server_params) as (read, write): + async with self._client( + url=self._server_params.url, + headers=self._server_params.headers, + timeout=self._server_params.timeout, + sse_read_timeout=self._server_params.sse_read_timeout, + ) as (read, write): async with self._session(read, write) as session: await session.initialize() tools_schema = await self._list_tools(session, mcp_tool_wrapper, llm)