在代理程式應用程式中使用參數化檢視區塊

如要在使用 AI 代理或大型語言模型 (LLM) 的 Bigtable 應用程式中安全地執行參數化檢視區塊，您必須建立嚴格的存取權控管機制，並傳遞頻外參數。

設定存取控管

Bigtable 邏輯檢視區塊會根據定義者的權限安全防護機制運作。 也就是說，當您查詢檢視表時，查詢會以定義檢視表的使用者權限執行，而非執行查詢的使用者權限。如要強制執行最小權限原則，請只授予應用程式服務帳戶存取檢視區塊的權限，並保留基礎來源資料表的權限。

如要設定存取控管，請按照下列步驟操作：

  1. 為應用程式建立專屬的 IAM 角色，並授予最低權限，例如 bigtable.reader 角色。
  2. 授予這個角色僅供檢視的權限。您可以使用 IAM 條件，將 bigtable.logicalViews.readRows 權限限制在特定檢視區塊。
  3. 如要嚴格控管存取權，請使用 IAM 拒絕政策，明確拒絕應用程式角色對基礎資料表的所有權限。

詳情請參閱「使用 IAM 控管 Bigtable 存取權」。

注入檢視區塊參數

在代理程式應用程式中，參數化邏輯檢視區塊的參數值 (例如使用者憑證或租戶界線)，必須由您信任的應用程式程式碼提供，而非 LLM 或使用者。這樣一來，資料庫輸入內容就不會受到不可信的使用者輸入內容和模型生成的查詢字串影響。

如要建構 AI 代理，可以使用 Agent Development Kit (ADK) 架構。ADK 提供下列元件，協助您整合參數化檢視區塊：

  • BigtableToolset：ADK 資料庫工具組，可設定及提供與資料庫相關的工具 (具體來說是 execute_sql_parameterized)，用來與 Bigtable 通訊。這組工具會自動擷取檢視區塊參數，並針對您的資料執行 SQL 查詢。
  • ToolContext：執行階段機制，可將工作階段專屬狀態和基礎架構屬性放入沙箱，讓工具解析敏感的篩選參數，而不向 LLM 公開這些參數。

下圖顯示各元件如何協同運作，將參數注入 Bigtable：

將檢視區塊參數注入 Bigtable 的程序。
圖 1. 將參數化檢視區塊整合至代理程式應用程式的程序 (按一下可放大)。

以下步驟進一步說明參數注入程序：

  1. 應用程式會驗證使用者身分，並取得經過驗證的使用者憑證和租戶機構。
  2. 應用程式收到使用者的自然語言查詢。
  3. 應用程式會執行 ADK 代理程式，並在安全的工作階段狀態中傳遞已驗證的使用者和租戶 ID。
  4. 代理程式決定查詢資料庫時，只會判斷自然語言篩選引數 (例如城市)，並叫用 execute_sql_parameterized 工具。
  5. 基礎資料庫工具會從 ToolContext 安全地擷取機密的使用者憑證和租戶界線，並針對 Bigtable 執行查詢。

以下範例說明如何使用 ADK 設定代理程式應用程式，查詢使用者的購買記錄。

設定 Bigtable 工具集

在 Python 應用程式中，使用您要帶外解析的參數名稱 (view_parameter_names)，設定 ADK BigtableToolset。這會將架構基礎架構屬性 (例如 user_id) 和應用程式工作階段變數 (例如 tenant_id) 直接對應至資料庫查詢。

import google.auth
from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.bigtable.bigtable_credentials import BigtableCredentialsConfig
from google.adk.tools.bigtable.bigtable_toolset import BigtableToolset

# 1. Initialize credentials (using Application Default Credentials here)
credentials, _ = google.auth.default()
credentials_config = BigtableCredentialsConfig(credentials=credentials)

# 2. Configure the BigtableToolset
# Passing view_parameter_names=["user_id", "tenant_id"] instructs the toolset
# to automatically extract both parameters from the ToolContext at runtime
# and inject them into the query's view_parameters.
bigtable_toolset = BigtableToolset(
    credentials_config=credentials_config,
    view_parameter_names=["user_id", "tenant_id"],
)

使用工具集初始化代理程式

直接將工具集傳遞至代理的 tools 清單。代理程式會自動偵測並公開強型別 execute_sql_parameterized 工具。

# 3. Create the agent and expose the toolset
agent = LlmAgent(
    model="MODEL_NAME",
    name="purchase_history_agent",
    description="An agent that retrieves multi-tenant purchase history.",
    instruction="You are an assistant that helps users find their purchase history within their tenant.",
    tools=[bigtable_toolset],
)

將 MODEL_NAME 替換成要使用的模型名稱，例如 gemini-2.5-flash。

執行具備代理功能的應用程式

執行代理程式時，請使用基礎架構使用者身分和應用程式機構狀態，初始化有效工作階段。

import asyncio
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

async def main():
    # 4. Initialize session service
    session_service = InMemorySessionService()

    # 5. Create a session for the authenticated user and store their specific
    # organization tenant in the state.
    authenticated_user_id = "user-anwesha-123"
    organization_tenant_id = "tenant-corp-alpha"

    session = await session_service.create_session(
        user_id=authenticated_user_id, # Resolved from tool_context.user_id
        # Resolved from tool_context.state["tenant_id"]
        state={"tenant_id": organization_tenant_id},
        app_name="purchase_history_app",
    )

    runner = Runner(
        app_name="purchase_history_app",
        agent=agent,
        session_service=session_service,
    )

    # 6. Simulate a user query
    user_query = "What did I buy in New York?"
    content = types.Content(role="user", parts=[types.Part(text=user_query)])

    # The runner runs the agent.
    # When the agent calls execute_sql_parameterized, the ADK framework resolves
    # both "user_id" and "tenant_id" out-of-band and passes them to Bigtable
    # as view parameters, completely hidden from the LLM.
    events = runner.run(
        session_id=session.id,
        user_id=session.user_id,
        new_message=content,
    )

    for event in events:
        if event.content and event.content.parts:
            print(f"Agent: {event.content.parts[0].text}")

if __name__ == "__main__":
    asyncio.run(main())

結果會列出已驗證使用者的購買記錄，並根據查詢中指定的城市篩選，且僅限於租戶邊界。

這個程序可確保安全參數是透過頻外注入，且完全不會受到語言模型操控。