自适应 Schema Linking 策略 - 从 Naive 到 RAG 的演进

在 Text-to-SQL 系统中，Schema Linking（模式链接）是连接自然语言和数据库结构的关键桥梁。当数据库包含数百张表、数千个字段时，如何快速准确地找到用户问题相关的表和字段，成为系统准确性和性能的核心挑战。AskTable 设计了一套自适应 Schema Linking 策略系统，能够根据元数据规模自动选择最优策略。

问题背景

不同规模的数据库需要不同的 Schema Linking 策略：

• •小型数据库（≤3 表，≤100 字段）：可以直接将所有元数据提供给 LLM
• •中型数据库（≤7 表，≤300 字段）：需要 LLM 推理筛选相关表
• •大型数据库（≤10000 表）：必须使用向量检索缩小范围
• •超大型数据库（>10000 表）：需要 Agent 多轮交互式检索

四种 Schema Linking 策略

1. Naive 策略 - 全量元数据

适用场景：小型数据库（≤3 表，≤100 字段）

核心思想：直接将所有可访问的元数据提供给 LLM，无需检索和筛选。

async def _naive_link(self, accessible_meta: MetaAdmin, question: Question) -> MetaContext:
    # 1. 提取关键词和子查询
    await self._rewrite_question(question)

    # 2. 检索示例值和训练样本
    values = await self._retrieve_values(question.keywords or [])
    pairs = await self._retrieve_examples(question.specification)

    # 3. 将示例值添加到字段描述中
    entities = _merge_values_fields(values)
    _add_context_to_meta(accessible_meta, entities)

    return {"meta": accessible_meta, "training_pairs": pairs}

优势：

• •简单直接，无需复杂检索
• •准确率高，不会遗漏相关表
• •延迟低，适合实时查询

2. Reasoning 策略 - LLM 推理筛选

适用场景：中型数据库（≤7 表，≤300 字段）

核心思想：先检索示例值，然后让 LLM 推理筛选相关表。

async def _reasoning_link(self, accessible_meta: MetaAdmin, question: Question) -> MetaContext:
    # 1. 提取关键词
    await self._rewrite_question(question)

    # 2. 检索示例值和训练样本
    values = await self._retrieve_values(question.keywords or [])
    pairs = await self._retrieve_examples(question.specification)

    # 3. 添加示例值到元数据
    entities = _merge_values_fields(values)
    _add_context_to_meta(accessible_meta, entities)

    # 4. LLM 推理筛选相关表
    table_of_interest = await self._pick_tables(
        accessible_meta, question.specification, pairs
    )

    # 5. 过滤元数据
    meta = accessible_meta.filter_tables_by_names(table_of_interest)
    return {"meta": meta, "training_pairs": pairs}

LLM 表筛选 Prompt：

table_of_interest = await prompt_generate(
    "query.select_tables_by_question",
    meta_data=meta_candidate.to_markdown(),
    question=specification,
    translation_examples=dict_to_markdown(training_pairs),
)

优势：

• •利用 LLM 的推理能力
• •减少无关表的干扰
• •保持较高准确率

3. RAG 策略 - 向量检索 + LLM 重排序

适用场景：大型数据库（≤10000 表）

核心思想：使用向量检索缩小范围，再用 LLM 重排序精选表。

async def _rag_link(self, accessible_meta: MetaAdmin, question: Question) -> MetaContext:
    # 1. 提取子查询和关键词
    await self._rewrite_question(question)

    # 2. 三路并行检索
    fields = await self._retrieve_fields(question.subqueries or [])  # 字段检索
    values = await self._retrieve_values(question.keywords or [])    # 值检索
    pairs = await self._retrieve_examples(question.specification)    # 示例检索

    # 3. 融合检索结果
    examples = _training_pair_to_entities(pairs, self.ds.dialect)
    entities = _merge_values_fields(values + fields + examples)
    _add_context_to_meta(accessible_meta, entities)

    # 4. 提取命中的表
    hit_table_names = set([(e["schema_name"], e["table_name"]) for e in entities])

    # 5. 如果命中表过多，使用 LLM 重排序
    if len(hit_table_names) > 3:
        fields_full_names = _get_field_full_names_from_entities(entities)
        hit_fields = accessible_meta.filter_fields_by_names(
            [convert_full_name_to_tuple(f) for f in fields_full_names]
        )
        # LLM 重排序
        table_of_interest = await self._pick_tables(
            hit_fields, question.specification, pairs
        )
        meta = accessible_meta.filter_tables_by_names(table_of_interest)
    else:
        meta = accessible_meta.filter_tables_by_names(list(hit_table_names))

    return {"meta": meta, "training_pairs": pairs}

三路检索详解：

1. •

   字段检索：根据子查询检索相关字段

   fields = await self.ds.retrieve_fields_by_question(subqueries)
   

2. •

   值检索：根据关键词检索示例值

   values = await self.ds.retrieve_values_by_question(keywords)
   

3. •

   示例检索：检索历史 SQL 示例

   pairs = await retrieve_training_pairs(
       datasource_id=self.ds.id,
       query=query,
       role_id=self.role.id
   )
   

