From 312c5691821787a93d64dea5183884c827f4a5d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 13 Jun 2024 09:30:50 -0700 Subject: [PATCH 1/5] services(openpipe): refactored so it's based on BaseOpenAILLMService --- src/pipecat/processors/frame_processor.py | 6 +- src/pipecat/services/openai.py | 35 ++-- src/pipecat/services/openpipe.py | 197 ++++++---------------- 3 files changed, 80 insertions(+), 158 deletions(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 1197f4490..a1dfebc9e 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -22,7 +22,11 @@ class FrameDirection(Enum): class FrameProcessor: - def __init__(self, name: str | None = None, loop: asyncio.AbstractEventLoop | None = None): + def __init__( + self, + name: str | None = None, + loop: asyncio.AbstractEventLoop | None = None, + **kwargs): self.id: int = obj_id() self.name = name or f"{self.__class__.__name__}#{obj_count(self)}" self._prev: "FrameProcessor" | None = None diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index c2f4d8915..0862de9e3 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -9,7 +9,7 @@ import base64 import io import json -from typing import AsyncGenerator, List, Literal +from typing import Any, AsyncGenerator, List, Literal from loguru import logger from PIL import Image @@ -70,17 +70,29 @@ class BaseOpenAILLMService(LLMService): def __init__(self, model: str, api_key=None, base_url=None, **kwargs): super().__init__(**kwargs) self._model: str = model - self._client = self.create_client(api_key=api_key, base_url=base_url) + self._client = self.create_client(api_key=api_key, base_url=base_url, **kwargs) - def create_client(self, api_key=None, base_url=None): + def create_client(self, api_key=None, base_url=None, **kwargs): return AsyncOpenAI(api_key=api_key, base_url=base_url) def can_generate_metrics(self) -> bool: return True + async def get_chat_completions( + self, + context: OpenAILLMContext, + messages: List[ChatCompletionMessageParam]) -> AsyncStream[ChatCompletionChunk]: + chunks = await self._client.chat.completions.create( + model=self._model, + stream=True, + messages=messages, + tools=context.tools, + tool_choice=context.tool_choice, + ) + return chunks + async def _stream_chat_completions( - self, context: OpenAILLMContext - ) -> AsyncStream[ChatCompletionChunk]: + self, context: OpenAILLMContext) -> AsyncStream[ChatCompletionChunk]: logger.debug(f"Generating chat: {context.get_messages_json()}") messages: List[ChatCompletionMessageParam] = context.get_messages() @@ -97,15 +109,10 @@ class BaseOpenAILLMService(LLMService): del message["data"] del message["mime_type"] - chunks: AsyncStream[ChatCompletionChunk] = ( - await self._client.chat.completions.create( - model=self._model, - stream=True, - messages=messages, - tools=context.tools, - tool_choice=context.tool_choice, - ) - ) + try: + chunks = await self.get_chat_completions(context, messages) + except Exception as e: + logger.error(f"{self} exception: {e}") return chunks diff --git a/src/pipecat/services/openpipe.py b/src/pipecat/services/openpipe.py index e15118773..de9d9fcfa 100644 --- a/src/pipecat/services/openpipe.py +++ b/src/pipecat/services/openpipe.py @@ -1,159 +1,70 @@ -from pipecat.services.ai_services import LLMService -from openpipe import AsyncOpenAI as OpenPipeAI -from openpipe import AsyncStream -import os +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from typing import Dict, List + +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.openai import BaseOpenAILLMService + from loguru import logger -import secrets -import time -import base64 -from openai.types.chat import (ChatCompletionMessageParam, ChatCompletionChunk) -from typing import List -from pipecat.processors.frame_processor import FrameDirection -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame -from pipecat.frames.frames import ( - ErrorFrame, - Frame, - LLMFullResponseEndFrame, - LLMFullResponseStartFrame, - LLMMessagesFrame, - LLMResponseEndFrame, - LLMResponseStartFrame, - TextFrame, - URLImageRawFrame, - VisionImageRawFrame -) + +try: + from openpipe import AsyncOpenAI as OpenPipeAI, AsyncStream + from openai.types.chat import (ChatCompletionMessageParam, ChatCompletionChunk) +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use OpenPipe, you need to `pip install pipecat-ai[openpipe]`. Also, set `OPENPIPE_API_KEY` and `OPENAI_API_KEY` environment variables.") + raise Exception(f"Missing module: {e}") -class BaseOpenPipeLLMService(LLMService): +class OpenPipeLLMService(BaseOpenAILLMService): def __init__( self, - model: str, - c_id=None, - api_key=None, - openpipe_api_key=None, - openpipe_base_url=None, - prompt=None): - super().__init__() - self._model = model - self._client = self.create_client( - api_key=api_key, + model: str = "gpt-4o", + api_key: str | None = None, + base_url: str | None = None, + openpipe_api_key: str | None = None, + openpipe_base_url: str = "https://app.openpipe.ai/api/v1", + tags: Dict[str, str] | None = None, + **kwargs): + super().__init__( + model, + api_key, + base_url, openpipe_api_key=openpipe_api_key, - openpipe_base_url=openpipe_base_url) - self.c_id = c_id if c_id else secrets.token_urlsafe(16) - self.prompt = prompt - logger.debug(f"Client Created: {self._client}") + openpipe_base_url=openpipe_base_url, + **kwargs) + self._tags = tags - def create_client(self, api_key=None, openpipe_api_key=None, openpipe_base_url=None): - # Set up the OpenPipe client with the provided API keys and base URL + def create_client(self, api_key=None, base_url=None, **kwargs): + openpipe_api_key = kwargs.get("openpipe_api_key") or "" + openpipe_base_url = kwargs.get("openpipe_base_url") or "" client = OpenPipeAI( - api_key=api_key or os.environ.get("OPENAI_API_KEY"), + api_key=api_key, + base_url=base_url, openpipe={ - "api_key": openpipe_api_key or os.environ.get("OPENPIPE_API_KEY"), - "base_url": openpipe_base_url or "https://app.openpipe.ai/api/v1" + "api_key": openpipe_api_key, + "base_url": openpipe_base_url } ) return client - async def _stream_chat_completions(self, context): - logger.debug(f"Generating chat: {context.get_messages_json()}") - - messages: List[ChatCompletionMessageParam] = context.get_messages() - - # base64 encode any images - for message in messages: - if message.get("mime_type") == "image/jpeg": - encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8") - text = message["content"] - message["content"] = [ - {"type": "text", "text": text}, - {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{encoded_image}"}} - ] - del message["data"] - del message["mime_type"] - - start_time = time.time() - # Stream chat completions using the OpenPipe client - chunks = ( - await self._client.chat.completions.create( - model=self._model, - stream=True, - messages=messages, - openpipe={ - "tags": {"conversation_id": self.c_id, - "prompt": self.prompt}, - "log_request": True - } - ) + async def get_chat_completions( + self, + context: OpenAILLMContext, + messages: List[ChatCompletionMessageParam]) -> AsyncStream[ChatCompletionChunk]: + chunks = await self._client.chat.completions.create( + model=self._model, + stream=True, + messages=messages, + openpipe={ + "tags": self._tags, + "log_request": True + } ) - - logger.debug(f"OpenPipe LLM TTFB: {time.time() - start_time}") - return chunks - - async def _process_context(self, context): - function_name = "" - arguments = "" - - chunk_stream: AsyncStream[ChatCompletionChunk] = ( - await self._stream_chat_completions(context) - ) - - await self.push_frame(LLMFullResponseStartFrame()) - - async for chunk in chunk_stream: - if len(chunk.choices) == 0: - continue - - if chunk.choices[0].delta.tool_calls: - # We're streaming the LLM response to enable the fastest response times. - # For text, we just yield each chunk as we receive it and count on consumers - # to do whatever coalescing they need (eg. to pass full sentences to TTS) - # - # If the LLM is a function call, we'll do some coalescing here. - # If the response contains a function name, we'll yield a frame to tell consumers - # that they can start preparing to call the function with that name. - # We accumulate all the arguments for the rest of the streamed response, then when - # the response is done, we package up all the arguments and the function name and - # yield a frame containing the function name and the arguments. - - tool_call = chunk.choices[0].delta.tool_calls[0] - if tool_call.function and tool_call.function.name: - function_name += tool_call.function.name - # yield LLMFunctionStartFrame(function_name=tool_call.function.name) - if tool_call.function and tool_call.function.arguments: - # Keep iterating through the response to collect all the argument fragments and - # yield a complete LLMFunctionCallFrame after run_llm_async - # completes - arguments += tool_call.function.arguments - elif chunk.choices[0].delta.content: - await self.push_frame(LLMResponseStartFrame()) - await self.push_frame(TextFrame(chunk.choices[0].delta.content)) - await self.push_frame(LLMResponseEndFrame()) - - await self.push_frame(LLMFullResponseEndFrame()) - - # if we got a function name and arguments, yield the frame with all the info so - # frame consumers can take action based on the function call. - # if function_name and arguments: - # yield LLMFunctionCallFrame(function_name=function_name, arguments=arguments) - - async def process_frame(self, frame: Frame, direction: FrameDirection): - context = None - if isinstance(frame, OpenAILLMContextFrame): - context: OpenAILLMContext = frame.context - elif isinstance(frame, LLMMessagesFrame): - context = OpenAILLMContext.from_messages(frame.messages) - elif isinstance(frame, VisionImageRawFrame): - context = OpenAILLMContext.from_image_frame(frame) - else: - await self.push_frame(frame, direction) - - if context: - await self._process_context(context) - - -class OpenPipeLLMService(BaseOpenPipeLLMService): - - def __init__(self, model="gpt-4o", cli_id=None, **kwargs): - super().__init__(model, cli_id, **kwargs) From 18604e1a397f02014c110581cb96ad39a5d7bf81 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 13 Jun 2024 09:30:57 -0700 Subject: [PATCH 2/5] re-add removed CHANGELOG lines --- CHANGELOG.md | 36 ++++++++++++++++++++++++++++++------ 1 file changed, 30 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fa9d56551..950668ca2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,22 +9,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Added a new `Service`. This service will let you run OpenAI through +- Added `OpenPipeLLMService`. This service will let you run OpenAI through OpenPipe's SDK. +- Allow specifying frame processors' name through a new `name` constructor + argument. + ### Changed -- `OpenPipe` can now be used. Can call OpenAI through OpenPipe SDK to get - LLM logging stored in OpenPipe. +- `FrameSerializer.deserialize()` can now return `None` in case it is not + possible to desearialize the given data. + +- `daily_rest.DailyRoomProperties` now allows extra unknown parameters. + +- Added `DeepgramSTTService`. This service has an ongoing websocket + connection. To handle this, it subclasses `AIService` instead of + `STTService`. The output of this service will be pushed from the same task, + except system frames like `StartFrame`, `CancelFrame` or + `StartInterruptionFrame`. ### Fixed -- None +- Fixed an issue where `DailyRoomProperties.exp` always had the same old + timestamp unless set by the user. + +- Fixed a couple of issues with `WebsocketServerTransport`. It needed to use + `push_audio_frame()` and also VAD was not working properly. + +- Fixed an issue that would cause LLM aggregator to fail with small + `VADParams.stop_secs` values. + +- Fixed an issue where `BaseOutputTransport` would send longer audio frames + preventing interruptions. ### Other -- Added new `openpipe` example. This example shows how to use OpenPipe to run - OpenAI LLMs and get the logs stored in OpenPipe. +- Added new `07h-interruptible-openpipe.py` example. This example shows how to + use OpenPipe to run OpenAI LLMs and get the logs stored in OpenPipe. + +- Added new `dialin-chatbot` example. This examples shows how to call the bot + using a phone number. ## [0.0.29] - 2024-06-12 From 9992d826b1f10440ec8cb6e9fa3fd8ae507e154e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 13 Jun 2024 09:31:18 -0700 Subject: [PATCH 3/5] examples: renamed 06b-listen... to 07h-inte... --- ...npipe.py => 07h-interruptible-openpipe.py} | 33 ++++++++----------- 1 file changed, 13 insertions(+), 20 deletions(-) rename examples/foundational/{06b-listen-and-respond-openpipe.py => 07h-interruptible-openpipe.py} (80%) diff --git a/examples/foundational/06b-listen-and-respond-openpipe.py b/examples/foundational/07h-interruptible-openpipe.py similarity index 80% rename from examples/foundational/06b-listen-and-respond-openpipe.py rename to examples/foundational/07h-interruptible-openpipe.py index 9cfd681b8..240574f5f 100644 --- a/examples/foundational/06b-listen-and-respond-openpipe.py +++ b/examples/foundational/07h-interruptible-openpipe.py @@ -12,12 +12,11 @@ import sys from pipecat.frames.frames import LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_response import ( LLMAssistantResponseAggregator, LLMUserResponseAggregator, ) -from pipecat.processors.logger import FrameLogger from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.openpipe import OpenPipeLLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -36,9 +35,6 @@ logger.add(sys.stderr, level="DEBUG") async def main(room_url: str, token): - - timestamp = int(time.time()) - async with aiohttp.ClientSession() as session: transport = DailyTransport( room_url, @@ -58,16 +54,16 @@ async def main(room_url: str, token): voice_id=os.getenv("ELEVENLABS_VOICE_ID"), ) + timestamp = int(time.time()) llm = OpenPipeLLMService( api_key=os.getenv("OPENAI_API_KEY"), + openpipe_api_key=os.getenv("OPENPIPE_API_KEY"), model="gpt-4o", - cli_id=f"cli-{timestamp}" + tags={ + "conversation_id": f"pipecat-{timestamp}" + } ) - fl = FrameLogger("!!! after LLM", "red") - fltts = FrameLogger("@@@ out of tts", "green") - flend = FrameLogger("### out of the end", "magenta") - messages = [ { "role": "system", @@ -78,18 +74,15 @@ async def main(room_url: str, token): tma_out = LLMAssistantResponseAggregator(messages) pipeline = Pipeline([ - transport.input(), - tma_in, - llm, - fl, - tts, - fltts, - transport.output(), - tma_out, - flend + transport.input(), # Transport user input + tma_in, # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + tma_out # Assistant spoken responses ]) - task = PipelineTask(pipeline) + task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): From befb8db120aab3d0e6607464a068fd31f58c82de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 13 Jun 2024 09:33:56 -0700 Subject: [PATCH 4/5] update pyproject and requirements --- linux-py3.10-requirements.txt | 18 ++++++++++++++---- macos-py3.10-requirements.txt | 18 ++++++++++++++---- pyproject.toml | 2 +- 3 files changed, 29 insertions(+), 9 deletions(-) diff --git a/linux-py3.10-requirements.txt b/linux-py3.10-requirements.txt index d68106ee6..3201e35f2 100644 --- a/linux-py3.10-requirements.txt +++ b/linux-py3.10-requirements.txt @@ -18,7 +18,9 @@ aiosignal==1.3.1 annotated-types==0.7.0 # via pydantic anthropic==0.25.9 - # via pipecat-ai (pyproject.toml) + # via + # openpipe + # pipecat-ai (pyproject.toml) anyio==4.4.0 # via # anthropic @@ -29,7 +31,9 @@ async-timeout==4.0.3 # aiohttp # langchain attrs==23.2.0 - # via aiohttp + # via + # aiohttp + # openpipe av==12.1.0 # via faster-whisper azure-cognitiveservices-speech==1.37.0 @@ -77,7 +81,7 @@ fal-client==0.4.0 # via pipecat-ai (pyproject.toml) faster-whisper==1.0.2 # via pipecat-ai (pyproject.toml) -filelock==3.14.0 +filelock==3.15.1 # via # huggingface-hub # pyht @@ -150,6 +154,7 @@ httpx==0.27.0 # deepgram-sdk # fal-client # openai + # openpipe httpx-sse==0.4.0 # via fal-client huggingface-hub==0.23.3 @@ -264,8 +269,9 @@ onnxruntime==1.18.0 openai==1.26.0 # via # langchain-openai + # openpipe # pipecat-ai (pyproject.toml) -openpipe==4.13.0 +openpipe==4.14.0 # via pipecat-ai (pyproject.toml) orjson==3.10.4 # via langsmith @@ -328,6 +334,8 @@ pytest==8.2.2 # via pytest-asyncio pytest-asyncio==0.23.7 # via cartesia +python-dateutil==2.9.0.post0 + # via openpipe python-dotenv==1.0.1 # via pipecat-ai (pyproject.toml) pyyaml==6.0.1 @@ -362,6 +370,8 @@ safetensors==0.4.3 # transformers scipy==1.13.1 # via pyloudnorm +six==1.16.0 + # via python-dateutil sniffio==1.3.1 # via # anthropic diff --git a/macos-py3.10-requirements.txt b/macos-py3.10-requirements.txt index 38adada24..7d735527c 100644 --- a/macos-py3.10-requirements.txt +++ b/macos-py3.10-requirements.txt @@ -18,7 +18,9 @@ aiosignal==1.3.1 annotated-types==0.7.0 # via pydantic anthropic==0.25.9 - # via pipecat-ai (pyproject.toml) + # via + # openpipe + # pipecat-ai (pyproject.toml) anyio==4.4.0 # via # anthropic @@ -29,7 +31,9 @@ async-timeout==4.0.3 # aiohttp # langchain attrs==23.2.0 - # via aiohttp + # via + # aiohttp + # openpipe av==12.1.0 # via faster-whisper azure-cognitiveservices-speech==1.37.0 @@ -77,7 +81,7 @@ fal-client==0.4.0 # via pipecat-ai (pyproject.toml) faster-whisper==1.0.2 # via pipecat-ai (pyproject.toml) -filelock==3.14.0 +filelock==3.15.1 # via # huggingface-hub # pyht @@ -147,6 +151,7 @@ httpx==0.27.0 # deepgram-sdk # fal-client # openai + # openpipe httpx-sse==0.4.0 # via fal-client huggingface-hub==0.23.3 @@ -230,8 +235,9 @@ onnxruntime==1.18.0 openai==1.26.0 # via # langchain-openai + # openpipe # pipecat-ai (pyproject.toml) -openpipe==4.13.0 +openpipe==4.14.0 # via pipecat-ai (pyproject.toml) orjson==3.10.4 # via langsmith @@ -294,6 +300,8 @@ pytest==8.2.2 # via pytest-asyncio pytest-asyncio==0.23.7 # via cartesia +python-dateutil==2.9.0.post0 + # via openpipe python-dotenv==1.0.1 # via pipecat-ai (pyproject.toml) pyyaml==6.0.1 @@ -328,6 +336,8 @@ safetensors==0.4.3 # transformers scipy==1.13.1 # via pyloudnorm +six==1.16.0 + # via python-dateutil sniffio==1.3.1 # via # anthropic diff --git a/pyproject.toml b/pyproject.toml index e777a4caf..ee2d720f1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,11 +47,11 @@ langchain = [ "langchain~=0.2.1", "langchain-community~=0.2.1", "langchain-opena local = [ "pyaudio~=0.2.0" ] moondream = [ "einops~=0.8.0", "timm~=0.9.16", "transformers~=4.40.2" ] openai = [ "openai~=1.26.0" ] +openpipe = [ "openpipe~=4.14.0" ] playht = [ "pyht~=0.0.28" ] silero = [ "torch~=2.3.0", "torchaudio~=2.3.0" ] websocket = [ "websockets~=12.0" ] whisper = [ "faster-whisper~=1.0.2" ] -openpipe = [ "openpipe~=4.13.0" ] [tool.setuptools.packages.find] # All the following settings are optional: From 099e65f3b6420615eaafd09e60b5e4daa38b7cc8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 13 Jun 2024 09:36:01 -0700 Subject: [PATCH 5/5] report processor name in error logs --- src/pipecat/processors/frameworks/langchain.py | 4 ++-- src/pipecat/services/anthropic.py | 2 +- src/pipecat/services/azure.py | 6 +++--- src/pipecat/services/cartesia.py | 4 ++-- src/pipecat/services/deepgram.py | 4 ++-- src/pipecat/services/elevenlabs.py | 2 +- src/pipecat/services/fal.py | 2 +- src/pipecat/services/google.py | 4 ++-- src/pipecat/services/moondream.py | 2 +- src/pipecat/services/openai.py | 7 ++++--- src/pipecat/services/playht.py | 2 +- src/pipecat/services/whisper.py | 2 +- src/pipecat/transports/base_input.py | 2 +- src/pipecat/transports/base_output.py | 6 +++--- 14 files changed, 25 insertions(+), 24 deletions(-) diff --git a/src/pipecat/processors/frameworks/langchain.py b/src/pipecat/processors/frameworks/langchain.py index 674d3aa97..c1f769a16 100644 --- a/src/pipecat/processors/frameworks/langchain.py +++ b/src/pipecat/processors/frameworks/langchain.py @@ -73,7 +73,7 @@ class LangchainProcessor(FrameProcessor): await self.push_frame(TextFrame(self.__get_token_value(token))) await self.push_frame(LLMResponseEndFrame()) except GeneratorExit: - logger.warning("Generator was closed prematurely") + logger.warning(f"{self} generator was closed prematurely") except Exception as e: - logger.error(f"An unknown error occurred: {e}") + logger.error(f"{self} an unknown error occurred: {e}") await self.push_frame(LLMFullResponseEndFrame()) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 0f3d7d241..688d43b14 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -122,7 +122,7 @@ class AnthropicLLMService(LLMService): await self.push_frame(LLMResponseEndFrame()) except Exception as e: - logger.error(f"Anthropic exception: {e}") + logger.error(f"{self} exception: {e}") finally: await self.push_frame(LLMFullResponseEndFrame()) diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index f4184f647..2821bb7b7 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -72,7 +72,7 @@ class AzureTTSService(TTSService): cancellation_details = result.cancellation_details logger.warning(f"Speech synthesis canceled: {cancellation_details.reason}") if cancellation_details.reason == CancellationReason.Error: - logger.error(f"Error details: {cancellation_details.error_details}") + logger.error(f"{self} error: {cancellation_details.error_details}") class AzureLLMService(BaseOpenAILLMService): @@ -143,7 +143,7 @@ class AzureImageGenServiceREST(ImageGenService): while status != "succeeded": attempts_left -= 1 if attempts_left == 0: - logger.error("Image generation timed out") + logger.error(f"{self} error: image generation timed out") yield ErrorFrame("Image generation timed out") return @@ -156,7 +156,7 @@ class AzureImageGenServiceREST(ImageGenService): image_url = json_response["result"]["data"][0]["url"] if json_response else None if not image_url: - logger.error("Image generation failed") + logger.error(f"{self} error: image generation failed") yield ErrorFrame("Image generation failed") return diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 161f0589b..f81226576 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -37,7 +37,7 @@ class CartesiaTTSService(TTSService): voice_id = voices[self._voice_name]["id"] self._voice = self._client.get_voice_embedding(voice_id=voice_id) except Exception as e: - logger.error(f"Cartesia initialization error: {e}") + logger.error(f"{self} initialization error: {e}") def can_generate_metrics(self) -> bool: return True @@ -60,4 +60,4 @@ class CartesiaTTSService(TTSService): await self.stop_ttfb_metrics() yield AudioRawFrame(chunk["audio"], chunk["sampling_rate"], 1) except Exception as e: - logger.error(f"Cartesia exception: {e}") + logger.error(f"{self} exception: {e}") diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index aa00d762a..810d7518d 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -74,7 +74,7 @@ class DeepgramTTSService(TTSService): return logger.error( - f"Error getting audio (status: {r.status}, error: {response_text})") + f"{self} error getting audio (status: {r.status}, error: {response_text})") yield ErrorFrame(f"Error getting audio (status: {r.status}, error: {response_text})") return @@ -83,7 +83,7 @@ class DeepgramTTSService(TTSService): frame = AudioRawFrame(audio=data, sample_rate=16000, num_channels=1) yield frame except Exception as e: - logger.error(f"Deepgram exception: {e}") + logger.error(f"{self} exception: {e}") class DeepgramSTTService(AIService): diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 256f077c3..c24736672 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -55,7 +55,7 @@ class ElevenLabsTTSService(TTSService): async with self._aiohttp_session.post(url, json=payload, headers=headers, params=querystring) as r: if r.status != 200: text = await r.text() - logger.error(f"Error getting audio (status: {r.status}, error: {text})") + logger.error(f"{self} error getting audio (status: {r.status}, error: {text})") yield ErrorFrame(f"Error getting audio (status: {r.status}, error: {text})") return diff --git a/src/pipecat/services/fal.py b/src/pipecat/services/fal.py index 14bb0a7f1..13b48de49 100644 --- a/src/pipecat/services/fal.py +++ b/src/pipecat/services/fal.py @@ -62,7 +62,7 @@ class FalImageGenService(ImageGenService): image_url = response["images"][0]["url"] if response else None if not image_url: - logger.error("Image generation failed") + logger.error(f"{self} error: image generation failed") yield ErrorFrame("Image generation failed") return diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index b1da9ddf6..5a20a269e 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -104,10 +104,10 @@ class GoogleLLMService(LLMService): logger.debug( f"LLM refused to generate content for safety reasons - {messages}.") else: - logger.error(f"Error {e}") + logger.error(f"{self} error: {e}") except Exception as e: - logger.error(f"Exception: {e}") + logger.error(f"{self} exception: {e}") finally: await self.push_frame(LLMFullResponseEndFrame()) diff --git a/src/pipecat/services/moondream.py b/src/pipecat/services/moondream.py index 508b8d24f..2c105c92d 100644 --- a/src/pipecat/services/moondream.py +++ b/src/pipecat/services/moondream.py @@ -71,7 +71,7 @@ class MoondreamService(VisionService): async def run_vision(self, frame: VisionImageRawFrame) -> AsyncGenerator[Frame, None]: if not self._model: - logger.error("Moondream model not available") + logger.error(f"{self} error: Moondream model not available") yield ErrorFrame("Moondream model not available") return diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 0862de9e3..ce480e783 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -270,7 +270,7 @@ class OpenAIImageGenService(ImageGenService): image_url = image.data[0].url if not image_url: - logger.error(f"No image provided in response: {image}") + logger.error(f"{self} No image provided in response: {image}") yield ErrorFrame("Image generation failed") return @@ -324,7 +324,8 @@ class OpenAITTSService(TTSService): ) as r: if r.status_code != 200: error = await r.text() - logger.error(f"Error getting audio (status: {r.status_code}, error: {error})") + logger.error( + f"{self} error getting audio (status: {r.status_code}, error: {error})") yield ErrorFrame(f"Error getting audio (status: {r.status_code}, error: {error})") return async for chunk in r.iter_bytes(8192): @@ -333,4 +334,4 @@ class OpenAITTSService(TTSService): frame = AudioRawFrame(chunk, 24_000, 1) yield frame except BadRequestError as e: - logger.error(f"Error generating TTS: {e}") + logger.error(f"{self} error generating TTS: {e}") diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index c44719047..6b885f811 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -80,4 +80,4 @@ class PlayHTTTSService(TTSService): frame = AudioRawFrame(chunk, 16000, 1) yield frame except Exception as e: - logger.error(f"Error generating TTS: {e}") + logger.error(f"{self} error generating TTS: {e}") diff --git a/src/pipecat/services/whisper.py b/src/pipecat/services/whisper.py index 6dfc5080e..9cd3150dc 100644 --- a/src/pipecat/services/whisper.py +++ b/src/pipecat/services/whisper.py @@ -72,8 +72,8 @@ class WhisperSTTService(STTService): async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: """Transcribes given audio using Whisper""" if not self._model: + logger.error(f"{self} error: Whisper model not available") yield ErrorFrame("Whisper model not available") - logger.error("Whisper model not available") return await self.start_ttfb_metrics() diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 0d4817e62..d7e3c1c9d 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -186,4 +186,4 @@ class BaseInputTransport(FrameProcessor): except queue.Empty: pass except BaseException as e: - logger.error(f"Error reading audio frames: {e}") + logger.error(f"{self} error reading audio frames: {e}") diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 86802d552..d46201294 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -201,7 +201,7 @@ class BaseOutputTransport(FrameProcessor): except queue.Empty: pass except BaseException as e: - logger.error(f"Error processing sink queue: {e}") + logger.error(f"{self} error processing sink queue: {e}") # # Push frames task @@ -270,7 +270,7 @@ class BaseOutputTransport(FrameProcessor): except queue.Empty: pass except Exception as e: - logger.error(f"Error writing to camera: {e}") + logger.error(f"{self} error writing to camera: {e}") # # Audio out @@ -286,5 +286,5 @@ class BaseOutputTransport(FrameProcessor): buffer = buffer[self._audio_chunk_size:] return buffer except BaseException as e: - logger.error(f"Error writing audio frames: {e}") + logger.error(f"{self} error writing audio frames: {e}") return buffer