一：核心概念与痛点分析

1.1. 结构化数据 vs 非结构化数据

核心痛点

1. 向量检索无法精确聚合 - "过去 30 天销售额"无法通过向量计算
2. Text-to-SQL 缺乏语义理解 - "推荐适合的产品" SQL 无法理解
3. 单一方案覆盖不全 - 同时需要精确计算和语义理解
4. 性能与成本平衡难 - 准确率、速度、成本的三角矛盾

1. 结构化数据与语义向量空间不匹配

大多数 RAG 实现依赖向量检索技术（embedding + 向量数据库）对文本进行相似度搜索。这对非结构化文本有效，但对表格 / 结构化数据效果差强人意：

• 表格数据的行列结构、数值字段、层次关系等语义在向量空间中往往被“扁平化”，导致 embedding 丢失结构信息。
• 在把表格拆为“自然语言段落”再向量化时，会丢失列与行之间的本质关系，使检索结果难以准确匹配真实结构语义。
• 向量模型并不擅长捕捉关系型数据的精确筛选条件，如数值范围、日期范围和条件逻辑（比如 SQL WHERE 子句，"过去 30 天销售额"）。

这类问题直接影响检索质量与后续生成的准确性，是结构化数据 RAG 最大的难点之一。

2. 查询语义对结构化数据的不匹配

用户往往以自然语言提问（例如 “过去三年中欧盟 vs 美国赢得的法律案件数对比？”），而结构化数据一般需要通过 SQL 或逻辑分析表达：

• RAG 传统检索不具备关系型查询能力，难以准确把语言查询转换为结构化查询（如 SQL）。
• 即使结合 LLM 做自然语言到 SQL 的转换，也存在生成错误、表名 / 字段拼写错误或语义误解等问题。
• 多表 join、聚合函数等复杂操作，对 LLM 推理要求更高且容易出现错误。

这会导致检索结果不准确或可执行查询失败。

3. 数据规模与实时性压力

企业结构化数据量通常很大，并且更新频繁：

• 实时或近实时更新数据仓库 / 湖难以及时同步到向量数据库或传统检索库。
• 向量数据库在大规模数据下需要高性能索引（如 HNSW / IVF），但这些不一定对结构化条件过滤友好。
• 查询性能和延迟是实时交互场景中的重要指标，但大数据规模会压制检索速度，造成用户体验不佳。

这就要求复杂的数据管道和高性能基础设施。

4. 复杂数据预处理与特征提取成本高

为让结构化数据适配 RAG，需要很多额外处理：

• 提取 schema 信息（表 / 列元数据）并构造描述性文本或 embeddings。
• 用额外 dual-encoder、cross-encoder 进行表格 reranking。
• 为数值字段构建 binning 或专门 embedding。
• 模式混合的表格、分类编码、缺失值处理等。

这些预处理步骤往往需要 ML 专业知识与工程投入。

下表总结结构化数据在 RAG 过程中的关键难点：

在企业 RAG 落地中，结构化数据之所以成为“硬骨头”，是因为它试图用概率性的模型（LLM / 向量）去解决确定性的逻辑问题（SQL / 计算）。

核心结论与建议：

• 不要试图全部向量化：对于强结构化数据，Text-to-SQL 或 Text-to-Pandas 仍是首选，但必须配合人工精修的 Schema 描述（给 LLM 看的数据库字典）。
• 采用“路由架构”（Router）：在 RAG 前端增加意图识别。如果是定性问题（“某产品的评价如何”），走向量检索；如果是定量问题（“某产品库存多少”），走 SQL 查询。
• 权限前置：不要依赖向量库做权限，而是在检索前就通过元数据过滤掉用户无权访问的数据范围。

结构化数据 RAG 的关键不在“模型能力”，而在“系统设计能力”。

• LLM ≠ 数据库
• RAG ≠ 万能查询引擎
• 企业级系统必须：
  • 控制数据边界
  • 显式建模语义
  • 解耦推理与执行

二：环境准备

# 安装依赖
pip install -q llama-index-core llama-index-llms-openai llama-index-embeddings-openai llama-index-llms-dashscope llama-index-embeddings-dashscope
pip install -q llama-index-vector-stores-milvus llama-index-retrievers-bm25 rank-bm25 llama-parse
pip install -q sqlalchemy pandas python-dotenv 

print("✅ 依赖安装完成")

# 全局配置
import os
import warnings
from dotenv import load_dotenv
from llama_index.llms.dashscope import DashScope
from llama_index.embeddings.dashscope import DashScopeEmbedding
from llama_index.core.settings import Settings

warnings.filterwarnings('ignore')
# 加载环境变量
load_dotenv(override=True)

# 设置 LLM
Settings.llm = DashScope(
    model_name="qwen-max",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    api_base=os.getenv("DASHSCOPE_BASE_URL", "https://api.dashscope.aliyuncs.com")
)

# 设置 Embedding模型
Settings.embed_model = DashScopeEmbedding(
    model_name="text-embedding-v4",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    api_base=os.getenv("DASHSCOPE_BASE_URL", "https://api.dashscope.aliyuncs.com")
)

# 基本使用
response = Settings.llm.complete("你好，请介绍一下通义千问。")
print(response)

response_embed = Settings.embed_model.get_query_embedding("你好，请介绍一下通义千问。")
print(response_embed)

2.1. Markdown 文档中的表格检索方式

1. 数据解析 (Parsing Phase)

• 目标：把“混沌”的 Markdown 文档拆解成此“条理分明”的结构化节点。
• 输入：原始 Markdown 文件（包含文本、标题、表格）。
• 工具：MarkdownElementNodeParser。
• 动作：
  • 扫描：扫描全文，识别 |---|---| 格式的表格。
  • 提取表格 (Object Extraction)：
    • 将识别到的表格从原文中“抠”出来，转换成 TableNode（本质是 DataFrame 结构，或者保留原始 Markdown 表格文本）。
    • 这部分内容不直接切片进入向量库，因为它包含了复杂的结构，直接切片会破坏语义。
  • 生成摘要 (Summary Generation)：
    • 关键一步：Parser 会调用 LLM（如 GPT-3.5），给每个提取出来的表格生成一段自然语言摘要（Summary）。
    • 例子：LLM 生成“这是一个关于公司各部门 2023 年 Q3 财务数据的表格，包含研发、销售部的收入和成本。”
  • 构建基座节点 (Base Nodes)：
    • 将剩余的普通正文切片成 TextNode。
    • 创建一个 IndexNode，内容是上面的表格摘要，并带有一个指针（index_id）指向那个完整的 TableNode。

import os
import pandas as pd
from llama_index.core import Document, VectorStoreIndex
from llama_index.core.node_parser import MarkdownElementNodeParser
from llama_index.core.retrievers import RecursiveRetriever
from llama_index.llms.openai import OpenAI
from llama_index.core.schema import TextNode, IndexNode # 确保导入这两个类

# 0. 准备测试数据 (确保表格格式标准)
markdown_content = """
# 公司季度财报
以下是我们在2023年第三季度的详细财务表现:
| 部门 | 收入(万元) | 成本(万元) | 利润(万元) |
|------|------------|------------|------------|
| 研发部 | 0 | 500 | -500 |
| 销售部 | 2000 | 800 | 1200 |
| 产品部 | 500 | 200 | 300 |
总结来看,销售部表现最为亮眼。
"""
doc = Document(text=markdown_content)

# 1. 创建解析器
node_parser = MarkdownElementNodeParser(
    llm=Settings.llm, 
    num_workers=1   # 并行解析数
)

# 2. 正确的解析流程
# 第一步: 从文档获取原始节点
raw_nodes = node_parser.get_nodes_from_documents([doc])

