AI Agent 语义理解与数据查询：从识别业务意图到生成可信 SQL

让 AI Agent 回答业务数据问题，难点并不只是把一句自然语言改写成 SQL。用户表达中的指标、区域和时间范围往往带有歧义，而数据库要求每个字段、口径与连接条件都精确无误。要让查询结果真正可信，需要把语义对齐、上下文管理、查询生成、安全检查和结果校验串成一条可观测的工程链路。

AI Agent 语义分析与数据引擎查询：从"听懂人话"到"查对数据"的技术拆解

本文适合关注 AI Agent、数据工程、NL2SQL 方向的开发者阅读，全文约 4000 字。

---

一、问题的起点：为什么 Agent 查数据这么难？

过去一年，"AI Agent + 数据分析" 是最热闹的落地场景之一。理想很丰满：用户对着对话框说 "帮我看看上季度华东区新客的复购率，对比去年同期" ，Agent 应该自动完成：

1. 理解业务语义（"新客"如何定义？华东区含哪些省份？"复购率"按 30 天还是 90 天窗口？）
2. 找到对应的数据表与字段（dwd_user_first_order / dws_trade_daily ...）
3. 生成可执行且正确的 SQL
4. 查询、校验、并用人话回答

但现实往往是：Agent 生成的 SQL 跑通了，结果却是错的——因为它悄悄用错了表、漏了 join 条件、或者把时间窗口理解成了自然月。

核心矛盾在于：语言是模糊的，数据是精确的。 语义分析层就是解决这个矛盾的第一道关卡。

---

二、整体架构：一个典型的"语义分析 → 数据引擎查询"流水线

plain

复制

用户提问
   │
   ▼
┌─────────────────────────────────────┐
│ 1. 意图理解 (Intent Recognition)      │
│    - 任务分类：查询 / 趋势 / 对比 / 归因  │
│    - 实体抽取：指标、维度、时间范围       │
└─────────────────────────────────────┘
   │
   ▼
┌─────────────────────────────────────┐
│ 2. 语义对齐 (Semantic Mapping)        │
│    - 业务术语 → 标准指标定义            │
│    - 维度 → 可用字段映射               │
│    - 歧义消解 / 澄清追问               │
└─────────────────────────────────────┘
   │
   ▼
┌─────────────────────────────────────┐
│ 3. 上下文管理 (Context Management)    │
│    - 多轮对话状态                     │
│    - 指代消解（"那再按渠道拆开看"）      │
└─────────────────────────────────────┘
   │
   ▼
┌─────────────────────────────────────┐
│ 4. 查询生成 (Query Generation)        │
│    - Text2SQL / DSL 生成              │
│    - 基于元数据的约束生成              │
└─────────────────────────────────────┘
   │
   ▼
┌─────────────────────────────────────┐
│ 5. 执行与校验 (Execution & Guarding)  │
│    - SQL 安全检查                     │
│    - 结果合理性校验                   │
│    - 失败自愈（重试 / 改写）            │
└─────────────────────────────────────┘
   │
   ▼
自然语言回答 + 图表 + 可追溯的 SQL

每一层都可能出错，下面我们逐层拆解关键技术。

---

三、语义分析层：让 Agent "听懂业务"

3.1 意图识别：先判断"这活儿该怎么干"

用户的同一句话可能对应完全不同的任务：

"看一下上个月的销售额"

可能是：

• 取数：返回一个数字
• 归因分析：销售额为什么涨了/跌了？
• 趋势查看：按月画出折线图

实践中常用两阶段方法：

Python

复制

# 阶段1：任务分类（可以用小模型 + 指令微调，成本低、延迟小）
task = classify_intent(
    user_query="看一下上个月的销售额",
    schema={
        "fetch":    "取数/查询类",
        "analyze":  "归因/对比分析类",
        "trend":    "时间序列趋势类",
    }
)
# → task = "fetch"

关键经验：意图分类不要交给大模型自由生成，用受约束的分类（function calling / 结构化输出），准确率提升非常明显。

3.2 实体抽取：把句子拆成"指标 + 维度 + 时间"

这是语义分析的核心。一个查询可以被建模为：

