BigQuery 回调处理器集成

将 LangChain 和 LangGraph 的事件记录到 Google BigQuery，以进行实时分析，从而监控、审计和分析您的 LLM 应用。

BigQuery 回调处理器

社区Python预览

Google BigQuery 是一个无服务器且具有成本效益的企业数据仓库，可在云中跨云工作，并随您的数据扩展。

该 BigQueryCallbackHandler 允许您将 LangChain 和 LangGraph 的事件记录到 Google BigQuery。这对于监控、审计和分析您的 LLM 应用程序的性能非常有用。

主要功能：

• LangGraph 支持：自动检测具有 AGENT_STARTING / AGENT_COMPLETED 事件和顶层 INVOCATION_STARTING / INVOCATION_COMPLETED 边界的 LangGraph 节点（词汇表与 Google ADK’s BigQueryAgentAnalyticsPlugin 对齐）
• 自动创建的分析视图：每个事件类型一个带类型列的 CREATE OR REPLACE VIEW（v_llm_response.usage_total_tokens 而不是 JSON_VALUE(...)）
• 自动模式升级：现有表上的增量 ALTER TABLE ADD COLUMN，由模式版本标签控制，因此每个版本最多运行一次
• 子智能体归因：当元数据中没有显式设置 agent 时，agent BigQuery 列自动派生自 langgraph_node，因此多智能体图可以按子智能体标记，而无需任何用户更改
• 丰富的 LLM 遥测：Token 使用（prompt_tokens / completion_tokens / total_tokens / cached_content_token_count）、model_version、完整的 usage_metadata、cache_metadata，以及 llm_config（温度、top_p、…）和 tools 在 LLM_REQUEST 上的信息
• 延迟跟踪：所有 LLM 和工具调用的内置延迟测量
• 事件过滤：可配置的允许列表/拒绝列表，以及一个可选的 skip_internal_chain_events 启发式算法，它会丢弃噪声框架链（ChannelWrite、RunnableLambda、…），而不会破坏跟踪连续性
• 图上下文管理器：具有精确时间的显式 INVOCATION_* 边界
• flush() 请求之间：在不销毁处理器的情况下清空队列
• 实时仪表板：FastAPI 监控 Web 应用程序，具有实时事件流

预览版本

BigQuery 回调处理器处于 预览 状态。API 和功能可能会发生变化。 更多信息，请参阅 发布阶段描述。

BigQuery 存储写入 API

此功能使用 BigQuery 存储写入 API，这是一个付费服务。 关于成本的信息，请参阅 BigQuery 文档。

安装

您需要安装 langchain-google-community 并添加 bigquery 依赖项。对于此示例，您还需要 langchain-google-genai 和 langgraph。

pip install "langchain-google-community[bigquery]" langchain langchain-google-genai langgraph

前置条件

1. Google Cloud 项目，并已启用 BigQuery API。
2. BigQuery 数据集：在使用回调处理器之前，创建一个数据集来存储日志表。如果数据集不存在，回调处理器会自动在该数据集中创建必要的日志表。
3. Google Cloud 存储 桶（可选）：如果您计划记录多模态内容（图像、音频等），建议创建 GCS 桶以转储大文件。
4. 身份验证：
   • 本地：运行 gcloud auth application-default login。
   • 云端：确保您的服务账户具有所需的权限。

IAM 权限

为了让回调处理器正常工作，运行应用程序的主机（例如服务账户、用户账户）需要以下 Google Cloud 角色：

• roles/bigquery.jobUser 在项目级别运行 BigQuery 查询。
• roles/bigquery.dataEditor 在表级别写入日志/事件数据。
• 如果使用 GCS 转储：roles/storage.objectCreator 和 roles/storage.objectViewer 在目标桶上。

与 LangGraph 智能体一起使用

要使用 BigQueryCallbackHandler 与 LangGraph 智能体一起使用，请使用您的 Google Cloud 项目 ID 和数据集 ID 实例化它。处理器在首次运行时创建事件表（以及每个事件类型的分析视图）。使用 graph_context() 方法跟踪顶层调用边界——它在进入时发出 INVOCATION_STARTING，并在异常时发出 INVOCATION_COMPLETED（或 INVOCATION_ERROR）并具有准确的延迟。

通过 config 对象中的 metadata 字典传递 session_id、user_id，以及（可选的）agent。如果 agent 未设置，处理器会自动从 metadata['langgraph_node'] 中派生它，以便每个子智能体的事件都能正确归因。

