导读：本文是 “数据拾光者” 专栏的第一百一十八篇文章，这个系列聚焦自然语言处理和大模型 相关实践。本篇主要学习智能问数开源项目SQLBot 源码，尤其是Text2Sql部分的详细流程。

关键词：Agent Skills,Claude,MCP,渐进式披露, 智能问数, 低代码开发
欢迎转载，转载请注明出处以及链接，更多关于自然语言处理、推荐系统优质内容请关注如下频道。
知乎专栏：数据拾光者
公众号：数据拾光者

最近在做智能问数项目，之前也调研了一些开源和闭源的好的智能问数产品和项目：

AI那些趣事系列116：智能问数：让人人都能玩转数据的 AI 黑科技

最后我们打算借鉴开源项目SQLBot来构建我们的智能问数产品。本篇是学习SQLBot的源码实践，站在巨人的肩膀上才能看的更远，至少能学习先进经验，避开一些显而易见的大坑。下面是SQLBot 的 Text2SQL  源码学习笔记。

SQLBot 的 Text2SQL 核心是基于大模型 + RAG 实现自然语言到 SQL 的转换，全流程围绕「上下文构建→Prompt 编排 →大模型调用→SQL 生成→验证优化」展开，以下结合源码拆解每一步细节。

整体流程总览

主要流程包括：用户自然语言提问，上下文收集（元数据+术语+模板），Prompt 结构化构建（规则+示例+元数据），调用大模型生成 SQL，SQL 验证/执行（有效性校验），结果返回+数据训练（迭代优化）。

步骤 1：上下文收集（Text2SQL 核心基础）

大模型生成准确 SQL 的前提是「知道业务上下文 + 数据库规则」，SQLBot 从数据源元数据 、术语库 、SQL 模板  三个维度收集上下文，是生成高质量 SQL 的核心。

1. 数据源元数据获取（数据库表 / 字段信息）

对应文件：SQLBot/backend/apps/db/db_sql.py
作用：动态获取不同数据源（MySQL/PostgreSQL/SQL Server 等）的版本、表、字段信息，作为大模型生成 SQL 的「数据字典」。

核心函数解析：

get_version_sql：获取数据库版本（如 ClickHouse 不同版本表注释语法不同，需版本适配）；

defget_version_sql(ds:CoreDatasource,conf:DatasourceConf):
    ifequals_ignore_case(ds.type,"mysql","doris","starrocks"):
        return"SELECT VERSION()"
    elifequals_ignore_case(ds.type,"sqlServer"):
        return"select SERVERPROPERTY('ProductVersion')"
# 其他数据库（PG/Oracle/CK等）适配逻辑...

• get_table_sql

  ：获取指定数据库 / 模式下的表名 + 表注释，适配不同数据库的系统表语法；
  例如 PostgreSQL 适配：

elif equals_ignore_case(ds.type, "pg", "excel"):    return """  SELECT c.relname AS TABLE_NAME,         COALESCE(COALESCE(d.description, obj_description(c.oid)), '') AS TABLE_COMMENT  FROM pg_class c           LEFT JOIN pg_namespace n ON n.oid = c.relnamespace           LEFT JOIN pg_description d ON d.objoid = c.oid AND d.objsubid = 0  WHERE n.nspname = :param    AND c.relkind IN ('r', 'v', 'p', 'm')    AND c.relname NOT LIKE 'pg_%'    AND c.relname NOT LIKE 'sql_%'""", conf.dbSchema

• get_field_sql

  ：获取指定表的字段名、数据类型、字段注释，是生成 SQL 字段选择的核心依据；
  例如 MySQL 适配：

elif equals_ignore_case(ds.type, "mysql"):    sql1 = """    SELECT COLUMN_NAME,DATA_TYPE,COLUMN_COMMENT    FROM INFORMATION_SCHEMA.COLUMNS    WHERE TABLE_SCHEMA = :param1    """    sql2 = " AND TABLE_NAME = :param2" if table_name else ""    return sql1 + sql2, conf.database, table_name

关键逻辑：

• 适配 10+ 种数据库（MySQL/PG/SQL Server/Oracle/CK/DM 等）的系统表语法；

• 通过 equals_ignore_case 做数据库类型大小写兼容；

• 支持参数化查询（:param/%s），避免 SQL 注入。

2. 术语库整合（业务术语→数据库映射）

对应文件：SQLBot/backend/apps/terminology/curd/terminology.py
作用：将业务术语（如「订单金额」）映射到数据库表 / 字段（如 orders.amount），解决「大模型不懂业务术语」的问题。

核心函数：build_terminology_query

