mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Azure chat client (#185)
* updated openai, fcc works, with sample * reduced files in openai * Add azure chat client * fix tests * Update python/packages/main/tests/unit/test_openai_chat_completion_base.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/packages/azure/agent_framework/azure/__init__.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * Update python/packages/azure/agent_framework/azure/_azure_openai_settings.py Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> * PR comments * fix bad merge * disable tests for now * actually disable tests for azure * fix tests, align test files with merge changes * update code for new project structure * PR comments * add streaming integration tests. Fix flakiness --------- Co-authored-by: eavanvalkenburg <github@vanvalkenburg.eu> Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
peterychang
committed by
GitHub
Unverified
GitHub
parent
fa96f74ee9
commit
f0dc661c3e
@@ -2,9 +2,16 @@
|
||||
|
||||
import importlib.metadata
|
||||
|
||||
from ._chat_client import AzureChatClient
|
||||
from ._entra_id_authentication import get_entra_auth_token
|
||||
|
||||
try:
|
||||
__version__ = importlib.metadata.version(__name__)
|
||||
except importlib.metadata.PackageNotFoundError:
|
||||
__version__ = "0.0.0" # Fallback for development mode
|
||||
|
||||
__all__ = ["__version__"]
|
||||
__all__ = [
|
||||
"AzureChatClient",
|
||||
"__version__",
|
||||
"get_entra_auth_token",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,232 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import Mapping
|
||||
from copy import deepcopy
|
||||
from typing import Any, TypeVar
|
||||
from uuid import uuid4
|
||||
|
||||
from agent_framework import (
|
||||
ChatFinishReason,
|
||||
ChatResponse,
|
||||
ChatResponseUpdate,
|
||||
FunctionCallContent,
|
||||
FunctionResultContent,
|
||||
TextContent,
|
||||
)
|
||||
from agent_framework.exceptions import ServiceInitializationError
|
||||
from agent_framework.openai import OpenAIModelTypes
|
||||
from agent_framework.openai._chat_client import OpenAIChatClientBase
|
||||
from openai.lib.azure import AsyncAzureADTokenProvider, AsyncAzureOpenAI
|
||||
from openai.types.chat.chat_completion import ChatCompletion, Choice
|
||||
from openai.types.chat.chat_completion_chunk import ChatCompletionChunk
|
||||
from openai.types.chat.chat_completion_chunk import Choice as ChunkChoice
|
||||
from pydantic import SecretStr, ValidationError
|
||||
from pydantic.networks import AnyUrl
|
||||
|
||||
from ._shared import (
|
||||
DEFAULT_AZURE_API_VERSION,
|
||||
DEFAULT_AZURE_TOKEN_ENDPOINT,
|
||||
AzureOpenAIConfigBase,
|
||||
AzureOpenAISettings,
|
||||
)
|
||||
|
||||
logger: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
TChatResponse = TypeVar("TChatResponse", ChatResponse, ChatResponseUpdate)
|
||||
|
||||
|
||||
class AzureChatClient(AzureOpenAIConfigBase, OpenAIChatClientBase):
|
||||
"""Azure Chat completion class."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
api_key: str | None = None,
|
||||
deployment_name: str | None = None,
|
||||
endpoint: str | None = None,
|
||||
base_url: str | None = None,
|
||||
api_version: str | None = None,
|
||||
ad_token: str | None = None,
|
||||
ad_token_provider: AsyncAzureADTokenProvider | None = None,
|
||||
token_endpoint: str | None = None,
|
||||
default_headers: Mapping[str, str] | None = None,
|
||||
async_client: AsyncAzureOpenAI | None = None,
|
||||
env_file_path: str | None = None,
|
||||
env_file_encoding: str | None = None,
|
||||
instruction_role: str | None = None,
|
||||
) -> None:
|
||||
"""Initialize an AzureChatCompletion service.
|
||||
|
||||
Args:
|
||||
api_key (str | None): The optional api key. If provided, will override the value in the
|
||||
env vars or .env file.
|
||||
deployment_name (str | None): The optional deployment. If provided, will override the value
|
||||
(chat_deployment_name) in the env vars or .env file.
|
||||
endpoint (str | None): The optional deployment endpoint. If provided will override the value
|
||||
in the env vars or .env file.
|
||||
base_url (str | None): The optional deployment base_url. If provided will override the value
|
||||
in the env vars or .env file.
|
||||
api_version (str | None): The optional deployment api version. If provided will override the value
|
||||
in the env vars or .env file.
|
||||
ad_token (str | None): The Azure Active Directory token. (Optional)
|
||||
ad_token_provider (AsyncAzureADTokenProvider): The Azure Active Directory token provider. (Optional)
|
||||
token_endpoint (str | None): The token endpoint to request an Azure token. (Optional)
|
||||
default_headers (Mapping[str, str]): The default headers mapping of string keys to
|
||||
string values for HTTP requests. (Optional)
|
||||
async_client (AsyncAzureOpenAI | None): An existing client to use. (Optional)
|
||||
env_file_path (str | None): Use the environment settings file as a fallback to using env vars.
|
||||
env_file_encoding (str | None): The encoding of the environment settings file, defaults to 'utf-8'.
|
||||
instruction_role (str | None): The role to use for 'instruction' messages, for example, summarization
|
||||
prompts could use `developer` or `system`. (Optional)
|
||||
"""
|
||||
try:
|
||||
# Filter out any None values from the arguments
|
||||
azure_openai_settings = AzureOpenAISettings(
|
||||
api_key=SecretStr(api_key) if api_key else None,
|
||||
base_url=AnyUrl(base_url) if base_url else None,
|
||||
endpoint=AnyUrl(endpoint) if endpoint else None,
|
||||
chat_deployment_name=deployment_name,
|
||||
api_version=api_version or DEFAULT_AZURE_API_VERSION,
|
||||
env_file_path=env_file_path,
|
||||
env_file_encoding=env_file_encoding,
|
||||
token_endpoint=token_endpoint or DEFAULT_AZURE_TOKEN_ENDPOINT,
|
||||
)
|
||||
except ValidationError as exc:
|
||||
raise ServiceInitializationError(f"Failed to validate settings: {exc}") from exc
|
||||
|
||||
if not azure_openai_settings.chat_deployment_name:
|
||||
raise ServiceInitializationError("chat_deployment_name is required.")
|
||||
|
||||
super().__init__(
|
||||
deployment_name=azure_openai_settings.chat_deployment_name,
|
||||
endpoint=azure_openai_settings.endpoint,
|
||||
base_url=azure_openai_settings.base_url,
|
||||
api_version=azure_openai_settings.api_version,
|
||||
api_key=azure_openai_settings.api_key.get_secret_value() if azure_openai_settings.api_key else None,
|
||||
ad_token=ad_token,
|
||||
ad_token_provider=ad_token_provider,
|
||||
token_endpoint=azure_openai_settings.token_endpoint,
|
||||
default_headers=default_headers,
|
||||
ai_model_type=OpenAIModelTypes.CHAT,
|
||||
client=async_client,
|
||||
instruction_role=instruction_role,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, settings: dict[str, Any]) -> "AzureChatClient":
|
||||
"""Initialize an Azure OpenAI service from a dictionary of settings.
|
||||
|
||||
Args:
|
||||
settings: A dictionary of settings for the service.
|
||||
should contain keys: service_id, and optionally:
|
||||
ad_auth, ad_token_provider, default_headers
|
||||
"""
|
||||
return AzureChatClient(
|
||||
api_key=settings.get("api_key"),
|
||||
deployment_name=settings.get("deployment_name"),
|
||||
endpoint=settings.get("endpoint"),
|
||||
base_url=settings.get("base_url"),
|
||||
api_version=settings.get("api_version"),
|
||||
ad_token=settings.get("ad_token"),
|
||||
ad_token_provider=settings.get("ad_token_provider"),
|
||||
default_headers=settings.get("default_headers"),
|
||||
env_file_path=settings.get("env_file_path"),
|
||||
)
|
||||
|
||||
def _create_chat_message_content(
|
||||
self, response: ChatCompletion, choice: Choice, response_metadata: dict[str, Any]
|
||||
) -> ChatResponse:
|
||||
"""Create an Azure chat message content object from a choice."""
|
||||
content = super()._create_chat_message_content(response, choice, response_metadata)
|
||||
return self._add_tool_message_to_chat_message_content(content, choice)
|
||||
|
||||
def _create_streaming_chat_message_content(
|
||||
self,
|
||||
chunk: ChatCompletionChunk,
|
||||
choice: ChunkChoice,
|
||||
chunk_metadata: dict[str, Any],
|
||||
) -> ChatResponseUpdate:
|
||||
"""Create an Azure streaming chat message content object from a choice."""
|
||||
content = super()._create_streaming_chat_message_content(chunk, choice, chunk_metadata)
|
||||
assert isinstance(content, ChatResponseUpdate) and isinstance(choice, ChunkChoice) # nosec # noqa: S101
|
||||
return self._add_tool_message_to_chat_message_content(content, choice)
|
||||
|
||||
def _add_tool_message_to_chat_message_content(
|
||||
self,
|
||||
content: TChatResponse,
|
||||
choice: Choice | ChunkChoice,
|
||||
) -> TChatResponse:
|
||||
if tool_message := self._get_tool_message_from_chat_choice(choice=choice):
|
||||
if not isinstance(tool_message, dict):
|
||||
# try to json, to ensure it is a dictionary
|
||||
try:
|
||||
tool_message = json.loads(tool_message)
|
||||
except json.JSONDecodeError:
|
||||
logger.warning("Tool message is not a dictionary, ignore context.")
|
||||
return content
|
||||
function_call = FunctionCallContent(
|
||||
call_id=str(uuid4()),
|
||||
name="Azure-OnYourData",
|
||||
arguments={"query": tool_message.get("intent", [])},
|
||||
)
|
||||
result = FunctionResultContent(
|
||||
call_id=function_call.call_id,
|
||||
result=tool_message["citations"],
|
||||
exception=function_call.exception,
|
||||
additional_properties=function_call.additional_properties,
|
||||
)
|
||||
|
||||
inner_content = content.messages[0].contents if isinstance(content, ChatResponse) else content.contents
|
||||
|
||||
inner_content.insert(0, function_call)
|
||||
inner_content.insert(1, result)
|
||||
return content
|
||||
|
||||
def _get_tool_message_from_chat_choice(self, choice: Choice | ChunkChoice) -> dict[str, Any] | None:
|
||||
"""Get the tool message from a choice."""
|
||||
content = choice.message if isinstance(choice, Choice) else choice.delta
|
||||
# When you enable asynchronous content filtering in Azure OpenAI, you may receive empty deltas
|
||||
if content and content.model_extra is not None:
|
||||
return content.model_extra.get("context", None)
|
||||
# openai allows extra content, so model_extra will be a dict, but we need to check anyway, but no way to test.
|
||||
return None # pragma: no cover
|
||||
|
||||
@staticmethod
|
||||
def split_message(message: "ChatResponse") -> ChatResponse:
|
||||
"""Split an Azure On Your Data response into separate ChatMessages within the ChatResponse.
|
||||
|
||||
If the message does not have three contents, and those three are one each of:
|
||||
FunctionCallContent, FunctionResultContent, and TextContent,
|
||||
it will not return three messages, potentially only one or two.
|
||||
|
||||
The order of the returned messages is as expected by OpenAI.
|
||||
"""
|
||||
if len(message.messages) == 0:
|
||||
return message
|
||||
if len(message.messages[0].contents) != 3:
|
||||
return message
|
||||
messages = {
|
||||
"tool_call": deepcopy(message.messages[0]),
|
||||
"tool_result": deepcopy(message.messages[0]),
|
||||
"assistant": deepcopy(message.messages[0]),
|
||||
}
|
||||
for key, msg in messages.items():
|
||||
if key == "tool_call":
|
||||
msg.contents = [item for item in msg.contents if isinstance(item, FunctionCallContent)]
|
||||
message.finish_reason = ChatFinishReason.TOOL_CALLS
|
||||
if key == "tool_result":
|
||||
msg.contents = [item for item in msg.contents if isinstance(item, FunctionResultContent)]
|
||||
if key == "assistant":
|
||||
msg.contents = [item for item in msg.contents if isinstance(item, TextContent)]
|
||||
|
||||
return ChatResponse(
|
||||
response_id=message.response_id,
|
||||
conversation_id=message.conversation_id,
|
||||
messages=[messages["tool_call"], messages["tool_result"], messages["assistant"]],
|
||||
created_at=message.created_at,
|
||||
model_id=message.ai_model_id,
|
||||
usage_details=message.usage_details,
|
||||
finish_reason=message.finish_reason,
|
||||
additional_properties=message.additional_properties,
|
||||
)
|
||||
@@ -0,0 +1,37 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import logging
|
||||
|
||||
from agent_framework.exceptions import ServiceInvalidAuthError
|
||||
from azure.core.exceptions import ClientAuthenticationError
|
||||
from azure.identity import DefaultAzureCredential
|
||||
|
||||
logger: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def get_entra_auth_token(token_endpoint: str) -> str | None:
|
||||
"""Retrieve a Microsoft Entra Auth Token for a given token endpoint.
|
||||
|
||||
The token endpoint may be specified as an environment variable, via the .env
|
||||
file or as an argument. If the token endpoint is not provided, the default is None.
|
||||
|
||||
Args:
|
||||
token_endpoint: The token endpoint to use to retrieve the authentication token.
|
||||
|
||||
Returns:
|
||||
The Azure token or None if the token could not be retrieved.
|
||||
"""
|
||||
if not token_endpoint:
|
||||
raise ServiceInvalidAuthError(
|
||||
"A token endpoint must be provided either in settings, as an environment variable, or as an argument."
|
||||
)
|
||||
|
||||
credential = DefaultAzureCredential()
|
||||
|
||||
try:
|
||||
auth_token = credential.get_token(token_endpoint)
|
||||
except ClientAuthenticationError:
|
||||
logger.error(f"Failed to retrieve Azure token for the specified endpoint: `{token_endpoint}`.")
|
||||
return None
|
||||
|
||||
return auth_token.token if auth_token else None
|
||||
@@ -0,0 +1,261 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
|
||||
import logging
|
||||
from collections.abc import Awaitable, Callable, Mapping
|
||||
from copy import copy
|
||||
from typing import Any, ClassVar, Final
|
||||
|
||||
from agent_framework import AFBaseSettings, HttpsUrl
|
||||
from agent_framework.exceptions import ServiceInitializationError
|
||||
from agent_framework.openai import OpenAIHandler, OpenAIModelTypes
|
||||
from agent_framework.telemetry import USER_AGENT_KEY
|
||||
from openai.lib.azure import AsyncAzureOpenAI
|
||||
from pydantic import ConfigDict, SecretStr, validate_call
|
||||
|
||||
from ._entra_id_authentication import get_entra_auth_token
|
||||
|
||||
logger: logging.Logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
DEFAULT_AZURE_API_VERSION: Final[str] = "2024-10-21"
|
||||
DEFAULT_AZURE_TOKEN_ENDPOINT: Final[str] = "https://cognitiveservices.azure.com/.default" # noqa: S105
|
||||
|
||||
|
||||
class AzureOpenAISettings(AFBaseSettings):
|
||||
"""AzureOpenAI model settings.
|
||||
|
||||
The settings are first loaded from environment variables with the prefix 'AZURE_OPENAI_'.
|
||||
If the environment variables are not found, the settings can be loaded from a .env file
|
||||
with the encoding 'utf-8'. If the settings are not found in the .env file, the settings
|
||||
are ignored; however, validation will fail alerting that the settings are missing.
|
||||
|
||||
Optional settings for prefix 'AZURE_OPENAI_' are:
|
||||
- chat_deployment_name: str - The name of the Azure Chat deployment. This value
|
||||
will correspond to the custom name you chose for your deployment
|
||||
when you deployed a model. This value can be found under
|
||||
Resource Management > Deployments in the Azure portal or, alternatively,
|
||||
under Management > Deployments in Azure AI Foundry.
|
||||
(Env var AZURE_OPENAI_CHAT_DEPLOYMENT_NAME)
|
||||
- responses_deployment_name: str - The name of the Azure Responses deployment. This value
|
||||
will correspond to the custom name you chose for your deployment
|
||||
when you deployed a model. This value can be found under
|
||||
Resource Management > Deployments in the Azure portal or, alternatively,
|
||||
under Management > Deployments in Azure AI Foundry.
|
||||
(Env var AZURE_OPENAI_RESPONSES_DEPLOYMENT_NAME)
|
||||
- text_deployment_name: str - The name of the Azure Text deployment. This value
|
||||
will correspond to the custom name you chose for your deployment
|
||||
when you deployed a model. This value can be found under
|
||||
Resource Management > Deployments in the Azure portal or, alternatively,
|
||||
under Management > Deployments in Azure AI Foundry.
|
||||
(Env var AZURE_OPENAI_TEXT_DEPLOYMENT_NAME)
|
||||
- embedding_deployment_name: str - The name of the Azure Embedding deployment. This value
|
||||
will correspond to the custom name you chose for your deployment
|
||||
when you deployed a model. This value can be found under
|
||||
Resource Management > Deployments in the Azure portal or, alternatively,
|
||||
under Management > Deployments in Azure AI Foundry.
|
||||
(Env var AZURE_OPENAI_EMBEDDING_DEPLOYMENT_NAME)
|
||||
- text_to_image_deployment_name: str - The name of the Azure Text to Image deployment. This
|
||||
value will correspond to the custom name you chose for your deployment
|
||||
when you deployed a model. This value can be found under
|
||||
Resource Management > Deployments in the Azure portal or, alternatively,
|
||||
under Management > Deployments in Azure AI Foundry.
|
||||
(Env var AZURE_OPENAI_TEXT_TO_IMAGE_DEPLOYMENT_NAME)
|
||||
- audio_to_text_deployment_name: str - The name of the Azure Audio to Text deployment. This
|
||||
value will correspond to the custom name you chose for your deployment
|
||||
when you deployed a model. This value can be found under
|
||||
Resource Management > Deployments in the Azure portal or, alternatively,
|
||||
under Management > Deployments in Azure AI Foundry.
|
||||
(Env var AZURE_OPENAI_AUDIO_TO_TEXT_DEPLOYMENT_NAME)
|
||||
- text_to_audio_deployment_name: str - The name of the Azure Text to Audio deployment. This
|
||||
value will correspond to the custom name you chose for your deployment
|
||||
when you deployed a model. This value can be found under
|
||||
Resource Management > Deployments in the Azure portal or, alternatively,
|
||||
under Management > Deployments in Azure AI Foundry.
|
||||
(Env var AZURE_OPENAI_TEXT_TO_AUDIO_DEPLOYMENT_NAME)
|
||||
- realtime_deployment_name: str - The name of the Azure Realtime deployment. This value
|
||||
will correspond to the custom name you chose for your deployment
|
||||
when you deployed a model. This value can be found under
|
||||
Resource Management > Deployments in the Azure portal or, alternatively,
|
||||
under Management > Deployments in Azure AI Foundry.
|
||||
(Env var AZURE_OPENAI_REALTIME_DEPLOYMENT_NAME)
|
||||
- api_key: SecretStr - The API key for the Azure deployment. This value can be
|
||||
found in the Keys & Endpoint section when examining your resource in
|
||||
the Azure portal. You can use either KEY1 or KEY2.
|
||||
(Env var AZURE_OPENAI_API_KEY)
|
||||
- base_url: HttpsUrl | None - base_url: The url of the Azure deployment. This value
|
||||
can be found in the Keys & Endpoint section when examining
|
||||
your resource from the Azure portal, the base_url consists of the endpoint,
|
||||
followed by /openai/deployments/{deployment_name}/,
|
||||
use endpoint if you only want to supply the endpoint.
|
||||
(Env var AZURE_OPENAI_BASE_URL)
|
||||
- endpoint: HttpsUrl - The endpoint of the Azure deployment. This value
|
||||
can be found in the Keys & Endpoint section when examining
|
||||
your resource from the Azure portal, the endpoint should end in openai.azure.com.
|
||||
If both base_url and endpoint are supplied, base_url will be used.
|
||||
(Env var AZURE_OPENAI_ENDPOINT)
|
||||
- api_version: str | None - The API version to use. The default value is "2024-02-01".
|
||||
(Env var AZURE_OPENAI_API_VERSION)
|
||||
- token_endpoint: str - The token endpoint to use to retrieve the authentication token.
|
||||
The default value is "https://cognitiveservices.azure.com/.default".
|
||||
(Env var AZURE_OPENAI_TOKEN_ENDPOINT)
|
||||
"""
|
||||
|
||||
env_prefix: ClassVar[str] = "AZURE_OPENAI_"
|
||||
|
||||
chat_deployment_name: str | None = None
|
||||
responses_deployment_name: str | None = None
|
||||
text_deployment_name: str | None = None
|
||||
embedding_deployment_name: str | None = None
|
||||
text_to_image_deployment_name: str | None = None
|
||||
audio_to_text_deployment_name: str | None = None
|
||||
text_to_audio_deployment_name: str | None = None
|
||||
realtime_deployment_name: str | None = None
|
||||
endpoint: HttpsUrl | None = None
|
||||
base_url: HttpsUrl | None = None
|
||||
api_key: SecretStr | None = None
|
||||
api_version: str = DEFAULT_AZURE_API_VERSION
|
||||
token_endpoint: str = DEFAULT_AZURE_TOKEN_ENDPOINT
|
||||
|
||||
def get_azure_openai_auth_token(self, token_endpoint: str | None = None) -> str | None:
|
||||
"""Retrieve a Microsoft Entra Auth Token for a given token endpoint for the use with Azure OpenAI.
|
||||
|
||||
The required role for the token is `Cognitive Services OpenAI Contributor`.
|
||||
The token endpoint may be specified as an environment variable, via the .env
|
||||
file or as an argument. If the token endpoint is not provided, the default is None.
|
||||
The `token_endpoint` argument takes precedence over the `token_endpoint` attribute.
|
||||
|
||||
Args:
|
||||
token_endpoint: The token endpoint to use. Defaults to `https://cognitiveservices.azure.com/.default`.
|
||||
|
||||
Returns:
|
||||
The Azure token or None if the token could not be retrieved.
|
||||
|
||||
Raises:
|
||||
ServiceInitializationError: If the token endpoint is not provided.
|
||||
"""
|
||||
endpoint_to_use = token_endpoint or self.token_endpoint
|
||||
if endpoint_to_use is None: # type: ignore
|
||||
raise ServiceInitializationError("Please provide a token endpoint to retrieve the authentication token.")
|
||||
return get_entra_auth_token(endpoint_to_use)
|
||||
|
||||
|
||||
class AzureOpenAIConfigBase(OpenAIHandler):
|
||||
"""Internal class for configuring a connection to an Azure OpenAI service."""
|
||||
|
||||
@validate_call(config=ConfigDict(arbitrary_types_allowed=True))
|
||||
def __init__(
|
||||
self,
|
||||
deployment_name: str,
|
||||
ai_model_type: OpenAIModelTypes,
|
||||
endpoint: HttpsUrl | None = None,
|
||||
base_url: HttpsUrl | None = None,
|
||||
api_version: str = DEFAULT_AZURE_API_VERSION,
|
||||
api_key: str | None = None,
|
||||
ad_token: str | None = None,
|
||||
ad_token_provider: Callable[[], str | Awaitable[str]] | None = None,
|
||||
token_endpoint: str | None = None,
|
||||
default_headers: Mapping[str, str] | None = None,
|
||||
client: AsyncAzureOpenAI | None = None,
|
||||
instruction_role: str | None = None,
|
||||
**kwargs: Any,
|
||||
) -> None:
|
||||
"""Internal class for configuring a connection to an Azure OpenAI service.
|
||||
|
||||
The `validate_call` decorator is used with a configuration that allows arbitrary types.
|
||||
This is necessary for types like `HttpsUrl` and `OpenAIModelTypes`.
|
||||
|
||||
Args:
|
||||
deployment_name (str): Name of the deployment.
|
||||
ai_model_type (OpenAIModelTypes): The type of OpenAI model to deploy.
|
||||
endpoint (HttpsUrl): The specific endpoint URL for the deployment. (Optional)
|
||||
base_url (Url): The base URL for Azure services. (Optional)
|
||||
api_version (str): Azure API version. Defaults to the defined DEFAULT_AZURE_API_VERSION.
|
||||
api_key (str): API key for Azure services. (Optional)
|
||||
ad_token (str): Azure AD token for authentication. (Optional)
|
||||
ad_token_provider (Callable[[], Union[str, Awaitable[str]]]): A callable
|
||||
or coroutine function providing Azure AD tokens. (Optional)
|
||||
token_endpoint (str): Azure AD token endpoint use to get the token. (Optional)
|
||||
default_headers (Union[Mapping[str, str], None]): Default headers for HTTP requests. (Optional)
|
||||
client (AsyncAzureOpenAI): An existing client to use. (Optional)
|
||||
instruction_role (str | None): The role to use for 'instruction' messages, for example, summarization
|
||||
prompts could use `developer` or `system`. (Optional)
|
||||
kwargs: Additional keyword arguments.
|
||||
|
||||
"""
|
||||
# Merge APP_INFO into the headers if it exists
|
||||
merged_headers = dict(copy(default_headers)) if default_headers else {}
|
||||
|
||||
if not client:
|
||||
# If the client is None, the api_key is none, the ad_token is none, and the ad_token_provider is none,
|
||||
# then we will attempt to get the ad_token using the default endpoint specified in the Azure OpenAI
|
||||
# settings.
|
||||
if not api_key and not ad_token_provider and not ad_token and token_endpoint:
|
||||
ad_token = get_entra_auth_token(token_endpoint)
|
||||
|
||||
if not api_key and not ad_token and not ad_token_provider:
|
||||
raise ServiceInitializationError(
|
||||
"Please provide either api_key, ad_token or ad_token_provider or a client."
|
||||
)
|
||||
|
||||
if not endpoint and not base_url:
|
||||
raise ServiceInitializationError("Please provide an endpoint or a base_url")
|
||||
|
||||
args: dict[str, Any] = {
|
||||
"default_headers": merged_headers,
|
||||
}
|
||||
if api_version:
|
||||
args["api_version"] = api_version
|
||||
if ad_token:
|
||||
args["azure_ad_token"] = ad_token
|
||||
if ad_token_provider:
|
||||
args["azure_ad_token_provider"] = ad_token_provider
|
||||
if api_key:
|
||||
args["api_key"] = api_key
|
||||
if base_url:
|
||||
args["base_url"] = str(base_url)
|
||||
if endpoint and not base_url:
|
||||
args["azure_endpoint"] = str(endpoint)
|
||||
# TODO (eavanvalkenburg): Remove the check on model type when the package fixes: https://github.com/openai/openai-python/issues/2120
|
||||
if deployment_name and ai_model_type != OpenAIModelTypes.REALTIME:
|
||||
args["azure_deployment"] = deployment_name
|
||||
|
||||
if "websocket_base_url" in kwargs:
|
||||
args["websocket_base_url"] = kwargs.pop("websocket_base_url")
|
||||
|
||||
client = AsyncAzureOpenAI(**args)
|
||||
args = {
|
||||
"ai_model_id": deployment_name,
|
||||
"client": client,
|
||||
"ai_model_type": ai_model_type,
|
||||
}
|
||||
if instruction_role:
|
||||
args["instruction_role"] = instruction_role
|
||||
super().__init__(**args, **kwargs)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Convert the configuration to a dictionary."""
|
||||
client_settings = {
|
||||
"base_url": str(self.client.base_url),
|
||||
"api_version": self.client._custom_query["api-version"], # type: ignore
|
||||
"api_key": self.client.api_key,
|
||||
"ad_token": getattr(self.client, "_azure_ad_token", None),
|
||||
"ad_token_provider": getattr(self.client, "_azure_ad_token_provider", None),
|
||||
"default_headers": {k: v for k, v in self.client.default_headers.items() if k != USER_AGENT_KEY},
|
||||
}
|
||||
base = self.model_dump(
|
||||
exclude={
|
||||
"prompt_tokens",
|
||||
"completion_tokens",
|
||||
"total_tokens",
|
||||
"api_type",
|
||||
"org_id",
|
||||
"ai_model_type",
|
||||
"service_id",
|
||||
"client",
|
||||
},
|
||||
by_alias=True,
|
||||
exclude_none=True,
|
||||
)
|
||||
base.update(client_settings)
|
||||
return base
|
||||
Reference in New Issue
Block a user