LangGraph 功能，如 graph_name 和 graph_context()，需要 langchain-google-community>=4.0.0。

import os
from datetime import datetime

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool
from langchain_google_community.callbacks.bigquery_callback import (
    BigQueryCallbackHandler,
    BigQueryLoggerConfig,
)
from langchain_google_genai import ChatGoogleGenerativeAI

# 1. Define tools for the agent
@tool
def get_current_time() -> str:
    """Returns the current local time."""
    now = datetime.now()
    return f"Current time: {now.strftime('%I:%M:%S %p')} on {now.strftime('%B %d, %Y')}"

@tool
def get_weather(city: str) -> str:
    """Returns the current weather for a specific city."""
    # Simulated weather data (use real API in production)
    weather_data = {
        "new york": {"temp": 22, "condition": "Clear"},
        "tokyo": {"temp": 24, "condition": "Sunny"},
        "london": {"temp": 14, "condition": "Overcast"},
    }
    city_lower = city.lower()
    if city_lower in weather_data:
        data = weather_data[city_lower]
        return f"Weather in {city.title()}: {data['temp']}°C, {data['condition']}"
    return f"Weather data for '{city}' not available."

@tool
def convert_currency(amount: float, from_currency: str, to_currency: str) -> str:
    """Convert an amount between currencies."""
    rates = {"USD": 1.0, "EUR": 1.08, "GBP": 1.27, "JPY": 0.0067}
    from_curr, to_curr = from_currency.upper(), to_currency.upper()
    if from_curr not in rates or to_curr not in rates:
        return f"Unknown currency"
    result = amount * rates[from_curr] / rates[to_curr]
    return f"{amount} {from_curr} = {result:.2f} {to_curr}"

def run_agent_example(project_id: str):
    """Run a LangGraph agent with BigQuery logging."""

    dataset_id = "agent_analytics"

    # 2. Configure the callback handler.
    # `table_id` defaults to "agent_events"; pass it explicitly only if you
    # want a different table. The handler creates the table and the
    # per-event-type `v_*` analytics views on first run.
    config = BigQueryLoggerConfig(
        batch_size=1,
        batch_flush_interval=0.5,
        custom_tags={"env": "prod", "service": "travel"},  # static tags on every event
    )

    handler = BigQueryCallbackHandler(
        project_id=project_id,
        dataset_id=dataset_id,
        config=config,
        graph_name="travel_assistant",  # used for INVOCATION_* events + root_agent_name
    )

    # 3. Create the LLM and agent
    # Use project parameter for Gemini Enterprise Agent Platform, or api_key for Gemini Developer API
    llm = ChatGoogleGenerativeAI(
        model="gemini-2.5-flash",
        project=project_id,  # For Gemini Enterprise Agent Platform
    )
    tools = [get_current_time, get_weather, convert_currency]
    agent = create_agent(llm, tools)

    # 4. Run with graph_context for INVOCATION_STARTING / INVOCATION_COMPLETED
    query = "What time is it? What's the weather in Tokyo? How much is 100 USD in EUR?"

    run_metadata = {
        "session_id": "session-001",
        "user_id": "user-123",
        # `agent` is optional — when omitted the handler falls back to
        # metadata['langgraph_node'] for per-sub-agent attribution.
    }

    with handler.graph_context("travel_assistant", metadata=run_metadata):
        result = agent.invoke(
            {"messages": [HumanMessage(content=query)]},
            config={
                "callbacks": [handler],
                "metadata": run_metadata,
            },
        )

    print(f"Response: {result['messages'][-1].content}")

    # 5. Block until pending rows are written (between request boundaries).
    # Does NOT shut the handler down — subsequent invocations still work.
    handler.flush(timeout=5.0)

    # 6. Final cleanup at process exit.
    handler.shutdown()

if __name__ == "__main__":
    project_id = os.environ.get("GCP_PROJECT_ID", "your-project-id")
    run_agent_example(project_id)

配置选项

您可以使用 BigQueryLoggerConfig 自定义回调处理器。

enabled · bool · default: "True"

要禁用处理器将数据记录到 BigQuery 表中，请将此参数设置为 False。

clustering_fields · List[str] · default: "['event_type', 'agent', 'user_id']"

自动创建 BigQuery 表时用于聚类的字段。

gcs_bucket_name · str · default: "None"

要转储到 GCS 桶的大型内容（图像、块、大文本）的名称。如果未提供，大型内容可能会被截断或替换为占位符。

