diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 2badfed96..abeb1ea11 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -4,6 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""Base OpenAI LLM service implementation.""" + import base64 import json from typing import Any, Dict, List, Mapping, Optional @@ -39,16 +41,39 @@ from pipecat.utils.tracing.service_decorators import traced_llm class BaseOpenAILLMService(LLMService): - """This is the base for all services that use the AsyncOpenAI client. + """Base class for all services that use the AsyncOpenAI client. This service consumes OpenAILLMContextFrame frames, which contain a reference - to an OpenAILLMContext frame. The OpenAILLMContext object defines the context - sent to the LLM for a completion. This includes user, assistant and system messages - as well as tool choices and the tool, which is used if requesting function - calls from the LLM. + to an OpenAILLMContext object. The context defines what is sent to the LLM for + completion, including user, assistant, and system messages, as well as tool + choices and function call configurations. + + Args: + model: The OpenAI model name to use (e.g., "gpt-4.1", "gpt-4o"). + api_key: OpenAI API key. If None, uses environment variable. + base_url: Custom base URL for OpenAI API. If None, uses default. + organization: OpenAI organization ID. + project: OpenAI project ID. + default_headers: Additional HTTP headers to include in requests. + params: Input parameters for model configuration and behavior. + **kwargs: Additional arguments passed to the parent LLMService. """ class InputParams(BaseModel): + """Input parameters for OpenAI model configuration. + + Parameters: + frequency_penalty: Penalty for frequent tokens (-2.0 to 2.0). + presence_penalty: Penalty for new tokens (-2.0 to 2.0). + seed: Random seed for deterministic outputs. + temperature: Sampling temperature (0.0 to 2.0). + top_k: Top-k sampling parameter (currently ignored by OpenAI). + top_p: Top-p (nucleus) sampling parameter (0.0 to 1.0). + max_tokens: Maximum tokens in response (deprecated, use max_completion_tokens). + max_completion_tokens: Maximum completion tokens to generate. + extra: Additional model-specific parameters. + """ + frequency_penalty: Optional[float] = Field( default_factory=lambda: NOT_GIVEN, ge=-2.0, le=2.0 ) @@ -110,6 +135,19 @@ class BaseOpenAILLMService(LLMService): default_headers=None, **kwargs, ): + """Create an AsyncOpenAI client instance. + + Args: + api_key: OpenAI API key. + base_url: Custom base URL for the API. + organization: OpenAI organization ID. + project: OpenAI project ID. + default_headers: Additional HTTP headers. + **kwargs: Additional client configuration arguments. + + Returns: + Configured AsyncOpenAI client instance. + """ return AsyncOpenAI( api_key=api_key, base_url=base_url, @@ -124,11 +162,25 @@ class BaseOpenAILLMService(LLMService): ) def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as OpenAI service supports metrics generation. + """ return True async def get_chat_completions( self, context: OpenAILLMContext, messages: List[ChatCompletionMessageParam] ) -> AsyncStream[ChatCompletionChunk]: + """Get streaming chat completions from OpenAI API. + + Args: + context: The LLM context containing tools and configuration. + messages: List of chat completion messages to send. + + Returns: + Async stream of chat completion chunks. + """ params = { "model": self.model_name, "stream": True, @@ -274,6 +326,15 @@ class BaseOpenAILLMService(LLMService): await self.run_function_calls(function_calls) async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames for LLM completion requests. + + Handles OpenAILLMContextFrame, LLMMessagesFrame, VisionImageRawFrame, + and LLMUpdateSettingsFrame to trigger LLM completions and manage settings. + + Args: + frame: The frame to process. + direction: The direction of frame processing. + """ await super().process_frame(frame, direction) context = None diff --git a/src/pipecat/services/openai/llm.py b/src/pipecat/services/openai/llm.py index 44e0a40ce..c27eab867 100644 --- a/src/pipecat/services/openai/llm.py +++ b/src/pipecat/services/openai/llm.py @@ -4,6 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""OpenAI LLM service implementation with context aggregators.""" + import json from dataclasses import dataclass from typing import Any, Optional @@ -26,17 +28,46 @@ from pipecat.services.openai.base_llm import BaseOpenAILLMService @dataclass class OpenAIContextAggregatorPair: + """Pair of OpenAI context aggregators for user and assistant messages. + + Parameters: + _user: User context aggregator for processing user messages. + _assistant: Assistant context aggregator for processing assistant messages. + """ + _user: "OpenAIUserContextAggregator" _assistant: "OpenAIAssistantContextAggregator" def user(self) -> "OpenAIUserContextAggregator": + """Get the user context aggregator. + + Returns: + The user context aggregator instance. + """ return self._user def assistant(self) -> "OpenAIAssistantContextAggregator": + """Get the assistant context aggregator. + + Returns: + The assistant context aggregator instance. + """ return self._assistant class OpenAILLMService(BaseOpenAILLMService): + """OpenAI LLM service implementation. + + Provides a complete OpenAI LLM service with context aggregation support. + Uses the BaseOpenAILLMService for core functionality and adds OpenAI-specific + context aggregator creation. + + Args: + model: The OpenAI model name to use. Defaults to "gpt-4.1". + params: Input parameters for model configuration. + **kwargs: Additional arguments passed to the parent BaseOpenAILLMService. + """ + def __init__( self, *, @@ -53,14 +84,15 @@ class OpenAILLMService(BaseOpenAILLMService): user_params: LLMUserAggregatorParams = LLMUserAggregatorParams(), assistant_params: LLMAssistantAggregatorParams = LLMAssistantAggregatorParams(), ) -> OpenAIContextAggregatorPair: - """Create an instance of OpenAIContextAggregatorPair from an - OpenAILLMContext. Constructor keyword arguments for both the user and - assistant aggregators can be provided. + """Create OpenAI-specific context aggregators. + + Creates a pair of context aggregators optimized for OpenAI's message format, + including support for function calls, tool usage, and image handling. Args: - context (OpenAILLMContext): The LLM context. - user_params (LLMUserAggregatorParams, optional): User aggregator parameters. - assistant_params (LLMAssistantAggregatorParams, optional): User aggregator parameters. + context: The LLM context to create aggregators for. + user_params: Parameters for user message aggregation. + assistant_params: Parameters for assistant message aggregation. Returns: OpenAIContextAggregatorPair: A pair of context aggregators, one for @@ -75,11 +107,32 @@ class OpenAILLMService(BaseOpenAILLMService): class OpenAIUserContextAggregator(LLMUserContextAggregator): + """OpenAI-specific user context aggregator. + + Handles aggregation of user messages for OpenAI LLM services. + Inherits all functionality from the base LLMUserContextAggregator. + """ + pass class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): + """OpenAI-specific assistant context aggregator. + + Handles aggregation of assistant messages for OpenAI LLM services, + with specialized support for OpenAI's function calling format, + tool usage tracking, and image message handling. + """ + async def handle_function_call_in_progress(self, frame: FunctionCallInProgressFrame): + """Handle a function call in progress. + + Adds the function call to the context with an IN_PROGRESS status + to track ongoing function execution. + + Args: + frame: Frame containing function call progress information. + """ self._context.add_message( { "role": "assistant", @@ -104,6 +157,14 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): ) async def handle_function_call_result(self, frame: FunctionCallResultFrame): + """Handle the result of a function call. + + Updates the context with the function call result, replacing any + previous IN_PROGRESS status. + + Args: + frame: Frame containing the function call result. + """ if frame.result: result = json.dumps(frame.result) await self._update_function_call_result(frame.function_name, frame.tool_call_id, result) @@ -113,6 +174,13 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): ) async def handle_function_call_cancel(self, frame: FunctionCallCancelFrame): + """Handle a cancelled function call. + + Updates the context to mark the function call as cancelled. + + Args: + frame: Frame containing the function call cancellation information. + """ await self._update_function_call_result( frame.function_name, frame.tool_call_id, "CANCELLED" ) @@ -129,6 +197,14 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator): message["content"] = result async def handle_user_image_frame(self, frame: UserImageRawFrame): + """Handle a user image frame from a function call request. + + Marks the associated function call as completed and adds the image + to the context for processing. + + Args: + frame: Frame containing the user image and request context. + """ await self._update_function_call_result( frame.request.function_name, frame.request.tool_call_id, "COMPLETED" )