跳到内容

pydantic_ai.providers

基类: ABC, Generic[InterfaceClient]

提供程序的抽象类。

提供程序负责为 API 提供经过身份验证的客户端。

每个提供程序仅支持特定的接口。一个接口可以被多个提供程序支持。

例如,OpenAIModel 接口可以被 OpenAIProvider 和 DeepSeekProvider 支持。

源代码位于 pydantic_ai_slim/pydantic_ai/providers/__init__.py
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class Provider(ABC, Generic[InterfaceClient]):
    """Abstract class for a provider.

    The provider is in charge of providing an authenticated client to the API.

    Each provider only supports a specific interface. A interface can be supported by multiple providers.

    For example, the OpenAIModel interface can be supported by the OpenAIProvider and the DeepSeekProvider.
    """

    _client: InterfaceClient

    @property
    @abstractmethod
    def name(self) -> str:
        """The provider name."""
        raise NotImplementedError()

    @property
    @abstractmethod
    def base_url(self) -> str:
        """The base URL for the provider API."""
        raise NotImplementedError()

    @property
    @abstractmethod
    def client(self) -> InterfaceClient:
        """The client for the provider."""
        raise NotImplementedError()

name abstractmethod property

name: str

提供程序名称。

base_url abstractmethod property

base_url: str

提供程序 API 的基本 URL。

client abstractmethod property

client: InterfaceClient

提供程序的客户端。

GoogleVertexProvider

基类: Provider[AsyncClient]

Vertex AI API 的提供程序。

源代码位于 pydantic_ai_slim/pydantic_ai/providers/google_vertex.py
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
class GoogleVertexProvider(Provider[httpx.AsyncClient]):
    """Provider for Vertex AI API."""

    @property
    def name(self) -> str:
        return 'google-vertex'

    @property
    def base_url(self) -> str:
        return (
            f'https://{self.region}-aiplatform.googleapis.com/v1'
            f'/projects/{self.project_id}'
            f'/locations/{self.region}'
            f'/publishers/{self.model_publisher}/models/'
        )

    @property
    def client(self) -> httpx.AsyncClient:
        return self._client

    @overload
    def __init__(
        self,
        *,
        service_account_file: Path | str | None = None,
        project_id: str | None = None,
        region: VertexAiRegion = 'us-central1',
        model_publisher: str = 'google',
        http_client: httpx.AsyncClient | None = None,
    ) -> None: ...

    @overload
    def __init__(
        self,
        *,
        service_account_info: Mapping[str, str] | None = None,
        project_id: str | None = None,
        region: VertexAiRegion = 'us-central1',
        model_publisher: str = 'google',
        http_client: httpx.AsyncClient | None = None,
    ) -> None: ...

    def __init__(
        self,
        *,
        service_account_file: Path | str | None = None,
        service_account_info: Mapping[str, str] | None = None,
        project_id: str | None = None,
        region: VertexAiRegion = 'us-central1',
        model_publisher: str = 'google',
        http_client: httpx.AsyncClient | None = None,
    ) -> None:
        """Create a new Vertex AI provider.

        Args:
            service_account_file: Path to a service account file.
                If not provided, the service_account_info or default environment credentials will be used.
            service_account_info: The loaded service_account_file contents.
                If not provided, the service_account_file or default environment credentials will be used.
            project_id: The project ID to use, if not provided it will be taken from the credentials.
            region: The region to make requests to.
            model_publisher: The model publisher to use, I couldn't find a good list of available publishers,
                and from trial and error it seems non-google models don't work with the `generateContent` and
                `streamGenerateContent` functions, hence only `google` is currently supported.
                Please create an issue or PR if you know how to use other publishers.
            http_client: An existing `httpx.AsyncClient` to use for making HTTP requests.
        """
        if service_account_file and service_account_info:
            raise ValueError('Only one of `service_account_file` or `service_account_info` can be provided.')

        self._client = http_client or cached_async_http_client()
        self.service_account_file = service_account_file
        self.service_account_info = service_account_info
        self.project_id = project_id
        self.region = region
        self.model_publisher = model_publisher

        self._client.auth = _VertexAIAuth(service_account_file, service_account_info, project_id, region)
        self._client.base_url = self.base_url

__init__

