让 AI Agent 根据一句自然语言需求或者一个需求文档就输出准确的 SQL，听起来像是个模型能力问题，实际上是个工程体系问题。单靠换更强的模型、调更好的 prompt，效果很快就会碰到天花板。真正能让这套系统在生产环境跑起来的，是一套由"元数据 → 语义层 → 案例库 → 知识库 → Skill系统 → 校验体系"六个模块组成的工程架构。

这六个模块各自解决一个具体问题：元数据让 Agent 知道有哪些表和字段；语义层把业务指标封装成 Agent 能理解的定义；案例库提供从需求到 SQL 的推理样本；知识库补充业务背景和决策记录；Skill系统把这些能力编排成自动化流水线；校验体系则确保输出的 SQL 经得起审查。少了任何一个，系统都会在某个环节卡住或者出错。

下面逐一拆解每个模块的设计思路和落地细节。

整体架构概览

先看全局。整个系统可以分成三层：输入层接收需求，处理层完成推理，支撑层提供知识，校验层保障质量。

flowchart TB subgraph input["输入层"] A["需求文档 / 自然语言描述"] end subgraph core["处理层"] B["需求分析 Skill"] C["语义理解引擎"] D["SQL 生成引擎"] end subgraph foundation["基础支撑层"] E["元数据管理"] F["语义层"] G["案例库"] H["知识库"] end subgraph validation["校验层"] I["AI 自动评估"] J["人工核验"] K["持续迭代"] end A --> B B --> C C --> D E --> C F --> C G --> C H --> B D --> I I --> J J --> K K --> E K --> F K --> G

支撑层提供事实，处理层完成推理，校验层保障质量——三层之间通过标准化接口通信，每层都可以独立迭代。

元数据管理：让 Agent 看懂数据资产

元数据是 Agent 理解数据世界的入口。如果元数据质量差——表名是拼音缩写、字段没有中文注释、血缘关系缺失——那 Agent 即使推理能力再强，也只能在一堆 t_ord_dtl_001 这样的表名面前干瞪眼。

元数据的三层分类

企业里的元数据大致分三类，每类对 Agent 的作用不同：

类别 典型内容 Agent 拿它做什么
技术元数据 表名、列名、数据类型、主外键 知道有哪些表、表里有什么字段
业务元数据 中文含义、计算口径、数据 owner 理解字段背后的业务含义
操作元数据 血缘关系、更新频率、质量评分 判断该用哪张表、哪张表更靠谱

技术元数据通常能从数据库系统里自动采集，业务元数据则需要人工标注或者从文档中抽取，操作元数据往往需要在数据 pipeline 运行过程中积累。

采集与标注实践

以 OpenMetadata 为例，它提供了统一的元数据管理平台，支持 70 多种数据源连接器。采集配置本身不复杂：

source:
  type: mysql
  serviceName: "business_db"
  serviceConnection:
    config:
      username: "${DB_USER}"
      password: "${DB_PASSWORD}"
      hostAndPort: "db.example.com:3306"
  sourceConfig:
    includeViews: true
    includeTables: true
    enableDataProfiler: true

真正花时间的是业务语义标注。每个高频使用的表和字段，最好都有：

  • 中文名称：order_amount → "订单金额"
  • 业务定义：这个字段到底包不包含运费？扣不扣优惠券？
  • 计算公式：衍生字段是怎么算出来的
  • 使用示例：典型的查询场景长什么样
{
  "column": "order_amount",
  "chinese_name": "订单金额",
  "business_definition": "用户实际支付的订单总金额，包含商品价格和运费，不包含优惠券抵扣",
  "calculation": "order_amount = item_price * quantity + shipping_fee",
  "data_type": "DECIMAL(18,2)",
  "example_query": "查询上月订单总金额超过1000元的用户"
}

血缘关系

数据血缘解决的是"这个指标从哪来"的问题。用户说"我要看 GMV"，Agent 需要通过血缘找到 GMV 最终落在哪张宽表里，中间经过了哪些加工步骤。

