GCP AI 與大數據：從 BigQuery 到生成式 AI 的完整工具鏈

數據是現在企業最重要的資產，這點大家都同意。但光有數據還不夠，你得能儲存、處理、分析這些數據，再拿來訓練 AI 模型，甚至接上生成式 AI 提供智能服務。

GCP 把這一整套大數據與 AI 工具都備齊了，從資料倉儲 BigQuery、數據處理服務 Dataflow/Dataproc、訊息佇列 Pub/Sub，到機器學習平台 Vertex AI 和生成式 AI 的 Gemini API。這篇就帶你把這些工具的核心功能、怎麼選，以及怎麼把它們兜成一套智能客服系統的實戰案例都搞懂。

讀者將學到什麼？

✅ BigQuery 資料倉儲：如何使用 PB 級分析引擎處理海量數據 ✅ Dataflow vs Dataproc：批次處理與串流處理的選型決策樹 ✅ Pub/Sub 訊息佇列：建構事件驅動的即時數據架構 ✅ Vertex AI 與 AutoML：無程式碼訓練機器學習模型 ✅ Gemini API 生成式 AI：企業級整合與實際應用 ✅ 實戰案例：端到端的智能客服系統架構設計

BigQuery 資料倉儲：PB 級數據分析的基石

為什麼選擇 BigQuery？

BigQuery 是 GCP 的全代管資料倉儲服務,在 2026 年 Gartner Magic Quadrant 也被評為 Leader。它的核心優勢有這些：

🆕 2025-2026 BigQuery 重大更新

1. BigQuery Earth Engine 整合（正式推出，GA）

功能：直接在 BigQuery 中查詢 Earth Engine 地理空間數據集
數據規模：存取 PB 級的衛星影像、氣候數據、地形資料
用途：環境監測、氣候變遷分析、農業優化、城市規劃
範例：分析特定區域過去 20 年的森林覆蓋變化

2. Serverless Spark in BigQuery（正式推出，GA）

功能：在 BigQuery 中直接執行 Spark 作業，無需管理 Dataproc 集群
優勢：自動擴展、不用管基礎設施、跟 BigQuery 數據接得很順
用途：複雜的 ETL 處理、機器學習特徵工程、大規模數據轉換
成本：按實際計算資源使用計費，相較於自管 Dataproc 集群可降低營運與閒置成本（實際節省幅度依工作負載而定）

使用範例：

from google.cloud import bigquery

client = bigquery.Client()

# 使用 BigQuery Spark 執行數據處理
spark_job = client.create_spark_job(
    spark_application='gs://my-bucket/spark-app.jar',
    main_class='com.example.DataProcessor'
)

1. 無伺服器架構（Serverless）

不用管任何基礎設施。下面這些你都不用煩：

伺服器容量規劃
備份與災難恢復
版本升級與維護

你只要上傳數據、寫 SQL 查詢，BigQuery 會自動配運算資源跑分析。

2. PB 級規模處理

BigQuery 可處理 Petabyte（PB）規模的資料，同時保持低成本。舉例來說：

掃描 1 TB 數據的成本僅約 $5 美元
查詢 10 億筆記錄通常在秒級完成

3. 近即時分析

傳統資料庫得先建索引、預先載入數據才能查。BigQuery 不一樣，它支援即時串流寫入（每秒數百萬筆），寫進去馬上就能查。

4. AI 輔助查詢優化（2026 新功能）

Gemini in BigQuery 現已納入定價模型，提供：

自然語言轉 SQL：直接用中文問「過去 30 天銷售額最高的前 10 個產品」，自動生成 SQL
查詢優化建議：AI 會分析你的 SQL 並提出效能改進建議
異常偵測：自動標記數據異常，如突然的流量激增或銷售下降

BigQuery SQL 查詢優化技巧

BigQuery 雖然快，但查詢寫得不好還是會多花錢。下面是 2026 年最新的幾招優化技巧：

1. Partitioned Tables（分區表）

問題：查詢包含數十億筆歷史數據的表格時，即使只需要最近一週的數據，也會掃描整張表。

解決方案：根據時間或整數範圍分區。

-- 建立按日期分區的表格
CREATE TABLE `project.dataset.sales`
PARTITION BY DATE(order_date)
AS SELECT * FROM `project.dataset.raw_sales`;

-- 查詢時只掃描特定分區
SELECT SUM(amount)
FROM `project.dataset.sales`
WHERE DATE(order_date) BETWEEN '2026-01-01' AND '2026-01-07';

效益：查詢只掃描 7 天的數據分區，成本降低 99%（假設總數據涵蓋 2 年）。

2. Clustering（聚類）