__init__(
    *,
    service_account_file: Path | str | None = None,
    project_id: str | None = None,
    region: VertexAiRegion = "us-central1",
    model_publisher: str = "google",
    http_client: AsyncClient | None = None
) -> None
__init__(
    *,
    service_account_info: Mapping[str, str] | None = None,
    project_id: str | None = None,
    region: VertexAiRegion = "us-central1",
    model_publisher: str = "google",
    http_client: AsyncClient | None = None
) -> None
__init__(
    *,
    service_account_file: Path | str | None = None,
    service_account_info: Mapping[str, str] | None = None,
    project_id: str | None = None,
    region: VertexAiRegion = "us-central1",
    model_publisher: str = "google",
    http_client: AsyncClient | None = None
) -> None

创建一个新的 Vertex AI 提供程序。

参数

名称 类型 描述 默认值
service_account_file Path | str | None

服务帐户文件的路径。如果未提供,将使用 service_account_info 或默认环境凭据。

None
service_account_info Mapping[str, str] | None

已加载的 service_account_file 内容。如果未提供,将使用 service_account_file 或默认环境凭据。

None
project_id str | None

要使用的项目 ID,如果未提供,将从凭据中获取。

None
region VertexAiRegion

用于发出请求的区域。

'us-central1'
model_publisher str

要使用的模型发布者,我找不到可用的发布者列表,并且从试错来看,非 Google 模型似乎不适用于 generateContentstreamGenerateContent 函数,因此目前仅支持 google。如果您知道如何使用其他发布者,请创建 issue 或 PR。

'google'
http_client AsyncClient | None

用于发出 HTTP 请求的现有 httpx.AsyncClient

None
源代码位于 pydantic_ai_slim/pydantic_ai/providers/google_vertex.py
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def __init__(
    self,
    *,
    service_account_file: Path | str | None = None,
    service_account_info: Mapping[str, str] | None = None,
    project_id: str | None = None,
    region: VertexAiRegion = 'us-central1',
    model_publisher: str = 'google',
    http_client: httpx.AsyncClient | None = None,
) -> None:
    """Create a new Vertex AI provider.

    Args:
        service_account_file: Path to a service account file.
            If not provided, the service_account_info or default environment credentials will be used.
        service_account_info: The loaded service_account_file contents.
            If not provided, the service_account_file or default environment credentials will be used.
        project_id: The project ID to use, if not provided it will be taken from the credentials.
        region: The region to make requests to.
        model_publisher: The model publisher to use, I couldn't find a good list of available publishers,
            and from trial and error it seems non-google models don't work with the `generateContent` and
            `streamGenerateContent` functions, hence only `google` is currently supported.
            Please create an issue or PR if you know how to use other publishers.
        http_client: An existing `httpx.AsyncClient` to use for making HTTP requests.
    """
    if service_account_file and service_account_info:
        raise ValueError('Only one of `service_account_file` or `service_account_info` can be provided.')

    self._client = http_client or cached_async_http_client()
    self.service_account_file = service_account_file
    self.service_account_info = service_account_info
    self.project_id = project_id
    self.region = region
    self.model_publisher = model_publisher

    self._client.auth = _VertexAIAuth(service_account_file, service_account_info, project_id, region)
    self._client.base_url = self.base_url

OpenAIProvider

基类: Provider[AsyncOpenAI]

OpenAI API 的提供程序。