connection_id · str · default: "None"

要用作 ObjectRef 列的授权者的 BigQuery 连接 ID（例如，us.my-connection）。对于使用 BigQuery ML 的 ObjectRef 需要。

max_content_length · int · default: "512000"

(500 KB) 在转储到 GCS（如果配置了）或截断之前，BigQuery 中存储的文本内容的最大长度（以字符为单位）。

batch_size · int · default: "1"

在写入 BigQuery 之前要批处理的事件数。

batch_flush_interval · float · default: "1.0"

在刷新部分批次之前要等待的最大时间（以秒为单位）。

shutdown_timeout · float · default: "10.0"

关闭期间等待日志刷新的秒数。

event_allowlist · List[str] · default: "None"

要记录的事件类型列表。如果 None，则记录所有事件，但排除 event_denylist 中的事件。

event_denylist · List[str] · default: "None"

要跳过记录的事件类型列表。

log_multi_modal_content · bool · default: "True"

是否记录详细的内容部分（包括 GCS 引用）。

table_id · str · default: "agent_events"

如果未显式提供给回调处理器构造函数，则要使用的默认表 ID。

retry_config · RetryConfig · default: "RetryConfig()"

写入 BigQuery 失败时的重试逻辑配置（最大重试次数、延迟、乘数）。

queue_max_size · int · default: "10000"

在丢弃新事件之前，内部缓冲队列中可以保留的最大事件数。

skip_internal_chain_events · bool · default: "False"

当 True 时，丢弃 CHAIN_* 事件，这些事件由框架内部的 Runnables 发出（ChannelWrite、ChannelRead、Branch、RunnableLambda、RunnableSequence、RunnableParallel、RunnableAssign、RunnablePassthrough、RunnableBinding、Pregel、__start__、__end__）。跳过的运行仍然在跟踪注册表中注册，因此子 LLM/工具事件将真实的图根作为其 trace_id（不会出现中断的跟踪）。每次抑制都会记录一条 DEBUG 行，以便启发式算法可审计。

custom_tags · dict[str, Any] · default: "{}"

每个事件行写入 attributes.custom_tags 的静态标签。适用于按部署、队列或实验切片仪表板（例如 {"env": "prod", "agent_role": "sales"}）。

log_session_metadata · bool · default: "True"

当 True 时，将用户提供的 RunnableConfig 元数据（减去我们已提升为第一类列的键，如 session_id、user_id、agent、langgraph_node）在 attributes.session_metadata 下转储。

content_formatter · Callable[[Any, str], Any] · default: "None"

可选的 (raw_content, event_type) -> formatted 钩子在内容解析前被调用。可用于 PII 匿名化或强制自定义有效载荷。失败时回退到原始内容并发出警告——格式化器不能破坏智能体。

auto_schema_upgrade · bool · default: "True"

当 True 时，加性地 ALTER TABLE ADD COLUMN 任何此处理器未来版本添加到事件模式中的新字段。由 langchain_bq_schema_version 表标签控制，以便差异化最多运行一次每个模式版本。永远不会丢弃、重命名或重新类型列。

create_views · bool · default: "True"

当 True 时，自动 CREATE OR REPLACE 每个事件类型的分析视图在事件表旁边。每个视图将 JSON 列展平为类型化的顶层列（见下文自动创建的分析视图）。

view_prefix · str · default: "v"

自动创建视图名称的前缀（v_llm_request, v_tool_completed, …）。当多个处理器实例共享一个数据集时按表设置，以避免冲突。

以下代码示例展示了如何为 BigQuery 回调处理器定义带事件过滤的配置：

from langchain_google_community.callbacks.bigquery_callback import (
    BigQueryCallbackHandler,
    BigQueryLoggerConfig,
)

# 1. Configure BigQueryLoggerConfig
config = BigQueryLoggerConfig(
    enabled=True,
    event_allowlist=["LLM_REQUEST", "LLM_RESPONSE"],  # Only log these specific events
    shutdown_timeout=10.0,  # Wait up to 10s for logs to flush on exit
    max_content_length=500,  # Truncate content to 500 characters
)

# 2. Initialize the Callback Handler
handler = BigQueryCallbackHandler(
    project_id="your-project-id",
    dataset_id="your_dataset",
    table_id="your_table",
    config=config,
)

模式和生产环境设置