# 第二步: 从节点中提取基座节点和表格对象
base_nodes, objects = node_parser.get_nodes_and_objects(raw_nodes)
print(f"解析完成: 生成了 {len(base_nodes)} 个基座节点, {len(objects)} 个表格对象")

# 3. 验证表格对象
print(f"\nobjects 类型: {type(objects)}")

if objects:
    for i, obj in enumerate(objects):
        print(f"  对象 {i}:")
        print(f"    类型: {type(obj).__name__}")
        print(f"    ID: {obj.node_id if hasattr(obj, 'node_id') else 'N/A'}")
        # 如果是表格节点,尝试显示内容
        if hasattr(obj, 'text'):
            print(f"    内容预览: {obj.text[:100]}...")
        if hasattr(obj, 'metadata'):
            print(f"    元数据: {obj.metadata}")
else:
    print("\n⚠️ 未提取到表格,调试信息:")
    for i, node in enumerate(raw_nodes):
        print(f"  节点 {i}: {type(node).__name__} - {node.text[:60]}...")

2. 检索与生成 (Retrieval & Generation Phase)

• 目标：用户提问，系统精准找到表格并回答。
• 工具：RecursiveRetriever (递归检索器)。
• 流程：
  • 第一级检索 (Level-1 Retrieval)：
    • 用户问：“销售部门去年的利润怎么样？”
    • 检索器在 Vector Store 中搜索。
    • 由于表格摘要里有“...销售部的收入和成本...”，语义匹配度很高，检索器找回了这个 IndexNode（表格摘要）。
  • 递归解析 (Recursive Step)：
    • RecursiveRetriever 拿到 IndexNode 后，发现它手里捏着一个 index_id。
    • 它根据这个 ID 去 DocStore (objects) 里抓取完整的表格对象。
    • 现在，检索器手里拿到了完整的 Markdown 表格或 DataFrame。
  • 上下文组装 (Synthesis)：
    • 如果检索到了其他普通 TextNode（正文），也一并带上。
    • 最终的 Prompt 变成了：
      [上下文]
      1. (正文片段) 公司 Q3 战略重点是...
      2. (表格内容) | 部门 | 收入 | ... | 销售部 | 2000 | ...
         [问题]
         销售部门去年的利润怎么样？
  • LLM 生成：
    • LLM 看到完整的表格数据，轻松计算或提取出“1200 万元”，生成最终回答。

# 4. 过滤 base_nodes (只保留 TextNode 和 IndexNode)
base_nodes_filtered = [n for n in raw_nodes if isinstance(n, (TextNode, IndexNode))]
print(f"\n过滤后基座节点数: {len(base_nodes_filtered)}")

# 5. 建立向量索引 (只对摘要建索引)
vector_index = VectorStoreIndex(nodes=base_nodes_filtered)
vector_retriever = vector_index.as_retriever(similarity_top_k=2)

# 6. 构建递归检索器
# 将 objects 列表转换为字典 {node_id: node}
objects_dict = {obj.node_id: obj for obj in objects}

recursive_retriever = RecursiveRetriever(
    "vector",
    retriever_dict={"vector": vector_retriever},
    node_dict=objects_dict,  # 传入表格对象字典
    verbose=True  # 开启日志，可以看到递归过程
)

# 7. 执行检索测试
query = "销售部的利润是多少？"

retrieved_nodes = recursive_retriever.retrieve(query)

for i, node_with_score in enumerate(retrieved_nodes):
    print(f"\n--- 结果 {i+1} (相关度: {node_with_score.score:.3f}) ---")
    print(f"节点类型: {type(node_with_score.node).__name__}")
    print(f"内容:\n{node_with_score.node.text[:300]}...")
    
# 8. 结合 LLM 生成最终答案
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core import Settings

query_engine = RetrieverQueryEngine.from_args(
    recursive_retriever,
    llm=Settings.llm
)

response = query_engine.query(query)
print(f"\n问题: {query}")
print(f"\n答案: {response}")
print(f"\n来源节点数: {len(response.source_nodes)}")

2.2. PDF 文档中表格解析检索

• 核心架构：LlamaParse + MarkdownElementNodeParser
• 实现思路
  • PDF → Markdown 转换：使用 LlamaParse（LlamaIndex 官方的 PDF 解析工具）将 PDF 转换为结构化的 Markdown 格式，表格会被精准保留为 Markdown 表格
  • Markdown → 节点拆分：使用 MarkdownElementNodeParser 提取表格并生成摘要
  • 递归检索：与之前 Markdown 案例完全一致

重要说明

1. LlamaParse API Key 需要在 https://cloud.llamaindex.ai 注册账号 免费版有每日解析页数限制(通常 1000 页 / 天) 将 API Key 替换代码中的 "llx-..."
2. 成本优化 LlamaParse 是付费服务，如果文档较大建议先测试几页，可以通过 pages=[1, 2, 3] 参数只解析指定页
3. 备选方案 (如果不想用 LlamaParse) 使用开源工具如 pdfplumber 或 camelot 提取表格，然后手工转换为 Markdown 但这种方式对复杂表格(跨行跨列)的处理效果通常不如 LlamaParse

import os
from llama_index.core import Document, VectorStoreIndex, Settings
from llama_index.core.node_parser import MarkdownElementNodeParser
from llama_index.core.retrievers import RecursiveRetriever
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.schema import TextNode, IndexNode
from llama_parse import LlamaParse
from dotenv import load_dotenv

load_dotenv(override=True)

# ========== 步骤 1: 使用 LlamaParse 解析 PDF ==========
parser = LlamaParse(
    result_type="markdown",  # 输出格式为 Markdown (保留表格结构)
    verbose=True,
    language="ch_sim",  # 简体中文
    # 可选: 提供自定义的解析指令
    # parsing_instruction="重点提取财务数据表格，保留表头和数值"
)

# 解析 PDF 文件
pdf_path = "dataset/招股意向书.pdf"
# 添加错误处理
try:
    documents = parser.load_data(pdf_path)
    
    # 检查是否为空
    if not documents:
        print("⚠️ 警告: 解析结果为空!")
        print("可能的原因:")
        print("1. PDF 文件路径不正确")
        print("2. LlamaParse API 调用失败")
        print("3. API Key 未正确配置")
        print(f"\n请检查:")
        print(f"- 文件是否存在: {os.path.exists(pdf_path)}")
        print(f"- API Key 是否已设置: {bool(os.getenv('LLAMA_CLOUD_API_KEY'))}")
    else:
        print(f"预览第一个片段:\n{documents[0].text[:300]}...\n")
        
except Exception as e:
    print(f"❌ PDF 解析失败: {e}")
    print("\n调试信息:")
    print(f"- 文件路径: {pdf_path}")
    print(f"- 文件存在: {os.path.exists(pdf_path)}")
    print(f"- API Key 已设置: {bool(os.getenv('LLAMA_CLOUD_API_KEY'))}")
    raise

# ========== 步骤 2: 使用 MarkdownElementNodeParser 提取表格 ==========
node_parser = MarkdownElementNodeParser(
    llm=Settings.llm,  # 用于生成表格摘要
    num_workers=4      # 并行处理
)

# 从文档获取原始节点
raw_nodes = node_parser.get_nodes_from_documents(documents)

# 提取基座节点和表格对象
base_nodes, objects = node_parser.get_nodes_and_objects(raw_nodes)

# 调试:查看提取到的表格
if objects:
    for i, obj in enumerate(objects[:3]):  # 只显示前 3 个
        print(f"  表格 {i+1}:")
        print(f"    ID: {obj.node_id}")
        print(f"    内容预览: {obj.text[:100]}...\n")

