Merge pull request #235 from pipecat-ai/aleix/openpipe-refactoring

openpipe refactoring

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 
+- `FrameSerializer.deserialize()` can now return `None` in case it is not
-  LLM logging stored in OpenPipe.
+  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
+- Added new `07h-interruptible-openpipe.py` example. This example shows how to
-  OpenAI LLMs and get the logs stored in OpenPipe.
+  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
 

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(),
+            transport.input(),   # Transport user input
-            tma_in,
+            tma_in,              # User responses
-            llm,
+            llm,                 # LLM
-            fl,
+            tts,                 # TTS
-            tts,
+            transport.output(),  # Transport bot output
-            fltts,
+            tma_out              # Assistant spoken responses
-            transport.output(),
-            tma_out,
-            flend
         ])
 
-        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):

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

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

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:

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

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())

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())
 

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
 

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}")

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):

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
 

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
 

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())
 

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
 

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
+            self, context: OpenAILLMContext) -> AsyncStream[ChatCompletionChunk]:
-    ) -> 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] = (
+        try:
-            await self._client.chat.completions.create(
+            chunks = await self.get_chat_completions(context, messages)
-                model=self._model,
+        except Exception as e:
-                stream=True,
+            logger.error(f"{self} exception: {e}")
-                messages=messages,
-                tools=context.tools,
-                tool_choice=context.tool_choice,
-            )
-        )
 
         return chunks
 
@@ -263,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
 
@@ -317,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):
@@ -326,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}")

src/pipecat/services/openpipe.py

@@ -1,159 +1,70 @@
-from pipecat.services.ai_services import LLMService
+#
-from openpipe import AsyncOpenAI as OpenPipeAI
+# Copyright (c) 2024, Daily
-from openpipe import AsyncStream
+#
-import os
+# 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
+try:
-import base64
+    from openpipe import AsyncOpenAI as OpenPipeAI, AsyncStream
-from openai.types.chat import (ChatCompletionMessageParam, ChatCompletionChunk)
+    from openai.types.chat import (ChatCompletionMessageParam, ChatCompletionChunk)
-from typing import List
+except ModuleNotFoundError as e:
-from pipecat.processors.frame_processor import FrameDirection
+    logger.error(f"Exception: {e}")
-from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext, OpenAILLMContextFrame
+    logger.error(
-from pipecat.frames.frames import (
+        "In order to use OpenPipe, you need to `pip install pipecat-ai[openpipe]`. Also, set `OPENPIPE_API_KEY` and `OPENAI_API_KEY` environment variables.")
-    ErrorFrame,
+    raise Exception(f"Missing module: {e}")
-    Frame,
-    LLMFullResponseEndFrame,
-    LLMFullResponseStartFrame,
-    LLMMessagesFrame,
-    LLMResponseEndFrame,
-    LLMResponseStartFrame,
-    TextFrame,
-    URLImageRawFrame,
-    VisionImageRawFrame
-)
 
 
-class BaseOpenPipeLLMService(LLMService):
+class OpenPipeLLMService(BaseOpenAILLMService):
 
     def __init__(
             self,
-            model: str,
+            model: str = "gpt-4o",
-            c_id=None,
+            api_key: str | None = None,
-            api_key=None,
+            base_url: str | None = None,
-            openpipe_api_key=None,
+            openpipe_api_key: str | None = None,
-            openpipe_base_url=None,
+            openpipe_base_url: str = "https://app.openpipe.ai/api/v1",
-            prompt=None):
+            tags: Dict[str, str] | None = None,
-        super().__init__()
+            **kwargs):
-        self._model = model
+        super().__init__(
-        self._client = self.create_client(
+            model,
-            api_key=api_key,
+            api_key,
+            base_url,
             openpipe_api_key=openpipe_api_key,
-            openpipe_base_url=openpipe_base_url)
+            openpipe_base_url=openpipe_base_url,
-        self.c_id = c_id if c_id else secrets.token_urlsafe(16)
+            **kwargs)
-        self.prompt = prompt
+        self._tags = tags
-        logger.debug(f"Client Created: {self._client}")
 
-    def create_client(self, api_key=None, openpipe_api_key=None, openpipe_base_url=None):
+    def create_client(self, api_key=None, base_url=None, **kwargs):
-        # Set up the OpenPipe client with the provided API keys and base URL
+        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"),
+                "api_key": openpipe_api_key,
-                "base_url": openpipe_base_url or "https://app.openpipe.ai/api/v1"
+                "base_url": openpipe_base_url
             }
         )
         return client
 
-    async def _stream_chat_completions(self, context):
+    async def get_chat_completions(
-        logger.debug(f"Generating chat: {context.get_messages_json()}")
+            self,
-
+            context: OpenAILLMContext,
-        messages: List[ChatCompletionMessageParam] = context.get_messages()
+            messages: List[ChatCompletionMessageParam]) -> AsyncStream[ChatCompletionChunk]:
-
+        chunks = await self._client.chat.completions.create(
-        # base64 encode any images
+            model=self._model,
-        for message in messages:
+            stream=True,
-            if message.get("mime_type") == "image/jpeg":
+            messages=messages,
-                encoded_image = base64.b64encode(message["data"].getvalue()).decode("utf-8")
+            openpipe={
-                text = message["content"]
+                "tags": self._tags,
-                message["content"] = [
+                "log_request": True
-                    {"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
-                }
-            )
         )
-
-        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)

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}")

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()

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}")

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