問題：即使使用分區，單一分區內的數據仍可能雜亂無章。

解決方案：根據常用查詢欄位自動排序。

-- 建立分區 + 聚類的表格
CREATE TABLE `project.dataset.sales`
PARTITION BY DATE(order_date)
CLUSTER BY customer_id, product_id
AS SELECT * FROM `project.dataset.raw_sales`;

效益：查詢特定客戶或產品時，BigQuery 可跳過大量不相關的區塊，進一步減少掃描成本。

3. Materialized Views（物化視圖）

問題：某些複雜的聚合查詢（如每日銷售總額）會被重複執行。

解決方案：建立快取視圖，自動更新。

-- 建立物化視圖
CREATE MATERIALIZED VIEW `project.dataset.daily_sales`
AS
SELECT
  DATE(order_date) AS sale_date,
  SUM(amount) AS total_sales
FROM `project.dataset.sales`
GROUP BY sale_date;

-- 查詢物化視圖（極快）
SELECT * FROM `project.dataset.daily_sales`
WHERE sale_date = '2026-01-15';

效益：物化視圖會自動增量更新，查詢速度提升 10-100 倍。

BigQuery 與其他資料倉儲比較

面向	BigQuery	AWS Redshift	Azure Synapse
架構	Serverless	Cluster-based	Hybrid
擴展性	自動擴展	需手動調整節點	自動擴展（Serverless Pool）
查詢語言	Standard SQL	PostgreSQL SQL	T-SQL
AI 整合	Gemini 原生整合	SageMaker 外掛	Azure ML 外掛
成本模式	按查詢計費	按集群運行時間	按 DWU 或查詢計費

結論：如果你已經在用 GCP，或者需要 AI 原生整合（像 Vertex AI、Gemini），那 BigQuery 就是首選。

實戰範例：分析網站流量數據

假設你有一個電商網站，每天產生數百萬筆點擊日誌。使用 BigQuery，你可以：

-- 1. 匯入 Google Analytics 4 數據（自動同步）
-- 2. 查詢過去 7 天流量最高的頁面

SELECT
  page_location,
  COUNT(*) AS page_views,
  COUNT(DISTINCT user_pseudo_id) AS unique_users
FROM `project.analytics.events_*`
WHERE _TABLE_SUFFIX BETWEEN '20260115' AND '20260121'
  AND event_name = 'page_view'
GROUP BY page_location
ORDER BY page_views DESC
LIMIT 10;

執行時間：約 2-3 秒（即使掃描 1 億筆記錄）

Dataflow vs Dataproc：數據處理服務的選型策略

當數據量大到一台機器處理不來時，你就需要分散式數據處理框架。GCP 在這塊有兩個主力：Dataflow 和 Dataproc。問題是，該選哪一個？

🆕 2025-2026 Dataflow 重大更新

1. Dataflow Prime 成為預設模式（2026 年 8 月 4 日起）

變更：所有新的 Dataflow 作業將自動使用 Dataflow Prime
優勢：自動垂直與水平自動擴展、更好的資源利用率、降低成本
效能提升：相較於經典模式，處理速度提升 20-30%
成本優化：自動調整資源配置，減少閒置成本

2. ⚠️ Dataflow SQL 棄用（2026 年 1 月 31 日）

狀態：Dataflow SQL 已於 2026 年 1 月 31 日正式棄用
替代方案：使用 BigQuery + Serverless Spark 或 Apache Beam SQL
遷移建議：現有 Dataflow SQL 用戶應盡快遷移至 BigQuery

Dataflow Prime 範例：

# 使用 Dataflow Prime（自動啟用）
gcloud dataflow flex-template run my-pipeline \
  --template-file-gcs-location=gs://my-bucket/template.json \
  --region=asia-east1 \
  --enable-streaming-engine  # Prime 自動優化

核心差異一覽表

面向	Dataflow	Dataproc
架構	Serverless（無伺服器）	Managed Clusters（代管集群）
程式模型	Apache Beam（統一 batch + streaming）	Hadoop/Spark（主要 batch）
管理複雜度	低（自動擴展）	中（需管理集群配置）
啟動時間	快（秒級）	慢（分鐘級，需啟動 VM）
成本模式	按使用量計費	按集群運行時間計費
適合場景	即時串流、無維護需求	批次處理、遷移 Hadoop/Spark 工作

Dataflow：Serverless 串流與批次處理

什麼是 Apache Beam？

Dataflow 用的是 Apache Beam 的統一程式模型，好處是同一套程式碼就能同時處理批次和串流數據。

Beam 的核心概念：

Pipeline（管道）
 ├─ PCollection（數據集合）
 ├─ Transform（轉換操作）
 └─ I/O Connector（輸入輸出）