# ========== 步骤 3: 构建索引和递归检索器 ==========
# 过滤节点(只保留 TextNode 和 IndexNode)
base_nodes_filtered = [n for n in base_nodes if isinstance(n, (TextNode, IndexNode))]

# 建立向量索引
vector_index = VectorStoreIndex(nodes=base_nodes_filtered)
vector_retriever = vector_index.as_retriever(similarity_top_k=5)

# 构建 objects 字典
objects_dict = {obj.node_id: obj for obj in objects}

# 构建递归检索器
recursive_retriever = RecursiveRetriever(
    "vector",
    retriever_dict={"vector": vector_retriever},
    node_dict=objects_dict,
    verbose=True
)

# 构建查询引擎
query_engine = RetrieverQueryEngine.from_args(
    recursive_retriever,
    llm=Settings.llm
)

# ========== 步骤 4: 测试查询 ==========
# 示例查询(根据招股意向书的实际内容调整)
test_queries = [
    "体外诊断项目中，“输血项目”的检测范围和用途是什么?",
]
for query in test_queries:
    print(f"\n问题: {query}")
    response = query_engine.query(query)
    print(f"答案: {response}\n")
    print("-" * 60)

三：PandasQueryEngine - 小数据量高效方案

在处理小于 1 万行的结构化数据时，PandasQueryEngine 是最直接高效的解决方案：

• ✅ 零索引成本: 无需向量化，直接操作内存中的 DataFrame
• ✅ 精准计算: LLM 生成 Pandas 代码，计算 100% 准确
• ✅ 快速迭代: 适合探索性数据分析
• ⚠️ 适用场景: 数据量 < 1 万行，单表或少量表

3.1. 入门：基础查询

import pandas as pd
import time
from typing import Dict, Any, List
from llama_index.experimental.query_engine import PandasQueryEngine
from llama_index.llms.openai import OpenAI
from llama_index.core import Settings

import logging
# 禁用 Phoenix/OpenInference 的错误日志，在后续有提到的监控工具
logging.getLogger("openinference.instrumentation.llama_index").setLevel(logging.CRITICAL)

# 加载产品数据
df = pd.read_csv("dataset/products.csv")
df.head()

数据预览:

1. 创建 PandasQueryEngine

PandasQueryEngine 的核心参数:

# 优化的指令字符串 - 关键配置
instruction_str = (
    "1. 将查询转换为可执行的 Pandas 代码 \n"
    "2. 代码必须是单行表达式，可以被 eval() 执行 \n"
    "3. DataFrame 变量名为 'df' \n"
    "4. 只输出表达式本身，不要有其他文字 \n"
    "5. 不要使用引号包裹表达式 \n"
    "\n 示例: \n"
    "问题: 找出价格最高的产品 \n"
    "代码: df.loc[df['price'].idxmax()]['name'] \n"
)

# 创建查询引擎
query_engine = PandasQueryEngine(
    df=df,   # DataFrame 读取的数据
    verbose=True,  # 显示生成的代码
    instruction_str=instruction_str,
    synthesize_response=True  # 综合回答
)

# 场景 1: 简单聚合查询
query = "电子产品类别有多少个产品？"
print(f"📝 查询: {query}")

response = query_engine.query(query)
print(f"💬 回答: {response}") # 💬 回答: 根据查询结果，电子产品类别中共有 6 种不同的产品。

# 场景 2: 排序和筛选
query = "列出价格(price)最高的5个产品，显示名称和价格"
print(f"\n📝 查询: {query}")

response = query_engine.query(query)
print(f"💬 回答: {response}")

3.2. 进阶：复杂查询与性能监控

在生产环境中需要:

• 🔄 错误重试: 处理 LLM 偶发错误
• ⏱️ 性能监控: 跟踪查询耗时
• 📊 统计分析: 了解查询成功率

class AdvancedPandasQueryEngine:
    """增强的 Pandas 查询引擎 - 带错误处理和性能监控"""
    def __init__(self, df: pd.DataFrame, llm_model: str = "gpt-4o-mini"):
        self.df = df
        self.llm = OpenAI(model=llm_model, temperature=0)
        
        # 优化的指令（中文提示效果更好）
        instruction_str = (
            "你是数据分析专家，将自然语言转换为 Pandas 代码。\n"
            "要求: \n"
            "1. 生成单行可 eval() 执行的表达式 \n"
            "2. DataFrame 变量名为 'df' \n"
            "3. 只输出代码，无其他文字 \n"
            "4. 优先使用向量化操作 \n"
        )
        
        self.query_engine = PandasQueryEngine(
            df=df, 
            verbose=True,
            instruction_str=instruction_str,
            synthesize_response=True
        )
        self.query_history = []
    
    def query(self, query_str: str, max_retries: int = 2) -> Dict[str, Any]:
        """执行查询，带重试和性能记录"""
        start_time = time.time()
        
        for attempt in range(max_retries + 1):
            try:
                response = self.query_engine.query(query_str)
                execution_time = time.time() - start_time
                
                self.query_history.append({
                    "query": query_str,
                    "time": execution_time,
                    "success": True
                })
                
                return {"success": True, "result": str(response), "time": execution_time}
                
            except Exception as e:
                if attempt == max_retries:
                    self.query_history.append({
                        "query": query_str,
                        "time": time.time() - start_time,
                        "success": False,
                        "error": str(e)
                    })
                    return {"success": False, "error": str(e)}
    
    def get_stats(self) -> Dict[str, Any]:
        """获取性能统计"""
        if not self.query_history:
            return {"message": "没有查询历史"}
        
        successful = [q for q in self.query_history if q["success"]]
        return {
            "total": len(self.query_history),
            "success": len(successful),
            "success_rate": f"{len(successful)/len(self.query_history)*100:.1f}%",
            "avg_time": f"{sum(q['time'] for q in successful)/len(successful):.2f}s" if successful else "N/A"
        }

# 初始化高级引擎
advanced_engine = AdvancedPandasQueryEngine(df)

# 进阶查询示例
advanced_queries = [
    "计算每个品牌(brand)的产品数量，按数量降序排列",
    "找出评分(rating)高于 4.5 的产品名称和价格",
    "显示各类别(category)的平均价格"
]

for query in advanced_queries:
    result = advanced_engine.query(query)
    if result["success"]:
        print(f"💬 回答: {result['result']}")
        print(f"⏱️  耗时: {result['time']:.2f}s")
    else:
        print(f"❌ 错误: {result['error']}")

# 显示统计
stats = advanced_engine.get_stats()
for k, v in stats.items():
    print(f"   {k}: {v}")

3.3. 高阶：Chain of Tables（表格思维链）

普通的 Text-to-SQL（或 Text-to-Pandas）往往试图用一句超级复杂的 SQL / 代码解决所有问题，这很容易出错（LLM 手滑写错代码）。对于复杂的多步骤分析，使用表格思维链技术:

原始表格 → [筛选] → 中间表1 → [聚合] → 中间表2 → [比较] → 最终结果

核心思想:

1. 分解复杂问题为简单步骤
2. 逐步执行每个操作
3. 保留中间结果便于调试
4. 记录操作历史可追溯

• 核心实现：

  • 这个实现是一个人机协同的 Chain-of-Table：人（或上层 Agent）：负责规划思维链（High-level reasoning），决定先 filter 再 aggregate。
  • Python 类：负责维护中间状态（State Management），传递 current_df。
  • LLM：负责执行具体的原子操作（Atomic Execution），即把自然语言转成 Pandas 代码。