如果表不存在，插件会自动创建。但对于生产环境，我们建议使用以下 DDL 手动创建表，该 DDL 使用 JSON 类型以实现灵活性，并使用 REPEATED RECORDs 来处理多模态内容。

推荐 DDL:

CREATE TABLE `your-gcp-project-id.adk_agent_logs.agent_events`
(
  timestamp TIMESTAMP NOT NULL OPTIONS(description="The UTC timestamp when the event occurred."),
  event_type STRING OPTIONS(description="The category of the event."),
  agent STRING OPTIONS(description="The name of the agent."),
  session_id STRING OPTIONS(description="A unique identifier for the conversation session."),
  invocation_id STRING OPTIONS(description="A unique identifier for a single turn."),
  user_id STRING OPTIONS(description="The identifier of the end-user."),
  trace_id STRING OPTIONS(description="OpenTelemetry trace ID."),
  span_id STRING OPTIONS(description="OpenTelemetry span ID."),
  parent_span_id STRING OPTIONS(description="OpenTelemetry parent span ID."),
  content JSON OPTIONS(description="The primary payload of the event."),
  content_parts ARRAY<STRUCT<
    mime_type STRING,
    uri STRING,
    object_ref STRUCT<
      uri STRING,
      version STRING,
      authorizer STRING,
      details JSON
    >,
    text STRING,
    part_index INT64,
    part_attributes STRING,
    storage_mode STRING
  >> OPTIONS(description="For multi-modal events, contains a list of content parts."),
  attributes JSON OPTIONS(description="Arbitrary key-value pairs."),
  latency_ms JSON OPTIONS(description="Latency measurements."),
  status STRING OPTIONS(description="The outcome of the event."),
  error_message STRING OPTIONS(description="Detailed error message."),
  is_truncated BOOLEAN OPTIONS(description="Flag indicating if content was truncated.")
)
PARTITION BY DATE(timestamp)
CLUSTER BY event_type, agent, user_id;

自动创建的分析视图

当处理器创建事件表时，它还会在旁边创建一个 CREATE OR REPLACE VIEW 每个事件类型（由 create_views 控制，默认 True）。每个视图将 JSON 列展平为 类型化的顶层列，以便分析查询不必每次都拼写 JSON_VALUE(...)：

-- With the auto-view
SELECT agent, SUM(usage_total_tokens) AS tokens
FROM `PROJECT.DATASET.v_llm_response`
GROUP BY agent;

-- Equivalent against the raw table
SELECT agent,
       SUM(CAST(JSON_VALUE(attributes, '$.usage.total_tokens') AS INT64)) AS tokens
FROM `PROJECT.DATASET.agent_events`
WHERE event_type = 'LLM_RESPONSE'
GROUP BY agent;

默认视图名称（可通过 view_prefix 配置）以及每个视图在始终包含的列之上添加的类型化列：

视图	额外类型化列
v_invocation_starting	(无——仅包含始终的列)
v_invocation_completed / v_invocation_error	total_ms
v_agent_starting	node_name, step
v_agent_completed	node_name, step, total_ms
v_agent_error	node_name, total_ms
v_llm_request	model, request_content, llm_config, tools
v_llm_response	response, usage_prompt_tokens, usage_completion_tokens, usage_total_tokens, usage_cached_tokens, context_cache_hit_rate, total_ms, ttft_ms, model_version, usage_metadata, cache_metadata
v_llm_error	total_ms
v_tool_starting	tool_name, tool_args
v_tool_completed	tool_name, tool_result, total_ms
v_tool_error	tool_name, total_ms
v_retriever_start	query
v_retriever_end / v_retriever_error	total_ms

每个视图还公开原始表的始终包含的列 (timestamp, event_type, agent, session_id, invocation_id, user_id, trace_id, span_id, parent_span_id, status, error_message, is_truncated) 以及从 attributes JSON 提取的三个列：root_agent_name, custom_tags, session_metadata。

自动模式升级

现有表在处理器的模式获得新列时自动加性升级。处理器在启动时读取表，并为任何新字段运行 ALTER TABLE ADD COLUMN，由 langchain_bq_schema_version 表标签控制，以便差异化最多运行一次 每个模式版本。永远不要丢弃、重命名或重新类型列。 使用 auto_schema_upgrade=False 禁用。

子智能体归因

对于多智能体 LangGraph 部署，agent BigQuery 列自动派生于此回退链：

