티스토리 수익 글 보기
# BigQuery Agent Analytics plugin for ADK
Supported in ADKPython v1.21.0Java v1.5.0Kotlin v0.8.0
The BigQuery Agent Analytics Plugin significantly enhances Agent Development Kit (ADK) by providing a robust solution for in-depth agent behavior analysis. Using the ADK Plugin architecture and the **BigQuery Storage Write API**, it captures and logs critical operational events directly into a Google BigQuery table, empowering you with advanced capabilities for debugging, real-time monitoring, and comprehensive offline performance evaluation.
The plugin also provides **Auto Schema Upgrade** (safely add new columns to existing tables), **Tool Provenance** tracking (LOCAL, MCP, SUB_AGENT, A2A, TRANSFER_AGENT, TRANSFER_A2A), **HITL Event Tracing** for human-in-the-loop interactions, and **Automatic View Creation** (generate flat, query-friendly event views).
Support for **ADK 2.0** multi-agent workflows extends tracing to agent transfers, state checkpoints, event compaction, and long-running tools. It adds four new event types: `AGENT_TRANSFER`, `AGENT_STATE_CHECKPOINT`, `EVENT_COMPACTION`, and `TOOL_PAUSED`. It also stamps an `attributes.adk` envelope on every row so you can reconstruct the agent execution graph and join a paused tool to the row that resumes it. In **Java**, this support currently covers the `TOOL_PAUSED` event and its pause/resume pairing keys only (without the `attributes.adk` envelope). See [Agent workflow and pause/resume events (ADK 2.0)](#adk-2-events) for details.
The plugin includes three reliability and observability fixes (Java: v1.7.0 or later):
– **Cross-region Storage Write API routing.** Writes to BigQuery datasets outside the `US` multi-region (for example `EU` or `northamerica-northeast1`) now route to the region that owns the write stream. Previously they could fail with a “session not found” / stream-not-found error and silently drop every row.
– **Delivery and content-incident observability.** Delivery losses are tracked per reason. Python also counts formatter and parser failures where a sentinel row is written. The counters are exposed through `BigQueryAgentAnalyticsPlugin.get_drop_stats()` (Python) or `getDropStats()` (Java), so a host can poll and export them to its own monitoring. The reason keys and semantics differ per language; see [Dropped-event observability](#dropped-event-observability).
– **No duplicate spans in Cloud Trace.** When Agent Engine telemetry (`GOOGLE_CLOUD_AGENT_ENGINE_ENABLE_TELEMETRY=true`) or any other Cloud Trace exporter is wired to the global tracer provider, the plugin no longer produces a duplicate span next to each framework span. The plugin still inherits `trace_id` from the ambient OTel span, so BigQuery rows continue to join cleanly to Cloud Trace traces.
In Python v2.7.0 and later, every row also receives a stable `event_id` before it enters the write queue. The ID is preserved across Storage Write API retries, so consumers can identify retry duplicates. An opt-in `exactly_once_delivery` mode uses committed streams and explicit offsets to prevent duplicates from ambiguous retries within a live processor. This mode does not guarantee lossless delivery; see [Delivery and deduplication](#delivery-and-deduplication).
The same Python release adds model and workflow termination details. Final `LLM_RESPONSE` rows include `finish_reason` and, when supplied by the model, a sanitized `error_message`. Workflow nodes can emit `NODE_OUTPUT` and `NODE_ERROR`, and unhandled agent or run exceptions emit `AGENT_ERROR` and `INVOCATION_ERROR`.
BigQuery Storage Write API
This feature uses **BigQuery Storage Write API**, which is a paid service. For information on costs, see the [BigQuery documentation](https://cloud.google.com/bigquery/pricing?e=48754805&hl=en#data-ingestion-pricing).
Kotlin support
The **Kotlin** plugin logs invocation lifecycle events. It writes an `INVOCATION_STARTING` row when an invocation begins and an `INVOCATION_COMPLETED` row when it ends, and creates the partitioned, clustered events table on first use if it does not already exist.
Rows are inserted one at a time through `tabledata.insertAll`, synchronously on the invocation path, rather than through the Storage Write API used by Python and Java.
The following are not implemented in Kotlin: LLM, tool, agent, state, HITL and A2A events; the ADK 2.0 workflow events; automatic view creation; Auto Schema Upgrade; tool provenance; GCS offloading; and drop statistics.
## Use cases
– **Agent workflow debugging and analysis:** Capture a wide range of *plugin lifecycle events* (LLM calls, tool usage) and *agent-yielded events* (user input, model responses), into a well-defined schema.
– **High-volume analysis and debugging:** Logging operations are performed asynchronously using the Storage Write API to allow high throughput and low latency.
– **Multimodal Analysis**: Log and analyze text, images, and other modalities. Large files are offloaded to GCS, making them accessible to BigQuery ML via Object Tables.
– **Distributed Tracing**: Built-in support for OpenTelemetry-style tracing (`trace_id`, `span_id`) to visualize agent execution flows.
– **Tool Provenance**: Track the origin of each tool call (local function, MCP server, sub-agent, A2A remote agent, or transfer agent).
– **Human-in-the-Loop (HITL) Tracing**: Dedicated event types for credential requests, confirmation prompts, and user input requests.
– **Agent Workflow Tracing** (ADK 2.0): Capture agent transfers, state checkpoints, event compaction, and long-running tool pause/resume, with an `attributes.adk` envelope for reconstructing the execution graph.
– **Queryable Event Views**: Automatically create flat, per-event-type BigQuery views (e.g., `v_llm_request`, `v_tool_completed`) to simplify downstream analytics by unnesting JSON payload data.
### Captured events summary
The following table lists all event types the plugin logs. For detailed payload examples, see [Event types and payloads](#event-types). The **View** column shows the optional BigQuery view. Python creates views by default; Java creates them only when `createViews(true)` is configured.
In **Kotlin**, the plugin logs `INVOCATION_STARTING` and `INVOCATION_COMPLETED` only and creates no views, so the other rows and the entire **View** column apply to Python and Java.
The table is the union of the Python and Java event sets. `INVOCATION_ERROR`, `AGENT_ERROR`, `AGENT_TRANSFER`, `AGENT_STATE_CHECKPOINT`, `EVENT_COMPACTION`, `NODE_OUTPUT`, and `NODE_ERROR` are Python-only. Java emits `TOOL_PAUSED`, but none of the other workflow-specific events. The remaining rows apply to both languages.
| Event Type | Captured When | Key Payload Fields | View |
| ————————————- | ——————————————————————- | ————————————————————————————————————————————— | —————————– |
| `USER_MESSAGE_RECEIVED` | A user message enters the invocation | text summary / content parts | `v_user_message_received` |
| `INVOCATION_STARTING` | An invocation begins | *(common columns only)* | `v_invocation_starting` |
| `INVOCATION_COMPLETED` | An invocation ends | *(common columns only)* | `v_invocation_completed` |
| `INVOCATION_ERROR` | An invocation fails with an unhandled exception | error message, sanitized traceback | `v_invocation_error` |
| `AGENT_STARTING` | Agent execution begins | instruction text; in ADK Python v2.12.0 and later, also instruction source, static instruction, description, model, and sub-agent names | `v_agent_starting` |
| `AGENT_COMPLETED` | Agent execution ends | latency | `v_agent_completed` |
| `AGENT_ERROR` | Agent execution fails with an unhandled exception | error message, sanitized traceback, latency | `v_agent_error` |
| `LLM_REQUEST` | A model request is sent | model, prompt, config, tools | `v_llm_request` |
| `LLM_RESPONSE` | A model response is received | response, usage tokens, cache metadata, finish reason, latency, TTFT | `v_llm_response` |
| `LLM_ERROR` | A model call fails | error message, latency | `v_llm_error` |
| `TOOL_STARTING` | A tool begins execution | tool name, args, origin | `v_tool_starting` |
| `TOOL_COMPLETED` | A tool succeeds | tool name, result, origin, latency | `v_tool_completed` |
| `TOOL_ERROR` | A tool fails | tool name, args, origin, error, latency | `v_tool_error` |
| `STATE_DELTA` | Session state changes | state delta | `v_state_delta` |
| `HITL_CREDENTIAL_REQUEST` | Credential request is emitted | synthetic tool name, args | `v_hitl_credential_request` |
| `HITL_CONFIRMATION_REQUEST` | Confirmation request is emitted | synthetic tool name, args | `v_hitl_confirmation_request` |
| `HITL_INPUT_REQUEST` | User input request is emitted | synthetic tool name, args | `v_hitl_input_request` |
| `HITL_CREDENTIAL_REQUEST_COMPLETED` | User provides credential response | synthetic tool name, result | *(base table only)* |
| `HITL_CONFIRMATION_REQUEST_COMPLETED` | User provides confirmation response | synthetic tool name, result | *(base table only)* |
| `HITL_INPUT_REQUEST_COMPLETED` | User provides input response | synthetic tool name, result | *(base table only)* |
| `A2A_INTERACTION` | Remote A2A call completes | response, task ID, context ID, request/response | `v_a2a_interaction` |
| `AGENT_RESPONSE` | Final agent response is yielded | response (content), source event ID/author/branch (attributes) | `v_agent_response` |
| `AGENT_TRANSFER` | One agent hands off control to another | from agent, to agent, source event ID | `v_agent_transfer` |
| `AGENT_STATE_CHECKPOINT` | An agent snapshots its state (or marks the end of its run) | agent state, end-of-agent flag, source event ID | `v_agent_state_checkpoint` |
| `EVENT_COMPACTION` | A window of events is compacted into a summary | window start/end timestamps, compacted content | `v_event_compaction` |
| `TOOL_PAUSED` | A long-running tool (or HITL request) suspends, awaiting resumption | tool name, args, pause kind, function call ID | `v_tool_paused` |
| `NODE_OUTPUT` | A workflow node emits a final structured output | output, node path, run ID, parent run ID | `v_node_output` |
| `NODE_ERROR` | A workflow node ends with a non-model error | error code, error message, node path, run ID, parent run ID | `v_node_error` |
## Installation
For Python, install ADK with the dedicated BigQuery Agent Analytics extra. The extra includes the BigQuery clients, Cloud Storage client, and `pyarrow` required by the plugin:
“`bash
pip install “google-adk[bigquery-analytics]>=2.7.0″
“`
The `pyarrow` dependency is no longer included in the general `gcp` extra. If `pyarrow` is missing, the plugin’s import error identifies the `bigquery-analytics` extra to install.
## Quickstart
Add the plugin to your agent’s `App` object. For prerequisites, see [Prerequisites](#prerequisites).
agent.py
“`python
import os
from google.adk.agents import Agent
from google.adk.apps import App
from google.adk.models.google_llm import Gemini
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryAgentAnalyticsPlugin
os.environ[‘GOOGLE_CLOUD_PROJECT’] = ‘your-gcp-project-id’
os.environ[‘GOOGLE_CLOUD_LOCATION’] = ‘us-central1’
os.environ[‘GOOGLE_GENAI_USE_ENTERPRISE’] = ‘True’
plugin = BigQueryAgentAnalyticsPlugin(
project_id=”your-gcp-project-id”,
dataset_id=”your-big-query-dataset-id”,
)
root_agent = Agent(
model=Gemini(model=”gemini-flash-latest”),
name=’my_agent’,
instruction=”You are a helpful assistant.”,
)
app = App(
name=”my_agent”,
root_agent=root_agent,
plugins=[plugin],
)
“`
Add the plugin to your runner’s plugins list. For prerequisites, see [Prerequisites](#prerequisites).
Agent.java
“`java
import com.google.adk.agents.LlmAgent;
import com.google.adk.agents.RunConfig;
import com.google.adk.models.Gemini;
import com.google.adk.plugins.Plugin;
import com.google.adk.plugins.agentanalytics.BigQueryAgentAnalyticsPlugin;
import com.google.adk.plugins.agentanalytics.BigQueryLoggerConfig;
import com.google.adk.runner.InMemoryRunner;
import com.google.common.collect.ImmutableList;
public final class Agent {
public static void main(String[] args) throws Exception {
Plugin bqLoggingPlugin = new BigQueryAgentAnalyticsPlugin(
BigQueryLoggerConfig.builder()
.projectId(“your-gcp-project-id”)
.datasetId(“your-big-query-dataset-id”)
.tableName(“agent_events”) // Optional; default in v1.8.0+
.build());
InMemoryRunner runner = new InMemoryRunner(
LlmAgent.builder()
.model(Gemini.builder().modelName(“gemini-2.5-flash”).build())
.name(“my_agent”)
.instruction(“You are a helpful assistant.”)
.build(),
“my_agent”,
ImmutableList.of(bqLoggingPlugin));
// Use runner …
// Close runner to flush and close plugin
runner.close().blockingAwait();
}
}
“`
Add the plugin to your agent’s `App` object. For prerequisites, see [Prerequisites](#prerequisites). The plugin is JVM-only and ships outside core, so add the integrations artifact:
build.gradle.kts
“`kotlin
implementation(“com.google.adk:google-adk-kotlin-integrations:1.4.0”)
“`
BigQueryAnalyticsExample.kt
“`kotlin
import com.google.adk.kt.agents.Instruction
import com.google.adk.kt.agents.LlmAgent
import com.google.adk.kt.apps.App
import com.google.adk.kt.models.Gemini
import com.google.adk.kt.plugins.agentanalytics.BigQueryAgentAnalyticsPlugin
import com.google.adk.kt.plugins.agentanalytics.BigQueryLoggerConfig
val analyticsAgent =
LlmAgent(
name = “my_agent”,
model = Gemini(name = “gemini-flash-latest”),
instruction = Instruction(“You are a helpful assistant.”),
)
/**
* Wraps [analyticsAgent] in an [App] whose invocations are logged to BigQuery.
*
* The plugin creates the day-partitioned table on first use, so the credentials
* in scope need permission to create a table in the dataset, not only to insert
* rows. Without explicit `credentials`, application default credentials are used.
*
* Logging failures never fail the turn: a table that cannot be created, or a row
* that cannot be inserted, is logged and the invocation carries on.
*/
fun analyticsApp(
projectId: String,
datasetId: String,
datasetLocation: String,
): App {
val plugin =
BigQueryAgentAnalyticsPlugin(
config =
BigQueryLoggerConfig(
projectId = projectId,
datasetId = datasetId,
// Defaults to “US”; pass your dataset’s location instead.
location = datasetLocation,
),
)
return App(
appName = “my_agent”,
rootAgent = analyticsAgent,
plugins = listOf(plugin),
)
}
“`
The plugin creates the events table on first use, so the credentials in scope need permission to create a table in the dataset, not only to insert rows. Set `location` to your dataset’s location; it defaults to `”US”`. For the full set of options, see [Configuration options](#configuration-options).
Logging never fails the turn: if the table cannot be created or a row cannot be inserted, the plugin logs the error and the invocation continues. When rows are missing, enable logging for `com.google.adk.kt.plugins.agentanalytics.BigQueryAgentAnalyticsPlugin` — logs are emitted under that class name, not under the plugin’s ADK name (`bigquery_agent_analytics`).
### Run and test agent
Test the plugin by running the agent and making a few requests through the chat interface, such as “tell me what you can do” or “List datasets in my cloud project “. These actions create events which are recorded in your Google Cloud project BigQuery instance. Once these events have been processed, you can view the data for them in the [BigQuery Console](https://console.cloud.google.com/bigquery), using this query:
“`sql
SELECT timestamp, event_type, content
FROM `your-gcp-project-id.your-big-query-dataset-id.agent_events`
ORDER BY timestamp DESC
LIMIT 20;
“`
Full example with GCS offloading, OpenTelemetry, and BigQuery tools
my_bq_agent/agent.py
“`python
# my_bq_agent/agent.py
import os
import google.auth
from google.adk.apps import App
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryAgentAnalyticsPlugin, BigQueryLoggerConfig
from google.adk.agents import Agent
from google.adk.models.google_llm import Gemini
from google.adk.tools.bigquery import BigQueryToolset, BigQueryCredentialsConfig
# — OpenTelemetry note (no setup required for BQAA) —
# The BQAA plugin does NOT export OTel spans of its own. It tracks the
# parent-child hierarchy on an internal stack: the root invocation span
# reuses the ambient OTel span’s id (as a 16-hex string) when one is
# active, and child BQAA spans are generated internally as 16-hex
# strings. The plugin’s `trace_id`
# column inherits from whichever OpenTelemetry span is active in the
# surrounding runtime when the agent runs:
# * Agent Engine wires its invocation span automatically, so
# `trace_id` in BigQuery joins to Cloud Trace out of the box.
# * Locally, framework-instrumented runners open an invocation span
# for you.
# * If neither is available, the plugin falls back to a per-invocation
# trace_id and the parent-child hierarchy is still preserved in
# BigQuery; no OTel setup needed.
# Setting a bare `TracerProvider` with no ambient span will NOT cause
# `trace_id` to be populated with a “real” OTel id; only an *active*
# span does. See the “Tracing and observability” section for details.
# — Configuration —
PROJECT_ID = os.environ.get(“GOOGLE_CLOUD_PROJECT”, “your-gcp-project-id”)
DATASET_ID = os.environ.get(“BIG_QUERY_DATASET_ID”, “your-big-query-dataset-id”)
# GOOGLE_CLOUD_LOCATION must be a valid Agent Platform region (e.g., “us-central1”).
# BQ_LOCATION is the BigQuery dataset location, which can be a multi-region
# like “US” or “EU”, or a single region like “us-central1”.
VERTEX_LOCATION = os.environ.get(“GOOGLE_CLOUD_LOCATION”, “us-central1”)
BQ_LOCATION = os.environ.get(“BQ_LOCATION”, “US”)
GCS_BUCKET = os.environ.get(“GCS_BUCKET_NAME”, “your-gcs-bucket-name”) # Optional
if PROJECT_ID == “your-gcp-project-id”:
raise ValueError(“Please set GOOGLE_CLOUD_PROJECT or update the code.”)
# — CRITICAL: Set environment variables BEFORE Gemini instantiation —
os.environ[‘GOOGLE_CLOUD_PROJECT’] = PROJECT_ID
os.environ[‘GOOGLE_CLOUD_LOCATION’] = VERTEX_LOCATION
os.environ[‘GOOGLE_GENAI_USE_ENTERPRISE’] = ‘True’
# — Initialize the Plugin with Config —
bq_config = BigQueryLoggerConfig(
enabled=True,
gcs_bucket_name=GCS_BUCKET, # Enable GCS offloading for multimodal content
log_multi_modal_content=True,
max_content_length=500 * 1024, # 500 KB limit for inline text
batch_size=1, # Default is 1 for low latency, increase for high throughput
shutdown_timeout=10.0
)
bq_logging_plugin = BigQueryAgentAnalyticsPlugin(
project_id=PROJECT_ID,
dataset_id=DATASET_ID,
table_id=”agent_events”, # default table name is agent_events
config=bq_config,
location=BQ_LOCATION
)
# — Initialize Tools and Model —
credentials, _ = google.auth.default(scopes=[“https://www.googleapis.com/auth/cloud-platform”])
bigquery_toolset = BigQueryToolset(
credentials_config=BigQueryCredentialsConfig(credentials=credentials)
)
llm = Gemini(model=”gemini-flash-latest”)
root_agent = Agent(
model=llm,
name=’my_bq_agent’,
instruction=”You are a helpful assistant with access to BigQuery tools.”,
tools=[bigquery_toolset]
)
# — Create the App —
app = App(
name=”my_bq_agent”,
root_agent=root_agent,
plugins=[bq_logging_plugin],
)
“`
“`java
package adk.plugins.agentanalytics.demo;
import static java.nio.charset.StandardCharsets.UTF_8;
import static java.util.Collections.singletonList;
import com.google.adk.agents.LlmAgent;
import com.google.adk.agents.RunConfig;
import com.google.adk.events.Event;
import com.google.adk.models.Gemini;
import com.google.adk.plugins.Plugin;
import com.google.adk.plugins.agentanalytics.BigQueryAgentAnalyticsPlugin;
import com.google.adk.plugins.agentanalytics.BigQueryLoggerConfig;
import com.google.adk.runner.InMemoryRunner;
import com.google.adk.sessions.Session;
import com.google.adk.tools.FunctionTool;
import com.google.adk.tools.ToolContext;
import com.google.genai.types.Content;
import com.google.genai.types.GenerateContentConfig;
import com.google.genai.types.Part;
import io.opentelemetry.sdk.OpenTelemetrySdk;
import io.opentelemetry.sdk.common.CompletableResultCode;
import io.opentelemetry.sdk.trace.SdkTracerProvider;
import io.opentelemetry.sdk.trace.data.SpanData;
import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
import io.opentelemetry.sdk.trace.export.SpanExporter;
import io.reactivex.rxjava3.core.Flowable;
import java.util.Collection;
import java.util.Scanner;
/** Demo agent showing how to use BigQueryAgentAnalyticsPlugin. */
public final class BqDemoAgent {
private static final String PROJECT_ID = “your-gcp-project-id”;
private static final String DATASET_ID = “your-gcp-dataset_id”;
private static final String TABLE_ID = “your-gcp-table”;
private static final String GCS_BUCKET_NAME = “your-gcs-bucket-name”;
private static final String API_KEY = “your-api_key”;
// A simple tool to demonstrate tool execution logging
public static String reverseString(String input, ToolContext toolContext) {
return new StringBuilder(input).reverse().toString();
}
public static void main(String[] args) throws Exception {
// 0. Initialize OpenTelemetry
initOpenTelemetry();
// 1. Configure the BigQuery Logger
BigQueryLoggerConfig config =
BigQueryLoggerConfig.builder()
.projectId(PROJECT_ID)
.datasetId(DATASET_ID)
.tableName(TABLE_ID)
.gcsBucketName(GCS_BUCKET_NAME)
.createViews(true)
.build();
// 2. Create the plugin instance
Plugin bqLoggingPlugin = new BigQueryAgentAnalyticsPlugin(config);
// 3. Initialize the model (Gemini)
Gemini model =
Gemini.builder()
.modelName(“gemini-3-flash-preview”) // Use appropriate model
.apiKey(API_KEY)
.build();
// 4. Create the agent with the tool and plugin
LlmAgent agent =
LlmAgent.builder()
.model(model)
.name(“bq_demo_agent”)
.instruction(
“You are a helpful assistant. You have a tool ‘reverseString’ that you can use to”
+ ” reverse text.”)
.tools(FunctionTool.create(BqDemoAgent.class, “reverseString”))
.generateContentConfig(GenerateContentConfig.builder().temperature(0.5f).build())
.build();
// 5. Initialize the runner
InMemoryRunner runner =
new InMemoryRunner(agent, “bq_demo_agent”, singletonList(bqLoggingPlugin));
// 6. Create a session
Session session =
runner.sessionService().createSession(runner.appName(), “demo_user”).blockingGet();
RunConfig runConfig = RunConfig.builder().build();
System.out.println(“Agent ready. Type ‘quit’ to exit.”);
try (Scanner scanner = new Scanner(System.in, UTF_8)) {
while (true) {
System.out.print(“\nUser: “);
String userInput = scanner.nextLine();
if (userInput.trim().equalsIgnoreCase(“quit”)) {
break;
}
Content userMsg = Content.fromParts(Part.fromText(userInput));
// Run the agent and stream events
Flowable events =
runner.runAsync(session.userId(), session.id(), userMsg, runConfig);
System.out.print(“Agent: “);
events.blockingForEach(
event -> {
if (event.finalResponse()) {
System.out.println(event.stringifyContent());
}
});
}
} finally {
System.out.println(“Closing runner (flushing remaining logs)…”);
runner.close().blockingAwait();
System.out.println(“Done.”);
}
}
private static void initOpenTelemetry() {
PrintingSpanExporter exporter = new PrintingSpanExporter();
SdkTracerProvider tracerProvider =
SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter)).build();
OpenTelemetrySdk.builder().setTracerProvider(tracerProvider).buildAndRegisterGlobal();
}
private static class PrintingSpanExporter implements SpanExporter {
@Override
public CompletableResultCode export(Collection spans) {
for (SpanData span : spans) {
System.out.println(“— Span: ” + span.getName() + ” —“);
System.out.println(” TraceId: ” + span.getTraceId());
System.out.println(” SpanId: ” + span.getSpanId());
System.out.println(” ParentSpanId: ” + span.getParentSpanId());
System.out.println(” Attributes: ” + span.getAttributes());
System.out.println(“————————“);
}
return CompletableResultCode.ofSuccess();
}
@Override
public CompletableResultCode flush() {
return CompletableResultCode.ofSuccess();
}
@Override
public CompletableResultCode shutdown() {
return CompletableResultCode.ofSuccess();
}
}
private BqDemoAgent() {}
}
“`
Deploying to Agent Runtime?
See [Deploy to Agent Runtime](#deploy-agent-runtime).
## Prerequisites
– **Google Cloud Project** with the **BigQuery API** enabled.
– **BigQuery Dataset:** Create a dataset to store logging tables before using the plugin. The plugin automatically creates the necessary events table within the dataset if the table does not exist.
– **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.
– **Authentication:**
– **Local:** Run `gcloud auth application-default login`.
– **Cloud:** Ensure your service account has the required permissions.
Note: Gemini model selector `gemini-flash-latest`
Most code examples in ADK documentation use `gemini-flash-latest` to select the [latest available](https://ai.google.dev/gemini-api/docs/models#latest) Gemini Flash version. However, if you access Gemini from a regional endpoint, such as `us-central1`, this selection string may not work. In that case, use a specific model version string from the [Gemini models](https://ai.google.dev/gemini-api/docs/models) page or Google Cloud [Gemini models](https://docs.cloud.google.com/vertex-ai/generative-ai/docs/models) list.
### IAM permissions
For the agent to work properly, the principal (e.g., service account, user account) under which the agent 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.
## Configuration options
### Constructor parameters
The `BigQueryAgentAnalyticsPlugin` constructor accepts these parameters. It also accepts `**kwargs`, which are forwarded directly to `BigQueryLoggerConfig` (see below).
| Parameter | Type | Default | Use when |
| ———————— | ———————————————– | ———— | —————————————————————————————————————————————————————————————————————————————————————————– |
| `project_id` | `str` | *(required)* | Select the Google Cloud project |
| `dataset_id` | `str` | *(required)* | Select the BigQuery dataset |
| `table_id` | `Optional[str]` | `None` | Use a custom table name (overrides config `table_id`) |
| `config` | `Optional[BigQueryLoggerConfig]` | `None` | Pass a config object for detailed tuning |
| `location` | `str` | `”US”` | Match the BigQuery dataset location (e.g., `”US”`, `”EU”`, `”us-central1″`) |
| `credentials` | `Optional[google.auth.credentials.Credentials]` | `None` | Use explicit service-account, impersonated, or cross-project credentials instead of [ADC](https://cloud.google.com/docs/authentication/application-default-credentials) |
| `credentials_identifier` | `Optional[str]` | `None` | Choose explicitly which plugin instances share a [background writer](#shared-background-writer). Use the same identifier only for instances with equivalent credentials and writer settings. Takes precedence over `config.credentials_identifier` without modifying `config` |
“`python
plugin = BigQueryAgentAnalyticsPlugin(
project_id=”my-project”,
dataset_id=”my_dataset”,
batch_size=10, # forwarded to BigQueryLoggerConfig
shutdown_timeout=5.0, # forwarded to BigQueryLoggerConfig
)
“`
### BigQueryLoggerConfig options
All options below are optional and have sensible defaults. Pass them to `BigQueryLoggerConfig` or as `**kwargs` to the plugin constructor.
| Option | Type | Default | Use when |
| ——————————– | ————————————— | ———————————— | —————————————————————————————————————————————————————————————————————————————————————————————- |
| `enabled` | `bool` | `True` | Temporarily disable logging |
| `table_id` | `str` | `”agent_events”` | Use a custom table name (constructor value takes precedence) |
| `clustering_fields` | `List[str]` | `[“event_type”, “agent”, “user_id”]` | Customize table clustering on creation |
| `gcs_bucket_name` | `Optional[str]` | `None` | Offload large text and multimodal content to GCS |
| `connection_id` | `Optional[str]` | `None` | Use BigQuery ObjectRef / object tables (e.g., `us.my-connection`) |
| `max_content_length` | `int` | `500 * 1024` | Control inline payload size before offloading/truncating. Use `-1` for no limit; other values must be at least `1` |
| `batch_size` | `int` | `1` | Tune write throughput vs. latency |
| `batch_flush_interval` | `float` | `1.0` | Flush partial batches periodically (seconds) |
| `shutdown_timeout` | `float` | `10.0` | Wait for final flush on shutdown (seconds) |
| `event_allowlist` | `Optional[List[str]]` | `None` | Log only selected [event types](#event-types) |
| `event_denylist` | `Optional[List[str]]` | `None` | Skip sensitive or noisy [event types](#event-types) |
| `content_formatter` | `Optional[Callable]` | `None` | Apply custom masking/formatting per event (receives `(content, event_type)`) |
| `log_multi_modal_content` | `bool` | `True` | Capture `content_parts` details including GCS references |
| `queue_max_size` | `int` | `10000` | Bound the in-memory event queue |
| `retry_config` | `RetryConfig` | `RetryConfig()` | Tune retry behavior (`max_retries=3`, `initial_delay=1.0`, `multiplier=2.0`, `max_delay=10.0`) |
| `log_session_metadata` | `bool` | `True` | Add session info to `attributes` (`session_id`, `app_name`, `user_id`, `state`). Keys prefixed `temp:` are [redacted](#built-in-redaction). |
| `custom_tags` | `Dict[str, Any]` | `{}` | Add static tags (e.g., `{“env”: “prod”}`) to every event’s `attributes` |
| `auto_schema_upgrade` | `bool` | `True` | Automatically add new columns to existing tables (additive only) |
| `create_views` | `bool` | `True` | Create per-event-type BigQuery views |
| `view_prefix` | `str` | `”v”` | Avoid view-name collisions when multiple plugins share a dataset (e.g., `”v_staging”`) |
| `enable_otel_correlation` | `bool` | `False` | Capture the ambient OpenTelemetry span context into `attributes.otel.{span_id, trace_id}` as a best-effort Cloud Trace join key |
| `custom_metadata_allowlist` | `Optional[List[str]]` | `None` | Capture selected `event.custom_metadata` keys into `attributes.custom_metadata.*`: exact keys or `”prefix*”` patterns. In ADK Python v2.12.0 and later, final `LLM_RESPONSE` rows capture keys from `llm_response.custom_metadata` |
| `payload_column_denylist` | `Optional[List[str]]` | `None` | Project payload columns (`content`, `content_parts`, `attributes`, `latency_ms`) out of the table at write time |
| `final_response_tool_names` | `FrozenSet[str]` | `frozenset()` | Log call arguments from selected successful tools as `AGENT_RESPONSE` payloads |
| `flush_on_run_end` | `bool` | `True` | Wait for queued rows to finish writing at the end of each run |
| `exactly_once_delivery` | `bool` | `False` | Use committed streams and explicit offsets to prevent ambiguous-retry duplicates within a live processor |
| `use_dedicated_background_loop` | `Optional[bool]` | `None` | Choose where the batch writer runs: `None` uses the [shared background writer](#shared-background-writer) thread when `flush_on_run_end=False`, `True` always uses it, and `False` keeps the writer on the caller’s event loop |
| `credentials_identifier` | `Optional[str]` | `None` | Choose explicitly which plugin instances share a background writer; the constructor argument takes precedence. Use the same identifier only for instances with equivalent credentials and writer settings |
| `debug_content_formatter_errors` | `bool` | `False` | Append a bounded traceback of an exception raised by `content_formatter` to the plugin’s local Python log warning. The traceback can include unformatted content and is never written to BigQuery |
| `on_schema_error` | `Optional[Callable[[Exception], None]]` | `None` | In ADK Python v2.12.0 and later, run a synchronous callback with the exception each time the [table readiness check](#table-readiness-callbacks) fails to read, create, or upgrade the table. Values that aren’t synchronous callables raise `ValueError` when the plugin is constructed |
| `on_schema_ready` | `Optional[Callable[[], None]]` | `None` | In ADK Python v2.12.0 and later, run a synchronous no-argument callback each time the [table readiness check](#table-readiness-callbacks) succeeds, including any requested views. Values that aren’t synchronous callables raise `ValueError` when the plugin is constructed |
The following code sample shows how to define a configuration for the BigQuery Agent Analytics plugin:
“`python
import json
import re
from typing import Any
from google.adk.plugins.bigquery_agent_analytics_plugin import BigQueryLoggerConfig
def redact_dollar_amounts(event_content: Any, event_type: str) -> str:
“””
Custom formatter to redact dollar amounts (e.g., $600, $12.50)
and ensure JSON output if the input is a dict.
Args:
event_content: The raw content of the event.
event_type: The event type string (e.g., “LLM_REQUEST”, “LLM_RESPONSE”).
“””
text_content = “”
if isinstance(event_content, dict):
text_content = json.dumps(event_content)
else:
text_content = str(event_content)
# Regex to find dollar amounts: $ followed by digits, optionally with commas or decimals.
# Examples: $600, $1,200.50, $0.99
redacted_content = re.sub(r’\$\d+(?:,\d{3})*(?:\.\d+)?’, ‘xxx’, text_content)
return redacted_content
config = BigQueryLoggerConfig(
enabled=True,
event_allowlist=[“LLM_REQUEST”, “LLM_RESPONSE”], # Only log these events
# event_denylist=[“TOOL_STARTING”], # Skip these events
shutdown_timeout=10.0, # Wait up to 10s for logs to flush on exit
max_content_length=500, # Truncate content to 500 chars
content_formatter=redact_dollar_amounts, # Redact the dollar amounts in the logging content
queue_max_size=10000, # Max events to hold in memory
auto_schema_upgrade=True, # Automatically add new columns to existing tables
create_views=True, # Automatically create per-event-type views
# retry_config=RetryConfig(max_retries=3), # Optional: Configure retries
)
plugin = BigQueryAgentAnalyticsPlugin(
project_id=”my-project”,
dataset_id=”my_dataset”,
config=config,
)
“`
### Trace correlation, metadata capture, and column projection
Supported in ADKPython v2.4.0
Three options control what extra context lands in `attributes` and whether payload columns are written at all. Each is listed in the `BigQueryLoggerConfig` options table above; the notes below add the cross-option rules the flat table can’t express:
– **`enable_otel_correlation`**: The captured span context is a best-effort Cloud Trace correlation key, not a foreign key; when disabled (the default) no `attributes.otel` is written.
– **`custom_metadata_allowlist`**: Leaving it unset preserves the previous behavior, where only the built-in `a2a:*` capture runs. Keys are read from `event.custom_metadata`. In ADK Python v2.12.0 and later, final (non-partial) `LLM_RESPONSE` rows read them from `llm_response.custom_metadata`. Captured values pass the same safety pipeline as all other logged content (truncation, sensitive-key redaction, circular-reference handling).
– **`payload_column_denylist`**: Only `content`, `content_parts`, `attributes`, and `latency_ms` may be listed; identity and correlation columns are protected and raise `ValueError`. The projection is applied schema-first, so the table schema, the written rows, and the auto-created views stay consistent (views drop derived columns that depend on a denied column). Denying `attributes` also disables `attributes.otel` and `attributes.custom_metadata`, and combining it with a non-empty `custom_metadata_allowlist` is rejected at construction.
“`python
config = BigQueryLoggerConfig(
enable_otel_correlation=True, # join key against Cloud Trace
custom_metadata_allowlist=[“ticket_id”, “exp:*”], # capture selected custom_metadata keys
# payload_column_denylist=[“content_parts”], # don’t persist multimodal payloads
)
“`
### Final-answer capture and end-of-run flushing
Use `final_response_tool_names` when an agent delivers its final answer by calling a dedicated tool instead of yielding a plain-text final event. On a successful matching tool call, the plugin writes the tool’s call arguments as an `AGENT_RESPONSE` row and adds `source_tool` to `attributes`.
The `flush_on_run_end` option defaults to `True`, which makes `after_run_callback` wait for the current event loop’s write queue. Set it to `False` to remove that flush from the response path; the background writer will continue draining the queue, so rows may appear in BigQuery shortly after the run returns. In Python v2.10.0 and later, this setting also moves the writer to the [shared background writer](#shared-background-writer).
“`python
config = BigQueryLoggerConfig(
final_response_tool_names=frozenset({“submit_final_response”}),
flush_on_run_end=False,
)
“`
### Shared background writer
Supported in ADKPython v2.10.0
When `flush_on_run_end=False`, the plugin runs its batch writer on a dedicated, process-lifetime background event loop thread instead of the caller’s event loop. Set `use_dedicated_background_loop=True` to use that thread even when `flush_on_run_end=True`, or `False` to keep the writer on the caller’s event loop.
Plugin instances in the same process that target the same table with equivalent credentials share one background writer and BigQuery write client, so short-lived plugin instances, such as one per request, don’t each open a new writer. Equivalence is derived from the credential identity (for example, the service account email, scopes, and project). Set `credentials_identifier` to choose explicitly which plugin instances share a writer. Use the same identifier only for instances with equivalent credentials and writer settings: the shared writer uses the credentials and writer settings, such as `batch_size` and `retry_config`, of the instance that creates it.
Calling `shutdown()` or `close()` with default arguments detaches the plugin instance from the shared writer but leaves the writer running, so queued rows continue to be written. At interpreter exit, the plugin makes a best-effort attempt to drain active writers. To drain and stop the shared writer explicitly, for example at the end of a standalone script or a test, pass `close_background_transport=True` or call `close_shared_background_transports()`:
“`python
# Drain and stop the shared writer for this plugin’s table and credentials.
await plugin.close(close_background_transport=True)
# Or drain and stop every shared background writer in the process.
await BigQueryAgentAnalyticsPlugin.close_shared_background_transports(timeout=10.0)
“`
### Delivery and deduplication
Supported in ADKPython v2.7.0
Every row receives a 32-character hexadecimal `event_id` before enqueue. The same ID is reused when the Storage Write API retries that row, making it the deduplication key in the default delivery mode:
“`sql
SELECT *
FROM `your-gcp-project-id.adk_agent_logs.agent_events`
QUALIFY
event_id IS NULL
OR ROW_NUMBER() OVER (PARTITION BY event_id ORDER BY timestamp) = 1;
“`
The `event_id IS NULL` condition retains rows written before the column was introduced.
Set `exactly_once_delivery=True` to use one loop-local committed stream and explicit offsets. This prevents a retry whose first result was ambiguous from creating a duplicate within the lifetime of that processor. It can consume additional BigQuery `CreateWriteStream` quota when a stream must be rotated.
“`python
config = BigQueryLoggerConfig(exactly_once_delivery=True)
“`
Despite its name, this option is not a lossless-delivery guarantee. Batches can still be dropped after retry exhaustion, an offset conflict, or a replacement-stream failure. After a failed stream rotation, events arriving during the 30-second rotation backoff are also dropped. Monitor the `offset_conflict` and other [drop reasons](#dropped-event-observability), and retain `event_id` as a consumer deduplication key.
### Table readiness callbacks
Supported in ADKPython v2.12.0
Use `on_schema_error` and `on_schema_ready` to observe the table readiness check that the plugin runs during setup. The check reads, creates, or upgrades the BigQuery table and creates any requested views.
– `on_schema_error` is called with the exception each time the check fails to read, create, or upgrade the table. It fires at most once per check, so it can be called repeatedly while setup retries. It isn’t called when a view can’t be created or a label-only table refresh fails; the plugin logs those failures and continues.
– `on_schema_ready` is called with no arguments each time the check succeeds, which means that the table exists or was created or upgraded and every requested view was created.
Both callbacks run on a worker thread, not the event loop, so keep them fast and non-blocking. Exceptions that a callback raises are logged and suppressed. Values that aren’t synchronous callables raise `ValueError` when the plugin is constructed.
After a successful check, later setups of the same plugin skip it, including after `close()`, after a fork, and in copies pickled afterwards. Neither callback fires again, and later table outages appear only as [dropped rows](#dropped-event-observability) when rows are written. Later setups run the check again only if the setup that was waiting for it was cancelled, or after a `create_analytics_views()` call fails to create a view. A `create_analytics_views()` call that creates every view makes later setups skip the check, and its own view creation never invokes either callback.
If you serialize the plugin with the standard `pickle` module, the callbacks must be picklable, for example module-level functions. Lambdas and closures can’t be pickled with `pickle`, and bound methods pickle their instance.
In Java, all configuration is managed via the `BigQueryLoggerConfig` builder.
#### BigQueryLoggerConfig Builder options
| Builder Method | Type | Default | Description |
| ——————————— | ———————————— | ———————————— | —————————————————————————————- |
| `enabled(boolean)` | `boolean` | `true` | Temporarily disable logging |
| `projectId(String)` | `String` | *(required)* | Select the Google Cloud project |
| `datasetId(String)` | `String` | *(required)* | Select the BigQuery dataset |
| `tableName(String)` | `String` | `”agent_events”` | Use a custom table name |
| `location(String)` | `String` | `”us”` | Match the BigQuery dataset location |
| `clusteringFields(List)` | `List` | `[“event_type”, “agent”, “user_id”]` | Customize table clustering on creation |
| `gcsBucketName(String)` | `String` | `””` | Offload large text and multimodal content to GCS |
| `connectionId(String)` | `String` | `null` | Use BigQuery ObjectRef / object tables |
| `maxContentLength(int)` | `int` | `500 * 1024` | Control inline payload size before offloading/truncating |
| `batchSize(int)` | `int` | `1` | Tune write throughput vs. latency |
| `batchFlushInterval(Duration)` | `Duration` | `Duration.ofSeconds(1)` | Flush partial batches periodically |
| `shutdownTimeout(Duration)` | `Duration` | `Duration.ofSeconds(10)` | Wait for final flush on shutdown |
| `eventAllowlist(List)` | `List` | `[]` | Log only selected event types |
| `eventDenylist(List)` | `List` | `[]` | Skip sensitive or noisy event types |
| `contentFormatter(BiFunction)` | `BiFunction` | `null` | Apply custom masking/formatting per event |
| `logMultiModalContent(boolean)` | `boolean` | `true` | Capture `content_parts` details including GCS references |
| `queueMaxSize(int)` | `int` | `10000` | Bound the in-memory event queue |
| `retryConfig(RetryConfig)` | `RetryConfig` | `RetryConfig.builder().build()` | Tune retry behavior |
| `logSessionMetadata(boolean)` | `boolean` | `true` | Add session info to `attributes` |
| `customTags(Map)` | `Map` | `{}` | Add static tags to every event’s `attributes` |
| `autoSchemaUpgrade(boolean)` | `boolean` | `true` | Automatically add new columns to existing tables |
| `createViews(boolean)` | `boolean` | `false` | Create per-event-type BigQuery views (Note: defaults to `false`, unlike Python’s `true`) |
| `viewPrefix(String)` | `String` | `”v”` | Avoid view-name collisions |
| `credentials(Credentials)` | `Credentials` | `null` | Use explicit service-account credentials |
In Java v1.8.0 and later, `datasetId` is required and `tableName` defaults to `”agent_events”`. Java v1.7.0 and earlier defaulted these values to `”agent_analytics”` and `”events”`, respectively.
The following code sample shows how to define a configuration for the BigQuery Agent Analytics plugin in Java:
“`java
import com.google.adk.plugins.agentanalytics.BigQueryAgentAnalyticsPlugin;
import com.google.adk.plugins.agentanalytics.BigQueryLoggerConfig;
import java.time.Duration;
import java.util.function.BiFunction;
// Custom formatter to redact dollar amounts
BiFunction redactDollarAmounts = (content, eventType) -> {
String textContent = content.toString();
return textContent.replaceAll(“\\$\\d+(?:,\\d{3})*(?:\\.\\d+)?”, “xxx”);
};
BigQueryLoggerConfig config = BigQueryLoggerConfig.builder()
.enabled(true)
.projectId(“my-project”)
.datasetId(“my_dataset”)
.tableName(“agent_events”)
.batchSize(1)
.batchFlushInterval(Duration.ofMillis(500))
.contentFormatter(redactDollarAmounts)
.autoSchemaUpgrade(true)
.createViews(true)
.build();
BigQueryAgentAnalyticsPlugin plugin = new BigQueryAgentAnalyticsPlugin(config);
“`
In Kotlin, all configuration is managed via the `BigQueryLoggerConfig` data class, which the plugin takes as its only required argument.
#### BigQueryLoggerConfig properties
| Option | Type | Default | Use when |
| ————- | ————– | —————- | ————————————————————————————————————————————— |
| `projectId` | `String` | *(required)* | Select the Google Cloud project |
| `datasetId` | `String` | *(required)* | Select the BigQuery dataset |
| `enabled` | `Boolean` | `true` | Temporarily disable logging |
| `location` | `String` | `”US”` | Match the BigQuery dataset location (for example, `”EU”` or `”us-central1″`) |
| `tableName` | `String` | `”agent_events”` | Use a custom table name |
| `credentials` | `Credentials?` | `null` | Use explicit service-account credentials instead of [ADC](https://cloud.google.com/docs/authentication/application-default-credentials) |
The following code sample shows how to define a configuration for the BigQuery Agent Analytics plugin in Kotlin:
“`kotlin
import com.google.adk.kt.plugins.agentanalytics.BigQueryAgentAnalyticsPlugin
import com.google.adk.kt.plugins.agentanalytics.BigQueryLoggerConfig
val config =
BigQueryLoggerConfig(
projectId = “my-project”,
datasetId = “my_dataset”,
location = “EU”,
tableName = “agent_events”,
)
val plugin = BigQueryAgentAnalyticsPlugin(config = config)
“`
The options listed under the **Python** and **Java** tabs, such as batching, content formatting, event allowlists, GCS offloading, and view creation, do not exist in Kotlin.
## Schema and production setup
### Schema Reference
The events table (`agent_events`) uses a flexible schema. The following table provides a comprehensive reference with example values.
| Field Name | Type | Mode | Description | Example Value |
| —————— | ———– | ———- | ————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————— | ———————————————————————————————————————————————————————————————————— |
| **timestamp** | `TIMESTAMP` | `REQUIRED` | UTC timestamp of event creation. Acts as the primary ordering key and the daily partitioning key. Precision is microsecond. | `2026-02-03 20:52:17 UTC` |
| **event_id** | `STRING` | `NULLABLE` | A 32-character hexadecimal ID assigned before enqueue. Storage Write API retries preserve it so consumers can identify duplicate rows. Rows written before schema version 2 have `NULL`. | `ca5e3c9d99e24e46b614f2f44f93bf6e` |
| **event_type** | `STRING` | `NULLABLE` | The canonical event category. Standard values include LLM, tool, agent, invocation, state, HITL, A2A, response, workflow, node output, and node error events described in [Event types and payloads](#event-types). Used for high-level filtering. | `LLM_REQUEST` |
| **agent** | `STRING` | `NULLABLE` | The name of the agent responsible for this event. Defined during agent initialization or via the `root_agent_name` context. | `my_bq_agent` |
| **session_id** | `STRING` | `NULLABLE` | A persistent identifier for the entire conversation thread. Stays constant across multiple turns and sub-agent calls. | `04275a01-1649-4a30-b6a7-5b443c69a7bc` |
| **invocation_id** | `STRING` | `NULLABLE` | The unique identifier for a single execution turn or request cycle. Corresponds to `trace_id` in many contexts. | `e-b55b2000-68c6-4e8b-b3b3-ffb454a92e40` |
| **user_id** | `STRING` | `NULLABLE` | The identifier of the user (human or system) initiating the session. Extracted from the `User` object or metadata. | `test_user` |
| **trace_id** | `STRING` | `NULLABLE` | Trace identifier. It is a 32-character hex OpenTelemetry trace ID when inherited from an ambient span (e.g. Agent Engine’s invocation span or the ADK Runner span), so BigQuery rows join cleanly to your existing Cloud Trace traces. Without an ambient span, Python also generates that format as its per-invocation fallback, while Java falls back to the ADK invocation ID. Links all operations within a single distributed request lifecycle. | `a2c7f13d3a3f0bbb8793692f76a6012a` |
| **span_id** | `STRING` | `NULLABLE` | 16-character hex Span ID identifying this specific atomic operation. **Tracked on the plugin’s internal stack, not exported as an OTel span.** The plugin does not call `tracer.start_span` against your configured OpenTelemetry provider. The root invocation span reuses the ambient OTel span’s id when one is active; child spans are generated internally (see [Tracing and observability](#tracing-and-observability)). | `3916f5762bcd4d42` |
| **parent_span_id** | `STRING` | `NULLABLE` | 16-character hex Span ID of the immediate caller. Used to reconstruct the parent-child execution tree (DAG). | `4c4a42bfdeb84934` |
| **content** | `JSON` | `NULLABLE` | The primary event payload. Structure is polymorphic based on `event_type`. | `{“system_prompt”: “You are…”, “prompt”: [{“role”: “user”, “content”: “hello”}], “response”: “Hi”, “usage”: {“total”: 15}}` |
| **attributes** | `JSON` | `NULLABLE` | Metadata/Enrichment (usage stats, model info, tool provenance, custom tags). | `{“model”: “gemini-flash-latest”, “usage_metadata”: {“total_token_count”: 15}, “session_metadata”: {“session_id”: “…”, “app_name”: “…”, “user_id”: “…”, “state”: {}}, “custom_tags”: {“env”: “prod”}}` |
| **latency_ms** | `JSON` | `NULLABLE` | Performance metrics. Standard keys are `total_ms` (wall-clock duration) and `time_to_first_token_ms` (streaming latency). | `{“total_ms”: 1250, “time_to_first_token_ms”: 450}` |
| **status** | `STRING` | `NULLABLE` | High-level outcome. Values: `OK` (success) or `ERROR` (failure). | `OK` |
| **error_message** | `STRING` | `NULLABLE` | Sanitized diagnostic message for exceptions and model termination details. In Python, it can be populated on rows whose `status` remains `OK`, such as a final `LLM_RESPONSE` or a row whose custom `content_formatter` failed (for example, `content_formatter raised ImportError`). | `Error 404: Dataset not found` |
| **is_truncated** | `BOOLEAN` | `NULLABLE` | `true` when content or metadata is truncated or replaced by a safety boundary, including the configured `max_content_length`, sanitizer depth or node budgets, and diagnostic-text sanitization. In ADK Python v2.12.0 and later, it is also set by redaction of common credential patterns in the formatted parts of a model response and by `[FORMATTING FAILED]`, the placeholder written when the plugin’s built-in formatting of response or compaction content fails (not to be confused with `[FORMATTER_FAILED]`, which marks a custom `content_formatter` failure). Ordinary structured sensitive-key redaction does not set it by itself. | `false` |
| **content_parts** | `RECORD` | `REPEATED` | Array of multi-modal segments (Text, Image, Blob). Used when content cannot be serialized as simple JSON (e.g., large binaries or GCS refs). | `[{“mime_type”: “text/plain”, “text”: “hello”}]` |
In Python, the `event_id` column is part of schema version 2. With `auto_schema_upgrade=True` (the default), the plugin adds it to an existing table automatically. If you manage the table schema yourself, add the column before using ADK Python v2.7.0 or later.
The Java schema does not include `event_id`. A table created from the manual DDL below remains compatible with Java because the column is nullable.
In **Kotlin**, the plugin creates the table with these same columns but populates only `timestamp`, `event_type`, `agent`, `session_id`, `invocation_id`, `user_id`, and `content`. The remaining columns are always null.
The plugin automatically creates the table if it does not exist. For production, you can optionally create the table manually using the DDL below.
Manual DDL for production setup
“`sql
CREATE TABLE `your-gcp-project-id.adk_agent_logs.agent_events`
(
timestamp TIMESTAMP NOT NULL OPTIONS(description=”The UTC time at which the event was logged.”),
event_id STRING OPTIONS(description=”Unique ID assigned before enqueue and preserved across Storage Write API retries.”),
event_type STRING OPTIONS(description=”Indicates the type of event being logged (e.g., ‘LLM_REQUEST’, ‘TOOL_COMPLETED’).”),
agent STRING OPTIONS(description=”The name of the ADK agent or author associated with the event.”),
session_id STRING OPTIONS(description=”A unique identifier to group events within a single conversation or user session.”),
invocation_id STRING OPTIONS(description=”A unique identifier for each individual agent execution or turn within a session.”),
user_id STRING OPTIONS(description=”The identifier of the user associated with the current session.”),
trace_id STRING OPTIONS(description=”32-char hex trace ID. Inherited from the ambient OpenTelemetry span when one is active; otherwise generated per invocation by the plugin.”),
span_id STRING OPTIONS(description=”16-char hex span ID for this specific operation. Tracked on the plugin’s internal stack; the root invocation span may reuse the ambient OTel span id, while child BQAA spans are generated internally. No OpenTelemetry span is created or exported.”),
parent_span_id STRING OPTIONS(description=”16-char hex span ID of the immediate caller, used to reconstruct the parent-child execution tree.”),
content JSON OPTIONS(description=”The event-specific data (payload) stored as JSON.”),
content_parts ARRAY,
text STRING,
part_index INT64,
part_attributes STRING,
storage_mode STRING
>> OPTIONS(description=”Detailed content parts for multi-modal data.”),
attributes JSON OPTIONS(description=”Arbitrary key-value pairs for additional metadata (e.g., ‘root_agent_name’, ‘model_version’, ‘usage_metadata’, ‘session_metadata’, ‘custom_tags’).”),
latency_ms JSON OPTIONS(description=”Latency measurements (e.g., total_ms).”),
status STRING OPTIONS(description=”The outcome of the event, typically ‘OK’ or ‘ERROR’.”),
error_message STRING OPTIONS(description=”Diagnostic message for errors and model termination details; may be populated on rows whose status is ‘OK’, such as LLM_RESPONSE rows and rows whose content_formatter failed.”),
is_truncated BOOLEAN OPTIONS(description=”Flag indicates if content was truncated.”)
)
PARTITION BY DATE(timestamp)
CLUSTER BY event_type, agent, user_id;
“`
### Automatically Created Views
Supported in ADKPython v1.27.0Java v1.5.0
In Python, `create_views=True` (the default) automatically generates views for each event type. In Java, set `createViews(true)`; its default is `false`. Kotlin does not create views. The views unnest common JSON structures into flat, typed columns, which avoids repetitive `JSON_VALUE` and `JSON_QUERY` expressions.
View names follow the convention `{view_prefix}_{event_type_lowercase}` (for example, with the default prefix `”v”`, `LLM_REQUEST` becomes `v_llm_request`). Set `view_prefix` in `BigQueryLoggerConfig` to a distinct value when multiple plugin instances write to different tables in the same dataset, preventing view-name collisions:
“`python
# Two plugins in the same dataset with distinct view prefixes
plugin_prod = BigQueryAgentAnalyticsPlugin(
project_id=PROJECT_ID, dataset_id=DATASET_ID,
table_id=”agent_events_prod”,
config=BigQueryLoggerConfig(view_prefix=”v_prod”),
)
# Creates views: v_prod_llm_request, v_prod_tool_completed, …
plugin_staging = BigQueryAgentAnalyticsPlugin(
project_id=PROJECT_ID, dataset_id=DATASET_ID,
table_id=”agent_events_staging”,
config=BigQueryLoggerConfig(view_prefix=”v_staging”),
)
# Creates views: v_staging_llm_request, v_staging_tool_completed, …
“`
You can also call the public async method `await plugin.create_analytics_views()` to manually refresh views, for example after a schema upgrade.
Every Python view includes these **common columns**: `timestamp`, `event_id`, `event_type`, `agent`, `session_id`, `invocation_id`, `user_id`, `trace_id`, `span_id`, `parent_span_id`, `status`, `error_message`, `is_truncated`. Java views include the same common columns except `event_id`.
The following table lists the Python views and their event-specific columns:
| View Name | Event-Specific Columns |
| ——————————— | ————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————————– |
| **`v_user_message_received`** | *(common columns only)* |
| **`v_llm_request`** | `model` (STRING), `request_content` (JSON), `llm_config` (JSON), `tools` (JSON) |
| **`v_llm_response`** | `response` (JSON), `usage_prompt_tokens` (INT64), `usage_completion_tokens` (INT64), `usage_total_tokens` (INT64), `usage_cached_tokens` (INT64), `usage_thinking_tokens` (INT64), `usage_tool_use_tokens` (INT64), `context_cache_hit_rate` (FLOAT64), `total_ms` (INT64), `ttft_ms` (INT64), `model_version` (STRING), `usage_metadata` (JSON), `cache_metadata` (JSON), `cache_type` (STRING), `finish_reason` (STRING) |
| **`v_llm_error`** | `total_ms` (INT64) |
| **`v_tool_starting`** | `tool_name` (STRING), `tool_args` (JSON), `tool_origin` (STRING) |
| **`v_tool_completed`** | `tool_name` (STRING), `tool_result` (JSON), `tool_origin` (STRING), `total_ms` (INT64), `pause_kind` (STRING), `function_call_id` (STRING) |
| **`v_tool_error`** | `tool_name` (STRING), `tool_args` (JSON), `tool_origin` (STRING), `total_ms` (INT64) |
| **`v_agent_starting`** | `agent_instruction` (STRING); in ADK Python v2.12.0 and later, also `instruction_source` (STRING), `static_instruction` (STRING), `agent_description` (STRING), `model` (STRING), `sub_agents` (JSON) |
| **`v_agent_completed`** | `total_ms` (INT64) |
| **`v_agent_error`** | `total_ms` (INT64), `error_traceback` (STRING) |
| **`v_invocation_starting`** | *(common columns only)* |
| **`v_invocation_completed`** | *(common columns only)* |
| **`v_invocation_error`** | `error_traceback` (STRING) |
| **`v_state_delta`** | `state_delta` (JSON) |
| **`v_hitl_credential_request`** | `tool_name` (STRING), `tool_args` (JSON) |
| **`v_hitl_confirmation_request`** | `tool_name` (STRING), `tool_args` (JSON) |
| **`v_hitl_input_request`** | `tool_name` (STRING), `tool_args` (JSON) |
| **`v_a2a_interaction`** | `response_content` (JSON), `a2a_task_id` (STRING), `a2a_context_id` (STRING), `a2a_request` (JSON), `a2a_response` (JSON) |
| **`v_agent_response`** | `response_text` (STRING), `source_event_id` (STRING), `source_event_author` (STRING), `source_event_branch` (STRING) |
| **`v_agent_transfer`** | `from_agent` (STRING), `to_agent` (STRING), `source_event_id` (STRING) |
| **`v_agent_state_checkpoint`** | `agent_state` (JSON), `agent_state_type` (STRING), `end_of_agent` (BOOL), `source_event_id` (STRING) |
| **`v_event_compaction`** | `start_seconds` (FLOAT64), `end_seconds` (FLOAT64), `window_start` (TIMESTAMP), `window_end` (TIMESTAMP), `compacted_content` (JSON, holding the formatted summary string) |
| **`v_tool_paused`** | `tool_name` (STRING), `tool_args` (JSON), `pause_kind` (STRING), `function_call_id` (STRING) |
| **`v_node_output`** | `node_path` (STRING), `node_run_id` (STRING), `node_parent_run_id` (STRING), `output` (JSON) |
| **`v_node_error`** | `node_path` (STRING), `node_run_id` (STRING), `node_parent_run_id` (STRING), `error_code` (STRING) |
The four workflow views (`v_agent_transfer`, `v_agent_state_checkpoint`, `v_event_compaction`, `v_tool_paused`) and the `pause_kind` / `function_call_id` columns on `v_tool_completed` come with the [ADK 2.0 workflow event support](#adk-2-events). In **Java** (v1.7.0+), only `v_tool_paused` and the `pause_kind` / `function_call_id` columns on `v_tool_completed` are created; `v_agent_transfer`, `v_agent_state_checkpoint`, and `v_event_compaction` are Python-only (the Java plugin does not emit those events). The `v_node_output` and `v_node_error` views are available in Python v2.7.0 and later.
The other Java view differences are:
– Java does not create `v_agent_error`, `v_invocation_error`, `v_node_output`, or `v_node_error` because it does not emit those events.
– Java’s `v_llm_response` ends at `usage_metadata`; it does not expose `usage_thinking_tokens`, `usage_tool_use_tokens`, `cache_metadata`, `cache_type`, or `finish_reason`.
– Java’s `v_agent_response` exposes `text_summary` instead of `response_text`.
– Java’s `v_agent_starting` exposes only `agent_instruction`.
– Java’s `v_a2a_interaction` omits `a2a_response`; the response remains available in `response_content`.
## Event types and payloads
The `content` column now 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`.
### LLM interactions (plugin lifecycle)
These events track the raw requests sent to and responses received from the LLM.
**1. LLM_REQUEST**
Captures the prompt sent to the model, including conversation history and system instructions.
“`json
{
“event_type”: “LLM_REQUEST”,
“content”: {
“system_prompt”: “You are a helpful assistant…”,
“prompt”: [
{
“role”: “user”,
“content”: “hello how are you today”
}
]
},
“attributes”: {
“root_agent_name”: “my_bq_agent”,
“model”: “gemini-flash-latest”,
“tools”: [
{
“name”: “list_dataset_ids”,
“description”: “…”
},
{
“name”: “execute_sql”,
“description”: “…”
}
],
“llm_config”: {
“temperature”: 0.5,
“top_p”: 0.9
}
}
}
“`
The automatically created `v_llm_request` view exposes the `tools` attribute as its `tools` (JSON) column. In Python, `tools` has one entry per tool with its `name` and, when available, its `description` and `parameters` (the tool’s parameter schema). These entries pass through the same size limit and sensitive-key redaction as other attributes. Java records only the tool names, for example `[“list_dataset_ids”, “execute_sql”]`.
**2. LLM_RESPONSE**
Captures the model’s output and token usage statistics.
“`json
{
“event_type”: “LLM_RESPONSE”,
“content”: {
“response”: “text: ‘Hello! I’m doing well…'”,
“usage”: {
“completion”: 19,
“prompt”: 10129,
“total”: 10148
}
},
“attributes”: {
“root_agent_name”: “my_bq_agent”,
“model_version”: “gemini-flash-latest”,
“usage_metadata”: {
“prompt_token_count”: 10129,
“candidates_token_count”: 19,
“total_token_count”: 10148
},
“finish_reason”: “STOP”
},
“latency_ms”: {
“time_to_first_token_ms”: 2579,
“total_ms”: 2579
}
}
“`
The Python plugin adds `finish_reason` only to final, non-partial responses. Its `v_llm_response` view exposes it as a `STRING`, along with the final response’s `cache_type`. When the model supplies termination diagnostics, it stores a sanitized value in the common `error_message` column even though the row’s `status` remains `OK`.
In Python, model finish and block reasons remain classified as `LLM_RESPONSE`. The plugin uses `LLM_ERROR` only when a model call raises an exception.
In Python, `content.response` summarizes the response parts, separated by `|`, for example `text: ‘…’` for text, `call: ` for a function call, `resp: ` for a function response, and `other` for other part types. Text longer than `max_content_length` is cut, which sets `is_truncated`. In ADK Python v2.12.0 and later, the summary also shows executable code as `Executable code ():
` and code execution results as `Code execution result ():