class ChainOfTablesEngine:
    """表格思维链引擎 - 多步骤推理（修正版）"""
    def __init__(self, df: pd.DataFrame):
        self.original_df = df.copy()    # 保存原始数据
        self.current_df = df.copy()    # 当前工作表
        self.operation_history = []    # 操作历史
        self.llm = Settings.llm        # LLM
    
    def filter(self, condition: str) -> pd.DataFrame:
        """筛选操作 - 真正更新 DataFrame"""
        rows_before = len(self.current_df)
        
        # 构造 prompt，让 LLM 生成筛选代码
        prompt = f"""
        你是一位 Pandas 专家。用户想要筛选数据。
        当前 DataFrame 的列名是: {list(self.current_df.columns)}
        用户的筛选条件是: {condition}

        请生成一行 Python 代码来完成筛选，要求：
        1. 变量名必须是 'df'
        2. 返回筛选后的 DataFrame
        3. 只输出代码，不要有任何解释

        示例: df[df['price'] > 5000]
        """
        
        # 调用 LLM 生成代码
        response = self.llm.complete(prompt)
        generated_code = response.text.strip()
        
        # 清理代码（去掉可能的 markdown 标记）
        if generated_code.startswith("```"):
            generated_code = generated_code.split("\n")[1]
        if generated_code.endswith("```"):
            generated_code = generated_code.rsplit("\n", 1)[0]
        generated_code = generated_code.strip()
        
        # 执行代码并更新 current_df
        try:
            df = self.current_df  # 为了代码能找到 df 变量
            filtered_df = eval(generated_code)   # 执行代码
            self.current_df = filtered_df    # 更新当前工作表
            print(f"✅ 筛选成功: {rows_before} 行 → {len(self.current_df)} 行")
        except Exception as e:
            print(f"❌ 代码执行失败: {e}")
            print(f"保持原数据不变")
        
        # 记录操作
        self.operation_history.append({
            'step': len(self.operation_history) + 1,
            'operation': 'FILTER',
            'description': condition,
            'code': generated_code,
            'rows': f"{rows_before} → {len(self.current_df)}"
        })
        
        return self.current_df
    
    def aggregate(self, aggregation: str) -> str:
        """聚合操作 - 基于当前表计算"""
        # 构造 prompt
        prompt = f"""
        你是一位 Pandas 专家。
        当前 DataFrame 有 {len(self.current_df)} 行数据，列名是: {list(self.current_df.columns)}
        用户想要: {aggregation}

        请生成一行 Python 代码来完成计算，要求：
        4. 变量名必须是 'df'
        5. 返回计算结果（数值或 Series）
        6. 只输出代码，不要解释

        示例: df['rating'].mean()
        """
        
        response = self.llm.complete(prompt)   # 调用 LLM 生成代码
        generated_code = response.text.strip()  # 去除前后的空格
        
        # 清理代码
        if generated_code.startswith("```"):
            generated_code = generated_code.split("\n")[1]
        if generated_code.endswith("```"):
            generated_code = generated_code.rsplit("\n", 1)[0]
        generated_code = generated_code.strip()
        
        # 执行计算
        try:
            df = self.current_df
            result = eval(generated_code)
            print(f"✅ 计算结果: {result}")
        except Exception as e:
            result = f"计算失败: {e}"
            print(f"❌ {result}")
        
        # 记录操作
        self.operation_history.append({
            'step': len(self.operation_history) + 1,
            'operation': 'AGGREGATE',
            'description': aggregation,
            'code': generated_code,
            'result': str(result)[:100]
        })
        
        return str(result)
    
    def reset(self):
        """重置到原始表"""
        self.current_df = self.original_df.copy()
        self.operation_history = []
    
    def show_history(self):
        """显示操作历史"""
        for op in self.operation_history:
            print(f"  步骤 {op['step']}: {op['operation']}")
            print(f"    描述: {op['description']}")
            print(f"    代码: {op.get('code', 'N/A')}")
            if 'rows' in op:
                print(f"    行数变化: {op['rows']}")
            if 'result' in op:
                print(f"    结果: {op['result']}")
            print()

# 重新初始化
cot_engine = ChainOfTablesEngine(df)

# 场景: 分析高端产品
print("\n📌 任务: 找出高价产品的平均评分")
print("思维链: 筛选(价格>5000) → 聚合(平均评分)")

# 步骤 1: 筛选
cot_engine.filter("价格 (price) 大于 5000")
# 步骤 2: 聚合
result = cot_engine.aggregate("计算评分 (rating) 的平均值")
print(f"\n🎯 最终结果: {result}")

# 显示完整历史
cot_engine.show_history()

# 重置并执行另一个分析
print("📌 任务: 各类别产品数量排名")
print("思维链: 分组 → 计数 → 排序")

cot_engine.reset()
result = cot_engine.aggregate("按类别 (category) 分组，计算每个类别的产品数量，按降序排列")
cot_engine.show_history()

# 可以通过 matplotlib 来对生成的 Pandas 数据框进行可视化
import matplotlib.pyplot as plt
import platform

# 1. 解决中文显示问题
system_name = platform.system()
if system_name == 'Darwin':  # macOS
    plt.rcParams['font.sans-serif'] = ['Arial Unicode MS']
elif system_name == 'Windows': # Windows
    plt.rcParams['font.sans-serif'] = ['SimHei']
else: # Linux
    plt.rcParams['font.sans-serif'] = ['WenQuanYi Micro Hei']
plt.rcParams['axes.unicode_minus'] = False 

# 2. 定义标题
title = '各类别产品数量分布'

# 3. 绘图
plt.figure(figsize=(10, 6))
ax = query_result.plot(kind='bar', color='steelblue')
plt.title(title)
plt.xlabel('类别')
plt.ylabel('数量')
plt.xticks(rotation=45, ha='right')

# 4. 添加数值标签
for p in ax.patches:
    ax.annotate(str(p.get_height()), (p.get_x() + p.get_width() / 2., p.get_height()),
                ha='center', va='center', xytext=(0, 5), textcoords='offset points')

plt.tight_layout()
plt.show()

3.4. PandasQueryEngine 总结

1. 适用场景对比

2. 优化技巧

1. 指令工程: 提供清晰的 instruction_str 指导 LLM
2. 错误处理: 实现重试机制应对 LLM 偶发错误
3. 性能监控: 记录查询耗时和成功率
4. 复杂问题: 使用 Chain of Tables 分解为简单步骤

3. 局限性

• ⚠️ 大数据量时内存开销大
• ⚠️ 复杂查询可能生成错误代码
• ⚠️ 不支持跨表 JOIN（需预处理）

四：向量检索 + BM25 基准

4.1. 方案原理

在处理结构化数据（如数据库表、Excel、CSV）的 RAG 场景中，我们将每一行数据视为一个独立的文档进行索引。然而，结构化数据具有其独特性：它既包含需要语义理解的非结构化字段（如“商品描述”、“用户评论”），也包含需要精确匹配的结构化字段（如“SKU 编码”、“具体参数”、“年份”）。面对这种混合特征，向量检索 + BM25 的组合拳是目前业界在语义检索方面的最佳实践方案。

1. 核心原理：行数据的双路召回

该方案首先将结构化表格的每一行数据进行序列化（Serialization），转化为包含字段名和字段值的文本段落（例如将 {'Name': 'iPhone 15', 'Price': 7999} 转换为 "Name: iPhone 15, Price: 7999"）。基于转换后的文本，系统同时维护两套索引：

