refactor(native): move the chat provider helpers out of main.py
Some checks failed
LiteLLM Rust / rust-lint (push) Has been cancelled
LiteLLM Rust / rust-test (push) Has been cancelled
LiteLLM Rust / rust-wheel (push) Has been cancelled

This commit is contained in:
narmstrong-berri 2026-10-16 20:01:06 +00:00
parent 3568674359
commit 557413fa96
32 changed files with 4465 additions and 4121 deletions

View file

@ -42,13 +42,13 @@ from litellm.llms.black_forest_labs.image_generation.handler import bfl_image_ge
from litellm.llms.openai.workload_identity import resolve_openai_workload_identity_config
from litellm.main import (
azure_chat_completions,
base_llm_aiohttp_handler,
base_llm_http_handler,
bedrock_image_edit,
bedrock_image_generation,
openai_chat_completions,
openai_image_variations,
)
from litellm.native.chat_providers.handlers import base_llm_aiohttp_handler
###########################################
from litellm.secret_managers.main import get_secret_str

View file

@ -1,7 +1,7 @@
import asyncio
import json
import time
from collections.abc import Callable, Coroutine
from collections.abc import Callable, Coroutine, Mapping
from typing import Final
import httpx
@ -115,7 +115,7 @@ class AzureOpenAIAssistantsAPIConfig:
def check_dynamic_azure_params(
azure_client_params: dict,
azure_client_params: Mapping[str, object],
azure_client: AzureOpenAI | AsyncAzureOpenAI | None,
) -> bool:
"""

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,273 @@
from collections.abc import Mapping, Sequence
from typing import Final, cast # noqa: TID251 # casts moved verbatim from litellm/main.py
from pydantic import TypeAdapter
import litellm
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.native.chat_providers.handlers import base_llm_http_handler
from litellm.native.chat_providers.support import dispatch_client_http, get_encoding
from litellm.secret_managers.main import get_secret_str
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.types.llms.openai import AllMessageValues
def complete_ragflow(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
try:
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
except Exception as e:
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": headers},
)
raise e
return response
def complete_a2a(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
(
api_base,
api_key,
headers,
) = litellm.A2AConfig.resolve_agent_config_from_registry(
agent_name=model,
api_base=api_base,
api_key=api_key,
headers=headers,
optional_params=optional_params,
)
# Fall back to environment variables and defaults
api_base = api_base or litellm.api_base or get_secret_str("A2A_API_BASE")
if api_base is None:
raise Exception(
"api_base is required for A2A provider. "
"Either provide api_base parameter, set A2A_API_BASE environment variable, "
"or register the agent in the proxy with model='a2a/<agent-name>'."
)
headers = headers or litellm.headers
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider=custom_llm_provider,
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
client=client,
provider_config=provider_config,
)
def complete_langgraph(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
from litellm.llms.langgraph.chat.transformation import LangGraphConfig
(
api_base,
api_key,
) = LangGraphConfig().get_openai_compatible_provider_info(
api_base=api_base or litellm.api_base,
api_key=api_key or litellm.api_key,
)
headers = headers or litellm.headers
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider=custom_llm_provider,
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
client=client,
)
def complete_microsoft_365_copilot(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
from litellm.llms.microsoft_365_copilot.chat.handler import acompletion, completion
from litellm.llms.microsoft_365_copilot.common_utils import as_secret_fields
messages: Final[Sequence[AllMessageValues]] = cast( # cast-ok: dispatch messages are prevalidated
Sequence[AllMessageValues],
ctx.messages,
)
optional_params_value: Final[object] = cast( # cast-ok: dispatch params are validated below
object,
ctx.optional_params,
)
optional_params: Final = TypeAdapter(Mapping[str, object]).validate_python(optional_params_value)
kwargs_value: Final[object] = cast( # cast-ok: dispatch kwargs are validated below
object,
ctx.kwargs,
)
kwargs: Final = TypeAdapter(Mapping[str, object]).validate_python(kwargs_value)
secret_fields: Final = as_secret_fields(kwargs.get("secret_fields"))
client: Final = dispatch_client_http(ctx)
if ctx.acompletion:
async_client: Final = client if isinstance(client, AsyncHTTPHandler) else None
return acompletion(
model=ctx.model,
messages=messages,
api_key=ctx.api_key,
litellm_params=ctx.litellm_params,
optional_params=optional_params,
secret_fields=secret_fields,
stream=ctx.stream is True,
timeout=ctx.timeout,
logging_obj=ctx.logging,
client=async_client,
shared_session=ctx.shared_session,
)
sync_client: Final = client if isinstance(client, HTTPHandler) else None
return completion(
model=ctx.model,
messages=messages,
api_key=ctx.api_key,
litellm_params=ctx.litellm_params,
optional_params=optional_params,
secret_fields=secret_fields,
stream=ctx.stream is True,
timeout=ctx.timeout,
logging_obj=ctx.logging,
client=sync_client,
)
def complete_langflow(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
from litellm.llms.langflow.chat.transformation import LangFlowConfig
(
api_base,
api_key,
) = LangFlowConfig().get_openai_compatible_provider_info(
api_base=api_base or litellm.api_base,
api_key=api_key or litellm.api_key,
)
headers = headers or litellm.headers
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider=custom_llm_provider,
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
client=client,
)

View file

@ -0,0 +1,131 @@
import os
from typing import Final, cast # noqa: TID251 # casts moved verbatim from litellm/main.py
import litellm
from litellm._logging import verbose_logger
from litellm.native.chat_providers.handlers import anthropic_chat_completions, base_llm_http_handler
from litellm.native.chat_providers.support import dispatch_client_http, get_encoding, print_verbose
from litellm.secret_managers.main import get_secret_bool
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.utils import get_secret
def complete_anthropic_text(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
custom_prompt_dict = ctx.custom_prompt_dict
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or litellm.anthropic_key or litellm.api_key or os.environ.get("ANTHROPIC_API_KEY")
custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
api_base = cast(
str | None,
api_base
or litellm.api_base
or get_secret("ANTHROPIC_API_BASE")
or get_secret("ANTHROPIC_BASE_URL")
or "https://api.anthropic.com/v1/complete",
)
# Check if we should disable automatic URL suffix appending
disable_url_suffix: Final = get_secret_bool("LITELLM_ANTHROPIC_DISABLE_URL_SUFFIX")
if api_base is not None and not disable_url_suffix and not api_base.endswith("/v1/complete"):
api_base += "/v1/complete"
elif disable_url_suffix:
verbose_logger.debug("LITELLM_ANTHROPIC_DISABLE_URL_SUFFIX is set, skipping /v1/complete suffix")
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="anthropic_text",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
)
def complete_anthropic(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
custom_prompt_dict = ctx.custom_prompt_dict
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
timeout: Final = ctx.timeout
api_key = api_key or litellm.anthropic_key or litellm.api_key or os.environ.get("ANTHROPIC_API_KEY")
custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
# call /messages
# default route for all anthropic models
api_base = cast(
str | None,
api_base
or litellm.api_base
or get_secret("ANTHROPIC_API_BASE")
or get_secret("ANTHROPIC_BASE_URL")
or "https://api.anthropic.com/v1/messages",
)
# Check if we should disable automatic URL suffix appending
disable_url_suffix: Final = get_secret_bool("LITELLM_ANTHROPIC_DISABLE_URL_SUFFIX")
if api_base is not None and not disable_url_suffix and not api_base.endswith("/v1/messages"):
api_base += "/v1/messages"
elif disable_url_suffix:
verbose_logger.debug("LITELLM_ANTHROPIC_DISABLE_URL_SUFFIX is set, skipping /v1/messages suffix")
response: Final = anthropic_chat_completions.completion(
model=model,
messages=messages,
api_base=api_base,
acompletion=acompletion,
custom_prompt_dict=litellm.custom_prompt_dict,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(), # for calculating input/output tokens
api_key=api_key,
logging_obj=logging,
headers=headers,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
)
if optional_params.get("stream", False) or acompletion is True:
## LOGGING
logging.post_call(
input=messages,
api_key=api_key,
original_response=response,
)
return response

View file

@ -0,0 +1,393 @@
from collections.abc import Mapping
from functools import partial
from typing import Final
import httpx
from pydantic import TypeAdapter
import litellm
from litellm._logging import verbose_logger
from litellm.llms.bedrock.chat.converse_transformation import AmazonConverseConfig
from litellm.llms.bedrock.chat.invoke_transformations.anthropic_claude3_transformation import (
AmazonAnthropicClaudeConfig,
)
from litellm.llms.bedrock.chat.native_output_fallback import NativeOutputAttemptResult, call_with_native_output_fallback
from litellm.llms.bedrock.common_utils import (
BedrockModelInfo,
NativeOutputFallback,
bedrock_route_for_request,
deployment_model_info,
without_bedrock_native_output_opt_in,
without_bedrock_route_prefix,
)
from litellm.llms.bedrock_mantle.chat.claude_transformation import bedrock_mantle_chat_config
from litellm.native.chat_providers.handlers import (
base_llm_http_handler,
bedrock_converse_chat_completion,
openai_like_chat_completion,
sagemaker_llm,
)
from litellm.native.chat_providers.support import (
dispatch_client_http,
get_encoding,
print_verbose,
provider_config_items,
)
from litellm.secret_managers.main import get_secret_str
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager, get_secret
_ADDITIONAL_DROP_PARAMS_ADAPTER: Final = TypeAdapter(list[str])
_OPTIONAL_PARAMS_ADAPTER: Final = TypeAdapter(dict[str, object])
def complete_bedrock_mantle(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = api_base or litellm.api_base or get_secret("BEDROCK_MANTLE_API_BASE")
api_key = api_key or litellm.api_key or get_secret("BEDROCK_MANTLE_API_KEY")
headers = headers or litellm.headers
config: Final = bedrock_mantle_chat_config(model).get_config()
for k, v in provider_config_items(config):
if k not in optional_params:
optional_params[k] = v
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider=custom_llm_provider,
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
client=client,
)
def complete_amazon_nova(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
api_base = ctx.api_base
api_key = ctx.api_key
custom_llm_provider: Final = ctx.custom_llm_provider
custom_prompt_dict: Final = ctx.custom_prompt_dict
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
timeout: Final = ctx.timeout
api_key = api_key or litellm.amazon_nova_api_key or get_secret_str("AMAZON_NOVA_API_KEY") or litellm.api_key
api_base = (
api_base or litellm.api_base or get_secret_str("AMAZON_NOVA_API_BASE") or "https://api.nova.amazon.com/v1"
)
return openai_like_chat_completion.completion(
model=model,
messages=messages,
api_base=api_base,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
custom_prompt_dict=custom_prompt_dict,
)
def complete_sagemaker_chat(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
custom_llm_provider=custom_llm_provider,
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
client=client,
)
def complete_sagemaker(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
custom_prompt_dict: Final = ctx.custom_prompt_dict
hf_model_name: Final = ctx.hf_model_name
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
return sagemaker_llm.completion(
model=model,
messages=messages,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
custom_prompt_dict=custom_prompt_dict,
hf_model_name=hf_model_name,
logger_fn=logger_fn,
encoding=get_encoding(),
logging_obj=logging,
acompletion=acompletion,
)
_AWS_BEDROCK_CLIENT_CREDENTIAL_KEYS: Final = (
"aws_access_key_id",
"aws_secret_access_key",
"aws_session_token",
"aws_region_name",
)
def _native_output_fallback(
ctx: CompletionDispatchContext,
first_attempt_params: Mapping[str, object],
first_attempt_litellm_params: Mapping[str, object],
) -> NativeOutputFallback:
litellm_params: Final = without_bedrock_native_output_opt_in(first_attempt_litellm_params)
remapped_params: Final = ctx.optional_params_for_model_info(deployment_model_info(litellm_params))
return NativeOutputFallback(
optional_params={
**{key: value for key, value in remapped_params.items() if key != "aws_bedrock_client"},
**{
key: first_attempt_params[key]
for key in _AWS_BEDROCK_CLIENT_CREDENTIAL_KEYS
if key in first_attempt_params
},
},
litellm_params=litellm_params,
)
def _converse_native_output_fallback(
ctx: CompletionDispatchContext,
first_attempt_params: Mapping[str, object],
first_attempt_litellm_params: Mapping[str, object],
) -> NativeOutputFallback | None:
fallback: Final = _native_output_fallback(ctx, first_attempt_params, first_attempt_litellm_params)
return None if AmazonConverseConfig.sends_native_text_format(fallback.optional_params) else fallback
def _native_output_attempt_timeout(
ctx: CompletionDispatchContext, fallback: NativeOutputFallback | None
) -> float | str | httpx.Timeout | None:
if fallback is None:
return ctx.timeout
retry_timeout: Final = ctx.timeout_for_optional_params(fallback.optional_params)
ctx.logging.record_applied_timeout(retry_timeout)
return retry_timeout
def complete_bedrock(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_prompt_dict = ctx.custom_prompt_dict
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
if "aws_bedrock_client" in optional_params:
verbose_logger.warning(
"'aws_bedrock_client' is a deprecated param. Please move to another auth method - https://docs.litellm.ai/docs/providers/bedrock#boto3---authentication."
)
# Extract credentials for legacy boto3 client and pass thru to httpx
aws_bedrock_client: Final = optional_params.pop("aws_bedrock_client")
creds: Final = aws_bedrock_client._get_credentials().get_frozen_credentials()
if creds.access_key:
optional_params["aws_access_key_id"] = creds.access_key
if creds.secret_key:
optional_params["aws_secret_access_key"] = creds.secret_key
if creds.token:
optional_params["aws_session_token"] = creds.token
if "aws_region_name" not in optional_params or optional_params["aws_region_name"] is None:
optional_params["aws_region_name"] = aws_bedrock_client.meta.region_name
additional_drop_params: Final = (
_ADDITIONAL_DROP_PARAMS_ADAPTER.validate_python(ctx.kwargs["additional_drop_params"])
if ctx.kwargs.get("additional_drop_params") is not None
else None
)
bedrock_route: Final = bedrock_route_for_request(model, ctx.request_params, additional_drop_params)
if bedrock_route == "claude_platform":
provider_config = ProviderConfigManager.get_provider_chat_config(
model=model,
provider=LlmProviders.BEDROCK,
)
model = BedrockModelInfo.get_claude_platform_model(model)
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="bedrock",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
client=client,
provider_config=provider_config,
)
elif bedrock_route == "converse":
model = without_bedrock_route_prefix(model)
converse_model: Final = model
def converse_attempt(fallback: NativeOutputFallback | None) -> NativeOutputAttemptResult:
return bedrock_converse_chat_completion.completion(
model=converse_model,
messages=messages,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
optional_params=optional_params if fallback is None else dict(fallback.optional_params),
litellm_params=litellm_params if fallback is None else dict(fallback.litellm_params),
logger_fn=logger_fn,
encoding=get_encoding(),
logging_obj=logging,
extra_headers=headers, # Use merged headers instead of original extra_headers
timeout=_native_output_attempt_timeout(ctx, fallback),
acompletion=acompletion,
client=client,
api_base=api_base,
api_key=api_key,
)
converse_params: Final = _OPTIONAL_PARAMS_ADAPTER.validate_python(optional_params)
response = call_with_native_output_fallback(
converse_attempt,
(
partial(_converse_native_output_fallback, ctx, converse_params, dict(litellm_params))
if AmazonConverseConfig().uses_opt_in_native_output(converse_model, converse_params, litellm_params)
else None
),
converse_model,
)
elif bedrock_route == "converse_like":
model = model.replace("converse_like/", "")
response = base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
custom_llm_provider="bedrock",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
client=client,
)
else:
def invoke_attempt(fallback: NativeOutputFallback | None) -> NativeOutputAttemptResult:
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params if fallback is None else dict(fallback.optional_params),
litellm_params=litellm_params if fallback is None else dict(fallback.litellm_params),
custom_llm_provider="bedrock",
timeout=_native_output_attempt_timeout(ctx, fallback),
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
client=client,
)
invoke_params: Final = _OPTIONAL_PARAMS_ADAPTER.validate_python(optional_params)
response = call_with_native_output_fallback(
invoke_attempt,
(
partial(_native_output_fallback, ctx, invoke_params, dict(litellm_params))
if isinstance(provider_config, AmazonAnthropicClaudeConfig)
and provider_config.uses_opt_in_native_output(model, invoke_params, litellm_params)
else None
),
model,
)
return response

View file

@ -0,0 +1,406 @@
from typing import Final, cast # noqa: TID251 # casts moved verbatim from litellm/main.py
import openai
import litellm
from litellm.litellm_core_utils.prompt_templates.factory import stringify_json_tool_call_content
from litellm.llms.azure.azure import check_dynamic_azure_params
from litellm.native.chat_providers.handlers import (
azure_anthropic_chat_completions,
azure_chat_completions,
azure_o1_chat_completions,
azure_text_completions,
base_llm_http_handler,
)
from litellm.native.chat_providers.support import (
dispatch_client_azure,
dispatch_client_http,
get_encoding,
print_verbose,
provider_config_items,
)
from litellm.secret_managers.main import get_secret_str
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.utils import get_secret
def complete_azure(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
_azure_detection_model: Final = ctx._azure_detection_model
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
api_version = ctx.api_version
client: Final = dispatch_client_azure(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
extra_headers: Final = ctx.extra_headers
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
max_retries: Final = ctx.max_retries
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
timeout: Final = ctx.timeout
dynamic_params = False
if client is not None and (isinstance(client, openai.AzureOpenAI) or isinstance(client, openai.AsyncAzureOpenAI)):
dynamic_params = check_dynamic_azure_params(
azure_client_params={"api_version": api_version},
azure_client=client,
)
api_type: Final = get_secret("AZURE_API_TYPE") or "azure"
api_base = api_base or litellm.api_base or get_secret("AZURE_API_BASE")
api_version = (
api_version or litellm.api_version or get_secret_str("AZURE_API_VERSION") or litellm.AZURE_DEFAULT_API_VERSION
)
api_key = (
api_key
or litellm.api_key
or litellm.azure_key
or get_secret_str("AZURE_OPENAI_API_KEY")
or get_secret_str("AZURE_API_KEY")
)
azure_ad_token: Final = optional_params.get("extra_body", {}).pop("azure_ad_token", None) or get_secret_str(
"AZURE_AD_TOKEN"
)
azure_ad_token_provider_value: Final = litellm_params.get("azure_ad_token_provider", None)
azure_ad_token_provider: Final = azure_ad_token_provider_value if callable(azure_ad_token_provider_value) else None
headers = headers or litellm.headers
if extra_headers is not None:
optional_params["extra_headers"] = extra_headers
if max_retries is not None:
optional_params["max_retries"] = max_retries
if litellm.AzureOpenAIO1Config().is_o_series_model(model=_azure_detection_model):
## LOAD CONFIG - if set
config = litellm.AzureOpenAIO1Config.get_config()
for k, v in provider_config_items(config):
if (
k not in optional_params
): # completion(top_k=3) > azure_config(top_k=3) <- allows for dynamic variables to be passed in
optional_params[k] = v
response = azure_o1_chat_completions.completion(
model=model,
messages=messages,
headers=headers,
api_key=api_key,
api_base=api_base,
api_version=api_version,
dynamic_params=dynamic_params,
azure_ad_token=azure_ad_token,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
logging_obj=logging,
acompletion=acompletion,
timeout=timeout,
client=client, # pass AsyncAzureOpenAI, AzureOpenAI client
custom_llm_provider=custom_llm_provider,
)
else:
## LOAD CONFIG - if set
config = litellm.AzureOpenAIConfig.get_config()
for k, v in provider_config_items(config):
if (
k not in optional_params
): # completion(top_k=3) > azure_config(top_k=3) <- allows for dynamic variables to be passed in
optional_params[k] = v
## COMPLETION CALL
response = azure_chat_completions.completion(
model=model,
messages=messages,
headers=headers,
api_key=api_key,
api_base=api_base,
api_version=api_version,
api_type=api_type,
dynamic_params=dynamic_params,
azure_ad_token=azure_ad_token,
azure_ad_token_provider=azure_ad_token_provider,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
logging_obj=logging,
acompletion=acompletion,
timeout=timeout,
client=client, # pass AsyncAzureOpenAI, AzureOpenAI client
)
if optional_params.get("stream", False):
## LOGGING
logging.post_call(
input=messages,
api_key=api_key,
original_response=response,
additional_args={
"headers": headers,
"api_version": api_version,
"api_base": api_base,
},
)
return response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract
def complete_azure_text(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
api_version = ctx.api_version
client: Final = dispatch_client_azure(ctx)
extra_headers: Final = ctx.extra_headers
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
timeout: Final = ctx.timeout
api_type: Final = get_secret_str("AZURE_API_TYPE") or "azure"
api_base = api_base or litellm.api_base or get_secret_str("AZURE_API_BASE")
if api_base is None:
raise ValueError(
"api_base is required for Azure OpenAI LLM provider. Either set it dynamically or set the AZURE_API_BASE environment variable."
)
api_version = api_version or litellm.api_version or get_secret_str("AZURE_API_VERSION")
api_key = (
api_key
or litellm.api_key
or litellm.azure_key
or get_secret_str("AZURE_OPENAI_API_KEY")
or get_secret_str("AZURE_API_KEY")
)
azure_ad_token: Final = optional_params.get("extra_body", {}).pop("azure_ad_token", None) or get_secret_str(
"AZURE_AD_TOKEN"
)
azure_ad_token_provider_value: Final = litellm_params.get("azure_ad_token_provider", None)
azure_ad_token_provider: Final = azure_ad_token_provider_value if callable(azure_ad_token_provider_value) else None
headers = headers or litellm.headers
if extra_headers is not None:
optional_params["extra_headers"] = extra_headers
## LOAD CONFIG - if set
config: Final = litellm.AzureOpenAIConfig.get_config()
for k, v in provider_config_items(config):
if (
k not in optional_params
): # completion(top_k=3) > azure_config(top_k=3) <- allows for dynamic variables to be passed in
optional_params[k] = v
## COMPLETION CALL
response: Final = azure_text_completions.completion(
model=model,
messages=messages,
headers=headers,
api_key=api_key,
api_base=api_base,
api_version=cast(str, api_version),
api_type=api_type,
azure_ad_token=azure_ad_token,
azure_ad_token_provider=azure_ad_token_provider,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
logging_obj=logging,
acompletion=acompletion,
timeout=timeout,
client=client, # pass AsyncAzureOpenAI, AzureOpenAI client
)
if optional_params.get("stream", False) or acompletion is True:
## LOGGING
logging.post_call(
input=messages,
api_key=api_key,
original_response=response,
additional_args={
"headers": headers,
"api_version": api_version,
"api_base": api_base,
},
)
return response
def complete_azure_ai(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
extra_headers: Final = ctx.extra_headers
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
from litellm.llms.azure_ai.common_utils import AzureFoundryModelInfo
azure_ai_route: Final = AzureFoundryModelInfo.get_azure_ai_route(model)
# Check if this is an agents route - model format: azure_ai/agents/<agent_id>
if azure_ai_route == "agents":
from litellm.llms.azure_ai.agents import AzureAIAgentsConfig
api_base = AzureFoundryModelInfo.get_api_base(api_base)
if api_base is None:
raise ValueError(
"Azure AI Agents requests require an api_base. Set `api_base` or the AZURE_AI_API_BASE env var."
)
api_key = AzureFoundryModelInfo.get_api_key(api_key)
response = AzureAIAgentsConfig.completion(
model=model,
messages=messages,
api_base=api_base,
api_key=api_key,
model_response=model_response,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
acompletion=acompletion,
stream=stream,
headers=headers or litellm.headers,
)
# Check if this is a Claude model - route to Azure Anthropic handler
elif "claude" in model.lower():
# Use Azure Anthropic handler for Claude models
api_base = AzureFoundryModelInfo.get_api_base(api_base)
if api_base is None:
raise ValueError(
"Azure Anthropic requests require an api_base. Set `api_base` or the AZURE_AI_API_BASE env var."
)
api_key = AzureFoundryModelInfo.get_api_key(api_key)
# Ensure the URL ends with /v1/messages for Anthropic
if api_base:
api_base = api_base.rstrip("/")
if not api_base.endswith("/v1/messages"):
if "/anthropic" in api_base:
parts: Final = api_base.split("/anthropic", 1)
api_base = parts[0] + "/anthropic"
else:
api_base = api_base + "/anthropic"
api_base = api_base + "/v1/messages"
response = azure_anthropic_chat_completions.completion(
model=model,
messages=messages,
api_base=api_base,
acompletion=acompletion,
custom_prompt_dict=litellm.custom_prompt_dict,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
headers=headers,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
)
if optional_params.get("stream", False) or acompletion is True:
## LOGGING
logging.post_call(
input=messages,
api_key=api_key,
original_response=response,
)
response = response
else:
# Non-Claude models use standard Azure AI flow
api_base = AzureFoundryModelInfo.get_api_base(api_base)
# set API KEY
api_key = AzureFoundryModelInfo.get_api_key(api_key)
headers = headers or litellm.headers
if extra_headers is not None:
optional_params["extra_headers"] = extra_headers
## FOR COHERE
if "command-r" in model: # make sure tool call in messages are str
messages = stringify_json_tool_call_content(messages=messages)
## COMPLETION CALL
try:
response = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client, # pass AsyncOpenAI, OpenAI client
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
)
except Exception as e:
## LOGGING - log the original exception returned
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": headers},
)
raise e
if optional_params.get("stream", False):
## LOGGING
logging.post_call(
input=messages,
api_key=api_key,
original_response=response,
additional_args={"headers": headers},
)
return response

View file

@ -0,0 +1,147 @@
import time
from typing import Final
import httpx
import litellm
from litellm.exceptions import LiteLLMUnknownProvider
from litellm.llms.custom_llm import CustomLLM, custom_chat_llm_router
from litellm.native.chat_providers.support import get_encoding, print_verbose
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.utils import CustomStreamWrapper
def _custom_api_first_output(resp: httpx.Response | None) -> str:
return resp.json()["data"][0]["output"][0]
def complete_custom(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
api_base: Final = ctx.api_base
headers: Final = ctx.headers
kwargs: Final = ctx.kwargs
max_tokens: Final = ctx.max_tokens
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
temperature: Final = ctx.temperature
top_p: Final = ctx.top_p
url: Final = litellm.api_base or api_base or ""
if url is None or url == "":
raise ValueError("api_base not set. Set api_base or litellm.api_base for custom endpoints")
"""
assume input to custom LLM api bases follow this format:
resp = litellm.module_level_client.post(
api_base,
json={
'model': 'meta-llama/Llama-2-13b-hf', # model name
'params': {
'prompt': ["The capital of France is P"],
'max_tokens': 32,
'temperature': 0.7,
'top_p': 1.0,
'top_k': 40,
}
}
)
"""
prompt: Final = " ".join([message["content"] for message in messages])
resp: Final = litellm.module_level_client.post(
url,
headers=headers,
json={
"model": model,
"params": {
"prompt": [prompt],
"max_tokens": max_tokens,
"temperature": temperature,
"top_p": top_p,
"top_k": kwargs.get("top_k"),
},
**kwargs.get("extra_body", {}),
},
)
"""
assume all responses from custom api_bases of this format:
{
'data': [
{
'prompt': 'The capital of France is P',
'output': ['The capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France is PARIS.\nThe capital of France'],
'params': {'temperature': 0.7, 'top_k': 40, 'top_p': 1}}],
'message': 'ok'
}
]
}
"""
string_response: Final = _custom_api_first_output(resp)
## RESPONSE OBJECT
model_response.choices[0].message.content = string_response
model_response.created = int(time.time())
model_response.model = model
return model_response
def complete_custom_providers(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = ctx.client
custom_llm_provider: Final = ctx.custom_llm_provider
custom_prompt_dict: Final = ctx.custom_prompt_dict
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
custom_handler: CustomLLM | None = None
for item in litellm.custom_provider_map:
if item["provider"] == custom_llm_provider:
custom_handler = item["custom_handler"]
if custom_handler is None:
raise LiteLLMUnknownProvider(model=model, custom_llm_provider=custom_llm_provider)
## ROUTE LLM CALL ##
handler_fn: Final = custom_chat_llm_router(async_fn=acompletion, stream=stream, custom_llm=custom_handler)
headers = headers or litellm.headers or {}
## CALL FUNCTION
response: Final = handler_fn(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
print_verbose=print_verbose,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
timeout=timeout,
custom_prompt_dict=custom_prompt_dict,
client=client, # pass AsyncOpenAI, OpenAI client
encoding=get_encoding(),
)
if stream is True:
return CustomStreamWrapper(
completion_stream=response,
model=model,
custom_llm_provider=custom_llm_provider,
logging_obj=logging,
)
return response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract

View file

@ -0,0 +1,299 @@
from typing import Final
import openai
import litellm
from litellm.constants import NADIR_DEFAULT_API_BASE
from litellm.native.chat_providers.handlers import base_llm_http_handler
from litellm.native.chat_providers.support import dispatch_client_http, get_encoding, provider_config_items
from litellm.secret_managers.main import get_secret_str
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.utils import get_secret
def complete_cometapi(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or litellm.cometapi_key or get_secret_str("COMETAPI_KEY") or litellm.api_key
api_base = api_base or litellm.api_base or get_secret_str("COMETAPI_API_BASE") or "https://api.cometapi.com/v1"
## COMPLETION CALL
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
## LOGGING
logging.post_call(input=messages, api_key=api_key, original_response=response)
return response
def complete_openrouter(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = api_base or litellm.api_base or get_secret_str("OPENROUTER_API_BASE") or "https://openrouter.ai/api/v1"
api_key = (
api_key
or litellm.api_key
or litellm.openrouter_key
or get_secret_str("OPENROUTER_API_KEY")
or get_secret_str("OR_API_KEY")
)
openrouter_site_url: Final = get_secret("OR_SITE_URL") or "https://litellm.ai"
openrouter_app_name: Final = get_secret("OR_APP_NAME") or "liteLLM"
openrouter_headers: Final = {
"HTTP-Referer": openrouter_site_url,
"X-Title": openrouter_app_name,
}
_headers: Final = headers or litellm.headers
if _headers:
openrouter_headers.update(_headers)
headers = openrouter_headers
## Load Config
config: Final = litellm.OpenrouterConfig.get_config()
for k, v in provider_config_items(config):
if k == "extra_body":
# we use openai 'extra_body' to pass openrouter specific params - transforms, route, models
if "extra_body" in optional_params:
optional_params[k].update(v)
else:
optional_params[k] = v
elif k not in optional_params:
optional_params[k] = v
## COMPLETION CALL
response: Final = base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="openrouter",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
client=client,
)
## LOGGING
logging.post_call(input=messages, api_key=openai.api_key, original_response=response)
return response
def complete_nadir(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
api_base: Final = ctx.api_base or litellm.api_base or get_secret_str("NADIR_API_BASE") or NADIR_DEFAULT_API_BASE
api_key: Final = ctx.api_key
response: Final = base_llm_http_handler.completion(
model=ctx.model,
stream=ctx.stream,
messages=ctx.messages,
acompletion=ctx.acompletion,
api_base=api_base,
model_response=ctx.model_response,
optional_params=ctx.optional_params,
litellm_params=ctx.litellm_params,
shared_session=ctx.shared_session,
custom_llm_provider="nadir",
timeout=ctx.timeout,
headers=ctx.headers or litellm.headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=ctx.logging,
client=ctx.client,
)
ctx.logging.post_call(input=ctx.messages, api_key=api_key, original_response=response)
return response
def complete_vercel_ai_gateway(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = (
api_base
or litellm.api_base
or get_secret_str("VERCEL_AI_GATEWAY_API_BASE")
or "https://ai-gateway.vercel.sh/v1"
)
api_key = api_key or litellm.api_key or get_secret("VERCEL_AI_GATEWAY_API_KEY")
vercel_site_url: Final = get_secret("VERCEL_SITE_URL") or "https://litellm.ai"
vercel_app_name: Final = get_secret("VERCEL_APP_NAME") or "liteLLM"
vercel_headers: Final = {
"http-referer": vercel_site_url,
"x-title": vercel_app_name,
}
_headers: Final = headers or litellm.headers
if _headers:
vercel_headers.update(_headers)
headers = vercel_headers
## Load Config
config: Final = litellm.VercelAIGatewayConfig.get_config()
for k, v in provider_config_items(config):
if k == "extra_body":
# we use openai 'extra_body' to pass vercel specific params - providerOptions
if "extra_body" in optional_params:
optional_params[k].update(v)
else:
optional_params[k] = v
elif k not in optional_params:
optional_params[k] = v
## COMPLETION CALL
response: Final = base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="vercel_ai_gateway",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
client=client,
)
## LOGGING
logging.post_call(input=messages, api_key=openai.api_key, original_response=response)
return response
def complete_edenai(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
api_base: Final = litellm.EdenAIChatConfig.get_api_base(ctx.api_base)
api_key: Final = litellm.EdenAIChatConfig.get_api_key(ctx.api_key or litellm.api_key)
response: Final = base_llm_http_handler.completion(
model=ctx.model,
messages=ctx.messages,
api_base=api_base,
custom_llm_provider="edenai",
model_response=ctx.model_response,
encoding=get_encoding(),
logging_obj=ctx.logging,
optional_params=ctx.optional_params,
timeout=ctx.timeout,
litellm_params=ctx.litellm_params,
shared_session=ctx.shared_session,
acompletion=ctx.acompletion,
stream=ctx.stream,
api_key=api_key,
headers=ctx.headers or litellm.headers,
client=dispatch_client_http(ctx),
provider_config=ctx.provider_config,
)
ctx.logging.post_call(input=ctx.messages, api_key=api_key, original_response=response)
return response
def complete_fal_ai(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
if ctx.stream:
raise litellm.FalAIError(
status_code=400,
message="fal_ai chat completions do not support streaming",
)
api_base: Final = litellm.FalAIChatConfig.get_api_base(ctx.api_base)
api_key: Final = litellm.FalAIChatConfig.get_api_key(ctx.api_key or litellm.api_key)
response: Final = base_llm_http_handler.completion(
model=ctx.model,
messages=ctx.messages,
api_base=api_base,
custom_llm_provider="fal_ai",
model_response=ctx.model_response,
encoding=get_encoding(),
logging_obj=ctx.logging,
optional_params=ctx.optional_params,
timeout=ctx.timeout,
litellm_params=ctx.litellm_params,
shared_session=ctx.shared_session,
acompletion=ctx.acompletion,
stream=ctx.stream,
api_key=api_key,
headers=ctx.headers or litellm.headers,
client=dispatch_client_http(ctx),
provider_config=ctx.provider_config,
)
ctx.logging.post_call(input=ctx.messages, api_key=api_key, original_response=response)
return response

View file

@ -0,0 +1,311 @@
from typing import Final
import litellm
from litellm.litellm_core_utils.core_helpers import safe_deep_copy
from litellm.llms.gemini.common_utils import get_api_key_from_env
from litellm.llms.vertex_ai import vertex_ai_non_gemini
from litellm.llms.vertex_ai.common_utils import VertexAIModelRoute, get_vertex_ai_model_route
from litellm.native.chat_providers.handlers import (
base_llm_http_handler,
gdc_transformation,
vertex_chat_completion,
vertex_gemma_chat_completion,
vertex_model_garden_chat_completion,
vertex_partner_models_chat_completion,
)
from litellm.native.chat_providers.support import dispatch_client_http, get_encoding, print_verbose
from litellm.secret_managers.main import get_secret_str
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.utils import CustomStreamWrapper, get_secret
def complete_vertex_ai_beta(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
timeout: Final = ctx.timeout
vertex_ai_project: Final = (
optional_params.pop("vertex_project", None)
or optional_params.pop("vertex_ai_project", None)
or litellm.vertex_project
or get_secret("VERTEXAI_PROJECT")
)
vertex_ai_location: Final = (
optional_params.pop("vertex_location", None)
or optional_params.pop("vertex_ai_location", None)
or litellm.vertex_location
or get_secret("VERTEXAI_LOCATION")
)
vertex_credentials: Final = (
optional_params.pop("vertex_credentials", None)
or optional_params.pop("vertex_ai_credentials", None)
or get_secret("VERTEXAI_CREDENTIALS")
)
gemini_api_key: Final = (
api_key
or get_api_key_from_env()
or get_secret("PALM_API_KEY") # older palm api key should also work
or litellm.api_key
)
api_base = api_base or litellm.api_base or get_secret("GEMINI_API_BASE")
new_params: Final = safe_deep_copy(optional_params or {})
return vertex_chat_completion.completion(
model=model,
messages=messages,
model_response=model_response,
print_verbose=print_verbose,
optional_params=new_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
vertex_location=vertex_ai_location,
vertex_project=vertex_ai_project,
vertex_credentials=vertex_credentials,
gemini_api_key=gemini_api_key,
logging_obj=logging,
acompletion=acompletion,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
client=client,
api_base=api_base,
extra_headers=headers,
)
def complete_vertex_ai(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
custom_prompt_dict: Final = ctx.custom_prompt_dict
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
vertex_ai_project: Final = (
optional_params.pop("vertex_project", None)
or optional_params.pop("vertex_ai_project", None)
or litellm.vertex_project
or get_secret("VERTEXAI_PROJECT")
)
vertex_ai_location: Final = (
optional_params.pop("vertex_location", None)
or optional_params.pop("vertex_ai_location", None)
or litellm.vertex_location
or get_secret("VERTEXAI_LOCATION")
)
vertex_credentials: Final = (
optional_params.pop("vertex_credentials", None)
or optional_params.pop("vertex_ai_credentials", None)
or get_secret("VERTEXAI_CREDENTIALS")
)
api_base = api_base or litellm.api_base or get_secret("VERTEXAI_API_BASE")
new_params: Final = safe_deep_copy(optional_params or {})
model_route: Final = get_vertex_ai_model_route(model=model, litellm_params=litellm_params)
if model_route == VertexAIModelRoute.PARTNER_MODELS:
model_response = vertex_partner_models_chat_completion.completion(
model=model,
messages=messages,
model_response=model_response,
print_verbose=print_verbose,
optional_params=new_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
api_base=api_base,
vertex_location=vertex_ai_location,
vertex_project=vertex_ai_project,
vertex_credentials=vertex_credentials,
logging_obj=logging,
acompletion=acompletion,
headers=headers,
custom_prompt_dict=custom_prompt_dict,
timeout=timeout,
client=client,
)
elif model_route == VertexAIModelRoute.GEMINI:
model_response = vertex_chat_completion.completion(
model=model,
messages=messages,
model_response=model_response,
print_verbose=print_verbose,
optional_params=new_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
vertex_location=vertex_ai_location,
vertex_project=vertex_ai_project,
vertex_credentials=vertex_credentials,
gemini_api_key=None,
logging_obj=logging,
acompletion=acompletion,
timeout=timeout,
custom_llm_provider=custom_llm_provider,
client=client,
api_base=api_base,
extra_headers=headers,
)
elif model_route == VertexAIModelRoute.GEMMA:
# Vertex Gemma Models with custom prediction endpoint
model_response = vertex_gemma_chat_completion.completion(
model=model,
messages=messages,
model_response=model_response,
print_verbose=print_verbose,
optional_params=new_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
api_base=api_base,
vertex_location=vertex_ai_location,
vertex_project=vertex_ai_project,
vertex_credentials=vertex_credentials,
logging_obj=logging,
acompletion=acompletion,
headers=headers,
custom_prompt_dict=custom_prompt_dict,
timeout=timeout,
client=client,
)
elif model_route == VertexAIModelRoute.MODEL_GARDEN:
# Vertex Model Garden - OpenAI compatible models
model_response = vertex_model_garden_chat_completion.completion(
model=model,
messages=messages,
model_response=model_response,
print_verbose=print_verbose,
optional_params=new_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
api_base=api_base,
vertex_location=vertex_ai_location,
vertex_project=vertex_ai_project,
vertex_credentials=vertex_credentials,
logging_obj=logging,
acompletion=acompletion,
headers=headers,
custom_prompt_dict=custom_prompt_dict,
timeout=timeout,
client=client,
)
elif model_route == VertexAIModelRoute.AGENT_ENGINE:
# Vertex AI Agent Engine (Reasoning Engines)
from litellm.llms.vertex_ai.agent_engine.transformation import (
VertexAgentEngineConfig,
)
vertex_agent_engine_config: Final = VertexAgentEngineConfig()
# Update litellm_params with vertex credentials
litellm_params["vertex_project"] = vertex_ai_project
litellm_params["vertex_location"] = vertex_ai_location
litellm_params["vertex_credentials"] = vertex_credentials
model_response = base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
model_response=model_response,
optional_params=new_params,
litellm_params=litellm_params,
encoding=get_encoding(),
api_key=None,
api_base=api_base,
logging_obj=logging,
acompletion=acompletion,
timeout=timeout,
client=client,
custom_llm_provider="vertex_ai",
provider_config=vertex_agent_engine_config,
headers=headers or {},
)
else: # VertexAIModelRoute.NON_GEMINI
model_response = vertex_ai_non_gemini.completion(
model=model,
messages=messages,
model_response=model_response,
print_verbose=print_verbose,
optional_params=new_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
vertex_location=vertex_ai_location,
vertex_project=vertex_ai_project,
vertex_credentials=vertex_credentials,
logging_obj=logging,
acompletion=acompletion,
)
if "stream" in optional_params and optional_params["stream"] is True and acompletion is False:
return CustomStreamWrapper(
model_response,
model,
custom_llm_provider="vertex_ai",
logging_obj=logging,
)
return model_response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract
def complete_gdc(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or litellm.gdc_key or get_secret_str("GDC_API_KEY") or litellm.api_key
api_base = api_base or litellm.gdc_api_base or get_secret_str("GDC_API_BASE") or litellm.api_base
return base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=gdc_transformation,
)

View file

@ -0,0 +1,51 @@
from typing import Final
from litellm.llms.anthropic.chat import AnthropicChatCompletion
from litellm.llms.azure.azure import AzureChatCompletion
from litellm.llms.azure.chat.o_series_handler import AzureOpenAIO1ChatCompletion
from litellm.llms.azure.completion.handler import AzureTextCompletion
from litellm.llms.azure_ai.anthropic.handler import AzureAnthropicChatCompletion
from litellm.llms.bedrock.chat import BedrockConverseLLM
from litellm.llms.bytez.chat.transformation import BytezChatConfig
from litellm.llms.codestral.completion.handler import CodestralTextCompletion
from litellm.llms.custom_httpx.aiohttp_handler import BaseLLMAIOHTTPHandler
from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
from litellm.llms.gdc.chat.transformation import GDCGeminiConfig
from litellm.llms.lemonade.chat.transformation import LemonadeChatConfig
from litellm.llms.openai.completion.handler import OpenAITextCompletion
from litellm.llms.openai.openai import OpenAIChatCompletion
from litellm.llms.openai_like.chat.handler import OpenAILikeChatHandler
from litellm.llms.ovhcloud.chat.transformation import OVHCloudChatConfig
from litellm.llms.predibase.chat.handler import PredibaseChatCompletion
from litellm.llms.sagemaker.completion.handler import SagemakerLLM
from litellm.llms.sap.chat.handler import GenAIHubOrchestration
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import VertexLLM
from litellm.llms.vertex_ai.vertex_ai_partner_models.main import VertexAIPartnerModels
from litellm.llms.vertex_ai.vertex_gemma_models.main import VertexAIGemmaModels
from litellm.llms.vertex_ai.vertex_model_garden.main import VertexAIModelGardenModels
from litellm.llms.watsonx.chat.handler import WatsonXChatHandler
openai_chat_completions = OpenAIChatCompletion()
openai_text_completions: Final = OpenAITextCompletion()
sap_gen_ai_hub_chat_completions: Final = GenAIHubOrchestration()
anthropic_chat_completions: Final = AnthropicChatCompletion()
azure_anthropic_chat_completions: Final = AzureAnthropicChatCompletion()
azure_chat_completions = AzureChatCompletion()
azure_o1_chat_completions: Final = AzureOpenAIO1ChatCompletion()
azure_text_completions: Final = AzureTextCompletion()
predibase_chat_completions: Final = PredibaseChatCompletion()
codestral_text_completions: Final = CodestralTextCompletion()
bedrock_converse_chat_completion: Final = BedrockConverseLLM()
vertex_chat_completion: Final = VertexLLM()
vertex_partner_models_chat_completion: Final = VertexAIPartnerModels()
vertex_gemma_chat_completion: Final = VertexAIGemmaModels()
vertex_model_garden_chat_completion: Final = VertexAIModelGardenModels()
gdc_transformation: Final = GDCGeminiConfig()
sagemaker_llm: Final = SagemakerLLM()
watsonx_chat_completion: Final = WatsonXChatHandler()
openai_like_chat_completion: Final = OpenAILikeChatHandler()
base_llm_http_handler = BaseLLMHTTPHandler()
base_llm_aiohttp_handler = BaseLLMAIOHTTPHandler()
bytez_transformation: Final = BytezChatConfig()
ovhcloud_transformation: Final = OVHCloudChatConfig()
lemonade_transformation: Final = LemonadeChatConfig()

View file

@ -0,0 +1,905 @@
import os
from types import MappingProxyType
from typing import Final, cast # noqa: TID251 # casts moved verbatim from litellm/main.py
import httpx
import litellm
from litellm._logging import verbose_logger
from litellm.llms.cohere.common_utils import CohereModelInfo
from litellm.llms.deprecated_providers import aleph_alpha
from litellm.llms.nlp_cloud.chat.handler import completion as nlp_cloud_chat_completion
from litellm.llms.replicate.chat.handler import completion as replicate_chat_completion
from litellm.native.chat_providers.handlers import (
base_llm_http_handler,
bytez_transformation,
codestral_text_completions,
openai_like_chat_completion,
predibase_chat_completions,
)
from litellm.native.chat_providers.support import (
dispatch_client_http,
get_encoding,
print_verbose,
provider_config_items,
)
from litellm.secret_managers.main import get_secret_str
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.utils import CustomStreamWrapper, get_secret
def complete_deepseek(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
try:
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
except Exception as e:
## LOGGING - log the original exception returned
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": headers},
)
raise e
return response
def complete_fireworks_ai(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
optional_params: Final = (
provider_config.map_extra_body_params(optional_params=ctx.optional_params, model=model)
if isinstance(provider_config, litellm.FireworksAIConfig)
else ctx.optional_params
)
try:
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
except Exception as e:
## LOGGING - log the original exception returned
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": headers},
)
raise e
return response
def complete_together_ai(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
try:
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
except Exception as e:
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args=MappingProxyType({"headers": headers}),
)
raise
return response
def complete_scaledown(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
try:
return base_llm_http_handler.completion(
model=ctx.model,
messages=ctx.messages,
headers=ctx.headers,
model_response=ctx.model_response,
api_key=ctx.api_key,
api_base=ctx.api_base,
acompletion=ctx.acompletion,
logging_obj=ctx.logging,
optional_params=ctx.optional_params,
litellm_params=ctx.litellm_params,
shared_session=ctx.shared_session,
timeout=cast( # cast-ok: CompletionTimeout.resolve normalizes this value
float | httpx.Timeout, ctx.timeout
),
client=dispatch_client_http(ctx),
custom_llm_provider=ctx.custom_llm_provider,
encoding=get_encoding(),
stream=ctx.stream,
provider_config=ctx.provider_config,
)
except Exception as exc:
ctx.logging.post_call(
input=ctx.messages,
api_key=ctx.api_key,
original_response=str(exc),
additional_args={"headers": ctx.headers},
)
raise
def complete_xai(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
try:
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
except Exception as e:
## LOGGING - log the original exception returned
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": headers},
)
raise e
return response
def complete_groq(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = (
api_base # for deepinfra/perplexity/anyscale/groq/friendliai we check in get_llm_provider and pass in the api base from there
or litellm.api_base
or get_secret("GROQ_API_BASE")
or "https://api.groq.com/openai/v1"
)
# set API KEY
api_key = (
api_key
or litellm.api_key # for deepinfra/perplexity/anyscale/friendliai we check in get_llm_provider and pass in the api key from there
or litellm.groq_key
or get_secret("GROQ_API_KEY")
)
headers = headers or litellm.headers
## LOAD CONFIG - if set
config: Final = litellm.GroqChatConfig.get_config()
for k, v in provider_config_items(config):
if (
k not in optional_params
): # completion(top_k=3) > openai_config(top_k=3) <- allows for dynamic variables to be passed in
optional_params[k] = v
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider=custom_llm_provider,
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
client=client,
)
def complete_minimax(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or get_secret_str("MINIMAX_API_KEY") or litellm.api_key
api_base = api_base or litellm.api_base or get_secret_str("MINIMAX_API_BASE") or "https://api.minimax.io/v1"
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
api_base=api_base,
custom_llm_provider=custom_llm_provider,
model_response=model_response,
encoding=get_encoding(),
logging_obj=logging,
optional_params=optional_params,
timeout=timeout,
litellm_params=litellm_params,
shared_session=shared_session,
acompletion=acompletion,
stream=stream,
api_key=api_key,
headers=headers,
client=client,
provider_config=provider_config,
)
logging.post_call(input=messages, api_key=api_key, original_response=response)
return response
def complete_mistral(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or litellm.api_key or get_secret("MISTRAL_API_KEY")
api_base = api_base or litellm.api_base or get_secret("MISTRAL_API_BASE") or "https://api.mistral.ai/v1"
return base_llm_http_handler.completion(
model=model,
messages=messages,
api_base=api_base,
custom_llm_provider=custom_llm_provider,
model_response=model_response,
encoding=get_encoding(),
logging_obj=logging,
optional_params=optional_params,
timeout=timeout,
litellm_params=litellm_params,
shared_session=shared_session,
acompletion=acompletion,
stream=stream,
api_key=api_key,
headers=headers,
client=client,
provider_config=provider_config,
)
def complete_replicate(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key: Final = ctx.api_key
custom_prompt_dict = ctx.custom_prompt_dict
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response = ctx.model_response
optional_params: Final = ctx.optional_params
replicate_key: Final = (
api_key
or litellm.replicate_key
or litellm.api_key
or get_secret("REPLICATE_API_KEY")
or get_secret("REPLICATE_API_TOKEN")
)
api_base = api_base or litellm.api_base or get_secret("REPLICATE_API_BASE") or "https://api.replicate.com/v1"
custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
model_response = replicate_chat_completion(
model=model,
messages=messages,
api_base=api_base,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(), # for calculating input/output tokens
api_key=replicate_key,
logging_obj=logging,
custom_prompt_dict=custom_prompt_dict,
acompletion=acompletion,
headers=headers,
)
if optional_params.get("stream", False) is True:
## LOGGING
logging.post_call(
input=messages,
api_key=replicate_key,
original_response=model_response,
)
return model_response
def complete_nlp_cloud(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key: Final = ctx.api_key
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
nlp_cloud_key: Final = api_key or litellm.nlp_cloud_key or get_secret("NLP_CLOUD_API_KEY") or litellm.api_key
api_base = api_base or litellm.api_base or get_secret("NLP_CLOUD_API_BASE") or "https://api.nlpcloud.io/v1/gpu/"
response = nlp_cloud_chat_completion(
model=model,
messages=messages,
api_base=api_base,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
api_key=nlp_cloud_key,
logging_obj=logging,
)
if "stream" in optional_params and optional_params["stream"] is True:
# don't try to access stream object,
response = CustomStreamWrapper(
response,
model,
custom_llm_provider="nlp_cloud",
logging_obj=logging,
)
if optional_params.get("stream", False) or acompletion is True:
## LOGGING
logging.post_call(
input=messages,
api_key=api_key,
original_response=response,
)
return response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract
def complete_aleph_alpha(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
api_base = ctx.api_base
api_key: Final = ctx.api_key
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response = ctx.model_response
optional_params: Final = ctx.optional_params
aleph_alpha_key: Final = (
api_key
or litellm.aleph_alpha_key
or get_secret("ALEPH_ALPHA_API_KEY")
or get_secret("ALEPHALPHA_API_KEY")
or litellm.api_key
)
api_base = (
api_base or litellm.api_base or get_secret("ALEPH_ALPHA_API_BASE") or "https://api.aleph-alpha.com/complete"
)
model_response = aleph_alpha.completion(
model=model,
messages=messages,
api_base=api_base,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
default_max_tokens_to_sample=litellm.max_tokens,
api_key=aleph_alpha_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
)
if "stream" in optional_params and optional_params["stream"] is True:
# don't try to access stream object,
return CustomStreamWrapper(
model_response,
model,
custom_llm_provider="aleph_alpha",
logging_obj=logging,
)
return model_response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract
def complete_cohere_chat(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key: Final = ctx.api_key
extra_headers: Final = ctx.extra_headers
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
cohere_key: Final = (
api_key
or litellm.cohere_key
or get_secret_str("COHERE_API_KEY")
or get_secret_str("CO_API_KEY")
or litellm.api_key
)
cohere_route: Final = CohereModelInfo.get_cohere_route(model)
verbose_logger.debug("Cohere route: %s", cohere_route)
# Set API base based on route
if cohere_route == "v2":
api_base = api_base or litellm.api_base or get_secret_str("COHERE_API_BASE") or "https://api.cohere.com/v2/chat"
# Remove v2/ prefix from model name for the actual API call
if "v2/" in model:
model = model.replace("v2/", "")
else:
api_base = api_base or litellm.api_base or get_secret_str("COHERE_API_BASE") or "https://api.cohere.ai/v1/chat"
headers = headers or litellm.headers or {}
if headers is None:
headers = {}
if extra_headers is not None:
headers.update(extra_headers)
verbose_logger.debug("Model: %s, API Base: %s", model, api_base)
verbose_logger.debug("Provider Config: %s", provider_config)
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="cohere_chat",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=cohere_key,
provider_config=provider_config,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
)
def complete_maritalk(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
api_base = ctx.api_base
api_key: Final = ctx.api_key
custom_prompt_dict: Final = ctx.custom_prompt_dict
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
maritalk_key: Final = api_key or litellm.maritalk_key or get_secret("MARITALK_API_KEY") or litellm.api_key
api_base = api_base or litellm.api_base or get_secret("MARITALK_API_BASE") or "https://chat.maritaca.ai/api"
return openai_like_chat_completion.completion(
model=model,
messages=messages,
api_base=api_base,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
api_key=maritalk_key,
logging_obj=logging,
custom_llm_provider="maritalk",
custom_prompt_dict=custom_prompt_dict,
)
def complete_huggingface(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
huggingface_key: Final = (
api_key
or litellm.huggingface_key
or os.environ.get("HF_TOKEN")
or os.environ.get("HUGGINGFACE_API_KEY")
or litellm.api_key
)
hf_headers: Final = headers or litellm.headers
return base_llm_http_handler.completion(
model=model,
messages=messages,
headers=hf_headers,
model_response=model_response,
api_key=huggingface_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
)
def complete_compactifai(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or get_secret_str("COMPACTIFAI_API_KEY") or litellm.api_key
api_base = api_base or "https://api.compactif.ai/v1"
## COMPLETION CALL
return base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
def complete_predibase(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
custom_prompt_dict: Final = ctx.custom_prompt_dict
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
timeout: Final = ctx.timeout
tenant_id: Final = (
optional_params.pop("tenant_id", None)
or optional_params.pop("predibase_tenant_id", None)
or litellm.predibase_tenant_id
or get_secret("PREDIBASE_TENANT_ID")
)
if tenant_id is None:
raise ValueError(
"Missing Predibase Tenant ID - Required for making the request. Set dynamically (e.g. `completion(..tenant_id=<MY-ID>)`) or in env - `PREDIBASE_TENANT_ID`."
)
api_base = (
api_base
or optional_params.pop("api_base", None)
or optional_params.pop("base_url", None)
or litellm.api_base
or get_secret("PREDIBASE_API_BASE")
)
api_key = api_key or litellm.api_key or litellm.predibase_key or get_secret("PREDIBASE_API_KEY")
_model_response: Final = predibase_chat_completions.completion(
model=model,
messages=messages,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
logging_obj=logging,
acompletion=acompletion,
api_base=api_base,
custom_prompt_dict=custom_prompt_dict,
api_key=api_key,
tenant_id=tenant_id,
timeout=timeout,
)
if "stream" in optional_params and optional_params["stream"] is True and acompletion is False:
return _model_response
return _model_response
def complete_text_completion_codestral(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
custom_prompt_dict: Final = ctx.custom_prompt_dict
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = (
api_base
or optional_params.pop("api_base", None)
or optional_params.pop("base_url", None)
or litellm.api_base
or "https://codestral.mistral.ai/v1/fim/completions"
)
api_key = api_key or litellm.api_key or get_secret("CODESTRAL_API_KEY")
text_completion_model_response: Final = litellm.TextCompletionResponse(stream=stream)
_model_response: Final = codestral_text_completions.completion(
model=model,
messages=messages,
model_response=text_completion_model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
logging_obj=logging,
acompletion=acompletion,
api_base=api_base,
custom_prompt_dict=custom_prompt_dict,
api_key=api_key,
timeout=timeout,
)
if "stream" in optional_params and optional_params["stream"] is True and acompletion is False:
return _model_response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract
return _model_response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract
def complete_bytez(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or litellm.bytez_key or get_secret_str("BYTEZ_API_KEY") or litellm.api_key
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=bytez_transformation,
)
return response

View file

@ -0,0 +1,388 @@
from typing import Final, cast # noqa: TID251 # casts moved verbatim from litellm/main.py
import openai
import litellm
from litellm._logging import verbose_logger
from litellm.llms.base_llm.chat.transformation import with_attribution_headers
from litellm.llms.custom_httpx.http_handler import http2_enabled
from litellm.native.chat_providers.handlers import (
base_llm_aiohttp_handler,
base_llm_http_handler,
openai_chat_completions,
openai_text_completions,
)
from litellm.native.chat_providers.support import (
dispatch_client_http,
dispatch_client_openai,
dispatch_metadata,
get_encoding,
print_verbose,
provider_config_items,
)
from litellm.responses.bridge_check import resolve_openai_api_base
from litellm.secret_managers.main import get_secret_bool, get_secret_str
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.utils import get_requester_metadata, get_secret
def complete_text_completion_openai(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_openai(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
text_completion: Final = ctx.text_completion
timeout: Final = ctx.timeout
openai.api_type = "openai"
api_base = (
api_base
or litellm.api_base
or get_secret("OPENAI_BASE_URL")
or get_secret("OPENAI_API_BASE")
or "https://api.openai.com/v1"
)
openai.api_version = None
# set API KEY
api_key = api_key or litellm.api_key or litellm.openai_key or get_secret("OPENAI_API_KEY")
headers = headers or litellm.headers
## LOAD CONFIG - if set
config: Final = litellm.OpenAITextCompletionConfig.get_config()
for k, v in provider_config_items(config):
if (
k not in optional_params
): # completion(top_k=3) > openai_text_config(top_k=3) <- allows for dynamic variables to be passed in
optional_params[k] = v
if litellm.organization:
openai.organization = litellm.organization
## COMPLETION CALL
_response = openai_text_completions.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
print_verbose=print_verbose,
api_key=api_key,
custom_llm_provider=custom_llm_provider,
api_base=api_base,
acompletion=acompletion,
client=client, # pass AsyncOpenAI, OpenAI client
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
timeout=timeout,
)
if optional_params.get("stream", False) is False and acompletion is False and text_completion is False:
# convert to chat completion response
_response = litellm.OpenAITextCompletionConfig().convert_to_chat_model_response_object(
response_object=_response, model_response_object=model_response
)
if optional_params.get("stream", False) or acompletion is True:
## LOGGING
logging.post_call(
input=messages,
api_key=api_key,
original_response=_response,
additional_args={"headers": headers},
)
return _response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract
def complete_aiohttp_openai(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
if http2_enabled():
verbose_logger.warning(
"litellm.http2 is enabled but aiohttp_openai/ always uses aiohttp, which has no HTTP/2 client; this request stays on HTTP/1.1"
)
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
extra_headers: Final = ctx.extra_headers
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = (
api_base # for deepinfra/perplexity/anyscale/groq/friendliai we check in get_llm_provider and pass in the api base from there
or litellm.api_base
or get_secret("OPENAI_BASE_URL")
or get_secret("OPENAI_API_BASE")
or "https://api.openai.com/v1"
)
# set API KEY
api_key = (
api_key
or litellm.api_key # for deepinfra/perplexity/anyscale/friendliai we check in get_llm_provider and pass in the api key from there
or litellm.openai_key
or get_secret("OPENAI_API_KEY")
)
headers = headers or litellm.headers
if extra_headers is not None:
optional_params["extra_headers"] = extra_headers
return base_llm_aiohttp_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
)
def complete_custom_openai(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
custom_prompt_dict: Final = ctx.custom_prompt_dict
extra_headers = ctx.extra_headers
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
metadata: Final = dispatch_metadata(ctx)
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
organization = ctx.organization
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
# for deepinfra/perplexity/anyscale/groq/friendliai we check in get_llm_provider and pass in the api base from there
api_base = resolve_openai_api_base(api_base)
organization = (
organization
or litellm.organization
or get_secret("OPENAI_ORGANIZATION")
or None # default - https://github.com/openai/openai-python/blob/284c1799070c723c6a553337134148a7ab088dd8/openai/util.py#L105
)
openai.organization = organization
# set API KEY
api_key = (
api_key
or litellm.api_key # for deepinfra/perplexity/anyscale/friendliai we check in get_llm_provider and pass in the api key from there
or litellm.openai_key
or get_secret("OPENAI_API_KEY")
)
headers = headers or litellm.headers
outbound_headers: Final = (
headers
if provider_config is None
else with_attribution_headers(provider_config.get_attribution_headers(), headers)
)
# Add GitHub Copilot headers (same as /responses endpoint does)
if custom_llm_provider == "github_copilot":
from litellm.llms.github_copilot.authenticator import Authenticator
from litellm.llms.github_copilot.common_utils import (
get_copilot_default_headers,
pin_session_authorization,
)
from litellm.llms.github_copilot.per_user_auth import (
require_github_copilot_user_session,
)
user_session: Final = require_github_copilot_user_session(litellm_params)
copilot_headers: Final = get_copilot_default_headers(
user_session.token if user_session is not None else Authenticator().get_api_key()
)
if extra_headers:
copilot_headers.update(
cast("dict[str, str]", extra_headers) # cast-ok: extra_headers is a str-valued request dict
)
if user_session is not None:
pin_session_authorization(copilot_headers, user_session.token)
extra_headers = copilot_headers
use_base_llm_http_handler: Final = get_secret_bool("EXPERIMENTAL_OPENAI_BASE_LLM_HTTP_HANDLER")
if extra_headers is not None and not use_base_llm_http_handler:
optional_params["extra_headers"] = extra_headers
if litellm.enable_preview_features and metadata is not None: # [PREVIEW] allow metadata to be passed to OPENAI
openai_metadata: Final = get_requester_metadata(metadata)
if openai_metadata is not None:
optional_params["metadata"] = openai_metadata
## LOAD CONFIG - if set
config: Final = litellm.OpenAIConfig.get_config()
for k, v in provider_config_items(config):
if (
k not in optional_params
): # completion(top_k=3) > openai_config(top_k=3) <- allows for dynamic variables to be passed in
optional_params[k] = v
## COMPLETION CALL
try:
if use_base_llm_http_handler:
response = base_llm_http_handler.completion(
model=model,
messages=messages,
api_base=api_base,
custom_llm_provider=custom_llm_provider,
model_response=model_response,
encoding=get_encoding(),
logging_obj=logging,
optional_params=optional_params,
timeout=timeout,
litellm_params=litellm_params,
shared_session=shared_session,
acompletion=acompletion,
stream=stream,
api_key=api_key,
headers=outbound_headers,
client=client,
provider_config=provider_config,
)
else:
response = openai_chat_completions.completion(
model=model,
messages=messages,
headers=outbound_headers,
model_response=model_response,
print_verbose=print_verbose,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
timeout=timeout,
custom_prompt_dict=custom_prompt_dict,
client=client, # pass AsyncOpenAI, OpenAI client
organization=organization,
custom_llm_provider=custom_llm_provider,
shared_session=shared_session,
)
except Exception as e:
## LOGGING - log the original exception returned
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": outbound_headers},
)
raise e
if optional_params.get("stream", False):
## LOGGING
logging.post_call(
input=messages,
api_key=api_key,
original_response=response,
additional_args={"headers": outbound_headers},
)
return response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract
def complete_text_completion_inception(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_openai(ctx)
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
text_completion: Final = ctx.text_completion
timeout: Final = ctx.timeout
passed_api_base: Final = api_base or optional_params.pop("api_base", None) or optional_params.pop("base_url", None)
api_base = passed_api_base or get_secret_str("INCEPTION_API_BASE") or "https://api.inceptionlabs.ai/v1"
# FIM is served at `/v1/fim/completions`; the OpenAI client appends
# `/completions`, so point it at the `/v1/fim` base.
api_base = api_base.rstrip("/")
if not api_base.endswith("/fim"):
api_base += "/fim"
# Don't forward the server-managed Inception key to a caller-supplied
# api_base; only resolve it for the default/server base, or when the
# caller passes their own key.
if passed_api_base is None or api_key:
api_key = api_key or litellm.inception_key or get_secret_str("INCEPTION_API_KEY")
_response = openai_text_completions.completion(
model=model,
messages=messages,
model_response=model_response,
print_verbose=print_verbose,
api_key=api_key,
custom_llm_provider="text-completion-inception",
api_base=api_base,
acompletion=acompletion,
client=client,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
timeout=timeout,
)
if optional_params.get("stream", False) is False and acompletion is False and text_completion is False:
_response = litellm.OpenAITextCompletionConfig().convert_to_chat_model_response_object(
response_object=_response, model_response_object=model_response
)
if optional_params.get("stream", False) or acompletion is True:
logging.post_call(
input=messages,
api_key=api_key,
original_response=_response,
additional_args={"headers": headers},
)
return _response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract

View file

@ -0,0 +1,623 @@
from typing import Final
import litellm
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.llms.databricks.common_utils import DatabricksBase
from litellm.native.chat_providers.handlers import (
base_llm_http_handler,
ovhcloud_transformation,
sap_gen_ai_hub_chat_completions,
watsonx_chat_completion,
)
from litellm.native.chat_providers.support import (
dispatch_client_http,
get_encoding,
print_verbose,
provider_config_items,
)
from litellm.secret_managers.main import get_secret_str
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.utils import get_secret
def complete_heroku(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
try:
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
except Exception as e:
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": headers},
)
raise e
return response
def complete_gigachat(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = (
api_key
or litellm.api_key
or litellm.gigachat_key
or get_secret("GIGACHAT_API_KEY")
or get_secret("GIGACHAT_CREDENTIALS")
)
headers = headers or litellm.headers or {}
## COMPLETION CALL
try:
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
except Exception as e:
## LOGGING - log the original exception returned
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": headers},
)
raise e
return response
def complete_sap(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
headers = headers or litellm.headers
## LOAD CONFIG - if set
config: Final = litellm.GenAIHubOrchestrationConfig.get_config()
for k, v in provider_config_items(config):
if (
k not in optional_params
): # completion(top_k=3) > openai_config(top_k=3) <- allows for dynamic variables to be passed in
optional_params[k] = v
return sap_gen_ai_hub_chat_completions.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
shared_session=shared_session,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
api_key=api_key,
api_base=api_base,
stream=stream,
)
def complete_oci(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
return base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
)
def complete_databricks(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = DatabricksBase.resolve_api_base(ctx.api_base)
api_key: Final = DatabricksBase.resolve_api_key(ctx.api_key)
client: Final = dispatch_client_http(ctx)
headers = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
headers = headers or litellm.headers
## COMPLETION CALL
try:
response: Final = base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
custom_llm_provider="databricks",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
client=client,
)
except Exception as e:
## LOGGING - log the original exception returned
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": headers},
)
raise e
if optional_params.get("stream", False):
## LOGGING
logging.post_call(
input=messages,
api_key=api_key,
original_response=response,
additional_args={"headers": headers},
)
return response
def complete_databricks_agent(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
from litellm.llms.databricks.agent.transformation import DatabricksAgentConfig
api_base, api_key = DatabricksAgentConfig.resolve_api_base_and_key(
api_base=ctx.api_base or litellm.api_base,
api_key=ctx.api_key or litellm.api_key,
)
return base_llm_http_handler.completion(
model=ctx.model,
stream=ctx.stream,
messages=ctx.messages,
acompletion=ctx.acompletion,
api_base=api_base,
model_response=ctx.model_response,
optional_params=ctx.optional_params,
litellm_params={**ctx.litellm_params, "api_base": api_base},
shared_session=ctx.shared_session,
custom_llm_provider=ctx.custom_llm_provider,
timeout=ctx.timeout,
headers=ctx.headers or litellm.headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=ctx.logging,
client=dispatch_client_http(ctx),
)
def complete_datarobot(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
stream: Final = ctx.stream
timeout: Final = ctx.timeout
return base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=provider_config,
)
def complete_watsonx(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_prompt_dict: Final = ctx.custom_prompt_dict
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
timeout: Final = ctx.timeout
return watsonx_chat_completion.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
print_verbose=print_verbose,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
timeout=timeout,
custom_prompt_dict=custom_prompt_dict,
client=client, # pass AsyncOpenAI, OpenAI client
encoding=get_encoding(),
custom_llm_provider="watsonx",
)
def complete_watsonx_text(
ctx: CompletionDispatchContext,
) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = (
api_key
or optional_params.pop("apikey", None)
or get_secret_str("WATSONX_APIKEY")
or get_secret_str("WATSONX_API_KEY")
or get_secret_str("WX_API_KEY")
)
api_base = (
api_base
or optional_params.pop(
"url",
optional_params.pop("api_base", optional_params.pop("base_url", None)),
)
or get_secret_str("WATSONX_API_BASE")
or get_secret_str("WATSONX_URL")
or get_secret_str("WX_URL")
or get_secret_str("WML_URL")
)
wx_credentials: Final = optional_params.pop(
"wx_credentials",
optional_params.pop("watsonx_credentials", None), # follow {provider}_credentials, same as vertex ai
)
token: str | None = None
if wx_credentials is not None:
api_base = wx_credentials.get("url", api_base)
api_key = wx_credentials.get("apikey", wx_credentials.get("api_key", api_key))
token = wx_credentials.get(
"token",
wx_credentials.get(
"watsonx_token", None
), # follow format of {provider}_token, same as azure - e.g. 'azure_ad_token=..'
)
if token is not None:
optional_params["token"] = token
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="watsonx_text",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
client=client,
)
def complete_cloudflare(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
custom_prompt_dict = ctx.custom_prompt_dict
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or litellm.cloudflare_api_key or litellm.api_key or get_secret("CLOUDFLARE_API_KEY")
api_base = api_base or litellm.api_base or get_secret("CLOUDFLARE_API_BASE")
custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="cloudflare",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
)
def complete_snowflake(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key: Final = ctx.api_key
injected_client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
client: Final = (
injected_client if injected_client is not None else (HTTPHandler(timeout=timeout) if stream is False else None)
) # Keep this here, otherwise, the httpx.client closes and streaming is impossible
try:
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
)
except Exception as e:
## LOGGING - log the original exception returned
logging.post_call(
input=messages,
api_key=api_key,
original_response=str(e),
additional_args={"headers": headers},
)
raise e
return response
def complete_gradient_ai(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key: Final = ctx.api_key
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = litellm.api_base or api_base
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="gradient_ai",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
)
def complete_ovhcloud(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or litellm.ovhcloud_key or get_secret_str("OVHCLOUD_API_KEY") or litellm.api_key
api_base = (
api_base
or litellm.api_base
or get_secret_str("OVHCLOUD_API_BASE")
or "https://oai.endpoints.kepler.ai.cloud.ovh.net/v1"
)
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=ovhcloud_transformation,
)
return response

View file

@ -0,0 +1,325 @@
import os
from typing import Final
import litellm
from litellm.llms.oobabooga.chat import oobabooga
from litellm.llms.petals.completion import handler as petals_handler
from litellm.llms.vllm.completion import handler as vllm_handler
from litellm.native.chat_providers.handlers import base_llm_http_handler, lemonade_transformation
from litellm.native.chat_providers.support import dispatch_client_http, get_encoding, print_verbose
from litellm.secret_managers.main import get_secret_str
from litellm.types.completion import CompletionDispatchContext, CompletionDispatchResult
from litellm.utils import CustomStreamWrapper, get_secret
def complete_hosted_vllm(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
provider_config: Final = ctx.provider_config
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = api_base or litellm.api_base or get_secret_str("HOSTED_VLLM_API_BASE")
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
api_base=api_base,
custom_llm_provider=custom_llm_provider,
model_response=model_response,
encoding=get_encoding(),
logging_obj=logging,
optional_params=optional_params,
timeout=timeout,
litellm_params=litellm_params,
shared_session=shared_session,
acompletion=acompletion,
stream=stream,
api_key=api_key,
headers=headers,
client=client,
provider_config=provider_config,
)
logging.post_call(input=messages, api_key=api_key, original_response=response)
return response
def complete_oobabooga(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
api_base: Final = ctx.api_base
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response = ctx.model_response
optional_params: Final = ctx.optional_params
model_response = oobabooga.completion(
model=model,
messages=messages,
model_response=model_response,
api_base=api_base,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
api_key=None,
logger_fn=logger_fn,
encoding=get_encoding(),
logging_obj=logging,
)
if "stream" in optional_params and optional_params["stream"] is True:
# don't try to access stream object,
return CustomStreamWrapper(
model_response,
model,
custom_llm_provider="oobabooga",
logging_obj=logging,
)
return model_response # pyright: ignore[reportReturnType] # provider SDK return type is broader than the dispatch contract
def complete_vllm(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
custom_prompt_dict = ctx.custom_prompt_dict
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response = ctx.model_response
optional_params: Final = ctx.optional_params
custom_prompt_dict = custom_prompt_dict or litellm.custom_prompt_dict
model_response = vllm_handler.completion(
model=model,
messages=messages,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
logging_obj=logging,
)
if "stream" in optional_params and optional_params["stream"] is True: ## [BETA]
# don't try to access stream object,
return CustomStreamWrapper(
model_response,
model,
custom_llm_provider="vllm",
logging_obj=logging,
)
## RESPONSE OBJECT
return model_response
def complete_ollama(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key: Final = ctx.api_key
client: Final = dispatch_client_http(ctx)
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = litellm.api_base or api_base or get_secret("OLLAMA_API_BASE") or "http://localhost:11434"
if api_key is not None and "Authorization" not in headers:
headers["Authorization"] = f"Bearer {api_key}"
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="ollama",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
client=client,
)
def complete_ollama_chat(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = litellm.api_base or api_base or get_secret("OLLAMA_API_BASE") or "http://localhost:11434"
api_key = api_key or litellm.ollama_key or os.environ.get("OLLAMA_API_KEY") or litellm.api_key
if api_key is not None and "Authorization" not in headers:
headers["Authorization"] = f"Bearer {api_key}"
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider="ollama_chat",
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging, # model call logging done inside the class as we make need to modify I/O to fit aleph alpha's requirements
client=client,
)
def complete_triton(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base = ctx.api_base
api_key: Final = ctx.api_key
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
shared_session: Final = ctx.shared_session
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_base = litellm.api_base or api_base
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
shared_session=shared_session,
custom_llm_provider=custom_llm_provider,
timeout=timeout,
headers=headers,
encoding=get_encoding(),
api_key=api_key,
logging_obj=logging,
)
def complete_petals(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
api_base = ctx.api_base
client: Final = dispatch_client_http(ctx)
litellm_params: Final = ctx.litellm_params
logger_fn: Final = ctx.logger_fn
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response = ctx.model_response
optional_params: Final = ctx.optional_params
stream = ctx.stream
api_base = api_base or litellm.api_base
stream = optional_params.pop("stream", False)
model_response = petals_handler.completion(
model=model,
messages=messages,
api_base=api_base,
model_response=model_response,
print_verbose=print_verbose,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=get_encoding(),
logging_obj=logging,
client=client,
)
if stream is True: ## [BETA]
# Fake streaming for petals
resp_string: Final = model_response["choices"][0]["message"]["content"]
return CustomStreamWrapper(
resp_string,
model,
custom_llm_provider="petals",
logging_obj=logging,
)
return model_response
def complete_lemonade(ctx: CompletionDispatchContext) -> CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
api_key = ctx.api_key
client: Final = dispatch_client_http(ctx)
custom_llm_provider: Final = ctx.custom_llm_provider
headers: Final = ctx.headers
litellm_params: Final = ctx.litellm_params
logging: Final = ctx.logging
messages: Final = ctx.messages
model: Final = ctx.model
model_response: Final = ctx.model_response
optional_params: Final = ctx.optional_params
stream: Final = ctx.stream
timeout: Final = ctx.timeout
api_key = api_key or litellm.lemonade_key or get_secret_str("LEMONADE_API_KEY") or litellm.api_key
response: Final = base_llm_http_handler.completion(
model=model,
messages=messages,
headers=headers,
model_response=model_response,
api_key=api_key,
api_base=api_base,
acompletion=acompletion,
logging_obj=logging,
optional_params=optional_params,
litellm_params=litellm_params,
timeout=timeout,
client=client,
custom_llm_provider=custom_llm_provider,
encoding=get_encoding(),
stream=stream,
provider_config=lemonade_transformation,
)
return response

View file

@ -0,0 +1,47 @@
from collections.abc import Iterable, Mapping
from functools import cache
import openai
import litellm
from litellm._lazy_imports import get_default_encoding
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.tokenizer import Encoding as Tokenizer
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.types.completion import CompletionDispatchContext
def dispatch_metadata(ctx: CompletionDispatchContext) -> Mapping[str, object] | None:
return ctx.metadata
def dispatch_client_http(ctx: CompletionDispatchContext) -> HTTPHandler | AsyncHTTPHandler | None:
return ctx.client
def dispatch_client_azure(
ctx: CompletionDispatchContext,
) -> openai.AzureOpenAI | openai.AsyncAzureOpenAI | HTTPHandler | AsyncHTTPHandler | None:
return ctx.client
def dispatch_client_openai(ctx: CompletionDispatchContext) -> openai.OpenAI | openai.AsyncOpenAI | None:
return ctx.client
def provider_config_items(config: Mapping[str, object]) -> Iterable[tuple[str, object]]:
return config.items()
def print_verbose(print_statement: object) -> None:
try:
verbose_logger.debug(print_statement)
if litellm.set_verbose:
print(print_statement) # noqa: T201
except Exception:
pass
@cache
def get_encoding() -> Tokenizer:
return get_default_encoding()

View file

@ -47,7 +47,7 @@ def s3_object_reader(bucket_name: str) -> SyncBucketObjectReader:
import boto3
from botocore.credentials import Credentials
from litellm.main import bedrock_converse_chat_completion
from litellm.native.chat_providers.handlers import bedrock_converse_chat_completion
credentials: Final[Credentials] = bedrock_converse_chat_completion.get_credentials()
s3_client: Final = boto3.client(

View file

@ -68,9 +68,9 @@ def _is_openai_backed_api_base(api_base: str) -> bool:
return hostname is not None and (hostname == _OPENAI_API_HOST or hostname.endswith(f".{_OPENAI_API_HOST}"))
def _resolve_openai_api_base(api_base: str | None) -> str:
def resolve_openai_api_base(api_base: str | None) -> str:
"""Effective OpenAI base a chat request will hit: arg > global > env > default. The bridge gate
and the ``_complete_custom_openai`` chat handler MUST resolve this identically, or a custom base
and the ``complete_custom_openai`` chat handler MUST resolve this identically, or a custom base
set via ``litellm.api_base`` or ``OPENAI_BASE_URL``/``OPENAI_API_BASE`` is invisible to the gate,
which then misreads it as the default OpenAI endpoint and bridges a request the backend can't serve."""
return (
@ -239,7 +239,7 @@ def responses_api_bridge_check(
# exactly as the chat handler does, so a custom base set via litellm.api_base or
# OPENAI_BASE_URL/OPENAI_API_BASE isn't misread as the default and bridged to a /responses route it
# lacks. A whitespace-only base collapses to the default too.
resolved_api_base: Final = _resolve_openai_api_base(api_base).strip()
resolved_api_base: Final = resolve_openai_api_base(api_base).strip()
on_foundry_openai_endpoint: Final = custom_llm_provider == "azure_ai" and azure_ai_supports_native_responses(
model, api_base
)

View file

@ -245,7 +245,7 @@ class CompletionDispatchContext:
_CompletionDispatchContext = CompletionDispatchContext
_CompletionDispatchResult = Union[
CompletionDispatchResult = Union[
Coroutine[Any, Any, Union["ModelResponse", "CustomStreamWrapper"]],
"ModelResponse",
"CustomStreamWrapper",

View file

@ -172,7 +172,7 @@ def test_azure_o_series_routing():
assert "stream" not in mock_create.call_args.kwargs["body"]
@patch("litellm.main.azure_o1_chat_completions._get_openai_client")
@patch("litellm.native.chat_providers.handlers.azure_o1_chat_completions._get_openai_client")
def test_openai_o_series_max_retries_0(mock_get_openai_client):
import litellm

View file

@ -728,7 +728,7 @@ async def test_ensure_initialize_azure_sdk_client_always_used_azure_text(call_ty
# Get appropriate input for this call type
input_kwarg = test_inputs.get(call_type.value, {})
patch_target = "litellm.main.azure_text_completions.initialize_azure_sdk_client"
patch_target = "litellm.native.chat_providers.handlers.azure_text_completions.initialize_azure_sdk_client"
# Mock the initialize_azure_sdk_client function
with patch(patch_target) as mock_init_azure:

View file

@ -21,7 +21,7 @@ def test_main_azure_ai_claude_completion_passes_timeout_to_azure_anthropic_handl
captured.update(kwargs)
return ModelResponse()
with patch("litellm.main.azure_anthropic_chat_completions") as mock_azure_anthropic:
with patch("litellm.native.chat_providers.azure.azure_anthropic_chat_completions") as mock_azure_anthropic:
mock_azure_anthropic.completion = MagicMock(
side_effect=fake_azure_anthropic_completion
)

View file

@ -686,7 +686,7 @@ class TestGDCGeminiConfig:
class TestCompleteGDC:
@patch("litellm.main.base_llm_http_handler.completion")
def test_complete_gdc_resolves_key_and_base(self, mock_completion, monkeypatch):
from litellm.main import gdc_transformation
from litellm.native.chat_providers.handlers import gdc_transformation
mock_completion.return_value = MagicMock()
monkeypatch.setattr(litellm, "gdc_key", "resolved-key", raising=False)

View file

@ -28,7 +28,7 @@ def test_gemini_completion_no_api_key():
del os.environ[key]
# Test without mock_response to ensure actual API key validation
with pytest.raises(Exception, match='in _complete_vertex_ai_beta') as exc_info:
with pytest.raises(Exception, match='in complete_vertex_ai_beta') as exc_info:
completion(
model="gemini/gemini-1.5-flash",
messages=[{"role": "user", "content": "Test message"}],
@ -60,7 +60,7 @@ def test_gemini_completion_no_api_key_with_mock():
with patch("litellm.get_secret") as mock_get_secret:
mock_get_secret.return_value = None
with pytest.raises(Exception, match='in _complete_vertex_ai_beta') as exc_info:
with pytest.raises(Exception, match='in complete_vertex_ai_beta') as exc_info:
completion(
model="gemini/gemini-1.5-flash",
messages=[{"role": "user", "content": "Test message"}],

View file

@ -550,7 +550,7 @@ def test_e2e_completion_per_user_credential(per_user_credential):
def test_e2e_completion_per_user_session_token_wins_over_caller_authorization(per_user_credential):
"""The session token reaches the wire even when the caller passes their own
Authorization in extra_headers (chat dispatches through _complete_custom_openai)."""
Authorization in extra_headers (chat dispatches through complete_custom_openai)."""
with (
github_http(_exchange_responder()),
_no_authenticator(),

View file

@ -1545,9 +1545,9 @@ def test_system_prompt_only_adds_blank_user_message():
@pytest.fixture
def _cached_vertex_credentials(monkeypatch: pytest.MonkeyPatch) -> str:
from litellm.main import (
from litellm.main import vertex_embedding
from litellm.native.chat_providers.handlers import (
vertex_chat_completion,
vertex_embedding,
vertex_model_garden_chat_completion,
vertex_partner_models_chat_completion,
)

View file

@ -15,6 +15,7 @@ import pytest
import litellm
from litellm._uuid import uuid
from litellm.native.chat_providers.handlers import vertex_partner_models_chat_completion
from litellm.litellm_core_utils.prompt_templates.factory import (
anthropic_messages_pt,
convert_to_anthropic_tool_result,
@ -406,7 +407,7 @@ async def test_vertex_ai_anthropic_async_completion_inlines_https_images_off_the
with (
patch.dict(sys.modules, {"vertexai": vertexai}),
patch.object( # test-quality-ok: litellm.acompletion has no seam for Vertex token minting
litellm.main.vertex_partner_models_chat_completion,
vertex_partner_models_chat_completion,
"_ensure_access_token",
return_value=("token", "test-project"),
),

View file

@ -538,7 +538,7 @@ async def test_vertex_ai_anthropic_prompt_caching_does_not_add_beta_header(
monkeypatch: pytest.MonkeyPatch,
sync_mode: bool,
) -> None:
from litellm.main import vertex_partner_models_chat_completion
from litellm.native.chat_providers.handlers import vertex_partner_models_chat_completion
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
monkeypatch.setitem(

View file

@ -1320,7 +1320,7 @@ def _gemma_cached_access_token():
"""Serve a fake token from the handler's credential cache so no auth round-trip runs."""
from types import SimpleNamespace
from litellm.main import vertex_gemma_chat_completion
from litellm.native.chat_providers.handlers import vertex_gemma_chat_completion
cache = vertex_gemma_chat_completion._credentials_project_mapping
key = (_FAKE_GEMMA_CREDENTIALS, "test")

View file

@ -20,7 +20,7 @@ class TestGetFileContentsFromS3:
"""Test suite for S3 config loading functionality."""
@patch("boto3.client")
@patch("litellm.main.bedrock_converse_chat_completion")
@patch("litellm.native.chat_providers.handlers.bedrock_converse_chat_completion")
@patch("yaml.safe_load")
def test_get_file_contents_from_s3_no_temp_file_creation(
self, mock_yaml_load, mock_bedrock, mock_boto3_client

View file

@ -5243,7 +5243,7 @@ def test_bridged_responses_with_openai_http_handler_keeps_forwarded_headers_out_
def test_aiohttp_openai_warns_only_when_http2_enabled(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, http2_on: bool
):
from litellm.main import base_llm_aiohttp_handler
from litellm.native.chat_providers.handlers import base_llm_aiohttp_handler
monkeypatch.setattr(litellm, "http2", http2_on)
monkeypatch.delenv("LITELLM_HTTP2", raising=False)