RAG实战：CSV、Excel与数据库结构化数据导入全攻略

1. 为什么表格和数据库是 RAG 数据导入的硬骨头做过 RAG 项目的人都有一个共识PDF 和纯文本的解析虽然烦但至少路子是清晰的大不了上 OCR、调分段策略。真正让人头疼的是表格和数据库这类结构化数据。你拿到的 CSV 可能有几十列、编码格式五花八门、缺失值满天飞Excel 更离谱一个文件里塞了七八个 sheet每个 sheet 的表头还不一样有的合并单元格有的前几行是说明文字。数据库那边呢表结构复杂、字段类型不统一、还有各种关联查询的需求。我前后在三个 RAG 项目里处理过结构化数据导入踩的坑足够写一本小册子。这一篇就把 CSV、Excel 和通过 LlamaHub 连接数据库这三条路完整走一遍从数据探查、清洗、分块策略到最终灌入向量库每一步都给出可直接复现的操作和参数。这篇文章适合已经在做 RAG 项目、需要把结构化数据接入知识库的开发者。如果你还没搭好基础的 RAG 框架建议先看前两篇关于文档解析和向量化的内容这篇默认你已经有了基本的 RAG 管线。提示结构化数据导入 RAG 的核心矛盾在于——表格的强结构性和向量检索的语义模糊性之间存在天然张力。后面所有的策略选择本质上都是在调和这个矛盾。2. 结构化数据导入 RAG 的整体设计思路2.1 三种数据源的特性对比与选型逻辑在动手写代码之前先把 CSV、Excel、数据库这三种数据源的核心差异理清楚。很多人一上来就写 loader结果写到一半发现数据格式跟自己预想的完全不一样返工成本极高。维度CSVExcel数据库结构复杂度低纯文本行列中高多 sheet、合并单元格、公式高多表关联、外键、索引编码问题常见GBK/UTF-8/UTF-8-BOM较少xlsx 内部统一取决于数据库字符集数据量级中大规模百万行常见中小规模十万行以内可大可小取决于查询更新频率批量导入为主人工维护为主实时或准实时解析工具pandas、csv 模块openpyxl、pandasSQLAlchemy、各数据库驱动RAG 适配难点表头语义弱、数值列多多 sheet 结构不统一需要 SQL 查询转自然语言选型的核心判断依据是数据是否频繁更新、结构是否稳定、是否需要行级检索。如果数据每周更新一次且结构固定CSV 批量导入是最省事的如果数据由业务人员手工维护、格式经常变Excel 解析要做得足够健壮如果需要实时查询或做聚合分析那必须走数据库连接的路子。2.2 分块策略表格数据怎么切才不丢语义这是整个流程里最关键的决策点。文本分块可以按 token 数切但表格不行——你把一行数据切成两半语义就碎了。我的经验是采用行级序列化 语义增强的策略。具体来说分三步第一步把每一行数据转成一段自然语言描述。比如一行销售数据{日期: 2024-01-15, 产品: A, 销量: 1200, 区域: 华东}转成2024年1月15日产品A在华东区域的销量为1200件。这样做的目的是让向量模型能理解这行数据的语义而不是把它当成一堆无意义的数字。第二步给每段描述加上表名和字段说明作为前缀。比如【销售表】日期2024-01-15产品A……。这个前缀在检索时能提供关键的上下文信息尤其是当你有多个表的时候。第三步根据行数和字段数决定是否合并。如果一行数据字段很少比如只有三四个字段可以把相邻的几行合并成一段避免碎片化。一般控制在每段 200-500 字比较合适。注意不要试图把整个表格塞进一个 chunk。我试过把一张 500 行的表直接转成一段文本结果向量化之后检索效果极差因为语义被稀释得太厉害了。2.3 元数据设计让检索能精确定位结构化数据导入 RAG 有一个纯文本没有的优势——你可以给每个 chunk 附加丰富的元数据。这些元数据在检索阶段可以做过滤大幅提升准确率。我通常会给每个 chunk 附加以下元数据source_type: csv / excel / databasetable_name: 来源表名或 sheet 名row_index: 原始行号方便溯源columns: 涉及的字段列表import_time: 导入时间戳这些元数据在 LangChain 或 LlamaIndex 里都可以通过Document对象的metadata字段传入。检索时可以用filter参数做精确过滤比如只检索销售表里华东区域的数据。3. CSV 导入实战从编码探测到批量入库3.1 编码探测与分隔符识别CSV 最坑的地方就是编码。国内的数据源大量使用 GBK 或 GB2312而 Python 默认按 UTF-8 读取直接报UnicodeDecodeError。更隐蔽的是 UTF-8-BOM文件头多了\xef\xbb\xbf三个字节导致第一列列名读出来带个不可见字符后面按列名取值全部失败。我的处理方案是用chardet做编码探测然后统一转成 UTF-8import chardet import pandas as pd def detect_encoding(file_path, sample_size100000): with open(file_path, rb) as f: raw f.read(sample_size) result chardet.detect(raw) return result[encoding], result[confidence] def read_csv_robust(file_path): encoding, confidence detect_encoding(file_path) if confidence 0.7: # 置信度低时尝试常见编码 for enc in [utf-8, gbk, gb2312, latin-1]: try: df pd.read_csv(file_path, encodingenc, nrows5) encoding enc break except (UnicodeDecodeError, pd.errors.ParserError): continue df pd.read_csv(file_path, encodingencoding) # 去除列名中的 BOM 和空白 df.columns df.columns.str.replace(\ufeff, ).str.strip() return df分隔符也不能想当然。虽然叫 CSV逗号分隔但实际项目中我遇到过用分号、制表符、甚至竖线分隔的。pandas 的sepNone配合enginepython可以自动嗅探分隔符但大文件下性能很差。我的做法是先读前几行手动判断确定分隔符后再全量读取。3.2 缺失值与异常值处理CSV 里的缺失值有无数种表示方式空字符串、NULL、N/A、-、nan、None。如果不统一处理转成自然语言描述时会出现销量为N/A这种毫无意义的句子。import numpy as np def clean_dataframe(df): # 统一缺失值标识 null_markers [, NULL, N/A, n/a, -, nan, None, null, NA] df df.replace(null_markers, np.nan) # 数值列填充或标记 for col in df.select_dtypes(include[np.number]).columns: if df[col].isna().sum() / len(df) 0.5: df df.drop(columns[col]) # 缺失超过50%直接丢弃 else: df[col] df[col].fillna(0) # 文本列用未知填充 for col in df.select_dtypes(include[object]).columns: df[col] df[col].fillna(未知) return df这里有个经验缺失率超过 50% 的列直接丢掉。我见过一张表 30 列里有 18 列全是空的这些列不仅没有信息量还会在序列化时产生大量噪音拖累检索效果。3.3 行级序列化与批量向量化清洗完之后就是核心的序列化环节。我的做法是给每一行生成一段描述文本然后批量送入 embedding 模型def row_to_text(row, table_name, columns): parts [f【{table_name}】] for col in columns: val row[col] if isinstance(val, float) and val int(val): val int(val) # 避免出现 1200.0 这种 parts.append(f{col}{val}) return .join(parts) def dataframe_to_documents(df, table_name): documents [] columns df.columns.tolist() for idx, row in df.iterrows(): text row_to_text(row, table_name, columns) doc Document( page_contenttext, metadata{ source_type: csv, table_name: table_name, row_index: int(idx), columns: ,.join(columns) } ) documents.append(doc) return documents批量向量化的时候注意控制 batch size。我实测下来OpenAI 的 embedding 接口 batch size 设在 100-500 之间比较稳太大容易超时太小则吞吐量上不去。如果用本地模型比如 BGE 系列batch size 可以设到 64-128取决于显存大小。实操心得百万行级别的 CSV 不要一次性全部加载到内存。用 pandas 的chunksize参数分块读取每块处理完就写入向量库处理完释放内存。我处理过一个 200 万行的 CSV16GB 内存的机器如果不分块直接 OOM。4. Excel 导入实战多 Sheet 与合并单元格的破解之道4.1 多 Sheet 遍历与表头识别Excel 文件最大的特点就是一个文件多个 sheet而且每个 sheet 的结构可能完全不同。有的 sheet 是数据表有的是说明页有的是图表。你不能无脑把所有 sheet 都当数据表处理。我的策略是先做一轮 sheet 扫描判断每个 sheet 是否是有效数据表import openpyxl def scan_excel_sheets(file_path): wb openpyxl.load_workbook(file_path, read_onlyTrue, data_onlyTrue) sheet_info [] for sheet_name in wb.sheetnames: ws wb[sheet_name] # 读取前10行做判断 rows [] for i, row in enumerate(ws.iter_rows(max_row10, values_onlyTrue)): rows.append(row) if i 9: break # 判断依据非空单元格比例、是否有表头特征 non_empty sum(1 for row in rows for cell in row if cell is not None) total sum(len(row) for row in rows) density non_empty / total if total 0 else 0 sheet_info.append({ name: sheet_name, density: density, is_data_sheet: density 0.3 and len(rows) 2 }) wb.close() return sheet_infodata_onlyTrue这个参数很关键。Excel 里的公式单元格如果不加这个参数读出来的是公式本身比如SUM(A1:A10)而不是计算结果。加上之后读到的才是值。但要注意如果这个 Excel 文件从来没有被 Excel 软件打开保存过缓存的计算值可能不存在读出来是 None。这种情况就需要用formulas库或者手动计算。4.2 合并单元格的向下填充合并单元格是 Excel 解析的头号杀手。比如部门这一列第一行是销售部合并了下面 5 行openpyxl 读出来只有第一行有值后面 5 行都是 None。如果不处理序列化出来就是部门未知。处理方案是检测合并区域并向下填充def fill_merged_cells(ws): merged_ranges list(ws.merged_cells.ranges) for merge_range in merged_ranges: min_row, min_col merge_range.min_row, merge_range.min_col max_row, max_col merge_range.max_row, merge_range.max_col top_left_value ws.cell(rowmin_row, columnmin_col).value for row in range(min_row, max_row 1): for col in range(min_col, max_col 1): ws.cell(rowrow, columncol).value top_left_value return ws这个操作要在读取数据之前做。注意read_only模式下不能修改单元格需要先以普通模式加载填充完再读。大文件下这会比较慢但合并单元格的处理没有更好的办法。4.3 表头行自动定位很多业务人员做的 Excel 前几行是标题、说明、日期之类的真正的表头在第 3 行甚至第 5 行。如果直接用header0读列名全是Unnamed: 1这种。我的做法是扫描前 10 行找到最像表头的那一行。判断标准是该行非空单元格数量最多、且该行以下的数据行类型一致性最高。def find_header_row(df_raw): best_row 0 best_score 0 for i in range(min(10, len(df_raw))): row df_raw.iloc[i] non_null row.notna().sum() # 表头行通常非空值多且都是字符串 str_count sum(1 for v in row if isinstance(v, str)) score non_null str_count * 0.5 if score best_score: best_score score best_row i return best_row找到表头行之后用pd.read_excel(file_path, sheet_namesheet, headerbest_row)重新读取即可。踩坑记录有一次处理一个财务报表表头在第 4 行但第 3 行有个合并单元格写着2024年度我一开始没处理结果这一行被当成了数据行序列化出来变成2024年度未知未知未知……污染了好几个 chunk。后来加了表头定位逻辑才解决。5. 通过 LlamaHub 连接数据库的完整流程5.1 DatabaseReader 的配置与连接LlamaHub 提供了DatabaseReader这个工具可以直接把数据库查询结果转成 Document 对象。它底层用的是 SQLAlchemy所以理论上支持所有 SQLAlchemy 兼容的数据库——MySQL、PostgreSQL、SQLite、Oracle 都行。先安装依赖pip install llama-index-readers-database sqlalchemy pymysql连接配置的核心是构造连接字符串。以 MySQL 为例from llama_index.readers.database import DatabaseReader connection_string mysqlpymysql://user:passwordhost:3306/dbname?charsetutf8mb4 reader DatabaseReader(uriconnection_string) documents reader.load_data( querySELECT id, product_name, category, price, stock FROM products WHERE status active, metadata{table: products, source: mysql} )DatabaseReader默认会把查询结果的每一行转成一段文本格式是column: value的拼接。这个默认行为对于简单的表够用但对于字段多、需要语义增强的场景建议自己写转换逻辑。5.2 SQL 查询转自然语言的策略直接用SELECT *把整张表拉出来是最粗暴的做法效果往往不好。更好的方式是根据 RAG 的检索需求定制查询语句并且在查询层面就做好语义增强。我的做法是写一个 SQL 模板用CONCAT在数据库层面就把行数据拼成自然语言SELECT CONCAT( 产品, product_name, 属于, category, 类目当前售价, price, 元库存, stock, 件 ) AS description FROM products WHERE status active;这样查出来的每一行本身就是一段完整的自然语言描述直接就能转成 Document省去了 Python 层的转换。而且数据库的字符串拼接性能通常比 Python 循环处理要快。对于需要多表关联的场景用 JOIN 把关联字段拼进来SELECT CONCAT( 订单, o.order_id, 由客户, c.customer_name, 于, o.order_date, 购买总金额, o.total_amount, 元 ) AS description FROM orders o JOIN customers c ON o.customer_id c.id WHERE o.order_date 2024-01-01;5.3 增量同步与去重机制数据库数据是动态变化的不可能每次全量重新导入。需要设计增量同步机制。最简单的方案是用时间戳字段做增量from datetime import datetime, timedelta last_sync_time get_last_sync_time() # 从配置或数据库读取 query f SELECT CONCAT(...) AS description, id, updated_at FROM products WHERE updated_at {last_sync_time} 去重的话用id字段作为唯一标识在写入向量库时先删除相同id的旧记录再插入新的。ChromaDB 和 Milvus 都支持按 metadata 过滤删除。注意增量同步要考虑时区问题。如果数据库存的是 UTC 时间而你的同步脚本用的是本地时间会出现漏同步或重复同步。统一用 UTC 时间戳做比较最稳妥。6. 常见问题与排查技巧实录6.1 编码与格式类问题速查问题现象根本原因解决方案UnicodeDecodeError: utf-8 codec cant decode文件是 GBK 编码用 chardet 探测后指定 encodinggbk第一列列名带不可见字符UTF-8-BOM 文件头读取后.str.replace(\ufeff, )所有列名都是 Unnamed表头不在第一行用 find_header_row 定位后重新读取数值列读出来是字符串列中有非数字字符如1,200先.str.replace(,, )再pd.to_numericExcel 公式读出来是 None文件未被 Excel 保存过无缓存值用 formulas 库计算或要求用户另存6.2 向量化与检索效果差的自查清单结构化数据导入后检索效果不好通常不是 embedding 模型的问题而是数据预处理的问题。按以下顺序排查检查序列化文本是否可读随便抽几个 Document 的page_content打印出来如果读起来像天书向量模型也理解不了。检查 chunk 长度分布如果大量 chunk 只有十几个字说明分块太碎如果大量超过 1000 字说明合并过度。检查元数据是否完整用filter检索时能不能精确命中目标表。检查是否有重复数据同一行数据被导入了多次检索时全是重复结果。检查数值列的处理像1200.0这种浮点表示在语义上和1200有细微差异统一转成整数。6.3 性能优化百万行数据的处理策略处理百万行级别的数据性能瓶颈通常在三个地方文件读取、序列化转换、向量化写入。文件读取用 pandas 的chunksize分块每块 5 万行左右。序列化转换用df.apply比逐行iterrows快 10 倍以上。向量化写入用批量接口并且开多线程并发。from concurrent.futures import ThreadPoolExecutor def process_chunk(chunk_df, table_name): docs dataframe_to_documents(chunk_df, table_name) # 批量向量化并写入 vector_store.add_documents(docs) return len(docs) with pd.read_csv(file_path, chunksize50000, encodingutf-8) as reader: with ThreadPoolExecutor(max_workers4) as executor: futures [] for chunk in reader: chunk clean_dataframe(chunk) futures.append(executor.submit(process_chunk, chunk, sales)) for f in futures: print(f处理了 {f.result()} 行)我实测下来200 万行数据用这个方案从读取到全部写入向量库大约 40 分钟。瓶颈在 embedding API 的调用上如果换成本地 GPU 推理可以压缩到 10 分钟以内。6.4 几个容易忽略的细节日期格式统一。CSV 里的日期可能是2024/1/15、2024-01-15、15/01/2024各种格式。序列化之前统一转成YYYY-MM-DD格式否则检索2024年1月的时候会漏掉很多。大数字的可读性。1200000和1,200,000在语义上是一样的但向量模型可能处理不同。我通常会在序列化时加上千分位或者转成120万这种中文习惯的表达。布尔值的处理。数据库里的0/1、true/false、Y/N要统一转成是/否否则检索是否有效的时候匹配不上。空表的处理。有些 sheet 或查询结果可能是空的要在代码里加判断避免生成空的 Document 污染向量库。最后分享一个排查技巧当你觉得检索效果不对劲的时候把用户的 query 和检索到的 top-5 Document 一起打印出来逐条看为什么这条被召回了、那条没被召回。十有八九问题出在序列化文本的表述方式上而不是向量模型本身。