源代码位于 pydantic_ai_slim/pydantic_ai/providers/openai.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class OpenAIProvider(Provider[AsyncOpenAI]):
    """Provider for OpenAI API."""

    @property
    def name(self) -> str:
        return 'openai'  # pragma: no cover

    @property
    def base_url(self) -> str:
        return self._base_url

    @property
    def client(self) -> AsyncOpenAI:
        return self._client

    def __init__(
        self,
        base_url: str | None = None,
        api_key: str | None = None,
        openai_client: AsyncOpenAI | None = None,
        http_client: httpx.AsyncClient | None = None,
    ) -> None:
        """Create a new OpenAI provider.

        Args:
            base_url: The base url for the OpenAI requests. If not provided, the `OPENAI_BASE_URL` environment variable
                will be used if available. Otherwise, defaults to OpenAI's base url.
            api_key: The API key to use for authentication, if not provided, the `OPENAI_API_KEY` environment variable
                will be used if available.
            openai_client: An existing
                [`AsyncOpenAI`](https://github.com/openai/openai-python?tab=readme-ov-file#async-usage)
                client to use. If provided, `base_url`, `api_key`, and `http_client` must be `None`.
            http_client: An existing `httpx.AsyncClient` to use for making HTTP requests.
        """
        self._base_url = base_url or os.getenv('OPENAI_BASE_URL', 'https://api.openai.com/v1')
        # This is a workaround for the OpenAI client requiring an API key, whilst locally served,
        # openai compatible models do not always need an API key, but a placeholder (non-empty) key is required.
        if api_key is None and 'OPENAI_API_KEY' not in os.environ and openai_client is None:
            api_key = 'api-key-not-set'

        if openai_client is not None:
            assert base_url is None, 'Cannot provide both `openai_client` and `base_url`'
            assert http_client is None, 'Cannot provide both `openai_client` and `http_client`'
            assert api_key is None, 'Cannot provide both `openai_client` and `api_key`'
            self._client = openai_client
        elif http_client is not None:
            self._client = AsyncOpenAI(base_url=self.base_url, api_key=api_key, http_client=http_client)
        else:
            self._client = AsyncOpenAI(base_url=self.base_url, api_key=api_key, http_client=cached_async_http_client())

__init__

__init__(
    base_url: str | None = None,
    api_key: str | None = None,
    openai_client: AsyncOpenAI | None = None,
    http_client: AsyncClient | None = None,
) -> None

创建一个新的 OpenAI 提供程序。

参数

名称 类型 描述 默认值
base_url str | None

OpenAI 请求的基本 url。如果未提供,则如果可用,将使用 OPENAI_BASE_URL 环境变量。否则,默认为 OpenAI 的基本 url。

None
api_key str | None

用于身份验证的 API 密钥,如果未提供,则如果可用,将使用 OPENAI_API_KEY 环境变量。

None
openai_client AsyncOpenAI | None

要使用的现有 AsyncOpenAI 客户端。如果提供,base_urlapi_keyhttp_client 必须为 None

None
http_client AsyncClient | None

用于发出 HTTP 请求的现有 httpx.AsyncClient

None
源代码位于 pydantic_ai_slim/pydantic_ai/providers/openai.py
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def __init__(
    self,
    base_url: str | None = None,
    api_key: str | None = None,
    openai_client: AsyncOpenAI | None = None,
    http_client: httpx.AsyncClient | None = None,
) -> None:
    """Create a new OpenAI provider.

    Args:
        base_url: The base url for the OpenAI requests. If not provided, the `OPENAI_BASE_URL` environment variable
            will be used if available. Otherwise, defaults to OpenAI's base url.
        api_key: The API key to use for authentication, if not provided, the `OPENAI_API_KEY` environment variable
            will be used if available.
        openai_client: An existing
            [`AsyncOpenAI`](https://github.com/openai/openai-python?tab=readme-ov-file#async-usage)
            client to use. If provided, `base_url`, `api_key`, and `http_client` must be `None`.
        http_client: An existing `httpx.AsyncClient` to use for making HTTP requests.
    """
    self._base_url = base_url or os.getenv('OPENAI_BASE_URL', 'https://api.openai.com/v1')
    # This is a workaround for the OpenAI client requiring an API key, whilst locally served,
    # openai compatible models do not always need an API key, but a placeholder (non-empty) key is required.
    if api_key is None and 'OPENAI_API_KEY' not in os.environ and openai_client is None:
        api_key = 'api-key-not-set'

    if openai_client is not None:
        assert base_url is None, 'Cannot provide both `openai_client` and `base_url`'
        assert http_client is None, 'Cannot provide both `openai_client` and `http_client`'
        assert api_key is None, 'Cannot provide both `openai_client` and `api_key`'
        self._client = openai_client
    elif http_client is not None:
        self._client = AsyncOpenAI(base_url=self.base_url, api_key=api_key, http_client=http_client)
    else:
        self._client = AsyncOpenAI(base_url=self.base_url, api_key=api_key, http_client=cached_async_http_client())

DeepSeekProvider

基类: Provider[AsyncOpenAI]

DeepSeek API 的提供程序。

