Bootstrapping a Production AI Travel Agent with Gemini Enterprise Agent Platform

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Offvia Travel Concierge -- Schema-Constrained Intent Parser (Part 1)
Bootstrapped on Google Cloud Gemini Enterprise Agent Platform Express Mode.
"""

from __future__ import annotations

import os
import sys
import logging
from datetime import date, datetime
from time import perf_counter
from uuid import uuid4
from zoneinfo import ZoneInfo
from typing import List, Optional

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

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("offvia.intent_parser")

MODEL_ID = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
REFERENCE_TIMEZONE = os.getenv("OFFVIA_TIMEZONE", "Europe/Rome")


class HotelCriteria(BaseModel):
    preferred_area: Optional[str] = Field(default=None, description="Neighborhood, e.g. Shibuya")
    min_star_rating: Optional[float] = Field(default=None, ge=1.0, le=5.0)
    required_amenities: List[str] = Field(default_factory=list)
    free_cancellation_required: bool = Field(default=False)


class TripIntent(BaseModel):
    # Separate raw traveler inputs from authoritative domain resolution
    origin_city: str = Field(description="Origin city, e.g. San Francisco")
    destination_city: str = Field(description="Destination city, e.g. Tokyo")
    supplied_origin_iata: Optional[str] = Field(
        default=None, 
        description="Populate ONLY if the traveler explicitly specified a 3-letter IATA code (e.g. SFO)"
    )
    airport_resolution_required: bool = Field(
        default=True, 
        description="True when downstream tools must resolve city to airport codes (e.g. Tokyo -> HND/NRT)"
    )
    
    trip_duration_days: int = Field(ge=1, le=60)
    
    # Preserve original wording for auditing alongside normalized ISO date
    departure_date_expression: Optional[str] = Field(default=None, description="Raw expression, e.g. 'next Wednesday'")
    departure_date: Optional[str] = Field(default=None, description="Resolved date in YYYY-MM-DD format")
    
    budget_limit: float = Field(gt=0)
    budget_currency: str = Field(description="3-letter currency code, e.g. USD")
    non_stop_required: bool = Field(default=False)
    
    hotel_preferences: HotelCriteria
    special_constraints: List[str] = Field(default_factory=list)
    unresolved_questions: List[str] = Field(default_factory=list)

    @field_validator("budget_currency")
    @classmethod
    def validate_currency(cls, v: str) -> str:
        code = v.strip().upper()
        if len(code) != 3 or not code.isalpha():
            raise ValueError("budget_currency must be a 3-letter code")
        return code

    @field_validator("departure_date")
    @classmethod
    def validate_date(cls, v: Optional[str]) -> Optional[str]:
        if v:
            date.fromisoformat(v)
        return v


def build_client() -> genai.Client:
    api_key = os.getenv("GEMINI_API_KEY")
    project_id = os.getenv("GOOGLE_CLOUD_PROJECT")

    # Express Mode: Rapid local prototyping via API key
    if api_key:
        logger.info("Initializing Google GenAI client in Express Mode (API Key)")
        return genai.Client(
            api_key=api_key,
            http_options=types.HttpOptions(
                retry_options=types.HttpRetryOptions(attempts=4, initial_delay=1.0, max_delay=8.0)
            ),
        )

    # Enterprise Mode: Production runtime with Workload Identity / ADC
    if project_id:
        logger.info("Initializing Google GenAI client in Enterprise Mode (ADC / Workload Identity)")
        return genai.Client(
            enterprise=True,
            project=project_id,
            location=os.getenv("GOOGLE_CLOUD_LOCATION", "global"),
            http_options=types.HttpOptions(
                retry_options=types.HttpRetryOptions(attempts=4, initial_delay=1.0, max_delay=8.0)
            ),
        )

    raise RuntimeError(
        "Neither GEMINI_API_KEY nor GOOGLE_CLOUD_PROJECT is configured. "
        "Export GEMINI_API_KEY for Express Mode or authenticate via ADC."
    )


def parse_traveler_request(client: genai.Client, request_text: str, ref_date: date, tz: str) -> TripIntent:
    system_instruction = (
        "You are the intent ingestion component for Offvia travel platform. "
        "Extract only facts stated by the traveler. If the user states a city rather than an airport code, "
        "leave supplied_origin_iata as null and set airport_resolution_required to true. "
        "Capture raw date phrasing in departure_date_expression and normalized date in departure_date."
    )
    prompt = f"Reference Date: {ref_date.isoformat()}\nTimezone: {tz}\n\nQuery:\n{request_text}"

    response = client.models.generate_content(
        model=MODEL_ID,
        contents=prompt,
        config=types.GenerateContentConfig(
            system_instruction=system_instruction,
            response_mime_type="application/json",
            response_schema=TripIntent,
        ),
    )
    return TripIntent.model_validate_json(response.text)


if __name__ == "__main__":
    req_id = str(uuid4())[:8]
    sample_request = (
        "Find me a four-day tech-conference trip to Tokyo from San Francisco for no more than "
        "$2,200, departing next Wednesday. I need a non-stop flight, a four-star hotel near "
        "Shibuya with dedicated high-speed Wi-Fi, and free cancellation in case my visa decision is delayed."
    )

    ref_date = datetime.now(ZoneInfo(REFERENCE_TIMEZONE)).date()
    logger.info("Processing request [%s] via model [%s]", req_id, MODEL_ID)

    client = build_client()
    start = perf_counter()

    try:
        intent = parse_traveler_request(client, sample_request, ref_date, REFERENCE_TIMEZONE)
        elapsed_ms = (perf_counter() - start) * 1000
        logger.info("Request [%s] parsed in %.0f ms", req_id, elapsed_ms)
        print("\n=== Offvia Parsed Travel Contract ===")
        print(intent.model_dump_json(indent=2))
    except ValidationError as e:
        logger.error("Schema validation failed: %s", e)
        sys.exit(1)
    finally:
        client.close()

{
  "origin_city": "San Francisco",
  "destination_city": "Tokyo",
  "supplied_origin_iata": null,
  "airport_resolution_required": true,
  "trip_duration_days": 4,
  "departure_date_expression": "next Wednesday",
  "departure_date": "2026-09-30",
  "budget_limit": 2200.0,
  "budget_currency": "USD",
  "non_stop_required": true,
  "hotel_preferences": {
    "preferred_area": "Shibuya",
    "min_star_rating": 4.0,
    "required_amenities": [
      "dedicated high-speed Wi-Fi"
    ],
    "free_cancellation_required": true
  },
  "special_constraints": [
    "Japanese visa decision might be delayed"
  ],
  "unresolved_questions": []
}