實戰範例：即時計算網站訪問量

假設你有一條即時的網站日誌串流（透過 Pub/Sub 傳送），想每分鐘算一次訪問量：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# 定義 Dataflow Pipeline
with beam.Pipeline(options=PipelineOptions()) as p:
    (p
     | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(
         topic='projects/my-project/topics/website-logs')
     | 'Parse JSON' >> beam.Map(lambda msg: json.loads(msg))
     | 'Extract Page' >> beam.Map(lambda log: log['page_url'])
     | 'Count per Minute' >> beam.WindowInto(
         beam.window.FixedWindows(60))  # 60 秒視窗
     | 'Count' >> beam.combiners.Count.PerElement()
     | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
         table='my-project:analytics.page_views')
    )

執行方式：

python dataflow_pipeline.py \
  --project=my-project \
  --runner=DataflowRunner \
  --region=asia-east1 \
  --temp_location=gs://my-bucket/temp

Dataflow 會自動：

啟動 VM workers
根據數據量自動擴展（auto-scaling）
處理失敗重試
完成後自動關閉資源

成本：只在 pipeline 執行時計費，無閒置成本。

Dataproc：Hadoop/Spark 集群的代管服務

適合場景

選擇 Dataproc 的時機：

✅ 已有 Hadoop/Spark 工作負載要從地端遷移到雲端
✅ 團隊精通 Spark、Hive、Presto 等工具
✅ 需要執行特殊工具（如 HBase、Flink、自訂 JAR 檔）
✅ 偏好手動控制集群配置（機器類型、節點數量）

實戰範例：批次處理銷售數據

假設你每天都有銷售數據（CSV 檔放在 Cloud Storage），想算出每個產品的總銷售額：

# spark_job.py
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SalesAnalysis").getOrCreate()

# 讀取 Cloud Storage 中的 CSV
df = spark.read.csv("gs://my-bucket/sales/*.csv", header=True)

# 計算每個產品的總銷售額
result = df.groupBy("product_id").sum("amount")

# 寫回 BigQuery
result.write \
  .format("bigquery") \
  .option("table", "my-project:analytics.product_sales") \
  .save()

建立 Dataproc 集群：

gcloud dataproc clusters create my-cluster \
  --region=asia-east1 \
  --num-workers=5 \
  --worker-machine-type=n1-standard-4

提交作業：

gcloud dataproc jobs submit pyspark \
  --cluster=my-cluster \
  --region=asia-east1 \
  gs://my-bucket/spark_job.py

成本優化技巧：

使用 Preemptible VM（可中斷式 VM）作為 workers，成本降低 80%
作業完成後立即刪除集群（--max-idle 參數自動化）

選型決策樹

需要即時串流處理？
├─ Yes → Dataflow
└─ No → 需要批次處理

    團隊已熟悉 Hadoop/Spark？
    ├─ Yes → Dataproc
    └─ No → 偏好無維護體驗？
        ├─ Yes → Dataflow
        └─ No → Dataproc（但需學習成本）

關鍵原則：

優先選 Dataflow：新專案、需要串流處理、或者懶得管集群的話
選 Dataproc：手上有舊的 Hadoop/Spark 程式碼、或團隊本來就熟這些工具

Pub/Sub 訊息佇列：建構事件驅動的即時架構

什麼是 Pub/Sub？

Cloud Pub/Sub 是 GCP 的全代管訊息佇列服務，基於發布-訂閱模式（Publish-Subscribe Pattern）。

⚠️ 2025-2026 Pub/Sub 重要通知

Pub/Sub Lite 已於 2026 年 3 月 18 日正式關閉

狀態：Pub/Sub Lite 已停止服務（2026/3/18 EOL）
原因：Cloud Pub/Sub 標準版已大幅降價，Lite 版的成本優勢已不明顯
替代方案：遷移至 Cloud Pub/Sub 標準版
考試提醒：ACE 考試中看到 Pub/Sub Lite 選項，直接排除

標準版 vs Lite 版對比（Lite 已關閉）：

特性	Pub/Sub 標準版	~~Pub/Sub Lite~~（已關閉）
可用性 SLA	99.95%	已停止服務
自動擴展	✅ 全自動	已停止服務
全球部署	✅ 支援	已停止服務
定價	按訊息量計費	已停止服務
推薦度	✅ 唯一選擇	❌ 已關閉

核心概念

Publisher（發布者）
    ↓
  Topic（主題）
    ├─ Subscription A → Subscriber A
    ├─ Subscription B → Subscriber B
    └─ Subscription C → Subscriber C

特點：

