wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add OpenAI LLM docstrings

Browse Source
This commit is contained in:
Mark Backman
2025-06-26 00:06:57 -04:00
parent 5b8f1fe3e3
commit ac61139243
2 changed files with 148 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
src/pipecat/services/openai/base_llm.py
View File

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

88
src/pipecat/services/openai/llm.py
View File

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