由于现有数据库问答效果不佳，所以对优化方法进行了调研，记录一下。

当前问题：我当前的项目还存在数据库对话准确性非常低的情况。我的数据库很大，有2.7T。但是库表之间没有外键约束，表的属性命名也存在不规范。再无法改动数据库的情况下，我怎么优化数据库问答效果？

方案一

阶段 0：前置准备（非运行时）

1. 构建 Schema 知识图谱（KG）

   • 从 DDL 提取表、字段、主键、外键

   • 用 LLM 为每个字段生成自然语言语义描述（如 “SAMPLE_NOK_QTY = 不合格样本数量”）

   • 存入图数据库（如 Neo4j）或向量+结构化混合存储

⚠️ 这一步只需做一次，后续问答都复用此 KG。

---

阶段 1：运行时 —— 智能体推理循环（Single Agent with Reflection）

用户提问 → 启动一个 具备工具调用能力的智能体，执行以下步骤：

---

🔁 Step 1：【意图理解 + Schema 检索】

• 目标：从自然语言中识别业务概念（如“成品检测”、“不合格率”），并映射到数据库对象。

• 操作：

  • 调用 Schema Navigator Tool（连接知识图谱）

    • 输入：用户问题

    • 输出：相关表列表、关键字段、外键路径、语义解释

• 输出示例：

  相关表: QMS_PRODUCT_INSP (成品检测), QMS_PROCESS_TASK (质检任务)
  关键字段: SAMPLE_NOK_QTY, SAMPLE_SIZE, CHECK_TIME
  关系: QMS_PRODUCT_INSP.QMS_PRODUCT_TASK_ID → QMS_PROCESS_TASK.QMS_PROCESS_TASK_ID
  

✅ 这就是你所说的“抽取相关表的 agent”，但实际是 智能体调用的一个工具。

---

🔁 Step 2：【SQL 生成】

• 目标：基于检索到的 schema 上下文，生成可执行 SQL。

• 操作：

  • 将用户问题 + schema 上下文 拼接成 prompt

  • 调用 LLM（如 GPT-4、Claude 3.5、SQLCoder）生成 SQL

• 关键技巧：

  • 在 prompt 中强制要求：“只使用上述表和字段，不要假设不存在的列”

  • 输出格式约束：sql ...

✅ 这对应你所说的“根据表生成 SQL 的 agent”。

---

🔁 Step 3：【SQL 执行 + 结果验证】

• 目标：执行 SQL 并判断结果是否回答了原始问题。

• 操作：

  • 调用 Database Executor Tool（安全沙箱环境）

    • 执行 SQL，返回前 10 行 + 列名 + 行数

  • 调用 Answer Validator Sub-Agent（或 LLM 自评）：

    • Prompt: “用户问‘上月不合格率最高的产品’，当前 SQL 返回了 TASK_NO 和 fail_rate，是否满足需求？”

    • 输出：✅ 满足 / ❌ 不满足（原因：缺少产品名称、时间范围错误等）

✅ 这就是“执行 agent 后检查是否满足问题”。

---

🔁 Step 4：【反思与迭代（Reflection）】

• 如果验证失败：

  • 智能体进入 反思模式：

    • 分析失败原因（如“未关联产品主数据表”、“时间字段用错”）

    • 更新内部状态（如“需要额外获取 BASE_PART_ID”）

    • 回到 Step 1，重新调用 Schema Navigator，扩大检索范围

  • 最多重试 2~3 次，避免无限循环

✅ 这完全对应你所说的“如果不满足就反思，回到抽取步骤”。

---

🔁 Step 5：【自然语言总结】

• 目标：将 SQL 结果转化为用户友好的答案。

• 操作：

  • 调用 LLM，输入：

    • 原始问题

    • 执行结果（表格数据）

    • 相关业务上下文（如“不合格率 = NOK / 总样本”）

  • 生成简洁、准确、带单位的自然语言回答

  • 可选：附上数据来源（“数据来自 QMS_PRODUCT_INSP 表”）

