Vertex AI 结构化输出，使用 Instructor 的完整指南¶

Google Cloud 的 Vertex AI 提供企业级 AI 能力，具有强大的扩展性和安全功能。本指南将展示如何使用 Instructor 与 Vertex AI 结合，实现类型安全、经验证的响应。

快速入门¶

安装支持 Vertex AI 的 Instructor。您可以通过运行以下命令来完成。

pip install "instructor[vertexai]"

简单用户示例 (同步)¶

import instructor
import vertexai  # type: ignore
from vertexai.generative_models import GenerativeModel  # type: ignore
from pydantic import BaseModel

vertexai.init()


class User(BaseModel):
    name: str
    age: int


client = instructor.from_vertexai(
    client=GenerativeModel("gemini-1.5-pro-preview-0409"),
    mode=instructor.Mode.VERTEXAI_TOOLS,
)

# note that client.chat.completions.create will also work
resp = client.create(
    messages=[
        {
            "role": "user",
            "content": "Extract Jason is 25 years old.",
        }
    ],
    response_model=User,
)

print(resp)
#> User(name='Jason', age=25)

简单用户示例 (异步)¶

import asyncio
import instructor
import vertexai  # type: ignore
from vertexai.generative_models import GenerativeModel  # type: ignore
from pydantic import BaseModel

vertexai.init()


class User(BaseModel):
    name: str
    age: int


client = instructor.from_vertexai(
    client=GenerativeModel("gemini-1.5-pro-preview-0409"),
    mode=instructor.Mode.VERTEXAI_TOOLS,
    _async=True,
)

async def extract_user():
    user = await client.create(
        messages=[
            {
                "role": "user",
                "content": "Extract Jason is 25 years old.",
            }
        ],
        response_model=User,
    )
    return user


# Run async function
user = asyncio.run(extract_user())
print(user)  # User(name='Jason', age=25)

流式支持¶

Instructor 现在支持 Vertex AI 的流式传输功能！您可以使用 create_partial 进行增量模型构建，以及使用 create_iterable 进行集合的流式传输。

流式部分响应¶

import vertexai  # type: ignore
from vertexai.generative_models import GenerativeModel  # type: ignore
import instructor
from pydantic import BaseModel
from instructor.dsl.partial import Partial

vertexai.init()

class UserExtract(BaseModel):
    name: str
    age: int

client = instructor.from_vertexai(
    client=GenerativeModel("gemini-2.0-flash"),
    mode=instructor.Mode.VERTEXAI_TOOLS,
)

# Stream partial responses
response_stream = client.chat.completions.create(
    response_model=Partial[UserExtract],
    stream=True,
    messages=[
        {"role": "user", "content": "Anibal is 23 years old"},
    ],
)

for partial_user in response_stream:
    print(f"Received update: {partial_user}")
# Output might show:
# Received update: UserExtract(name='Anibal', age=None)
# Received update: UserExtract(name='Anibal', age=23)

流式可迭代集合¶

import vertexai  # type: ignore
from vertexai.generative_models import GenerativeModel  # type: ignore
import instructor
from pydantic import BaseModel

vertexai.init()

class UserExtract(BaseModel):
    name: str
    age: int

client = instructor.from_vertexai(
    client=GenerativeModel("gemini-2.0-flash"),
    mode=instructor.Mode.VERTEXAI_TOOLS,
)

# Stream iterable responses
response_stream = client.chat.completions.create_iterable(
    response_model=UserExtract,
    messages=[
        {"role": "user", "content": "Make up two people"},
    ],
)

for user in response_stream:
    print(f"Generated user: {user}")
# Output:
# Generated user: UserExtract(name='Sarah Johnson', age=32)
# Generated user: UserExtract(name='David Chen', age=27)

异步流式¶

您也可以使用这两种流式传输方法的异步版本

import asyncio
import vertexai  # type: ignore
from vertexai.generative_models import GenerativeModel  # type: ignore
import instructor
from pydantic import BaseModel
from instructor.dsl.partial import Partial

vertexai.init()

class UserExtract(BaseModel):
    name: str
    age: int

client = instructor.from_vertexai(
    client=GenerativeModel("gemini-2.0-flash"),
    mode=instructor.Mode.VERTEXAI_TOOLS,
    _async=True,
)

async def stream_partial():
    response_stream = await client.chat.completions.create(
        response_model=Partial[UserExtract],
        stream=True,
        messages=[
            {"role": "user", "content": "Anibal is 23 years old"},
        ],
    )

    async for partial_user in response_stream:
        print(f"Received update: {partial_user}")

async def stream_iterable():
    response_stream = client.chat.completions.create_iterable(
        response_model=UserExtract,
        messages=[
            {"role": "user", "content": "Make up two people"},
        ],
    )

    async for user in response_stream:
        print(f"Generated user: {user}")

# Run async functions
asyncio.run(stream_partial())
asyncio.run(stream_iterable())

更新与兼容性¶

Instructor 与 Vertex AI 的最新 API 版本保持兼容。查看更新日志获取更新。

现已添加对部分响应和可迭代集合的流式支持，同时提供同步和异步接口。