混合检索 RAG 与 SQL Agent 架构研究：从 RRF 重排到执行引导的 Text-to-SQL 闭环

企业知识访问的双重范式：非结构化 RAG 与结构化 SQL Agent

在实际业务系统中，外部知识通常分布在两类截然不同的存储介质中：

非结构化技术文档、手册与工单：适合通过分块、稠密向量与稀疏倒排的**混合检索增强生成（Hybrid RAG）**回答定性与流程类问题；
高维关系型数据表（OLTP / OLAP）：当问题涉及精确聚合、分组排序、多表关联或时间窗口统计时，向量相似度检索在原理上无法保证完备性与算术正确性，必须通过 Text-to-SQL Agent 将自然语言编译为关系代数查询并在数据库引擎上执行。

本文分别剖析这两套系统在生产级精度下的关键算法与工程闭环。

第一部分：混合检索 RAG 管线设计

单纯依赖固定窗口切分（如按 500 字符硬切）加上单一稠密向量检索（Bi-Encoder Cosine Similarity）的朴素 RAG，在面对专有名词、型号编码（如 PRO630、ERR_CONN_RESET）或跨段落条件约束时召回率会显著退化。高可靠 RAG 管线由**结构感知分块、稀疏+稠密双路召回、倒数秩融合（RRF）与交叉编码器重排（Cross-Encoder Reranking）**四级组成。

1. 结构感知分块与父子上下文绑定

在切分 Markdown 或技术文档时，应遵循语法树边界（标题层级 # / ##、代码块围栏、表格完整性），避免将代码块或公式从中间截断。为了兼顾“检索向量的语义聚焦度”与“生成上下文的完整性”，常采用**小块检索、大块回填（Small-to-Big / Parent-Child Chunking）**策略，或为每个子块前置注入所属章节的面包屑路径（Contextual Header）：

[文档：PRO630 运动学手册 > 章节：解析逆运动学 > 子节：奇异解与分支合并]
当 sin(theta_5) = 0 时，不能使用除以 s_5 的角度和公式……

2. BM25 稀疏检索与稠密向量的互补性

对于查询 $Q = \{q_1, q_2, \dots, q_m\}$ 与文档块 $D$ ，Okapi BM25 打分定义为：

\operatorname{BM25}(Q, D) = \sum_{i=1}^{m} \operatorname{IDF}(q_i) \cdot \frac{f(q_i, D) \cdot (k_1 + 1)}{f(q_i, D) + k_1 \cdot \left(1 - b + b \cdot \frac{|D|}{\text{avgdl}}\right)}

其中 $f(q_i, D)$ 为词项在文档块中的词频， $k_1 \in [1.2, 2.0]$ 控制词频饱和速度， $b \approx 0.75$ 控制文档长度归一化。

BM25（精确词法匹配）：对低频实体名、函数签名、报错码具有极高区分度，但无法处理同义词改写；
Dense Embedding（双塔语义匹配）：将查询和文档映射到单位超球面 $\mathbb{S}^{d-1}$ ，通过 HNSW 图索引检索 $\langle e_Q, e_D \rangle$ ，擅长捕捉意图与同义表述，但会将仅相差一个数字编号的两个不同型号压缩到相近子空间。

3. 倒数秩融合（Reciprocal Rank Fusion, RRF）

由于 BM25 分数无上界且随查询词数剧烈波动，而余弦相似度局限在 $[-1, 1]$ ，直接对两路原始分数做线性加权 $\alpha s_{\text{dense}} + (1-\alpha) s_{\text{bm25}}$ 极易受分布漂移影响。**倒数秩融合（RRF）**丢弃绝对分值，仅利用候选文档在各检索器中的名次 $r_m(d) \in \{1, 2, \dots\}$ 进行非线性合并：

\operatorname{RRF}(d) = \sum_{m \in \{\text{dense}, \text{bm25}\}} \frac{w_m}{k + r_m(d)}

平滑常数通常取 $k = 60$ 。RRF 对单路异常高分具有天然免疫性，同时对同时出现在两路前列的交集文档赋予显著增益。

4. Cross-Encoder 精排与阈值截断