✅ 这就是“最后总结回答的 agent”。

---

🧩 整体架构图（逻辑流）

用户问题
   │
   ▼
[智能体 Controller]
   │
   ├───▶ 调用 Schema Navigator Tool → 获取相关表/字段（基于 KG）
   │
   ├───▶ 生成 SQL（LLM + schema 上下文）
   │
   ├───▶ 调用 DB Executor → 获取结果
   │
   ├───▶ 调用 Validator → 判断是否满足问题？
   │           │
   │           ├─ 是 → 生成自然语言回答
   │           │
   │           └─ 否 → 反思原因 → 更新查询意图 → 回到 Schema Navigator
   │
   ▼
最终回答（含溯源）

💡 注意：这不是 5 个独立 agent，而是一个 带状态记忆和工具调用能力的单一智能体，通过 LangGraph 的 StateGraph 或 AutoGen 的 ConversableAgent + Tool 实现。

方案二

用户问题
    ↓
智能路由（分类问题类型）
    ↓
表关系推断引擎
    ↓
Schema增强模块
    ↓
多模型SQL生成（多个候选）
    ↓
SQL验证与执行
    ↓
结果合理性检查
    ↓
LLM生成自然语言回答

1. Schema信息增强策略

1.1 构建知识图谱辅助理解

python

# 示例：自动构建表关系图谱
def build_table_relationships():
    # 1. 基于列名相似性推断关系
    # 2. 基于数据值分布分析关联
    # 3. 基于查询日志发现表连接模式
    # 4. 创建虚拟外键映射表
    return {
        "table1": {
            "likely_related_to": ["table2", "table3"],
            "join_candidates": [
                {"column": "user_id", "related_table": "users", "confidence": 0.8}
            ]
        }
    }

1.2 智能Schema提取与增强

python

class SchemaEnhancer:
    def __init__(self):
        self.schema_cache = {}
        
    def enrich_schema(self, table_name):
        # 1. 获取基本结构
        schema = self.get_table_schema(table_name)
        
        # 2. 添加示例数据（采样）
        sample_data = self.get_sample_data(table_name, limit=3)
        schema["sample_data"] = sample_data
        
        # 3. 添加列统计信息
        column_stats = self.analyze_column_patterns(table_name)
        schema["column_patterns"] = column_stats
        
        # 4. 添加语义标签（通过大模型分析）
        semantic_tags = self.generate_semantic_tags(schema)
        schema["semantic_info"] = semantic_tags
        
        return schema

2. 查询优化策略

2.1 两阶段查询生成

python

def two_stage_sql_generation(user_query, relevant_tables):
    """
    第一阶段：生成查询计划
    第二阶段：生成具体SQL
    """
    
    # 阶段1：生成查询结构
    query_plan_prompt = f"""
    基于以下表结构，分析查询需求：
    用户问题：{user_query}
    相关表：{relevant_tables}
    
    请输出：
    1. 需要查询的主要字段
    2. 可能的表连接关系
    3. 过滤条件的关键词
    4. 排序和分组需求
    """
    
    query_plan = llm_generate(query_plan_prompt)
    
    # 阶段2：生成具体SQL
    sql_generation_prompt = f"""
    根据查询计划生成SQL：
    查询计划：{query_plan}
    表结构详情：{enhanced_schemas}
    
    注意：
    1. 如果没有明确外键，使用语义相似的列进行JOIN
    2. 处理命名不规范的情况
    3. 考虑使用子查询简化复杂逻辑
    """
    
    return llm_generate(sql_generation_prompt)

2.2 动态提示词优化

python

