`pydantic_ai.models.vertexai`

已弃用

VertexAIModel 已弃用。您可以使用 GoogleVertexProvider 通过 Vertex AI API 进行身份验证，然后使用 GeminiModel 使用 Gemini API。

用于 Gemini 模型的 *-aiplatform.googleapis.com API 的自定义接口。

此模型继承自 GeminiModel，仅更改了 URL 和身份验证方法，它依赖于 VertexAI generateContent 和 streamGenerateContent 函数端点，这些端点与等效的 Gemini 端点具有相同的模式。

设置

有关如何设置此模型的身份验证以及与 GeminiModel 使用的 generativelanguage.googleapis.com API 进行比较的详细信息，请参阅通过 VertexAI 的 Gemini 模型配置。

使用示例

使用已在您的环境中使用“应用程序默认凭据”配置的默认 Google 项目

vertex_example_env.py

from pydantic_ai import Agent
from pydantic_ai.models.vertexai import VertexAIModel

model = VertexAIModel('gemini-1.5-flash')  # (1)!
agent = Agent(model)
result = agent.run_sync('Tell me a joke.')
print(result.data)
#> Did you hear about the toothpaste scandal? They called it Colgate.

VertexAIModel 已弃用，您应该使用 GeminiModel 和 GoogleVertexProvider 代替。

或使用服务帐户 JSON 文件

vertex_example_service_account.py

from pydantic_ai import Agent
from pydantic_ai.models.vertexai import VertexAIModel

model = VertexAIModel(  # (1)!
    'gemini-1.5-flash',
    service_account_file='path/to/service-account.json',
)
agent = Agent(model)
result = agent.run_sync('Tell me a joke.')
print(result.data)
#> Did you hear about the toothpaste scandal? They called it Colgate.

VertexAIModel 已弃用，您应该使用 GeminiModel 和 GoogleVertexProvider 代替。

VERTEX_AI_URL_TEMPLATE `module-attribute`

VERTEX_AI_URL_TEMPLATE = "https://{region}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{region}/publishers/{model_publisher}/models/{model}:"

Vertex AI 的 URL 模板。

有关更多信息，请参阅 generateContent 文档和 streamGenerateContent 文档。

模板按如下方式使用

region 替换为 region 参数，请参阅可用区域
model_publisher 替换为 model_publisher 参数
model 替换为 model_name 参数
project_id 替换为来自 auth/credentials 的 project_id
function（generateContent 或 streamGenerateContent）添加到 URL 的末尾

VertexAIModel `dataclass`

基类：GeminiModel

一个通过 *-aiplatform.googleapis.com VertexAI API 使用 Gemini 的模型。

源代码位于 pydantic_ai_slim/pydantic_ai/models/vertexai.py