双塔向量模型（Bi-Encoder）将查询 $Q$ 与文档 $D$ 独立编码为单向量，丢失了词元间的细粒度交叉注意力（Token-level Cross-Attention）。在 RRF 选出 Top- $K$ （如 $K = 50$ ）候选后，使用 Cross-Encoder 重排模型（如 bge-reranker-v2-m3）将拼接序列 [CLS] Q [SEP] D [SEP] 送入完整 Transformer 编码器：

s_{\text{rerank}}(Q, D) = \sigma\bigl(W \cdot h_{\text{[CLS]}}(Q \oplus D)\bigr)

精排后仅保留 Top- $N$ （如 $N = 5$ ）且满足最低相关性阈值 $s_{\text{rerank}} \ge \tau_{\min}$ 的片段送入 LLM，并在 System Prompt 中强制要求每条结论标注引用块编号 [1]、[2]，对无证据支撑的问题显式拒答。

第二部分：SQL Agent 架构与执行引导闭环

在包含数百张表、数千个字段的企业数仓中，直接将全量 CREATE TABLE DDL 塞入上下文不仅会撑爆 Token 预算，还会引发严重的字段张冠李戴（Hallucinated Columns）与关联路径遗漏。一个生产级 SQL Agent 的工作流由四个核心模块组成：

用户问题 ---> [1. 实体值检索与 Schema Linking] ---> [2. 外键连接子图剪枝]
                                                           |
                                                           v
最终回答 <--- [4. 数据库只读执行与报错自修正] <--- [3. SQL 生成与 AST 静态校验]

1. 值级倒排检索与 Schema Linking

用户提问往往只提及业务枚举值而非列名（例如问“华东大区上月已签收的订单额”，其中 '华东' 存于 regions.region_name，而 '已签收' 在 orders.status 中实际编码为 'DELIVERED'）。因此 Schema Linking 必须分两路进行：

值级链接（Value Linking）：对数据库中低基数枚举列（Cardinality $< 1000$ ）的去重值构建 MinHash LSH 或 BM25 索引，提取问题中的命名实体去匹配真实库内字面量，将匹配到的 table.column = 'exact_value' 作为注释注入对应列定义；
表列语义链接（Table/Column Linking）：将每张表的描述、字段名及注释向量化，检索与问题最相关的种子表集合 $T_{\text{seed}}$ 。

2. 基于外键图的最小连接子图扩展

两张业务表之间往往需要通过中间桥接表（Junction Table）关联，而中间表名未必出现在用户问题中。将数据库模式建模为无向图 $G = (\mathcal{T}, \mathcal{E}_{\text{FK}})$ ，其中顶点为表，边为主外键引用关系。给定检索出的种子表集合 $T_{\text{seed}}$ ，通过计算包含 $T_{\text{seed}}$ 的近似**斯坦纳树（Steiner Tree）**或限定跳数（ $\le 2$ 跳）的最短路闭包，自动补全所有必需的中间连接表，再裁剪掉与查询无关的宽表冗余字段，生成精简 DDL。

3. SQL AST 静态安全网关

决不能将模型生成的 SQL 字符串直接拼接到数据库连接上执行。即使配置了数据库只读账号，也必须在应用层通过 SQL 解析器（如 sqlglot）将 SQL 解析为抽象语法树（AST），执行三项强制检查与改写：

只读语法白名单：根节点类型必须严格为 SELECT 或 UNION，拒绝任何包含 INSERT、UPDATE、DELETE、DROP、ALTER、COPY、pg_sleep 等危险节点的多语句注入；
笛卡尔积与无界扫描拦截：检查多表 JOIN 是否缺失 ON 条件；
强制注入 LIMIT：若顶层 SELECT 未指定 LIMIT 或上限超过安全阈值（如 $1000$ 行），在 AST 上自动追加或收紧 LIMIT，防止千万级明细拉爆内存与上下文窗口。

4. 执行引导的自修正环路（Execution-Guided Self-Correction）

生成 SQL 后的常见错误分为三类：语法/方言错误（如在 PostgreSQL 中误用 MySQL 的 DATE_FORMAT）、Schema 引用错误（列名拼写偏差或聚合未入 GROUP BY）、以及空结果集陷阱（过滤条件与库内实际枚举值不匹配导致返回 0 rows）。