def build_optimized_prompt(user_query, schemas):
    prompt_template = """
    你是一个专业的SQL生成专家，特别擅长处理无外键约束、命名不规范的数据库。
    
    ## 数据库特点
    1. 没有外键约束，需要智能推断表关系
    2. 列名不规范（如：cust_id, customerID, cid可能都表示客户ID）
    3. 数据量大（2.7TB），需要考虑查询性能
    
    ## 用户问题
    {user_query}
    
    ## 相关表结构
    {schemas}
    
    ## 查询提示
    1. 先分析表之间的语义关系
    2. 使用模糊匹配处理列名不一致问题
    3. 优先使用INNER JOIN，必要时使用LEFT JOIN
    4. 考虑添加LIMIT限制结果集大小
    
    ## 生成规则
    请生成符合{db_type}语法的SQL
    """
    return prompt_template.format(...)

3. 数据采样与示例学习

3.1 构建查询-示例对

sql

-- 自动收集高频查询模式
SELECT 
    query_text,
    execution_count,
    tables_used
FROM query_history
WHERE success = true
ORDER BY execution_count DESC
LIMIT 1000;

3.2 在提示词中添加成功案例

python

def add_example_queries(prompt, similar_queries):
    examples_section = "\n## 类似成功案例\n"
    for example in similar_queries[:3]:
        examples_section += f"问题：{example['question']}\n"
        examples_section += f"SQL：{example['sql']}\n\n"
    
    return prompt + examples_section

4. 执行反馈与迭代优化

4.1 SQL验证与修正机制

python

class SQLValidator:
    def validate_and_correct(self, sql, user_query):
        try:
            # 尝试执行
            result = self.execute_sql(sql)
            
            # 检查结果合理性
            if self.is_result_suspicious(result):
                # 结果可疑，尝试修正
                corrected_sql = self.correct_sql(sql, result, user_query)
                return corrected_sql
                
            return sql
        except Exception as e:
            # 语法错误，让LLM修正
            correction_prompt = f"""
            SQL执行错误：{str(e)}
            原SQL：{sql}
            问题：{user_query}
            请修正SQL
            """
            return llm_generate(correction_prompt)

4.2 结果验证与重试

python

def execute_with_fallback(sql_attempts, user_query):
    """
    尝试多个SQL版本，选择最合理的结果
    """
    for sql in sql_attempts:
        try:
            result = execute_sql(sql)
            
            # 验证结果是否合理
            if validate_result(result, user_query):
                return result
        except:
            continue
    
    # 所有尝试失败，返回降级查询
    return execute_degraded_query(user_query)

零散的优化方法

这一部分比较散乱，想到哪写到哪
现在生成SQL不准确，很大一部分原因是数据库当初设计的时候没有按照规范，如果不做数据治理改变数据库的话只能治标不治本。
可以使用映射表，把数据库的表名和字段映射成模型更容易理解的形态，这是让模型理解数据库。
还可以建立专业术语表，把专业术语和数据库里面的表和字段对应起来，这是让模型理解用户问题。

这些都是在模型外层做的工作，如果数据库表多，字段多，那么解释表和字段的作用就会占据模型大部分注意力，让模型没办法去很好地解决问题。因此我建议将公司项目目前的给一个问题，然后给一段很长地数据库schema和解释信息，然后让QWEN-3抽取跟问题相关的表改为使用向量库匹配表。但是效果不一定会更好，只是会更快。或者结合起来，用向量库做初步筛选，然后用模型做细致的抽取，减轻模型的负担。

对模型可以使用正例负例进行微调，让模型学习到数据库的相关知识。

整体历程

要做数据库问答，大致就是一下几步：
1.自然语言转变为对应业务的问题
2.对应业务的问题转变为可执行的SQL语句
3.可执行的SQL语句执行后返回大模型生成自然语言的回答

难点主要在第1、2步。

第一步难点在于自然语言要跟业务对齐，比如我在实际生成中遇到的问题，2025年每个月完成了多少订单？
 

1. 日志里确实是两套完全不同的 SQL

第一次（大约 02:05:07）

系统最终喂给总结模型的 system 里这段 SQL：

