Bigquery callback handler integration

Log events from LangChain and LangGraph to Google BigQuery for monitoring, auditing, and analyzing your LLM applications with real-time analytics.

BigQuery Callback Handler

CommunityPythonPreview

Google BigQuery is a serverless and cost-effective enterprise data warehouse that works across clouds and scales with your data.

The BigQueryCallbackHandler allows you to log events from LangChain and LangGraph to Google BigQuery. This is useful for monitoring, auditing, and analyzing the performance of your LLM applications.

Key features:

• LangGraph support: automatic detection of LangGraph nodes with AGENT_STARTING / AGENT_COMPLETED events and top-level INVOCATION_STARTING / INVOCATION_COMPLETED boundaries (vocabulary aligned with Google ADK’s BigQueryAgentAnalyticsPlugin)
• Auto-created analytics views: one typed-column CREATE OR REPLACE VIEW per event type (v_llm_response.usage_total_tokens instead of JSON_VALUE(...))
• Auto schema upgrade: additive ALTER TABLE ADD COLUMN on existing tables, gated by a schema-version label so it runs at most once per version
• Sub-agent attribution: the agent BigQuery column is auto-derived from langgraph_node when no explicit agent is set in metadata, so multi-agent graphs are tagged per sub-agent without any user changes
• Rich LLM telemetry: token usage (prompt_tokens / completion_tokens / total_tokens / cached_content_token_count), model_version, full usage_metadata, cache_metadata, plus llm_config (temperature, top_p, …) and tools on LLM_REQUEST
• Latency tracking: built-in latency measurement for all LLM and tool calls
• Event filtering: configurable allowlist / denylist plus an opt-in skip_internal_chain_events heuristic that drops noisy framework chains (ChannelWrite, RunnableLambda, …) without breaking trace continuity
• Graph context manager: explicit INVOCATION_* boundaries with accurate timing
• flush() between requests: drain the queue without tearing the handler down
• Real-time dashboard: FastAPI monitoring webapp with live event streaming

Preview release

The BigQuery Callback Handler is in Preview. APIs and functionality are subject to change. For more information, see the launch stage descriptions.

BigQuery Storage Write API

This feature uses the BigQuery Storage Write API, which is a paid service. For information on costs, see the BigQuery documentation.

Installation

You need to install langchain-google-community with bigquery extra dependencies. For this example, you will also need langchain-google-genai and langgraph.

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

Prerequisites

1. Google Cloud Project with the BigQuery API enabled.
2. BigQuery Dataset: Create a dataset to store logging tables before using the callback handler. The callback handler automatically creates the necessary events table within the dataset if the table does not exist.
3. Google Cloud Storage Bucket (Optional): If you plan to log multimodal content (images, audio, etc.), creating a GCS bucket is recommended for offloading large files.
4. Authentication:
   • Local: Run gcloud auth application-default login.
   • Cloud: Ensure your service account has the required permissions.

IAM Permissions

For the callback handler to work properly, the principal (e.g., service account, user account) under which the application is running needs these Google Cloud roles:

• roles/bigquery.jobUser at Project Level to run BigQuery queries.
• roles/bigquery.dataEditor at Table Level to write log/event data.
• If using GCS offloading: roles/storage.objectCreator and roles/storage.objectViewer on the target bucket.

Use with LangGraph agent

To use the BigQueryCallbackHandler with a LangGraph agent, instantiate it with your Google Cloud project ID and dataset ID. The handler creates the events table (and per-event-type analytics views) on first run. Use the graph_context() method to track top-level invocation boundaries — it emits INVOCATION_STARTING on enter and INVOCATION_COMPLETED (or INVOCATION_ERROR on exception) with accurate latency.

Pass session_id, user_id, and (optionally) agent via the metadata dictionary in the config object when invoking the agent. If agent is not set, the handler auto-derives it from metadata['langgraph_node'] so each sub-agent’s events are correctly attributed.

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 Vertex AI, or api_key for Gemini Developer API
    llm = ChatGoogleGenerativeAI(
        model="gemini-2.5-flash",
        project=project_id,  # For Vertex AI
    )
    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)

Configuration options

