企业级对话分析落地：从自然语言查询到可治理的数据智能体

预计阅读时间：10 分钟

企业部署生成式 AI，真正困难的部分并不是给聊天窗口接上一个大模型，而是让自然语言查询继承现有数据平台的语义、安全边界和运维能力。面对营收、供应链或客户数据，一条看似合理但口径错误的 SQL，就足以让整个系统失去信任。

Google Cloud 的对话分析能力正在从单点试验走向统一的数据入口：BigQuery 和 Looker 中的 Conversational Analytics 已正式可用，Conversational Analytics API 也已正式可用；AlloyDB、Cloud SQL 和 Spanner 的相关能力处于预览阶段。更重要的是，这套能力开始覆盖多云数据、企业语义层、成本治理、可观测性以及主动分析工作流。

覆盖更多数据，不等于放松数据边界

对话分析现在可以面向 BigQuery、Looker，以及预览阶段的 AlloyDB、Cloud SQL 和 Spanner。它还可以分析 Lakehouse Managed Service 表、Apache Iceberg REST Catalog 和联邦 AWS S3 Unity Catalog 中的数据。这意味着企业不必先把全部数据复制到单一存储，再让智能体开始工作。

使用入口也不局限于一个聊天机器人。数据人员可以在 BigQuery Studio、BigQuery Data Canvas 和 Database Studio 中使用相关能力；业务用户则可以从 Looker、Data Studio 或 Gemini Enterprise 发起问题。通过 API、SDK 和 MCP 工具，团队还可以把数据智能体嵌入内部应用、Slack 机器人或多智能体系统。

但“能够查询”与“应该允许查询”必须分开设计。一个面向全公司的智能体至少需要处理四层边界：

身份边界：当前用户是谁，继承了哪些角色。
数据边界：允许访问哪些项目、数据集、表、行和列。
计算边界：单次查询最多扫描多少字节，是否允许执行高成本操作。
发布边界：哪些智能体可以进入 Gemini Enterprise 或其他共享入口。

Google Cloud 提供 CMEK、Private IP、VPC、数据驻留和基于角色的访问控制。数据库侧还可以借助安全视图和参数化视图，把权限落实到行列级别。这里的关键原则是：权限必须由数据平台执行，不能只写在提示词里。

语义层决定回答是否可信

通用大模型可以生成语法正确的 SQL，却未必理解企业内部的“活跃客户”“净收入”或“准时交付率”。这些指标通常包含时间窗口、退款处理、测试账号排除和组织权限等复杂规则。如果智能体临时猜测表连接和过滤条件，相同问题可能得到不同答案。

对话分析因此需要与数据模型共同设计。Knowledge Catalog 可以为数据发现、术语表、表说明和连接关系提供上下文；BigQuery Graphs 和 Spanner Graphs 可以表达跨多跳关系；Looker 的 LookML 语义层则能让智能体使用集中治理的指标定义。

“Golden Queries”也值得纳入工程流程。它们不是更多的提示词，而是一组经过业务和数据团队验证的查询逻辑。智能体遇到关键指标时，应优先复用这些逻辑，而不是重新猜测 SQL。例如：

# golden-queries.yaml
queries:
  - id: net_revenue_by_region
    description: 按区域统计已确认净收入
    owners:
      - finance-analytics
    allowed_roles:
      - finance_viewer
      - finance_analyst
    parameters:
      - start_date
      - end_date
      - region
    metric_definition: confirmed_sales - refunds - credits
    freshness_slo_minutes: 60

  - id: delayed_shipments
    description: 查找超过承诺交付日期的未完成订单
    owners:
      - supply-chain-data
    allowed_roles:
      - operations_viewer
    parameters:
      - warehouse_id
      - as_of_date
    freshness_slo_minutes: 15

这不是 Google Cloud API 的固定配置格式，而是一种可以直接改造的内部登记方式。生产系统可以把它同步到语义层、目录系统或智能体工具注册表，并在发布前要求负责人审批。

在调用 API 前增加策略闸门

即使底层平台具备权限控制，应用侧仍应限制数据源、查询规模和请求超时。下面是一个可运行的 Python 示例：默认使用模拟响应演示策略检查；接入真实 Conversational Analytics API 时，只需要替换 call_provider，并按照实际 SDK 或 API 文档构造请求。

运行前安装依赖：

python -m pip install requests pyyaml

创建 analytics_agent.py：

import json
import os
from dataclasses import dataclass
from typing import Any

import requests


@dataclass(frozen=True)
class Policy:
    allowed_sources: frozenset[str]
    max_bytes_processed: int
    timeout_seconds: int = 30


POLICY = Policy(
    allowed_sources=frozenset({"finance_curated", "supply_chain_curated"}),
    max_bytes_processed=10 * 1024**3,  # 10 GiB
)


def call_provider(question: str, source: str) -> dict[str, Any]:
    endpoint = os.getenv("CA_API_ENDPOINT")
    token = os.getenv("CA_ACCESS_TOKEN")

    if not endpoint:
        return {
            "answer": "模拟结果：西区净收入环比下降 4.2%。",
            "source": source,
            "estimated_bytes_processed": 2 * 1024**3,
            "trace_id": "demo-trace-001",
        }

    response = requests.post(
        endpoint,
        headers={"Authorization": f"Bearer {token}"},
        json={"question": question, "source": source},
        timeout=POLICY.timeout_seconds,
    )
    response.raise_for_status()
    return response.json()


def ask(question: str, source: str) -> dict[str, Any]:
    if source not in POLICY.allowed_sources:
        raise PermissionError(f"Data source is not allowed: {source}")

    result = call_provider(question, source)
    scanned = int(result.get("estimated_bytes_processed", 0))
    if scanned > POLICY.max_bytes_processed:
        raise RuntimeError(
            f"Query exceeds byte limit: {scanned} > {POLICY.max_bytes_processed}"
        )
    return result


if __name__ == "__main__":
    output = ask(
        "上个月各区域净收入发生了什么变化？",
        "finance_curated",
    )
    print(json.dumps(output, ensure_ascii=False, indent=2))

执行：

python analytics_agent.py

示例中的请求字段和响应字段是应用层假设，不代表正式 API 契约。接入时应使用官方 SDK 支持的认证与请求模型，并从服务端查询作业或追踪数据中读取实际扫描量，不能信任客户端自行上报的成本数字。

从问答升级为主动分析

对话分析的下一阶段并不是让用户提出更多问题，而是在指标变化时自动启动调查。预览阶段的 Agentic Workflows 可以安排日报、周报和多维深挖；流式异常检测还可以在指标偏离基线时触发智能体。

一个供应链工作流可以按以下顺序执行：

检测准时交付率是否超出正常波动区间。
分析仓库、承运商、商品类别和目的地区域等 10 到 20 个因素。
调用财务数据智能体，估算延迟对毛利和退款的影响。
将证据、指标口径和追踪标识发送到团队聊天频道。
由负责人确认后，再触发业务动作。

这里必须区分“自动分析”和“自动决策”。前者可以扩大覆盖面，后者可能影响库存、价格或客户权益，应增加人工审批、幂等控制和回滚机制。

上线前检查什么

企业级对话分析应先从一个指标稳定、权限明确、查询量可控的业务域开始，而不是立即开放整个数据仓库。上线前至少确认以下事项：

核心指标是否由 LookML、验证查询或其他语义资产统一定义。
行列级权限是否由数据库或数据仓库强制执行。
是否设置单次查询字节上限、并发限制和预算告警。
是否记录查询标签、用户反馈、延迟、令牌消耗和工具调用。
是否能通过 OpenTelemetry 汇总智能体健康状态和调用链。
高风险回答是否展示数据来源、口径、时间范围和追踪标识。
预览功能是否满足组织对稳定性、区域和合规性的要求。

真正可扩展的对话分析，不是把数据库包装成聊天界面，而是把数据目录、语义层、权限系统、成本控制和评估闭环组合成一个可信入口。自然语言只是交互方式，企业已有的数据治理能力才是系统能够进入生产环境的基础。