plain

复制

Query = f(指标, 维度列表, 过滤条件, 时间范围)

例如：

"帮我看看上季度华东区新客的复购率，对比去年同期"

JSON

复制

{
  "metrics": [
    {"name": "复购率", "window": "90天", "definition": "二次购买用户数 / 首购用户数"}
  ],
  "dimensions": [
    {"name": "华东区", "mapping": "region IN ('上海','江苏','浙江','安徽','福建','江西','山东')"}
  ],
  "filters": [
    {"field": "is_new_customer", "op": "=", "value": true}
  ],
  "time_range": {
    "relative": "上季度",
    "resolve": {"from": "2026-04-01", "to": "2026-06-30"}
  },
  "compare": {"type": "同比", "offset": "-1年"}
}

LLM + 约束解码（Structured Output） 是目前的主流做法。注意两点：

1. 时间词必须归一化："上个月"这种相对时间，要在执行时基于当前日期解析，而不是让模型"猜"日期。
2. 指标必须挂接语义定义："复购率"在不同公司定义完全不同，抽取出名字后必须映射到指标平台的准确定义。

3.3 业务语义层（Semantic Layer）：最不能省的一层

这是从 demo 走向生产的关键。做法是把公司口径沉淀成一个指标语义层，Agent 只跟语义层打交道，不直接碰表：

plain

复制

用户说的"GMV"
     │
     ▼
指标平台定义：GMV = SUM(pay_amount) - SUM(refund_amount)
                过滤：order_status = '已完成' AND is_test = 0
                时间口径：支付时间
                可用维度：date / region / channel / category

落到工程上，就是给 LLM 的上下文里注入一份受控元数据卡片：

JSON

复制

{
  "metric": "gmv",
  "sql_template": "SUM(pay_amount) - SUM(refund_amount)",
  "filters": ["order_status = '已完成'", "is_test = 0"],
  "time_field": "pay_time",
  "dimensions": ["date", "region", "channel", "category"]
}

这样 Text2SQL 就从"让模型自由写 SQL"变成"让模型在模板约束下拼装 SQL"，正确率有数量级提升。

---

四、上下文管理：多轮对话的暗坑

单轮查询好做，多轮对话才是真正的产品形态：

用户："上季度华东的销售额" Agent：【给出结果】 用户： "那华南呢？" （指代 + 省略） 用户： "这个结果按渠道拆一下" （在上一轮基础上追加维度）

常见方案：

1. 对话状态结构化：每轮结束后维护一个 QueryContext 对象（指标、维度、时间、过滤都在里面），下一轮在其上做增量修改，而不是每次从头理解。
2. 历史 SQL 感知：把上一轮生成的 SQL 也放进上下文，让模型做"改写"而非"重写"。

Python

复制

context = {
    "metrics": ["gmv"],
    "dimensions": ["region"],
    "filters": [{"field": "quarter", "value": "2026Q2"}],
    "last_sql": "SELECT region, SUM(...) ..."
}

# 用户说"那华南呢" → diff 操作：dimensions.region = '华南'

---

五、查询生成与执行：Text2SQL 的正确打开方式

5.1 生成策略的选择

表格

复制

方案	适用场景	特点
纯 LLM Text2SQL	表少、schema 简单	灵活但幻觉多
LLM + 语义层模板	指标口径固定的 BI 场景	生产首选，正确率高
LLM + RAG（检索相似 SQL 示例）	表多、join 复杂	用历史好 SQL 做 few-shot
LLM + AST 约束生成（如木兰/Mozi 类思路）	强安全要求	生成中间 DSL 再编译成 SQL

实际工程中通常是 "语义层模板为主 + RAG 相似样例为辅" 的混合策略。

5.2 上下文注入：该给模型看什么

把几百张表的 schema 全塞给模型既贵又糟。正确姿势是先检索、再生成：

1. 用用户问题 + 抽取出的实体，在向量库中检索相关表/字段/指标（Top-K）
2. 只把相关的 DDL、字段注释、指标定义注入 prompt
3. 生成 SQL

5.3 执行前的 Guardrail

生成完的 SQL 绝不能直接跑生产库：