def build_terminology_query(session: SessionDep, oid: int, name: Optional[str] = None,                            paginate: bool = True, current_page: int = 1, page_size: int = 10,                            dslist: Optional[list[int]] = None):    # 1. 基础查询+数据源筛选（只查当前数据源关联的术语）    parent_ids_subquery, child = get_terminology_base_query(oid, name)    if dslist:        ds_filter_condition = or_(            *[Terminology.datasource_ids.contains([ds_id]) for ds_id in dslist],            Terminology.datasource_ids == []        )        parent_ids_subquery = parent_ids_subquery.where(ds_filter_condition)    # 2. 聚合术语关联信息（同义词/数据源名称）    children_subquery = (        select(child.pid, func.jsonb_agg(child.word).filter(child.word.isnot(None)).label('other_words'))        .where(child.pid.isnot(None)).group_by(child.pid).subquery()    )    # 3. 最终查询：返回术语+同义词+数据源名称+状态等    stmt = (        select(            Terminology.id, Terminology.word, Terminology.description,            children_subquery.c.other_words, func.jsonb_agg(CoreDatasource.name).label('datasource_names')        )        .outerjoin(children_subquery, Terminology.id == children_subquery.c.pid)        .where(Terminology.id.in_(paginated_parent_ids), Terminology.oid == oid)        .group_by(...)    )    return stmt, total_count, total_pages, current_page, page_size

关键逻辑：

• 按「数据源」筛选术语（只加载当前数据源相关的业务术语）；

• 聚合术语的「同义词 / 别名」（other_words），让大模型理解「订单金额」=「交易金额」= orders.amount；

• 关联数据源名称，确保术语与数据源匹配。

3. SQL 模板加载（数据库语法规则 + 示例）

对应文件：

• 模板文件：SQLBot/backend/templates/sql_examples/PostgreSQL.yaml、Microsoft_SQL_Server.yaml

• 模板加载函数：SQLBot/backend/apps/template/generate_sql/generator.py

作用：为不同数据库提供「语法规则 + 正确 / 错误示例」，约束大模型生成符合数据库规范的 SQL。

（1）模板核心结构（以 PostgreSQL 为例）

template:  quot_rule: |  # 引号规则（强制双引号包裹标识符）    <rule>      必须对数据库名、表名、字段名、别名外层加双引号（"）。      <note>        1. 点号（.）不能包含在引号内，必须写成 "schema"."table"        2. 即使标识符不含特殊字符，也需强制加双引号      </note>    </rule>  limit_rule: |  # 分页规则（LIMIT 语法）    <rule>当需要限制行数时，必须使用标准的LIMIT语法</rule>  other_rule: |  # 通用规则（表别名、禁止*、中英别名等）    <rule>必须为每个表生成别名（不加AS）</rule>    <rule>禁止使用星号(*)，必须明确字段名</rule>    <rule>中文/特殊字符字段需保留原名并添加英文别名</rule>  basic_example: |  # 正确/错误示例（核心参考）    <example>      <input>查询 TEST.ORDERS 表的前100条订单（含中文字段和百分比）</input>      <output-bad>        SELECT * FROM TEST.ORDERS LIMIT 100  -- 错误：未加引号、使用星号      </output-bad>      <output-good>        SELECT           "t1"."订单ID" AS "order_id",          "t1"."金额" AS "amount",          ROUND("t1"."折扣率" * 100, 2) || '%' AS "discount_percent"        FROM "TEST"."ORDERS" "t1"        LIMIT 100      </output-good>    </example>

（2）模板加载函数

def get_sql_example_template(db_type: Union[str, DB]):    template = get_base_sql_template(db_type)    return template['template']

• 根据数据库类型（如 PostgreSQL/SQL Server）加载对应的模板；

• 提取模板中的「规则 + 示例」，作为 Prompt 的核心约束。

步骤 2：Prompt 结构化构建

SQLBot 未直接暴露 Prompt 构建的完整代码，但结合上下文和模板，Prompt 会按以下结构编排（核心逻辑）：

【系统指令】你是专业的 Text2SQL 助手，需遵循以下规则生成 SQL：1. 语法规则：{quot_rule + limit_rule + other_rule}（来自 SQL 模板）2. 输出格式：返回 JSON 结构 {"success":bool,"sql":"","tables":[""],"chart-type":""}3. 禁止生成危险 SQL（DROP/ALTER 等）【数据字典】当前数据源（PostgreSQL）的表/字段信息：- 表："sample_country_gdp"（国家GDP表）  字段："country"（国家）、"continent"（大洲）、"year"（年份）、"gdp"（GDP值）【业务术语】- "中国GDP" → "sample_country_gdp"."country" = '中国' AND "sample_country_gdp"."gdp"- "年度GDP" → "sample_country_gdp"."year" + "sample_country_gdp"."gdp"【参考示例】{basic_example}（来自 SQL 模板的正确/错误示例）【用户问题】{用户的自然语言提问，如：查询2024年中国的GDP}【生成要求】1. 严格遵循语法规则，使用双引号包裹所有标识符；2. 字段必须加英文别名，禁止使用*；3. 生成的 SQL 需可直接执行，无需修改；4. 返回指定 JSON 格式，仅输出 JSON 内容。

步骤 3：大模型调用（生成 SQL）

对应文件：SQLBot/backend/apps/ai_model/model_factory.py
作用：初始化大模型客户端，将构建好的 Prompt 传入大模型，生成 SQL。

核心代码：OpenAIAzureLLM 类