@deprecated('Please use `GeminiModel(provider=GoogleVertexProvider(...))` instead.')
@dataclass(init=False)
class VertexAIModel(GeminiModel):
    """A model that uses Gemini via the `*-aiplatform.googleapis.com` VertexAI API."""

    service_account_file: Path | str | None
    project_id: str | None
    region: VertexAiRegion
    model_publisher: Literal['google']
    url_template: str

    _model_name: GeminiModelName = field(repr=False)
    _system: str = field(default='vertex_ai', repr=False)

    # TODO __init__ can be removed once we drop 3.9 and we can set kw_only correctly on the dataclass
    def __init__(
        self,
        model_name: GeminiModelName,
        *,
        service_account_file: Path | str | None = None,
        project_id: str | None = None,
        region: VertexAiRegion = 'us-central1',
        model_publisher: Literal['google'] = 'google',
        http_client: AsyncHTTPClient | None = None,
        url_template: str = VERTEX_AI_URL_TEMPLATE,
    ):
        """Initialize a Vertex AI Gemini model.

        Args:
            model_name: The name of the model to use. I couldn't find a list of supported Google models, in VertexAI
                so for now this uses the same models as the [Gemini model][pydantic_ai.models.gemini.GeminiModel].
            service_account_file: Path to a service account file.
                If not provided, the 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.
            url_template: URL template for Vertex AI, see
                [`VERTEX_AI_URL_TEMPLATE` docs][pydantic_ai.models.vertexai.VERTEX_AI_URL_TEMPLATE]
                for more information.
        """
        self._model_name = model_name
        self.service_account_file = service_account_file
        self.project_id = project_id
        self.region = region
        self.model_publisher = model_publisher
        self.client = http_client or cached_async_http_client()
        self.url_template = url_template

        self._auth = None
        self._url = None
        warnings.warn(
            'VertexAIModel is deprecated, please use `GeminiModel(provider=GoogleVertexProvider(...))` instead.',
            DeprecationWarning,
        )
        self._provider = None

    async def ainit(self) -> None:
        """Initialize the model, setting the URL and auth.

        This will raise an error if authentication fails.
        """
        if self._url is not None and self._auth is not None:
            return

        if self.service_account_file is not None:
            creds: BaseCredentials | ServiceAccountCredentials = _creds_from_file(self.service_account_file)
            assert creds.project_id is None or isinstance(creds.project_id, str)
            creds_project_id: str | None = creds.project_id
            creds_source = 'service account file'
        else:
            creds, creds_project_id = await _async_google_auth()
            creds_source = '`google.auth.default()`'

        if self.project_id is None:
            if creds_project_id is None:
                raise UserError(f'No project_id provided and none found in {creds_source}')
            project_id = creds_project_id
        else:
            project_id = self.project_id

        self._url = self.url_template.format(
            region=self.region,
            project_id=project_id,
            model_publisher=self.model_publisher,
            model=self._model_name,
        )
        self._auth = BearerTokenAuth(creds)

    async def request(
        self,
        messages: list[ModelMessage],
        model_settings: ModelSettings | None,
        model_request_parameters: ModelRequestParameters,
    ) -> tuple[ModelResponse, usage.Usage]:
        await self.ainit()
        return await super().request(messages, model_settings, model_request_parameters)

    @asynccontextmanager
    async def request_stream(
        self,
        messages: list[ModelMessage],
        model_settings: ModelSettings | None,
        model_request_parameters: ModelRequestParameters,
    ) -> AsyncIterator[StreamedResponse]:
        await self.ainit()
        async with super().request_stream(messages, model_settings, model_request_parameters) as value:
            yield value

    @property
    def model_name(self) -> GeminiModelName:
        """The model name."""
        return self._model_name

    @property
    def system(self) -> str:
        """The system / model provider."""
        return self._system

init

__init__(
    model_name: GeminiModelName,
    *,
    service_account_file: Path | str | None = None,
    project_id: str | None = None,
    region: VertexAiRegion = "us-central1",
    model_publisher: Literal["google"] = "google",
    http_client: AsyncClient | None = None,
    url_template: str = VERTEX_AI_URL_TEMPLATE
)

初始化 Vertex AI Gemini 模型。

参数

名称	类型	描述	默认值
`model_name`	`GeminiModelName`	要使用的模型名称。我找不到 VertexAI 中支持的 Google 模型列表，因此目前这使用与 Gemini 模型相同的模型。	必需
`service_account_file`	`Path \| str \| None`	服务帐户文件的路径。如果未提供，将使用默认环境凭据。	`None`
`project_id`	`str \| None`	要使用的项目 ID，如果未提供，将从凭据中获取。	`None`
`region`	`VertexAiRegion`	要向其发出请求的区域。	`'us-central1'`
`model_publisher`	`Literal['google']`	要使用的模型发布者，我找不到可用的发布者的好列表，并且从试错法来看，非 Google 模型似乎不适用于 `generateContent` 和 `streamGenerateContent` 函数，因此目前仅支持 `google`。如果您知道如何使用其他发布者，请创建问题或 PR。	`'google'`
`http_client`	`AsyncClient \| None`	用于发出 HTTP 请求的现有 `httpx.AsyncClient`。	`None`
`url_template`	`str`	Vertex AI 的 URL 模板，有关更多信息，请参阅 `VERTEX_AI_URL_TEMPLATE` 文档。	`VERTEX_AI_URL_TEMPLATE`

源代码位于 pydantic_ai_slim/pydantic_ai/models/vertexai.py

