使用 Gemini 3 Flash 和 AlloyDB AI 构建自主供应链

1. 概览

“会读的聊天机器人”时代即将结束。我们正迈入智能体视觉时代。

在此 Codelab 中,我们将实现确定性 AI 工程,即构建不进行猜测的 AI 系统的实践。如果让标准 AI 模型数出复杂图片中的物品数量,它们往往会“产生幻觉”(猜测)。在供应链中,猜测是危险的。如果 AI 猜测您有 12 件商品,而您实际上有 15 件,则会触发代价高昂的错误。

我们将利用 Gemini 3 Flash 中的全新思考、行动、观察循环来构建自主供应链代理。它不仅会看,还会调查。

确定性架构

我们将从“盲”系统和“失忆”系统开始。您将逐一手动“唤醒”它的感官:

17191a91a9c54146.png

  1. Eyes(视觉智能体):我们为 Gemini 3 Flash 启用了代码执行功能。该模型不是预测令牌来猜测数字,而是编写 Python 代码 (OpenCV) 来确定性地统计像素。
  2. 记忆(供应商代理):我们通过 ScaNN(可扩缩的最近邻)启用 AlloyDB AI。这样,代理就可以在数百万个选项中,在几毫秒内准确回忆出某个零件的供应商。
  3. 握手(A2A 协议):我们使用标准化的 agent_card.json 实现代理对代理通信,使 Vision Agent 能够自主向 Supplier Agent 订购库存。

构建内容

  • 一种对摄像头画面执行“视觉数学”运算的 Vision Agent
  • 由 AlloyDB ScaNN 提供支持的供应商代理,可实现高速向量搜索。
  • 一个 Control Tower 前端,具有实时 WebSocket 更新功能,可直观呈现自主循环。

学习内容

  • 如何使用 Gemini API 启用 Agentic Vision 和 gemini-3-flash-preview。
  • 如何在 AlloyDB 中使用 <=>(余弦距离)运算符实现向量搜索。
  • 如何使用身份验证代理将 Cloud Shell 连接到 AlloyDB。

要求

  • 一个浏览器,例如 ChromeFirefox
  • 启用了结算功能的 Google Cloud 项目。
  • 适用于 Vision Agent 的 Gemini API 密钥(免费层级可在 Google AI Studio 中获取)。

2. 准备工作

创建项目

  1. Google Cloud Console 的项目选择器页面上,选择或创建一个 Google Cloud 项目
  2. 确保您的 Cloud 项目已启用结算功能。了解如何检查项目是否已启用结算功能
  1. 您将使用 Cloud Shell,它是在 Google Cloud 中运行的命令行环境。点击 Google Cloud 控制台顶部的“激活 Cloud Shell”。

“激活 Cloud Shell”按钮图片

  1. 连接到 Cloud Shell 后,您可以使用以下命令检查自己是否已通过身份验证,以及项目是否已设置为您的项目 ID:
gcloud auth list

大功告成!

现在,您可以开始一键设置了。下一部分将:

  • 自动打开 Cloud Shell
  • 克隆存储库
  • 通过交互式教程引导您完成整个设置过程

3. 在 Cloud Shell 中一键设置

我们已将设置流程简化为引导式 Cloud Shell 教程。所有操作均自动化完成:基础架构预配、AlloyDB 设置、Auth Proxy 配置和数据库初始填充。

启动 Cloud Shell 教程

⚠️ 重要提示 - 点击之前:点击下方按钮后,您会看到一个安全对话框,询问您是否在 Cloud Shell 中打开。此消息会在克隆代码库之前显示。

您必须:

  1. ✅ 勾选“信任代码库”复选框
  2. ✅ 点击“确认

否则,代码库将不会被克隆。

准备好了么?点击以下链接,打开包含分步教程的项目:

接下来会发生什么

  1. Cloud Shell 会打开,并预先克隆代码库
  2. 右侧会显示一个教程面板,其中包含分步说明
  3. 系统会引导您完成以下操作:
  • 获取 Gemini API 密钥(提供免费层级)
  • 在终端中设置 GCP 项目
  • 正在运行设置(检查 API,根据需要启用 API,预配 AlloyDB:约 15 分钟)
  • 进行 2 项关键代码更改(启用视觉和记忆)
  • 创建代理卡片 (A2A 协议)
  • 启动所有服务