1. metadata["agent"] — 显式用户提供的值（最高优先级）
2. metadata["langgraph_node"] — 活动的 LangGraph 节点，因此每个 子智能体的事件都带有节点名称标记
3. metadata["checkpoint_ns"] — LangGraph 检查点命名空间
4. handler.graph_name — 顶层 INVOCATION_* 事件的回退

因此，一个多智能体图（例如 supervisor → TheCritic, TheMeteo, …）产生的遥测数据中，每个事件都归因于发起的子智能体，而无需任何用户更改。

事件类型和有效载荷

content 列包含特定于 event_type 的 JSON 对象。content_parts 列提供了内容的结构化视图，特别适用于图像或卸载数据。

内容截断

• 可变内容字段被截断到 max_content_length（在 BigQueryLoggerConfig中配置，默认500KB）。
• 如果 gcs_bucket_name 已配置，则大型内容会转存到 GCS 而不是被截断，并在 content_parts.object_ref 中存储引用。

content 始终包含 summary 键

每个事件行的 content JSON 对象包含一个 summary 字符串，其中包含有效负载的可读预览（最多 max_content_length 字符）。 summary 在下方的每个事件表中被省略以保持结构可读，但在磁盘上始终存在。

大语言模型交互

这些事件追踪发送给大语言模型和接收到的原始响应。

事件类型	内容 (JSON) 结构	属性 (JSON)
LLM_REQUEST	对话模型: {"messages": [<dumped messages>]} 旧版模型: {"prompt": [<prompts>]}	{"tags": ["..."], "model": "...", "llm_config": {"temperature": 0.2, ...}, "tools": ["get_weather", ...]}
LLM_RESPONSE	{"response": "<generation text>"}	{"usage": {"prompt_tokens": 100, "completion_tokens": 25, "total_tokens": 125}, "model_version": "gemini-2.5-flash-001", "usage_metadata": {"cached_content_token_count": 30, ...}, "cache_metadata": {...}}
LLM_ERROR	{"data": null}（实际的异常文本位于 error_message 列中）	{}

子智能体（LangGraph 节点）和调用生命周期

这些事件来自 LangGraph 的节点和图上下文生命周期。 agent 在未设置显式 agent 时自动从 metadata['langgraph_node'] 派生，因此事件按子智能体进行标记。

事件类型	描述
AGENT_STARTING / AGENT_COMPLETED / AGENT_ERROR	LangGraph 节点开始 / 结束 / 出错
INVOCATION_STARTING / INVOCATION_COMPLETED / INVOCATION_ERROR	顶层图调用开始 / 结束 / 出错（由 handler.graph_context() 发出）

工具使用

这些事件追踪智能体执行工具的情况。工具名称也显示在 attributes.tool_name 中，并且（对于自动视图）作为类型的 tool_name 列显示。

事件类型	内容 (JSON) 结构
TOOL_STARTING	{"tool": "<name>", "input": "<input string>"} — 例如 {"tool": "get_weather", "input": "city='Paris'"}
TOOL_COMPLETED	{"tool": "<name>", "result": "<output string>"} — 例如 {"tool": "get_weather", "result": "25°C, Sunny"}
TOOL_ERROR	{"data": null}（实际的异常文本位于 error_message 列中）

链执行

这些事件在非图 LangChain Runnable 生命周期中触发（图调用和 LangGraph 节点使用上面列出的 INVOCATION_* / AGENT_* 事件）。

事件类型	内容 (JSON) 结构
CHAIN_START	{"data": "<JSON-stringified inputs>"}
CHAIN_END	{"data": "<JSON-stringified outputs>"}
CHAIN_ERROR	{"data": null}（见 error_message 列）

检索器使用

这些事件追踪检索器的执行。

事件类型	内容 (JSON) 结构
RETRIEVER_START	{"data": "<query string>"} — 例如 {"data": "What is the capital of France?"}
RETRIEVER_END	{"data": "<JSON-stringified list of documents>"}
RETRIEVER_ERROR	{"data": null}（实际的异常文本位于 error_message 列中）

智能体操作

这些事件来自旧版 LangChain AgentExecutor 风格的智能体（on_agent_action / on_agent_finish）。data 字段包含一个序列化的 JSON 字符串，表示操作 / 完成有效负载。

事件类型	内容 (JSON) 结构
AGENT_ACTION	{"data": "{\"tool\": \"Calculator\", \"input\": \"2 + 2\"}"}
AGENT_FINISH	{"data": "{\"output\": \"The answer is 4\"}"}

其他事件

事件类型	内容（JSON）结构
TEXT	{"data": "<text>"} — 例如 {"data": "Some logging text..."}

