fix: add proper type annotations for embedding() function (#12262)

Fixes #9526

The embedding() function can return either an EmbeddingResponse or a
Coroutine[Any, Any, EmbeddingResponse] depending on whether aembedding=True
is passed. This adds @overload decorators to properly indicate the return
type based on the aembedding parameter.

This resolves type checking errors when using embedding() with aembedding=True
where the type checker couldn't determine that a coroutine was being returned.
This commit is contained in:
Cole McIntosh 2025-07-05 16:54:22 -06:00 committed by GitHub
parent 2812ffc0c8
commit bbc3f74833
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -2870,9 +2870,9 @@ def completion( # type: ignore # noqa: PLR0915
"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
)
optional_params[
"aws_region_name"
] = aws_bedrock_client.meta.region_name
bedrock_route = BedrockModelInfo.get_bedrock_route(model)
if bedrock_route == "converse":
@ -3498,6 +3498,82 @@ async def aembedding(*args, **kwargs) -> EmbeddingResponse:
)
# Overload for when aembedding=True (returns coroutine)
@overload
def embedding(
model,
input=[],
# Optional params
dimensions: Optional[int] = None,
encoding_format: Optional[str] = None,
timeout=600, # default to 10 minutes
# set api_base, api_version, api_key
api_base: Optional[str] = None,
api_version: Optional[str] = None,
api_key: Optional[str] = None,
api_type: Optional[str] = None,
caching: bool = False,
user: Optional[str] = None,
custom_llm_provider=None,
litellm_call_id=None,
logger_fn=None,
*,
aembedding: Literal[True],
**kwargs,
) -> Coroutine[Any, Any, EmbeddingResponse]:
...
# Overload for when aembedding=False or not specified (returns EmbeddingResponse)
@overload
def embedding(
model,
input=[],
# Optional params
dimensions: Optional[int] = None,
encoding_format: Optional[str] = None,
timeout=600, # default to 10 minutes
# set api_base, api_version, api_key
api_base: Optional[str] = None,
api_version: Optional[str] = None,
api_key: Optional[str] = None,
api_type: Optional[str] = None,
caching: bool = False,
user: Optional[str] = None,
custom_llm_provider=None,
litellm_call_id=None,
logger_fn=None,
*,
aembedding: Literal[False] = False,
**kwargs,
) -> EmbeddingResponse:
...
# Overload for when aembedding is not specified at all (returns EmbeddingResponse)
@overload
def embedding(
model,
input=[],
# Optional params
dimensions: Optional[int] = None,
encoding_format: Optional[str] = None,
timeout=600, # default to 10 minutes
# set api_base, api_version, api_key
api_base: Optional[str] = None,
api_version: Optional[str] = None,
api_key: Optional[str] = None,
api_type: Optional[str] = None,
caching: bool = False,
user: Optional[str] = None,
custom_llm_provider=None,
litellm_call_id=None,
logger_fn=None,
**kwargs,
) -> EmbeddingResponse:
...
@client
def embedding( # noqa: PLR0915
model,
@ -4651,9 +4727,9 @@ def adapter_completion(
new_kwargs = translation_obj.translate_completion_input_params(kwargs=kwargs)
response: Union[ModelResponse, CustomStreamWrapper] = completion(**new_kwargs) # type: ignore
translated_response: Optional[Union[BaseModel, AdapterCompletionStreamWrapper]] = (
None
)
translated_response: Optional[
Union[BaseModel, AdapterCompletionStreamWrapper]
] = None
if isinstance(response, ModelResponse):
translated_response = translation_obj.translate_completion_output_params(
response=response
@ -5644,9 +5720,9 @@ def stream_chunk_builder( # noqa: PLR0915
]
if len(content_chunks) > 0:
response["choices"][0]["message"]["content"] = (
processor.get_combined_content(content_chunks)
)
response["choices"][0]["message"][
"content"
] = processor.get_combined_content(content_chunks)
reasoning_chunks = [
chunk
@ -5657,9 +5733,9 @@ def stream_chunk_builder( # noqa: PLR0915
]
if len(reasoning_chunks) > 0:
response["choices"][0]["message"]["reasoning_content"] = (
processor.get_combined_reasoning_content(reasoning_chunks)
)
response["choices"][0]["message"][
"reasoning_content"
] = processor.get_combined_reasoning_content(reasoning_chunks)
audio_chunks = [
chunk