解耦：發布者不需知道誰在訂閱
一對多：一個 Topic 可以有多個 Subscription
非同步：發布者發送訊息後立即返回，不等待處理完成

應用場景：事件驅動架構

場景 1：日誌聚合

假設你有 100 台伺服器，每台都會吐日誌。用 Pub/Sub 的話：

伺服器 1 → Pub/Sub Topic "logs"
伺服器 2 → Pub/Sub Topic "logs"
...
伺服器 100 → Pub/Sub Topic "logs"

Subscription "log-to-bigquery"
  → Dataflow → BigQuery（長期儲存）

Subscription "log-to-alerting"
  → Cloud Functions → 發送告警 Email

好處：

伺服器只要丟到 Pub/Sub 就好，後續怎麼處理它不用管
想加新的訂閱者（例如多接一個監控工具）隨時都行，不會動到現有系統

場景 2：微服務解耦

假設你有個電商平台，訂單服務每次都得通知庫存服務、物流服務、發票服務：

傳統做法（緊耦合）：

訂單服務 → HTTP 呼叫庫存服務
         → HTTP 呼叫物流服務
         → HTTP 呼叫發票服務

問題：

物流服務一掛掉，整個訂單流程跟著卡住
想加新服務，就得回頭改訂單服務的程式碼

使用 Pub/Sub（解耦）：

訂單服務 → Pub/Sub Topic "order-created"

Subscription "inventory" → 庫存服務
Subscription "shipping" → 物流服務
Subscription "invoicing" → 發票服務

好處：

訂單服務只要發一次訊息就好
各服務各跑各的，互不干擾
想加新服務（例如會員積分）只要加個訂閱，訂單服務一行都不用動

實戰範例：即時網站活動追蹤

假設你想記下使用者在網站上的每一次點擊，而且要即時分析：

1. 前端發送事件到 Pub/Sub

// 前端 JavaScript
function trackClick(element) {
  fetch('/api/track', {
    method: 'POST',
    body: JSON.stringify({
      event: 'click',
      element: element.id,
      timestamp: Date.now(),
    }),
  });
}

// 後端 API (Cloud Run)
const { PubSub } = require('@google-cloud/pubsub');
const pubsub = new PubSub();

app.post('/api/track', async (req, res) => {
  const topic = pubsub.topic('user-events');
  await topic.publishMessage({
    data: Buffer.from(JSON.stringify(req.body)),
  });
  res.status(200).send('Event tracked');
});

2. Dataflow 即時處理並寫入 BigQuery

# 使用前面的 Dataflow 範例
(p
 | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(
     topic='projects/my-project/topics/user-events')
 | 'Count clicks per page' >> ...
 | 'Write to BigQuery' >> ...
)

3. Looker Studio 即時儀表板

連接 BigQuery 表格，建立即時儀表板顯示：

每分鐘點擊量
最熱門頁面
使用者地理分布

延遲：從點擊到儀表板顯示，通常在 5-10 秒內。

Pub/Sub vs 其他 Message Queue

工具	優勢	適合場景
RabbitMQ	功能豐富、支援多種協定	複雜路由邏輯
Kafka	高吞吐量、日誌儲存	大數據流處理、事件溯源
Cloud Pub/Sub	全代管、自動擴展、GCP 深度整合	GCP 生態系、無維護需求
Amazon SQS	AWS 原生整合	AWS 生態系

選 Pub/Sub 的理由：

✅ 完全不用管基礎設施
✅ 自動擴展，撐得住每秒數百萬訊息
✅ 跟 BigQuery、Dataflow、Cloud Functions 接得很順
✅ 全球低延遲（跨區域也能發布-訂閱）

Vertex AI 與 AutoML：機器學習的民主化

Vertex AI 2026 年架構：三大核心

照 2026 年 5 月的最新更新，Vertex AI 現在收斂成三大核心功能：

1. Model Garden（模型花園）

功能：預訓練模型庫，包括：

🆕 2026 最新 Gemini 模型家族：

Gemini 3.1 Pro：旗艦級多模態推理模型，支援長達 200 萬 tokens 上下文
Gemini 3.1 Flash：速度與智慧最佳平衡，推薦大多數企業場景
Gemini 2.5 Pro / Flash：上一代模型，仍廣泛使用，性價比高
Gemini Computer Use：可控制電腦界面的 AI 代理（截圖→決策→操作）
Imagen 3：圖像生成
Veo：影片生成
Chirp：語音識別

🆕 2026 熱門第三方模型：

Llama 4：Meta 最新開源模型
DeepSeek R1 / V3：開源推理模型
Kimi K2：中文超長文本模型
Mistral Large 2：企業級推理模型
Anthropic Claude 4：強推理與程式碼能力