本教程是互动式的,每个步骤都有编号,并会跟踪您的进度。

替代方案:手动设置

如果您更喜欢手动控制:

  1. 打开 Cloud Shell 并验证项目是否已设置
gcloud config get-value project
  1. 根据需要设置项目
gcloud config set project YOUR_PROJECT_ID
  1. 克隆存储库
git clone https://github.com/MohitBhimrajka/visual-commerce-gemini-3-alloydb.git
cd visual-commerce-gemini-3-alloydb
  1. 运行设置
sh setup.sh

按照设置脚本中的屏幕说明操作。

后续步骤:本教程将指导您完成剩余步骤。完成后,请继续前往第 4 部分,了解后台发生了什么

4. 幕后揭秘:身份验证代理和数据库初始配置

问题:AlloyDB 位于专用 VPC 内。Cloud Shell 位于该边界之外。无法直接连接。

修复AlloyDB Auth Proxy 会在 Cloud Shell 上从 127.0.0.1:5432 创建一个安全且经过 IAM 身份验证的隧道,连接到您的 AlloyDB 实例。如果您的实例已启用公共 IP,则代理会使用该 IP;否则,代理会通过 VPC 的专用 IP 进行连接。

setup.sh 的作用

  1. 自动检测到您的 AlloyDB 实例(集群、区域、项目)
  2. 创建了包含所有凭据(GEMINI_API_KEY、DB_PASS、AlloyDB 详细信息)的 .env
  3. 已下载并启动 Auth Proxy(如果适用,请使用 --public-ip
  4. 使用 8 个示例商品目录零件和 ScaNN 索引为数据库设定了初始值

您的 .env 文件已准备就绪。所有未来的运行都会自动加载凭据。

验证是否成功

确保您仍位于代码库根目录中

pwd  # Should end with: visual-commerce-gemini-3-alloydb

检查身份验证代理是否正在运行

ps aux | grep alloydb-auth-proxy

创建了哪些内容

  • 包含 8 个部分和 768 维嵌入的库存表
  • 用于快速向量搜索的 ScaNN 索引 (idx_inventory_scann)

5. 第 1 步:记忆(供应商代理)

供应商代理使用 AlloyDB ScaNN 记住数百万个零件。我们首先将其作为 A2A 服务器启动,然后修复矢量查询。

The Audit: The Amnesiac

如果您现在查询供应商代理(使用占位 SQL),它会返回找到的第一行,而不是最接近的匹配项。它没有相似性的概念。他患有失忆症。

启动供应商代理

A2A 服务器 (main.py) 会委托给 agent_executor.py,后者会将协议与 inventory.py 中的业务逻辑相关联。

pkill -f uvicorn #Kill all uvicorn processes

第 1 步:前往代理目录

cd agents/supplier-agent

第 2 步:安装依赖项

pip install -r requirements.txt

第 3 步:启动代理服务器

uvicorn main:app --host 0.0.0.0 --port 8082 > /dev/null 2>&1 &

> /dev/null 2>&1 & 在后台运行服务器并抑制输出,以免中断终端。

第 4 步:验证代理是否正在运行(启动后等待 2-3 秒)

curl http://localhost:8082/.well-known/agent-card.json

预期输出:包含代理配置的 JSON(应返回且无错误)

真实语义嵌入向量

在设置过程中,数据库已使用通过 Google Gen AI SDK 的 text-embedding-005 模型生成的真实语义嵌入进行初始填充。这样可确保相似度匹配准确无误,而不是随机向量。对于 13 个示例商品,种子流程大约需要 10 秒时间,使用并行嵌入生成来创建有意义的 768 维向量,以捕获每个部分的语义含义。

AlloyDB 绕行:为什么选择 ScaNN?

修复:实现 <=> 运算符

代理随附一个占位查询。我们需要启用 ScaNN 向量搜索。

第 1 步:打开商品目录文件

cd agents/supplier-agent

第 2 步:在 inventory.py 中找到 TODO

在第 47-60 行附近查找 find_supplier() 函数。您会看到:

# ============================================================
# CODELAB STEP 1: Implement ScaNN Vector Search
# ============================================================
# TODO: Replace this placeholder query with ScaNN vector search

sql = "SELECT part_name, supplier_name FROM inventory LIMIT 1;"
cursor.execute(sql)

第 3 步:将占位 SQL 替换为 ScaNN 向量搜索

删除以下两行代码:

sql = "SELECT part_name, supplier_name FROM inventory LIMIT 1;"
cursor.execute(sql)

并将其替换为:

sql = """
SELECT part_name, supplier_name
FROM inventory
ORDER BY part_embedding <=> %s::vector
LIMIT 1;
"""
cursor.execute(sql, (embedding_vector,))

此功能的作用

  • <=> 是 PostgreSQL 中的余弦距离运算符
  • ORDER BY part_embedding <=> %s::vector 可查找最接近的匹配项(距离最短 = 语义最接近)
  • %s::vector 将您的嵌入数组转换为 PostgreSQL 的向量类型
  • LIMIT 1 仅返回最接近的匹配项
  • ScaNN 索引会自动加快此查询的速度!

第 4 步:保存文件(Ctrl+S 或 Cmd+S)

现在,代理将使用语义搜索,而不是返回随机结果!

验证

测试 A2A 发现和资产清单:

curl http://localhost:8082/.well-known/agent-card.json

44500cdeeca3cee8.png

python3 -c "
from inventory import find_supplier
import json
vec = [0.1]*768
r = find_supplier(vec)
if r:
    result = {'part': r[0], 'supplier': r[1]}
    if len(r) > 2:
        result['distance'] = float(r[2]) if r[2] else None
    print(json.dumps(result))
else:
    print('No result found')
"

预期:agent-card.json 返回代理卡片。此 Python 代码段会从种子数据中返回零件和供应商。

6. 第 2 步:眼睛(视觉代理)

在数据库可访问的情况下,我们使用 Gemini 3 Flash 唤醒眼睛。Vision Agent 通过代码执行功能执行“视觉数学”运算。A2A 服务器 (main.py) 委托给 agent_executor.py,后者调用 agent.py 进行 Gemini 分析。

The Audit: The Hallucination

如果您向标准多模态模型询问“这张杂乱的图片中有多少个盒子?”,它会将图片视为静态快照并进行猜测。

  • 模型回答:“我看到大约 12 个盒子。”
  • 实际情况:有 15 个盒子。
  • 结果:供应链故障。

解决方案:唤醒“思考-行动-观察”循环

我们启用了代码执行ThinkingConfig,以便模型编写 Python (OpenCV) 来确定性地进行计数。

  1. 打开 agents/vision-agent/agent.py。
  2. 找到 GenerateContentConfig 部分。
  3. 取消注释 thinking_config=types.ThinkingConfig(...) 块和 tools=[types.Tool(code_execution=...)]。
  4. 客户端已配置为使用环境变量中的 GEMINI_API_KEY。

文件:agents/vision-agent/agent.py

config = types.GenerateContentConfig(
    temperature=0,
    # CODELAB STEP 1: Uncomment to enable reasoning
    thinking_config=types.ThinkingConfig(
        thinking_level="LOW",  # Valid: "MINIMAL", "LOW", "MEDIUM", "HIGH"
        include_thoughts=False    # Set to True for debugging
    ),
    # CODELAB STEP 2: Uncomment to enable code execution
    tools=[types.Tool(code_execution=types.ToolCodeExecution)]
)

为什么 thinking_level="LOW"?

对于此特定任务(通过执行代码来统计商品数量),“低”可提供足够的推理预算来:

  • 规划 Python 脚本结构
  • 确定要使用的图片处理方法
  • 验证数量是否与边界框数量一致

使用“高”会增加 2-3 倍的延迟时间和费用,但不会提高确定性任务的准确率。对于复杂的多步推理(例如,“分析此次供应链中断,并推荐 3 个备选供应商,附上理由。”

成本效益优化是生产 AI 工程的关键技能:根据任务复杂程度调整推理深度。

启动 Vision 代理

🔄 路径检查:如果您仍在 agents/supplier-agent/ 中,请先使用 cd ../.. 返回到代码库根目录

第 1 步:前往 Vision 代理目录

cd agents/vision-agent

第 2 步:安装依赖项

pip install -r requirements.txt

第 3 步:启动视觉代理服务器

uvicorn main:app --host 0.0.0.0 --port 8081 > /dev/null 2>&1 &

> /dev/null 2>&1 & 在后台运行服务器并抑制输出,以免中断终端。

验证

测试 A2A 发现

curl http://localhost:8081/.well-known/agent-card.json

预期:包含代理名称和技能的 JSON。您将在第 8 步中使用控制塔界面测试实际的视觉计数。

dc9bc53007336472.png

7. 第 3 步:握手(A2A 智能体卡片)

我们的代理会看到问题(视觉),并知道供应商(记忆)。A2A 协议支持动态发现,即前端通过读取每个代理的卡片来了解如何与该代理通信。

A2A 与传统 REST API

传统 REST

A2A 协议

端点发现

配置中的硬编码网址

通过 /.well-known/agent-card.json 实现动态

功能说明

API 文档(面向人类)

技能(机器可读)

集成

每项服务的手动代码

语义匹配:“我需要库存搜索”→ 发现技能

已添加新代理

更新所有客户端的配置

零配置 - 自动发现

实际益处:在传统微服务中,如果您添加了第三个“物流代理”,则需要使用其网址和 API 契约来更新控制塔的代码。借助 A2A,控制塔会自动发现智能体,并通过自然语言技能说明了解其功能。

因此,A2A 支持即插即用型智能体组合,这是自主系统的架构模式。

解决方法:创建代理卡片

我们需要定义供应商代理可以执行哪些操作。

  1. 将 agents/supplier-agent/agent_card_skeleton.json 复制到 agents/supplier-agent/agent_card.json。
  2. 修改文件以替换占位符:

之前(框架)

{
  "name": "___FILL: agent-name ___",
  "description": "___FILL: what-this-agent-does ___"
}

修改后

{
  "name": "Acme Supplier Agent",
  "description": "Autonomous fulfillment for industrial parts via AlloyDB ScaNN.",
  "version": "1.0.0",
  "skills": [{
    "id": "search_inventory",
    "name": "Search Inventory",
    "description": "Searches the warehouse database for semantic matches using AlloyDB ScaNN vector search.",
    "tags": ["inventory", "search", "alloydb"],
    "examples": ["Find stock for Industrial Widget X-9", "Who supplies ball bearings?"]
  }]
}
  1. 重启供应商代理,以便加载新卡片:

第 1 步:停止正在运行的代理

pkill -f "uvicorn main:app.*8082"

第 2 步:前往代理目录

cd agents/supplier-agent

第 3 步:重新启动代理

uvicorn main:app --host 0.0.0.0 --port 8082 > /dev/null 2>&1 &

> /dev/null 2>&1 & 在后台运行服务器并抑制输出,以免中断终端。

第 4 步:验证新的客服卡片(开始后等待 2-3 秒)

curl http://localhost:8082/.well-known/agent-card.json

预期输出:包含您填写完毕的名称、说明和技能的 JSON。

dd352ca2e7e6109a.png

8. 第 4 步:控制塔

使用 FastAPI + WebSockets 运行 Control Tower 前端。它通过 A2A 发现代理,并通过实时更新编排整个循环。

启动所有服务

启动所有服务的最简单方法

验证您是否位于代码库根目录中

pwd  # Should end with: visual-commerce-gemini-3-alloydb

然后进行以下操作:

sh run.sh

此单个命令会启动

  • AlloyDB Auth Proxy(如果未运行)
  • 端口 8081 上的 Vision Agent
  • 端口 8082 上的供应商代理
  • 端口 8080 上的 Control Tower

等待约 10 秒,让所有服务完成初始化。

测试系统

访问 Control Tower

  1. 点击 Cloud Shell 工具栏中的网页预览按钮(眼睛图标 👁️)
  2. 选择“在端口 8080 上预览
  3. 系统会在新标签页中打开 Control Tower 信息中心

运行演示

  1. 右上角:连接状态(绿色“直播”圆点)、演示/自动模式切换开关和音频控件
  2. 中心:主要工作流画布,可用于上传图片和直观呈现分析结果
  3. 侧边栏(在分析期间显示):工作流时间轴(左侧)、进度跟踪和代码查看器(右侧)

方法 1:快速入门(推荐)

  1. 在首页上,您会看到一个“快速入门”部分,其中包含示例图片
  2. 点击任意示例图片即可自动开始分析
  3. 观看自主工作流(约 30-45 秒)

方法 2:自行上传

  1. 拖放仓库/货架图片(PNG、JPG,大小不超过 10MB),或点击“浏览”
  2. 点击“Initiate Autonomous Workflow”(启动自主工作流)
  3. 观察 4 阶段流水线

会发生什么情况

  1. 智能体发现:A2A 协议模态框会显示 Vision Agent 和 Supplier Agent 卡片,其中包含技能和端点
  2. 视觉分析:Gemini 3 Flash 生成并执行 Python 代码 (OpenCV) 来统计商品数量。进度条显示子步骤。检测到的项目上的边界框叠加层。结果徽章显示“✓ 经过代码验证”或“~ 估计”
  3. 供应商匹配:AlloyDB ScaNN 向量搜索动画。搜索查询显示(例如,“工业金属盒”)。结果卡片显示匹配的零件、供应商和置信度分数
  4. 订单已下达:包含订单 ID、数量和详细信息的收据卡片

提示:请保持演示模式开启状态(右上角),以便在演示时在每个阶段暂停。在 AUTO 模式下,工作流会持续运行。

1a031c4fd407a183.png

刚刚发生了什么

控制塔使用 A2A 协议通过 /.well-known/agent-card.json 发现两个代理,编排视觉分析(Gemini 3 Flash,可执行代码),执行向量搜索 (AlloyDB ScaNN),并下达自主订单,所有这些操作均通过实时 WebSocket 更新完成。每个代理都通过 A2A 标准公开其功能,从而实现即插即用的组合,而无需自定义 SDK。了解详情:A2A 协议

问题排查

与路径相关的错误

  • 运行命令时显示“No such file or directory”:您不在代码库根目录下。
# Check where you are
pwd

# If you're lost, navigate to home and back to repo
cd
cd visual-commerce-gemini-3-alloydb

服务错误

  • Address already in use”(地址已被使用):之前运行的进程仍处于活跃状态。
# Kill all services and restart
pkill -f uvicorn
sh run.sh  # Or manually restart individual agents
  • 服务未启动:检查端口是否被占用:
# Check which processes are using the ports
lsof -i :8080  # Control Tower
lsof -i :8081  # Vision Agent
lsof -i :8082  # Supplier Agent
  • AlloyDB 拒绝连接:验证 Auth Proxy 是否正在运行:
ps aux | grep alloydb-auth-proxy

AlloyDB 连接问题

如果您看到“与服务器 127.0.0.1 的连接失败,端口为 5432”:

如果您看到与 127.0.0.1 上端口 5432 的服务器的连接失败,请执行以下操作

  1. 检查 Auth Proxy:ps aux | grep alloydb-auth-proxy
  2. 验证是否已启用公共 IP:gcloud alloydb instances describe INSTANCE_NAME –cluster=CLUSTER_NAME –region=us-central1 –format="value(ipAddress)"
  3. 对于本地开发(而非 Cloud Shell)
  4. 问题:Cloud Shell 可自动运行,但本地机器需要授权网络
  5. 解决方案:重新运行 sh setup.sh,并在出现提示时选择选项 1(授权 0.0.0.0/0)
  6. 安全注意事项:即使使用 0.0.0.0/0,连接也需要:
  7. 有效的 GCP 凭据(应用默认凭据)
  8. 数据库密码
  9. mTLS 加密(由 Auth Proxy 处理)

9. 清理

为避免产生费用,请使用自动清理脚本销毁所有资源

# From repo root
sh cleanup.sh

此操作可安全移除

  • AlloyDB 集群(主要费用驱动因素)
  • Cloud Run 服务(如果已部署)
  • 关联的服务账号

在删除任何内容之前,脚本会提示您进行确认。

10. 参考资料和拓展阅读材料

本 Codelab 中的所有技术声明均已通过 Google Cloud 和 Google AI 官方文档验证。

官方文档

Gemini 3 Flash

AlloyDB AI 和 ScaNN

价格信息

经过验证的性能声明

功能

领取

来源

ScaNN 与 HNSW(过滤后)

速度提升 10 倍

Google Cloud 博客(已验证)

ScaNN 与 HNSW(标准)

速度达到原来的 4 倍

Google Cloud 博客(已验证)

ScaNN 内存占用

小 3-4 倍

Google Cloud 博客(已验证)

ScaNN 索引构建时间

速度达到原来的 8 倍

Google Cloud 博客(已验证)

代码执行超时

最长 30 秒

Google Cloud 文档(已验证)

代码执行文件 I/O

不支持

Google Cloud 文档(已验证)

温度为 0 时的行为

确定性输出

已经过社区验证

其他资源

Agent-to-Agent (A2A) 协议

  • A2A 可标准化代理发现和通信
  • /.well-known/agent-card.json提供的代理卡片
  • 自主智能体协作的新兴标准

ScaNN 研究

  • 基于 12 年的 Google 研究
  • 为 Google 搜索和 YouTube 提供支持,规模达数十亿
  • 正式版发布时间:2024 年 10 月
  • 第一个适用于百万到十亿个向量的 PostgreSQL 向量索引

11. 挑战模式:提升智能体技能

您已构建一个可正常运行的自主供应链。准备好进一步探索了吗?这些挑战会将您学到的模式应用于新问题。

挑战 1:基于图片的搜索(多模态嵌入)

当前流程:视觉智能体清点商品数量 → 生成文本查询 → 供应商智能体嵌入文本 → 搜索 AlloyDB

挑战:完全绕过文本,直接将剪裁后的图片发送给供应商代理。

提示

  1. Vision Agent 的代码执行可以从货架图片中剪裁单个商品
  2. Vertex AI 的 multimodalembedding@001 模型可以直接嵌入图片
  3. 修改 inventory.py 以接受图片字节,而不是文本
  4. 更新了 A2A 技能说明,以表明“接受:image/jpeg 或 text”

重要性:对于外观复杂的零件(颜色变化、损坏、包装差异),视觉搜索的准确性更高。

挑战 2:可观测性 - 以透明赢得信任

当前状态:系统正常运行,但您无法查看“幕后”情况

挑战:检查 AlloyDB 的查询日志,以证明向量搜索正在执行。

步骤

  • 默认情况下,AlloyDB 会启用 Query Insights。如需对此进行验证,请运行以下命令:
gcloud alloydb instances describe INSTANCE_NAME \
  --cluster=CLUSTER_NAME \
  --region=us-central1 \
  --format="value(queryInsightsConfig.queryPlansPerMinute)"
  • 通过界面运行供应商搜索
  • 查看实际执行的 SQL:
gcloud logging read \
  'resource.type="alloydb.googleapis.com/Instance" AND textPayload:"ORDER BY part_embedding"' \
  --limit 5 \
  --format=json

预期输出:您将看到确切的 ORDER BY part_embedding <=> $1::vector LIMIT 1 查询以及执行时间。

重要性:可观测性有助于建立信任。当利益相关方询问“此代理如何做出决策?”时,您可以向他们展示查询计划,而不仅仅是输出。

挑战 3:多智能体组合

挑战:添加第三个代理(物流代理),该代理可根据仓库位置和商品重量计算运费。

架构

  • Vision Agent 输出:商品数量
  • 供应商代理的输出:供应商位置
  • 物流代理(新)输入:目的地、重量 → 输出:运费 + 预计到达时间

提示:A2A 协议可轻松实现此目的,只需创建一个具有 calculate_shipping 技能的新代理卡片即可。控制塔会自动发现该账号。

您正在学习的模式:这是面向智能体的架构的核心,即由小型、可组合的专家构建的复杂系统。

12. 总结

您已成功从生成式 AI 改用智能体 AI

我们构建了:

  • Vision:我们已将“猜测”替换为代码执行(通过 API 密钥使用 Gemini 3 Flash)。
  • 内存:我们将“慢速搜索”替换为 AlloyDB ScaNN(通过 GCP)。
  • 操作:我们将“API 集成”替换为 A2A 协议

混合架构的优势:

此 Codelab 演示了一种混合方法:

  • Vision Agent:使用 Gemini API(API 密钥)- 简单易用,提供免费层级,无需 GCP 结算
  • 供应商代理:使用 GCP (Vertex AI + AlloyDB) - 企业级、符合法规要求

这是自治经济的架构。您可以保留此代码。

后续步骤