RAG 系统中 Excel/表格数据的正确处理方式

RAG 系统中 Excel/表格数据的正确处理方式：从向量检索到 Text-to-SQL 的架构演进

前言

在构建 RAG（检索增强生成）系统时，很多人会踩一个坑：把 Excel 表格和 PDF/DOCX 文档一视同仁，全部丢给 MinerU 解析成 Markdown，再切片向量化存入 Milvus。

这样做会导致一系列严重问题：

• ❌ 无法做聚合操作（SUM、AVG、COUNT）
• ❌ 无法排序、比较、筛选
• ❌ 浪费 LLM Token 去"理解"纯数字文本
• ❌ 查询精度极低，"产品A 3月销售额"这种精确问题答不上来

本文将完整梳理这个问题的根因分析、方案对比、架构设计，以及最终的落地方案。

一、问题复现：Excel 走 RAG 链路的灾难

1.1 当前链路

Excel 文件 → MinerU 解析 → Markdown（HTML <table>）→ 切片 → Embedding → Milvus 向量存储

1.2 一个典型的失败案例

用户上传了一份2024年Q1销售报表.xlsx：

用户提问：“Q1 各产品总销售额排名？”

RAG 系统的回答：语义检索返回了"可能相关"的文本片段，LLM 试图从模糊的 chunk 中拼凑答案——结果要么答错，要么回答"知识库中未找到相关内容"。

1.3 根因分析

本质原因：Excel 是结构化数据，应该走 SQL 路径，而不是 RAG 语义检索路径。

结论：不是所有数据都适合向量化。结构化数据需要结构化的查询方式。

二、方案对比：临时表 vs 永久表

2.1 方案 A：临时表 + Text-to-SQL

上传 Excel → 存储原始文件 → 提取文件名/表头 ↓ 用户提问 → LLM 生成 SQL → 创建临时表 → 导入数据 → 执行 SQL → 返回结果

优点：

• 不污染数据库，数据随用随弃

致命缺点：

2.2 方案 B：永久数据库表（✅ 推荐）

上传 Excel → 解析 Schema → CREATE TABLE（永久）→ 批量 INSERT → 建立索引 ↓ 用户提问 → LLM Text-to-SQL → 直接查询永久表 → 返回结果

优点：

需要注意的问题：

2.3 对比总结

三、架构设计：与 RAG 知识库隔离

3.1 为什么必须隔离

核心原则：RAG 管语义，SQL 管数据，Agent 负责路由判断。

3.2 整体架构图

┌──────────────────────────────────────────────────────────────┐ │ 用户上传文件 │ └──────────────────┬───────────────────────────────────────────┘ │ ┌──────┴──────┐ │ 文件类型判断 │ └──────┬──────┘ │ ┌─────────────┼─────────────┐ ▼ ▼ PDF/DOCX/PPTX/图片 XLSX/XLS/CSV │ │ ▼ ▼ ┌──────────┐ ┌──────────────┐ │ module_rag│ │ module_data │ │ │ │ │ │ MinerU │ │ 解析 Schema │ │ ↓ │ │ ↓ │ │ 切片 │ │ CREATE TABLE │ │ ↓ │ │ ↓ │ │ Embedding│ │ 批量 INSERT │ │ ↓ │ │ ↓ │ │ Milvus │ │ 存储元数据 │ └──────────┘ └──────────────┘ │ │ ▼ ▼ ┌──────────────────────────────────────────┐ │ Agent (FunctionAgent) │ │ │ │ ┌────────────────┐ ┌────────────────┐ │ │ │ knowledge_search│ │ data_query │ │ │ │ Tool │ │ Tool │ │ │ └───────┬────────┘ └───────┬────────┘ │ │ │ │ │ │ 语义类问题 数据类问题 │ │ "退货政策是什么" "销售额前10排名" │ └──────────────────────────────────────────┘

3.3 模块划分

module_data/ ← 新建：结构化数据管理模块 ├── controller/ │ └── data_source_controller.py ← Excel 上传/管理 API ├── service/ │ ├── data_source_service.py ← 业务逻辑（解析、建表、导入） │ └── text_to_sql_service.py ← Text-to-SQL 核心 ├── dao/ │ └── data_source_dao.py ← 数据源 CRUD ├── entity/ │ ├── do/ │ │ └── data_source_do.py ← 数据源表 ORM 模型 │ └── vo/ │ └── data_source_vo.py ← 请求/响应模型 └── tools/ └── sql_tool.py ← 给 Agent 用的 SQL Tool module_agent/ ← 现有：Agent 模块（改造） ├── tools/ │ ├── rag_tool.py ← 现有：知识检索 Tool │ └── (sql_tool.py 从 module_data 导入) ├── service/ │ ├── agent_factory.py ← 改造：支持注入多个 Tool │ └── agent_service.py ← 改造：多 Tool 协同