開源模型：Llama 系列、Mistral 系列、Stable Diffusion 專業模型：醫療影像分析、金融詐欺偵測

使用方式：

from google.cloud import aiplatform

aiplatform.init(project='my-project', location='us-central1')

# 使用 Gemini 3.1 Pro 進行文本生成
model = aiplatform.Model("publishers/google/models/gemini-3.1-pro")
response = model.predict(instances=[{
    "prompt": "解釋量子運算的基本原理"
}])
print(response.predictions[0]['content'])

# 使用 Gemini 3.1 Flash 進行音訊轉文字 + 分析
model_audio = aiplatform.Model("publishers/google/models/gemini-3.1-flash")
response_audio = model_audio.predict(instances=[{
    "audio": audio_bytes,  # 直接輸入音訊
    "prompt": "總結這段對話的重點"
}])

2. Model Builder（模型建構器）

包含兩種訓練方式：

A. AutoML（無程式碼訓練）

適合場景：

資料科學團隊資源有限
想快速驗證 ML 可行性（POC）
不需要複雜的模型自訂

支援的數據類型：

類型	任務	範例
圖像	分類、物體偵測、圖像分割	判斷照片中是否有受損車輛
文本	分類、情感分析、實體提取	客服訊息自動分類
表格	回歸、分類、預測	預測客戶流失率
影片	動作識別、物體追蹤	監控影片中的異常行為

實戰範例：訓練圖像分類模型

假設你手上有 10,000 張車輛照片，每張都標好了「正常」或「受損」：

上傳數據到 Cloud Storage：

gsutil -m cp -r ./car-images gs://my-bucket/car-dataset/

建立 Dataset（在 Vertex AI Console）：
- 選擇「Image」類型
- 匯入 CSV 標籤檔（格式：gs://path/to/image.jpg,normal）
訓練 AutoML 模型：
- 選擇「AutoML Image Classification」
- 設定訓練預算（如 1 小時）
- 點擊「Train」
評估模型：
- Vertex AI 會自動提供混淆矩陣、精確率、召回率
- 範例結果：Precision 95%、Recall 92%
部署到端點：

endpoint = model.deploy(
    machine_type="n1-standard-4",
    min_replica_count=1,
    max_replica_count=5
)

# 預測
prediction = endpoint.predict(instances=[{
    "content": base64.b64encode(image_bytes).decode('utf-8')
}])
print(prediction.predictions[0]['label'])  # 'normal' 或 'damaged'

成本：訓練 1 小時約 $19 美元，部署後按請求次數計費。

B. Custom Training（自訂訓練）

如果 AutoML 滿足不了你的需求（例如要特殊架構、自訂損失函數），就改用自訂訓練：

from google.cloud import aiplatform

# 準備 Docker 容器（包含訓練程式碼）
job = aiplatform.CustomTrainingJob(
    display_name="fraud-detection",
    container_uri="asia-east1-docker.pkg.dev/my-project/my-repo/fraud-model:v1"
)

# 執行訓練
job.run(
    dataset=dataset,
    model_display_name="fraud-model-v1",
    machine_type="n1-standard-8",
    accelerator_type="NVIDIA_TESLA_T4",
    accelerator_count=2
)

3. Agent Builder（代理建構器）

功能：建構對話式 AI 代理，整合搜尋與資料來源。

應用場景：

企業知識庫搜尋（如內部文件查詢）
客服聊天機器人
電商產品推薦對話

🆕 2025-2026 重大更新：Sessions 與 Memory Bank（正式推出）

根據 Vertex AI Agent Builder 發布公告，Sessions 和 Memory Bank 已經在 2026 年正式推出（GA）：

Sessions（對話會話）：

功能：追蹤多輪對話上下文，實現連貫對話
用途：記住使用者先前的問題和答案，提供更智能的回應
計費：從 2026 年 1 月 28 日起開始計費

Memory Bank（記憶庫）：

功能：長期儲存使用者偏好、對話歷史、個人化資訊
用途：跨會話記住使用者資訊，提供個人化服務
範例：記住使用者的購物偏好、語言選擇、過往問題

計費標準（2026 年 1 月 28 日起）：

Sessions：每 1,000 次會話 $1 美元
Memory Bank：每 GB 儲存 $0.10/月

實戰範例：建立具有記憶功能的客服機器人

from google.cloud import discoveryengine

# 建立數據儲存（匯入 FAQ 文件）
data_store = discoveryengine.DataStore.create(
    project='my-project',
    location='global',
    data_store_id='customer-support-faq'
)

# 上傳文件（支援 PDF、HTML、TXT）
data_store.import_documents(
    gcs_source='gs://my-bucket/faq-docs/*.pdf'
)