• 向量检索（Vector Retrieval）：负责“懂意”。 它将行文本映射到高维语义空间。当用户询问“适合移动办公的轻薄本”时，即使表格中没有“轻薄”二字，只写了“重量 1.1 kg”，向量模型也能捕捉到两者间的语义关联，从而召回相关记录。它弥补了传统数据库无法进行模糊语义查询的短板。
• BM25 检索（Keyword Retrieval）：负责“精准”。 它基于概率统计模型（TF-IDF 的改进版），对稀有且明确的关键词极其敏感。在结构化数据中，往往存在大量的专有名词（如“RTX 4090”、“ISO 9001”）。向量模型容易忽略这些低频词的精确差异，而 BM25 能确保这些核心参数必须出现在召回结果中，充当了“模糊 SQL”的硬过滤器。

2. 方案 1：表格行级向量化

# 示例：向量检索实现
from llama_index.core import VectorStoreIndex, Document
import pandas as pd
from sqlalchemy import create_engine

# 数据库连接
db_url = "sqlite:///dataset/ecommerce.db"
engine = create_engine(db_url)

# 加载数据
df = pd.read_sql("SELECT * FROM products LIMIT 15", engine)

# 将每行转为文本
documents = []
for _, row in df.iterrows():
    text = f"""
    产品名称: {row.get('name', row.get('product_name', 'Unknown'))}
    类别: {row.get('category', 'N/A')}
    价格: {row.get('price', row.get('unit_price', 0))}元
    """
    documents.append(Document(text=text, metadata=row.to_dict()))

# 创建向量索引
vector_index = VectorStoreIndex.from_documents(
    documents,
    insert_batch_size=10  # ✅ DashScope 同步接口限制：每批最多 10 条
)
vector_retriever = vector_index.as_retriever(similarity_top_k=3)

3. 方案 2：分块向量化（推荐用于大规模数据）

# 示例：分块向量化实现
from llama_index.core import VectorStoreIndex, Document
import pandas as pd
from sqlalchemy import create_engine

# 数据库连接
db_url = "sqlite:///dataset/ecommerce.db"
engine = create_engine(db_url)

# 加载数据
df = pd.read_sql("SELECT * FROM orders", engine)

# ========== 分块参数 ==========
CHUNK_SIZE = 10  # 每个分块包含的行数

# 将多行合并为一个文档（分块向量化）
documents = []
for chunk_start in range(0, len(df), CHUNK_SIZE):
    # 取出当前分块的数据
    chunk_df = df.iloc[chunk_start:chunk_start + CHUNK_SIZE]
    
    # 将分块内所有行拼接为文本
    text_parts = []
    for _, row in chunk_df.iterrows():
        item = f"产品id: {row.get('product_id')} | 用户id: {row.get('user_id')} | 数量: {row.get('quantity')} | 订单日期: {row.get('order_date')} | 订单金额: {row.get('amount')}元 | 订单状态: {row.get('status')} "
        text_parts.append(item)
    
    # 合并为单个文档
    chunk_text = "\n".join(text_parts)
    
    # 元数据记录分块信息
    metadata = {
        "chunk_id": chunk_start // CHUNK_SIZE,
        "row_range": f"{chunk_start}-{min(chunk_start + CHUNK_SIZE, len(df))}",
        "status": list(chunk_df['status'].unique())  # 分块包含的状态
    }
    
    documents.append(Document(text=chunk_text, metadata=metadata))

print(f"生成 {len(documents)} 个分块文档 (每块 {CHUNK_SIZE} 行)")

# 创建向量索引
vector_index = VectorStoreIndex.from_documents(
    documents,
    insert_batch_size=10  # DashScope 限制
)
vector_retriever = vector_index.as_retriever(similarity_top_k=3)

4. 方案 3：混合索引（最灵活）

4.2. Milvus 向量数据库存储 + BM25 检索

"""
Milvus 混合向量化示例

核心优势:
Milvus 原生支持元数据（标量字段）过滤
性能优异，适合大规模生产环境
支持本地 Lite 模式（./milvus.db）和服务器模式
"""
from typing import Optional, Dict, Any, List, Tuple, Union
import pandas as pd
import numpy as np
from sqlalchemy import create_engine
from tqdm import tqdm
from llama_index.core import (
    VectorStoreIndex,
    Document,
    StorageContext,
    Settings
)
from llama_index.core.vector_stores import (
    MetadataFilters,
    MetadataFilter,
    FilterOperator
)
from llama_index.vector_stores.milvus import MilvusVectorStore
from llama_index.embeddings.dashscope import DashScopeEmbedding
import os
from dotenv import load_dotenv
from sqlalchemy import inspect

# 加载环境变量
load_dotenv()

