mirror of
https://github.com/BerriAI/litellm.git
synced 2025-04-25 02:34:29 +00:00
1105e35538
* fix(o_series_transformation.py): add 'reasoning_effort' as o series model param Closes https://github.com/BerriAI/litellm/issues/8182 * fix(main.py): ensure `reasoning_effort` is a mapped openai param * refactor(azure/): rename o1_[x] files to o_series_[x] * refactor(base_llm_unit_tests.py): refactor testing for o series reasoning effort * test(test_azure_o_series.py): have azure o series tests correctly inherit from base o series model tests * feat(base_utils.py): support translating 'developer' role to 'system' role for non-openai providers Makes it easy to switch from openai to anthropic * fix: fix linting errors * fix(base_llm_unit_tests.py): fix test * fix(main.py): add missing param
127 lines
3.8 KiB
Python
127 lines
3.8 KiB
Python
"""
|
|
Utility functions for base LLM classes.
|
|
"""
|
|
|
|
import copy
|
|
from abc import ABC, abstractmethod
|
|
from typing import List, Optional, Type, Union
|
|
|
|
from openai.lib import _parsing, _pydantic
|
|
from pydantic import BaseModel
|
|
|
|
from litellm.types.llms.openai import AllMessageValues
|
|
from litellm.types.utils import ProviderSpecificModelInfo
|
|
|
|
|
|
class BaseLLMModelInfo(ABC):
|
|
def get_provider_info(
|
|
self,
|
|
model: str,
|
|
) -> Optional[ProviderSpecificModelInfo]:
|
|
return None
|
|
|
|
@abstractmethod
|
|
def get_models(self) -> List[str]:
|
|
pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def get_api_key(api_key: Optional[str] = None) -> Optional[str]:
|
|
pass
|
|
|
|
@staticmethod
|
|
@abstractmethod
|
|
def get_api_base(api_base: Optional[str] = None) -> Optional[str]:
|
|
pass
|
|
|
|
|
|
def _dict_to_response_format_helper(
|
|
response_format: dict, ref_template: Optional[str] = None
|
|
) -> dict:
|
|
if ref_template is not None and response_format.get("type") == "json_schema":
|
|
# Deep copy to avoid modifying original
|
|
modified_format = copy.deepcopy(response_format)
|
|
schema = modified_format["json_schema"]["schema"]
|
|
|
|
# Update all $ref values in the schema
|
|
def update_refs(schema):
|
|
stack = [(schema, [])]
|
|
visited = set()
|
|
|
|
while stack:
|
|
obj, path = stack.pop()
|
|
obj_id = id(obj)
|
|
|
|
if obj_id in visited:
|
|
continue
|
|
visited.add(obj_id)
|
|
|
|
if isinstance(obj, dict):
|
|
if "$ref" in obj:
|
|
ref_path = obj["$ref"]
|
|
model_name = ref_path.split("/")[-1]
|
|
obj["$ref"] = ref_template.format(model=model_name)
|
|
|
|
for k, v in obj.items():
|
|
if isinstance(v, (dict, list)):
|
|
stack.append((v, path + [k]))
|
|
|
|
elif isinstance(obj, list):
|
|
for i, item in enumerate(obj):
|
|
if isinstance(item, (dict, list)):
|
|
stack.append((item, path + [i]))
|
|
|
|
update_refs(schema)
|
|
return modified_format
|
|
return response_format
|
|
|
|
|
|
def type_to_response_format_param(
|
|
response_format: Optional[Union[Type[BaseModel], dict]],
|
|
ref_template: Optional[str] = None,
|
|
) -> Optional[dict]:
|
|
"""
|
|
Re-implementation of openai's 'type_to_response_format_param' function
|
|
|
|
Used for converting pydantic object to api schema.
|
|
"""
|
|
if response_format is None:
|
|
return None
|
|
|
|
if isinstance(response_format, dict):
|
|
return _dict_to_response_format_helper(response_format, ref_template)
|
|
|
|
# type checkers don't narrow the negation of a `TypeGuard` as it isn't
|
|
# a safe default behaviour but we know that at this point the `response_format`
|
|
# can only be a `type`
|
|
if not _parsing._completions.is_basemodel_type(response_format):
|
|
raise TypeError(f"Unsupported response_format type - {response_format}")
|
|
|
|
if ref_template is not None:
|
|
schema = response_format.model_json_schema(ref_template=ref_template)
|
|
else:
|
|
schema = _pydantic.to_strict_json_schema(response_format)
|
|
|
|
return {
|
|
"type": "json_schema",
|
|
"json_schema": {
|
|
"schema": schema,
|
|
"name": response_format.__name__,
|
|
"strict": True,
|
|
},
|
|
}
|
|
|
|
|
|
def map_developer_role_to_system_role(
|
|
messages: List[AllMessageValues],
|
|
) -> List[AllMessageValues]:
|
|
"""
|
|
Translate `developer` role to `system` role for non-OpenAI providers.
|
|
"""
|
|
new_messages: List[AllMessageValues] = []
|
|
for m in messages:
|
|
if m["role"] == "developer":
|
|
new_messages.append({"role": "system", "content": m["content"]})
|
|
else:
|
|
new_messages.append(m)
|
|
return new_messages
|