# 建立對話引擎
agent = discoveryengine.ConversationalSearchService.create_conversation(
    data_store_name=data_store.name
)

# 使用者提問
response = agent.converse(
    query="如何退換貨？"
)
print(response.reply.summary)  # AI 自動從文件中擷取答案

成本：按查詢次數計費，每 1,000 次查詢約 $3 美元。

AutoML vs 自訂訓練 vs 預訓練模型

方案	開發時間	精確度	成本	適合場景
預訓練模型（Model Garden）	幾小時	中等	低	通用任務（如文本摘要、圖像標註）
AutoML	1-2 天	高	中	特定領域（如車損偵測、客訴分類）
自訂訓練	1-4 週	非常高	高	複雜任務（如醫療診斷、自動駕駛）

建議流程：

先拿 Model Garden 的預訓練模型試試
精確度不夠的話，用 AutoML 快速訓一個
還是要再調的話，才投資源去做 自訂訓練

Gemini API 生成式 AI：企業級整合指南

Gemini Developer API vs Vertex AI Gemini API

Google 提供兩種 Gemini API 產品：

面向	Gemini Developer API	Vertex AI Gemini API
定位	個人開發、快速原型	企業級應用
免費額度	✅ 有免費層級（依模型而定的 RPM/RPD 限制，例如每日數百至上千次請求）	❌ 無免費額度
SLA 保證	❌	✅ 99.9% uptime
資料隱私	可能用於模型改進	保證不用於訓練
GCP 整合	基本	深度整合（BigQuery、Vertex AI）
認證方式	API Key	Google Cloud Service Account
適合場景	學習、實驗、小型專案	生產環境、企業應用

結論：

開發階段：用 Gemini Developer API（免費又快）
上線階段：換成 Vertex AI Gemini API（有企業級保障）

統一 SDK：一套程式碼，兩種 Backend

2026 年推出的 Google Gen AI SDK，讓你能輕鬆切換 backend：

from google import genai

# 使用 Gemini Developer API
client = genai.Client(api_key='YOUR_API_KEY')

# 或使用 Vertex AI backend
client = genai.Client(
    vertexai=True,
    project='my-project',
    location='us-central1'
)

# 程式碼完全相同
response = client.models.generate_content(
    model='gemini-3.1-pro',
    contents='解釋 Kubernetes 的核心概念'
)
print(response.text)

支援語言：Python、JavaScript/TypeScript、Go、Java

實戰應用：建構智能文件摘要系統

假設你有一堆法律合約文件（PDF），想讓系統自動產摘要：

from google import genai
from google.cloud import storage
import base64

client = genai.Client(
    vertexai=True,
    project='my-project',
    location='us-central1'
)

# 讀取 PDF（從 Cloud Storage）
storage_client = storage.Client()
bucket = storage_client.bucket('my-bucket')
blob = bucket.blob('contracts/contract-2026-01.pdf')
pdf_bytes = blob.download_as_bytes()

# 使用 Gemini 3.1 Pro（支援長文本 + 多模態）
response = client.models.generate_content(
    model='gemini-3.1-pro',
    contents=[
        "請摘要以下合約的關鍵條款、雙方權利義務、以及注意事項：",
        {
            "mime_type": "application/pdf",
            "data": base64.b64encode(pdf_bytes).decode('utf-8')
        }
    ]
)

print(response.text)

輸出範例：

【合約摘要】
- 甲方：ABC 公司，乙方：XYZ 供應商
- 合約期限：2026/01/01 ~ 2026/12/31
- 主要條款：
  1. 乙方每月提供不少於 1,000 件產品
  2. 甲方需於 30 天內付款
  3. 違約金為合約總額的 10%
- 注意事項：
  - 第 5 條規定不可轉包，需特別留意
  - 第 12 條有獨家供應限制

成本：Gemini 3.1 Pro 每 100 萬 tokens 約 $1.25～$10 美元（依 input/output 計費）。

企業級功能：Gemini for Google Cloud

Gemini for Google Cloud 是針對企業的生成式 AI 助理，整合到 GCP 各項服務：

1. Gemini Cloud Assist

協助開發者撰寫 Terraform、Kubernetes YAML：

你：「建立一個 GKE 集群，3 個節點，使用 e2-standard-4 機器」