高级分析查询

当您的智能体运行并记录事件后，您可以对 agent_events 表执行功率分析。

1. 重建跟踪（对话回合）

使用 trace_id 将属于单个执行流程的所有事件（链、LLM、工具）分组。

SELECT
  timestamp,
  event_type,
  span_id,
  parent_span_id,
  -- Extract summary or specific content based on event type
  COALESCE(
    JSON_VALUE(content, '$.messages[0].content'),
    JSON_VALUE(content, '$.summary'),
    JSON_VALUE(content)
  ) AS summary,
  JSON_VALUE(latency_ms, '$.total_ms') AS duration_ms
FROM
  `your-gcp-project-id.adk_agent_logs.agent_events`
WHERE
    -- Replace with a specific trace_id from your logs
  trace_id = '019bb986-a0db-7da1-802d-2725795ab340'
ORDER BY
  timestamp ASC;

2. 分析 LLM 延迟与 Token 使用

计算您的 LLM 调用的平均延迟和总 Token 使用量。

SELECT
  JSON_VALUE(attributes, '$.model') AS model,
  COUNT(*) AS total_calls,
  AVG(CAST(JSON_VALUE(latency_ms, '$.total_ms') AS FLOAT64)) AS avg_latency_ms,
  SUM(CAST(JSON_VALUE(attributes, '$.usage.total_tokens') AS INT64)) AS total_tokens
FROM
  `your-gcp-project-id.adk_agent_logs.agent_events`
WHERE
  event_type = 'LLM_RESPONSE'
GROUP BY
  1;

3. 使用 BigQuery 远程模型（Gemini）分析多模态内容

如果您将图像卸载到 GCS，可以使用 BigQuery ML 直接分析它们。

SELECT
  logs.session_id,
  -- Get a signed URL for the image (optional, for viewing)
  STRING(OBJ.GET_ACCESS_URL(parts.object_ref, "r").access_urls.read_url) as signed_url,
  -- Analyze the image using a remote model (e.g., gemini-2.5-flash)
  AI.GENERATE(
    ('Describe this image briefly. What company logo?', parts.object_ref)
  ) AS generated_result
FROM
  `your-gcp-project-id.adk_agent_logs.agent_events` logs,
  UNNEST(logs.content_parts) AS parts
WHERE
  parts.mime_type LIKE 'image/%'
ORDER BY logs.timestamp DESC
LIMIT 1;

4. 分析跨度层级与持续时间

使用跨度 ID 可视化智能体操作的执行流程和性能（LLM 调用、工具使用）。

SELECT
  span_id,
  parent_span_id,
  event_type,
  timestamp,
  -- Extract duration from latency_ms for completed operations
  CAST(JSON_VALUE(latency_ms, '$.total_ms') AS INT64) as duration_ms,
  -- Identify the specific tool or operation
  COALESCE(
    JSON_VALUE(content, '$.tool'),
    'LLM_CALL'
  ) as operation
FROM `your-gcp-project-id.adk_agent_logs.agent_events`
WHERE trace_id = 'your-trace-id'
  AND event_type IN ('LLM_RESPONSE', 'TOOL_COMPLETED')
ORDER BY timestamp ASC;

5. 查询卸载内容（获取签名 URL）

SELECT
  timestamp,
  event_type,
  part.mime_type,
  part.storage_mode,
  part.object_ref.uri AS gcs_uri,
  -- Generate a signed URL to read the content directly (requires connection_id configuration)
  STRING(OBJ.GET_ACCESS_URL(part.object_ref, 'r').access_urls.read_url) AS signed_url
FROM `your-gcp-project-id.adk_agent_logs.agent_events`,
UNNEST(content_parts) AS part
WHERE part.storage_mode = 'GCS_REFERENCE'
ORDER BY timestamp DESC
LIMIT 10;

6. 高级 SQL 场景

这些高级模式展示了如何会话化数据、分析工具使用以及使用 BigQuery ML 执行根本原因分析。

