通过 Private Service Connect 端点使用 Python SDK 访问 Gemini 3 Pro 对话

1. 概览

Vertex AI API 可以通过互联网访问,但在企业中,您可能希望以私密方式访问 Vertex AI API,而无需通过互联网。在本实验中,您将首先执行以下操作。

  • 通过 Vertex 使用 Python SDK 访问 Gemini 3 Pro API
  • 此脚本将在虚拟机实例上运行
  • 连接将通过 Cloud NAT 连接到公共互联网。

然后,您将创建指向 Google API 的 Private Service Connect 端点,并更改流量流以使用专用端点连接到 Gemini Chat API。配置将是 Terraform、gcloud 和控制台的组合。

在本实验中,您将创建以下模式。

图 1.

304c579d349aec90.png

2. 目标

在本实验中,您将学习如何执行以下任务:

  • 设置虚拟机实例以使用 Python SDK
  • 通过 Python 脚本连接到 Gemini Chat
  • 配置 PSC 端点以连接到 Googleapis
  • 验证与 Googleais 的连接路径
  • 配置手动 DNS 条目

自定进度的环境设置

  1. 登录 Google Cloud 控制台,然后创建一个新项目或重复使用现有项目。如果您还没有 Gmail 或 Google Workspace 账号,则必须创建一个

295004821bab6a87.png

37d264871000675d.png

96d86d3d5655cdbe.png

  • 项目名称是此项目参与者的显示名称。它是 Google API 尚未使用的字符串。您可以随时对其进行更新。
  • 项目 ID 在所有 Google Cloud 项目中是唯一的,并且是不可变的(一经设置便无法更改)。Cloud 控制台会自动生成一个唯一字符串;通常情况下,您无需关注该字符串。在大多数 Codelab 中,您都需要引用项目 ID(通常用 PROJECT_ID 标识)。如果您不喜欢生成的 ID,可以再随机生成一个 ID。或者,您也可以尝试自己的项目 ID,看看是否可用。完成此步骤后便无法更改该 ID,并且此 ID 在项目期间会一直保留。
  • 此外,还有第三个值,即部分 API 使用的项目编号,供您参考。如需详细了解所有这三个值,请参阅文档
  1. 接下来,您需要在 Cloud 控制台中启用结算功能,以便使用 Cloud 资源/API。运行此 Codelab 应该不会产生太多的费用(如果有的话)。若要关闭资源以避免产生超出本教程范围的结算费用,您可以删除自己创建的资源或删除项目。Google Cloud 新用户符合参与 300 美元免费试用计划的条件。

启动 Cloud Shell

虽然可以通过笔记本电脑对 Google Cloud 进行远程操作,但在此 Codelab 中,您将使用 Google Cloud Shell,这是一个在云端运行的命令行环境。

Google Cloud 控制台 中,点击右上角工具栏中的 Cloud Shell 图标:

激活 Cloud Shell

预配和连接到环境应该只需要片刻时间。完成后,您应该会看到如下内容:

Google Cloud Shell 终端的屏幕截图,显示环境已连接

这个虚拟机已加载了您需要的所有开发工具。它提供了一个持久的 5 GB 主目录,并且在 Google Cloud 中运行,大大增强了网络性能和身份验证功能。您在此 Codelab 中的所有工作都可以在浏览器中完成。您无需安装任何程序。

3. 任务 1. 使用 Terraform 设置环境

我们将创建一个具有防火墙规则和子网的自定义 VPC。打开 Cloud 控制台,然后选择您要使用的项目。

  1. 打开位于控制台右上角的 Cloud Shell,确保您在 Cloud Shell 中看到正确的项目 ID,并确认所有允许访问的提示。b51b80043d3bac90.png
  2. 创建名为 terraform-build 的文件夹并移至该文件夹
mkdir terraform-build  && cd terraform-build
  1. 创建 main.tfvariable.tf 文件。