You can customize the callback handler using BigQueryLoggerConfig.

enabled · bool · default: "True"

To disable the handler from logging data to the BigQuery table, set this parameter to False.

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

The fields used to cluster the BigQuery table when it is automatically created.

gcs_bucket_name · str · default: "None"

The name of the GCS bucket to offload large content (images, blobs, large text) to. If not provided, large content may be truncated or replaced with placeholders.

connection_id · str · default: "None"

The BigQuery connection ID (e.g., us.my-connection) to use as the authorizer for ObjectRef columns. Required for using ObjectRef with BigQuery ML.

max_content_length · int · default: "512000"

(500 KB) The maximum length (in characters) of text content to store inline in BigQuery before offloading to GCS (if configured) or truncating.

batch_size · int · default: "1"

The number of events to batch before writing to BigQuery.

batch_flush_interval · float · default: "1.0"

The maximum time (in seconds) to wait before flushing a partial batch.

shutdown_timeout · float · default: "10.0"

Seconds to wait for logs to flush during shutdown.

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

A list of event types to log. If None, all events are logged except those in event_denylist.

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

A list of event types to skip logging.

log_multi_modal_content · bool · default: "True"

Whether to log detailed content parts (including GCS references).

table_id · str · default: "agent_events"

The default table ID to use if not explicitly provided to the callback handler constructor.

retry_config · RetryConfig · default: "RetryConfig()"

Configuration for retry logic (max retries, delay, multiplier) when writing to BigQuery fails.

queue_max_size · int · default: "10000"

The maximum number of events to hold in the internal buffer queue before dropping new events.

skip_internal_chain_events · bool · default: "False"

When True, drop CHAIN_* events emitted by framework-internal Runnables (ChannelWrite, ChannelRead, Branch, RunnableLambda, RunnableSequence, RunnableParallel, RunnableAssign, RunnablePassthrough, RunnableBinding, Pregel, __start__, __end__). Skipped runs are still registered in the trace registry so child LLM/tool events keep the real graph root as their trace_id (no broken traces). Each suppression logs a DEBUG line so the heuristic is auditable.

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

Static tags written to attributes.custom_tags on every event row. Useful for slicing dashboards by deployment, cohort, or experiment (e.g. {"env": "prod", "agent_role": "sales"}).

log_session_metadata · bool · default: "True"

When True, dumps the user-supplied RunnableConfig metadata (minus keys we already promote to first-class columns like session_id, user_id, agent, langgraph_node) under attributes.session_metadata.

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

Optional (raw_content, event_type) -> formatted hook invoked before content parsing. Useful for PII redaction or coercing custom payloads. Failures fall back to raw content with a warning — the formatter cannot break the agent.

auto_schema_upgrade · bool · default: "True"

When True, additively ALTER TABLE ADD COLUMN any new fields that future versions of this handler add to the events schema. Gated by a langchain_bq_schema_version table label so the diff runs at most once per schema version. Never drops, renames, or retypes columns.

create_views · bool · default: "True"

When True, automatically CREATE OR REPLACE per-event-type analytics views beside the events table. Each view unnests the JSON columns into typed top-level columns (see Auto-created analytics views below).

view_prefix · str · default: "v"

Prefix for auto-created view names (v_llm_request, v_tool_completed, …). Set per-table when several handler instances share one dataset to avoid collisions.

The following code sample shows how to define a configuration for the BigQuery callback handler with event filtering:

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,
)

Schema and production setup

The plugin automatically creates the table if it does not exist. However, for production, we recommend creating the table manually using the following DDL, which utilizes the JSON type for flexibility and REPEATED RECORDs for multimodal content.

Recommended 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;

Auto-created analytics views

When the handler creates the events table, it also creates one CREATE OR REPLACE VIEW per event type beside it (controlled by create_views, default True). Each view unnests the JSON columns into typed top-level columns so analytics queries don’t have to spell JSON_VALUE(...) every time:

-- 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;

The default view names (configurable via view_prefix) and the typed columns each one adds on top of the always-included columns:

View	Extra typed columns
v_invocation_starting	(none — only the always-included columns)
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