def __init__(
    self,
    model_name: GeminiModelName,
    *,
    service_account_file: Path | str | None = None,
    project_id: str | None = None,
    region: VertexAiRegion = 'us-central1',
    model_publisher: Literal['google'] = 'google',
    http_client: AsyncHTTPClient | None = None,
    url_template: str = VERTEX_AI_URL_TEMPLATE,
):
    """Initialize a Vertex AI Gemini model.

    Args:
        model_name: The name of the model to use. I couldn't find a list of supported Google models, in VertexAI
            so for now this uses the same models as the [Gemini model][pydantic_ai.models.gemini.GeminiModel].
        service_account_file: Path to a service account file.
            If not provided, the 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.
        url_template: URL template for Vertex AI, see
            [`VERTEX_AI_URL_TEMPLATE` docs][pydantic_ai.models.vertexai.VERTEX_AI_URL_TEMPLATE]
            for more information.
    """
    self._model_name = model_name
    self.service_account_file = service_account_file
    self.project_id = project_id
    self.region = region
    self.model_publisher = model_publisher
    self.client = http_client or cached_async_http_client()
    self.url_template = url_template

    self._auth = None
    self._url = None
    warnings.warn(
        'VertexAIModel is deprecated, please use `GeminiModel(provider=GoogleVertexProvider(...))` instead.',
        DeprecationWarning,
    )
    self._provider = None

ainit `async`

ainit() -> None

初始化模型，设置 URL 和身份验证。

如果身份验证失败，这将引发错误。

源代码位于 pydantic_ai_slim/pydantic_ai/models/vertexai.py

async def ainit(self) -> None:
    """Initialize the model, setting the URL and auth.

    This will raise an error if authentication fails.
    """
    if self._url is not None and self._auth is not None:
        return

    if self.service_account_file is not None:
        creds: BaseCredentials | ServiceAccountCredentials = _creds_from_file(self.service_account_file)
        assert creds.project_id is None or isinstance(creds.project_id, str)
        creds_project_id: str | None = creds.project_id
        creds_source = 'service account file'
    else:
        creds, creds_project_id = await _async_google_auth()
        creds_source = '`google.auth.default()`'

    if self.project_id is None:
        if creds_project_id is None:
            raise UserError(f'No project_id provided and none found in {creds_source}')
        project_id = creds_project_id
    else:
        project_id = self.project_id

    self._url = self.url_template.format(
        region=self.region,
        project_id=project_id,
        model_publisher=self.model_publisher,
        model=self._model_name,
    )
    self._auth = BearerTokenAuth(creds)

model_name `property`

model_name: GeminiModelName

模型名称。

system `property`

system: str

系统/模型提供商。

BearerTokenAuth `dataclass`

使用 google-auth 生成的 bearer token 进行身份验证。

源代码位于 pydantic_ai_slim/pydantic_ai/models/vertexai.py

@dataclass
class BearerTokenAuth:
    """Authentication using a bearer token generated by google-auth."""

    credentials: BaseCredentials | ServiceAccountCredentials
    token_created: datetime | None = field(default=None, init=False)

    async def headers(self) -> dict[str, str]:
        if self.credentials.token is None or self._token_expired():
            await run_in_executor(self._refresh_token)
            self.token_created = datetime.now()
        return {'Authorization': f'Bearer {self.credentials.token}'}

    def _token_expired(self) -> bool:
        if self.token_created is None:
            return True
        else:
            return (datetime.now() - self.token_created) > MAX_TOKEN_AGE

    def _refresh_token(self) -> str:
        self.credentials.refresh(Request())
        assert isinstance(self.credentials.token, str), f'Expected token to be a string, got {self.credentials.token}'
        return self.credentials.token

VertexAiRegion `module-attribute`

VertexAiRegion = Literal[
    "asia-east1",
    "asia-east2",
    "asia-northeast1",
    "asia-northeast3",
    "asia-south1",
    "asia-southeast1",
    "australia-southeast1",
    "europe-central2",
    "europe-north1",
    "europe-southwest1",
    "europe-west1",
    "europe-west2",
    "europe-west3",
    "europe-west4",
    "europe-west6",
    "europe-west8",
    "europe-west9",
    "me-central1",
    "me-central2",
    "me-west1",
    "northamerica-northeast1",
    "southamerica-east1",
    "us-central1",
    "us-east1",
    "us-east4",
    "us-east5",
    "us-south1",
    "us-west1",
    "us-west4",
]

Vertex AI 可用的区域。

更多详情请访问此处。

pydantic_ai.models.vertexai

设置

使用示例

VERTEX_AI_URL_TEMPLATE module-attribute

VertexAIModel dataclass

__init__

ainit async

model_name property

system property

BearerTokenAuth dataclass

VertexAiRegion module-attribute

`pydantic_ai.models.vertexai`

VERTEX_AI_URL_TEMPLATE `module-attribute`

VertexAIModel `dataclass`

init

ainit `async`

model_name `property`

system `property`

BearerTokenAuth `dataclass`

VertexAiRegion `module-attribute`