模块 1：Google Cloud 生成式 AI 概览
本课程将带你深入理解 Google Cloud Platform (GCP) 上的生成式 AI 服务。Google Cloud 提供了一套完整的 AI/ML 工具链，从 Vertex AI 到 Generative AI Studio，帮助开发者快速构建和部署智能应用。
什么是 Google Cloud 生成式 AI？
Google Cloud 的生成式 AI 建立在 PaLM、Gemini 等大语言模型的基础上，通过 Vertex AI 平台提供统一的模型访问、微调和部署能力。开发者无需从零训练模型，即可利用 Google 的前沿 AI 能力。
核心优势：Google Cloud AI 将 Google Research 最新的模型研究成果（如 Gemini、PaLM 2、Imagen）通过 API 开放给全球开发者，让任何人都能快速接入最先进的 AI 能力。
核心服务一览
Vertex AI
GCP 的统一机器学习平台，提供模型训练、部署、监控的全流程管理。支持 AutoML 和自定义训练。
Generative AI Studio
低代码的交互式界面，允许你通过提示工程快速测试和迭代语言模型、图像模型的效果。
Model Garden
Google 提供的模型仓库，包含 Google 自研模型和开源模型，一键部署即可使用。
PaLM API / Gemini API
直接访问 Google 大语言模型的 REST API，支持文本生成、对话、嵌入等多种任务。
快速上手：调用 Gemini API
import google.generativeai as genai
# 配置 API Key
genai.configure(api_key="YOUR_API_KEY")
# 选择 Gemini Pro 模型
model = genai.GenerativeModel('gemini-pro')
# 生成文本
response = model.generate_content(
"解释量子计算的基本原理，使用中文回答"
)
print(response.text)这段代码展示了如何使用 Google 的 Python SDK 调用 Gemini Pro 模型。首先配置 API Key，然后初始化模型实例，最后通过 generate_content 方法发送提示并获取生成结果。
Google Cloud AI 技术栈全景
架构层次：
- 应用层 — Generative AI Studio、Document AI、Dialogflow CX
- 平台层 — Vertex AI（训练/部署/管道/MLOps）
- 模型层 — Gemini、PaLM 2、Imagen、Chirp（语音）
- 基础设施层 — TPU v5、GPU（A100/H100）、GKE
# 使用 gcloud CLI 启用 Vertex AI API
gcloud services enable aiplatform.googleapis.com
# 列出可用的基础模型
gcloud ai models list \
--region=us-central1 \
--filter="displayName:gemini"使用 gcloud CLI 命令行工具启用 Vertex AI 服务，并列出所有可用的 Gemini 模型。这是在 GCP 上开始使用生成式 AI 的第一步。
GCP AI 服务与其他云厂商对比
Google Cloud vs AWS
Google 优势在于自研模型（Gemini）和多模态能力；AWS Bedrock 提供更多第三方模型选择。Google 在 TPU 硬件上有独特优势。
Google Cloud vs Azure
Azure 与 OpenAI 深度集成（GPT 系列），Google 则以 Gemini 和 PaLM 为核心。选择取决于你的模型偏好和现有云基础设施。
Vertex AI 特有优势
内置 MLOps 管道、Feature Store、Model Monitoring，以及与 BigQuery、Dataflow 的深度集成，适合端到端 ML 工作流。
环境准备与项目设置
# 安装 Google Cloud AI Python SDK
pip install google-cloud-aiplatform
pip install google-generativeai
# 设置应用默认凭证
gcloud auth application-default login
# 设置项目 ID
export GOOGLE_CLOUD_PROJECT="your-project-id"
export GOOGLE_CLOUD_LOCATION="us-central1"这三步是使用 Google Cloud AI 服务的基础准备工作：安装 Python SDK 包、配置认证凭证、设置项目和区域环境变量。us-central1 是 Vertex AI 支持最全面的区域。
Google Cloud 生成式 AI 的核心平台是什么？
模块 2：Vertex AI 核心概念
在这一模块中，我们将深入探讨 Vertex AI 的核心概念，包括模型注册、端点部署、批量预测和 MLOps 管道。
Vertex AI 架构解析
Vertex AI 是 Google Cloud 的统一机器学习平台，它将模型生命周期管理的所有环节整合到一个平台中。从数据准备到模型训练、部署、监控，Vertex AI 提供了完整的工具链。
关键概念：
- Model — 注册在 Vertex AI 中的模型工件，可以是自定义训练的，也可以是从 Model Garden 导入的
- Endpoint — 模型部署的在线推理服务，支持自动扩缩容
- Pipeline — 基于 Kubeflow Pipelines 的 ML 工作流编排
- Feature Store — 用于存储和提供 ML 特征的统一服务
模型类型与选择指南
Gemini Pro
通用大语言模型，支持文本生成、对话、代码编写、推理分析。最多支持 32K tokens 输入。
Gemini Ultra
最强大的 Gemini 模型，处理高度复杂的推理和编码任务。适合需要深度分析的场景。
Gemini Nano
轻量级模型，专为设备端部署优化。适合移动应用和边缘计算场景。
Imagen / Chirp
Imagen 用于图像生成，Chirp 用于语音识别。多模态能力的核心组件。
初始化 Vertex AI SDK
from google.cloud import aiplatform
# 初始化 Vertex AI SDK
aiplatform.init(
project="your-project-id",
location="us-central1",
staging_bucket="gs://your-bucket/staging"
)
# 从 Model Garden 加载预训练模型
model = aiplatform.Model(
model_name="publishers/google/models/gemini-pro"
)
# 部署模型到端点
endpoint = model.deploy(
machine_type="n1-standard-4",
accelerator_type="NVIDIA_TESLA_T4",
accelerator_count=1,
min_replica_count=1,
max_replica_count=3,
)这段代码展示了完整的 Vertex AI SDK 初始化和模型部署流程。首先设置项目和区域，然后从 Model Garden 选择 Gemini Pro 模型，最后部署到一个自动扩缩容的在线端点。
Prompt Engineering 核心技巧
import google.generativeai as genai
genai.configure(api_key="YOUR_API_KEY")
# 系统指令：定义 AI 的角色和行为
model = genai.GenerativeModel(
'gemini-pro',
generation_config=genai.GenerationConfig(
temperature=0.7,
top_p=0.95,
top_k=40,
max_output_tokens=2048,
),
safety_settings={
'HATE': 'BLOCK_NONE',
'HARASSMENT': 'BLOCK_NONE',
}
)
# Few-shot 示例：通过示例引导模型输出格式
prompt = """
你是一个专业的技术文档翻译。
示例：
Input: "The API returns a JSON response"
Output: "该 API 返回 JSON 格式的响应"
现在翻译以下内容：
Input: "Vertex AI provides a unified platform for ML"
Output:
"""
response = model.generate_content(prompt)
print(response.text)这段代码展示了 Prompt Engineering 的两个核心技巧：系统指令通过 generation_config 控制模型的创造性和安全性；Few-shot 通过在提示中提供输入输出示例，引导模型按指定格式生成结果。
Token 计算与成本优化
成本管理要点：
- Gemini Pro 输入：$0.00025/1K characters
- Gemini Pro 输出：$0.0005/1K characters
- 使用
max_output_tokens控制输出长度 - 批量预测比在线请求便宜约 50%
- 缓存常用系统指令可减少重复 token 消耗
# 计算 token 用量
def count_tokens(text: str, model_name: str = "gemini-pro"):
model = genai.GenerativeModel(model_name)
token_count = model.count_tokens(text)
print(f"Token 数量: {token_count.total_tokens}")
return token_count.total_tokens
# 批量预测以降低成本
from google.cloud import aiplatform
batch_prediction_job = aiplatform.BatchPredictionJob(
display_name="gemini-batch-analysis",
model_name="publishers/google/models/gemini-pro",
input_config={
"instances_format": "jsonl",
"gcs_source": {"uris": ["gs://bucket/input.jsonl"]},
},
output_config={
"predictions_format": "jsonl",
"gcs_destination": {"output_uri_prefix": "gs://bucket/output/"},
},
)Token 计数帮助你预估成本，count_tokens 方法可在发送请求前检查 token 用量。批量预测通过异步处理大量请求来节省成本，适合离线数据分析场景。
Embeddings 与向量搜索
from vertexai.language_models import TextEmbeddingModel
# 初始化嵌入模型
embedding_model = TextEmbeddingModel.from_pretrained("text-embedding-004")
# 生成文本嵌入向量
embeddings = embedding_model.get_embeddings([
"Google Cloud 生成式 AI 提供 Vertex AI 平台",
"Gemini 是 Google 最新的大语言模型",
"机器学习需要大量训练数据"
])
for i, emb in enumerate(embeddings):
print(f"文本 {i+1}: 向量维度 = {len(emb.values)}")
print(f" 前5维: {emb.values[:5]}")
# 使用 Vertex AI Vector Search 进行相似度检索
from google.cloud import aiplatform
index = aiplatform.MatchingEngineIndex.create_tree_ah_index(
display_name="doc-embeddings-index",
dimensions=768,
approximate_neighbors_count=10,
)嵌入模型将文本转换为高维向量（768维），用于语义搜索、文本分类和聚类。Vertex AI Vector Search（基于 ScaNN 算法）提供毫秒级的向量相似度检索能力，是构建 RAG 应用的核心组件。
以下哪个是 Vertex AI 的核心组件？
模块 3：实践应用 — 构建 AI 应用
本模块将带你通过实际案例学习如何在 Google Cloud 上构建生成式 AI 应用，包括 RAG（检索增强生成）、对话系统和内容生成。
实战 1：构建 RAG（检索增强生成）应用
RAG 是目前最流行的 LLM 应用架构，它将检索系统和生成模型结合，使 AI 能够基于私有数据回答问题。
from vertexai.language_models import TextEmbeddingModel
from vertexai.preview.generative_models import GenerativeModel
import numpy as np
class SimpleRAG:
def __init__(self, project_id: str, location: str = "us-central1"):
import vertexai
vertexai.init(project=project_id, location=location)
self.embedding_model = TextEmbeddingModel.from_pretrained(
"text-embedding-004"
)
self.generation_model = GenerativeModel("gemini-pro")
self.documents = []
self.embeddings = []
def index_documents(self, docs: list[str]):
"""将文档分块并生成嵌入向量"""
self.documents = docs
self.embeddings = self.embedding_model.get_embeddings(docs)
# 转换为 numpy 数组便于相似度计算
self.embedding_matrix = np.array(
[e.values for e in self.embeddings]
)
def retrieve(self, query: str, top_k: int = 3) -> list[str]:
"""检索与查询最相关的文档"""
query_embedding = self.embedding_model.get_embeddings([query])
query_vec = np.array(query_embedding[0].values)
# 余弦相似度
similarities = np.dot(self.embedding_matrix, query_vec) / (
np.linalg.norm(self.embedding_matrix, axis=1) * np.linalg.norm(query_vec)
)
top_indices = np.argsort(similarities)[-top_k:][::-1]
return [self.documents[i] for i in top_indices]
def generate(self, query: str) -> str:
"""基于检索结果生成回答"""
relevant_docs = self.retrieve(query)
prompt = f"""基于以下参考资料回答问题。如果无法从参考资料中找到答案，请说明。
参考资料：
{chr(10).join(f'- {doc}' for doc in relevant_docs)}
问题：{query}
回答："""
response = self.generation_model.generate_content(prompt)
return response.text
# 使用示例
rag = SimpleRAG(project_id="your-project-id")
rag.index_documents([
"Vertex AI 是 Google Cloud 的统一 ML 平台",
"Gemini 支持文本、图像、音频的多模态理解",
"BigQuery 是 Google Cloud 的数据仓库服务",
"Cloud Functions 支持事件驱动的无服务器计算",
])
answer = rag.generate("Google Cloud 的 ML 平台是什么？")
print(answer)这是一个完整的 RAG 实现流程：索引阶段将文档转换为嵌入向量存储；检索阶段计算查询与文档的余弦相似度，返回最相关的 top_k 文档；生成阶段将检索到的文档作为上下文注入提示，让模型基于事实生成回答。
实战 2：多轮对话系统
import google.generativeai as genai
class ChatAssistant:
def __init__(self, api_key: str, system_instruction: str = ""):
genai.configure(api_key=api_key)
self.model = genai.GenerativeModel(
'gemini-pro',
system_instruction=system_instruction,
)
self.chat = self.model.start_chat(history=[])
def send_message(self, message: str) -> str:
response = self.chat.send_message(message)
return response.text
def get_history(self):
return self.chat.history
# 创建一个技术支持助手
assistant = ChatAssistant(
api_key="YOUR_API_KEY",
system_instruction="""你是 Google Cloud 技术支持专家。
回答问题时：
1. 先确认问题关键点
2. 提供具体解决方案和代码示例
3. 如有必要，推荐官方文档链接
4. 使用中文回答"""
)
# 模拟多轮对话
print(assistant.send_message("如何在 Vertex AI 上部署自定义模型？"))
print("---")
print(assistant.send_message("那自定义容器呢？我需要什么 Dockerfile？"))
print("---")
print(assistant.send_message("如何设置自动扩缩容策略？"))Gemini 的 Chat Session 自动管理对话历史，无需手动传递上下文。system_instruction 定义了 AI 的角色和行为规范。每次 send_message 会自动将完整历史发送给模型，确保连贯的多轮对话。
实战 3：Function Calling（函数调用）
Function Calling 是 Gemini 的重要能力，让模型能够调用外部 API、查询数据库或执行其他操作。模型不会直接执行函数，而是生成结构化的调用参数，由你的代码来执行。
import google.generativeai as genai
import json
genai.configure(api_key="YOUR_API_KEY")
# 定义可用函数
def get_weather(location: str) -> dict:
"""模拟天气 API"""
return {
"location": location,
"temperature": 22,
"condition": "晴天",
"humidity": 65,
}
def get_cloud_status(project_id: str) -> dict:
"""查询 GCP 项目状态"""
return {
"project": project_id,
"services": ["compute", "aiplatform", "storage"],
"status": "healthy",
"region": "us-central1",
}
# 声明函数给模型
model = genai.GenerativeModel('gemini-pro', tools=[
{
"function_declarations": [
{
"name": "get_weather",
"description": "获取指定位置的天气信息",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "城市名称，如'北京'、'上海'"
}
},
"required": ["location"]
}
},
{
"name": "get_cloud_status",
"description": "查询 Google Cloud 项目运行状态",
"parameters": {
"type": "object",
"properties": {
"project_id": {
"type": "string",
"description": "GCP 项目 ID"
}
},
"required": ["project_id"]
}
}
]
}
])
chat = model.start_chat()
response = chat.send_message("北京今天天气怎么样？")
# 处理函数调用
if response.parts[0].function_call:
fc = response.parts[0].function_call
print(f"模型请求调用: {fc.name}({dict(fc.args)})")
# 执行函数
if fc.name == "get_weather":
result = get_weather(**dict(fc.args))
# 将结果返回给模型
response = chat.send_message(
genai.protos.Content(
parts=[genai.protos.Part(
function_response=genai.protos.FunctionResponse(
name=fc.name,
response={"result": result}
)
)]
)
)
print(response.text)Function Calling 的工作流程：1) 定义函数签名和参数 schema；2) 用户提问时，模型判断是否需要调用函数；3) 模型返回函数调用请求（函数名 + 参数）；4) 你的代码执行实际函数；5) 将结果返回模型，模型生成最终回答。这使得 AI 可以与真实世界交互。
实战 4：内容生成流水线
from vertexai.preview.generative_models import GenerativeModel
from vertexai.language_models import TextEmbeddingModel
import json
class ContentPipeline:
"""AI 内容生成流水线"""
def __init__(self, project_id: str):
import vertexai
vertexai.init(project=project_id, location="us-central1")
self.model = GenerativeModel("gemini-pro")
def generate_blog_post(self, topic: str, style: str = "技术教程") -> dict:
"""生成技术博客文章"""
prompt = f"""请为主题 "{topic}" 生成一篇 {style} 风格的技术博客。
要求：
1. 标题吸引眼球
2. 包含 3-5 个代码示例
3. 每个代码示例有详细中文解释
4. 总结部分包含关键要点
以 JSON 格式返回：
{{
"title": "文章标题",
"summary": "100字以内的摘要",
"sections": [
{{"heading": "小节标题", "content": "内容", "code": "代码示例"}}
],
"key_takeaways": ["要点1", "要点2"]
}}"""
response = self.model.generate_content(prompt)
try:
return json.loads(response.text)
except json.JSONDecodeError:
return {"raw": response.text}
def generate_api_docs(self, endpoint: str, method: str) -> str:
"""生成 API 文档"""
prompt = f"""为以下 API 端点生成 OpenAPI 3.0 格式的文档：
端点: {endpoint}
方法: {method}
包含：
- 完整的请求参数说明
- 响应示例（成功和错误）
- 认证要求
- 速率限制说明"""
response = self.model.generate_content(prompt)
return response.text
# 使用示例
pipeline = ContentPipeline(project_id="your-project-id")
# 生成博客
blog = pipeline.generate_blog_post("Google Cloud Vertex AI 入门指南")
print(f"生成文章: {blog.get('title', 'N/A')}")
# 生成 API 文档
docs = pipeline.generate_api_docs("/v1/models/gemini-pro:generateContent", "POST")
print(docs[:500])内容生成流水线展示了如何将 Gemini 的生成能力集成到自动化工作流中。博客生成器接收主题和风格参数，输出结构化的 JSON 内容。API 文档生成器则根据端点信息自动生成标准格式的文档。这种模式可以扩展到报告生成、邮件撰写、数据摘要等场景。
错误处理与最佳实践
重试机制
使用指数退避策略处理速率限制和临时故障。Google API 默认重试 3 次，建议自定义重试逻辑。
输入验证
发送前验证输入长度和格式，避免超出 token 限制。使用 count_tokens 预检。
安全过滤
配置 safety_settings 根据应用场景调整内容安全级别，避免误拦截。
异步调用
使用 asyncio + aiohttp 并发调用 API，提升吞吐量。注意并发限制。
import asyncio
import aiohttp
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def call_gemini_async(session, prompt: str) -> str:
"""带重试的异步 Gemini API 调用"""
url = f"https://generativelanguage.googleapis.com/v1/models/gemini-pro:generateContent"
params = {"key": "YOUR_API_KEY"}
payload = {
"contents": [{"parts": [{"text": prompt}]}],
"generationConfig": {
"temperature": 0.7,
"maxOutputTokens": 1024,
}
}
async with session.post(url, params=params, json=payload) as resp:
if resp.status == 429:
raise Exception("Rate limited - will retry")
resp.raise_for_status()
data = await resp.json()
return data["candidates"][0]["content"]["parts"][0]["text"]
async def batch_generate(prompts: list[str]) -> list[str]:
"""批量异步生成"""
async with aiohttp.ClientSession() as session:
tasks = [call_gemini_async(session, p) for p in prompts]
return await asyncio.gather(*tasks)
# 使用示例
prompts = [
"用一句话解释机器学习",
"用一句话解释深度学习",
"用一句话解释强化学习",
]
results = asyncio.run(batch_generate(prompts))
for r in results:
print(r)生产环境的最佳实践：使用 tenacity 库实现指数退避重试；使用 aiohttp 进行异步并发调用提升吞吐量；asyncio.gather 同时处理多个请求。注意 API 密钥应使用环境变量而非硬编码。
RAG 应用的核心组件是什么？
模块 4：进阶主题 — 模型微调与 MLOps
在本模块中，我们将探讨 Google Cloud AI 的高级特性，包括模型微调、Vertex AI Pipelines、Model Monitoring 和 Responsible AI。
模型微调（Fine-tuning）
当预训练模型无法满足特定领域需求时，微调是提升模型表现的关键手段。Vertex AI 支持对 Gemini、PaLM 等模型进行有监督微调。
from google.cloud import aiplatform
from google.cloud.aiplatform.compat.types import (
pipeline_state as ps
)
aiplatform.init(project="your-project-id", location="us-central1")
# 准备微调数据集（JSONL 格式）
# 每行一个训练样本，包含 input_text 和 output_text
training_data = """
{"input_text": "解释 Vertex AI", "output_text": "Vertex AI 是 Google Cloud 的统一 ML 平台..."}
{"input_text": "什么是 Gemini？", "output_text": "Gemini 是 Google 的多模态大语言模型..."}
{"input_text": "如何使用 BigQuery？", "output_text": "BigQuery 是无服务器数据仓库，支持 SQL 查询..."}
"""
# 上传训练数据到 GCS
from google.cloud import storage
client = storage.Client()
bucket = client.bucket("your-training-bucket")
blob = bucket.blob("training/fine_tune_data.jsonl")
blob.upload_from_string(training_data.strip())
# 启动微调任务
tuning_job = aiplatform.SupervisedTuningJob(
display_name="gemini-domain-finetune",
base_model="gemini-pro",
training_data_uri="gs://your-training-bucket/training/fine_tune_data.jsonl",
tuned_model_display_name="gemini-domain-v1",
epoch_count=4,
learning_rate_multiplier=1.0,
batch_size=8,
)
tuning_job.run()
print(f"微调状态: {tuning_job.state}")
print(f"微调后模型: {tuning_job.tuned_model_endpoint_name}")微调流程：1) 准备 JSONL 格式的训练数据（每行包含输入和期望输出）；2) 上传到 Cloud Storage；3) 配置微调参数（epochs、学习率、batch size）；4) 提交微调任务到 Vertex AI；5) 获取微调后的模型端点。微调通常需要几十分钟到几小时。
Vertex AI Pipelines 构建 MLOps 工作流
from google.cloud import aiplatform
from kfp import dsl
from kfp.v2 import compiler
@dsl.component(packages_to_install=["google-cloud-aiplatform"])
def train_model(dataset_uri: str, model_name: str) -> str:
from google.cloud import aiplatform
aiplatform.init(project="your-project-id", location="us-central1")
# 自定义训练任务
job = aiplatform.CustomTrainingJob(
display_name=model_name,
script_path="train.py",
container_uri="us-docker.pkg.dev/vertex-ai/training/tf-cpu.2-12:latest",
requirements=["google-cloud-aiplatform", "scikit-learn"],
)
model = job.run(
dataset=dataset_uri,
model_display_name=model_name,
args=["--epochs", "10", "--batch_size", "32"],
)
return model.resource_name
@dsl.component(packages_to_install=["google-cloud-aiplatform"])
def deploy_model(model_resource_name: str) -> str:
from google.cloud import aiplatform
aiplatform.init(project="your-project-id", location="us-central1")
model = aiplatform.Model(model_resource_name)
endpoint = model.deploy(
machine_type="n1-standard-4",
min_replica_count=1,
max_replica_count=5,
)
return endpoint.resource_name
@dsl.pipeline(name="ml-training-pipeline")
def ml_pipeline(dataset_uri: str, model_name: str):
train_task = train_model(
dataset_uri=dataset_uri,
model_name=model_name,
)
deploy_task = deploy_model(
model_resource_name=train_task.output,
)
deploy_task.after(train_task)
# 编译并运行管道
compiler.Compiler().compile(
pipeline_func=ml_pipeline,
package_path="ml_pipeline.json"
)
job = aiplatform.PipelineJob(
display_name="ml-training-pipeline-run",
template_path="ml_pipeline.json",
parameter_values={
"dataset_uri": "gs://your-bucket/datasets/train.csv",
"model_name": "production-model-v1",
},
)
job.run()Vertex AI Pipelines 基于 Kubeflow Pipelines 构建，允许你将 ML 工作流定义为有向无环图（DAG）。每个组件是一个独立的函数，可以运行在不同的容器环境中。@dsl.component 装饰器定义组件，@dsl.pipeline 定义管道拓扑。组件之间通过输入输出参数传递数据。
Model Monitoring（模型监控）
为什么需要模型监控？模型部署后，输入数据分布可能随时间变化（数据漂移），导致模型性能下降。Vertex AI Model Monitoring 自动检测预测请求中的异常分布，并在超过阈值时发送告警。
from google.cloud import aiplatform
aiplatform.init(project="your-project-id", location="us-central1")
# 为已部署的端点启用模型监控
endpoint = aiplatform.Endpoint("your-endpoint-id")
# 配置特征漂移检测
monitoring_job = aiplatform.ModelDeploymentMonitoringJob(
display_name="production-model-monitor",
endpoint=endpoint.resource_name,
logging_sampling_strategy={
"random_sample_config": {"sample_rate": 0.8}
},
model_monitoring_objective_configs=[
{
"objective_config": {
"training_dataset": {
"data_format": "csv",
"gcs_source": {"uris": ["gs://bucket/training_data.csv"]},
},
"training_prediction_skew_config": {
"skew_thresholds": {
"feature_a": {"value": 0.05},
"feature_b": {"value": 0.1},
}
},
"drift_thresholds": {
"feature_a": {"value": 0.05},
"feature_b": {"value": 0.1},
}
}
}
],
model_monitoring_alert_config={
"email_alert_config": {
"user_emails": ["team@example.com"],
}
},
)模型监控配置了三种检测机制：训练/预测偏差检测（skew）比较生产数据与训练数据的分布差异；特征漂移检测（drift）追踪时间维度上的分布变化；告警配置在超过阈值时通过邮件通知团队。
Responsible AI 与安全最佳实践
内容安全过滤
Gemini 内置安全过滤器，可检测和阻止有害内容输出。支持按类别配置过滤级别，适应不同应用场景。
数据隐私保护
Google Cloud 默认加密所有数据（静态和传输中）。Vertex AI 不使用客户数据训练基础模型，确保数据主权。
公平性评估
使用 Vertex AI Model Evaluation 评估模型在不同人口统计组的表现差异，识别和缓解偏见。
可解释性工具
Vertex Explainable AI 提供特征归因分析，帮助理解模型为什么做出特定预测，增强透明度和信任。
# 配置安全过滤和内容审核
import google.generativeai as genai
safety_config = {
generativeai.HarmCategory.HARM_CATEGORY_HARASSMENT:
generativeai.HarmBlockThreshold.BLOCK_ONLY_HIGH,
generativeai.HarmCategory.HARM_CATEGORY_HATE_SPEECH:
generativeai.HarmBlockThreshold.BLOCK_ONLY_HIGH,
generativeai.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT:
generativeai.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
generativeai.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT:
generativeai.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
}
model = genai.GenerativeModel(
'gemini-pro',
safety_settings=safety_config,
)
response = model.generate_content("解释量子计算原理")
# 检查安全评级
for candidate in response.candidates:
print(f"安全评级: {candidate.safety_ratings}")
# 如被过滤，给出替代响应
if candidate.finish_reason == 3: # SAFETY
print("内容被安全过滤器拦截")安全配置通过 safety_settings 字典控制每个伤害类别的过滤阈值。BLOCK_ONLY_HIGH 只拦截高风险内容，BLOCK_MEDIUM_AND_ABOVE 拦截中等及以上风险。生成后检查 safety_ratings 了解每个类别的风险评分，实现多层安全防护。
多模态应用开发
import google.generativeai as genai
from PIL import Image
import io
genai.configure(api_key="YOUR_API_KEY")
# 使用 Gemini Pro Vision 处理图像
vision_model = genai.GenerativeModel('gemini-pro-vision')
# 从本地加载图像
img = Image.open("architecture_diagram.png")
# 发送图像 + 文本提示
response = vision_model.generate_content([
"请分析这张架构图，解释其组件和数据流。使用中文回答。",
img
])
print(response.text)
# 视频理解（Gemini 1.5 Pro）
video_model = genai.GenerativeModel('gemini-1.5-pro')
# 上传视频文件
video_file = genai.upload_file(
path="product_demo.mp4",
display_name="Product Demo Video"
)
# 等待视频处理完成
import time
while video_file.state.name == "PROCESSING":
print("视频处理中...")
time.sleep(10)
video_file = genai.get_file(video_file.name)
response = video_model.generate_content([
video_file,
"请总结这个产品演示视频的要点，包括关键功能和时间节点。"
])
print(response.text)Gemini 的多模态能力支持图像和视频理解。图像分析直接传入 PIL Image 对象，模型可以识别图表、截图、文档等。视频理解通过文件上传 API 处理，Gemini 1.5 Pro 支持最长 1 小时的视频输入，可以定位特定事件的时间节点。
模型微调需要准备什么格式的训练数据？
模块 5：总结与综合练习
在本模块中，我们将回顾 Google Cloud 生成式 AI 的所有核心知识点，并通过综合练习巩固所学内容。
课程知识图谱
核心知识回顾：
- 模块 1：GCP AI 服务概览 — Vertex AI、Generative AI Studio、Model Garden
- 模块 2：核心概念 — 模型类型、SDK 初始化、Prompt Engineering、Embeddings
- 模块 3：实践应用 — RAG 构建、对话系统、Function Calling、内容生成流水线
- 模块 4：进阶主题 — 模型微调、MLOps 管道、模型监控、Responsible AI、多模态
综合练习 1：构建完整的 AI 客服系统
"""综合练习：Google Cloud AI 客服系统"""
import google.generativeai as genai
from vertexai.language_models import TextEmbeddingModel
from google.cloud import aiplatform
import numpy as np
import json
class AICustomerService:
"""基于 Gemini + RAG 的智能客服"""
def __init__(self, project_id: str, api_key: str):
genai.configure(api_key=api_key)
aiplatform.init(project=project_id, location="us-central1")
# 初始化模型
self.chat_model = genai.GenerativeModel(
'gemini-pro',
system_instruction="""你是专业的客服代表。
规则：
1. 礼貌、专业、有同理心
2. 优先使用知识库中的信息回答
3. 无法回答时诚实告知并提供升级路径
4. 记录用户反馈用于改进""",
)
self.embedding_model = TextEmbeddingModel.from_pretrained("text-embedding-004")
# 知识库
self.knowledge_base = {}
self.embeddings_cache = {}
def load_knowledge_base(self, category: str, documents: list[str]):
"""加载知识库文档"""
self.knowledge_base[category] = documents
embeddings = self.embedding_model.get_embeddings(documents)
self.embeddings_cache[category] = np.array([e.values for e in embeddings])
def search_knowledge(self, query: str, category: str = None, top_k: int = 3) -> list[str]:
"""语义检索知识库"""
query_emb = self.embedding_model.get_embeddings([query])
query_vec = np.array(query_emb[0].values)
categories = [category] if category else list(self.knowledge_base.keys())
results = []
for cat in categories:
if cat not in self.embeddings_cache:
continue
emb_matrix = self.embeddings_cache[cat]
sims = np.dot(emb_matrix, query_vec) / (
np.linalg.norm(emb_matrix, axis=1) * np.linalg.norm(query_vec)
)
top_idx = np.argsort(sims)[-top_k:][::-1]
for idx in top_idx:
results.append({
"text": self.knowledge_base[cat][idx],
"score": float(sims[idx]),
"category": cat,
})
results.sort(key=lambda x: x["score"], reverse=True)
return [r["text"] for r in results[:top_k]]
def handle_query(self, user_id: str, query: str) -> dict:
"""处理用户查询"""
# 1. 检索相关知识
relevant_docs = self.search_knowledge(query)
# 2. 构建增强提示
context = "\n".join(f"- {doc}" for doc in relevant_docs)
enhanced_query = f"""参考知识：
{context}
用户问题：{query}
请基于参考知识回答用户问题。如果知识库中没有相关信息，请说明。"""
# 3. 生成回答
chat = self.chat_model.start_chat()
response = chat.send_message(enhanced_query)
return {
"user_id": user_id,
"query": query,
"answer": response.text,
"sources": relevant_docs,
"model": "gemini-pro",
}
# 初始化系统
service = AICustomerService(
project_id="your-project-id",
api_key="YOUR_API_KEY"
)
# 加载知识库
service.load_knowledge_base("billing", [
"账单每月1号生成，可在控制台查看明细",
"支持信用卡和支付宝付款",
"退款申请需在消费后30天内提交",
"企业客户可申请月结",
])
service.load_knowledge_base("technical", [
"API 调用频率限制为每分钟60次",
"使用 API Key 进行认证",
"支持 REST 和 gRPC 两种调用方式",
"错误码 429 表示请求过于频繁，请稍后重试",
])
# 处理用户查询
result = service.handle_query("user_123", "怎么付款？账单什么时候出？")
print(json.dumps(result, ensure_ascii=False, indent=2))这个综合练习整合了课程所有核心知识：RAG（嵌入检索增强）、Prompt Engineering（系统指令和上下文注入）、Function Calling（扩展为知识库检索）、错误处理（降级策略）。这是一个可以直接部署到生产环境的 AI 客服系统骨架。
综合练习 2：多模态文档分析器
"""综合练习：多模态文档分析器"""
import google.generativeai as genai
from PIL import Image
genai.configure(api_key="YOUR_API_KEY")
class DocumentAnalyzer:
"""支持文本、图像、表格的多模态文档分析"""
def __init__(self):
self.model = genai.GenerativeModel('gemini-1.5-pro')
self.vision_model = genai.GenerativeModel('gemini-pro-vision')
def analyze_image(self, image_path: str, prompt: str = None) -> str:
"""分析图像内容"""
img = Image.open(image_path)
analysis_prompt = prompt or """
请详细分析这张图片，包括：
1. 内容描述
2. 如果是图表，提取关键数据
3. 如果是截图，解释界面元素
4. 如果是文档，提取关键信息
"""
response = self.vision_model.generate_content([analysis_prompt, img])
return response.text
def extract_structured_data(self, content: str, schema: dict) -> dict:
"""从非结构化内容中提取结构化数据"""
prompt = f"""从以下内容中提取结构化数据。
目标 JSON Schema:
{json.dumps(schema, ensure_ascii=False, indent=2)}
内容：
{content}
请严格按照 schema 格式返回 JSON，不要包含其他文字。"""
response = self.model.generate_content(prompt)
try:
# 尝试解析 JSON（可能包裹在 ```json``` 中）
text = response.text
if "```json" in text:
text = text.split("```json")[1].split("```")[0]
return json.loads(text.strip())
except:
return {"error": "Failed to parse", "raw": response.text}
def summarize_document(self, content: str, max_length: int = 200) -> str:
"""生成文档摘要"""
prompt = f"""请用中文生成以下文档的摘要，不超过 {max_length} 字。
重点突出：
1. 核心主题
2. 关键结论
3. 行动建议（如有）
文档内容：
{content}"""
response = self.model.generate_content(prompt)
return response.text
def compare_documents(self, doc1: str, doc2: str) -> str:
"""比较两份文档的差异"""
prompt = f"""请比较以下两份文档的异同：
文档 A:
{doc1}
文档 B:
{doc2}
请列出：
1. 相同点
2. 不同点
3. 哪份文档更完整/准确"""
response = self.model.generate_content(prompt)
return response.text
# 使用示例
analyzer = DocumentAnalyzer()
# 分析架构图
# result = analyzer.analyze_image("architecture.png")
# print(result)
# 提取结构化数据
text_content = """
Google Cloud Next '24 大会上宣布了 Gemini 1.5 Pro 正式发布。
该模型支持 100 万 token 上下文窗口，价格为企业版 $0.0125/千字符。
首发区域包括 us-central1、europe-west1 和 asia-northeast1。
"""
schema = {
"type": "object",
"properties": {
"event": {"type": "string"},
"product": {"type": "string"},
"features": {"type": "array", "items": {"type": "string"}},
"pricing": {"type": "string"},
"regions": {"type": "array", "items": {"type": "string"}}
}
}
data = analyzer.extract_structured_data(text_content, schema)
print(data)多模态文档分析器结合了图像理解和文本处理能力：analyze_image 处理视觉内容；extract_structured_data 通过 JSON Schema 约束输出格式；summarize_document 生成精准摘要；compare_documents 比较文档差异。这个工具可以集成到文档管理、合规审查、竞品分析等场景。
学习路径与进阶方向
Google Cloud AI 认证
考取 Google Cloud Professional Machine Learning Engineer 认证，系统化验证你的 AI/ML 工程能力。
深入学习 TensorFlow
Vertex AI 深度集成 TensorFlow。掌握 TF Extended (TFX) 可以构建更强大的 ML 生产管道。
MLOps 工程实践
学习 CI/CD for ML、模型版本管理、A/B 测试、特征工程平台等 MLOps 核心实践。
LLMOps 新兴领域
LLM 运维（LLMOps）是 MLOps 的演进，关注 prompt 版本管理、模型评估、幻觉检测等新挑战。
常用资源与参考链接
官方资源：
- Vertex AI 文档 — cloud.google.com/vertex-ai/docs
- Gemini API 参考 — ai.google.dev/docs
- Google Cloud AI 博客 — cloud.google.com/blog/topics/ai-ml
- Generative AI Studio — console.cloud.google.com/vertex-ai/generative
- Model Garden — console.cloud.google.com/vertex-ai/models
- Google AI 开发者社区 — discuss.ai.google.dev
以下哪些是本课程涵盖的核心技能？