数据分析 Copilot：自然语言驱动的 BI 探索与状态管理

在很多团队里，数据分析的日常沟通往往是这样开始的：业务同学跑来问“上个月华东区复购率怎么掉了”，分析师转头打开数仓写 SQL，拉出数据按品类、城市一层层拆解，最后再把图表贴回群里。这套流程不仅响应慢，而且绝大多数时间都耗在重复的“取数-切片-再取数”上。

为了让业务能自助探索，业界做过很多尝试。从最早的拖拽式仪表盘，到后来的 Text2SQL，大家都希望业务自己动动手就能拿到答案。但现实很骨感：拖拽面板对不懂数据模型的业务来说依然有门槛；而直接把 DDL 丢给大模型的 Text2SQL，在复杂多轮分析中往往几句话就跑偏了。

真正好用的数据分析 Copilot，核心并不是写单句 SQL，而是在多轮对话中精确维持分析状态与业务口径。

---

一、为什么直接 Text2SQL 搞不定多轮 BI 分析？

在实际业务场景中，几乎没有人会“一问即止”。数据探索是一个典型的“假设-验证-细分-归因”链路。

比如业务的提问过程通常是连贯的：

1. “看下今年 7 月华东区的 GMV 和下单人数。”
2. “把女装品类单独拉出来，按周看趋势。”
3. “把退款率加上，按城市拆开排个名。”
4. “那看下北京的数据呢？”

如果直接依赖大模型生成原始 SQL，会在第二轮和第四轮迅速崩塌：

• 口径黑盒与指标漂移：大模型不知道数仓里的 GMV 到底是 pay_amount 还是 order_amount - refund_amount。上一轮用了支付表，下一轮可能就被模型换成了订单快照表，前后口径根本对不上。
• 上下文继承与覆盖混乱：当用户问“那看下北京的数据呢”，系统需要把过滤条件中的“华东区”替换为“北京/华北区”，同时保留“女装品类”、“今年 7 月”、“退款率”等已有上下文。单纯靠 Prompt 拼接历史对话，大模型极容易丢掉时间窗口或把旧过滤条件死板地拼在 WHERE 后面，产生互斥条件（如 region = '华东' AND city = '北京'）。
• 权限与数据越权风险：自由生成的 SQL 无法严格受控，业务只要多问一句“看看友商团队的转化数据”，大模型就可能生成绕过行级权限（Row-Level Security）的查询。

因此，生产环境的方案必须是 LLM + 语义层（Semantic Layer / Metrics Store）：LLM 负责把自然语言解析成结构化的查询意图（DSL），状态机负责维护上下文槽位，最后由底层的语义编译器组装成带权限约束的确定性 SQL。

+-------------------------------------------------------------------+
|                        用户自然语言输入                           |
|      "上月华东区女装退款率是多少？按城市拆开看下排名"             |
+-------------------------------------------------------------------+
                                  |
                                  v
+-------------------------------------------------------------------+
|                   1. 意图解析 (LLM / NLU)                         |
|   - 抽取指标: [refund_rate]                                       |
|   - 抽取维度: [city]                                              |
|   - 抽取过滤: [dt='2026-07', category='女装', region='华东']      |
|   - 分析动作: DRILL_DOWN (下钻) / RANK (排序)                     |
+-------------------------------------------------------------------+
                                  |
                                  v
+-------------------------------------------------------------------+
|               2. 多轮对话状态机 (Dialogue State)                  |
|   - 槽位继承: 继承上文的时间窗与过滤条件                          |
|   - 冲突覆盖: 新条件替换旧同类维度 (如区域切换)                   |
|   - 下钻栈维护: 当前聚合维度 [category, city]                     |
+-------------------------------------------------------------------+
                                  |
                                  v
+-------------------------------------------------------------------+
|               3. 语义层编译器 (Semantic Compiler)                |
|   - 指标公式映射: refund_rate -> SUM(refund_amt) / SUM(pay_amt)   |
|   - 关联路径推导: 自动 JOIN 维表 (dim_city, dim_category)        |
|   - 强制权限注入: WHERE tenant_id = 'xxx' AND org_id IN (...)      |
+-------------------------------------------------------------------+
                                  |
                                  v
+-------------------------------------------------------------------+
|                 4. 引擎执行与确定性口径回显                       |
|   返回数据表格/图表 + 回显: "已按[实付退款口径]聚合，时间: 2026-07"|
+-------------------------------------------------------------------+

---

二、对话状态的核心要素：槽位管理与生命周期