Python

复制

def guard(sql: str) -> bool:
    return all([
        check_readonly(sql),        # 只允许 SELECT / WITH
        check_table_whitelist(sql), # 只能访问白名单表
        check_row_limit(sql),       # 强制 LIMIT，防全表扫描
        check_cost_estimate(sql),   # 预估扫描量超限则拦截
        check_syntax_by_explain(sql) # EXPLAIN 预检
    ])

5.4 执行后的校验与自愈

结果校验同样重要，常见检查：

• 空结果告警：查询结果为空时，先检查是不是 join 条件写错，而不是直接告诉用户"没有数据"
• 量级合理性：日活通常不会突然变成 3 亿，用历史分布做 sanity check
• 失败自愈：SQL 报错后，把错误信息回灌给模型重写，重试 1~2 次

Python

复制

for attempt in range(2):
    try:
        result = engine.execute(sql)
        if sanity_check(result):
            return result
    except Exception as e:
        sql = rewrite_with_error_feedback(sql, str(e))
raise QueryFailedError()

---

六、结果生成：最后一公里的体验

查到数据只是完成了一半，"怎么回答"决定产品体验：

1. 结论先行：先一句话回答，再给数据和 SQL。"上季度华东复购率为 18.2%，同比下降 2.1 个百分点。"
2. 自动选图：单值 → 指标卡；时间序列 → 折线；维度对比 → 柱状/条形，规则 + LLM 判断结合。
3. 可追溯：把最终 SQL、命中的指标定义、数据来源表都展示出来，让用户能审计——这是建立信任的关键，也是很多 Agent 产品忽略的。

---

七、一个简化版的伪代码串起来

Python

复制

def agent_pipeline(user_query: str, session_id: str):
    ctx = load_context(session_id)                    # 多轮上下文

    intent  = classify_intent(user_query)              # 1. 意图
    s   = extract_s(user_query, intent)        # 2. 实体抽取
    s   = disambiguate(s, ctx)                 #    歧义消解/澄清追问
    ctx     = merge_context(ctx, s)                # 3. 上下文合并

    meta    = retrieve_metadata(ctx)                   #    RAG 检索相关表/指标
    plan    = build_query_plan(ctx, meta)              #    语义层对齐成查询计划
    sql     = generate_sql(plan, meta, ctx)            # 4. 查询生成

    if not guard(sql):
        return ask_clarification("查询有风险，请确认...")

    for _ in range(2):
        try:
            result = engine.execute(sql)
            if sanity_check(result, ctx):
                break
            sql = rewrite(sql, "结果异常")
        except Exception as e:
            sql = rewrite(sql, str(e))
    else:
        return fallback_response()

    answer = compose_answer(result, ctx, sql)          # 5. 生成回答 + 图表 + SQL 溯源
    save_context(session_id, ctx, sql)
    return answer

---

八、落地过程中的几个关键教训

1. 不要指望一个 Prompt 解决所有问题。把语义分析、查询生成、校验拆开，每一层都可以独立优化和评测。
2. 评测集是生命线。沉淀一批真实业务问题 + 标准 SQL（以及标准答案数据），任何 prompt / 模型 / 元数据改动都跑一遍回归。
3. 能模板化就不要自由生成。业务指标是收敛的，语义层模板能覆盖 80% 的场景；剩下 20% 的 ad-hoc 查询才交给自由 Text2SQL + 强校验。
4. 澄清优于猜测。置信度低时主动追问"你说的复购率是指 90 天内再次下单吗？"，一次追问的成本远低于一次错误答案的代价。
5. 链路全程可观测。记录每一层的输入输出，线上 case 可以快速定位是语义抽错了、元数据没检索到、还是 SQL 生成失误。

---

九、小结

"AI Agent 语义分析 + 数据引擎查询"本质上是一个 「模糊语言 → 精确语义 → 受控执行 → 可信回答」 的翻译系统。LLM 负责理解与生成，但决定系统上限的，是语义层建设、元数据质量、校验与自愈的工程细节。

如果说 LLM 是大脑，那么语义层是字典，数据引擎是双手，而 Guardrail 则是安全带——缺一不可。