class OpenAIAzureLLM(BaseLLM):    def _init_llm(self) -> AzureChatOpenAI:        # 提取 Azure OpenAI 配置（API版本、部署名）        api_version = self.config.additional_params.get("api_version")        deployment_name = self.config.additional_params.get("deployment_name")        if api_version: self.config.additional_params.pop("api_version")        if deployment_name: self.config.additional_params.pop("deployment_name")        # 初始化 Azure OpenAI 客户端        return AzureChatOpenAI(            azure_endpoint=self.config.api_base_url,  # Azure 端点            api_key=self.config.api_key or 'Empty',    # API 密钥            model_name=self.config.model_name,         # 模型名称            api_version=api_version,                   # API 版本            deployment_name=deployment_name,           # 部署名            streaming=True,                            # 流式返回            **self.config.additional_params,           # 其他参数（温度/最大令牌等）        )

关键逻辑：

• 适配 Azure OpenAI 私有化部署（企业级常用）；

• 支持流式输出（streaming=True），实时返回生成的 SQL；

• 配置可扩展（通过 additional_params 传入温度、top_p 等参数，控制生成效果）；

• 调用流程：初始化客户端 → 传入 Prompt → 大模型返回 JSON 格式的 SQL 结果。

步骤 4：SQL 验证与执行

1. SQL 执行验证

对应文件：SQLBot/frontend/src/api/datasource.ts
作用：执行生成的 SQL，验证语法正确性、数据有效性。

execSql: (id: number | string, sql: string) =>    request.post(`/datasource/execSql/${id}`, { sql: sql })

• 传入「数据源 ID + 生成的 SQL」，调用后端执行；

• 若执行报错（如语法错误、字段不存在），则反馈给大模型重新生成，或记录错误信息。

2. SQL 校准与迭代优化

对应文件：SQLBot/backend/apps/data_training/curd/data_training.py
作用：通过用户交互数据（问题 + 正确 SQL）训练优化，实现「越问越准」。

核心代码：

def build_data_training_query(session: SessionDep, oid: int, name: Optional[str] = None,                              paginate: bool = True, current_page: int = 1, page_size: int = 10):    # 1. 查询用户标注的「问题-SQL」训练数据    parent_ids_subquery = get_data_training_base_query(oid, name)    count_stmt = select(func.count()).select_from(parent_ids_subquery.subquery())    total_count = session.execute(count_stmt).scalar()    # 2. 分页获取训练数据（关联数据源、大模型信息）    stmt = (        select(            DataTraining.id, DataTraining.question, DataTraining.create_time,            CoreDatasource.name, AssistantModel.name.label('advanced_application_name')        )        .outerjoin(CoreDatasource, DataTraining.datasource == CoreDatasource.id)        .outerjoin(AssistantModel, DataTraining.advanced_application == AssistantModel.id)        .where(DataTraining.id.in_(paginated_parent_ids))        .order_by(DataTraining.create_time.desc())    )    return stmt, total_count, total_pages, current_page, page_size

关键逻辑：

• 存储用户的「自然语言问题 + 人工校准的 SQL」；

• 后续生成 SQL 时，将这些训练数据加入 Prompt 参考；

• 基于用户交互数据持续迭代，优化生成效果（对应项目核心优势「越问越准」）。

核心亮点总结

1. 多数据库适配

   ：通过 db_sql.py 适配 10+ 种数据库的元数据查询和 SQL 语法；

2. 业务术语对齐

   ：术语库解决「大模型不懂业务」的问题，提升 SQL 贴合业务场景；

3. 规则化约束

   ：通过 SQL 模板强制语法规则（引号、分页、别名等），减少生成错误；

4. 迭代优化

   ：数据训练模块基于用户交互持续校准，实现「越问越准」；

5. 安全可控

   ：结合项目核心优势中的「工作空间隔离 + 细粒度权限」，确保 SQL 执行的安全性。

核心流程示例说明

以「用户提问：查询 2024 年中国的 GDP」为例：

1.上下文收集：

• • 元数据：获取 PostgreSQL 中 sample_country_gdp 表的字段（country/continent/year/gdp）；

  • 术语库：「中国 GDP」→ sample_country_gdp"."country" = '中国' AND "sample_country_gdp"."gdp"；

  • 模板：加载 PostgreSQL 引号规则、LIMIT 规则、示例；

2.Prompt 构建：整合上述信息，生成结构化 Prompt；

3.大模型调用：AzureChatOpenAI 接收 Prompt，生成 SQL：sql

SELECT "country" AS "country_name", "gdp" AS "gdp_usd" 
FROM "Sample_Database"."sample_country_gdp" 
WHERE "year" = '2024' AND "country" = '中国'

4.SQL 验证：调用 execSql 执行 SQL，返回 2024 年中国 GDP 数据；

5.迭代优化：若生成错误，记录问题和正确 SQL 到数据训练模块，后续优化 Prompt。

最新最全的文章请关注我的微信公众号或者知乎专栏：数据拾光者。

码字不易，欢迎小伙伴们关注和分享。