用于 ADK 的 Google Cloud Spanner 工具¶

Supported in ADKPython v1.11.0Experimental

Google Cloud Spanner 是一个全托管的分布式数据库，支持 SQL 和向量搜索。ADK Spanner 工具让你的智能体能够探索数据库模式、运行 SQL 查询，并对你的 Spanner 数据执行向量相似性搜索。

实验性功能

此功能为实验性功能，可能会在未来的版本中更新。

可用工具¶

SpannerToolset 提供以下工具：

list_table_names：获取 GCP Spanner 数据库中的表名。
list_table_indexes：获取 GCP Spanner 数据库中的表索引。
list_table_index_columns：获取 GCP Spanner 数据库中的表索引列。
list_named_schemas：获取 Spanner 数据库的命名模式。
get_table_schema：获取 Spanner 数据库表模式和元数据信息。
execute_sql：在 Spanner 数据库中运行 SQL 查询并获取结果。
similarity_search：使用文本查询在 Spanner 中进行相似性搜索。

与智能体配合使用¶

# Copyright 2025 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
# from google.adk.sessions import DatabaseSessionService
from google.adk.tools.google_tool import GoogleTool
from google.adk.tools.spanner import query_tool
from google.adk.tools.spanner.settings import SpannerToolSettings
from google.adk.tools.spanner.settings import Capabilities
from google.adk.tools.spanner.spanner_credentials import SpannerCredentialsConfig
from google.adk.tools.spanner.spanner_toolset import SpannerToolset
from google.genai import types
from google.adk.tools.tool_context import ToolContext
import google.auth
from google.auth.credentials import Credentials

# Define constants for this example agent
AGENT_NAME = "spanner_agent"
APP_NAME = "spanner_app"
USER_ID = "user1234"
SESSION_ID = "1234"
GEMINI_MODEL = "gemini-2.5-flash"

# Define Spanner tool config with read capability set to allowed.
tool_settings = SpannerToolSettings(capabilities=[Capabilities.DATA_READ])

# Define a credentials config - in this example we are using application default
# credentials
# https://cloud.google.com/docs/authentication/provide-credentials-adc
application_default_credentials, _ = google.auth.default()
credentials_config = SpannerCredentialsConfig(
    credentials=application_default_credentials
)

# Instantiate a Spanner toolset
spanner_toolset = SpannerToolset(
    credentials_config=credentials_config, spanner_tool_settings=tool_settings
)

# Optional
# Create a wrapped function tool for the agent on top of the built-in
# `execute_sql` tool in the Spanner toolset.
# For example, this customized tool can perform a dynamically-built query.
def count_rows_tool(
    table_name: str,
    credentials: Credentials,  # GoogleTool handles `credentials`
    settings: SpannerToolSettings,  # GoogleTool handles `settings`
    tool_context: ToolContext,  # GoogleTool handles `tool_context`
):
  """Counts the total number of rows for a specified table.

  Args:
    table_name: The name of the table for which to count rows.

  Returns:
      The total number of rows in the table.
  """

  # Replace the following settings for a specific Spanner database.
  PROJECT_ID = "<PROJECT_ID>"
  INSTANCE_ID = "<INSTANCE_ID>"
  DATABASE_ID = "<DATABASE_ID>"

  query = f"""
  SELECT count(*) FROM {table_name}
    """

  return query_tool.execute_sql(
      project_id=PROJECT_ID,
      instance_id=INSTANCE_ID,
      database_id=DATABASE_ID,
      query=query,
      credentials=credentials,
      settings=settings,
      tool_context=tool_context,
  )

# Agent Definition
spanner_agent = Agent(
    model=GEMINI_MODEL,
    name=AGENT_NAME,
    description=(
        "Agent to answer questions about Spanner database and execute SQL queries."
    ),
    instruction="""
        You are a data assistant agent with access to several Spanner tools.
        Make use of those tools to answer the user's questions.
    """,
    tools=[
        spanner_toolset,
        # Add customized Spanner tool based on the built-in Spanner toolset.
        GoogleTool(
            func=count_rows_tool,
            credentials_config=credentials_config,
            tool_settings=tool_settings,
        ),
    ],
)


# Session and Runner
session_service = InMemorySessionService()

# Optionally, Spanner can be used as the Database Session Service for production.
# Note that it's suggested to use a dedicated instance/database for storing sessions.
# session_service_spanner_db_url = "spanner+spanner:///projects/PROJECT_ID/instances/INSTANCE_ID/databases/my-adk-session"
# session_service = DatabaseSessionService(db_url=session_service_spanner_db_url)

session = asyncio.run(
    session_service.create_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID
    )
)
runner = Runner(
    agent=spanner_agent, app_name=APP_NAME, session_service=session_service
)


# Agent Interaction
def call_agent(query):
    """
    Helper function to call the agent with a query.
    """
    content = types.Content(role="user", parts=[types.Part(text=query)])
    events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)

    print("USER:", query)
    for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print("AGENT:", final_response)

# Replace the Spanner database and table names below with your own.
call_agent("List all tables in projects/<PROJECT_ID>/instances/<INSTANCE_ID>/databases/<DATABASE_ID>")
call_agent("Describe the schema of <TABLE_NAME>")
call_agent("List the top 5 rows in <TABLE_NAME>")

向量相似性搜索¶

vector_store_similarity_search 工具使智能体能够对配置为向量存储的 Spanner 表执行语义搜索。此功能对于构建具有上下文感知能力的 RAG 应用至关重要；它允许 AI 模型根据语义含义而非精确关键词匹配来检索数据库上下文。通过配置 SpannerVectorStoreSettings，你的智能体可以更好地理解用户查询背后的意图，并基于最相关的 Spanner 数据来支撑其回答。