flowchart LR subgraph source["数据源"] ODS_ORD["ods_order_detail"] ODS_PAY["ods_payment_record"] end subgraph dwd["DWD 层"] DWD_ORD["dwd_order_fact"] end subgraph dws["DWS 层"] DWS_GMV["dws_daily_gmv"] end subgraph ads["ADS 层"] ADS_REPORT["ads_gmv_report"] end ODS_ORD --> DWD_ORD ODS_PAY --> DWD_ORD DWD_ORD --> DWS_GMV DWS_GMV --> ADS_REPORT

SQL-PaLM 论文的研究结果表明，schema linking——也就是准确定位到相关的表和列——是 Text-to-SQL 任务中准确率的最大瓶颈。元数据标注的质量直接决定了这一环节的上限。

语义层与指标体系：让自然语言准确映射到数据

有了元数据，Agent 知道了有哪些表和字段。但用户说的是"GMV"、"获客成本"、"复购率"，不是 dws_daily_gmv.gmv_amount。语义层的作用就是在业务语言和技术实现之间搭一座桥。

语义层解决什么问题

没有语义层的时候，每个分析师写 SQL 都按自己的理解来。同一个"收入"指标，A 用的是含税金额，B 用的是不含税金额，C 把退款也扣掉了。三个人三个数。

语义层要做的事：

  1. 指标定义标准化：把"GMV"、"DAU"这些指标的计算逻辑封装成统一定义，所有人查出来都是同一个数
  2. 维度统一管理：确保"按地区"在所有指标里含义一致
  3. Join 路径自动导航：Agent 不需要知道底层表怎么关联，语义图会自动找到路径

指标定义 样例

MetricFlow 用 YAML 定义语义模型和指标。先看一个语义模型的定义：

semantic_models:
  - name: orders
    description: "订单事实表，记录每笔订单的详细信息"
    model: ref('fct_orders')
  
    entities:
      - name: order_id
        type: primary
        expr: "order_id"
      - name: customer
        type: foreign
        expr: "customer_id"
  
    dimensions:
      - name: order_status
        type: categorical
        expr: "status"
      - name: order_date
        type: time
        type_params:
          time_granularity: day
      - name: is_food_order
        type: categorical
        expr: "case when order_type = 'food' then true else false end"
  
    measures:
      - name: order_total
        agg: sum
        expr: "order_amount"
      - name: order_count
        agg: sum
        expr: "1"

然后是指标定义。MetricFlow 支持简单指标、派生指标等多种类型：

metrics:
  - name: revenue
    description: "总营收，所有已完成订单的金额总和"
    type: simple
    label: "营收"
    type_params:
      measure: order_total
  
  - name: food_revenue_pct
    description: "餐饮营收占比"
    type: derived
    label: "餐饮营收占比"
    type_params:
      expr: "food_revenue / revenue"
      metrics:
        - name: food_revenue
        - name: revenue

定义好之后，Agent 生成 SQL 时就不需要自己拼 JOIN 和聚合逻辑了，直接通过 MetricFlow 的 API 查询指标即可。

语义层架构

以Cube为例，看一下语义层架构的四个支柱：

flowchart TB subgraph modeling["数据建模"] direction TB CUBES["Cubes - 数据模型"] Measures["Measures - 指标定义"] JOINS["Joins - 关联关系"] end subgraph access["访问控制"] direction TB RLS["行级安全"] POLICY["策略代码"] end subgraph caching["缓存层"] direction TB PREAGG["Pre-aggregations"] STORE["Cube Store"] end subgraph apis["API 层"] direction TB SQL_API["SQL API"] REST_API["REST API"] GRAPHQL_API["GraphQL API"] MCP_API["MCP Server"] end modeling --> access access --> caching caching --> apis

Cube 的 Semantic SQL 在标准 SQL 基础上增加了 MEASURE 函数，Agent 可以在治理后的指标之上做派生计算，而不是直接操作裸表。同时它提供了 MCP Server 接口，AI Agent 可以通过 MCP 协议直接接入。

对 Agent 来说，语义层解决了三个实际问题：指标歧义（"收入"到底是哪个）、JOIN 复杂度（不用自己拼表关系）、口径一致性（谁查都是同一个逻辑）。

案例库：从需求到 SQL 的案例引擎

案例库不是 SQL 模板库。SQL 模板只能处理固定模式的需求，而真实的业务需求千变万化。案例库要做的，是提供从需求描述到最终 SQL 的完整推理链路——包括需求是怎么拆解的、选了哪些表、为什么这么写 JOIN。