Every view also exposes the always-included columns from the raw table (timestamp, event_type, agent, session_id, invocation_id, user_id, trace_id, span_id, parent_span_id, status, error_message, is_truncated) plus three columns lifted from the attributes JSON: root_agent_name, custom_tags, session_metadata.

Auto schema upgrade

Existing tables are auto-upgraded additively when the handler’s schema gains new columns. The handler reads the table at startup and runs ALTER TABLE ADD COLUMN for any new fields, gated by a langchain_bq_schema_version table label so the diff runs at most once per schema version. Never drops, renames, or retypes columns. Disable with auto_schema_upgrade=False.

Sub-agent attribution

For multi-agent LangGraph deployments, the agent BigQuery column is auto-derived from this fallback chain:

1. metadata["agent"] — explicit user-supplied value (highest priority)
2. metadata["langgraph_node"] — the active LangGraph node, so each sub-agent’s events are tagged with the node name
3. metadata["checkpoint_ns"] — LangGraph checkpoint namespace
4. handler.graph_name — fallback for top-level INVOCATION_* events

A multi-agent graph (e.g. supervisor → TheCritic, TheMeteo, …) thus produces telemetry where each event is attributed to the originating sub-agent without any user changes.

Event types and payloads

The content column contains a JSON object specific to the event_type. The content_parts column provides a structured view of the content, especially useful for images or offloaded data.

Content Truncation

• Variable content fields are truncated to max_content_length (configured in BigQueryLoggerConfig, default 500KB).
• If gcs_bucket_name is configured, large content is offloaded to GCS instead of being truncated, and a reference is stored in content_parts.object_ref.

content always carries a summary key

Every event row’s content JSON object includes a summary string with a human-readable preview of the payload (capped at max_content_length). The summary is omitted from the per-event tables below to keep the shapes readable, but it is always present on disk.

LLM interactions

These events track the raw requests sent to and responses received from the LLM.