源代码位于 pydantic_ai_slim/pydantic_ai/providers/deepseek.py
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class DeepSeekProvider(Provider[AsyncOpenAI]):
    """Provider for DeepSeek API."""

    @property
    def name(self) -> str:
        return 'deepseek'

    @property
    def base_url(self) -> str:
        return 'https://api.deepseek.com'

    @property
    def client(self) -> AsyncOpenAI:
        return self._client

    @overload
    def __init__(self) -> None: ...

    @overload
    def __init__(self, *, api_key: str) -> None: ...

    @overload
    def __init__(self, *, api_key: str, http_client: AsyncHTTPClient) -> None: ...

    @overload
    def __init__(self, *, openai_client: AsyncOpenAI | None = None) -> None: ...

    def __init__(
        self,
        *,
        api_key: str | None = None,
        openai_client: AsyncOpenAI | None = None,
        http_client: AsyncHTTPClient | None = None,
    ) -> None:
        api_key = api_key or os.getenv('DEEPSEEK_API_KEY')
        if api_key is None and openai_client is None:
            raise ValueError(
                'Set the `DEEPSEEK_API_KEY` environment variable or pass it via `DeepSeekProvider(api_key=...)`'
                'to use the DeepSeek provider.'
            )

        if openai_client is not None:
            self._client = openai_client
        elif http_client is not None:
            self._client = AsyncOpenAI(base_url=self.base_url, api_key=api_key, http_client=http_client)
        else:
            self._client = AsyncOpenAI(base_url=self.base_url, api_key=api_key, http_client=cached_async_http_client())

BedrockProvider

基类: Provider[BaseClient]

AWS Bedrock 的提供程序。

源代码位于 pydantic_ai_slim/pydantic_ai/providers/bedrock.py
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
class BedrockProvider(Provider[BaseClient]):
    """Provider for AWS Bedrock."""

    @property
    def name(self) -> str:
        return 'bedrock'

    @property
    def base_url(self) -> str:
        return self._client.meta.endpoint_url

    @property
    def client(self) -> BaseClient:
        return self._client

    @overload
    def __init__(self, *, bedrock_client: BaseClient) -> None: ...

    @overload
    def __init__(
        self,
        *,
        region_name: str | None = None,
        aws_access_key_id: str | None = None,
        aws_secret_access_key: str | None = None,
        aws_session_token: str | None = None,
    ) -> None: ...

    def __init__(
        self,
        *,
        bedrock_client: BaseClient | None = None,
        region_name: str | None = None,
        aws_access_key_id: str | None = None,
        aws_secret_access_key: str | None = None,
        aws_session_token: str | None = None,
    ) -> None:
        """Initialize the Bedrock provider.

        Args:
            bedrock_client: A boto3 client for Bedrock Runtime. If provided, other arguments are ignored.
            region_name: The AWS region name.
            aws_access_key_id: The AWS access key ID.
            aws_secret_access_key: The AWS secret access key.
            aws_session_token: The AWS session token.
        """
        if bedrock_client is not None:
            self._client = bedrock_client
        else:
            try:
                self._client = boto3.client(  # type: ignore[reportUnknownMemberType]
                    'bedrock-runtime',
                    aws_access_key_id=aws_access_key_id,
                    aws_secret_access_key=aws_secret_access_key,
                    aws_session_token=aws_session_token,
                    region_name=region_name,
                )
            except NoRegionError as exc:  # pragma: no cover
                raise ValueError('You must provide a `region_name` or a boto3 client for Bedrock Runtime.') from exc

__init__

__init__(*, bedrock_client: BaseClient) -> None
__init__(
    *,
    region_name: str | None = None,
    aws_access_key_id: str | None = None,
    aws_secret_access_key: str | None = None,
    aws_session_token: str | None = None
) -> None
__init__(
    *,
    bedrock_client: BaseClient | None = None,
    region_name: str | None = None,
    aws_access_key_id: str | None = None,
    aws_secret_access_key: str | None = None,
    aws_session_token: str | None = None
) -> None

初始化 Bedrock 提供程序。

参数

名称 类型 描述 默认值
bedrock_client BaseClient | None

用于 Bedrock Runtime 的 boto3 客户端。如果提供,则忽略其他参数。

None
region_name str | None

AWS 区域名称。

None
aws_access_key_id str | None

AWS 访问密钥 ID。

None
aws_secret_access_key str | None

AWS secret access key。

None
aws_session_token str | None

AWS 会话令牌。