一条语料长什么样

{
  "id": "CASE-2026-0042",
  "requirement_text": "查询上个月各渠道的新用户数量和获客成本",
  "requirement_analysis": {
    "核心指标": ["用户数量", "获客成本"],
    "维度": ["渠道"],
    "时间范围": "上个月",
    "过滤条件": "新用户（首次注册时间在上月）",
    "隐含逻辑": "获客成本 = 渠道总投入 / 新用户数量"
  },
  "involved_tables": ["dwd_user_register", "dws_channel_cost", "dim_channel"],
  "sql": "SELECT c.channel_name, COUNT(DISTINCT u.user_id) AS new_user_count, cc.total_cost / COUNT(DISTINCT u.user_id) AS cac FROM dwd_user_register u JOIN dim_channel c ON u.channel_id = c.channel_id JOIN dws_channel_cost cc ON c.channel_id = cc.channel_id AND cc.month = DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 MONTH), '%Y%m') WHERE DATE_FORMAT(u.register_time, '%Y%m') = DATE_FORMAT(DATE_SUB(CURDATE(), INTERVAL 1 MONTH), '%Y%m') GROUP BY c.channel_name, cc.total_cost ORDER BY new_user_count DESC",
  "validation_result": "PASS",
  "tags": ["获客分析", "渠道分析"]
}

每条案例语料都包含需求原文、需求分析（指标/维度/过滤条件/隐含逻辑）、涉及的表、最终 SQL、验证结果和标签。Agent 拿到新需求时，先检索相似的语料，参考历史案例的推理路径来生成 SQL。

案例语料从哪来

四个来源并行积累：

  1. 历史需求工单：BI 团队过去处理过的需求，是最直接的来源
  2. 人工构造：针对高频业务场景，由分析师主动编写典型案例
  3. 用户反馈：用户通过 AI 系统提交需求后，人工审核的结果可以回流成语料
  4. 错误案例：Agent 写错的 SQL 同样有价值，标注清楚错在哪里、怎么改

检索策略

新需求进来后，怎么找到最相关的历史案例？通常结合两种方式：

def retrieve_similar_cases(requirement_text, top_k=5):
    # 语义相似度：用 embedding 做向量检索
    query_embedding = embed(requirement_text)
    semantic_matches = vector_search(query_embedding, top_k=top_k * 2)
  
    # 关键词匹配：提取指标、维度、表名做精确检索
    entities = extract_entities(requirement_text)
    keyword_matches = keyword_search(entities, top_k=top_k * 2)
  
    # 融合排序
    candidates = merge_and_rerank(semantic_matches, keyword_matches)
    return candidates[:top_k]

检索到的案例会作为 few-shot examples 放进 prompt，帮 Agent 理解当前需求应该映射到哪些表、用什么 SQL 模式。

知识库：业务背景的深层理解

案例库教 Agent "怎么做"，知识库告诉它"为什么这么做"。

举个例子：用户说"查华东区的 GMV"。如果 Agent 不知道"华东区"包含哪些省份，它可能就只查了"华东"这个字段值，漏掉了上海、江苏、浙江的数据。又或者，"GMV"的定义在三个月前改过——之前包含未付款订单，现在只算已付款的——如果 Agent 不知道这个变更，查出来的数就对不上。

知识库存什么

类别 举例 解决什么问题
业务术语表 "GMV" = 成交总额，包含未付款订单 术语歧义
决策记录 "Q3 起获客成本不再包含品牌广告" 口径变更
组织架构 "华东区 = 上海 + 江苏 + 浙江" 维度理解
业务规则 "退款完成 7 天后才从 GMV 扣除" 时间逻辑
数据使用说明 "dws_daily_gmv 不含测试订单" 选错数据源

怎么组织

以结构化的 YAML 文件管理为例：

# knowledge/business_terms.yaml
terms:
  - term: "GMV"
    full_name: "Gross Merchandise Volume"
    chinese_name: "成交总额"
    definition: "所有已创建订单的金额总和，包含已支付和未支付订单"
    exclusion: "不包含测试订单、内部员工订单"
    related_metrics: ["净GMV", "实际收入"]
    last_updated: "2026-06-15"
    owner: "数据平台组"