-- 1. Sessionize Conversation History (Create View)
-- Consolidates all events into a single row per session with a formatted history.
CREATE OR REPLACE VIEW `your-project.your-dataset.agent_sessions` AS
SELECT
  session_id,
  user_id,
  MIN(timestamp) AS session_start,
  MAX(timestamp) AS session_end,
  ARRAY_AGG(
    STRUCT(timestamp, event_type, TO_JSON_STRING(content) as content, error_message)
    ORDER BY timestamp ASC
  ) AS events,
  STRING_AGG(
      CASE
          -- LLM_REQUEST carries the user's prompt under content.messages
          WHEN event_type = 'LLM_REQUEST' THEN CONCAT('User: ', JSON_VALUE(content, '$.summary'))
          WHEN event_type = 'LLM_RESPONSE' THEN CONCAT('Agent: ', JSON_VALUE(content, '$.summary'))
          WHEN event_type = 'TOOL_STARTING' THEN CONCAT('SYS: Calling ', JSON_VALUE(content, '$.tool'))
          WHEN event_type = 'TOOL_COMPLETED' THEN CONCAT('SYS: Result from ', JSON_VALUE(content, '$.tool'))
          WHEN event_type = 'TOOL_ERROR' THEN CONCAT('SYS: ERROR in ', JSON_VALUE(content, '$.tool'))
          ELSE NULL
      END,
      '\n' ORDER BY timestamp ASC
  ) AS full_conversation
FROM
  `your-project.your-dataset.agent_events`
GROUP BY
  session_id, user_id;