class UniversalHybridVectorizer:
    """
    通用混合索引 - 使用 Milvus 支持元数据过滤
    
    核心功能:
    1. 支持 CSV / Excel / 数据库多种数据源
    2. 数据库自动推断字段：利用 sqlalchemy.inspect 扫描表结构，自动将字段归类：
            长文本（Text / LongText）：自动划为 text_fields，用于生成向量，提供语义搜索能力。
            数值 / 日期（Int / Float / Date）：自动划为 metadata_fields，用于范围查询（如“价格 > 5000”）。
            短文本（Tags / Category）：自动划为 metadata_fields，用于精确匹配（如“品牌=Apple”）。
    3. 使用 Milvus 实现高性能向量检索 + 标量过滤
    """
    def __init__(
        self, 
        db_url: Optional[str] = None,
        table: Optional[str] = None,
        # ===== 可选：手动覆盖自动推断 =====
        text_fields: Optional[List[str]] = None,
        metadata_fields: Optional[List[str]] = None,
        id_field: Optional[str] = None,
        numeric_fields: Optional[List[str]] = None,
        categorical_fields: Optional[List[str]] = None
    ):
        """
        初始化向量化器
        方式 1（推荐）：传入数据库连接，自动推断字段
            vectorizer = UniversalHybridVectorizer(
                db_url="sqlite:///ecommerce.db",
                table="products"
            )
        方式 2：手动指定所有字段（用于 CSV / Excel或需要精确控制）
            vectorizer = UniversalHybridVectorizer(
                text_fields=["name", "description"],
                metadata_fields=["price", "category"]
            )
        Args:
            db_url: 数据库连接字符串（可选）
            table: 表名，用于自动推断字段（可选）
            text_fields: 需要向量化的字段列表
            metadata_fields: 作为元数据的字段列表
            id_field: 唯一标识符字段名
            numeric_fields: 数值字段列表（用于范围查询）
            categorical_fields: 类别字段列表（用于精确 / IN 查询）
        """
        self.engine = None
        self.table = table
        
        # 如果提供了数据库连接，创建引擎
        if db_url:
            self.engine = create_engine(db_url)
        
        # 自动推断字段（如果提供了 db_url 和 table 且未手动指定字段）
        if self.engine and table and not text_fields:
            auto_fields = self._auto_detect_fields(table)
            self.text_fields = text_fields or auto_fields.get('text', [])
            self.metadata_fields = metadata_fields or auto_fields.get('metadata', [])
            self.id_field = id_field or auto_fields.get('id')
            self.numeric_fields = numeric_fields or auto_fields.get('numeric', [])
            self.categorical_fields = categorical_fields or auto_fields.get('categorical', [])
            
            print(f"📊 自动推断字段配置:")
            print(f"   文本字段: {self.text_fields}")
            print(f"   数值字段: {self.numeric_fields}")
            print(f"   分类字段: {self.categorical_fields}")
            print(f"   ID 字段: {self.id_field}")
        else:
            # 使用手动指定的字段
            self.text_fields = text_fields or []
            self.metadata_fields = metadata_fields or []
            self.id_field = id_field
            self.numeric_fields = numeric_fields or []
            self.categorical_fields = categorical_fields or []
        
        self.index = None
    
    def _auto_detect_fields(self, table: str) -> Dict[str, Any]:
        """
        从数据库表结构自动推断字段分类
        
        分类规则：
        - TEXT, LONGTEXT, VARCHAR(255+) → text_fields
        - INTEGER, FLOAT, DECIMAL → numeric_fields + metadata_fields
        - VARCHAR(<255) → categorical_fields + metadata_fields
        - PRIMARY KEY → id_field
        """
        # 获取表结构
        inspector = inspect(self.engine)
        # 获取表的列信息
        columns = inspector.get_columns(table)
        # 获取主键约束
        pk_constraint = inspector.get_pk_constraint(table)
        result = {
            'text': [],
            'metadata': [],
            'numeric': [],
            'categorical': [],
            'id': None
        }
        
        # 获取主键
        if pk_constraint and pk_constraint.get('constrained_columns'):
            result['id'] = pk_constraint['constrained_columns'][0]
        
        for col in columns:
            col_name = col['name']
            col_type = str(col['type']).upper()
            
            # 跳过主键（已单独处理）
            if col_name == result['id']:
                continue
            
            # 根据类型分类
            if 'TEXT' in col_type or 'LONGTEXT' in col_type:
                result['text'].append(col_name)
            elif 'VARCHAR' in col_type:
                # 解析 VARCHAR 长度
                import re
                match = re.search(r'VARCHAR\((\d+)\)', col_type)
                length = int(match.group(1)) if match else 50
                
                if length >= 255:
                    result['text'].append(col_name)
                else:
                    result['categorical'].append(col_name)
                    result['metadata'].append(col_name)
            elif any(t in col_type for t in ['INT', 'FLOAT', 'DECIMAL', 'NUMERIC', 'REAL']):
                result['numeric'].append(col_name)
                result['metadata'].append(col_name)
            elif any(t in col_type for t in ['DATE', 'TIME', 'DATETIME', 'TIMESTAMP']):
                result['metadata'].append(col_name)
            else:
                # 其他类型默认加入 metadata
                result['metadata'].append(col_name)
        
        return result
    
    def load_from_database(
        self,
        table: Optional[str] = None,
        query: Optional[str] = None,
        limit: Optional[int] = None,
        chunk_size: int = 1000
    ) -> List[Document]:
        """
        从数据库加载数据
        Args:
            table: 表名（默认使用初始化时的表）
            query: 自定义 SQL 查询（支持 JOIN，优先于 table）
            limit: 限制返回行数
            chunk_size: 分批读取的行数
        Examples:
            # 简单用法：加载初始化时指定的表
            docs = vectorizer.load_from_database()
            # 指定表和限制
            docs = vectorizer.load_from_database(table="products", limit=1000)
            # 自定义 SQL（支持 JOIN）
            docs = vectorizer.load_from_database(
                query=\"\"\"
                    SELECT p.*, r.content as review_content 
                    FROM products p 
                    LEFT JOIN reviews r ON p.id = r.product_id
                \"\"\"
            )
        """
        if not self.engine:
            raise ValueError("未配置数据库连接，请在初始化时传入 db_url")
        
        # 确定要使用的表
        target_table = table or self.table
        
        # 构建SQL
        if query:
            sql = query
            # 检测查询中的新列并尝试自动推断
            self._detect_new_columns_from_query(query)
        elif target_table:
            sql = f"SELECT * FROM {target_table}"
            if limit:
                sql += f" LIMIT {limit}"
            print(f"🗄️  加载表: {target_table}" + (f" (LIMIT {limit})" if limit else ""))
        else:
            raise ValueError("必须指定 table 或 query")
        
        # 分批读取
        all_documents = []
        chunks = pd.read_sql(sql, self.engine, chunksize=chunk_size)
        
        for df_chunk in tqdm(chunks, desc="处理数据库数据"):
            documents = self._convert_to_hybrid_documents(df_chunk)
            all_documents.extend(documents)
        
        print(f"✅ 生成 {len(all_documents)} 个文档")
        return all_documents
    
    def _convert_to_hybrid_documents(self, df: pd.DataFrame) -> List[Document]:
        """
        “结构化”转“非结构化”的关键中间层，转换为混合文档
        策略：
        1. 文本化（Text Parts）：将所有 text_fields 的内容拼接成一段自然语言描述（例如："name: iPhone 15 \n description: 最新款苹果手机..."），
            这部分会被 Embedding 模型转化为向量。
        2. 元数据化（Metadata）：将所有 metadata_fields（如价格、类别、库存）保留为原始键值对（Key-Value），这部分会被存入 Milvus 的标量字段中，
            用于后续的 Filter 操作。
        这实现了**“语义在向量里，属性在字段里”**的双重存储结构。
        """
        documents = []
        
        for idx, row in df.iterrows():
            # 1. 构建向量化文本
            text_parts = []
            for field in self.text_fields:
                if field in row and pd.notna(row[field]):
                    text_parts.append(f"{field}: {row[field]}")
            
            text = "\n".join(text_parts) if text_parts else "N/A"
            
            # 2. 构建元数据
            metadata = {}
            for field in self.metadata_fields:
                if field in row and pd.notna(row[field]):
                    value = row[field]
                    # 处理 pandas 特殊类型
                    if isinstance(value, (pd.Timestamp, pd.Timedelta)):
                        value = str(value)
                    metadata[field] = value
            
            # 3. 确定文档 ID
            if self.id_field and self.id_field in row:
                doc_id = str(row[self.id_field])
            else:
                doc_id = str(idx)
            
            # 4. 创建文档
            doc = Document(
                text=text,
                metadata=metadata,
                doc_id=doc_id
            )
            documents.append(doc)
        
        return documents
    
    def build_index(self, documents: List[Document], 
                      save_path: str = "./hybrid_index",
                      collection_name: str = "hybrid_collection",
                      milvus_uri: str = "./milvus_demo.db"):
        """
        构建混合索引（使用 Milvus 支持元数据过滤）
        Args:
            documents: 文档列表
            save_path: 索引保存路径（兼容参数，实际数据存储在 Milvus）
            collection_name: Milvus 集合名称
            milvus_uri: Milvus 连接 URI
                - 本地模式: "./milvus_demo.db" (使用 Milvus Lite)
                - 服务器模式: "http://localhost:19530"
        """
        from llama_index.vector_stores.milvus import MilvusVectorStore
        
        # 使用 Milvus 构建向量存储（支持元数据过滤！）
        # dim=1024 对应 DashScope text-embedding-v4 的向量维度
        vector_store = MilvusVectorStore(
            uri=milvus_uri,            # Milvus 连接 URI
            collection_name=collection_name,  # Milvus 集合名称
            dim=1024,          # 向量维度，最好跟 Embedding 模型长度一致
            overwrite=True  # 如果存在同名 collection 则覆盖
        )
        
        # 使用 MilvusVectorStore 构建存储上下文
        storage_context = StorageContext.from_defaults(
            vector_store=vector_store
        )
        
        # 使用 VectorStoreIndex 构建向量索引
        self.index = VectorStoreIndex.from_documents(
            documents,
            storage_context=storage_context,  # 使用 MilvusVectorStore
            show_progress=True,   # 显示进度
            insert_batch_size=10  # ✅ DashScope 同步接口限制：每批最多 10 条
        )
        
        # Milvus 数据是自托管的,不需要 persist 到文件系统
        print(f"✅ 索引已创建: collection={collection_name}, uri={milvus_uri}")
        
        vector_retriever = self.index.as_retriever(similarity_top_k=5)
        return vector_retriever