以下示例将一个 Spanner 表配置为向量存储，并将 vector_store_similarity_search 工具接入 RAG 智能体：

from google.adk.agents import LlmAgent
from google.adk.tools.spanner import SpannerCredentialsConfig, SpannerToolset
from google.adk.tools.spanner.settings import (
    Capabilities,
    SpannerToolSettings,
    SpannerVectorStoreSettings,
)

# 1. 定义带有向量存储设置的 Spanner 工具配置
my_vector_store_settings = SpannerVectorStoreSettings(
    project_id="your-gcp-project",
    instance_id="your-spanner-instance",
    database_id="your-database",
    table_name="my_products",
    content_column="productDescription",
    embedding_column="productDescriptionEmbedding",
    vector_length=768,
    vertex_ai_embedding_model_name="text-embedding-005",
    selected_columns=["productId", "productName", "productDescription"],
    nearest_neighbors_algorithm="EXACT_NEAREST_NEIGHBORS",
    top_k=3,
    distance_type="COSINE",
    additional_filter="inventoryCount > 0",
)

my_tool_settings = SpannerToolSettings(
    capabilities=[Capabilities.DATA_READ],
    vector_store_settings=my_vector_store_settings,
)

# 2. 初始化 Spanner 工具集
credentials_config = SpannerCredentialsConfig()
my_spanner_toolset = SpannerToolset(
    credentials_config=credentials_config,
    spanner_tool_settings=my_tool_settings,
    tool_filter=["vector_store_similarity_search"],
)

# 3. 在你的 RAG 智能体中使用工具集
my_rag_agent = LlmAgent(
    model="gemini-flash-latest",
    name="product_search_agent",
    instruction="""
    你是一个有用的助手，通过查找相似产品来回答用户问题。
    1. 始终使用 `vector_store_similarity_search` 工具来查找相关产品信息。
    2. 如果没有找到相关信息，请说明未找到匹配的产品。
    3. 在回答中清晰地展示相关产品详情。
    """,
    tools=[my_spanner_toolset],
)

配置¶

上面使用的 SpannerVectorStoreSettings 类定义了 vector_store_similarity_search 的运行方式。它接受以下参数：

必需参数¶

project_id：用于认证上下文的 Google Cloud 项目 ID。
instance_id：Spanner 实例 ID。
database_id：Spanner 数据库 ID。
table_name：包含向量嵌入的 Spanner 表。
embedding_column：存储向量嵌入的 ARRAY<FLOAT> 或 ARRAY<DOUBLE> 列。
content_column：包含要检索的原始文本或内容的列。
vector_length：嵌入向量的维度，必须与你的模型匹配。
vertex_ai_embedding_model_name：用于生成嵌入的模型，例如 "text-embedding-005"。

可选参数¶

selected_columns：你可以在搜索结果中包含的列列表，例如元数据或标识符。
nearest_neighbors_algorithm：你用于搜索的算法，例如 EXACT_NEAREST_NEIGHBORS 和 APPROXIMATE_NEAREST_NEIGHBORS。
- num_leaves_to_search：搜索的索引叶节点数量。仅在使用 APPROXIMATE_NEAREST_NEIGHBORS 时有效。
- vector_search_index_settings：向量索引设置。仅在使用 APPROXIMATE_NEAREST_NEIGHBORS 时需要。
top_k：每次查询检索的最近邻数量。
distance_type：用于相似度计算的距离度量，例如 COSINE 或 EUCLIDEAN。
additional_filter：在搜索期间应用的可选 SQL 过滤字符串，例如："inventoryCount > 0"。

Spanner 管理工具集¶

SpannerAdminToolset 支持对你的 Spanner 实例和数据库执行管理操作。请注意，这需要单独导入库。

请谨慎使用

此工具集可以创建、查看和修改 Spanner 实例和数据库，请谨慎授予访问权限。确保执行环境（如 Application Default Credentials 或 Service Account 密钥）仅限于授权项目，并使用最小必要的 IAM 权限，例如 roles/spanner.admin。

可用工具¶

list_instances：列出项目中的 Spanner 实例。
get_instance：获取 Spanner 实例的详细信息。
create_database：创建新的 Spanner 数据库。
list_databases：列出实例中的 Spanner 数据库。
create_instance：创建新的 Spanner 实例。
list_instance_configs：列出可用的 Spanner 实例配置。
get_instance_config：获取 Spanner 实例配置的详细信息。

配置¶

在使用此工具集之前，请设置所需的环境变量：

SPANNER_PROJECT：用于操作的 GCP 项目 ID。
SPANNER_INSTANCE（可选）：默认 Spanner 实例 ID。
SPANNER_DATABASE（可选）：默认数据库 ID。

与智能体配合使用¶

初始化 SpannerAdminToolset 以访问 Google Cloud Spanner 管理功能。然后将其传入 LlmAgent 的 tools 列表中，使你的智能体能够管理 Spanner 资源。

from google.adk.agents import LlmAgent
from google.adk.tools.spanner import SpannerAdminToolset

# 初始化 Spanner 管理工具集
spanner_admin_tools = SpannerAdminToolset()

# 将工具集注册到你的智能体，确保提供模型和指令
agent = LlmAgent(
    name="SpannerAdminAgent",
    model="gemini-flash-latest",
    instruction=(
        "你是一个得力的数据库管理员。使用 SpannerAdminToolset "
        "来管理并查询项目中的 Spanner 实例和数据库。"
    ),
    tools=[spanner_admin_tools]
)