四、核心实现设计

4.1 数据源注册表

CREATETABLEdata_source(idVARCHARPRIMARYKEY,-- UUIDfile_nameVARCHARNOTNULL,-- 原始文件名file_pathVARCHARNOTNULL,-- 文件存储路径table_nameVARCHARNOTNULL,-- 实际 PG 表名 (excel_{uuid})sheet_nameVARCHAR,-- 原始 Sheet 名column_info JSONBNOTNULL,-- 列信息table_descriptionTEXT,-- LLM 生成的表描述row_countINTEGER,-- 数据行数source_typeVARCHARNOTNULL,-- 'excel' | 'csv' | 'pdf_table'source_doc_idVARCHAR,-- 来源文档 ID（PDF 提取的表格）create_byVARCHAR,create_timeTIMESTAMPDEFAULTNOW(),del_flagCHAR(1)DEFAULT'0');

column_info示例：

[{"name":"产品","pg_type":"TEXT","nullable":false,"sample":"产品A","description":"产品名称"},{"name":"1月","pg_type":"NUMERIC","nullable":false,"sample":"12500","description":"1月销售额（元）"},{"name":"2月","pg_type":"NUMERIC","nullable":false,"sample":"15800","description":"2月销售额（元）"}]

4.2 动态建表的安全约束

importreimportuuid TABLE_PREFIX="excel_"def_sanitize_table_name(doc_id:str)->str:"""生成安全的表名：固定前缀 + UUID，杜绝 SQL 注入"""safe_id=re.sub(r'[^a-zA-Z0-9]','',doc_id)returnf"{TABLE_PREFIX}{safe_id}"def_sanitize_column_name(name:str)->str:"""列名清洗：保留中文、字母、数字、下划线"""cleaned=re.sub(r'[^\w\u4e00-\u9fff]','_',str(name).strip())returncleanedor'unnamed_column'# SQL 安全校验ALLOWED_SQL_KEYWORDS={'SELECT','FROM','WHERE','GROUP BY','ORDER BY','HAVING','LIMIT','OFFSET','JOIN','ON','AS','AND','OR','NOT','IN','BETWEEN','LIKE','IS','NULL','ASC','DESC','COUNT','SUM','AVG','MAX','MIN'}defvalidate_sql(sql:str)->bool:"""只允许 SELECT 查询，禁止任何写操作"""upper_sql=sql.upper().strip()ifnotupper_sql.startswith('SELECT'):raiseValueError('只允许 SELECT 查询')forbidden={'INSERT','UPDATE','DELETE','DROP','ALTER','CREATE','TRUNCATE','EXEC','EXECUTE','GRANT','REVOKE'}forkeywordinforbidden:ifre.search(rf'\b{keyword}\b',upper_sql):raiseValueError(f'禁止的 SQL 关键字:{keyword}')# 强制追加 LIMIT（防止大结果集）if'LIMIT'notinupper_sql:sql=sql.rstrip(';')+' LIMIT 1000'returnTrue

4.3 Schema 推断