touch main.tf variable.tf 
  1. 切换到 Cloud Shell 编辑器视图。选择编辑器,确保允许任何必要的提示,以便加载界面。
  2. 加载完成后,依次前往文件 > 打开文件夹,然后前往 /home/your-user-name/terraform-build 并选择确定以在编辑器中打开该文件夹。39b3eb9a3e077bfd.png
  3. 选择 variable.tf 文件,然后添加以下内容。将 your-project-id-here 文本替换为您的实际项目 ID(用英文引号括起来)
variable "project_id" {
  type = string
  default = "your-project-id-here"
}

variable "network_id" {
  type = string
  default = "python-net"
}
  1. 接下来,打开 main.tf 文件。我们将添加一些 Terraform 代码来执行各种操作,如下所述。

启用 API

resource "google_project_service" "default"

创建名为 python-net 的 VPC

resource "google_compute_network" "default"

添加子网

resource "google_compute_subnetwork" "default"

添加两条防火墙规则

resource "google_compute_firewall" "allow_icmp"resource "google_compute_firewall" "allow_ssh"

  1. 将以下内容复制并粘贴到 main.tf 文件中。
resource "google_project_service" "default" {
  project = var.project_id 
  for_each = toset([
    "dns.googleapis.com",
    "aiplatform.googleapis.com",
    "servicedirectory.googleapis.com"
  ])

  service            = each.value
  disable_on_destroy = false
}

resource "google_compute_network" "default" {
  project                 = var.project_id
  name                    = var.network_id
  auto_create_subnetworks = false
  mtu                     = 1460
  routing_mode            = "GLOBAL"
}

resource "google_compute_subnetwork" "default" {
  name          = "vm1-subnet"
  ip_cidr_range = "10.0.11.0/24"
  project       = var.project_id
  region        = "us-east1"
  stack_type    = "IPV4_ONLY"
  network       = google_compute_network.default.id
}

resource "google_compute_firewall" "allow_icmp" {
  name    = "allow-icmp-${google_compute_network.default.name}"
  network = google_compute_network.default.id
  project = var.project_id

  allow {
    protocol = "icmp"
  }

  source_ranges = ["0.0.0.0/0"]
  target_tags   = ["allow-icmp"]
}

resource "google_compute_firewall" "allow_ssh" {
  name    = "allow-ssh-${google_compute_network.default.name}"
  network = google_compute_network.default.id
  project = var.project_id

  allow {
    protocol = "tcp"
    ports    = ["22"]
  }

  source_ranges = ["0.0.0.0/0"]
  target_tags   = ["allow-ssh"]
}
  1. 切换回 Cloud Shell 终端,确保您位于 terraform-build 目录 cd terraform-build 中,然后运行以下命令

terraform init

初始化工作目录。此步骤会下载给定配置所需的提供程序。

terraform plan

生成执行计划,显示 Terraform 将采取哪些操作来部署您的基础设施。

  1. 现在,运行 terraform apply 命令以创建资源,然后输入 yes 以运行。

4. 任务 2. 使用 Terraform 创建 NAT 网关和虚拟机

我们需要授予出站外部互联网访问权限,因此我们来创建 Cloud NAT 网关并将其附加。

  1. 打开 Cloud Shell,前往 terraform-build 文件夹,然后创建以下文件(共三个文件)。我们稍后会修改这些内容。
touch nat-vm.tf psc.tf dns.tf
  1. 切换到 Cloud Shell 编辑器视图,选择 nat-vm.tf 文件,然后添加以下 Terraform 代码。此命令将创建一个 NAT 网关和两个虚拟机。

Terraform nat-vm.tf

resource "google_compute_router" "default" {
  name    = "py-outbound-nat"
  region  = "us-east1"
  network = google_compute_network.default.id
  project = var.project_id
  

 bgp {
  asn = 64514
  }
}

resource "google_compute_router_nat" "default" {
  name                               = "py-outbound-nat-gw"
  router                             = google_compute_router.default.name
  region                             = google_compute_router.default.region
  nat_ip_allocate_option             = "AUTO_ONLY"
  project                            = var.project_id
  source_subnetwork_ip_ranges_to_nat = "ALL_SUBNETWORKS_ALL_IP_RANGES"

  log_config {
    enable = true
    filter = "ERRORS_ONLY"
  }
}

