from os import PathLike from typing import IO, Any, Iterable, List, Literal, Mapping, Optional, Tuple, Union from openai._legacy_response import ( HttpxBinaryResponseContent as _HttpxBinaryResponseContent, ) from openai.lib.streaming._assistants import ( AssistantEventHandler, AssistantStreamManager, AsyncAssistantEventHandler, AsyncAssistantStreamManager, ) from openai.pagination import AsyncCursorPage, SyncCursorPage from openai.types import Batch, EmbeddingCreateParams, FileObject from openai.types.beta.assistant import Assistant from openai.types.beta.assistant_tool_param import AssistantToolParam from openai.types.beta.thread_create_params import ( Message as OpenAICreateThreadParamsMessage, ) from openai.types.beta.threads.message import Message as OpenAIMessage from openai.types.beta.threads.message_content import MessageContent from openai.types.beta.threads.run import Run from openai.types.chat import ChatCompletionChunk from openai.types.chat.chat_completion_audio_param import ChatCompletionAudioParam from openai.types.chat.chat_completion_content_part_input_audio_param import ( ChatCompletionContentPartInputAudioParam, ) from openai.types.chat.chat_completion_modality import ChatCompletionModality from openai.types.chat.chat_completion_prediction_content_param import ( ChatCompletionPredictionContentParam, ) from openai.types.embedding import Embedding as OpenAIEmbedding from openai.types.fine_tuning.fine_tuning_job import FineTuningJob from pydantic import BaseModel, Field from typing_extensions import Dict, Required, TypedDict, override FileContent = Union[IO[bytes], bytes, PathLike] FileTypes = Union[ # file (or bytes) FileContent, # (filename, file (or bytes)) Tuple[Optional[str], FileContent], # (filename, file (or bytes), content_type) Tuple[Optional[str], FileContent, Optional[str]], # (filename, file (or bytes), content_type, headers) Tuple[Optional[str], FileContent, Optional[str], Mapping[str, str]], ] EmbeddingInput = Union[str, List[str]] class HttpxBinaryResponseContent(_HttpxBinaryResponseContent): _hidden_params: dict = {} pass class NotGiven: """ A sentinel singleton class used to distinguish omitted keyword arguments from those passed in with the value None (which may have different behavior). For example: ```py def get(timeout: Union[int, NotGiven, None] = NotGiven()) -> Response: ... get(timeout=1) # 1s timeout get(timeout=None) # No timeout get() # Default timeout behavior, which may not be statically known at the method definition. ``` """ def __bool__(self) -> Literal[False]: return False @override def __repr__(self) -> str: return "NOT_GIVEN" NOT_GIVEN = NotGiven() class ToolResourcesCodeInterpreter(TypedDict, total=False): file_ids: List[str] """ A list of [file](https://platform.openai.com/docs/api-reference/files) IDs made available to the `code_interpreter` tool. There can be a maximum of 20 files associated with the tool. """ class ToolResourcesFileSearchVectorStore(TypedDict, total=False): file_ids: List[str] """ A list of [file](https://platform.openai.com/docs/api-reference/files) IDs to add to the vector store. There can be a maximum of 10000 files in a vector store. """ metadata: object """Set of 16 key-value pairs that can be attached to a vector store. This can be useful for storing additional information about the vector store in a structured format. Keys can be a maximum of 64 characters long and values can be a maxium of 512 characters long. """ class ToolResourcesFileSearch(TypedDict, total=False): vector_store_ids: List[str] """ The [vector store](https://platform.openai.com/docs/api-reference/vector-stores/object) attached to this thread. There can be a maximum of 1 vector store attached to the thread. """ vector_stores: Iterable[ToolResourcesFileSearchVectorStore] """ A helper to create a [vector store](https://platform.openai.com/docs/api-reference/vector-stores/object) with file_ids and attach it to this thread. There can be a maximum of 1 vector store attached to the thread. """ class OpenAICreateThreadParamsToolResources(TypedDict, total=False): code_interpreter: ToolResourcesCodeInterpreter file_search: ToolResourcesFileSearch class FileSearchToolParam(TypedDict, total=False): type: Required[Literal["file_search"]] """The type of tool being defined: `file_search`""" class CodeInterpreterToolParam(TypedDict, total=False): type: Required[Literal["code_interpreter"]] """The type of tool being defined: `code_interpreter`""" AttachmentTool = Union[CodeInterpreterToolParam, FileSearchToolParam] class Attachment(TypedDict, total=False): file_id: str """The ID of the file to attach to the message.""" tools: Iterable[AttachmentTool] """The tools to add this file to.""" class ImageFileObject(TypedDict): file_id: Required[str] detail: Optional[str] class ImageURLObject(TypedDict): url: Required[str] detail: Optional[str] class MessageContentTextObject(TypedDict): type: Required[Literal["text"]] text: str class MessageContentImageFileObject(TypedDict): type: Literal["image_file"] image_file: ImageFileObject class MessageContentImageURLObject(TypedDict): type: Required[str] image_url: ImageURLObject class MessageData(TypedDict): role: Literal["user", "assistant"] content: Union[ str, List[ Union[ MessageContentTextObject, MessageContentImageFileObject, MessageContentImageURLObject, ] ], ] attachments: Optional[List[Attachment]] metadata: Optional[dict] class Thread(BaseModel): id: str """The identifier, which can be referenced in API endpoints.""" created_at: int """The Unix timestamp (in seconds) for when the thread was created.""" metadata: Optional[object] = None """Set of 16 key-value pairs that can be attached to an object. This can be useful for storing additional information about the object in a structured format. Keys can be a maximum of 64 characters long and values can be a maxium of 512 characters long. """ object: Literal["thread"] """The object type, which is always `thread`.""" # OpenAI Files Types class CreateFileRequest(TypedDict, total=False): """ CreateFileRequest Used by Assistants API, Batches API, and Fine-Tunes API Required Params: file: FileTypes purpose: Literal['assistants', 'batch', 'fine-tune'] Optional Params: extra_headers: Optional[Dict[str, str]] extra_body: Optional[Dict[str, str]] = None timeout: Optional[float] = None """ file: FileTypes purpose: Literal["assistants", "batch", "fine-tune"] extra_headers: Optional[Dict[str, str]] extra_body: Optional[Dict[str, str]] timeout: Optional[float] class FileContentRequest(TypedDict, total=False): """ FileContentRequest Used by Assistants API, Batches API, and Fine-Tunes API Required Params: file_id: str Optional Params: extra_headers: Optional[Dict[str, str]] extra_body: Optional[Dict[str, str]] = None timeout: Optional[float] = None """ file_id: str extra_headers: Optional[Dict[str, str]] extra_body: Optional[Dict[str, str]] timeout: Optional[float] # OpenAI Batches Types class CreateBatchRequest(TypedDict, total=False): """ CreateBatchRequest """ completion_window: Literal["24h"] endpoint: Literal["/v1/chat/completions", "/v1/embeddings", "/v1/completions"] input_file_id: str metadata: Optional[Dict[str, str]] extra_headers: Optional[Dict[str, str]] extra_body: Optional[Dict[str, str]] timeout: Optional[float] class RetrieveBatchRequest(TypedDict, total=False): """ RetrieveBatchRequest """ batch_id: str extra_headers: Optional[Dict[str, str]] extra_body: Optional[Dict[str, str]] timeout: Optional[float] class CancelBatchRequest(TypedDict, total=False): """ CancelBatchRequest """ batch_id: str extra_headers: Optional[Dict[str, str]] extra_body: Optional[Dict[str, str]] timeout: Optional[float] class ListBatchRequest(TypedDict, total=False): """ ListBatchRequest - List your organization's batches Calls https://api.openai.com/v1/batches """ after: Union[str, NotGiven] limit: Union[int, NotGiven] extra_headers: Optional[Dict[str, str]] extra_body: Optional[Dict[str, str]] timeout: Optional[float] BatchJobStatus = Literal[ "validating", "failed", "in_progress", "finalizing", "completed", "expired", "cancelling", "cancelled", ] class ChatCompletionAudioDelta(TypedDict, total=False): data: str transcript: str expires_at: int id: str class ChatCompletionToolCallFunctionChunk(TypedDict, total=False): name: Optional[str] arguments: str class ChatCompletionAssistantToolCall(TypedDict): id: Optional[str] type: Literal["function"] function: ChatCompletionToolCallFunctionChunk class ChatCompletionToolCallChunk(TypedDict): # result of /chat/completions call id: Optional[str] type: Literal["function"] function: ChatCompletionToolCallFunctionChunk index: int class ChatCompletionDeltaToolCallChunk(TypedDict, total=False): id: str type: Literal["function"] function: ChatCompletionToolCallFunctionChunk index: int class ChatCompletionCachedContent(TypedDict): type: Literal["ephemeral"] class OpenAIChatCompletionTextObject(TypedDict): type: Literal["text"] text: str class ChatCompletionTextObject( OpenAIChatCompletionTextObject, total=False ): # litellm wrapper on top of openai object for handling cached content cache_control: ChatCompletionCachedContent class ChatCompletionImageUrlObject(TypedDict, total=False): url: Required[str] detail: str class ChatCompletionImageObject(TypedDict): type: Literal["image_url"] image_url: Union[str, ChatCompletionImageUrlObject] class ChatCompletionAudioObject(ChatCompletionContentPartInputAudioParam): pass OpenAIMessageContentListBlock = Union[ ChatCompletionTextObject, ChatCompletionImageObject, ChatCompletionAudioObject, ] OpenAIMessageContent = Union[ str, Iterable[OpenAIMessageContentListBlock], ] # The prompt(s) to generate completions for, encoded as a string, array of strings, array of tokens, or array of token arrays. AllPromptValues = Union[str, List[str], Iterable[int], Iterable[Iterable[int]], None] class OpenAIChatCompletionUserMessage(TypedDict): role: Literal["user"] content: OpenAIMessageContent class OpenAITextCompletionUserMessage(TypedDict): role: Literal["user"] content: AllPromptValues class ChatCompletionUserMessage(OpenAIChatCompletionUserMessage, total=False): cache_control: ChatCompletionCachedContent class OpenAIChatCompletionAssistantMessage(TypedDict, total=False): role: Required[Literal["assistant"]] content: Optional[Union[str, Iterable[ChatCompletionTextObject]]] name: Optional[str] tool_calls: Optional[List[ChatCompletionAssistantToolCall]] function_call: Optional[ChatCompletionToolCallFunctionChunk] class ChatCompletionAssistantMessage(OpenAIChatCompletionAssistantMessage, total=False): cache_control: ChatCompletionCachedContent class ChatCompletionToolMessage(TypedDict): role: Literal["tool"] content: Union[str, Iterable[ChatCompletionTextObject]] tool_call_id: str class ChatCompletionFunctionMessage(TypedDict): role: Literal["function"] content: Optional[Union[str, Iterable[ChatCompletionTextObject]]] name: str tool_call_id: Optional[str] class OpenAIChatCompletionSystemMessage(TypedDict, total=False): role: Required[Literal["system"]] content: Required[Union[str, List]] name: str class ChatCompletionSystemMessage(OpenAIChatCompletionSystemMessage, total=False): cache_control: ChatCompletionCachedContent ValidUserMessageContentTypes = [ "text", "image_url", "input_audio", ] # used for validating user messages. Prevent users from accidentally sending anthropic messages. AllMessageValues = Union[ ChatCompletionUserMessage, ChatCompletionAssistantMessage, ChatCompletionToolMessage, ChatCompletionSystemMessage, ChatCompletionFunctionMessage, ] class ChatCompletionToolChoiceFunctionParam(TypedDict): name: str class ChatCompletionToolChoiceObjectParam(TypedDict): type: Literal["function"] function: ChatCompletionToolChoiceFunctionParam ChatCompletionToolChoiceStringValues = Literal["none", "auto", "required"] ChatCompletionToolChoiceValues = Union[ ChatCompletionToolChoiceStringValues, ChatCompletionToolChoiceObjectParam ] class ChatCompletionToolParamFunctionChunk(TypedDict, total=False): name: Required[str] description: str parameters: dict class OpenAIChatCompletionToolParam(TypedDict): type: Union[Literal["function"], str] function: ChatCompletionToolParamFunctionChunk class ChatCompletionToolParam(OpenAIChatCompletionToolParam, total=False): cache_control: ChatCompletionCachedContent class Function(TypedDict, total=False): name: Required[str] """The name of the function to call.""" class ChatCompletionNamedToolChoiceParam(TypedDict, total=False): function: Required[Function] type: Required[Literal["function"]] """The type of the tool. Currently, only `function` is supported.""" class ChatCompletionRequest(TypedDict, total=False): model: Required[str] messages: Required[List[AllMessageValues]] frequency_penalty: float logit_bias: dict logprobs: bool top_logprobs: int max_tokens: int n: int presence_penalty: float response_format: dict seed: int service_tier: str stop: Union[str, List[str]] stream_options: dict temperature: float top_p: float tools: List[ChatCompletionToolParam] tool_choice: ChatCompletionToolChoiceValues parallel_tool_calls: bool function_call: Union[str, dict] functions: List user: str metadata: dict # litellm specific param class ChatCompletionDeltaChunk(TypedDict, total=False): content: Optional[str] tool_calls: List[ChatCompletionDeltaToolCallChunk] role: str ChatCompletionAssistantContentValue = ( str # keep as var, used in stream_chunk_builder as well ) class ChatCompletionResponseMessage(TypedDict, total=False): content: Optional[ChatCompletionAssistantContentValue] tool_calls: Optional[List[ChatCompletionToolCallChunk]] role: Literal["assistant"] function_call: Optional[ChatCompletionToolCallFunctionChunk] class ChatCompletionUsageBlock(TypedDict): prompt_tokens: int completion_tokens: int total_tokens: int class OpenAIChatCompletionChunk(ChatCompletionChunk): def __init__(self, **kwargs): # Set the 'object' kwarg to 'chat.completion.chunk' kwargs["object"] = "chat.completion.chunk" super().__init__(**kwargs) class Hyperparameters(BaseModel): batch_size: Optional[Union[str, int]] = None # "Number of examples in each batch." learning_rate_multiplier: Optional[Union[str, float]] = ( None # Scaling factor for the learning rate ) n_epochs: Optional[Union[str, int]] = ( None # "The number of epochs to train the model for" ) class FineTuningJobCreate(BaseModel): """ FineTuningJobCreate - Create a fine-tuning job Example Request ``` { "model": "gpt-3.5-turbo", "training_file": "file-abc123", "hyperparameters": { "batch_size": "auto", "learning_rate_multiplier": 0.1, "n_epochs": 3 }, "suffix": "custom-model-name", "validation_file": "file-xyz789", "integrations": ["slack"], "seed": 42 } ``` """ model: str # "The name of the model to fine-tune." training_file: str # "The ID of an uploaded file that contains training data." hyperparameters: Optional[Hyperparameters] = ( None # "The hyperparameters used for the fine-tuning job." ) suffix: Optional[str] = ( None # "A string of up to 18 characters that will be added to your fine-tuned model name." ) validation_file: Optional[str] = ( None # "The ID of an uploaded file that contains validation data." ) integrations: Optional[List[str]] = ( None # "A list of integrations to enable for your fine-tuning job." ) seed: Optional[int] = None # "The seed controls the reproducibility of the job." class LiteLLMFineTuningJobCreate(FineTuningJobCreate): custom_llm_provider: Literal["openai", "azure", "vertex_ai"] class Config: extra = "allow" # This allows the model to accept additional fields AllEmbeddingInputValues = Union[str, List[str], List[int], List[List[int]]] OpenAIAudioTranscriptionOptionalParams = Literal[ "language", "prompt", "temperature", "response_format", "timestamp_granularities" ]