多轮探索的本质是状态的转移。一个健壮的 BI 对话状态机必须管理以下四类核心槽位（Slots）：

1. 时间窗口槽位（Time Window）

时间是所有 BI 分析的第一约束。状态机需要记录绝对时间（如 2026-07-01 ~ 2026-07-31）和时间聚合粒度（如 day、week、month）。

• 继承规则：后续追问默认沿用上一轮时间。
• 重置规则：一旦识别到新的显式时间表达式（如“看下今年 Q2”），立即全量替换时间范围，但保留当前分析的指标与维度。

2. 活跃指标槽位（Active Metrics）

记录当前分析关注的核心指标列表，每个指标必须在语义层中拥有唯一的标准定义。

• 追加与替换：如果用户说“把毛利率也加上”，执行指标追加；如果用户说“换成看退款金额”，则执行指标替换。

3. 维度与下钻栈（Drill-down Stack）

记录当前 GROUP BY 的维度列表。

• 下钻（Drill-down）：在原有维度基础上追加更细粒度维度（例如从 [region] 下钻到 [region, city]）。
• 上卷（Roll-up）：移除部分细粒度维度（例如从 [city] 回退到 [region]）。
• 切片切换（Slice）：把某一维度变成固定过滤条件，同时引入新维度。

4. 过滤条件字典（Filter Predicates）

记录 WHERE 条件的键值对。

• 同维覆盖：如果上一轮有 region = '华东'，本轮用户提问“那华南呢”，状态机应覆盖同维度过滤值，而不是拼接 region = '华东' AND region = '华南'。
• 跨维叠加：不同维度的过滤条件（如 category = '女装'）继续保留。

---

三、生产级对话状态机与查询编译器实现

下面我们用 Python 实现一套可运行的对话状态机与语义编译器。代码重点展示槽位合并、同维冲突覆盖、指标公式展开与数据权限下推。

"""
bi_copilot.py
对话式 BI 状态管理与语义编译核心引擎
"""

import copy
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Set


class ActionType(Enum):
    QUERY = "QUERY"             # 初始查询
    DRILL_DOWN = "DRILL_DOWN"   # 追加维度下钻
    ROLL_UP = "ROLL_UP"         # 上卷减少维度
    FILTER_CHANGE = "FILTER"    # 修改或增加过滤条件
    METRIC_CHANGE = "METRIC"    # 更换/追加指标
    RESET = "RESET"             # 重置分析状态


@dataclass
class ParsedIntent:
    """LLM 解析自然语言后输出的结构化意图"""
    action: ActionType
    metrics: List[str] = field(default_factory=list)
    dimensions: List[str] = field(default_factory=list)
    filters: Dict[str, Any] = field(default_factory=dict)
    time_window: Optional[str] = None
    time_grain: Optional[str] = None
    order_by: Optional[str] = None
    limit: int = 100


@dataclass
class MetricDefinition:
    """语义层指标定义"""
    name: str
    display_name: str
    formula: str             # 聚合公式，如 SUM(pay_amt)
    base_table: str
    required_joins: List[str] = field(default_factory=list)


@dataclass
class DialogueState:
    """多轮分析上下文状态"""
    session_id: str
    current_time_window: str = "LAST_30_DAYS"
    current_time_grain: str = "day"
    active_metrics: List[str] = field(default_factory=list)
    active_dimensions: List[str] = field(default_factory=list)
    active_filters: Dict[str, Any] = field(default_factory=dict)
    history_stack: List[Dict[str, Any]] = field(default_factory=list)

    def snapshot(self) -> Dict[str, Any]:
        """保存快照，用于回退分析"""
        return {
            "time_window": self.current_time_window,
            "time_grain": self.current_time_grain,
            "metrics": list(self.active_metrics),
            "dimensions": list(self.active_dimensions),
            "filters": copy.deepcopy(self.active_filters),
        }

    def rollback(self) -> bool:
        """回退到上一次探索状态"""
        if not self.history_stack:
            return False
        last_state = self.history_stack.pop()
        self.current_time_window = last_state["time_window"]
        self.current_time_grain = last_state["time_grain"]
        self.active_metrics = last_state["metrics"]
        self.active_dimensions = last_state["dimensions"]
        self.active_filters = last_state["filters"]
        return True