None
源代码位于 pydantic_ai_slim/pydantic_ai/providers/bedrock.py
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def __init__(
    self,
    *,
    bedrock_client: BaseClient | None = None,
    region_name: str | None = None,
    aws_access_key_id: str | None = None,
    aws_secret_access_key: str | None = None,
    aws_session_token: str | None = None,
) -> None:
    """Initialize the Bedrock provider.

    Args:
        bedrock_client: A boto3 client for Bedrock Runtime. If provided, other arguments are ignored.
        region_name: The AWS region name.
        aws_access_key_id: The AWS access key ID.
        aws_secret_access_key: The AWS secret access key.
        aws_session_token: The AWS session token.
    """
    if bedrock_client is not None:
        self._client = bedrock_client
    else:
        try:
            self._client = boto3.client(  # type: ignore[reportUnknownMemberType]
                'bedrock-runtime',
                aws_access_key_id=aws_access_key_id,
                aws_secret_access_key=aws_secret_access_key,
                aws_session_token=aws_session_token,
                region_name=region_name,
            )
        except NoRegionError as exc:  # pragma: no cover
            raise ValueError('You must provide a `region_name` or a boto3 client for Bedrock Runtime.') from exc

GroqProvider

基类: Provider[AsyncGroq]

Groq API 的提供程序。

源代码位于 pydantic_ai_slim/pydantic_ai/providers/groq.py
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
class GroqProvider(Provider[AsyncGroq]):
    """Provider for Groq API."""

    @property
    def name(self) -> str:
        return 'groq'

    @property
    def base_url(self) -> str:
        return os.environ.get('GROQ_BASE_URL', 'https://api.groq.com')

    @property
    def client(self) -> AsyncGroq:
        return self._client

    @overload
    def __init__(self, *, groq_client: AsyncGroq | None = None) -> None: ...

    @overload
    def __init__(self, *, api_key: str | None = None, http_client: AsyncHTTPClient | None = None) -> None: ...

    def __init__(
        self,
        *,
        api_key: str | None = None,
        groq_client: AsyncGroq | None = None,
        http_client: AsyncHTTPClient | None = None,
    ) -> None:
        """Create a new Groq provider.

        Args:
            api_key: The API key to use for authentication, if not provided, the `GROQ_API_KEY` environment variable
                will be used if available.
            groq_client: An existing
                [`AsyncGroq`](https://github.com/groq/groq-python?tab=readme-ov-file#async-usage)
                client to use. If provided, `api_key` and `http_client` must be `None`.
            http_client: An existing `AsyncHTTPClient` to use for making HTTP requests.
        """
        if groq_client is not None:
            assert http_client is None, 'Cannot provide both `groq_client` and `http_client`'
            assert api_key is None, 'Cannot provide both `groq_client` and `api_key`'
            self._client = groq_client
        else:
            api_key = api_key or os.environ.get('GROQ_API_KEY')

            if api_key is None:
                raise ValueError(
                    'Set the `GROQ_API_KEY` environment variable or pass it via `GroqProvider(api_key=...)`'
                    'to use the Groq provider.'
                )
            elif http_client is not None:
                self._client = AsyncGroq(base_url=self.base_url, api_key=api_key, http_client=http_client)
            else:
                self._client = AsyncGroq(
                    base_url=self.base_url, api_key=api_key, http_client=cached_async_http_client()
                )

__init__

__init__(*, groq_client: AsyncGroq | None = None) -> None
__init__(
    *,
    api_key: str | None = None,
    http_client: AsyncClient | None = None
) -> None
__init__(
    *,
    api_key: str | None = None,
    groq_client: AsyncGroq | None = None,
    http_client: AsyncClient | None = None
) -> None

创建一个新的 Groq 提供程序。

参数

名称 类型 描述 默认值
api_key str | None

用于身份验证的 API 密钥,如果未提供,则如果可用,将使用 GROQ_API_KEY 环境变量。

None
groq_client AsyncGroq | None

要使用的现有 AsyncGroq 客户端。如果提供,api_keyhttp_client 必须为 None

None
http_client AsyncClient | None

用于发出 HTTP 请求的现有 AsyncHTTPClient

