Update OpenAILLMService subclasses to use the new build_chat_completion_params function

CHANGELOG.md

@@ -106,6 +106,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `LLMUserContextAggregator` and `LLMAssistantResponseAggregator` (or
   LLM-specific subclasses thereof) instead.
 
+## [Unreleased]
+
+### Added
+
+- For `OpenAILLMService` and its subclasses, added the ability to retry
+  executing a chat completion after a timeout period. The new args are
+  `timeout` and `retry_on_timeout`. This feature is disabled by default.
+
 ## [0.0.78] - 2025-08-07
 
 ### Added

src/pipecat/services/cerebras/llm.py

@@ -9,8 +9,7 @@
 from typing import List
 
 from loguru import logger
-from openai import AsyncStream
+from openai.types.chat import ChatCompletionMessageParam
-from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
 
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai.llm import OpenAILLMService
@@ -55,20 +54,13 @@ class CerebrasLLMService(OpenAILLMService):
         logger.debug(f"Creating Cerebras client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
-    async def get_chat_completions(
+    def build_chat_completion_params(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> AsyncStream[ChatCompletionChunk]:
+    ) -> dict:
-        """Create a streaming chat completion using Cerebras's API.
+        """Build parameters for Cerebras chat completion request.
 
-        Args:
+        Cerebras supports a subset of OpenAI parameters, focusing on core
-            context: The context object containing tools configuration
+        completion settings without advanced features like frequency/presence penalties.
-                and other settings for the chat completion.
-            messages: The list of messages comprising
-                the conversation history and current request.
-
-        Returns:
-            A streaming response of chat completion
-                chunks that can be processed asynchronously.
         """
         params = {
             "model": self.model_name,
@@ -83,6 +75,4 @@ class CerebrasLLMService(OpenAILLMService):
         }
 
         params.update(self._settings["extra"])
-
+        return params
-        chunks = await self._client.chat.completions.create(**params)
-        return chunks

src/pipecat/services/deepseek/llm.py

@@ -9,8 +9,7 @@
 from typing import List
 
 from loguru import logger
-from openai import AsyncStream
+from openai.types.chat import ChatCompletionMessageParam
-from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
 
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai.llm import OpenAILLMService
@@ -55,20 +54,12 @@ class DeepSeekLLMService(OpenAILLMService):
         logger.debug(f"Creating DeepSeek client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
-    async def get_chat_completions(
+    def _build_chat_completion_params(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> AsyncStream[ChatCompletionChunk]:
+    ) -> dict:
-        """Create a streaming chat completion using DeepSeek's API.
+        """Build parameters for DeepSeek chat completion request.
 
-        Args:
+        DeepSeek doesn't support some OpenAI parameters like seed and max_completion_tokens.
-            context: The context object containing tools configuration
-                and other settings for the chat completion.
-            messages: The list of messages comprising the conversation
-                history and current request.
-
-        Returns:
-            A streaming response of chat completion chunks that can be
-            processed asynchronously.
         """
         params = {
             "model": self.model_name,
@@ -85,6 +76,4 @@ class DeepSeekLLMService(OpenAILLMService):
         }
 
         params.update(self._settings["extra"])
-
+        return params
-        chunks = await self._client.chat.completions.create(**params)
-        return chunks

src/pipecat/services/fireworks/llm.py

@@ -54,20 +54,13 @@ class FireworksLLMService(OpenAILLMService):
         logger.debug(f"Creating Fireworks client with api {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
-    async def get_chat_completions(
+    def build_chat_completion_params(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ):
+    ) -> dict:
-        """Get chat completions from Fireworks API.
+        """Build parameters for Fireworks chat completion request.
 
-        Removes OpenAI-specific parameters not supported by Fireworks and
+        Fireworks doesn't support some OpenAI parameters like seed, max_completion_tokens,
-        configures the request with Fireworks-compatible settings.
+        and stream_options.
-
-        Args:
-            context: The OpenAI LLM context containing tools and settings.
-            messages: List of chat completion message parameters.
-
-        Returns:
-            Async generator yielding chat completion chunks from Fireworks API.
         """
         params = {
             "model": self.model_name,
@@ -83,6 +76,4 @@ class FireworksLLMService(OpenAILLMService):
         }
 
         params.update(self._settings["extra"])
-
+        return params
-        chunks = await self._client.chat.completions.create(**params)
-        return chunks

src/pipecat/services/openpipe/llm.py

@@ -13,14 +13,13 @@ enabling integration with OpenPipe's fine-tuning and monitoring capabilities.
 from typing import Dict, List, Optional
 
 from loguru import logger
-from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
+from openai.types.chat import ChatCompletionMessageParam
 
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.services.openai.llm import OpenAILLMService
 
 try:
     from openpipe import AsyncOpenAI as OpenPipeAI
-    from openpipe import AsyncStream
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error("In order to use OpenPipe, you need to `pip install pipecat-ai[openpipe]`.")
@@ -87,22 +86,27 @@ class OpenPipeLLMService(OpenAILLMService):
         )
         return client
 
-    async def get_chat_completions(
+    def build_chat_completion_params(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> AsyncStream[ChatCompletionChunk]:
+    ) -> dict:
-        """Generate streaming chat completions with OpenPipe logging.
+        """Build parameters for OpenPipe chat completion request.
+
+        Adds OpenPipe-specific logging and tagging parameters.
 
         Args:
-            context: The OpenAI LLM context containing conversation state.
+            context: The LLM context containing tools and configuration.
-            messages: List of chat completion message parameters.
+            messages: List of chat completion messages to send.
 
         Returns:
-            Async stream of chat completion chunks.
+            Dictionary of parameters for the chat completion request.
         """
-        chunks = await self._client.chat.completions.create(
+        # Start with base parameters
-            model=self.model_name,
+        params = super().build_chat_completion_params(context, messages)
-            stream=True,
+
-            messages=messages,
+        # Add OpenPipe-specific parameters
-            openpipe={"tags": self._tags, "log_request": True},
+        params["openpipe"] = {
-        )
+            "tags": self._tags,
-        return chunks
+            "log_request": True,
+        }
+
+        return params

src/pipecat/services/perplexity/llm.py

@@ -13,8 +13,8 @@ reporting patterns while maintaining compatibility with the Pipecat framework.
 
 from typing import List
 
-from openai import NOT_GIVEN, AsyncStream
+from openai import NOT_GIVEN
-from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
+from openai.types.chat import ChatCompletionMessageParam
 
 from pipecat.metrics.metrics import LLMTokenUsage
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
@@ -53,17 +53,12 @@ class PerplexityLLMService(OpenAILLMService):
         self._has_reported_prompt_tokens = False
         self._is_processing = False
 
-    async def get_chat_completions(
+    def build_chat_completion_params(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> AsyncStream[ChatCompletionChunk]:
+    ) -> dict:
-        """Get chat completions from Perplexity API using OpenAI-compatible parameters.
+        """Build parameters for Perplexity chat completion request.
 
-        Args:
+        Perplexity uses a subset of OpenAI parameters and doesn't support tools.
-            context: The context containing conversation history and settings.
-            messages: The messages to send to the API.
-
-        Returns:
-            A stream of chat completion chunks from the Perplexity API.
         """
         params = {
             "model": self.model_name,
@@ -83,8 +78,7 @@ class PerplexityLLMService(OpenAILLMService):
         if self._settings["max_tokens"] is not NOT_GIVEN:
             params["max_tokens"] = self._settings["max_tokens"]
 
-        chunks = await self._client.chat.completions.create(**params)
+        return params
-        return chunks
 
     async def _process_context(self, context: OpenAILLMContext):
         """Process a context through the LLM and accumulate token usage metrics.

src/pipecat/services/sambanova/llm.py

@@ -68,17 +68,20 @@ class SambaNovaLLMService(OpenAILLMService):  # type: ignore
         logger.debug(f"Creating SambaNova client with API {base_url}")
         return super().create_client(api_key, base_url, **kwargs)
 
-    async def get_chat_completions(
+    def build_chat_completion_params(
         self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam]
-    ) -> Any:
+    ) -> dict:
-        """Get chat completions from SambaNova API endpoint.
+        """Build parameters for SambaNova chat completion request.
+
+        SambaNova doesn't support some OpenAI parameters like frequency_penalty,
+        presence_penalty, and seed.
 
         Args:
-            context: OpenAI LLM context containing tools and configuration.
+            context: The LLM context containing tools and configuration.
-            messages: List of chat completion message parameters.
+            messages: List of chat completion messages to send.
 
         Returns:
-            Chat completion response stream from SambaNova API.
+            Dictionary of parameters for the chat completion request.
         """
         params = {
             "model": self.model_name,
@@ -94,9 +97,7 @@ class SambaNovaLLMService(OpenAILLMService):  # type: ignore
         }
 
         params.update(self._settings["extra"])
-
+        return params
-        chunks = await self._client.chat.completions.create(**params)
-        return chunks
 
     @traced_llm  # type: ignore
     async def _process_context(self, context: OpenAILLMContext) -> AsyncStream[ChatCompletionChunk]: