使用 Spanner Omni 在设备上运行数据智能体和 GraphRAG

1. 概览

此 Codelab 涵盖了 Spanner Omni 的基础知识，以及它为何能成为自主 AI 智能体的终极数据基础。此外，我们还将大致介绍如何端到端实现一个边缘部署的智能体式匹克球战术教练和比赛模拟器，该模拟器由 Spanner Omni（关系型 + 图形 GQL）、Python Flask 和 Gemini 3.8 Flash 提供支持。

问题

现代互动式体育平台和实时游戏环境需要同时处理三种不同的工作负载：

高并发操作事务：

球场预订、球员会员资格、结算和锦标赛日程安排需要严格的 ACID 可串行性，以消除重复预订冲突。

连接图遥测：

连续对打镜头序列、过渡马尔可夫链和区域覆盖图，这些都需要低延迟遍历，而无需庞大的 ETL 流水线。

亚毫秒级边缘响应能力：

快节奏的 60 FPS 物理模拟，其中广域云网络往返（100-200 毫秒）会引入不可接受的抖动。

过去，工程团队通过将零散的架构拼接在一起来解决这个问题，这些架构大致如下：PostgreSQL 用于事务数据，Neo4j 用于镜头图，Redis 用于边缘缓存，以及后台同步流水线。这会导致架构漂移、高运营复杂性和最终一致性延迟。

我们的做法！

通过在边缘部署 Google Cloud Spanner Omni 并将其与 Google Gemini 3.8 Flash 相结合，我们构建了 Spanner Omni Pickleball Game，这是一个单引擎体育游戏平台，它将关系型业务逻辑、原生 ISO GQL 击球遥测和实时战术 AI 教练功能融为一体。

Spanner Omni

Spanner Omni 将 Google 的全球级分布式数据库引擎引入到自行管理、本地、多云和经过网闸隔离的环境中，该引擎具有横向可伸缩性、ACID 事务和多模型功能（关系型、图、向量和分析）。Spanner Omni 是 Google Cloud Spanner 的可下载、自行管理版本。无论您选择在何处运行，它都能提供 Spanner 的核心分布式数据库功能：自动分片、基于 Paxos 的复制和强大的外部一致性：

学习内容

在此 Codelab 中，您将从传统的孤立式数据库架构过渡到现代化的统一边缘数据库解决方案。您将学习：

1. 使用 Docker 在本地部署和编排 Spanner Omni，以管理高性能的单实例关系型工作负载。
2. 设计一个统一的关系型和 ISO GQL 图表架构，用于映射事务型业务模型以及高维游戏遥测数据，而无需进行外部数据库同步。
3. 在 Python 读写事务中实现严格的 ACID 可串行化，以编程方式防止竞态条件并消除重复预订冲突。
4. 构建一个原生 GraphRAG 流水线，利用与 SQL 集成的 ISO GQL 图查询，将实时、具有上下文依据的比赛遥测数据直接输入 Google Gemini，以实现即时战术指导。

要求

• 运行 Linux (Ubuntu 22.04+)、macOS 或 Windows 的标准本地机器或开发虚拟机。
• 已安装并运行 Docker Desktop 或 Docker Engine。
• 已安装 Python 3.10 或更高版本。
• 文本编辑器或 IDE（例如 VS Code 或 PyCharm）。
• 有效的 Google Gemini API 密钥。（注意：在本地运行 Spanner Omni 时，无需 Google Cloud 结算账号，因为它在免费层级的自行管理开发者评估许可下运行）。

2. Spanner Omni 部署和配置

在此步骤中，我们将启动本地边缘数据库实例。Spanner Omni 在高度优化的容器中运行，让您可以在本地开发者笔记本电脑上获得全球规模的关系型数据库功能。

1. 打开终端，创建一个永久性 Docker 卷，用于存储数据库架构和游戏遥测数据：

docker volume create spanner_omni_data

2. 拉取并启动在单服务器开发者模式下配置的官方 Spanner Omni 容器映像：

docker run -d --name spanner-omni \
  -p 9010:9010 -p 9020:9020 \
  -v spanner_omni_data:/spanner \
  us-docker.pkg.dev/spanner-omni/images/spanner-omni:2026.r2.1-beta \
  start-single-server

3. 确保容器已成功启动：

docker ps

🛠️ 快速单行命令问题排查：如果由于端口 9010 或 9020 已被后台进程或旧版模拟器运行占用而导致启动失败，请在终端中执行以下清理脚本：

pkill -f "spanner" || true
fuser -k 9010/tcp 9020/tcp 2>/dev/null || true
docker rm -f spanner-local spanner-omni spanner-omni-local 2>/dev/null || true

3. 关系型和 ISO GQL 图 DDL 架构设计

启用 Spanner Omni 后，我们就可以建立数据库架构了。此架构作为我们的单一事实来源，将关系型业务实体（例如球场、球员和预订）与快速、互联的图表遥测架构（投篮、区域和转换）统一起来。

如需创建操作表和遥测表，请执行以下操作：

1. 使用 Spanner CLI 或应用客户端连接到数据库
2. 运行以下 DDL 语句

目前您无需采取任何行动，因为我们正在应用代码中解决此问题：

-- ============================================================================
-- 1. OPERATIONAL & FACILITY MANAGEMENT (RELATIONAL ACID)
-- ============================================================================

CREATE TABLE Courts (
    court_id INT64 NOT NULL,
    name STRING(100) NOT NULL,
    location STRING(100) NOT NULL,
    surface_type STRING(50) NOT NULL,
    is_indoor BOOL NOT NULL,
    hourly_rate FLOAT64 NOT NULL,
    has_lighting BOOL NOT NULL,
    created_at TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true)
) PRIMARY KEY (court_id);

CREATE TABLE Players (
    player_id INT64 NOT NULL,
    name STRING(100) NOT NULL,
    email STRING(100) NOT NULL,
    skill_level STRING(20) NOT NULL,
    phone STRING(50),
    created_at TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true)
) PRIMARY KEY (player_id);

CREATE TABLE Bookings (
    booking_id INT64 NOT NULL,
    court_id INT64 NOT NULL,
    player_id INT64 NOT NULL,
    booking_date DATE NOT NULL,
    start_hour INT64 NOT NULL,
    end_hour INT64 NOT NULL,
    player_name STRING(100) NOT NULL,
    court_name STRING(100) NOT NULL,
    status STRING(20) NOT NULL,
    notes STRING(MAX),
    created_at TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true)
) PRIMARY KEY (booking_id);

CREATE TABLE MatchRequests (
    match_id INT64 NOT NULL,
    player_id INT64 NOT NULL,
    player_name STRING(100) NOT NULL,
    court_id INT64,
    court_name STRING(100),
    play_date DATE NOT NULL,
    start_hour INT64 NOT NULL,
    skill_required STRING(20) NOT NULL,
    match_type STRING(20) NOT NULL,
    spots_open INT64 NOT NULL,
    status STRING(20) NOT NULL,
    notes STRING(MAX),
    created_at TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true)
) PRIMARY KEY (match_id);

-- ============================================================================
-- 2. TACTICAL GAME TELEMETRY & PROPERTY GRAPH (ISO GQL)
-- ============================================================================

CREATE TABLE CourtZones (
    zone_id STRING(50) NOT NULL,
    name STRING(100) NOT NULL,
    is_kitchen BOOL NOT NULL,
    side STRING(50) NOT NULL
) PRIMARY KEY (zone_id);

CREATE TABLE Shots (
    shot_id STRING(50) NOT NULL,
    player_id STRING(50) NOT NULL,
    shot_type STRING(100) NOT NULL,
    speed_mph FLOAT64,
    spin_type STRING(50),
    outcome STRING(50),
    from_zone_id STRING(50),
    to_zone_id STRING(50)
) PRIMARY KEY (shot_id);

CREATE TABLE ShotTransitions (
    transition_id STRING(50) NOT NULL,
    from_shot_id STRING(50) NOT NULL,
    to_shot_id STRING(50) NOT NULL,
    rally_depth INT64,
    pressure_level STRING(20)
) PRIMARY KEY (transition_id);

-- ============================================================================
-- 3. MATCH SESSIONS & AI COACHING HISTORY
-- ============================================================================