Agent 在受控重试环路（通常限制 max_retries = 3）中：

先通过 EXPLAIN 或带超时的只读事务执行 SQL；
若数据库返回异常，提取错误码、报错位置及问题子句回传给修正节点；
若执行成功但返回 0 行且查询包含字符串相等过滤 WHERE col = 'val'，自动触发一次针对该列的 SELECT DISTINCT col FROM tbl WHERE col ILIKE '%...%' LIMIT 10 探针查询，核实是否因枚举值别名导致误判。

AST 校验与执行自修正核心代码实现

下面给出结合 sqlglot AST 安全改写与数据库反馈循环的 Python 参考实现：

from dataclasses import dataclass
import sqlite3
import sqlglot
from sqlglot import exp

FORBIDDEN_EXPRESSIONS = (
    exp.Insert,
    exp.Update,
    exp.Delete,
    exp.Drop,
    exp.Alter,
    exp.Create,
    exp.Command,
    exp.Transaction,
)


@dataclass
class SQLValidationResult:
    ok: bool
    sanitized_sql: str = ""
    error: str = ""


def validate_and_rewrite_sql(
    raw_sql: str,
    dialect: str = "postgres",
    max_limit: int = 200,
) -> SQLValidationResult:
    try:
        statements = sqlglot.parse(raw_sql, read=dialect)
    except sqlglot.errors.ParseError as exc:
        return SQLValidationResult(ok=False, error=f"SQL syntax error: {exc}")

    if len(statements) != 1 or statements[0] is None:
        return SQLValidationResult(ok=False, error="Exactly one SQL statement is allowed.")

    tree = statements[0]
    if not isinstance(tree, (exp.Select, exp.Union)):
        return SQLValidationResult(
            ok=False,
            error=f"Only SELECT queries are permitted, got {type(tree).__name__}.",
        )

    for node in tree.walk():
        if isinstance(node, FORBIDDEN_EXPRESSIONS):
            return SQLValidationResult(
                ok=False,
                error=f"Forbidden SQL clause detected: {type(node).__name__}.",
            )
        if isinstance(node, exp.Join) and node.args.get("on") is None and node.args.get("using") is None:
            return SQLValidationResult(
                ok=False,
                error="Unconditional CROSS JOIN / Cartesian join is not allowed; specify an ON clause.",
            )

    if isinstance(tree, exp.Select):
        limit_node = tree.args.get("limit")
        if limit_node is None:
            tree = tree.limit(max_limit)
        else:
            try:
                current_limit = int(limit_node.expression.name)
                if current_limit > max_limit:
                    tree = tree.limit(max_limit)
            except ValueError:
                pass

    return SQLValidationResult(ok=True, sanitized_sql=tree.sql(dialect=dialect))


def execute_with_feedback(
    conn: sqlite3.Connection,
    candidate_sql: str,
    dialect: str = "sqlite",
) -> dict:
    check = validate_and_rewrite_sql(candidate_sql, dialect=dialect)
    if not check.ok:
        return {
            "status": "validation_error",
            "sql": candidate_sql,
            "feedback": check.error,
        }

    try:
        cur = conn.cursor()
        cur.execute(check.sanitized_sql)
        columns = [desc[0] for desc in (cur.description or [])]
        rows = cur.fetchall()
        return {
            "status": "ok",
            "sql": check.sanitized_sql,
            "columns": columns,
            "rows": rows,
            "row_count": len(rows),
        }
    except sqlite3.Error as exc:
        return {
            "status": "execution_error",
            "sql": check.sanitized_sql,
            "feedback": f"Database execution error: {exc}. Check table/column names and GROUP BY clauses.",
        }

将上述 execute_with_feedback 封装为 LangGraph 中的工具检查节点后，SQL Agent 即可在“Schema 子图检索 $\to$ SQL 生成 $\to$ AST 静态门禁 $\to$ 执行与异常自修正”的确定性轨道上运行，兼顾复杂分析查询的准确率与生产数据库的安全边界。