class StateManager:
    """负责上下文槽位合并与生命周期维护"""

    @staticmethod
    def apply_intent(state: DialogueState, intent: ParsedIntent) -> None:
        # 保存当前状态到历史栈
        state.history_stack.append(state.snapshot())

        # 1. 处理时间窗口
        if intent.time_window:
            state.current_time_window = intent.time_window
        if intent.time_grain:
            state.current_time_grain = intent.time_grain

        # 2. 处理指标 (若显式指定则根据动作追加或替换)
        if intent.metrics:
            if intent.action == ActionType.METRIC_CHANGE:
                # 显式追加或替换
                for m in intent.metrics:
                    if m not in state.active_metrics:
                        state.active_metrics.append(m)
            elif intent.action == ActionType.QUERY:
                # 全新提问，重置指标
                state.active_metrics = list(intent.metrics)

        # 3. 处理维度 (下钻 vs 上卷 vs 重置)
        if intent.action == ActionType.DRILL_DOWN:
            for dim in intent.dimensions:
                if dim not in state.active_dimensions:
                    state.active_dimensions.append(dim)
        elif intent.action == ActionType.ROLL_UP:
            for dim in intent.dimensions:
                if dim in state.active_dimensions:
                    state.active_dimensions.remove(dim)
        elif intent.action == ActionType.QUERY and intent.dimensions:
            state.active_dimensions = list(intent.dimensions)

        # 4. 处理过滤条件 (同字段覆盖，异字段累加)
        for key, val in intent.filters.items():
            state.active_filters[key] = val


class SemanticCompiler:
    """语义层 SQL 编译器：将状态转化为安全可执行的 SQL"""

    def __init__(self, metrics_catalog: Dict[str, MetricDefinition]):
        self.metrics_catalog = metrics_catalog

    def compile(self, state: DialogueState, user_tenant_id: str) -> str:
        if not state.active_metrics:
            raise ValueError("当前上下文没有选定任何分析指标，无法生成查询。")

        # 校验指标合法性
        select_exprs: List[str] = []
        tables_needed: Set[str] = set()

        for m_name in state.active_metrics:
            if m_name not in self.metrics_catalog:
                raise ValueError(f"未定义的业务指标: {m_name}")
            metric_def = self.metrics_catalog[m_name]
            select_exprs.append(f"{metric_def.formula} AS {m_name}")
            tables_needed.add(metric_def.base_table)
            for j in metric_def.required_joins:
                tables_needed.add(j)

        # 维度字段
        dim_cols = list(state.active_dimensions)
        select_clause = ", ".join(dim_cols + select_exprs) if dim_cols else ", ".join(select_exprs)

        # 主表（简化示意：取第一个指标的基表）
        base_table = list(tables_needed)[0]

        # 构造 WHERE 条件（强制注入租户隔离与时间分区）
        where_clauses = [
            f"tenant_id = '{user_tenant_id}'",
            f"dt = '{state.current_time_window}'"
        ]
        for f_key, f_val in state.active_filters.items():
            if isinstance(f_val, list):
                val_str = ", ".join(f"'{v}'" for v in f_val)
                where_clauses.append(f"{f_key} IN ({val_str})")
            else:
                where_clauses.append(f"{f_key} = '{f_val}'")

        where_stmt = " AND ".join(where_clauses)
        group_by_stmt = f" GROUP BY {', '.join(dim_cols)}" if dim_cols else ""

        sql = f"SELECT {select_clause} FROM {base_table} WHERE {where_stmt}{group_by_stmt}"
        return sql

多轮状态演化验证

下面我们用一个模拟的分析会话，展示状态机如何连续处理四轮追问并生成准确 SQL：

# 初始化指标字典（语义层元数据）
METRICS_CATALOG = {
    "gmv": MetricDefinition(
        name="gmv",
        display_name="成交总额",
        formula="SUM(pay_amount)",
        base_table="dws_trade_orders_daily"
    ),
    "refund_rate": MetricDefinition(
        name="refund_rate",
        display_name="退款率",
        formula="ROUND(SUM(refund_amount) * 1.0 / NULLIF(SUM(pay_amount), 0), 4)",
        base_table="dws_trade_orders_daily"
    ),
    "buyer_uv": MetricDefinition(
        name="buyer_uv",
        display_name="下单用户数",
        formula="COUNT(DISTINCT buyer_id)",
        base_table="dws_trade_orders_daily"
    ),
}

# 实例化状态机与编译器
state = DialogueState(session_id="session_1001")
compiler = SemanticCompiler(METRICS_CATALOG)
tenant_id = "org_cn_east"