CREATE TABLE GameSessions (
    session_id STRING(50) NOT NULL,
    player_id INT64 NOT NULL,
    player_score INT64 NOT NULL,
    bot_score INT64 NOT NULL,
    bot_difficulty STRING(20) NOT NULL,
    winner STRING(20) NOT NULL,
    rally_length INT64 NOT NULL,
    fault_type STRING(100),
    coaching_summary STRING(MAX),
    created_at TIMESTAMP NOT NULL OPTIONS (allow_commit_timestamp=true)
) PRIMARY KEY (session_id);

cat << 'EOF' > Dockerfile
FROM ollama/ollama:latest
ENV OLLAMA_HOST=0.0.0.0:8080
ENV OLLAMA_MODELS=/root/.ollama
ENV MODEL=gemma4:e2b
RUN ollama serve & sleep 5 && ollama pull $MODEL
ENTRYPOINT ["ollama", "serve"]
EOF

💡 为什么提交时间戳很重要：指定 OPTIONS (allow_commit_timestamp=true) 可在所有本地节点副本中实现单调、微秒级精确的排序。这对于将高速游戏事件（例如压线球视频片段）与相应的数据库击球遥测日志同步至关重要。

4. ACID 并发控制和重复预订防范

在活跃的运动场地预订门户中，多用户经常会争相预订同一场地时段。为防止竞态条件，我们将在 Spanner Omni 中编写一个 Python 读写事务，以确保绝对、防重复预订的可串行化。

您现在无需执行任何操作，只需观察代码：

from google.cloud import spanner

def create_booking_transaction(transaction, booking_data):
    # 1. Query for existing overlapping confirmed bookings on the same court
    query = """
        SELECT booking_id, player_name, start_hour, end_hour
        FROM Bookings
        WHERE court_id = @court_id
          AND booking_date = @booking_date
          AND status = 'CONFIRMED'
          AND NOT (end_hour <= @start_hour OR start_hour >= @end_hour)
    """
    params = {
        "court_id": booking_data["court_id"],
        "booking_date": booking_data["booking_date"],
        "start_hour": booking_data["start_hour"],
        "end_hour": booking_data["end_hour"],
    }
    param_types = {
        "court_id": spanner.param_types.INT64,
        "booking_date": spanner.param_types.DATE,
        "start_hour": spanner.param_types.INT64,
        "end_hour": spanner.param_types.INT64,
    }
    
    results = list(transaction.execute_sql(query, params=params, param_types=param_types))
    
    if results:
        conflict = results[0]
        raise ValueError(
            f"Court conflict: Already booked by {conflict[1]} ({conflict[2]}:00-{conflict[3]}:00)"
        )
    
    # 2. Insert new confirmed booking with Spanner commit timestamp
    insert_stmt = """
        INSERT INTO Bookings (
            booking_id, court_id, player_id, booking_date, start_hour, end_hour,
            player_name, court_name, status, notes, created_at
        ) VALUES (
            @booking_id, @court_id, @player_id, @booking_date, @start_hour, @end_hour,
            @player_name, @court_name, 'CONFIRMED', @notes, PENDING_COMMIT_TIMESTAMP()
        )
    """
    # execute_update statement writing to Spanner Omni...
    transaction.execute_update(insert_stmt, params=booking_data, param_types=booking_types)

5. 实时 60 FPS 物理引擎和可调节 DUPR 机器人引擎

应用的前端采用 2.5D HTML5 Canvas 模拟，以每秒 60 帧 (FPS) 的速度进行渲染。此渲染过程与数据库层完全解耦，以保持流畅的性能，仅在拉力赛结束时刷新遥测数据。

该模拟器配备了可调节的 OmniBot AI 引擎，该引擎利用基于真实世界 DUPR（动态通用匹克球评分）水平的难度矩阵。

您现在无需采取任何行动，只需查看详细信息即可。

6. ISO GQL 镜头遥测图处理

在每次拉力赛结束时，前端都会将连接的轨迹图发布到我们的后端端点 /api/rally-end。每个镜头都记录为一个关系节点，序列通过过渡边缘动态链接。

现在您无需执行任何操作，只需观察代码：

@app.route("/api/execute-shot", methods=["POST"])
def execute_shot():
    data = request.json or {}
    shot_id = f"shot_{int(time.time()*1000)}"
    
    # 1. Insert the Shot Node
    insert_shot_sql = """
        INSERT INTO Shots (shot_id, player_id, shot_type, speed_mph, spin_type, outcome, from_zone_id, to_zone_id)
        VALUES (@shot_id, @player_id, @shot_type, @speed, @spin, @outcome, @from_zone, @to_zone)
    """
    
    # 2. If part of a continuous rally, link with a ShotTransition Edge
    if data.get("prev_shot_id"):
        transition_id = f"trans_{int(time.time()*1000)}"
        insert_trans_sql = """
            INSERT INTO ShotTransitions (transition_id, from_shot_id, to_shot_id, rally_depth, pressure_level)
            VALUES (@trans_id, @from_shot, @to_shot, @depth, @pressure)
        """
        # Executed within a single, unified Spanner Omni transaction...

通过利用 Spanner 的原生 ISO GQL 图查询引擎，教练或分析程序可以即时查询复杂的深层跃点结构模式，而无需运行庞大的数据库联接：

-- Find highest-frequency transition sequences leading to player unforced errors
GRAPH PickleballTelemetry
MATCH (s1:Shots)-[t:ShotTransitions]->(s2:Shots)
WHERE s1.shot_type = 'Speedup' AND s2.outcome = 'FAULT_NET'
RETURN s1.speed_mph, s2.from_zone_id, COUNT(*) AS occurrences
ORDER BY occurrences DESC
LIMIT 5;

7. Gemini AI 战术指导和基于 RAG 的对话

现在，我们将激活应用的大脑：Google Gemini 3.8 Flash。我们的 Python 后端使用该模型提供三种实时 AI 功能，这些功能直接由 Spanner Omni 提供支持。

该系统使用 Google Gemini 3.8 Flash 来执行一些独特的实时 AI 工作流：

实时回合分析和战术裁判：

每得一分，Gemini 都会分析击球链、机器人难度和球场站位。

Spanner 多表接地 AI Concierge 和比赛分析师：

用户可以使用自然语言查询球场时间表、设施价格，并询问“我与 OmniBot 的比赛结果如何？”后端会跨四个网域（Courts、Bookings、GameSessions 和 Shots）动态查询 Spanner Omni，为 Gemini 提供依据，确保其不会产生幻觉。

Real-Time Performance Analytics API (/api/performance)：

直接从 Spanner Omni 的 GameSessions 表中汇总实时胜负比、平均回合深度和近期游戏会话，以便在玩家信息中心内显示。

您现在无需采取任何行动，只需查看详细信息即可。

8. 配置、运行和测试！

让我们在本地机器上启动完整的应用，以验证端到端系统。

1. 打开主机终端并克隆项目代码库：

git clone https://github.com/AbiramiSukumaran/pickleball

cd pickleball

2. 按照以下文档中的步骤获取 Gemini API 密钥：

https://medium.com/google-cloud/get-api-key-from-google-ai-studio-b2d2059a171d

3. 在编辑器中打开 .env 文件，然后配置环境详细信息：

SPANNER_EMULATOR_HOST="localhost:9010"
SPANNER_PROJECT="PROJECT_ID"
SPANNER_INSTANCE="omni-instance"
SPANNER_DATABASE="pickleball-db"
GEMINI_API_KEY="REPLACE_WITH_YOUR_ACTUAL_GEMINI_KEY"

4. 安装 Python 依赖项并运行应用：

pip install -r requirements.txt

python app.py

5. 打开网络浏览器并前往 http://localhost:5000，即可开始游戏！您可以在预约标签页中预订球场时间，与可调节的 DUPR OmniBot 对战，并在球场边咨询实时 Gemini Match Analyst。

打开网络浏览器，或点击指向在 localhost 中运行的应用的链接！

9. 恭喜！

完成此 Codelab 后，您见证了数据库工程领域的范式转变。多年来，在边缘（工厂车间、军事区域或球场边的体育网络）构建响应迅速的实时智能体，迫使人们在设计上做出痛苦的妥协：将孤立的事务系统、独立的图节点和向量搜索索引之间脆弱的 ETL 流水线拼接在一起。

Spanner Omni 打破了这种权衡。它将全球级分布式一致性、ISO GQL 图查询和关系型数据库性能整合到一个可下载的本地容器中。无论您从事哪个领域，Spanner Omni 都能确保您的边缘智能体拥有坚不可摧的标准答案。