Exporting Google ADK Agent Logs to BigQuery: A Complete Guide

Introduction

When building AI agents with Google’s Agent Development Kit (ADK), you quickly accumulate valuable execution data: conversation histories, tool calls, reasoning traces, and performance metrics. This data is a goldmine for understanding how your agents behave in production.

Exporting these logs to BigQuery transforms raw execution data into actionable insights. You can:

• Debug complex multi-turn conversations by replaying exact agent decision paths
• Monitor agent performance with custom dashboards and alerts
• Analyze usage patterns to identify common user queries and failure modes
• Train and improve your agents using historical interaction data

This guide walks you through setting up a complete logging pipeline from ADK agents to BigQuery.

Setting up Google Cloud and BigQuery

Before you start exporting logs, ensure your Google Cloud project is properly configured.

Prerequisites

1. A Google Cloud project with billing enabled
2. The gcloud CLI installed and authenticated
3. Python 3.9+ environment

Enable the BigQuery API

# Set your project
export PROJECT_ID="your-project-id"
gcloud config set project $PROJECT_ID

# Enable required APIs
gcloud services enable bigquery.googleapis.com
gcloud services enable logging.googleapis.com

Create a BigQuery Dataset

# Create a dataset for agent logs
bq mk --dataset --location=US $PROJECT_ID:adk_logs

Set up Authentication

For local development, use Application Default Credentials:

gcloud auth application-default login

For production, create a service account with BigQuery permissions:

# Create service account
gcloud iam service-accounts create adk-logger \
  --display-name="ADK Logger"

# Grant BigQuery permissions
gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="serviceAccount:adk-logger@$PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/bigquery.dataEditor"

# Download credentials
gcloud iam service-accounts keys create ./adk-logger-key.json \
  --iam-account=adk-logger@$PROJECT_ID.iam.gserviceaccount.com

# Set environment variable
export GOOGLE_APPLICATION_CREDENTIALS="./adk-logger-key.json"

Configuring ADK Log Structure

Design your log schema thoughtfully. A good schema captures essential agent execution data while remaining queryable.

Understanding ADK Event Types

ADK agents typically produce these event types:

Recommended Schema

Create the logs table with partitioning for performance:

CREATE TABLE `YOUR_PROJECT_ID.adk_logs.agent_executions` (
    session_id STRING NOT NULL,
    agent_name STRING,
    timestamp TIMESTAMP NOT NULL,
    event_type STRING,
    event_data JSON,
    duration_ms INT64,
    status STRING,
    user_id STRING,
    input_tokens INT64,
    output_tokens INT64,
    metadata JSON
)
PARTITION BY DATE(timestamp)
CLUSTER BY agent_name, event_type;

Define the Log Model

Use Pydantic for consistent log structure:

from pydantic import BaseModel
from datetime import datetime
from typing import Optional, Any

class AgentLog(BaseModel):
    session_id: str
    agent_name: str
    timestamp: datetime
    event_type: str
    event_data: dict[str, Any]
    duration_ms: Optional[int] = None
    status: str = "success"
    user_id: Optional[str] = None
    input_tokens: Optional[int] = None
    output_tokens: Optional[int] = None
    metadata: dict[str, Any] = {}

Code Examples for Logging Integration

Now let’s build the logging infrastructure that writes to BigQuery.

Basic Logger Class

from google.cloud import bigquery
from datetime import datetime
import uuid
import time
import json
from typing import Optional, Any

class ADKBigQueryLogger:
    """Logger that streams ADK agent events to BigQuery."""

    def __init__(
        self,
        project_id: str,
        dataset: str = "adk_logs",
        table: str = "agent_executions"
    ):
        self.client = bigquery.Client(project=project_id)
        self.table_id = f"{project_id}.{dataset}.{table}"
        self.session_id = str(uuid.uuid4())

    def log(
        self,
        agent_name: str,
        event_type: str,
        event_data: dict[str, Any],
        duration_ms: Optional[int] = None,
        status: str = "success",
        user_id: Optional[str] = None,
        input_tokens: Optional[int] = None,
        output_tokens: Optional[int] = None,
        metadata: Optional[dict] = None
    ) -> None:
        """Log an agent event to BigQuery."""

        row = {
            "session_id": self.session_id,
            "agent_name": agent_name,
            "timestamp": datetime.utcnow().isoformat(),
            "event_type": event_type,
            "event_data": json.dumps(event_data),
            "duration_ms": duration_ms,
            "status": status,
            "user_id": user_id,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "metadata": json.dumps(metadata or {})
        }

        errors = self.client.insert_rows_json(self.table_id, [row])
        if errors:
            print(f"BigQuery insert errors: {errors}")

Batched Logger for High Volume

For production workloads, batch inserts improve performance:

from collections import deque
import threading

class BatchedADKLogger(ADKBigQueryLogger):
    """Buffered logger for high-throughput scenarios."""

    def __init__(
        self,
        project_id: str,
        dataset: str = "adk_logs",
        table: str = "agent_executions",
        batch_size: int = 100,
        flush_interval: float = 5.0
    ):
        super().__init__(project_id, dataset, table)
        self.batch_size = batch_size
        self.flush_interval = flush_interval
        self._buffer: deque = deque()
        self._lock = threading.Lock()
        self._start_flush_thread()

    def log(self, **kwargs) -> None:
        """Add event to buffer, flush if batch size reached."""
        row = {
            "session_id": self.session_id,
            "timestamp": datetime.utcnow().isoformat(),
            **kwargs
        }

        with self._lock:
            self._buffer.append(row)
            if len(self._buffer) >= self.batch_size:
                self._flush_buffer()

    def _flush_buffer(self) -> None:
        """Send buffered rows to BigQuery."""
        if not self._buffer:
            return

        rows = list(self._buffer)
        self._buffer.clear()

        errors = self.client.insert_rows_json(self.table_id, rows)
        if errors:
            print(f"BigQuery batch insert errors: {errors}")

    def _start_flush_thread(self) -> None:
        """Background thread for periodic flushing."""
        def flush_loop():
            while True:
                time.sleep(self.flush_interval)
                with self._lock:
                    self._flush_buffer()

        thread = threading.Thread(target=flush_loop, daemon=True)
        thread.start()

Integrating with ADK Agents

Wrap your ADK agent with logging:

from google.adk.agents import Agent

class LoggedAgent:
    """ADK Agent wrapper with BigQuery logging."""

    def __init__(self, agent: Agent, logger: ADKBigQueryLogger):
        self.agent = agent
        self.logger = logger

    async def run(
        self,
        user_input: str,
        user_id: Optional[str] = None
    ) -> str:
        """Execute the agent with comprehensive logging."""
        start_time = time.time()

        # Log agent start
        self.logger.log(
            agent_name=self.agent.name,
            event_type="agent_start",
            event_data={"user_input": user_input},
            user_id=user_id
        )

        try:
            # Run the agent
            response = await self.agent.run(user_input)
            status = "success"
            error_message = None

            # Log successful response
            self.logger.log(
                agent_name=self.agent.name,
                event_type="llm_response",
                event_data={
                    "response_preview": response[:500],
                    "response_length": len(response)
                },
                user_id=user_id
            )

            return response

        except Exception as e:
            status = "error"
            error_message = str(e)

            self.logger.log(
                agent_name=self.agent.name,
                event_type="error",
                event_data={
                    "error_type": type(e).__name__,
                    "message": str(e)
                },
                status="error",
                user_id=user_id
            )
            raise

        finally:
            duration_ms = int((time.time() - start_time) * 1000)

            self.logger.log(
                agent_name=self.agent.name,
                event_type="agent_end",
                event_data={},
                duration_ms=duration_ms,
                status=status,
                user_id=user_id
            )

Decorator for Tool Logging

import functools

def logged_tool(logger: ADKBigQueryLogger, tool_name: str):
    """Decorator to automatically log tool invocations."""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            session_id = kwargs.get('session_id')
            user_id = kwargs.get('user_id')
            start_time = time.time()

            try:
                result = await func(*args, **kwargs)
                status = "success"
                return result
            except Exception as e:
                status = "error"
                raise
            finally:
                duration_ms = int((time.time() - start_time) * 1000)
                logger.log(
                    agent_name="tool_executor",
                    event_type="tool_call",
                    event_data={
                        "tool_name": tool_name,
                        "kwargs": {k: str(v)[:200] for k, v in kwargs.items()}
                    },
                    duration_ms=duration_ms,
                    status=status,
                    user_id=user_id
                )
        return wrapper
    return decorator

# Usage
@logged_tool(logger, "search_documents")
async def search_documents(query: str, limit: int = 5) -> list:
    """Search the document knowledge base."""
    return documents

Querying and Analyzing the Data

With logs flowing into BigQuery, you can write powerful analytics queries.

Daily Execution Summary

SELECT
    DATE(timestamp) as date,
    agent_name,
    COUNT(DISTINCT session_id) as total_sessions,
    COUNTIF(status = 'error') as error_count,
    ROUND(AVG(duration_ms), 2) as avg_latency_ms,
    SUM(input_tokens) as total_input_tokens,
    SUM(output_tokens) as total_output_tokens
FROM `YOUR_PROJECT_ID.adk_logs.agent_executions`
WHERE event_type = 'agent_end'
GROUP BY date, agent_name
ORDER BY date DESC, agent_name;

Error Analysis

SELECT
    agent_name,
    JSON_EXTRACT_SCALAR(event_data, '$.error_type') as error_type,
    JSON_EXTRACT_SCALAR(event_data, '$.message') as error_message,
    COUNT(*) as occurrence_count
FROM `YOUR_PROJECT_ID.adk_logs.agent_executions`
WHERE status = 'error'
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
GROUP BY agent_name, error_type, error_message
ORDER BY occurrence_count DESC
LIMIT 20;

Performance Metrics

SELECT
    agent_name,
    COUNT(*) as executions,
    ROUND(AVG(duration_ms), 2) as avg_ms,
    ROUND(MIN(duration_ms), 2) as min_ms,
    ROUND(MAX(duration_ms), 2) as max_ms,
    APPROX_QUANTILES(duration_ms, 100)[OFFSET(95)] as p95_ms
FROM `YOUR_PROJECT_ID.adk_logs.agent_executions`
WHERE event_type = 'agent_end'
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)
GROUP BY agent_name;

Tool Usage Analytics

SELECT
    JSON_EXTRACT_SCALAR(event_data, '$.tool_name') as tool_name,
    COUNT(*) as call_count,
    ROUND(AVG(duration_ms), 2) as avg_duration_ms,
    SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) as errors
FROM `YOUR_PROJECT_ID.adk_logs.agent_executions`
WHERE event_type = 'tool_call'
  AND timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
GROUP BY tool_name
ORDER BY call_count DESC;

Token Usage by User

SELECT
    user_id,
    COUNT(DISTINCT session_id) as sessions,
    SUM(input_tokens) as input_tokens,
    SUM(output_tokens) as output_tokens,
    ROUND(
        (SUM(input_tokens) * 0.00001 + SUM(output_tokens) * 0.00003),
        2
    ) as estimated_cost_usd
FROM `YOUR_PROJECT_ID.adk_logs.agent_executions`
WHERE user_id IS NOT NULL
  AND event_type = 'agent_end'
GROUP BY user_id
ORDER BY input_tokens DESC
LIMIT 50;

Conversation Replay

Reconstruct agent reasoning for debugging:

SELECT
    timestamp,
    event_type,
    JSON_EXTRACT_SCALAR(event_data, '$.user_input') as user_input,
    JSON_EXTRACT_SCALAR(event_data, '$.response_preview') as response,
    duration_ms
FROM `YOUR_PROJECT_ID.adk_logs.agent_executions`
WHERE session_id = 'YOUR_SESSION_ID'
ORDER BY timestamp;

Best Practices

1. Sanitize Sensitive Data

Never log PII or credentials:

SENSITIVE_FIELDS = {"api_key", "password", "token", "ssn", "email"}

def sanitize_for_logging(data: dict) -> dict:
    """Remove sensitive fields before logging."""
    return {
        k: "[REDACTED]" if k.lower() in SENSITIVE_FIELDS else v
        for k, v in data.items()
    }

2. Use Partitioned Tables

Partitioning by date improves query performance and reduces costs:

PARTITION BY DATE(timestamp)
CLUSTER BY agent_name, event_type

3. Implement Retention Policies

Set up automatic expiration for old data:

ALTER TABLE `YOUR_PROJECT_ID.adk_logs.agent_executions`
SET OPTIONS (
    partition_expiration_days = 90
);

4. Handle Failures Gracefully

Logging should never block your agent:

class ResilientLogger(ADKBigQueryLogger):
    """Logger that handles BigQuery failures gracefully."""

    def log(self, **kwargs) -> None:
        try:
            super().log(**kwargs)
        except Exception as e:
            logging.error(f"Failed to log to BigQuery: {e}")

5. Create Views for Common Queries

Simplify dashboard queries:

CREATE VIEW `YOUR_PROJECT_ID.adk_logs.daily_metrics` AS
SELECT
    DATE(timestamp) as date,
    agent_name,
    COUNT(*) as events,
    ROUND(AVG(duration_ms), 2) as avg_latency,
    SUM(input_tokens) as input_tokens,
    SUM(output_tokens) as output_tokens
FROM `YOUR_PROJECT_ID.adk_logs.agent_executions`
GROUP BY date, agent_name;

6. Sample High-Volume Logs

For agents with massive throughput, consider sampling:

import random

def should_log(sample_rate: float = 0.1) -> bool:
    """Determine if this request should be logged."""
    return random.random() < sample_rate

Conclusion

Exporting ADK agent logs to BigQuery unlocks powerful observability and analytics capabilities. With the right schema, logging infrastructure, and queries, you can:

• Debug production issues quickly by replaying agent executions
• Monitor performance and set up alerts for anomalies
• Understand user behavior and agent usage patterns
• Optimize token usage and reduce costs

Start with the basic logger for development, then graduate to batched inserts as your traffic grows. The investment in proper logging pays dividends when troubleshooting production issues or iterating on your agent’s behavior.

For further reading:

• BigQuery Best Practices
• Google ADK Documentation
• BigQuery Schema Design