From 5c86f8e687987cef915d8b9aecc03507876e11bc Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 7 Aug 2025 11:14:15 -0400 Subject: [PATCH 1/3] Add timeout/retry logic and refactor parameter building in BaseOpenAILLMService - Add timeout (default 5.0s) and retry_on_timeout parameters to constructor - Implement timeout/retry logic in get_chat_completions using asyncio.wait_for - Extract build_chat_completion_params() as public method for subclass customization --- src/pipecat/services/openai/base_llm.py | 44 ++++++++++++++++++++++--- 1 file changed, 40 insertions(+), 4 deletions(-) diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index f88134fa2..5fb11ea4d 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -6,6 +6,7 @@ """Base OpenAI LLM service implementation.""" +import asyncio import base64 import json from typing import Any, Dict, List, Mapping, Optional @@ -14,6 +15,7 @@ import httpx from loguru import logger from openai import ( NOT_GIVEN, + APITimeoutError, AsyncOpenAI, AsyncStream, DefaultAsyncHttpxClient, @@ -91,6 +93,8 @@ class BaseOpenAILLMService(LLMService): project=None, default_headers: Optional[Mapping[str, str]] = None, params: Optional[InputParams] = None, + timeout: Optional[float] = 5.0, + retry_on_timeout: Optional[bool] = False, **kwargs, ): """Initialize the BaseOpenAILLMService. @@ -103,6 +107,8 @@ class BaseOpenAILLMService(LLMService): project: OpenAI project ID. default_headers: Additional HTTP headers to include in requests. params: Input parameters for model configuration and behavior. + timeout: Request timeout in seconds. Defaults to 5.0 seconds. + retry_on_timeout: Whether to retry the request once if it times out. **kwargs: Additional arguments passed to the parent LLMService. """ super().__init__(**kwargs) @@ -119,6 +125,8 @@ class BaseOpenAILLMService(LLMService): "max_completion_tokens": params.max_completion_tokens, "extra": params.extra if isinstance(params.extra, dict) else {}, } + self._timeout = timeout + self._retry_on_timeout = retry_on_timeout self.set_model_name(model) self._client = self.create_client( api_key=api_key, @@ -175,7 +183,7 @@ class BaseOpenAILLMService(LLMService): async def get_chat_completions( self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] ) -> AsyncStream[ChatCompletionChunk]: - """Get streaming chat completions from OpenAI API. + """Get streaming chat completions from OpenAI API with optional timeout and retry. Args: context: The LLM context containing tools and configuration. @@ -184,6 +192,36 @@ class BaseOpenAILLMService(LLMService): Returns: Async stream of chat completion chunks. """ + params = self.build_chat_completion_params(context, messages) + + if self._retry_on_timeout: + try: + chunks = await asyncio.wait_for( + self._client.chat.completions.create(**params), timeout=self._timeout + ) + return chunks + except (APITimeoutError, asyncio.TimeoutError): + # Retry, this time without a timeout so we get a response + chunks = await self._client.chat.completions.create(**params) + return chunks + else: + chunks = await self._client.chat.completions.create(**params) + return chunks + + def build_chat_completion_params( + self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] + ) -> dict: + """Build parameters for chat completion request. + + Subclasses can override this to customize parameters for different providers. + + Args: + context: The LLM context containing tools and configuration. + messages: List of chat completion messages to send. + + Returns: + Dictionary of parameters for the chat completion request. + """ params = { "model": self.model_name, "stream": True, @@ -201,9 +239,7 @@ class BaseOpenAILLMService(LLMService): } params.update(self._settings["extra"]) - - chunks = await self._client.chat.completions.create(**params) - return chunks + return params async def _stream_chat_completions( self, context: OpenAILLMContext From 4c029fcfa7d3c21322864a5b7b9e471817b68b32 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 7 Aug 2025 11:21:42 -0400 Subject: [PATCH 2/3] Update OpenAILLMService subclasses to use the new build_chat_completion_params function --- CHANGELOG.md | 8 ++++++ src/pipecat/services/cerebras/llm.py | 24 ++++++------------ src/pipecat/services/deepseek/llm.py | 23 +++++------------ src/pipecat/services/fireworks/llm.py | 21 +++++----------- src/pipecat/services/openpipe/llm.py | 34 ++++++++++++++------------ src/pipecat/services/perplexity/llm.py | 20 ++++++--------- src/pipecat/services/sambanova/llm.py | 19 +++++++------- 7 files changed, 63 insertions(+), 86 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c6a93dff9..005f5293d 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/pipecat/services/cerebras/llm.py b/src/pipecat/services/cerebras/llm.py index fb8e21a2b..c3577af82 100644 --- a/src/pipecat/services/cerebras/llm.py +++ b/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 ChatCompletionChunk, ChatCompletionMessageParam +from openai.types.chat import 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]: - """Create a streaming chat completion using Cerebras's API. + ) -> dict: + """Build parameters for Cerebras chat completion request. - Args: - 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. + Cerebras supports a subset of OpenAI parameters, focusing on core + completion settings without advanced features like frequency/presence penalties. """ params = { "model": self.model_name, @@ -83,6 +75,4 @@ class CerebrasLLMService(OpenAILLMService): } params.update(self._settings["extra"]) - - chunks = await self._client.chat.completions.create(**params) - return chunks + return params diff --git a/src/pipecat/services/deepseek/llm.py b/src/pipecat/services/deepseek/llm.py index 55aaa341f..1e6bfcc5c 100644 --- a/src/pipecat/services/deepseek/llm.py +++ b/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 ChatCompletionChunk, ChatCompletionMessageParam +from openai.types.chat import 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]: - """Create a streaming chat completion using DeepSeek's API. + ) -> dict: + """Build parameters for DeepSeek chat completion request. - Args: - 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. + DeepSeek doesn't support some OpenAI parameters like seed and max_completion_tokens. """ params = { "model": self.model_name, @@ -85,6 +76,4 @@ class DeepSeekLLMService(OpenAILLMService): } params.update(self._settings["extra"]) - - chunks = await self._client.chat.completions.create(**params) - return chunks + return params diff --git a/src/pipecat/services/fireworks/llm.py b/src/pipecat/services/fireworks/llm.py index 9edd6215c..e28ff9759 100644 --- a/src/pipecat/services/fireworks/llm.py +++ b/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] - ): - """Get chat completions from Fireworks API. + ) -> dict: + """Build parameters for Fireworks chat completion request. - Removes OpenAI-specific parameters not supported by Fireworks and - configures the request with Fireworks-compatible settings. - - 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. + Fireworks doesn't support some OpenAI parameters like seed, max_completion_tokens, + and stream_options. """ params = { "model": self.model_name, @@ -83,6 +76,4 @@ class FireworksLLMService(OpenAILLMService): } params.update(self._settings["extra"]) - - chunks = await self._client.chat.completions.create(**params) - return chunks + return params diff --git a/src/pipecat/services/openpipe/llm.py b/src/pipecat/services/openpipe/llm.py index 25257e294..581bb045f 100644 --- a/src/pipecat/services/openpipe/llm.py +++ b/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]: - """Generate streaming chat completions with OpenPipe logging. + ) -> dict: + """Build parameters for OpenPipe chat completion request. + + Adds OpenPipe-specific logging and tagging parameters. Args: - context: The OpenAI LLM context containing conversation state. - messages: List of chat completion message parameters. + context: The LLM context containing tools and configuration. + 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( - model=self.model_name, - stream=True, - messages=messages, - openpipe={"tags": self._tags, "log_request": True}, - ) - return chunks + # Start with base parameters + params = super().build_chat_completion_params(context, messages) + + # Add OpenPipe-specific parameters + params["openpipe"] = { + "tags": self._tags, + "log_request": True, + } + + return params diff --git a/src/pipecat/services/perplexity/llm.py b/src/pipecat/services/perplexity/llm.py index ae80b3942..59cbe520b 100644 --- a/src/pipecat/services/perplexity/llm.py +++ b/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.types.chat import ChatCompletionChunk, ChatCompletionMessageParam +from openai import NOT_GIVEN +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]: - """Get chat completions from Perplexity API using OpenAI-compatible parameters. + ) -> dict: + """Build parameters for Perplexity chat completion request. - Args: - 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. + Perplexity uses a subset of OpenAI parameters and doesn't support tools. """ 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 chunks + return params async def _process_context(self, context: OpenAILLMContext): """Process a context through the LLM and accumulate token usage metrics. diff --git a/src/pipecat/services/sambanova/llm.py b/src/pipecat/services/sambanova/llm.py index c11489e66..815d1c8be 100644 --- a/src/pipecat/services/sambanova/llm.py +++ b/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: - """Get chat completions from SambaNova API endpoint. + ) -> dict: + """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. - messages: List of chat completion message parameters. + context: The LLM context containing tools and configuration. + 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"]) - - chunks = await self._client.chat.completions.create(**params) - return chunks + return params @traced_llm # type: ignore async def _process_context(self, context: OpenAILLMContext) -> AsyncStream[ChatCompletionChunk]: From 8714c9137fc3e36a12584f85ed7a184b4864f247 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 12 Aug 2025 14:44:34 -0400 Subject: [PATCH 3/3] Code review fixes --- CHANGELOG.md | 15 +++++---------- src/pipecat/services/openai/base_llm.py | 9 +++++---- 2 files changed, 10 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 005f5293d..da4b2be40 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 Gemini model can be prompted to insert styled speech to control the TTS output. +- For `OpenAILLMService` and its subclasses, added the ability to retry + executing a chat completion after a timeout period. The new args are + `retry_timeout_secs` and `retry_on_timeout`. This feature is disabled by + default. + - Added Exotel support to Pipecat's development runner. You can now connect using the runner with `uv run bot.py -t exotel` and an ngrok connection to HTTP port 7860. @@ -63,8 +68,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue that would cause system frames to not be processed with higher priority than other frames. This could cause slower interruption times. -### Fixed - - Fixed an issue where retrying a websocket connection error would result in an error. @@ -106,14 +109,6 @@ 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 diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 5fb11ea4d..ecde66eec 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -93,7 +93,7 @@ class BaseOpenAILLMService(LLMService): project=None, default_headers: Optional[Mapping[str, str]] = None, params: Optional[InputParams] = None, - timeout: Optional[float] = 5.0, + retry_timeout_secs: Optional[float] = 5.0, retry_on_timeout: Optional[bool] = False, **kwargs, ): @@ -107,7 +107,7 @@ class BaseOpenAILLMService(LLMService): project: OpenAI project ID. default_headers: Additional HTTP headers to include in requests. params: Input parameters for model configuration and behavior. - timeout: Request timeout in seconds. Defaults to 5.0 seconds. + retry_timeout_secs: Request timeout in seconds. Defaults to 5.0 seconds. retry_on_timeout: Whether to retry the request once if it times out. **kwargs: Additional arguments passed to the parent LLMService. """ @@ -125,7 +125,7 @@ class BaseOpenAILLMService(LLMService): "max_completion_tokens": params.max_completion_tokens, "extra": params.extra if isinstance(params.extra, dict) else {}, } - self._timeout = timeout + self._retry_timeout_secs = retry_timeout_secs self._retry_on_timeout = retry_on_timeout self.set_model_name(model) self._client = self.create_client( @@ -197,11 +197,12 @@ class BaseOpenAILLMService(LLMService): if self._retry_on_timeout: try: chunks = await asyncio.wait_for( - self._client.chat.completions.create(**params), timeout=self._timeout + self._client.chat.completions.create(**params), timeout=self._retry_timeout_secs ) return chunks except (APITimeoutError, asyncio.TimeoutError): # Retry, this time without a timeout so we get a response + logger.debug(f"{self}: Retrying chat completion due to timeout") chunks = await self._client.chat.completions.create(**params) return chunks else: