Grab 正在把 AI Agent 引入分析工作流,并将机械化分析工作占比从 2 月的 44% 降低到 6 月的 30%。这不是简单地把聊天机器人接到数据仓库,而是把 Agent 自主执行能力、认证数据、上下文管理和人工监督组合成一套可控的自助分析系统。
从“帮分析师写 SQL”转向“完成分析请求”
传统的 AI 数据助手通常停留在生成 SQL:用户提问,模型给出一段查询语句,分析师还要检查指标定义、确认数据表、运行查询并解释结果。
Grab 的方向更接近完整工作流自动化。自助分析能力逐渐覆盖三类请求:
- 指标请求:查询订单量、收入、转化率等已经定义好的业务指标。
- 数据请求:从认证数据集中筛选、聚合和比较数据。
- SQL 请求:根据自然语言生成或执行查询,并返回结果。
这意味着 Agent 的产出不应只是 SQL 文本,而应包括指标口径、数据来源、查询结果、必要的假设,以及无法确定时交给人的原因。机械性工作减少后,分析师可以把精力放在问题定义、因果分析和业务决策上。
四个决定自动化上限的组件
1. Agent 自主性
Agent 需要能够拆分任务、选择工具、执行查询并根据结果继续行动。例如,一次“比较两个市场本季度的活跃用户和订单转化率”的请求,可能需要依次完成:识别指标、定位认证数据集、生成查询、验证时间范围、执行查询和组织答案。
但自主性不能等同于无限权限。生产系统应为 Agent 设置工具白名单、查询超时、扫描数据量限制和只读凭证。对删除、写入或改变生产数据的操作,应明确禁止或要求人工批准。
2. 认证数据
自然语言转 SQL 最容易出错的地方,往往不是 SQL 语法,而是指标含义。例如,“活跃用户”可能按日去重,也可能按月去重;“收入”可能包含退款,也可能只统计已完成订单。
认证数据集和指标目录可以把这些定义固定下来。Agent 应优先使用经过审核的指标和数据资产,而不是在任意表之间自由猜测。回答中也应该显示使用了哪些指标定义和数据集,让用户能够追溯结果。
3. 上下文管理
分析问题很少只有一句话。用户可能先问“新加坡的订单量是多少”,接着问“和上个月相比呢”,再追问“排除取消订单后如何”。如果 Agent 无法维护时间范围、市场、过滤条件和指标粒度,结果就会在多轮对话中悄悄漂移。
可以把上下文拆成结构化状态,而不是只依赖聊天记录:
{
"market": "Singapore",
"metric": "completed_orders",
"time_range": {
"start": "2025-01-01",
"end": "2025-01-31"
},
"comparison": "previous_month",
"filters": {
"exclude_cancelled": true
}
}
这样做便于在每次查询前校验参数,也方便记录审计日志。实际字段名称需要根据企业的数据目录和指标系统调整。
4. 人工监督
自动化分析并不意味着完全移除分析师。人工监督应集中在高风险和高歧义请求上,而不是重新检查每一个简单查询。
一种可行的分级方式是:
- 认证指标、固定过滤条件、只读查询:Agent 可以直接返回。
- 使用非认证数据、跨域连接或复杂业务规则:要求分析师复核。
- 涉及敏感数据、对外发布或业务决策的结果:必须人工批准。
这种设计把人工时间用在判断和责任边界上,同时让低风险请求真正实现自助化。
一个可改造的最小实现
下面是一个简化的 Python 示例,展示如何在执行 SQL 前检查指标是否认证、是否使用只读查询,以及是否超过扫描限制。它不是 Grab 的内部实现,而是一种可以在企业数据平台中实践的最小模式。
运行前请将 run_query 替换为你的数据仓库客户端,并把示例目录换成实际的指标服务。
from dataclasses import dataclass
import re
CERTIFIED_METRICS = {
"completed_orders": {
"sql": "COUNT(DISTINCT CASE WHEN status = 'completed' THEN order_id END)",
"table": "analytics.orders",
},
"gross_booking_value": {
"sql": "SUM(CASE WHEN status = 'completed' THEN amount ELSE 0 END)",
"table": "analytics.orders",
},
}
@dataclass
class QueryRequest:
metric: str
market: str
start_date: str
end_date: str
def build_query(request: QueryRequest) -> str:
definition = CERTIFIED_METRICS.get(request.metric)
if definition is None:
raise ValueError("Metric is not certified")
if request.start_date >= request.end_date:
raise ValueError("Invalid time range")
return f"""
SELECT
market,
{definition['sql']} AS value
FROM {definition['table']}
WHERE market = '{request.market}'
AND event_date >= '{request.start_date}'
AND event_date < '{request.end_date}'
GROUP BY market
""".strip()
def validate_read_only(sql: str) -> None:
forbidden = re.compile(r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|MERGE)\b", re.I)
if forbidden.search(sql):
raise ValueError("Only read-only SQL is allowed")
def run_query(sql: str):
# Replace this with a read-only warehouse client.
print(sql)
return {"market": "Singapore", "value": 12345}
request = QueryRequest(
metric="completed_orders",
market="Singapore",
start_date="2025-01-01",
end_date="2025-02-01",
)
query = build_query(request)
validate_read_only(query)
print(run_query(query))
这个例子体现了三个关键边界:指标必须来自认证目录,查询参数必须经过校验,执行权限必须限制为只读。生产环境还应使用参数化查询,避免把用户输入直接拼接进 SQL;同时增加数据权限检查、敏感字段脱敏、查询成本控制和完整审计记录。
如何衡量 Agent 是否真的有效
机械工作占比从 44% 降至 30% 是一个重要结果,但单看自动化比例还不够。落地时可以同时观察:
- 自助分析请求的完成率和失败率。
- 从提问到可信结果的平均耗时。
- Agent 生成结果被分析师修改或驳回的比例。
- 使用认证指标的占比。
- 查询成本、扫描数据量和权限违规次数。
- 分析师把时间投入到解释、建模和决策支持的比例。
如果自动化率上升,但返工率、错误指标和数据仓库成本同步上升,系统只是把机械工作转移成了质量控制工作。真正健康的指标体系应同时覆盖效率、正确性、可追溯性和风险。
采用建议:先自动化确定性问题
企业可以从低风险、高重复的请求开始,例如认证指标查询、固定时间范围对比和只读聚合分析。随着评估数据集、反馈机制和权限控制成熟,再逐步扩大 Agent 的工具范围。
适合落地的检查清单如下:
- 为核心指标建立唯一、可审计的定义。
- 给 Agent 提供认证数据集和明确的工具接口。
- 把多轮上下文保存为结构化状态。
- 使用只读凭证、查询限额和超时保护。
- 按风险等级决定自动返回还是人工复核。
- 记录提示词、指标版本、生成 SQL、查询结果和审批记录。
- 用真实业务请求持续评估准确率与返工率。
Grab 的实践说明,分析自动化的关键不只是让模型“会写 SQL”,而是让它在正确的数据、明确的上下文和可控的权限范围内完成任务。Agent 负责处理重复路径,人负责定义问题、审查边界并承担关键判断,这种分工更有可能把 AI 从演示工具变成稳定的分析基础设施。