使用 BigQuery 智能体分析

BigQuery 智能体分析是一种开源解决方案，可让您通过将原始智能体互动（例如请求、响应、工具调用和错误）直接流式传输到 BigQuery，大规模捕获、分析和直观呈现多模态智能体互动数据。借助此解决方案，您可以执行 AI 驱动的评估、优化智能体提示，以及提取长期记忆以增强未来的互动。 代理开发套件 (ADK) 和 LangGraph（预览版）支持 BigQuery 智能体分析。

架构

BigQuery 智能体分析使用 BigQuery Storage Write API将智能体活动数据流式传输到 BigQuery，该 API 提供 高吞吐量、低延迟的日志流式传输，而不会阻止智能体执行。

数据传输包含以下阶段：

  1. 捕获 。使用智能体开发套件 (ADK) 中的插件或 LangGraph 中的回调拦截互动事件，以捕获智能体事件。
  2. 流式传输 。互动事件通过 Storage Write API 发送到 BigQuery。如果不存在标准化架构，智能体会自动创建一个。
  3. 消耗。分析和评估已记录的智能体数据。您可以使用 SQL 查询原始数据 ，在自定义信息中心跟踪指标，或使用 BigQuery 智能体分析 SDK 重建和评估复杂的多轮智能体执行轨迹。

图表：显示了代理活动数据从编排框架流入 BigQuery 以供分析。

智能体分析的优势

  • 只需一行代码即可启用全面日志记录，并自动执行架构管理。
  • 使用对象表记录和分析多模态数据，例如文本、图片、视频和音频。
  • 在稳健的预定义架构中跟踪运营指标，例如令牌消耗量和延迟时间。
  • 使用 BigQuery 生成式 AI 函数和向量搜索来发现优化机会。
  • 通过精细的访问权限控制、数据遮盖和加密来保护智能体日志。

捕获智能体日志数据的方式

如需将智能体的互动遥测数据（请求、响应、工具调用和错误日志）以原生方式捕获到 BigQuery 中，您可以通过多种方式记录事件数据：

  • 编排框架插件：使用智能体编排工具包提供的标准日志记录插件。例如，代理开发套件 (ADK) 中的 BigQueryAgentAnalyticsPlugin 会挂钩到智能体运行程序，以自动拦截、序列化和流式传输事件。
  • 框架回调处理程序：在热门 智能体环境中集成标准回调。例如，您可以使用 LangGraph 和 LangChain 中的内置 BigQuery 处理程序来拦截和转发轨迹。
  • 直接 API 注入：对于自定义或专有框架，请使用 Cloud de Confiance by S3NS 客户端库通过 Storage Write API 将结构化事件直接流式传输到 事件表中。

无论采用哪种方法，所有日志记录选项都使用低延迟、 高吞吐量的 BigQuery Storage Write API。 此 API 提供了一个稳健的流式传输端点，该端点会在提交行之前在内存中异步缓冲和序列化行（使用 PyArrow 引擎），确保可观测性流水线任务不会阻止面向用户的智能体执行轮次。

分析智能体日志数据的方式

如需了解和优化智能体的性能，请通过以下方式分析和评估互动日志：

  • 直接 SQL 查询：在 BigQuery 中运行自定义查询，以 计算令牌消耗量和执行延迟时间等指标。您还可以 使用 AI.GENERATE 自动执行错误根本原因分析，或与业务表执行 联接以衡量业务影响。
  • 互动式信息中心：将 数据洞察等可视化工具连接到预构建或自定义 BigQuery 视图，以跟踪智能体的运行状况、错误率和使用 趋势。
  • Jupyter 笔记本：在互动式环境中使用 Python 库、pandas 或 BigFrames 探索和试验日志 数据。
  • Python SDK：以编程方式直接在应用代码或自动评估流水线中查询、重建和 审核智能体执行轨迹。

使用智能体日志数据的示例

以下是在 BigQuery 中使用智能体日志数据的常见用例和示例。

可观测性和运营指标

  • 查询数据 以按智能体流程细分费用，并确定特定智能体（例如优化智能体）与最终响应的贡献相比，是否消耗了不成比例的令牌。
  • 使用 BigQuery 对话分析智能体 通过运行带有 the AI.GENERATE 函数的查询，进行 AI 驱动的根本原因分析。 例如，“分析此对话日志并说明失败的根本原因。”

智能体评估和质量分析

  • 使用 AI.SCORE 函数 对对话进行排名，并衡量智能体排名随时间的变化。
  • 使用带有 向量搜索的 SQL 查询来识别智能体未能帮助用户的对话集群，然后 将这些集群与用户的原始意图进行比较。这有助于您发现智能体工具或知识库中的差距。

业务洞见和情境化

如需情境化智能体数据，请将 agent_events 表与其他业务表联接。例如，显示与 AI 智能体互动的客户与使用搜索栏的客户的平均订单价值 (AOV)。

如需查看更多示例，请参阅高级分析查询。

使用 Jupyter 笔记本处理智能体日志

使用此示例 Colab Jupyter 笔记本 以互动方式查询、直观呈现和评估智能体日志。

使用 BigQuery 智能体分析 SDK

BigQuery 智能体分析 SDK 是一个开源 Python 库，可为长期 智能体可观测性提供使用和评估层。信息中心和笔记本非常适合临时探索，而 SDK 提供了一种大规模系统地分析和审核智能体行为的方法。

SDK 的功能

您可以使用 BigQuery 智能体分析 SDK 执行以下任务。如需查看日志分析的详细示例， 请参阅 SDK 的 GitHub 代码库。

  • 轨迹重建：将多态事件日志重建为事件的因果 链，以调试跨多轮的会话，包括嵌套的 工具和 LLM 调用。
  • 确定性评估和语义评估：根据 基于规则的标准（例如延迟时间、轮次计数和错误率）以及 语义标准（例如正确性、情绪和幻觉）评估智能体质量。
  • 轨迹匹配：将实际智能体执行路径与 预期黄金轨迹进行比较，以验证步骤效率以及工具 是否按正确的顺序使用。
  • 行为监控和漂移检测：对非确定性智能体输出执行统计分析 ，监控用户请求分布 ，并检测生产回归或语义漂移。
  • 长时程智能体记忆：为智能体提供跨会话情境、 用户个人资料语义检索以及以令牌预算为依据的情节记忆， 这些记忆以原生方式存储在 BigQuery 中。

记录和分析智能体活动

将 SDK 集成到智能体工作流中通常涉及以下步骤：

  1. 记录互动：在智能体编排框架中附加日志记录器插件（例如 ADK 中的BigQueryAgentAnalyticsPlugin）或回调处理程序。当用户与智能体互动时，日志会使用高吞吐量 Storage Write API 异步流式传输到 BigQuery。

  2. 初始化客户端：从 Python SDK 连接到日志数据集：

    from google.cloud import bigquery
from bigquery_agent_analytics import Client

client = Client(
    project_id="YOUR_PROJECT_ID",
    dataset_id="YOUR_DATASET_ID",
    table_id="agent_events",
)

    该代码段使用以下组件：

    • client：父级 Client 实例，用于以编程方式路由查询和管理活跃数据库连接。
    • project_id：包含目标数据集的项目 ID。 Cloud de Confiance by S3NS
    • dataset_id：存储日志的 BigQuery 数据集名称。
    • table_id：存储遥测事件的特定表（默认情况下为 agent_events）。

  3. 重建会话轨迹：提取特定对话会话以 直观呈现和查看确切的事件序列：

    trace = client.get_trace(
    session_id="YOUR_SESSION_ID"
)
trace.render()

    该代码段使用以下组件：

    • trace：已填充的 Trace 对象，其中包含所有用户和智能体操作的重建的因果关联分层 DAG 跨度树。
    • YOUR_SESSION_ID：您要检查的会话的唯一 ID。

  4. 运行自动评估：以编程方式对会话轨迹进行评分 或针对黄金测试套件检查回归：

    from bigquery_agent_analytics.evaluators import CodeEvaluator, LLMAsJudge
from bigquery_agent_analytics.grader_pipeline import GraderPipeline

# Create a grader pipeline with deterministic and semantic metrics
evaluator = GraderPipeline(
    graders=[
        CodeEvaluator.latency(threshold_ms=5000),
        LLMAsJudge.correctness(),
    ]
)
report = client.evaluate(evaluator, session_ids=["session_1", "session_2"])
print(f"Evaluation Pass Rate: {report.pass_rate:.2%}")

    该代码段使用以下组件：

    • evaluator：结构化的 GraderPipeline 编译逻辑，用于组成异构的基于规则的评估指标和语义评估指标。
    • session_ids：要在 BigQuery 中针对其运行批量评估的会话 ID 字符串列表。
    • report：生成的 EvaluationReport 对象，其中包含原始会话等级、指标摘要、试用统计信息和自动评分器反馈。

将 BigQuery 智能体分析集成到工作流中

如需将 BigQuery 智能体分析集成到工作流中，请参阅框架的文档：

后续步骤