resource "google_compute_instance" "vm1" {
  name         = "py-vm1"
  zone         = "us-east1-b"
  machine_type = "n2-standard-2"
  project      = var.project_id

  boot_disk {
    initialize_params {
      image = "debian-cloud/debian-12"
    }
  }

  network_interface {
    subnetwork = google_compute_subnetwork.default.id  
    stack_type = "IPV4_ONLY"
  }

  tags = ["allow-ssh", "allow-icmp"]

  metadata_startup_script = <<-EOF
    #! /bin/bash
    sudo apt-get update
    sudo apt-get install python3 python3-dev python3-venv git -y
    sudo apt-get install tcpdump dnsutils -y    
    
      mkdir -p ~/py-gem-env
      cd ~/py-gem-env
      python3 -m venv env
      source env/bin/activate
      pip install --upgrade pip
      pip install ipython google-genai
    '
  EOF
}
  1. 切换到 Cloud Shell 终端,确保您位于 terraform-build 文件夹中,然后运行 terraform plan,系统会显示将添加 4 个项目,然后运行 terraform apply 并输入 yes 以创建 NAT 网关和虚拟机。

5. 任务 3. 配置虚拟机并进行测试

  1. 前往“虚拟机实例”。选择以 py-vm1 开头的虚拟机。选择 SSH
  2. 通过 SSH 连接到 py-vm1 后,输入 sudo -i 以启用 root
  3. 激活 venv 环境:
cd py-gem-env
source env/bin/activate
  1. 现在,我们来验证此账号,以便稍后进行一些测试。在虚拟机中运行以下命令,并在出现提示时按 y
gcloud auth application-default login
  1. 接下来,复制以 https:// 开头的网址,在实验浏览器窗口中打开一个新标签页,然后粘贴该网址。接受提示。
  2. 看到以下内容后,选择“复制”,然后切换回虚拟机 py-vm1 会话,在 Enter authorization code:(输入授权代码:)提示处粘贴您复制的代码,然后按 Enter 键进行身份验证。

b703db7aa2aa286a.png

  1. 现在,我们来快速测试一下是否可以连接到 Vertex API,该 API 使用 *-aiplatform.googleapis.com,因此我们将对该地址执行 dig,以查看流量路由方式。
dig *-aiplatform.googleapis.com
  1. 您应该会看到类似的内容(地址会有所不同)。请注意,由于该 API 是公共 API,因此路径是通过公共 IP 地址实现的。请勿复制
; <<>> DiG 9.18.41-1~deb12u1-Debian <<>> *-aiplatform.googleapis.com
;; global options: +cmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 60947
;; flags: qr rd ra; QUERY: 1, ANSWER: 16, AUTHORITY: 0, ADDITIONAL: 1

;; OPT PSEUDOSECTION:
; EDNS: version: 0, flags:; udp: 512
;; QUESTION SECTION:
;*-aiplatform.googleapis.com.   IN      A

;; ANSWER SECTION:
*-aiplatform.googleapis.com. 300 IN     A       173.194.216.95
*-aiplatform.googleapis.com. 300 IN     A       108.177.11.95
*-aiplatform.googleapis.com. 300 IN     A       192.178.219.95
*-aiplatform.googleapis.com. 300 IN     A       74.125.134.95
*-aiplatform.googleapis.com. 300 IN     A       74.125.139.95
*-aiplatform.googleapis.com. 300 IN     A       108.177.12.95
*-aiplatform.googleapis.com. 300 IN     A       173.194.210.95
*-aiplatform.googleapis.com. 300 IN     A       74.125.26.95
*-aiplatform.googleapis.com. 300 IN     A       173.194.212.95
*-aiplatform.googleapis.com. 300 IN     A       172.217.204.95
  1. 现在,我们来使用 Python。输入 ipython 以激活 ipython 界面。
