bigquery-agent-analytics-sdk

# BigQuery Agent Analytics SDK > Skill by [ara.so](https://ara.so) — Data Skills collection The BigQuery Agent Analytics SDK is an open-source Python toolkit for analyzing AI agent telemetry stored in BigQuery. It provides observability (trace reconstruction, DAG visualization), evaluation (LLM-as-Judge, trajectory matching, multi-trial pass@k), and advanced analytics (Agent Context Graph for decision-trace extraction, drift detection, memory service). Built on top of BigQuery Agent Analytics, it's designed for ML engineers running agents in production who need to measure quality, understand behavior, and detect regressions at scale. ## Installation ```bash # Core SDK pip install bigquery-agent-analytics # With LLM judge support pip install bigquery-agent-analytics[llm] # With BigFrames support pip install bigquery-agent-analytics[bigframes] # All features pip install bigquery-agent-analytics[llm,bigframes] ``` **Prerequisites:** - Python 3.10+ - Google Cloud project with BigQuery enabled - Agent traces in BigQuery (via [ADK BigQuery Trace Exporter](https://github.com/google/adk-python/tree/main/contributing/extensions/bigquery_trace_exporter)) - `GOOGLE_APPLICATION_CREDENTIALS` or `gcloud auth application-default login` configured ## Core Client API ### Initialize Client ```python from bigquery_agent_analytics import Client # Basic initialization client = Client( project_id="my-gcp-project", dataset_id="agent_analytics" ) # With custom location client = Client( project_id="my-project", dataset_id="agent_traces", location="US" ) ``` ### Retrieve and Visualize Traces ```python # Get a single trace trace = client.get_trace("trace-abc-123") # Render as ASCII DAG trace.render() # Get trace as dictionary trace_dict = trace.to_dict() # Get all traces for a session traces = 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 range from datetime import datetime, timedelta end_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 ) ``` ### Event Semantics ```python 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)}") ``` ## CLI Commands The SDK includes `bqaa` (or `bq-agent-sdk`) CLI with 12+ commands: ### Diagnostics ```bash # Check connectivity and permissions bqaa diagnose --project-id $PROJECT_ID --dataset-id agent_analytics # List available traces bqaa list-traces --project-id $PROJECT_ID --dataset-id agent_analytics --limit 10 # Inspect a specific trace bqaa inspect-trace --project-id $PROJECT_ID --dataset-id agent_analytics \ --trace-id "trace-abc-123" # Seed sample events for testing bqaa seed-events --project-id $PROJECT_ID --dataset-id agent_analytics \ --sessions 5 ``` ### Views Management ```bash # Create per-event-type BigQuery views bqaa create-views --project-id $PROJECT_ID --dataset-id agent_analytics # List created views bqaa list-views --project-id $PROJECT_ID --dataset-id agent_analytics # Drop views bqaa drop-views --project-id $PROJECT_ID --dataset-id agent_analytics ``` ### Evaluation ```bash # Run system evaluation (latency, tokens, cost) bqaa evaluate --project-id $PROJECT_ID --dataset-id agent_analytics \ --evaluator system --output-table eval_results # Run LLM-as-Judge evaluation bqaa evaluate --project-id $PROJECT_ID --dataset-id agent_analytics \ --evaluator llm-judge --model-name gemini-2.0-flash-exp \ --criteria correctness,hallucination --output-table llm_eval_results # Run trajectory evaluation bqaa evaluate-trajectory --project-id $PROJECT_ID --dataset-id agent_analytics \ --golden-path "plan,execute,verify" --match-strategy exact ``` ### Agent Context Graph ```bash # Extract decision traces from agent context graph bqaa context-graph --project-id $PROJECT_ID --dataset-id agent_analytics \ --graph agent_decisions_graph --lookback-hours 24 --format json # Schedule periodic materialization (creates Cloud Scheduler + Cloud Run) bqaa schedule-context-graph --project-id $PROJECT_ID \ --dataset-id agent_analytics --graph agent_decisions_graph \ --cron "0 */6 * * *" --region us-central1 ``` ## Observability ### Create Event-Type Views ```python from bigquery_agent_analytics.views import ViewManager view_manager = ViewManager(client) # Create all standard views view_manager.create_all_views() # List created views views = view_manager.list_views() for view in views: print(f"{view['view_name']}: {view['event_type']}") # Query a specific view agent_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}") ``` ### Trace Visualization ```python # Render trace as text trace.render() # Export trace to JSON import json with open("trace.json", "w") as f: json.dump(trace.to_dict(), f, indent=2) # Analyze trace structure print(f"Total events: {len(trace.events)}") print(f"Trace duration: {trace.duration_ms}ms") print(f"Session: {trace.session_id}") # Filter events tool_events = [e for e in trace.events if is_tool_invocation(e)] print(f"Tool invocations: {len(tool_events)}") ``` ## Evaluation ### System Evaluator (Code-Based Metrics) ```python from bigquery_agent_analytics.evaluators import SystemEvaluator evaluator = SystemEvaluator(client) # Run evaluation on recent traces results = evaluator.evaluate( start_time=datetime.utcnow() - timedelta(hours=24), end_time=datetime.utcnow() ) # Results include: latency, token_count, turn_count, error_rate, cost for 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 BigQuery evaluator.save_results(results, output_table="eval_results") ``` ### LLM-as-Judge Evaluation ```python from bigquery_agent_analytics.evaluators import LLMAsJudge judge = LLMAsJudge( client=client, model_name="gemini-2.0-flash-exp", criteria=["correctness", "hallucination", "helpfulness"] ) # Evaluate traces llm_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 BigQuery judge.save_results(llm_results, output_table="llm_eval_results") ``` ### Trajectory Matching ```python 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 matching exact_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}") ``` ### Multi-Trial Evaluation (pass@k) ```python from bigquery_agent_analytics.multi_trial import MultiTrialEvaluator multi_trial = MultiTrialEvaluator(client) # Run N trials per task and compute pass@k results = 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 metrics for 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%}") ``` ### Grader Composition ```python from bigquery_agent_analytics.grader_pipeline import ( GraderPipeline, WeightedAverageStrategy, BinaryThresholdStrategy, MajorityVoteStrategy ) # Weighted average of multiple evaluators pipeline = 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 judges majority_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}") ``` ### Eval Suite Lifecycle Management ```python from bigquery_agent_analytics.eval_suite import EvalSuite suite = EvalSuite(client) # Define eval suite suite.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 suite suite_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()