print("=== 第一轮：业务提问 '看下今年7月华东区的GMV和下单人数' ===")
intent1 = ParsedIntent(
    action=ActionType.QUERY,
    metrics=["gmv", "buyer_uv"],
    dimensions=["region"],
    filters={"region": "华东"},
    time_window="2026-07"
)
StateManager.apply_intent(state, intent1)
print("生成的 SQL:\n", compiler.compile(state, tenant_id), "\n")

print("=== 第二轮：业务追问 '把女装品类加上，按城市拆开看' ===")
intent2 = ParsedIntent(
    action=ActionType.DRILL_DOWN,
    dimensions=["city"],
    filters={"category": "女装"}
)
StateManager.apply_intent(state, intent2)
print("生成的 SQL (继承了2026-07与华东区，并下钻了城市):\n", compiler.compile(state, tenant_id), "\n")

print("=== 第三轮：业务追问 '那华南呢？退款率顺便也看下' ===")
intent3 = ParsedIntent(
    action=ActionType.METRIC_CHANGE,
    metrics=["refund_rate"],
    filters={"region": "华南"}  # 覆盖上一轮的 region='华东'
)
StateManager.apply_intent(state, intent3)
print("生成的 SQL (同维覆盖为华南，追加退款率指标):\n", compiler.compile(state, tenant_id), "\n")

print("=== 第四轮：业务表示看错了，想要 '回退到上一轮' ===")
if state.rollback():
    print("回退成功！当前恢复的 SQL:\n", compiler.compile(state, tenant_id))

运行上述代码，每一轮生成的 SQL 都能严格对齐口径，并自动继承上下文与行级安全条件。

---

四、工程落地中的防御边界与避坑策略

在构建生产级 Copilot 时，有几个关键边界需要提前设防：

1. 遇到歧义主动反问，绝不擅自猜测

自然语言具有天然的模糊性。例如业务提问“看下高净值用户的流失情况”：

• 什么是“高净值”？是近一年消费满 5000 元，还是累计充值满 1 万元？
• 什么是“流失”？是连续 30 天未登录，还是未产生购买？

如果系统直接交给 LLM 脑补，得出的数据不仅毫无参考价值，还会误导运营决策。
正确的做法：系统应在语义层中维护预置标签（Tags）与规则，当检测到未对齐的高歧义词汇时，直接中断查询组装，返回单选/多选卡片：

“检测到您查询了【高净值用户】，请确认统计口径：
[A] 近 30 天累计实付 > 2000 元
[B] 会员等级为 V4 及以上”

2. 显式口径回显（Explainability First）

每次生成结果后，界面绝不能只扔出一个孤零零的数字或图表，必须伴随自然语言口径说明：

“已为您展示 2026-07 期间、华东大区、女装品类的【成交总额】与【退款率】。计算口径：实付金额总和，排除了测试与已取消订单。”

这种设计让业务能一眼核验系统是否理解正确，建立了人与系统之间的信任基础。

3. 主题突变（Topic Switch）检测

业务在分析过程中可能会突然跳出当前话题，例如上一句还在看订单退款，下一句突然问“对了，上周新注册了多少用户”。
如果状态机机械地把“新注册用户”指标塞进订单事实表的上下文里，就会生成错误的关联查询。

• 防御机制：当意图识别模块发现新提取的指标与当前活跃状态在数仓域中完全无交集（例如属于用户域 vs 交易域），应触发 Topic Switch 警告，自动归档上一轮探索上下文，开启全新的分析分支。

4. 大宽表查询限流与分区兜底

开放自由提问极易引发慢查询打爆数据库。例如业务问“看下所有用户的历史总购买频次”，如果不加时间分区限制，就会直接扫描全表几亿行。

• 编译器必须设置强制分区注入（如默认兜底近 90 天）。
• 设置 LIMIT 截断与单次查询计算资源上限，超时直接熔断并提示“查询涉及数据量过大，请增加维度筛选条件”。

---

五、总结

打造一个真正能进生产的自然语言数据分析 Copilot，技术重心从来不在 LLM 的 Prompt 调优上，而在于 语义层建模与状态机工程。

1. 解耦意图与编译：让大模型专注于自然语言理解与意图提取，由语义层编译器负责确定性的 SQL 生成与权限注入。
2. 状态精准管理：在多轮追问中，严密维护时间、指标、下钻维度与过滤条件的生命周期，做到可继承、可覆盖、可回退。
3. 把控业务边界：遇到歧义主动反问，生成结果显式回显口径，遇到性能与权限边界坚决拦截。

只有把数据口径的确定性握在系统手里，自然语言探索才能真正从“玩具”变成业务日常离不开的分析利器。