ipython

845ef4291429888a.png

  1. 现在,复制并粘贴以下内容。这会向 Gemini 询问“Google 徽标的所有颜色是什么”和“天空是什么颜色”。
from google import genai
from google.genai import types
import os
import sys

LOCATION = "global"
MODEL_ID = "gemini-3-pro-preview" 

try:
    client = genai.Client(vertexai=True, location=LOCATION)
    print(f"Successfully initialized Google Gen AI Client (Vertex AI mode) in {LOCATION}")
except Exception as e:
    print(f"Error initializing client: {e}")
    print("Ensure you have installed the library: `pip install google-genai`")
    print("And authenticated: `gcloud auth application-default login`")
    sys.exit(1)

class SimpleChatSession:
    def __init__(self, model_id):
        self.model_id = model_id
        self.history = [] 

    def send_message(self, prompt, stream=True):
        user_content = types.Content(
            role="user",
            parts=[types.Part(text=prompt)]
        )
        self.history.append(user_content)

        try:
            response_stream = client.models.generate_content_stream(
                model=self.model_id,
                contents=self.history,
                config=types.GenerateContentConfig(
                    temperature=0.7 
                )
            )

            accumulated_text = ""
            
            for chunk in response_stream:
                if chunk.text:
                    yield chunk.text
                    accumulated_text += chunk.text
            
            model_content = types.Content(
                role="model",
                parts=[types.Part(text=accumulated_text)]
            )
            self.history.append(model_content)

        except Exception as e:
            print(f"\n[Error during generation: {e}]")

def get_chat_response(session: SimpleChatSession, prompt: str):
    print(f"\n--- User: {prompt} ---")
    print("--- Model: ", end="")
    
    stream_generator = session.send_message(prompt)
    
    full_text = ""
    for chunk_text in stream_generator:
        full_text += chunk_text
        print(chunk_text, end="", flush=True)

    print() 
    return full_text

if __name__ == "__main__":
    chat_session = SimpleChatSession(MODEL_ID)

    get_chat_response(chat_session, "Hello.")
    get_chat_response(chat_session, "What are all the colors of the Google logo?")
    get_chat_response(chat_session, "What color is the sky?")
  1. 按两次 Enter 键即可运行并查看结果。
  2. 此请求通过公开 API 访问了 Vertex。
  3. 关闭 SSH 会话,然后继续。

6. 任务 4. 使用 Terraform 创建与 googleapis 的 PSC 端点

如需启用与 Vertex API 端点的专用连接,我们将为 Google API 创建一个 Private Service Connect 端点。这样一来,我们就可以使用分配的专用 IP 地址将流量路由到所需的 Google API(在本例中为 Vertex)。

  1. 如果尚未打开,请在编辑器视图中打开 Cloud Shell。我们将创建以下内容:
  • 为 PSC 端点 192.168.255.250 创建 IP 地址 (resource "google_compute_global_address" "default")
  • 创建指向 Google API 的 PSC 端点 (resource "google_compute_global_forwarding_rule" "default")

打开 terraform-build 文件夹中的 psc.tf 文件。将以下代码添加到文件中。

Terraform psc.tf

resource "google_compute_global_address" "default" {
  name         = "vertex-ip"
  purpose      = "PRIVATE_SERVICE_CONNECT"
  network      = google_compute_network.default.id
  address_type = "INTERNAL"
  address      = "192.168.255.250"
}

resource "google_compute_global_forwarding_rule" "default" {  
  name                  = "pscvertexgemini"
  target                = "all-apis"
  network               = google_compute_network.default.id
  ip_address            = google_compute_global_address.default.id
  load_balancing_scheme = ""
  }
  1. 切换到 Cloud Shell 终端,确保您位于 terraform-build 文件夹中。然后运行 terraform init 然后运行 terraform plan,系统会显示将添加 2 个项目,
    然后运行 terraform apply 并输入 yes 以创建 IP 和 PSC Google API 端点。
  2. 验证端点是否存在