Gemini Cloud Assist：
```terraform
resource "google_container_cluster" "primary" {
  name               = "my-gke-cluster"
  location           = "us-central1"
  initial_node_count = 3

  node_config {
    machine_type = "e2-standard-4"
  }
}


#### 2. **BigQuery 自然語言轉 SQL**

你：「過去 7 天銷售額最高的前 10 個產品」

Gemini in BigQuery 自動生成：

SELECT
  product_id,
  SUM(amount) AS total_sales
FROM `project.dataset.sales`
WHERE DATE(order_date) >= DATE_SUB(CURRENT_DATE(), INTERVAL 7 DAY)
GROUP BY product_id
ORDER BY total_sales DESC
LIMIT 10;


#### 3. **Colab Enterprise 的 Gemini Code Assist**

在 Jupyter notebooks 中：
- 自動補全程式碼
- 解釋複雜的數據分析
- 生成視覺化圖表程式碼

---

### OpenAI 相容模式：無痛遷移

如果你本來就在用 OpenAI API，Gemini API 有提供相容層：

```python
from openai import OpenAI

# 只需更改 base_url 和 api_key
client = OpenAI(
    base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
    api_key="YOUR_GEMINI_API_KEY"
)

# 程式碼與 OpenAI 完全相同
response = client.chat.completions.create(
    model="gemini-3.1-pro",
    messages=[
        {"role": "user", "content": "寫一個 Python 快速排序"}
    ]
)
print(response.choices[0].message.content)

好處：現有程式碼一行都不用改，把端點換掉就能試 Gemini。

實戰案例：建構智能客服系統的完整架構

接下來我們把前面這些工具全部兜起來，做一套端到端的智能客服系統。

系統需求

即時對話：使用者透過網頁或 LINE 提問
知識庫搜尋：從 FAQ、產品文件中找答案
智能回覆：使用生成式 AI 生成自然語言回覆
數據分析：分析常見問題、客戶滿意度
持續改進：根據數據優化模型

架構設計

[使用者] → [網頁/LINE Bot]
             ↓
        [Cloud Run API]
             ├─ Pub/Sub Topic "customer-queries"
             ├─ Vertex AI Agent Builder（搜尋 FAQ）
             └─ Gemini API（生成回覆）
             ↓
        [Pub/Sub Subscription]
             ↓
        [Dataflow]
             ↓
        [BigQuery]
             ↓
        [Looker Studio 儀表板]

步驟 1：建立知識庫（Vertex AI Agent Builder）

from google.cloud import discoveryengine

# 建立數據儲存
data_store = discoveryengine.DataStore.create(
    project='my-project',
    location='global',
    data_store_id='customer-faq'
)

# 匯入 FAQ 文件（支援 CSV、JSON、網站爬蟲）
data_store.import_documents(
    gcs_source='gs://my-bucket/faq/*.json'
)

# 建立搜尋引擎
search_engine = discoveryengine.Engine.create(
    project='my-project',
    location='global',
    engine_id='customer-support-search',
    data_store_ids=['customer-faq']
)

步驟 2：API 端點（Cloud Run）

from flask import Flask, request, jsonify
from google.cloud import discoveryengine, genai, pubsub

app = Flask(__name__)
genai_client = genai.Client(vertexai=True, project='my-project')
search_client = discoveryengine.SearchServiceClient()
pubsub_client = pubsub.PublisherClient()

@app.route('/chat', methods=['POST'])
def chat():
    user_query = request.json['message']

    # 1. 搜尋知識庫
    search_response = search_client.search(
        serving_config='projects/my-project/locations/global/engines/customer-support-search/servingConfigs/default_search',
        query=user_query
    )

    # 2. 提取搜尋結果
    context = "\n".join([
        result.document.derived_struct_data['snippets'][0]['snippet']
        for result in search_response.results[:3]
    ])

    # 3. 使用 Gemini 生成回覆
    response = genai_client.models.generate_content(
        model='gemini-3.1-flash',  # Flash 版本更快、更便宜
        contents=f"""
你是一個客服助理。請根據以下知識庫內容回答使用者問題：

【知識庫】
{context}

【使用者問題】
{user_query}

請用友善、專業的語氣回答，並盡量簡潔。如果知識庫中沒有相關資訊，請誠實告知。
        """
    )

    # 4. 記錄到 Pub/Sub（供後續分析）
    pubsub_client.publish(
        topic='projects/my-project/topics/customer-queries',
        data=json.dumps({
            'query': user_query,
            'response': response.text,
            'timestamp': datetime.now().isoformat()
        }).encode('utf-8')
    )

    return jsonify({'reply': response.text})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080)

步驟 3：部署到 Cloud Run

# 建立 Dockerfile
cat > Dockerfile <<EOF
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY app.py .
CMD ["python", "app.py"]
EOF

# 部署
gcloud run deploy customer-support-api \
  --source . \
  --region asia-east1 \
  --allow-unauthenticated

步驟 4：數據分析（Dataflow + BigQuery）