text

执行sql：
SELECT DATE_FORMAT(MES_PP_PRODWO.PRODUCE_END_TIME, %Y-%m) AS month,
       BASE_PRODUCT_LINE_ID,
       COUNT(*) AS completed_orders
FROM MES_PP_PRODWO
JOIN MES_WO_BATCH
  ON MES_PP_PRODWO.MES_PP_PRODWO_ID = MES_WO_BATCH.MES_PP_PRODWO_ID
JOIN EAI_OP_LOT_HISTORY
  ON MES_PP_PRODWO.WO_NO = EAI_OP_LOT_HISTORY.wo_no
WHERE YEAR(MES_PP_PRODWO.PRODUCE_END_TIME) = 2025
  AND MES_PP_PRODWO.PRODUCE_END_TIME IS NOT NULL
  AND MES_WO_BATCH.JOB_END_TIME IS NOT NULL
  AND EAI_OP_LOT_HISTORY.end_time IS NOT NULL
GROUP BY month, BASE_PRODUCT_LINE_ID;

结果是三列：(月份, 产线, 订单数量)。

第二次（大约 02:06:49）

这次生成并执行的是另一条 SQL：

text

SELECT DATE_FORMAT(PRODUCE_END_TIME, '%Y-%m') AS month,
       SUM(COMPLETION_QTY) AS total_completed
FROM MES_PP_PRODWO
WHERE PRODUCE_END_TIME >= '2025-01-01'
  AND PRODUCE_END_TIME < '2026-01-01'
GROUP BY month
ORDER BY month;

结果是两列：(月份, 完成数量总和)，并且你在日志里可以看到当时只选中了这一个表：

text

2026-01-04 02:06:46,689 - log_pro - INFO - ['MES_PP_PRODWO']

这说明第二次只用 MES_PP_PRODWO 一张表，通过 COMPLETION_QTY 做聚合。

---

2. 这两条 SQL 的语义其实不一样

虽然你问的自然语言是同一句：

2025年每月各完成了多少订单

但两个 SQL 体现了两种不同“订单”定义：

• SQL 1（三表 JOIN + COUNT(*)）：

  • 把 MES_PP_PRODWO + MES_WO_BATCH + EAI_OP_LOT_HISTORY 联起来；
  • 每条符合条件的行就算一个“完成的订单”；
  • 月 + 产线维度统计（GROUP BY month, BASE_PRODUCT_LINE_ID）。

• SQL 2（单表 + SUM(COMPLETION_QTY)）：

  • 只看 MES_PP_PRODWO 里每个工单的 COMPLETION_QTY；
  • 把 2025-01-01 ~ 2025-12-31 所有完成数量加起来，按月汇总；
  • 没有产线维度，只有月份。

所以即使是同一句自然语言，“每月完成多少订单”也有多种合理的“翻译”：
按工单行数？按批次？按 COMPLETION_QTY 数量？按产线拆分还是整体？

你的系统目前没有固定一个唯一的业务口径，全靠模型根据表描述“理解”，再加上 LLM 本身有随机性，这就为“同问不同 SQL”埋了雷。

对应的解决方案有以下几种，但是只是方案还没实施，如果真的实施了我再来更新效果。
 

A. 最推荐：建立「业务语义层/指标字典」（自然语言→口径→SQL 模板/字段）

目标：把“订单数、完成量、维修成本、本周”等这些业务概念固化，不让模型每次自己猜。

做法示例（概念层面）：

• 指标：订单数
  • 定义：以 MES_PP_PRODWO 的 “工单条数” 还是以 COMPLETION_QTY 求和？
  • 维度：月、产线、工厂等
  • 时间字段：PRODUCE_END_TIME
  • 过滤：只取状态=已完成、时间非空等
• 把这些存成一份配置（JSON/YAML/表），查询时先把用户问题解析到“指标ID + 维度 + 时间范围”，再套模板生成 SQL。