Event Type	Content (JSON) Structure	Attributes (JSON)
LLM_REQUEST	Chat Model: {"messages": [<dumped messages>]} Legacy Model: {"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} (the actual exception text lives in the error_message column)	{}

Sub-agent (LangGraph node) and invocation lifecycle

These events come from LangGraph’s node and graph-context lifecycle. agent is auto-derived from metadata['langgraph_node'] when no explicit agent is set, so events are tagged per sub-agent.

Event Type	Description
AGENT_STARTING / AGENT_COMPLETED / AGENT_ERROR	A LangGraph node begins / ends / errors
INVOCATION_STARTING / INVOCATION_COMPLETED / INVOCATION_ERROR	Top-level graph invocation begins / ends / errors (emitted by handler.graph_context())

Tool usage

These events track the execution of tools by the agent. The tool name is also surfaced in attributes.tool_name and (for the auto-views) as a typed tool_name column.

Event Type	Content (JSON) Structure
TOOL_STARTING	{"tool": "<name>", "input": "<input string>"} — e.g. {"tool": "get_weather", "input": "city='Paris'"}
TOOL_COMPLETED	{"tool": "<name>", "result": "<output string>"} — e.g. {"tool": "get_weather", "result": "25°C, Sunny"}
TOOL_ERROR	{"data": null} (the actual exception text lives in the error_message column)

Chain execution

These events fire for non-graph LangChain Runnable lifecycles (graph invocations and LangGraph nodes use the INVOCATION_* / AGENT_* events listed above instead).

Event Type	Content (JSON) Structure
CHAIN_START	{"data": "<JSON-stringified inputs>"}
CHAIN_END	{"data": "<JSON-stringified outputs>"}
CHAIN_ERROR	{"data": null} (see error_message column)

Retriever usage

These events track the execution of retrievers.

Event Type	Content (JSON) Structure
RETRIEVER_START	{"data": "<query string>"} — e.g. {"data": "What is the capital of France?"}
RETRIEVER_END	{"data": "<JSON-stringified list of documents>"}
RETRIEVER_ERROR	{"data": null} (the actual exception text lives in the error_message column)

Agent Actions

These events come from legacy LangChain AgentExecutor-style agents (on_agent_action / on_agent_finish). The data field contains a JSON-serialized string of the action / finish payload.

Event Type	Content (JSON) Structure
AGENT_ACTION	{"data": "{\"tool\": \"Calculator\", \"input\": \"2 + 2\"}"}
AGENT_FINISH	{"data": "{\"output\": \"The answer is 4\"}"}

Other Events

Event Type	Content (JSON) Structure
TEXT	{"data": "<text>"} — e.g. {"data": "Some logging text..."}

Advanced analysis queries

Once your agent is running and logging events, you can perform power analysis on the agent_events table.

1. Reconstruct a Trace (Conversation Turn)

Use the trace_id to group all events (Chain, LLM, Tool) belonging to a single execution flow.

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. Analyze LLM Latency & Token Usage

Calculate the average latency and total token usage for your LLM calls.

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. Analyze Multimodal Content with BigQuery Remote Model (Gemini)

If you are offloading images to GCS, you can use BigQuery ML to analyze them directly.

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. Analyze Span Hierarchy & Duration

Visualize the execution flow and performance of your agent’s operations (LLM calls, Tool usage) using span IDs.

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. Querying Offloaded Content (Get Signed URLs)

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. Advanced SQL Scenarios

These advanced patterns demonstrate how to sessionize data, analyze tool usage, and perform root cause analysis using 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;

Conversational Analytics in BigQuery

Conversational Analytics

You can also use BigQuery Conversational Analytics to analyze your agent logs using natural language. Just ask questions like:

• “Show me the error rate over time”
• “What are the most common tool calls?”
• “Identify sessions with high token usage”

Looker Studio Dashboard

You can visualize your agent’s performance using our prebuilt Looker Studio Dashboard template.

To connect this dashboard to your own BigQuery table, use the following link format, replacing the placeholders with your specific project, dataset, and table IDs:

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 integration

The BigQueryCallbackHandler provides enhanced support for LangGraph agents with automatic node detection, graph-level tracking, and latency measurements.

LangGraph event types

In addition to standard LangChain events, the callback handler automatically detects and logs LangGraph-specific events:

Event Type	Description
AGENT_STARTING	Emitted when a LangGraph node begins execution
AGENT_COMPLETED	Emitted when a LangGraph node completes successfully
AGENT_ERROR	Emitted when a LangGraph node fails
INVOCATION_STARTING	Emitted when a graph execution begins (via context manager)
INVOCATION_COMPLETED	Emitted when a graph execution completes
INVOCATION_ERROR	Emitted when a graph execution fails

Graph context manager

Use the graph_context() method to explicitly mark graph execution boundaries. This enables INVOCATION_STARTING and INVOCATION_COMPLETED events with accurate latency measurements:

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 tracking

The callback handler automatically tracks latency for all operations and stores measurements in the latency_ms JSON column:

-- 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 filtering

Use event_allowlist and event_denylist to control which events are logged:

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,
)

Or exclude noisy events:

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

Examples and resources

Example code

The following examples demonstrate various features of the BigQuery callback handler:

Example	Description
Basic example	Basic callback usage with LLM calls
LangGraph agent	Complete ReAct agent with 6 realistic tools
Async example	Async handler with concurrent queries
Event filtering	Allowlist/denylist configurations
Sample data generator	Generate sample data across multiple agent types

Analytics notebook

The LangGraph Agent Analytics notebook provides comprehensive BigQuery analytics queries for:

• Real-time event monitoring
• Tool usage analytics
• Latency analysis and trends
• Error debugging
• User engagement metrics
• Time-series visualization

Real-time monitoring dashboard

A FastAPI-based monitoring dashboard is available for real-time agent monitoring:

Features:

• Live event stream via Server-Sent Events (SSE)
• Interactive charts for event distribution and latency trends
• Session tracing with detailed timeline view
• 20+ REST API endpoints for analytics queries
• Auto-refresh every 5 seconds

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

Feedback

We welcome your feedback on BigQuery Agent Analytics. If you have questions, suggestions, or encounter any issues, please reach out to the team at [e-mail].

Additional resources

• BigQuery Storage Write API
• Introduction to Object Tables
• Example code and notebook

---

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.