-- 2. Tool Usage Analysis
-- Extract tool names from content (the auto-view `v_tool_completed`
-- exposes `tool_name` directly if you'd rather skip the JSON_VALUE).
SELECT
  JSON_VALUE(content, '$.tool') AS tool_name,
  event_type,
  COUNT(*) as count
FROM `your-project.your-dataset.agent_events`
WHERE event_type IN ('TOOL_STARTING', 'TOOL_COMPLETED', 'TOOL_ERROR')
GROUP BY 1, 2
ORDER BY tool_name, event_type;

-- 3. Granular Cost & Token Estimation
-- Estimate tokens based on content character length (approx 4 chars/token)
SELECT
  session_id,
  COUNT(*) as interaction_count,
  SUM(LENGTH(TO_JSON_STRING(content))) / 4 AS estimated_tokens,
  -- Example cost: $0.0001 per 1k tokens
  ROUND((SUM(LENGTH(TO_JSON_STRING(content))) / 4) / 1000 * 0.0001, 6) AS estimated_cost_usd
FROM `your-project.your-dataset.agent_events`
GROUP BY session_id
ORDER BY estimated_cost_usd DESC
LIMIT 5;

-- 4. AI-Powered Root Cause Analysis (Requires BigQuery ML)
-- Use Gemini to analyze failed sessions
SELECT
  session_id,
  AI.GENERATE(
    ('Analyze this conversation and explain the failure root cause. Log: ', full_conversation),
    connection_id => 'your-project.us.bqml_connection',
    endpoint => 'gemini-2.5-flash'
  ).result AS root_cause_explanation
FROM `your-project.your-dataset.agent_sessions`
WHERE error_message IS NOT NULL
LIMIT 5;

BigQuery 中的对话分析

对话分析

您还可以使用 BigQuery 对话分析 通过自然语言分析您的智能体日志。 只需提问，例如：

• “显示随时间的错误率”
• “最常用的工具调用是什么？”
• “识别 Token 使用量高的会话”

Looker Studio 仪表板

您可以使用我们预制的 Looker Studio 仪表板模板 可视化智能体的性能。

要将此仪表板连接到您自己的 BigQuery 表，请使用以下链接格式，并将占位符替换为您特定的项目、数据集和表 ID：

https://lookerstudio.google.com/reporting/create?c.reportId=f1c5b513-3095-44f8-90a2-54953d41b125&ds.ds3.connector=bigQuery&ds.ds3.type=TABLE&ds.ds3.projectId=<your-project-id>&ds.ds3.datasetId=<your-dataset-id>&ds.ds3.tableId=<your-table-id>

LangGraph 集成

BigQueryCallbackHandler 为 LangGraph 智能体提供增强支持，具有自动节点检测、图级跟踪和延迟测量。

LangGraph 事件类型

除了标准的 LangChain 事件外，回调处理器会自动检测并记录 LangGraph 特定的事件：

事件类型	描述
AGENT_STARTING	当一个 LangGraph 节点开始执行时触发
AGENT_COMPLETED	当一个 LangGraph 节点成功完成时触发
AGENT_ERROR	当一个 LangGraph 节点失败时触发
INVOCATION_STARTING	当图执行开始时触发（通过上下文管理器）
INVOCATION_COMPLETED	当图执行完成时触发
INVOCATION_ERROR	当图执行失败时触发

图上下文管理器

使用 graph_context() 方法显式标记图执行边界。这使 INVOCATION_STARTING 和 INVOCATION_COMPLETED 事件能够进行精确的延迟测量：

from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain_google_community.callbacks.bigquery_callback import (
    BigQueryCallbackHandler,
    BigQueryLoggerConfig,
)

# Initialize handler with graph name
handler = BigQueryCallbackHandler(
    project_id="your-project-id",
    dataset_id="agent_analytics",
    table_id="agent_events",
    graph_name="my_agent",
)

# Create your agent
agent = create_agent(llm, tools)

# Use the graph context manager for proper INVOCATION_STARTING/INVOCATION_COMPLETED events
run_metadata = {
    "session_id": "session-123",
    "user_id": "user-456",
    "agent": "my_agent",
}

with handler.graph_context("my_agent", metadata=run_metadata):
    result = agent.invoke(
        {"messages": [HumanMessage(content="What is the weather in Tokyo?")]},
        config={
            "callbacks": [handler],
            "metadata": run_metadata,
        },
    )

延迟跟踪

回调处理器自动跟踪所有操作的延迟，并将测量结果存储在 latency_ms JSON 列中：

-- Query latency by event type
SELECT
    event_type,
    agent,
    COUNT(*) as count,
    ROUND(AVG(CAST(JSON_EXTRACT_SCALAR(latency_ms, '$.total_ms') AS FLOAT64)), 2) as avg_latency_ms,
    ROUND(APPROX_QUANTILES(CAST(JSON_EXTRACT_SCALAR(latency_ms, '$.total_ms') AS FLOAT64), 100)[OFFSET(95)], 2) as p95_latency_ms
FROM `your-project.your-dataset.agent_events`
WHERE DATE(timestamp) = CURRENT_DATE()
  AND event_type IN ('LLM_RESPONSE', 'TOOL_COMPLETED', 'INVOCATION_COMPLETED')
GROUP BY event_type, agent
ORDER BY avg_latency_ms DESC;

事件过滤

使用 event_allowlist 和 event_denylist 控制记录哪些事件：

from langchain_google_community.callbacks.bigquery_callback import (
    BigQueryCallbackHandler,
    BigQueryLoggerConfig,
)

# Production config: Only log important events
config = BigQueryLoggerConfig(
    event_allowlist=[
        "LLM_RESPONSE",
        "LLM_ERROR",
        "TOOL_COMPLETED",
        "TOOL_ERROR",
        "INVOCATION_COMPLETED",
        "INVOCATION_ERROR",
    ],
)

handler = BigQueryCallbackHandler(
    project_id="your-project-id",
    dataset_id="agent_analytics",
    config=config,
)

或排除噪声事件：

# Exclude chain events but log everything else
config = BigQueryLoggerConfig(
    event_denylist=["CHAIN_START", "CHAIN_END"],
)

示例和资源

示例代码

以下示例展示了 BigQuery 回调处理器的各种功能：

示例	描述
基本示例	基本回调使用 LLM 调用
LangGraph 智能体	完整 ReAct 智能体，包含 6 个真实工具
异步示例	异步处理器，支持并发查询
事件过滤	允许列表/拒绝列表配置
样本数据生成器	跨多种智能体类型生成样本数据

分析笔记本

LangGraph 智能体分析笔记本 提供了全面的 BigQuery 分析查询，用于：

• 实时事件监控
• 工具使用分析
• 延迟分析和趋势
• 错误调试
• 用户参与度指标
• 时间序列可视化

实时监控仪表盘

一个基于 FastAPI 的监控仪表盘可用于实时智能体监控：

功能：

• 通过 Server-Sent Events (SSE) 的实时事件流
• 交互式图表，用于事件分布和延迟趋势
• 会话跟踪，带有详细时间线视图
• 20+ REST API 端点，用于分析查询
• 每 5 秒自动刷新

# Run the dashboard
cd libs/community/examples/bigquery_callback/webapp
pip install -r requirements.txt
uvicorn main:app --port 8001
# Open http://localhost:8001

反馈

我们欢迎您对 BigQuery 智能体分析的反馈。如果您有任何问题、建议或遇到任何问题，请通过 [e-mail] 联系团队。

其他资源

• BigQuery 存储写入 API
• 对象表入门
• 示例代码和笔记本

---

官方原文：src/oss/python/integrations/callbacks/google_bigquery.mdx · 非官方翻译，译文由 AI 生成，以英文原文为准。