用 BigQuery、Pub/Sub 与 ADK 构建事件驱动数据智能体

#用 BigQuery、Pub/Sub 与 ADK 构建事件驱动数据智能体

数据的价值取决于你对它采取行动的能力。在现代企业中，事件发生数小时甚至数分钟之后才反应，往往已经太晚。无论你处理的是金融欺诈，还是动态供应链中断，每一秒都重要。

但许多系统仍依赖缓慢的定时批处理作业，或不断拉取变更的脆弱微服务。等到问题浮出水面，常常已经太晚。这让人工调查人员忙于翻查日志和数据库查询来拼凑情况。这是一个缓慢、痛苦、无法扩展的过程。

进入事件驱动数据智能体

如果数据平台不必等待缓慢的流水线和人工分诊，而是在检测到异常的瞬间就推送警报，并触发自主人工智能智能体去调查和解决，会怎样？

这就是事件驱动数据智能体（Event-Driven Data Agent）架构的承诺。把 BigQuery 连续查询（continuous queries）、Pub/Sub 和 Vertex AI 智能体引擎（Vertex AI Agent Engine） 上的 ADK 智能体（ADK Agents） 组合起来，你可以构建一条实时分诊事件并自主调查它们的流水线。智能体使用进阶推理收集上下文、分析数据，并当场解决问题，或在需要人在回路（human-in-the-loop）干预时升级给一个人。

混合架构：它如何工作

这条事件驱动流水线使用三个核心构建块：

1. 检测（Detection）： BigQuery 连续查询监控实时数据流，并用基于规则的引擎检测异常。

2. 路由（Routing）： Pub/Sub 可靠地投递这些事件，使用单消息转换（Single Message Transforms，SMT）把载荷重塑成人工智能智能体所期望的精确格式，从而触发智能体流水线开始调查。

3. 处置（Resolution）： 一个用 ADK 构建的 Vertex AI 智能体接收该事件，使用自定义工具调查，并记录其决定。

下面逐一查看每个组件。为了具体化，我们将演练一个简单用例：实时检测并调查欺诈性金融交易。

第 1 部分：BigQuery 连续查询

BigQuery 连续查询让你使用标准 SQL 原生构建实时事件流。它们是持续运行的持久 SQL 查询，分析传入数据，并立即把 SQL 结果导出到 Pub/Sub 等目的地。

从拉取转变为在 BigQuery 内原生推送流式事件，意味着你可以用标准 SQL 在数据仓库内检测复杂异常（例如用户在你指定的窗口内于两个不同国家发生交易）。无需把数据搬到单独的流分析引擎。

这一转变由已公开发布预览的 BigQuery 连续查询有状态数据处理支撑。它原生支持流到流的 JOIN、窗口聚合和滚动窗口（tumbling windows）。它允许你关联不同的数据流，并直接在 BigQuery 中计算滚动平均值或总和等复杂指标，从而让任何 SQL 用户都能做流处理。这消除了为构建实时“行动系统（System of Action）”而依赖专门外部工具或深厚数据科学专业知识的需要。该系统在事件发生时检测并做出反应。这种方法也有助于管理大语言模型（LLM）的令牌（token）成本；用有状态 SQL 过滤出特定异常，可以确保智能体只处理它们所需的精确上下文，而不是被原始数据淹没。

实现这一点很直接。 把标准 SQL 查询与 EXPORT DATA 语句组合，就可以在匹配行出现的那一秒把它们直接路由到 Pub/Sub 主题：

EXPORT DATA OPTIONS (
  format = "CLOUD_PUBSUB",
  uri = "https://pubsub.googleapis.com/projects/YOUR_PROJECT_ID/topics/cymbal-bank-escalations-topic"
) AS (
  WITH TransactionHeuristics AS (
    SELECT
      *,
      _CHANGE_TIMESTAMP AS bq_changed_ts,
    FROM APPENDS(TABLE `cymbal_bank.retail_transactions`, CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  )
  SELECT
    TO_JSON_STRING(STRUCT(
      window_end,
      user_id,
      COUNT(*) AS tx_count,
      SUM(amount) AS total_window_spend,
      MAX_BY(merchant_name, amount) AS highest_value_merchant,
      MAX_BY(merchant_category_code, amount) AS highest_value_mcc,
      100 AS final_risk_score,
      STRUCT(
        APPROX_COUNT_DISTINCT(location_country) > 1 AS is_impossible_travel,
        LOGICAL_OR(NOT is_trusted_device) AS has_security_mismatch
      ) AS logic_signals
    )) AS data
  FROM TUMBLE(TABLE TransactionHeuristics, "bq_changed_ts", INTERVAL 2 MINUTE)
  GROUP BY window_start, window_end, user_id
  HAVING APPROX_COUNT_DISTINCT(location_country) > 1
);

第 2 部分：Pub/Sub 与单消息转换（SMT）

用 Pub/Sub 弥合模式差距。 连续查询导出的事件数据被直接发送到一个 Pub/Sub 主题。在这些原始数据能被人工智能智能体消费之前，载荷需要被转换成智能体所期望的模式。

不必部署专门的 Cloud Function 之类的东西来重新格式化这些消息，你可以完全在 Pub/Sub 订阅内使用单消息转换（Single Message Transform，SMT）处理。SMT 允许你在 Pub/Sub 内直接运行轻量的内联 JavaScript 用户定义函数（User-Defined Functions，UDF），即时映射、重塑或清理载荷。

例如，你可以定义一个带 JavaScript 片段的 transform.yaml，拦截 BigQuery 载荷并提取智能体引擎所期望的精确 query 格式：

function process(res) {
  let bq_payload = JSON.parse(res.message.data);
  res.message.data = JSON.stringify({"query": bq_payload});
  return res;
}

要配置路由流水线，创建一个 Pub/Sub 推送订阅（Push Subscription）。该订阅把每一条已转换的 BigQuery 事件自动推送到人工智能智能体的 webhook 终结点：

gcloud pubsub subscriptions create cymbal-bank-escalations-sub \
  --topic=projects/$PROJECT_ID/topics/cymbal-bank-escalations-topic \
  --message-transforms-file=setup/transform.yaml \
  --push-endpoint="https://YOUR_AGENT_WEBHOOK_URL" \
  --push-no-wrapper \
  --ack-deadline=600 \
 --push-auth-service-account="adk-agent-sa@$PROJECT_ID.iam.gserviceaccount.com"

注意上面的 push-endpoint 参数。这个 webhook URL 由最后一个架构部件生成：人工智能智能体本身。

第 3 部分：ADK 与 Vertex AI 智能体引擎

当智能体部署到 Vertex AI 智能体引擎 时，平台会自动配置一个专门用于接收这些传入事件的安全 streamQuery 终结点。

这是整个操作的大脑。 一旦检测到异常并经 Pub/Sub 路由，该消息就会触发部署在 Vertex AI 上的 ADK 智能体。

要实现推理循环， 你定义配备工具的智能体并部署它：

investigation_agent = Agent(
    model="gemini-2.5-flash",
    name="fraud_investigation_agent",
    description="Expert fraud analyst agent that autonomously investigates alerts...",
    instruction=(
        "You are an expert fraud investigator for Cymbal Bank. "
        "Your goal is to investigate financial transaction alerts, "
        "determine if they are fraudulent, and take appropriate action. "
        "Use the BigQuery toolset to analyze data in the transactions table.."
        "Use the Google Search toolset to search for the merchant..."
        "Conslidate your findings and use the escalate_to_human tool if required..."
    ),
    tools=[\
        bigquery_toolset,\
        google_search,\
    ],
)

配备了具体指令和这一自定义工具集之后，智能体通过主动收集外部上下文来自主调查警报。它可以查询 BigQuery 以获取用户的交易历史，分析收据等非结构化数据，或用 Google Search 为发现提供依据，以核验商户信誉。最终，它把交易归类为 FALSE_POSITIVE，或将其标记为 ESCALATION_NEEDED。

人在回路的优势

这种方法是该架构可扩展性的核心。通过有效滤除噪声，它大幅降低运营开销，并确保调查人员只把时间花在最复杂的案件上。由于 ADK 提供一系列工具与集成，你可以让智能体把事件升级到多种企业系统，用于人在回路的参与，或使用人在环上（human-on-the-loop）的可观测性把流水线端到端自动化。

汇总：智能体分析

流水线上线之后，工作从构建转向监控。与传统软件不同，自主智能体在后台持续运行。因为它们在幕后运行，深入观察它们在做什么、花了多长时间、成本是多少，就很关键。

在部署期间初始化 BigQuery 智能体分析插件（BigQuery Agent Analytics plugin） 后，ADK 会自动把全部追踪数据、工具使用和执行延迟直接记入 BigQuery：

把这些追踪数据与智能体输出的结构化决定连接起来，你就能得到丰富的分析。这使你能够构建动态仪表盘，并设置自定义警报，实时监控人工智能劳动力。你可以查看这个 Codelab，进一步了解智能体分析插件的使用。

结论

实时数据流与智能体式人工智能（Agentic AI）的汇合，正在改变我们处理运营警报的方式。

1. 用 BigQuery 连续查询实时检测。

2. 用 Pub/Sub SMT 转换并路由。

3. 用 Vertex AI 智能体引擎调查并处置。

4. 用 BigQuery 智能体分析插件分析。

这一架构使你能够构建一支主动、自主的劳动力，在异常发生的当下处理它们，并且全部位于受治理、可扩展、无服务器的 Google Cloud 环境中。

准备动手？

查看我们的 Codelab，获取从零构建这条 Cymbal Bank 流水线的分步指南。

PRIMARY_SOURCE: https://cloud.google.com/blog/topics/developers-practitioners/building-event-driven-data-agents-with-bigquery-pubsub-and-adk