None
源代码位于 pydantic_ai_slim/pydantic_ai/providers/groq.py
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    *,
    api_key: str | None = None,
    groq_client: AsyncGroq | None = None,
    http_client: AsyncHTTPClient | None = None,
) -> None:
    """Create a new Groq provider.

    Args:
        api_key: The API key to use for authentication, if not provided, the `GROQ_API_KEY` environment variable
            will be used if available.
        groq_client: An existing
            [`AsyncGroq`](https://github.com/groq/groq-python?tab=readme-ov-file#async-usage)
            client to use. If provided, `api_key` and `http_client` must be `None`.
        http_client: An existing `AsyncHTTPClient` to use for making HTTP requests.
    """
    if groq_client is not None:
        assert http_client is None, 'Cannot provide both `groq_client` and `http_client`'
        assert api_key is None, 'Cannot provide both `groq_client` and `api_key`'
        self._client = groq_client
    else:
        api_key = api_key or os.environ.get('GROQ_API_KEY')

        if api_key is None:
            raise ValueError(
                'Set the `GROQ_API_KEY` environment variable or pass it via `GroqProvider(api_key=...)`'
                'to use the Groq provider.'
            )
        elif http_client is not None:
            self._client = AsyncGroq(base_url=self.base_url, api_key=api_key, http_client=http_client)
        else:
            self._client = AsyncGroq(
                base_url=self.base_url, api_key=api_key, http_client=cached_async_http_client()
            )

AzureProvider

基类: Provider[AsyncOpenAI]

Azure OpenAI API 的提供程序。

请参阅 https://azure.microsoft.com/en-us/products/ai-foundry 了解更多信息。

源代码位于 pydantic_ai_slim/pydantic_ai/providers/azure.py
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class AzureProvider(Provider[AsyncOpenAI]):
    """Provider for Azure OpenAI API.

    See <https://azure.microsoft.com/en-us/products/ai-foundry> for more information.
    """

    @property
    def name(self) -> str:
        return 'azure'

    @property
    def base_url(self) -> str:
        assert self._base_url is not None
        return self._base_url

    @property
    def client(self) -> AsyncOpenAI:
        return self._client

    @overload
    def __init__(self, *, openai_client: AsyncAzureOpenAI) -> None: ...

    @overload
    def __init__(
        self,
        *,
        azure_endpoint: str | None = None,
        api_version: str | None = None,
        api_key: str | None = None,
        http_client: httpx.AsyncClient | None = None,
    ) -> None: ...

    def __init__(
        self,
        *,
        azure_endpoint: str | None = None,
        api_version: str | None = None,
        api_key: str | None = None,
        openai_client: AsyncAzureOpenAI | None = None,
        http_client: httpx.AsyncClient | None = None,
    ) -> None:
        """Create a new Azure provider.

        Args:
            azure_endpoint: The Azure endpoint to use for authentication, if not provided, the `AZURE_OPENAI_ENDPOINT`
                environment variable will be used if available.
            api_version: The API version to use for authentication, if not provided, the `OPENAI_API_VERSION`
                environment variable will be used if available.
            api_key: The API key to use for authentication, if not provided, the `AZURE_OPENAI_API_KEY` environment variable
                will be used if available.
            openai_client: An existing
                [`AsyncAzureOpenAI`](https://github.com/openai/openai-python#microsoft-azure-openai)
                client to use. If provided, `base_url`, `api_key`, and `http_client` must be `None`.
            http_client: An existing `httpx.AsyncClient` to use for making HTTP requests.
        """
        if openai_client is not None:
            assert azure_endpoint is None, 'Cannot provide both `openai_client` and `azure_endpoint`'
            assert http_client is None, 'Cannot provide both `openai_client` and `http_client`'
            assert api_key is None, 'Cannot provide both `openai_client` and `api_key`'
            self._base_url = str(openai_client.base_url)
            self._client = openai_client
        else:
            azure_endpoint = azure_endpoint or os.getenv('AZURE_OPENAI_ENDPOINT')
            if azure_endpoint is None:  # pragma: no cover
                raise ValueError(
                    'Must provide one of the `azure_endpoint` argument or the `AZURE_OPENAI_ENDPOINT` environment variable'
                )

            if api_key is None and 'OPENAI_API_KEY' not in os.environ:  # pragma: no cover
                raise ValueError(
                    'Must provide one of the `api_key` argument or the `OPENAI_API_KEY` environment variable'
                )

            if api_version is None and 'OPENAI_API_VERSION' not in os.environ:  # pragma: no cover
                raise ValueError(
                    'Must provide one of the `api_version` argument or the `OPENAI_API_VERSION` environment variable'
                )

            http_client = http_client or cached_async_http_client()
            self._client = AsyncAzureOpenAI(
                azure_endpoint=azure_endpoint,
                api_key=api_key,
                api_version=api_version,
                http_client=http_client,
            )
            self._base_url = str(self._client.base_url)