# 1. 创建向量化器
vectorizer = UniversalHybridVectorizer(
        db_url="sqlite:///dataset/ecommerce.db",
        table="products"
    )
# 自动打印：文本字段: ['description'], 数值字段: ['price', 'stock']...

# 2. 加载数据
docs = vectorizer.load_from_database()  # 使用初始化时的表

# 3. 构建 Milvus 检索器
vector_retriever = vectorizer.build_index(
    docs, 
    collection_name="products",
    milvus_uri="http://localhost:19530"
)

1. BM25 检索器

在本项目中，BM25 检索器的实现承担着能够“精确匹配关键词”的重要角色。其实现逻辑由底层的核心算法与上层的过滤增强两个层面构成，采用了一种典型的“后过滤”（Post-Filtering）架构。

核心算法原理（Standard BM25） LlamaIndex 的 BM25Retriever 基于经典的概率检索模型（通常封装 rank_bm25 库）。它在内存中工作，主要流程如下：

• 倒排索引构建：启动时，系统会将所有文档节点（Nodes）的文本进行分词，并建立词到文档的映射关系。
• 相关性评分：当用户查询时，算法根据词频（TF）和逆文档频率（IDF）计算文档得分。
  • 它会奖励稀有词的匹配（例如 “RTX4090” 的权重远高于“电脑”）。
  • 它会对文档长度进行归一化惩罚，防止长文档仅因包含更多词汇而获得不公平的高分。

目前主流 BM25 检索器框架对比

from llama_index.core.schema import TextNode
from llama_index.retrievers.bm25 import BM25Retriever

import pandas as pd
from sqlalchemy import create_engine

# 数据库连接
db_url = "sqlite:///dataset/ecommerce.db"
engine = create_engine(db_url)

# 加载 products 表数据
df = pd.read_sql("SELECT * FROM products", engine)
print(f"总数据量: {len(df)}行")

# 直接创建 Nodes（跳过 Documents）
nodes = []
for idx, row in df.iterrows():
    # 将所有字段都添加到 text 中
    text = f"""产品 ID: {row.get('product_id')}
            产品名称: {row.get('name')}
            类别: {row.get('category')}
            价格: {row.get('price')} 元
            库存: {row.get('stock')} 件
            评分: {row.get('rating')}
            描述: {row.get('description')}"""

    node = TextNode(
        text=text,
        metadata=row.to_dict(),
        id_=f"product_{idx}"
    )
    nodes.append(node)

# 创建 BM25 检索器
bm25_retriever = BM25Retriever.from_defaults(
    nodes=nodes,
    similarity_top_k=5
)

2. 创建 QueryFusionRetriever 混合检索器

from llama_index.core.retrievers import QueryFusionRetriever
from llama_index.core.query_engine import RetrieverQueryEngine

# 1. 创建 QueryFusionRetriever 混合检索器
retriever = QueryFusionRetriever(
    retrievers=[vector_retriever, bm25_retriever],  # 集成不同的检索器
    similarity_top_k=5,                 # 最终返回的文档数量
    num_queries=1,                      # 为原始查询生成的变体数量，（默认为 1，不生成变体）
    mode="reciprocal_rerank",           # 结果融合模式，'reciprocal_rerank' 是常用且效果好的模式
    use_async=False,                     # 是否异步执行，Jupyter Notebook 环境做测试不用开启异步
    verbose=True,                       # 是否打印调试信息
    retriever_weights=[0.8,0.2],        # 检索器权重，用于加权融合
)

# 2.构建 QueryEngine，查询引擎
retriever_query_engine = RetrieverQueryEngine.from_args(
    retriever=retriever
)

# 3. 构建测试问题
test_queries = [
    "适合办公的笔记本 (价格 5000 - 15000 元)",
    "适合办公的电子产品",    # 语义查询
    "价格在 1000 元以内的产品",  # 数值范围查询,
    "所有产品的平均价格是多少？", # 聚合查询
]

for query in test_queries:
    response = retriever_query_engine.query(query)
    print(f"回答: {response}")
    print("-" * 80)

3. 实现 rerank 重排序检索

from llama_index.core.postprocessor import SentenceTransformerRerank
from llama_index.core.query_engine import RetrieverQueryEngine

# 模型本地路径
local_model_path = "./models/bge-reranker-v2-m3"

# 1. 定义精排序器 (Reranker)
reranker = SentenceTransformerRerank(
    model=local_model_path,   # 指定模型路径
    top_n=3, # 精选出最相关的 4 个文档送给 LLM,
)

# 2.构建 QueryEngine
query_engine = RetrieverQueryEngine.from_args(
    retriever=retriever,
    node_postprocessors=[reranker]  # 添加精排序器
)

# 3. 执行查询
query = "请帮我查询适合办公的笔记本 (价格 5000 - 15000 元)有哪些？请把产品名称和价格都展示出来"
response = query_engine.query(query)

性能评估

测试结果 (基于 2 个测试查询):

核心问题:

• ❌ 无法精确筛选数值范围
• ❌ 无法进行聚合计算
• ❌ 无法排序和 Top-K 查询
• ✅ 语义理解能力较强

4.3. Attu Milvus 官方 GUI 管理工具

五：Text-to-SQL 基础

5.1. 方案原理

Text-to-SQL 是一种将自然语言自动转换为结构化查询语言（SQL）的技术方案。在处理结构化数据的 RAG 系统中，它是解决“精确统计”与“复杂逻辑计算”问题的终极武器，其本质是将 LLM 从“阅读者”升级为“操作者”。

1. 核心原理：语义到代码的精准翻译

不同于向量检索基于“语义相似度”的模糊匹配模式，Text-to-SQL 采用的是逻辑推理 + 代码生成模式。其核心在于利用 LLM 强大的理解力，将由人类发出的模糊指令（如“统计上季度销售额增长超过 20% 的部门”）翻译成数据库能够严格执行的标准 SQL 语句。在这个过程中，LLM 充当了一个高智商的中间层，它不仅要理解用户的意图，还要理解数据库的表结构（Schema）和表之间的关联关系（Relationship）。

用户查询 → LLM 生成 SQL → 数据库执行 → 结果 → LLM 生成答案

大概流程如下

2. 实现思路与关键流程

一个成熟的 Text-to-SQL 工作流通常包含以下四个核心步骤：

• 上下文注入 (Context Injection)：系统首先提取数据库的元数据（表名、字段名、字段类型、备注等），将其转化为 Prompt 的一部分。这一步至关重要，它相当于告诉 LLM “你面前有哪些数据表可以使用”。
• SQL 生成 (SQL Generation)：LLM 结合用户问题和表结构信息，推理并生成相应的 SQL 语句（例如 SELECT department, SUM(sales) ... GROUP BY ...）。高级实现还会包含 “SQL 语法检查器”，确保生成的代码没有语法错误。
• 沙箱执行 (Safe Execution)：生成的 SQL 被送入数据库引擎执行。为了安全起见，通常会限制执行权限（如仅允许 SELECT 操作），防止恶意删除或修改数据。
• 结果解读 (Answer Synthesis)：数据库返回的通常是行数据或统计数值（如 [(Marketing, 150000), (Sales, 200000)]）。最后一步是将这些冷冰冰的数据再次喂给 LLM，让其生成通俗易懂的自然语言回答（如“市场部和销售部表现优异...”）。

3. 核心优势与场景

Text-to-SQL 的最大价值在于计算的精确性。向量检索无法准确回答“所有产品的平均价格是多少”或“库存少于 10 的产品有哪些”这类问题，而 SQL 引擎天生就是为此设计的。因此，该方案特别适用于财务报表分析、业务数据统计、库存管理查询等对数值精度和聚合逻辑有严格要求的场景。