# dataflow_pipeline.py
import apache_beam as beam

with beam.Pipeline() as p:
    (p
     | 'Read from Pub/Sub' >> beam.io.ReadFromPubSub(
         topic='projects/my-project/topics/customer-queries')
     | 'Parse JSON' >> beam.Map(json.loads)
     | 'Extract fields' >> beam.Map(lambda x: {
         'timestamp': x['timestamp'],
         'query': x['query'],
         'response_length': len(x['response'])
     })
     | 'Write to BigQuery' >> beam.io.WriteToBigQuery(
         table='my-project:analytics.customer_queries',
         schema='timestamp:TIMESTAMP,query:STRING,response_length:INTEGER'
     )
    )

步驟 5：建立即時儀表板（Looker Studio）

連接 BigQuery 表格 analytics.customer_queries，建立圖表：

每小時提問量（折線圖）
最常見問題（表格，依出現次數排序）
平均回覆長度（單一數值）
未解決問題（篩選出回覆包含「無相關資訊」的查詢）

成本估算（每月 10,000 次查詢）

服務	用量	成本
Vertex AI Agent Builder	10,000 次搜尋	$30
Gemini 3.1 Flash	10,000 次生成	$20
Cloud Run	10,000 次請求	$2
Pub/Sub	10,000 次訊息	$0.4
Dataflow	持續運行（Streaming）	$50
BigQuery	儲存 + 查詢	$10
總計		$112

備註：如果使用 Gemini Developer API（免費額度），初期成本可降至 $60/月。

總結與下一步

本文重點回顧

✅ BigQuery：無伺服器資料倉儲，支援 PB 級分析與 AI 輔助查詢 ✅ Dataflow vs Dataproc：Serverless 串流處理 vs 代管 Hadoop/Spark 集群 ✅ Pub/Sub：事件驅動架構的核心，解耦微服務 ✅ Vertex AI：三大核心（Model Garden、Model Builder、Agent Builder） ✅ Gemini API：企業級生成式 AI，支援文本、圖像、影片多模態 ✅ 實戰案例：整合所有工具建構智能客服系統

學習路徑建議

1. 入門實作（1 週）

完成 BigQuery 的 Qwiklabs 實驗室
建立一個簡單的 Dataflow pipeline（批次處理 CSV 檔案）
使用 Gemini Developer API 建立一個簡單的聊天機器人

2. 進階實作（2-3 週）

使用 Vertex AI AutoML 訓練一個圖像分類模型
建構一個即時數據分析系統（Pub/Sub + Dataflow + BigQuery）
將 Gemini API 整合到現有應用（如客服系統、文件摘要）

3. 企業級專案（1-2 個月）

規劃完整的數據管道（ETL/ELT）
建立 ML Ops 流程（模型訓練、部署、監控）
導入 FinOps 文化，優化 GCP 成本

延伸學習資源

官方文件

實作教學

Cloud Ace Blog：繁體中文 GCP 教學
東東 GCP 教學：深入的技術解析
Google Codelabs：官方實作教學

社群資源

GCP Taiwan User Group：台灣 GCP 社群
Google Cloud Community：官方論壇

ACE 考試重點整理

必背知識點

BigQuery 是無伺服器資料倉儲，支援 SQL，適合 PB 級分析
Dataflow 是全託管 Apache Beam 管線，適合串流和批次處理
Dataproc 是託管 Spark/Hadoop，適合既有 Spark 工作負載遷移
Pub/Sub 是全託管訊息佇列，解耦生產者和消費者
Vertex AI 是統一 ML 平台，整合 AutoML 和自訂訓練
Gemini API 是 Google 最新生成式 AI 模型的存取方式

常見陷阱題

Q：需要對即時串流資料做聚合分析（如每分鐘平均溫度），選什麼？ A：Dataflow（Apache Beam 支援 windowing 和 watermark）。BigQuery 串流插入可以接收資料，但即時聚合用 Dataflow。

Q：團隊有大量既有 Spark 作業要遷移到 GCP，選什麼？ A：Dataproc。直接跑 Spark，程式碼幾乎不用改。Dataflow 需要改寫為 Beam。

Q：需要用 SQL 分析 Cloud Storage 中的 CSV 檔案，但不想載入，選什麼？ A：BigQuery External Table（聯邦查詢）。直接在 BigQuery 用 SQL 查詢 GCS 上的檔案。

準備好往 GCP 的資安世界鑽進去了嗎？

下一課 ACE-205：企業級資安防護——GCP 安全架構與合規指南，會帶你搞懂零信任架構、KMS 和 VPC Service Controls。

ACE 架構設計 — 4/5 完成查看系列全覽 →