__init__

__init__(*, openai_client: AsyncAzureOpenAI) -> None
__init__(
    *,
    azure_endpoint: str | None = None,
    api_version: str | None = None,
    api_key: str | None = None,
    http_client: AsyncClient | None = None
) -> None
__init__(
    *,
    azure_endpoint: str | None = None,
    api_version: str | None = None,
    api_key: str | None = None,
    openai_client: AsyncAzureOpenAI | None = None,
    http_client: AsyncClient | None = None
) -> None

创建一个新的 Azure 提供程序。

参数

名称 类型 描述 默认值
azure_endpoint str | None

用于身份验证的 Azure 终结点,如果未提供,则如果可用,将使用 AZURE_OPENAI_ENDPOINT 环境变量。

None
api_version str | None

用于身份验证的 API 版本,如果未提供,则如果可用,将使用 OPENAI_API_VERSION 环境变量。

None
api_key str | None

用于身份验证的 API 密钥,如果未提供,则如果可用,将使用 AZURE_OPENAI_API_KEY 环境变量。

None
openai_client AsyncAzureOpenAI | None

要使用的现有 AsyncAzureOpenAI 客户端。如果提供,base_urlapi_keyhttp_client 必须为 None

None
http_client AsyncClient | None

用于发出 HTTP 请求的现有 httpx.AsyncClient

None
源代码位于 pydantic_ai_slim/pydantic_ai/providers/azure.py
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def __init__(
    self,
    *,
    azure_endpoint: str | None = None,
    api_version: str | None = None,
    api_key: str | None = None,
    openai_client: AsyncAzureOpenAI | None = None,
    http_client: httpx.AsyncClient | None = None,
) -> None:
    """Create a new Azure provider.

    Args:
        azure_endpoint: The Azure endpoint to use for authentication, if not provided, the `AZURE_OPENAI_ENDPOINT`
            environment variable will be used if available.
        api_version: The API version to use for authentication, if not provided, the `OPENAI_API_VERSION`
            environment variable will be used if available.
        api_key: The API key to use for authentication, if not provided, the `AZURE_OPENAI_API_KEY` environment variable
            will be used if available.
        openai_client: An existing
            [`AsyncAzureOpenAI`](https://github.com/openai/openai-python#microsoft-azure-openai)
            client to use. If provided, `base_url`, `api_key`, and `http_client` must be `None`.
        http_client: An existing `httpx.AsyncClient` to use for making HTTP requests.
    """
    if openai_client is not None:
        assert azure_endpoint is None, 'Cannot provide both `openai_client` and `azure_endpoint`'
        assert http_client is None, 'Cannot provide both `openai_client` and `http_client`'
        assert api_key is None, 'Cannot provide both `openai_client` and `api_key`'
        self._base_url = str(openai_client.base_url)
        self._client = openai_client
    else:
        azure_endpoint = azure_endpoint or os.getenv('AZURE_OPENAI_ENDPOINT')
        if azure_endpoint is None:  # pragma: no cover
            raise ValueError(
                'Must provide one of the `azure_endpoint` argument or the `AZURE_OPENAI_ENDPOINT` environment variable'
            )

        if api_key is None and 'OPENAI_API_KEY' not in os.environ:  # pragma: no cover
            raise ValueError(
                'Must provide one of the `api_key` argument or the `OPENAI_API_KEY` environment variable'
            )

        if api_version is None and 'OPENAI_API_VERSION' not in os.environ:  # pragma: no cover
            raise ValueError(
                'Must provide one of the `api_version` argument or the `OPENAI_API_VERSION` environment variable'
            )

        http_client = http_client or cached_async_http_client()
        self._client = AsyncAzureOpenAI(
            azure_endpoint=azure_endpoint,
            api_key=api_key,
            api_version=api_version,
            http_client=http_client,
        )
        self._base_url = str(self._client.base_url)