Text-to-SQL 主流工具对比分析:

5.2. 创建测试数据库

我们创建一个电商数据库，包含:

• products 表: 15 个产品
• users 表: 10 个用户
• orders 表: 100 个订单
• reviews 表: 50 条评论

数据保存在 ecommerce.db 文件中，可以自行查看

# 创建电商数据库的完整代码会非常长
# 这里提供核心创建逻辑，完整版见项目文件
import sqlite3
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import random
from pathlib import Path
from sqlalchemy import create_engine, text
from llama_index.core import SQLDatabase, Settings

# 使用已有的测试数据库作为示例
db_path = "../large_data_test.db"
# 数据库文件路径
db_path = Path("dataset/ecommerce.db")

if not db_path.exists():
    print("❌ 数据库文件不存在！")
    print("请先运行: python init_database.py")
else:
    # 创建 SQLAlchemy 引擎
    engine = create_engine(f"sqlite:///{db_path}")
    # 创建 LlamaIndex SQLDatabase 对象
    sql_database = SQLDatabase(engine)
    print(f"   数据库路径: {db_path.absolute()}")
print("✅ 数据库创建完成")

1. 查看示例数据

# 数据库相关
import sqlite3
# 工具库
import pandas as pd
from rich import print as rprint
from rich.console import Console
from rich.table import Table

console = Console()

# 查看用户表前 5 条数据
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM users LIMIT 5"))
    df_users = pd.DataFrame(result.fetchall(), columns=result.keys())

display(df_users)

# 查看产品表前 5 条数据
with engine.connect() as conn:
    result = conn.execute(text("SELECT * FROM products LIMIT 5"))
    df_products = pd.DataFrame(result.fetchall(), columns=result.keys())

display(df_products)

2. 创建 SQL 查询引擎

# Text-to-SQL 实现
from llama_index.core import SQLDatabase
from llama_index.core.query_engine import NLSQLTableQueryEngine
from sqlalchemy import create_engine

# 创建 SQL 数据库连接
engine = create_engine("sqlite:///dataset/ecommerce.db")
sql_database = SQLDatabase(engine, include_tables=["products", "orders", "users","reviews"])

# 创建 SQL 查询引擎
sql_query_engine = NLSQLTableQueryEngine(
    sql_database=sql_database,
    synthesize_response=True
)

3. 查看数据库表结构

# 获取所有表名
table_names = sql_database.get_usable_table_names()

for table in table_names:
    console.print(f"  • {table}")

# 查看每个表的结构
for table in table_names:
    console.print(f"\n[yellow]{table}[/yellow]:")
    schema = sql_database.get_single_table_info(table)
    console.print(schema)

5.3. 性能评估

测试结果:

核心发现:

• ✅ 数值查询大幅提升（25% → 85%）
• ✅ 聚合统计几乎完美（95%）
• ❌ 语义理解能力下降（88% → 40%），在第三个问题中没有根据语义信息进行回答
• ⚠️ SQL 生成错误率约 15-20%

六：Text-to-SQL 优化

6.1. 优化技术概览

我们将应用 4 种优化技术:

1. Schema 增强 - 添加表和字段的详细描述
2. Few-shot 示例 - 提供高质量查询示例
3. SQL 验证 - 自动检测和修复错误
4. 查询缓存 - 减少重复 LLM 调用

预期提升: 75% → 92%

6.2. Schema 增强技术

1. 为什么需要 Schema 增强？

基础的表结构信息可能不足以让 LLM 准确理解数据含义。通过添加：

• 表的业务描述
• 字段的详细说明
• 字段之间的关系
• 示例值

可以显著提升 SQL 生成的准确性。

2. 创建增强的 Schema

from llama_index.core.objects import SQLTableSchema

# 为每个表创建增强的 Schema
table_schemas = [
    SQLTableSchema(
        table_name="users",
        context_str="""
        用户表，存储所有注册用户的基本信息。
        
        字段说明：
        - user_id: 用户唯一标识符
        - name: 用户姓名
        - email: 用户邮箱地址
        - region: 用户所在地区（如：北京、上海、广州等）
        - register_date: 用户注册日期
        - vip_level: VIP 等级（0=普通用户, 1-3=不同等级 VIP）
        - total_spent: 用户累计消费金额（单位：元）
        
        业务规则：
        - VIP 等级越高，享受的折扣越多
        - total_spent 会随着订单完成自动更新
        """
    ),
    SQLTableSchema(
        table_name="products",
        context_str="""
        产品表，存储所有可售卖的商品信息。
        
        字段说明：
        - product_id: 产品唯一标识符
        - name: 产品名称
        - category: 产品分类（电子产品、家用电器、服装鞋帽、家具等）
        - price: 产品价格（单位：元）
        - stock: 库存数量
        - rating: 产品评分（1-5 分，保留两位小数）
        - description: 产品详细描述
        
        业务规则：
        - rating 基于用户评论自动计算
        - stock 为 0 时产品不可购买
        """
    ),
    SQLTableSchema(
        table_name="orders",
        context_str="""
        订单表，记录所有用户的购买订单。
        
        字段说明：
        - order_id: 订单唯一标识符
        - user_id: 下单用户 ID（外键关联 users 表）
        - product_id: 购买产品 ID（外键关联 products 表）
        - quantity: 购买数量
        - amount: 订单总金额（单位：元）= price * quantity
        - order_date: 下单日期
        - status: 订单状态（已完成、配送中、已取消等）
        
        业务规则：
        - 只有状态为"已完成"的订单才计入销售额
        - amount 已包含所有折扣
        """
    ),
    SQLTableSchema(
        table_name="reviews",
        context_str="""
        评论表，存储用户对产品的评价。
        
        字段说明：
        - review_id: 评论唯一标识符
        - user_id: 评论用户 ID（外键关联 users 表）
        - product_id: 被评论产品 ID（外键关联 products 表）
        - rating: 评分（1-5 分整数）
        - comment: 评论内容
        - review_date: 评论日期
        
        业务规则：
        - 用户必须购买过产品才能评论
        - rating 影响产品的总体评分
        """
    )
]

3. 使用增强 Schema 创建查询引擎

# 创建带有增强 Schema 的查询引擎
enhanced_query_engine = NLSQLTableQueryEngine(
    sql_database=sql_database,      # SQL 数据库对象
    tables=["users", "products", "orders", "reviews"], # 指定可供查询的表列表
    table_schemas=table_schemas,    # 注入包含列描述的增强 Schema
    verbose=True,                   # 输出详细的 SQL 生成过程
    synthesize_response=True        # 将 SQL 执行结果转换为自然语言回复
)

4. 对比测试：基础 vs 增强

测试查询 1: 复杂业务查询

query = "哪些 VIP 用户的累计消费超过 10000 元？"

# 基础引擎
try:
    response_basic = sql_query_engine.query(query)
    console.print(f"回答: {response_basic}")
    console.print(f"SQL: {response_basic.metadata.get('sql_query', 'N/A')}")
except Exception as e:
    console.print(f"[red] 错误: {e}[/red]")

# 增强引擎
try:
    response_enhanced = enhanced_query_engine.query(query)
    console.print(f"回答: {response_enhanced}")
    console.print(f"SQL: {response_enhanced.metadata.get('sql_query', 'N/A')}")
except Exception as e:
    console.print(f"[red] 错误: {e}[/red]")

测试查询 2: 需要理解业务规则

query = "计算已完成订单的总销售额"

# 增强引擎（应该能正确理解"已完成"状态）
response = enhanced_query_engine.query(query)

6.3. Few-shot 示例优化

1. 什么是 Few-shot Learning？