# knowledge/org_structure.yaml
organizations:
  - name: "华东区"
    type: "region"
    includes: ["上海分公司", "江苏分公司", "浙江分公司"]
    parent: "全国"

与 Agent 的集成

知识库通过 RAG 方式接入 Agent 的工作流。当 Agent 遇到不确定的业务概念时，先去知识库里查：

sequenceDiagram participant U as 用户 participant A as Agent participant KB as 知识库 participant SL as 语义层 participant DB as 数据库 U->>A: "查询华东区上月的 GMV" A->>KB: 检索"华东区"的组织架构定义 KB-->>A: 华东区 = 上海 + 江苏 + 浙江 A->>KB: 检索"GMV"的业务定义 KB-->>A: GMV = 所有已创建订单金额总和 A->>SL: 查询 GMV 指标定义和对应表 SL-->>A: dws_daily_gmv, 按 region 维度筛选 A->>DB: SELECT ... WHERE region IN ('上海','江苏','浙江') DB-->>A: 查询结果 A-->>U: 华东区上月 GMV 为 xxx 万元

这样 Agent 不只是在"翻译"需求，而是在"理解"需求。

Skill 系统：从需求到 SQL 的自动化流水线

前面说的元数据、语义层、案例库、知识库，都是"知识"层面的东西。Skill 系统把这些知识串起来，变成可以自动执行的工作流。

一个 Skill 的定义

name: requirement_to_sql
description: "将自然语言需求转换为可执行的 SQL 查询"
version: "1.2.0"

inputs:
  - name: requirement_text
    type: string
    required: true

steps:
  - name: "需求理解"
    action: "analyze_requirement"
    prompt_template: |
      请分析以下需求，提取关键信息：
      需求：{requirement_text}
      请提取：核心指标、分析维度、过滤条件、时间范围、隐含逻辑
  
  - name: "语义检索"
    action: "semantic"
    depends_on: ["需求理解"]
    uses:
      - semantic_layer
  - name: "案例检索"
    action: "retrieve_cases"
    depends_on: ["需求理解"]
    uses:
      - case_corpus

  - name: "元数据"
    action: "research_metadata"
    depends_on: ["需求理解"]
    uses:
      - metadata_service
  
  - name: "SQL 生成"
    action: "generate_sql"
    depends_on: ["语义检索", "案例检索", "元数据"]
    prompt_template: |
      基于以下信息生成 SQL：
      需求分析：{step_需求理解.output}
      可用表和字段：{step_元数据.output}
      参考：{语义检索, step_案例检索.output， 元数据}
  
  - name: "SQL 校验"
    action: "validate_sql"
    depends_on: ["SQL 生成"]

outputs:
  - name: sql
    from: "step:SQL 生成.output"

每个 step 做一件事，step 之间有明确的依赖关系。Agent 按照这个定义一步步执行，遇到问题可以回溯到上一步重试。

编排与执行

实际运行时，Agent 通过 function calling 调用各个 Skill。Anthropic 的 tool use 模式就是一个典型的实现：Agent 根据任务需要，自主决定调用哪个工具、传什么参数、怎么处理返回结果。

def process_requirement(requirement_text):
    analysis = call_skill("requirement_analysis", {
        "text": requirement_text,
        "knowledge_base": load_knowledge_base()
    })
  
    schema_info = call_tool("metadata_lookup", {
        "metrics": analysis.metrics,
        "dimensions": analysis.dimensions
    })
  
    similar_cases = call_tool("case_search", {
        "query": requirement_text,
        "top_k": 5
    })
  
    sql = call_skill("sql_generation", {
        "requirement": analysis,
        "schema": schema_info,
        "examples": similar_cases
    })
  
    validation = call_skill("sql_validation", {"sql": sql})
  
    return {"sql": sql, "analysis": analysis, "validation": validation}

校验体系：保障 AI 输出质量

Agent 生成的 SQL 不能直接用。不是因为它一定错，而是因为你不知道它什么时候会错。校验体系的作用就是在 SQL 到达用户之前，把明显有问题的拦截下来。

三层校验