gcloud compute addresses list --filter="name=( 'vertex-ip' ...)"
gcloud compute forwarding-rules describe pscvertexgemini --global

7. 任务 5. 使用 Terraform 为 googleapis 创建手动 DNS 条目

您可以使用专用 DNS 创建指向 PSC 端点的手动 DNS 条目。这会影响您为其分配的所有媒体文件。

  1. 前往“网络服务”,然后选择“Cloud DNS”。
  2. 在相应可用区中,您应该会看到一个为 Google API 的 Private Service Connect 自动创建的可用区,其可用区类型为 Service Directory。此格式可用于连接到 PSC 端点:**SERVICE-ENDPOINT.p.googleapis.com。示例:aiplatform-pscvertexgemini.p.googleapis.com
  3. 在这种情况下,我们希望手动创建专用 DNS 条目。配置如下所示
  • 为“googleapis.com”创建一个名为“googleapis-private”的专用 DNS 区域,并将其限制为“python-net”网络。
  • 添加一条 A 记录,将“googleapis.com”映射到 IP 地址“192.168.255.250”。
  • 添加 CNAME 记录,以将“googleapis.com”的所有子网域(例如 www.googleapis.com)重定向到“googleapis.com”。
  1. 如果尚未打开,请在编辑器视图中打开 Cloud Shell。打开 terraform-build 文件夹中的 dns.tf 文件。将以下代码添加到文件中。

Terraform dns.tf

resource "google_dns_managed_zone" "private_zone" {
  name        = "googleapis-private"
  dns_name    = "googleapis.com."  
  visibility  = "private"
  project     = var.project_id     

  private_visibility_config {
    networks {
      network_url = google_compute_network.default.id  
    }
  }
}

resource "google_dns_record_set" "a_record" {
  name    = "googleapis.com."  
  type    = "A"
  ttl     = 300
  managed_zone = google_dns_managed_zone.private_zone.name
  project = var.project_id    

  rrdatas = ["192.168.255.250"]
}

resource "google_dns_record_set" "cname_record" {
 name    = "*.googleapis.com."
 type    = "CNAME"
 ttl     = 300
 managed_zone = google_dns_managed_zone.private_zone.name
 project = var.project_id    

 rrdatas = ["googleapis.com."]  
}
  1. 切换到 Cloud Shell 终端,确保您位于 terraform-build 文件夹中。然后运行 terraform plan,系统会显示将添加哪些项,
    然后运行 terraform apply 并输入 yes 以创建专用 DNS 条目。
  2. 您应该会看到包含 A 记录和 CNAME 的设置,如图所示 6d2fc061460cd983.png
  3. 接下来,我们验证 py-vm1 上这些更改的连接性

8. 任务 6. 通过 IP 地址验证端点连接

让我们使用专用端点连接到 Gemini。

  1. 前往虚拟机实例 py-vm1。选择“SSH”,并通过 SSH 连接到虚拟机
  2. 输入 sudo -i 以获取根访问权限
  3. 使用 ping 命令检查与 aiplatform.googleapis.com 的连接路径。这将对专用 DNS 中 googleapis 的 A 记录中的 IP 地址执行 ping 操作。此 IP 是 PSC 端点,因此您的 ping 将失败。
ping -c 2 aiplatform.googleapis.com
  1. 使用自动创建的 DNS 条目(适用于具有 aiplatform-pscvertexgemini.p.googleapis.com 的 PSC Google API)通过 ping 检查连接路径。这会指向 PSC 端点的 IP 地址,因此您的 ping 将失败。
ping -c 2 aiplatform-pscvertexgemini.p.googleapis.com
  1. 使用 dig 命令检查与 aiplatform.googleapis.com 的连接路径。这应该是 PSC 端点的 IP 地址。
dig aiplatform.googleapis.com
  1. 返回到控制台,然后打开另一个 py-vm1 虚拟机实例。选择 SSH,并通过 SSH 连接到虚拟机
  2. 输入 sudo -i 以获取根访问权限
  3. 运行以下命令,以在 TCP 转储中查看连接
