Migrated from Vertex AI SDK to google-genai — Pydantic-native structured output is convenient

The vertexai.generative_models used in my Google Chat Bot RAG pipeline became deprecated, so I migrated to the google-genai SDK. While migrating, I introduced Structured Output and found it convenient to pass Pydantic models directly to response_schema, so I'm sharing that here.

Prerequisites & Environment

• Python 3.14
• google-cloud-aiplatform 1.88.0 (continued use for vertexai.rag)
• google-genai 1.x (newly added)
• Pydantic 2.x
• Cloud Functions 2nd gen (Cloud Run based)

What Became Deprecated

On June 24, 2025, Google deprecated the following modules within the google-cloud-aiplatform package (deletion deadline: June 24, 2026).

• vertexai.generative_models
• vertexai.language_models
• vertexai.vision_models
• vertexai.caching
• vertexai.tuning

The migration target is the google-genai package.

What is NOT deprecated: Modules like vertexai.rag and others not in the list above remain available. If you're using RAG Engine, the retrieval part required no changes.

Migration Key Points

1. GenerativeModel Singleton → genai.Client Singleton

In the old SDK, the model object was stateful and system_instruction was passed in the constructor. In the new SDK, the client object holds only authentication information, and model settings are passed per request.

Before (old SDK):

from vertexai.generative_models import GenerativeModel

model = GenerativeModel(
    "gemini-2.5-flash",
    system_instruction=SYSTEM_PROMPT,
)
response = model.generate_content(prompt)
print(response.text)

After (new SDK):

from google import genai
from google.genai import types

client = genai.Client(
    vertexai=True,
    project="your-project-id",
    location="asia-northeast1",
)
response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=prompt,
    config=types.GenerateContentConfig(
        system_instruction=SYSTEM_PROMPT,
    ),
)
print(response.text)

Key differences:

• vertexai.init() → genai.Client(vertexai=True, project=..., location=...)
• GenerativeModel(model_id, system_instruction=...) → client.models.generate_content(model=..., config=GenerateContentConfig(system_instruction=...))
• Model object singleton → Client object singleton

2. Coexistence with vertexai.rag

vertexai.rag is not subject to deprecation, but it requires initialization via aiplatform.init(). Since genai.Client is initialized separately, both initializations needed to be managed.

from google import genai
from google.cloud import aiplatform
from vertexai import rag

_rag_initialized = False
_genai_client = None
_lock = threading.Lock()

def _ensure_rag_initialized():
    """Initialize aiplatform for vertexai.rag (not deprecated)"""
    global _rag_initialized
    if not _rag_initialized:
        with _lock:
            if not _rag_initialized:
                aiplatform.init(project=PROJECT_ID, location=LOCATION)
                _rag_initialized = True

def _get_genai_client() -> genai.Client:
    """Singleton for genai.Client"""
    global _genai_client
    if _genai_client is None:
        with _lock:
            if _genai_client is None:
                _genai_client = genai.Client(
                    vertexai=True,
                    project=PROJECT_ID,
                    location=LOCATION,
                )
    return _genai_client

For code that doesn't use vertexai.rag (scripts, etc.), aiplatform.init() is not needed.

3. Pydantic-Native Structured Output

This was the most welcome change in the migration. The old SDK's GenerationConfig.response_schema only accepted dictionary-format schemas (with type names like "STRING", "BOOLEAN" in uppercase), which was incompatible with the lowercase standard JSON Schema output by Pydantic's .model_json_schema().

Old SDK — required manually writing dictionary schemas:

from vertexai.generative_models import GenerativeModel, GenerationConfig

# Need to define a dictionary schema separately from the Pydantic model
RAG_ANSWER_SCHEMA = {
    "type": "OBJECT",
    "properties": {
        "answer": {"type": "STRING"},
        "is_answerable": {"type": "BOOLEAN"},
    },
    "required": ["answer", "is_answerable"],
}

response = model.generate_content(
    prompt,
    generation_config=GenerationConfig(
        response_mime_type="application/json",
        response_schema=RAG_ANSWER_SCHEMA,
    ),
)
# Manual JSON parsing
result = RagAnswer.model_validate_json(response.text)

New SDK — can pass Pydantic models directly:

from pydantic import BaseModel, Field
from google import genai
from google.genai import types

class RagAnswer(BaseModel):
    answer: str = Field(description="Answer text (Markdown format)")
    is_answerable: bool = Field(
        description="true if there is relevant information in the knowledge base and the question can be answered, "
        "false if there is no relevant information and the question cannot be answered",
    )

client = genai.Client(vertexai=True, project=PROJECT_ID, location=LOCATION)
response = client.models.generate_content(
    model="gemini-2.5-flash",
    contents=prompt,
    config=types.GenerateContentConfig(
        system_instruction=SYSTEM_PROMPT,
        response_schema=RagAnswer,  # Pass the Pydantic model directly
        response_mime_type="application/json",  # Required when using response_schema
    ),
)

result = response.parsed  # Validated RagAnswer instance
print(result.answer)
print(result.is_answerable)

The type returned by response.parsed is an instance of the Pydantic model passed to response_schema. The SDK handles both JSON parsing and validation, so there's no need to manually call model_validate_json().

4. usage_metadata Field Names Are the Same

The field names in response.usage_metadata used to get token counts are unchanged between the old and new SDKs.

usage = response.usage_metadata
print(usage.prompt_token_count)       # Input token count
print(usage.candidates_token_count)   # Output token count
print(usage.total_token_count)        # Total

Practical Example of Structured Output: RAG Fallback Detection

There was a problem I wanted to solve at the same time as the migration: detecting cases in the RAG pipeline where "there was no relevant information in the knowledge base and the question could not be answered."

Limitations of Keyword Matching

Initially, I instructed via system prompt to "tell the user 'No relevant information was found' if no information is found," and determined this by checking whether the response text contained that phrase.

_FALLBACK_INDICATOR = "該当する情報が見つかりませんでした"

def is_fallback_response(text: str) -> bool:
    return _FALLBACK_INDICATOR in text

However, Gemini paraphrases regardless of prompt instructions.

• "Information and procedures were not found"
• "It is not registered in the knowledge base"
• "Since there is no relevant information in the knowledge base…"

Every time keywords were added, new variations appeared, turning into a cat-and-mouse game.

Resolution with Structured Output

With structured output, you can have Gemini itself classify "whether it could answer" as a boolean.

class RagAnswer(BaseModel):
    answer: str = Field(description="Answer text (Markdown format)")
    is_answerable: bool = Field(
        description="true if there is relevant information in the knowledge base and the question can be answered, "
        "false if there is no relevant information and the question cannot be answered",
    )

The calling side just needs to check result.is_answerable.

result = generate_answer(user_text, contexts)
is_fallback = not result.is_answerable

Since it's a boolean value, there's no variation in expression. Maintaining a keyword list is no longer necessary.

Preparing for Parse Failures

Structured output is generally stable in returning JSON, but as a precaution, a fallback for parse failures is included.

response = client.models.generate_content(
    model=MODEL_ID,
    contents=prompt,
    config=types.GenerateContentConfig(
        system_instruction=SYSTEM_PROMPT,
        response_schema=RagAnswer,
        response_mime_type="application/json",
    ),
)

try:
    if isinstance(response.parsed, RagAnswer):
        return response.parsed
    if response.text is not None:
        return RagAnswer.model_validate_json(response.text)
except (ValueError, TypeError):
    logger.warning("Failed to parse structured response, returning fallback")
return RagAnswer(answer=NO_RESULT_MESSAGE, is_answerable=False)

If response.parsed is not a RagAnswer instance (due to SDK version differences, etc.), manual parsing from response.text is attempted, and if that also fails, a fallback RagAnswer is returned.

Dependency Changes

dependencies = [
    "google-cloud-aiplatform>=1.88.0",  # Maintained for vertexai.rag
    "google-genai>=1.0.0",              # New SDK (replacement for old vertexai.generative_models)
    "pydantic>=2.13.4",                 # Schema definition for structured output
]

google-cloud-aiplatform is needed as long as vertexai.rag continues to be used. Since google-genai does not depend on it internally, both are specified explicitly.

Summary

Migrating from vertexai.generative_models to the google-genai SDK was straightforward. The main changes are the following three:

• Initialization pattern: vertexai.init() + GenerativeModel() → genai.Client(vertexai=True)
• How settings are passed: Constructor → GenerateContentConfig per request
• Structured output: Manually writing dictionary schemas → Pass Pydantic models directly + retrieve with response.parsed

In particular, Pydantic-native structured output not only eliminates the double management of schema definitions, but also provides type-safe access via response.parsed, offering benefits that outweigh the effort of migration.

Coexistence with vertexai.rag is also problem-free — it only requires managing aiplatform.init() and genai.Client separately.

References

• Vertex AI SDK migration guide | Google Cloud
• Google Gen AI SDK documentation
• Structured outputs | Gemini API | Google AI for Developers
• Structured Outputs and Response Schemas | googleapis/python-genai | DeepWiki