flowchart LR subgraph L1["第一层：AI 自动评估"] direction LR SYNTAX["语法检查"] SAFETY["安全审查"] SEMANTIC["语义一致性"] PERFORMANCE["性能评估"] end subgraph L2["第二层：对抗审查"] direction LR REVIEWER["Reviewer Agent"] RED_TEAM["Red Team 测试"] EDGE_CASE["边界场景验证"] end subgraph L3["第三层：人工核验"] direction LR ANALYST["数据工程师审核"] SAMPLING["抽样验证"] FEEDBACK["反馈闭环"] end L1 --> L2 L2 --> L3 L3 -->|反馈| L1

第一层：AI 自动评估，SQL 生成后立刻跑：

def auto_validate(sql, requirement, schema_info):
    results = {}
    results["syntax"] = check_sql_syntax(sql)
    results["safety"] = check_safety(sql, 
        forbidden=["DROP", "DELETE", "UPDATE", "INSERT", "TRUNCATE"])
    results["schema"] = check_schema_consistency(sql, schema_info)
    results["semantic"] = check_semantic_match(sql, requirement)
    results["performance"] = estimate_query_cost(sql)
    return results

其中语义一致性检查最有价值——用另一个 LLM 来审查生成的 SQL 是否真的回答了用户的需求。这一步能抓出很多"语法正确但逻辑错误"的情况。

第二层：对抗审查，一个独立的 Reviewer Agent 专门找茬：笛卡尔积风险、JOIN 条件缺失、聚合粒度不对、数据倾斜隐患……这种对抗机制能显著提高输出质量。

第三层：人工核验，涉及财务数据或对外报告的场景，最终需要数据工程师过一遍。人看的不是语法，而是业务逻辑对不对、结果合不合理。

反馈闭环

校验结果不应该只是 PASS 或 FAIL。每一次失败都应该回流到系统中：

flowchart LR GEN["SQL 生成"] --> VALIDATE["校验"] VALIDATE -->|PASS| OUTPUT["输出结果"] VALIDATE -->|FAIL| ANALYZE["错误分析"] ANALYZE --> UPDATE_CORPUS["更新案例库"] ANALYZE --> UPDATE_KB["更新知识库"] ANALYZE --> UPDATE_PROMPT["优化 Skill Prompt"] UPDATE_CORPUS --> GEN UPDATE_KB --> GEN UPDATE_PROMPT --> GEN

失败的案例标注后加入语料库，新发现的业务规则加入知识库，prompt 的薄弱环节加固——系统的准确率就在这个闭环里一点点涨上去。

端到端工作流

把核心模块串起来，完整流程如下：

flowchart TB REQ["用户需求输入"] --> PARSE["需求分析"] PARSE --> SL["语义层查询"] PARSE --> META["元数据服务"] PARSE --> KB["知识库检索"] PARSE --> CASE["案例库检索"] SL --> GEN["SQL 生成"] META --> GEN KB -.-> GEN CASE -.-> GEN GEN --> VALIDATE["三层校验"] VALIDATE -->|PASS| RESULT["输出 SQL 与执行结果"] VALIDATE -->|FAIL| RETRY["修正重试"] RETRY --> GEN RESULT --> FEEDBACK["反馈收集"] FEEDBACK --> KB FEEDBACK --> CASE

四个设计原则贯穿始终：分层解耦（每个模块独立运作）、知识驱动（语义层、元数据、案例库与知识库并行检索，构成 Agent 的知识底座）、校验前置（核心生成环节设置多重检查点）、持续迭代（反馈闭环让系统越用越准）。

写在最后

AI 自助需求开发不是换个更强的模型就能解决的问题。它需要元数据管理让 Agent 看懂数据资产，需要语义层让自然语言准确映射到指标，需要案例库提供推理样本，需要知识库补充业务背景，需要 Skill 系统编排自动化流程，需要校验体系保障输出质量。六个模块协同建设，才能真正实现"用户提需求，Agent 出 SQL"的体验。

好消息是，这六个模块都不需要从零造轮子。OpenMetadata、Cube、MetricFlow 这些开源项目已经提供了成熟的基础设施，你要做的是把它们串起来，针对自己的业务场景做好标注、积累和校验，当然开源意味着不能满足我们所有诉求，需要根据实际情况决定自研还是使用开源。