sudo tcpdump -i any port 53 -n or host aiplatform.googleapis.com
  1. 现在,切换回虚拟机实例 py-vm1 的第一个 SSH 实例
  2. 使用以下命令激活环境
cd py-gem-env
source env/bin/activate
  1. 现在,我们来测试一下 Python。输入 ipython 以激活 ipython 界面。
ipython
  1. 现在,复制并粘贴以下内容。这会向 Gemini 提出以下问题:“请用一句话简要总结一下,在 AI 中,分词器是什么?”以及“鲸鱼很棒,对吗?”。
from google import genai
from google.genai import types
import os
import sys

LOCATION = "global"
MODEL_ID = "gemini-3-pro-preview" 

try:
    client = genai.Client(vertexai=True, location=LOCATION)
    print(f"Successfully initialized Google Gen AI Client (Vertex AI mode) in {LOCATION}")
except Exception as e:
    print(f"Error initializing client: {e}")
    print("Ensure you have installed the library: `pip install google-genai`")
    print("And authenticated: `gcloud auth application-default login`")
    sys.exit(1)

class SimpleChatSession:
    def __init__(self, model_id):
        self.model_id = model_id
        self.history = [] 

    def send_message(self, prompt, stream=True):
        user_content = types.Content(
            role="user",
            parts=[types.Part(text=prompt)]
        )
        self.history.append(user_content)

        try:
            response_stream = client.models.generate_content_stream(
                model=self.model_id,
                contents=self.history,
                config=types.GenerateContentConfig(
                    temperature=0.7 
                )
            )

            accumulated_text = ""
            
            for chunk in response_stream:
                if chunk.text:
                    yield chunk.text
                    accumulated_text += chunk.text
            
            model_content = types.Content(
                role="model",
                parts=[types.Part(text=accumulated_text)]
            )
            self.history.append(model_content)

        except Exception as e:
            print(f"\n[Error during generation: {e}]")

def get_chat_response(session: SimpleChatSession, prompt: str):
    print(f"\n--- User: {prompt} ---")
    print("--- Model: ", end="")
    
    stream_generator = session.send_message(prompt)
    
    full_text = ""
    for chunk_text in stream_generator:
        full_text += chunk_text
        print(chunk_text, end="", flush=True)

    print() 
    return full_text

if __name__ == "__main__":
    chat_session = SimpleChatSession(MODEL_ID)

    get_chat_response(chat_session, "Hello.")
    get_chat_response(chat_session, "In one short sentence summarize what is a tokenizer in the context of AI?")
    get_chat_response(chat_session, "Are whales awesome or not?")
  1. 两次 Enter 键即可运行并查看结果。
  2. 切换回第二个虚拟机实例 py-vm1。您应该会看到 TCPDUMP 的结果。您会注意到,虚拟机的入站和出站流量以及 IP 地址正在使用 PSC 端点 IP 地址连接到 aiplatform.googleapis.com

关闭与虚拟机实例 py-vm1 的所有 SSH 会话

9. 清理

  1. 前往 Cloud Shell,确保您位于 terraform-build 目录中 cd terraform-build
  2. 运行 terraform plan destroy 以查看将要进行的所有更改
terraform plan -destroy
  1. 然后运行命令 terraform destroy,接着输入 yes,这样一来,您使用 Terraform 在项目中创建的所有资源都将被移除。
terraform destroy 

10. 恭喜

恭喜!您已成功连接到 Vertex,并使用公共 API 地址和 Private Service Connect 端点(以非公开方式)访问了 Gemini 3 Pro。此功能可以将专用 API 连接扩展到通过(互连、Cross-Cloud Interconnect 和 VPC)连接的本地/其他云环境。

后续步骤/了解详情

您可以详细了解 Vertex AI 网络

Codelab:通过 Private Service Connect 端点使用 Python SDK 在 Vertex AI 上访问 Anthropic Claude

参与下一项实验

继续探索 Google Cloud,并查看以下其他 Google Cloud Skills Boost 实验: