Bigquery Agent Analytics Sdk

from bigquery_agent_analytics import Client
# Basic initializationclient = Client(    project_id="my-gcp-project",    dataset_id="agent_analytics")
# With custom locationclient = Client(    project_id="my-project",    dataset_id="agent_traces",    location="US")

# Get a single tracetrace = client.get_trace("trace-abc-123")
# Render as ASCII DAGtrace.render()
# Get trace as dictionarytrace_dict = trace.to_dict()
# Get all traces for a sessiontraces = client.get_traces_for_session("session-xyz-456")for trace in traces:    print(f"Trace: {trace.trace_id}, Events: {len(trace.events)}")
# Query traces by time rangefrom datetime import datetime, timedeltaend_time = datetime.utcnow()start_time = end_time - timedelta(hours=24)
recent_traces = client.get_traces(    start_time=start_time,    end_time=end_time,    limit=100)

from bigquery_agent_analytics.event_semantics import (    is_agent_start,    is_agent_end,    is_tool_invocation,    is_llm_call,    get_input_text,    get_output_text)
for event in trace.events:    if is_agent_start(event):        print(f"Agent started: {get_input_text(event)}")    elif is_tool_invocation(event):        print(f"Tool called: {event.get('tool_name')}")    elif is_llm_call(event):        print(f"LLM invoked: {event.get('model_name')}")    elif is_agent_end(event):        print(f"Agent response: {get_output_text(event)}")

from bigquery_agent_analytics.views import ViewManager
view_manager = ViewManager(client)
# Create all standard viewsview_manager.create_all_views()
# List created viewsviews = view_manager.list_views()for view in views:    print(f"{view['view_name']}: {view['event_type']}")
# Query a specific viewagent_starts = client.query("""    SELECT session_id, trace_id, input_text, timestamp    FROM `my-project.agent_analytics.agent_start_events`    WHERE DATE(timestamp) = CURRENT_DATE()    LIMIT 10""")
for row in agent_starts:    print(f"{row.session_id}: {row.input_text}")

# Render trace as texttrace.render()
# Export trace to JSONimport jsonwith open("trace.json", "w") as f:    json.dump(trace.to_dict(), f, indent=2)
# Analyze trace structureprint(f"Total events: {len(trace.events)}")print(f"Trace duration: {trace.duration_ms}ms")print(f"Session: {trace.session_id}")
# Filter eventstool_events = [e for e in trace.events if is_tool_invocation(e)]print(f"Tool invocations: {len(tool_events)}")

from bigquery_agent_analytics.evaluators import SystemEvaluator
evaluator = SystemEvaluator(client)
# Run evaluation on recent tracesresults = evaluator.evaluate(    start_time=datetime.utcnow() - timedelta(hours=24),    end_time=datetime.utcnow())
# Results include: latency, token_count, turn_count, error_rate, costfor result in results:    print(f"Trace {result['trace_id']}: "          f"latency={result['latency_ms']}ms, "          f"tokens={result['token_count']}, "          f"cost=${result['cost']:.4f}")
# Save to BigQueryevaluator.save_results(results, output_table="eval_results")

from bigquery_agent_analytics.evaluators import LLMAsJudge
judge = LLMAsJudge(    client=client,    model_name="gemini-2.0-flash-exp",    criteria=["correctness", "hallucination", "helpfulness"])
# Evaluate tracesllm_results = judge.evaluate_traces(    trace_ids=["trace-1", "trace-2", "trace-3"])
# Results include scores for each criterion (0.0-1.0)for result in llm_results:    print(f"Trace {result['trace_id']}:")    print(f"  Correctness: {result['correctness_score']:.2f}")    print(f"  Hallucination: {result['hallucination_score']:.2f}")    print(f"  Helpfulness: {result['helpfulness_score']:.2f}")
# Save to BigQueryjudge.save_results(llm_results, output_table="llm_eval_results")

from bigquery_agent_analytics.trace_evaluator import TrajectoryEvaluator
trajectory_eval = TrajectoryEvaluator(client)
# Define golden path (sequence of event types or tool names)golden_path = ["search_documents", "extract_entities", "summarize"]
# Exact matchingexact_matches = trajectory_eval.evaluate(    trace_ids=["trace-1", "trace-2"],    golden_path=golden_path,    match_strategy="exact")
# In-order matching (allows extra steps)in_order_matches = trajectory_eval.evaluate(    trace_ids=["trace-1", "trace-2"],    golden_path=golden_path,    match_strategy="in_order")
# Any-order matching (all steps present, any order)any_order_matches = trajectory_eval.evaluate(    trace_ids=["trace-1", "trace-2"],    golden_path=golden_path,    match_strategy="any_order")
for result in exact_matches:    print(f"Trace {result['trace_id']}: "          f"matches={result['matches']}, "          f"similarity={result['similarity']:.2f}")

from bigquery_agent_analytics.multi_trial import MultiTrialEvaluator
multi_trial = MultiTrialEvaluator(client)
# Run N trials per task and compute pass@kresults = multi_trial.evaluate(    task_ids=["task-1", "task-2", "task-3"],    trials_per_task=5,    k_values=[1, 3, 5],    evaluation_fn=lambda trace: judge.evaluate_traces([trace.trace_id])[0])
# Results include pass@1, pass@3, pass@5 metricsfor task_result in results:    print(f"Task {task_result['task_id']}:")    print(f"  pass@1: {task_result['pass_at_1']:.2%}")    print(f"  pass@3: {task_result['pass_at_3']:.2%}")    print(f"  pass@5: {task_result['pass_at_5']:.2%}")

from bigquery_agent_analytics.grader_pipeline import (    GraderPipeline,    WeightedAverageStrategy,    BinaryThresholdStrategy,    MajorityVoteStrategy)
# Weighted average of multiple evaluatorspipeline = GraderPipeline(strategy=WeightedAverageStrategy())pipeline.add_grader(    evaluator=judge,    criteria="correctness",    weight=0.6)pipeline.add_grader(    evaluator=judge,    criteria="helpfulness",    weight=0.4)
composite_score = pipeline.evaluate(trace_id="trace-123")print(f"Composite score: {composite_score:.2f}")
# Binary threshold (pass/fail)binary_pipeline = GraderPipeline(strategy=BinaryThresholdStrategy(threshold=0.7))binary_pipeline.add_grader(evaluator=judge, criteria="correctness")passed = binary_pipeline.evaluate(trace_id="trace-123")print(f"Passed: {passed}")
# Majority vote across multiple judgesmajority_pipeline = GraderPipeline(strategy=MajorityVoteStrategy())majority_pipeline.add_grader(evaluator=judge, criteria="correctness")majority_pipeline.add_grader(evaluator=judge, criteria="hallucination")majority_pipeline.add_grader(evaluator=judge, criteria="helpfulness")result = majority_pipeline.evaluate(trace_id="trace-123")print(f"Majority vote: {result}")

from bigquery_agent_analytics.eval_suite import EvalSuite
suite = EvalSuite(client)
# Define eval suitesuite.add_test(    test_id="greeting_test",    input_text="Hello, how are you?",    expected_tools=["greeting_handler"],    min_correctness=0.8)
suite.add_test(    test_id="search_test",    input_text="Find documents about AI",    expected_tools=["search_documents", "extract_entities"],    min_correctness=0.9)
# Run suitesuite_results = suite.run()
# Check graduation criteria (all tests pass threshold)if suite.check_graduation(min_pass_rate=0.95):    print("✓ Suite ready for production")    suite.graduate()
# Check saturation (no improvement over N runs)if suite.check_saturation(lookback_runs=10, improvement_threshold=0.01):    print("⚠ Suite saturated - consider adding new tests")
# Save suite resultssuite.save_results(output_table="eval_suite_results")

from bigquery_agent_analytics.context_graph import ContextGraph
graph = ContextGraph(    client=client,    graph_name="agent_decisions_graph")
# Materialize window (extract traces from agent_events)graph.materialize_window(    lookback_hours=24,    extract_fn=lambda events: {        "request_id": events[0].get("trace_id"),        "request_text": events[0].get("input_text"),        "timestamp": events[0].get("timestamp"),        "options": [            {"option_id": f"opt-{i}", "option_label": opt}            for i, opt in enumerate(events[0].get("options", []))        ],        "outcome": {            "outcome_id": f"out-{events[-1].get('trace_id')}",            "status": events[-1].get("status"),            "rationale": events[-1].get("rationale")        }    })
# Query graph with GQLdecision_traces = client.query("""SELECT * FROM GRAPH_TABLE(  agent_analytics.agent_decisions_graph  MATCH (req:DecisionRequest)-[eo:evaluatesOption]->(opt:DecisionOption),        (req)-[ri:resultedIn]->(out:DecisionOutcome)  COLUMNS (    req.request_text AS question,    opt.option_label AS considered,    out.status AS outcome,    out.rationale AS rationale  ))WHERE req.timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 24 HOUR)LIMIT 100""")
for row in decision_traces:    print(f"Q: {row.question} | Considered: {row.considered} | "          f"Outcome: {row.outcome} | Rationale: {row.rationale}")

from bigquery_agent_analytics.ai_ml_integration import AIMLIntegration
ai_ml = AIMLIntegration(client)
# Generate summaries with AI.GENERATEsummaries = ai_ml.generate_summaries(    input_table="agent_analytics.agent_end_events",    prompt_column="output_text",    model_name="gemini-2.0-flash-exp",    output_table="trace_summaries")
# Classify sentiment with AI.CLASSIFYclassifications = ai_ml.classify_sentiment(    input_table="agent_analytics.agent_end_events",    text_column="output_text",    output_table="sentiment_results")
# Generate embeddings with AI.EMBEDembeddings = ai_ml.generate_embeddings(    input_table="agent_analytics.agent_start_events",    text_column="input_text",    embedding_model="textembedding-gecko@003",    output_table="input_embeddings")
# Anomaly detection on latencyanomalies = ai_ml.detect_anomalies(    metric_table="eval_results",    metric_column="latency_ms",    timestamp_column="timestamp",    output_table="latency_anomalies")

from bigquery_agent_analytics.bigframes_evaluator import BigFramesEvaluatorimport bigframes.pandas as bpd
bf_eval = BigFramesEvaluator(client)
# Load traces as BigFrames DataFramedf = bpd.read_gbq("agent_analytics.agent_events")
# Compute metricsmetrics_df = bf_eval.compute_metrics(    df,    metrics=["latency", "token_count", "error_rate"])
# Aggregate by sessionsession_metrics = metrics_df.groupby("session_id").agg({    "latency_ms": "mean",    "token_count": "sum",    "error_rate": "mean"})
# Save resultssession_metrics.to_gbq("agent_analytics.session_metrics")

from bigquery_agent_analytics.categorical_evaluator import CategoricalEvaluator
cat_eval = CategoricalEvaluator(client)
# Define categoriescategories = {    "greeting": ["hello", "hi", "hey"],    "search": ["find", "search", "lookup"],    "summarize": ["summarize", "summary", "tldr"]}
# Classify tracesclassifications = cat_eval.classify_traces(    trace_ids=["trace-1", "trace-2", "trace-3"],    categories=categories)
# Compute per-category metricscategory_metrics = cat_eval.compute_category_metrics(    classifications=classifications,    evaluator=judge)
for category, metrics in category_metrics.items():    print(f"{category}: accuracy={metrics['accuracy']:.2%}, "          f"precision={metrics['precision']:.2%}")

from bigquery_agent_analytics.feedback import DriftDetector
drift = DriftDetector(client)
# Compare question distributions (golden vs production)drift_results = drift.detect_drift(    golden_table="agent_analytics.golden_questions",    production_table="agent_analytics.agent_start_events",    text_column="input_text",    lookback_days=7)
if drift_results["drift_detected"]:    print(f"⚠ Drift detected! KL divergence: {drift_results['kl_divergence']:.4f}")    print(f"Top drifting terms: {drift_results['top_drifting_terms']}")else:    print("✓ No significant drift detected")
# Save drift reportdrift.save_report(drift_results, output_table="drift_reports")

from bigquery_agent_analytics.insights import InsightsPipeline
pipeline = InsightsPipeline(client)
# Stage 1: Extract featurespipeline.add_stage(    name="feature_extraction",    query="""    SELECT trace_id, session_id, latency_ms, token_count, tool_count    FROM `agent_analytics.eval_results`    WHERE DATE(timestamp) = CURRENT_DATE()    """)
# Stage 2: Aggregate by sessionpipeline.add_stage(    name="session_aggregation",    query="""    SELECT session_id,           AVG(latency_ms) AS avg_latency,           SUM(token_count) AS total_tokens,           COUNT(*) AS trace_count    FROM {feature_extraction}    GROUP BY session_id    """)
# Stage 3: Identify outlierspipeline.add_stage(    name="outlier_detection",    query="""    SELECT * FROM {session_aggregation}    WHERE avg_latency > (SELECT APPROX_QUANTILES(avg_latency, 100)[OFFSET(95)]                         FROM {session_aggregation})    """)
# Run pipelineresults = pipeline.run()outlier_sessions = results["outlier_detection"]
for row in outlier_sessions:    print(f"Session {row.session_id}: avg_latency={row.avg_latency}ms")

from bigquery_agent_analytics.memory_service import MemoryService
memory = MemoryService(client)
# Store agent memorymemory.store(    session_id="session-123",    memory_key="user_preferences",    memory_value={"theme": "dark", "language": "en"},    ttl_hours=720  # 30 days)
# Retrieve memoryprefs = memory.retrieve(    session_id="session-123",    memory_key="user_preferences")print(f"User preferences: {prefs}")
# Cross-session retrieval (find similar sessions)similar_sessions = memory.find_similar_sessions(    current_session_id="session-123",    similarity_threshold=0.8,    max_results=5)
for session in similar_sessions:    print(f"Similar session: {session['session_id']}, "          f"similarity={session['similarity']:.2f}")

# Custom BigQuery clientfrom google.cloud import bigquery
bq_client = bigquery.Client(    project="my-project",    location="EU")
client = Client(    project_id="my-project",    dataset_id="agent_analytics",    bq_client=bq_client  # Use custom client)
# Custom job labelsclient.default_job_labels = {    "team": "ml-platform",    "env": "production"}

from datetime import datetime, timedelta
# Evaluate last hour every hourdef continuous_eval_job():    end_time = datetime.utcnow()    start_time = end_time - timedelta(hours=1)        # System metrics    system_eval = SystemEvaluator(client)    sys_results = system_eval.evaluate(start_time, end_time)    system_eval.save_results(sys_results, "hourly_system_metrics")        # LLM judge    judge = LLMAsJudge(client, model_name="gemini-2.0-flash-exp")    llm_results = judge.evaluate_traces(        trace_ids=[r["trace_id"] for r in sys_results]    )    judge.save_results(llm_results, "hourly_llm_scores")        # Alert on anomalies    ai_ml = AIMLIntegration(client)    anomalies = ai_ml.detect_anomalies(        metric_table="hourly_system_metrics",        metric_column="latency_ms",        timestamp_column="timestamp"    )        if len(anomalies) > 0:        alert_on_call(anomalies)
# Schedule with Cloud Scheduler# (or run as cron in your environment)

# Compare production against golden datasetgolden_traces = client.get_traces_for_session("golden-session-id")prod_traces = client.get_traces(    start_time=datetime.utcnow() - timedelta(hours=1),    end_time=datetime.utcnow())
trajectory_eval = TrajectoryEvaluator(client)
for prod_trace in prod_traces[:10]:  # Sample    for golden_trace in golden_traces:        result = trajectory_eval.evaluate(            trace_ids=[prod_trace.trace_id],            golden_path=extract_golden_path(golden_trace),            match_strategy="in_order"        )        if not result[0]["matches"]:            print(f"⚠ Trace {prod_trace.trace_id} diverged from golden path")

# Query job labels for cost attributioncost_query = """SELECT  labels.value AS team,  SUM(total_bytes_billed) / POW(10, 12) AS tb_billed,  SUM(total_slot_ms) / 1000 / 3600 AS slot_hoursFROM `region-us`.INFORMATION_SCHEMA.JOBS_BY_PROJECTCROSS JOIN UNNEST(labels) AS labelsWHERE labels.key = 'team'  AND creation_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)  AND state = 'DONE'GROUP BY teamORDER BY tb_billed DESC"""
cost_by_team = client.query(cost_query)for row in cost_by_team:    print(f"{row.team}: {row.tb_billed:.2f} TB billed, "          f"{row.slot_hours:.2f} slot-hours")

# Sample traces for human reviewimport random
all_traces = client.get_traces(    start_time=datetime.utcnow() - timedelta(days=1),    end_time=datetime.utcnow())
# Stratified sampling: by latency bucketslow_latency = [t for t in all_traces if t.duration_ms < 1000]mid_latency = [t for t in all_traces if 1000 <= t.duration_ms < 5000]high_latency = [t for t in all_traces if t.duration_ms >= 5000]
sample = (    random.sample(low_latency, min(10, len(low_latency))) +    random.sample(mid_latency, min(10, len(mid_latency))) +    random.sample(high_latency, min(