优点：同问同答、可审计、业务可控。
缺点：前期需要业务/数据团队一起把口径梳理清楚。

---

B. 约束大模型：禁止“自由发挥口径”，只在已定义口径中选

即使你暂时不做模板化 SQL，也可以让模型先选口径再写 SQL：

1. 先让模型只输出结构化选择：

• metric: “订单数(口径A)” 或 “完成量(口径B)”
• time_range: 本月/本周/2025年
• dimensions: 月/产线

1. 再把选择结果 + 对应口径说明塞回第二次生成 SQL。

优点：改动相对小，能显著降低漂移。
缺点：仍依赖模型，但“随机空间”小很多。

---

C. 把随机性降到最低 + 加缓存：同样输入直接复用结果

你现在同问不同答，一个重要原因是 LLM 调用带随机性且每次都会重新走“选表→写SQL”。

可做：

• 关键步骤温度调低（尤其是“抽表名/生成 SQL”）
• 对 “(用户问题 + kbId + schema版本)” 做缓存：第一次生成的 SQL/相关表，后续直接复用

优点：最快落地，能立刻减少抖动。
缺点：口径没统一时，只是“稳定住某一次的答案”，不保证它是对的。

---

D. SQL 生成走“候选多条 → 自动验证选最好”的路线

让模型一次给 2~3 条不同口径/不同写法的 SQL，然后用规则自动挑：

• 优先选择：只用允许表、字段都存在、返回列名符合期望、结果行数合理、执行计划不爆炸
• 对聚合类（SUM/COUNT）强制 IFNULL，避免 None

优点：容错强，能减少“偶然写坏 SQL”。
缺点：执行成本更高（多跑几条 SQL）。

---

E. 把“表选择”变成确定性：别让模型凭感觉选表

你当前链路里模型会先总结任务再抽表名，这一步很容易漂。

改进方向：

• 用规则/检索（如向量/关键词）先给出 TopN 表，再让模型只能在这些表里选
• 或者对关键业务域（生产/维修/库存）写死“优先表集合”

优点：表稳定了，SQL 波动会小很多。
缺点：需要维护检索与白名单。

---

F. 明确“指标口径冲突时要问澄清问题”

对于确实有歧义的问题（比如“订单数”到底是工单数还是完成量），让系统返回澄清而不是硬猜：

• “你想看：工单数量（COUNT工单）还是完成数量（SUM完成量）？”
• “是否需要按产线拆分？”

优点：从源头避免错口径。
缺点：交互多一步，部分场景不适合。

---

G. 增加“口径测试集”和回归测试（工程化保障）

把常问问题沉淀成用例：

• 输入：自然语言问题
• 期望：使用的表/字段、SQL 结构特征、结果列名
• 每次改提示词/模型/表结构都跑一遍回归

优点：长期稳定性最好。
缺点：需要持续维护测试集。

第二步难点在于自然语言生成SQL语句。

要生成SQL语句，首先要让模型清楚你的数据库里有哪些表，表之间有哪些关系，表有哪些字段，每个表每个字段分别代表什么含义，对应业务上的哪个流程。
这里分为两个流派：
1.不动模型
单纯使用Schema来让模型理解数据库。这需要一个业务映射表，一个庞大的详细的描述数据库的Schema，如果数据库很大，这个流派就需要先做相关的表筛选，可以使用把表信息存入向量库，然后用问题和表做相似度匹配。也可以做关键词匹配，计算问题的关键词和表或者字段里描述有多少重合。

2.动模型

使用一些特化的小参数模型做微调，需要一个高质量的训练语料

第三步难点在于数据库资料庞大

这一步的难易程度是根据企业的应用场景决定。就比如我们的现有场景还没有遇到数据库执行后返回的文本超长的问题，但是如果是电商，需要客户的好评、差评、反馈意见，那就很容易出问题。等后面真的在这一步遇到问题了，我再来更新。