优势：

• •高效处理大规模元数据
• •多路检索提高召回率
• •LLM 重排序提高精确率

4. Agentic 策略 - 多轮交互式检索

适用场景：超大型数据库（>10000 表）

核心思想：Agent 多轮交互，逐步缩小范围。

async def _agentic_link(self, accessible_meta: MetaAdmin, question: Question) -> MetaContext:
    # TODO: 实现 Agent 多轮交互式检索
    raise NotImplementedError("Agentic schema linking is not implemented")

设计思路：

1. •Agent 先检索 schema 级别的信息
2. •根据结果选择相关 schema
3. •在选定 schema 内检索 table
4. •最后检索 field 级别信息

自适应策略选择

系统根据元数据规模自动选择最优策略：

async def link(self, question: Question) -> MetaContext:
    accessible_meta = self._get_accessible_meta()

    if config.at_schema_linking_mode == SchemaLinkingMode.auto:
        # 自动模式：根据规模选择策略
        if accessible_meta.table_count <= 3 and accessible_meta.field_count <= 100:
            return await self._naive_link(accessible_meta, question)
        elif accessible_meta.table_count <= 7 and accessible_meta.field_count <= 300:
            return await self._reasoning_link(accessible_meta, question)
        elif accessible_meta.table_count <= 10000:
            return await self._rag_link(accessible_meta, question)
        else:
            return await self._agentic_link(accessible_meta, question)
    else:
        # 手动模式：使用指定策略
        return await self._strategy_map[config.at_schema_linking_mode](
            accessible_meta, question
        )

实体检索与融合

问题改写

将用户问题改写为多个子查询和关键词：

async def _rewrite_question(self, question: Question) -> None:
    response = await prompt_generate(
        "query.extract_keywords_from_question",
        QUESTION=question.text,
        SPECIFICATION=question.specification,
        EVIDENCE=question.evidence,
    )
    question.keywords = response["keywords"]
    question.subqueries = response["subqueries"]

示例：

• •原始问题：「上个月销售额最高的前 10 个产品是什么？」
• •子查询：

  ["销售额", "产品", "上个月", "前 10"]

• •关键词：

  ["销售额", "产品"]

实体融合

将字段、值、示例三路检索结果融合：

def _merge_values_fields(hits: list[RetrievedMetaEntity]) -> list[MetaEntity]:
    fields_buckets: dict[tuple, set] = {}

    # 按 (schema, table, field) 分组
    for hit in hits:
        index = (
            hit["payload"]["schema_name"],
            hit["payload"]["table_name"],
            hit["payload"]["field_name"],
        )
        if not fields_buckets.get(index):
            fields_buckets[index] = set()

        # 收集示例值
        if hit["payload"]["type"] == "value":
            fields_buckets[index].add(hit["payload"]["value"])

    # 构建实体列表
    fields_list = []
    for index, values in fields_buckets.items():
        fields_list.append({
            "schema_name": index[0],
            "table_name": index[1],
            "field_name": index[2],
            "sample_values": list(values),
        })

    return fields_list

上下文增强

将示例值添加到字段描述中：

def _add_context_to_meta(meta: MetaAdmin, entities: list[MetaEntity]):
    for entity in entities:
        if schema := meta.schemas.get(entity["schema_name"]):
            if table := schema.tables.get(entity["table_name"]):
                if field := table.fields.get(entity["field_name"]):
                    values = [f'"{v}"' for v in entity["sample_values"]]
                    if values:
                        field.curr_desc += f"(e.g. {','.join(values)})"

效果：

• •原始描述：

  region: 地区

• •增强后：

  region: 地区 (e.g. "华东", "华北", "华南")

性能优化

1. 并行检索

三路检索并行执行，减少延迟：

# 并行执行
fields, values, pairs = await asyncio.gather(
    self._retrieve_fields(question.subqueries),
    self._retrieve_values(question.keywords),
    self._retrieve_examples(question.specification)
)

2. 缓存策略

• •元数据缓存：可访问元数据缓存 5 分钟
• •检索结果缓存：相同问题的检索结果缓存 1 分钟
• •LLM 响应缓存：表筛选结果缓存 10 分钟

3. 检索优化

• •向量索引：使用 Qdrant 高性能向量检索
• •混合检索：向量检索 + 全文检索
• •重排序：使用 LLM 对检索结果重排序

实际效果

在不同规模数据库上的性能表现：

总结

AskTable 的自适应 Schema Linking 系统通过四种策略的组合，实现了从小型到超大型数据库的全覆盖：

1. •Naive 策略：简单直接，适合小型数据库
2. •Reasoning 策略：LLM 推理，适合中型数据库
3. •RAG 策略：向量检索 + LLM 重排序，适合大型数据库
4. •Agentic 策略：多轮交互，适合超大型数据库

通过自动策略选择、三路实体检索融合、上下文增强等技术，系统在保证准确率的同时，实现了高效的 Schema Linking。

相关资源

• •AskTable Schema Linking 文档
• •RAG 技术详解