importpandasaspddefinfer_schema(file_path:str,sheet_name:str)->list[dict]:"""读取 Excel 前 100 行，推断每列的数据类型"""df=pd.read_excel(file_path,sheet_name=sheet_name,nrows=100)columns=[]forcolindf.columns:pg_type=_pandas_dtype_to_pg_type(df[col].dtype)sample=str(df[col].dropna().iloc[0])ifnotdf[col].dropna().emptyelseNonecolumns.append({"name":_sanitize_column_name(col),"original_name":str(col),"pg_type":pg_type,"nullable":bool(df[col].isnull().any()),"sample":sample,"description":"",# 后续由 LLM 补充描述})returncolumnsdef_pandas_dtype_to_pg_type(dtype)->str:"""pandas 数据类型 → PostgreSQL 类型映射"""importnumpyasnpifnp.issubdtype(dtype,np.integer):return'BIGINT'elifnp.issubdtype(dtype,np.floating):return'NUMERIC'elifnp.issubdtype(dtype,np.datetime64):return'TIMESTAMP'else:return'TEXT'

4.4 SQL Tool（给 Agent 用）

fromllama_index.core.toolsimportFunctionTooldefcreate_sql_tool(data_source_ids:list[str]|None=None)->FunctionTool:""" 创建 SQL 查询工具，供 Agent 查询结构化数据。 Agent 用自然语言描述问题，系统自动： 1. 获取相关数据源的 Schema 2. 构建 Text-to-SQL Prompt 3. LLM 生成 SQL 4. 安全校验 + 执行 5. 返回格式化结果 """asyncdefexecute_data_query(query:str)->str:"""对用户上传的 Excel/CSV 数据执行查询。 当用户的问题涉及数据统计、排名、对比、聚合（如求和、平均、最大最小） 或需要精确数值时，必须调用此工具。 Args: query: 自然语言查询问题 """# 1. 获取数据源 Schemaschemas=awaitget_schemas_by_ids(data_source_ids)# 2. 构建 Text-to-SQL Promptprompt=_build_text_to_sql_prompt(query,schemas)# 3. LLM 生成 SQLsql=awaitllm_generate_sql(prompt)# 4. 安全校验validate_sql(sql)# 5. 执行查询result=awaitexecute_readonly_query(sql)# 6. 格式化返回returnformat_query_result(result)returnFunctionTool.from_defaults(async_fn=execute_data_query,name='data_query',description=('查询用户上传的 Excel/CSV 数据。''当问题涉及数据统计、排名、对比、聚合或精确数值时使用此工具。''参数: query=自然语言查询问题'),)

4.5 Agent 工厂改造（多 Tool 支持）

classAgentFactory:@classmethoddefcreate_agent(cls,kb_id:str,collection_name:str,data_source_ids:list[str]|None=None,# 新增参数system_prompt:str|None=None,)->FunctionAgent:tools=[]# 1. RAG 工具（始终可用）tools.append(create_rag_tool(kb_id,collection_name))# 2. SQL 工具（有数据源时才注入）ifdata_source_ids:tools.append(create_sql_tool(data_source_ids))# 系统提示也要调整final_prompt=system_promptor_build_system_prompt(has_sql_tool=bool(data_source_ids))returnFunctionAgent(name='knowledge_assistant',description='基于知识库和数据的智能问答助手',system_prompt=final_prompt,tools=tools,llm=get_llm(),streaming=True,)

五、进阶：PDF/DOCX 中的表格如何处理？

5.1 PDF 中的表格 vs Excel 文件的本质区别

5.2 三种处理策略

策略 A：作为 Chunk 的一部分向量化（默认）

MinerU 已经把表格转成了 Markdown/HTML 格式，切片时表格自然包含在 chunk 里：

## 3.2 费用明细 | 项目 | 单价(元) | 数量 | 合计(元) | |------|---------|------|---------| | 设计费 | 50,000 | 1 | 50,000 | | 施工费 | 30,000 | 3 | 90,000 | | 监理费 | 15,000 | 2 | 30,000 | 如上表所示，本项目总费用为 17 万元...

整块一起向量化，用户问"设计费是多少"时，语义检索能命中这个 chunk，LLM 直接从表格文本中读取答案。

适用场景：表格较小（< 50 行）、表格是描述性/摘要性的、用户问的是"是什么"而不是"算一下"。

策略 B：提取建表（和 Excel 一样）

把 PDF 中的表格从文档中"抠出来"，单独建一张 PostgreSQL 表。

适用场景：表格很大（100+ 行）、表格是结构化的数据集、用户经常需要对表格数据做聚合查询。

缺点：表格脱离了文档上下文，丢失语义；管理复杂。

策略 C：智能分层处理（✅ 最终推荐）

核心思路：默认走策略 A（向量化），遇到"数据密集型表格"时自动升级到策略 B。

MinerU 输出 Markdown │ 表格检测 │ ┌────┴────┐ │ 有表格？ │ └────┬────┘ │ ┌────┴────────────────────┐ ▼ ▼ 描述性表格 数据密集型表格 （< 20行， （> 50行， 有上下文包裹） 纯数据无叙述） │ │ ▼ ▼ 保留在 Chunk 中 提取建表到 PostgreSQL → 向量化到 Milvus → 原文替换为摘要 → 通过 sql_tool 查询

5.3 表格分类规则

defclassify_table(headers:list[str],row_count:int,surrounding_text:str)->bool:""" 判断表格是"描述性"还是"数据密集型" 返回 True 表示数据密集型，需要建表 判断依据： 1. 行数阈值（> 50 行） 2. 列头含数值关键词（金额、数量、合计、%...） 3. 周围文本是否缺少叙述性描述 """# 规则 1: 行数阈值ifrow_count>50:returnTrue# 规则 2: 列头含数值关键词占比超过 50%numeric_keywords=['金额','数量','合计','总计','单价','比例','%','元','万']numeric_count=sum(1forhinheadersifany(kwinhforkwinnumeric_keywords))ifnumeric_count>=len(headers)*0.5:returnTrue# 规则 3: 周围文本很短（< 50 字符），说明表格是独立的数据罗列iflen(surrounding_text.strip())<50:returnTruereturnFalse

5.4 数据密集型表格的处理流程

asyncdefprocess_data_heavy_table(table_info,document):"""处理数据密集型表格：提取建表 + 原文替换为摘要"""# 1. 提取数据，动态建 PostgreSQL 表table_name=f"doc_{document.id}_table_{uuid4().hex[:8]}"awaitcreate_table_from_html(table_name,table_info.table_html,table_info.headers)awaitinsert_data_to_table(table_name,table_info.table_html)# 2. 注册到 data_source 表（和 Excel 共用同一套体系）awaitsave_data_source(doc_id=document.id,table_name=table_name,column_info=table_info.headers,source_type='pdf_table',# 标记来源是 PDF 中的表格original_file=document.file_name,)# 3. 在原文中替换为摘要（保留上下文）summary=(f"\n> 📊 此处有一张数据表（{table_info.row_count}行），"f"列：{', '.join(table_info.headers)}。"f"可通过数据查询工具精确查询。\n")returnsummary# 替换原文中的表格 HTML

六、Text-to-SQL 查询流程详解

6.1 完整流程

用户: "Q1 各产品总销售额排名，取前10" │ ▼ Agent 判断: 这是数据查询问题，调用 data_query Tool │ ▼ data_query Tool 内部: │ ├── 1. 从 data_source 表获取 column_info + table_description │ ├── 2. 构建 Prompt: │ "你是一个 SQL 专家。以下是可用的数据表： │ 表名: excel_a1b2c3d4 (2024年Q1销售报表) │ 列: 产品(TEXT), 1月(NUMERIC), 2月(NUMERIC), 3月(NUMERIC) │ │ 用户问题: Q1 各产品总销售额排名，取前10 │ 请生成 SELECT 语句。" │ ├── 3. LLM 生成 SQL: │ SELECT "产品", ("1月"+"2月"+"3月") AS q1_total │ FROM excel_a1b2c3d4 │ ORDER BY q1_total DESC LIMIT 10 │ ├── 4. 安全校验: 只含 SELECT ✅，表名在白名单内 ✅ │ └── 5. 执行 SQL，返回结果 │ ▼ Agent 拿到结果，组织自然语言回答: "Q1 销售额前10的产品如下： 1. 产品A: 46,500 元 2. 产品B: 28,800 元 ..."

6.2 多 Tool 协同示例

用户: "合同里的费用明细是多少？另外和预算表对比一下" │ ▼ Agent 判断: 这个问题涉及两部分 │ ├── 1. "合同里的费用明细" │ → knowledge_search（语义检索合同 PDF 中的表格 chunk） │ └── 2. "和预算表对比" → data_query（SQL 查询预算表 Excel 的数据） │ ▼ Agent 综合两个 Tool 的结果，生成最终回答

七、总结

7.1 核心原则

1. 结构化数据走 SQL，非结构化数据走向量— 不要把所有东西都塞进 Milvus
2. RAG 管语义，SQL 管数据，Agent 管路由— 各司其职
3. Excel 永久建表— 一次导入，N 次查询，毫秒级响应
4. PDF 中的表格智能分层— 小表格向量化，大表格建表
5. 统一查询入口— 无论数据来源是 Excel 还是 PDF 提取，都通过同一个sql_tool查询

7.2 最终架构一览

7.3 避坑清单

作者：RAG 不是万能的。结构化数据天然就有更好的查询方式（SQL），强行走向量化是在用错误工具解决正确问题。好的架构是让每种数据走最适合它的路径。