services(openpipe): refactored so it's based on BaseOpenAILLMService

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/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
 

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)