表格和数据库这两类数据源在 RAG 项目里属于那种绕不过去、但踩坑率极高的存在。我做过不下十个知识库项目几乎每一个都会遇到这样的场景业务方丢过来一个几十兆的 Excel或者甩一个数据库连接串说把这些数据也喂给模型。看起来只是换个数据源读一下实际上从编码、表头识别、多 sheet 处理到分块策略、元数据保留、增量更新每一步都有讲究。这篇就专门聊 CSV、Excel 和通过 LlamaHub 连数据库这三条链路把我在实际项目里趟过的坑和总结出来的做法完整讲一遍。1. 为什么表格数据是 RAG 里最容易被低估的一环1.1 表格和纯文本的本质差异很多人做 RAG 的第一反应是把文档切块、向量化、检索这套流程对散文、说明文档、FAQ 效果很好但直接套到表格上就会出问题。原因在于表格的信息密度和结构特性和自然语言完全不同。一段 500 字的说明文字切块之后每块还能保持语义相对完整。但一张 20 行 10 列的表格如果你按字符数硬切很可能把一行数据切成两半或者把表头和表体分离。检索的时候模型拿到的是张三 销售部 12000这种没有列名的碎片根本不知道 12000 是工资还是业绩。我在一个 HR 知识库项目里就吃过这个亏。当时把员工信息表直接按行转成文本丢进去结果用户问研发部平均薪资是多少检索出来的片段全是孤立的数字模型只能瞎猜。后来改成每行都带上列名姓名张三部门研发部薪资12000检索准确率立刻上了一个台阶。所以处理表格数据核心思路不是切块而是结构化转述——把每一行或每一组行转成一句语义完整、自带上下文的话。1.2 表格数据在 RAG 中的三类典型用途搞清楚用途才能决定怎么处理。根据我的经验表格数据进 RAG 大致分三类第一类是精确查询型。比如产品价格表、库存表用户问XX 型号多少钱期望的是准确数字不是模糊匹配。这类数据其实更适合走结构化查询Text-to-SQL而不是纯向量检索。第二类是语义检索型。比如客户反馈表、工单记录用户问最近有没有关于物流慢的投诉需要的是语义相似度匹配。这类数据转成自然语言文本后走向量检索效果最好。第三类是混合型。比如财务报表既可能被问第三季度营收多少精确也可能被问哪个业务线增长最快需要推理。这类要同时保留结构化字段和文本描述。提示在动手写导入代码之前先想清楚你的表格数据属于哪一类。这直接决定了后面是走向量检索、结构化查询还是两者结合。我见过太多人上来就无脑向量化结果精确查询场景一塌糊涂。1.3 一个常见的认知误区有个误区特别普遍认为 CSV 和 Excel 是一回事只是格式不同。实际上在 RAG 处理里这两者的坑完全不一样。CSV 是纯文本编码问题、分隔符问题、引号转义问题是重灾区。而 Excel 是二进制格式多 sheet、合并单元格、公式、日期格式、隐藏行列才是主要麻烦。用同一套代码处理两者迟早翻车。后面我会分别讲。2. CSV 导入编码、分隔符与分块策略的连环坑2.1 编码问题为什么你的中文全是乱码CSV 导入第一个拦路虎永远是编码。国内业务方给的 CSV十有八九是 GBK 或 GB2312 编码而 Python 默认按 UTF-8 读结果就是满屏乱码。最稳妥的做法是先用chardet或charset-normalizer探测编码再读取import chardet 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] # 使用 encoding, confidence detect_encoding(sales.csv) print(f探测编码: {encoding}, 置信度: {confidence})但要注意chardet对小文件或者纯 ASCII 内容经常误判。我的经验是如果置信度低于 0.8或者探测结果是 ascii就手动按 GBK 试一次看中文是否正常。更省事的办法是直接写一个多编码回退逻辑def read_csv_safe(file_path): for enc in [utf-8-sig, utf-8, gbk, gb18030, latin-1]: try: with open(file_path, r, encodingenc) as f: content f.read() # 简单校验如果包含常见乱码字符则跳过 if \ufffd not in content: return content, enc except UnicodeDecodeError: continue raise ValueError(无法识别文件编码)这里utf-8-sig要放在utf-8前面因为带 BOM 的 UTF-8 文件用utf-8读会在第一个字段前多出一个不可见字符导致列名匹配失败。这个坑我踩过不止一次排查了半天才发现是 BOM 作祟。2.2 分隔符与引号转义标准 CSV 用逗号分隔但现实中你会遇到分号、制表符、竖线各种分隔符。更麻烦的是字段里本身包含逗号的情况比如北京市, 朝阳区这种地址。Python 的csv模块能自动处理引号转义但前提是你要正确指定delimiter和quotechar。我一般先用csv.Sniffer嗅探一下import csv def sniff_dialect(sample): try: dialect csv.Sniffer().sniff(sample, delimiters,;\t|) return dialect.delimiter, dialect.quotechar except csv.Error: return ,, # 默认回退Sniffer不是万能的遇到字段内容特别规整的数据也会误判。所以生产环境里我通常把嗅探结果作为默认值同时允许配置覆盖。2.3 从 DataFrame 到可检索文本的转换读进来之后用 pandas 处理是最顺手的。关键是怎么把 DataFrame 转成适合 RAG 的文本块。我的做法是分两种粒度。行级粒度适合记录型数据每行转成一句带列名的描述def row_to_text(row, columns): parts [f{col}{row[col]} for col in columns if pd.notna(row[col])] return .join(parts) # 示例输出姓名张三部门研发部薪资12000入职日期2022-03-15块级粒度适合需要保留表格整体结构的场景比如把连续 10 行打包成一个块前面加上表头说明def chunk_dataframe(df, chunk_size10, table_name数据表): chunks [] for i in range(0, len(df), chunk_size): sub_df df.iloc[i:ichunk_size] header f以下是{table_name}的第{i1}到{min(ichunk_size, len(df))}条记录\n header 列说明 、.join(df.columns) \n body \n.join([row_to_text(row, df.columns) for _, row in sub_df.iterrows()]) chunks.append(header body) return chunks块大小怎么定我的经验是控制在 500 到 800 字之间。太小了上下文不足太大了检索精度下降。10 行左右通常比较合适但具体要看每行的字段数量。2.4 元数据保留别让检索丢了来源一个容易被忽略的点是元数据。每个块都应该带上来源信息比如文件名、sheet 名、行号范围。这样检索出来之后用户能知道数据从哪来也方便做引用溯源。chunk_metadata { source: sales_2024.csv, row_range: f{i1}-{min(ichunk_size, len(df))}, table_type: sales, import_time: datetime.now().isoformat() }在 LangChain 或 LlamaIndex 里这些元数据会跟着 Document 对象一起存进向量库检索时可以按元数据过滤。比如只检索某个文件的内容或者只检索最近导入的数据。3. Excel 导入多 Sheet、合并单元格与公式的实战处理3.1 多 Sheet 遍历与命名策略Excel 和 CSV 最大的区别就是多 sheet。一个工作簿里可能有汇总明细备注好几个表每个表结构还不一样。用 pandas 读的时候sheet_nameNone会返回一个字典key 是 sheet 名value 是对应的 DataFrameimport pandas as pd def read_excel_all_sheets(file_path): sheets pd.read_excel(file_path, sheet_nameNone, engineopenpyxl) result {} for sheet_name, df in sheets.items(): if df.empty: continue result[sheet_name] df return result这里有个细节空 sheet 要跳过否则会产生一堆无意义的空块。另外 sheet 名本身是重要的元数据一定要保留因为用户提问时可能会说汇总表里那个数据。engine 的选择也有讲究。openpyxl处理.xlsxxlrd处理老的.xls。现在新项目基本都用.xlsx所以openpyxl是默认选择。但如果你的数据源里有历史遗留的.xls文件就得做格式判断def get_engine(file_path): if file_path.endswith(.xls): return xlrd return openpyxl3.2 合并单元格最隐蔽的数据杀手合并单元格是 Excel 导入里最坑的问题没有之一。pandas 读取时合并区域只有左上角有值其他单元格是 NaN。如果你直接处理就会得到一堆缺失值。比如一个部门列合并了 5 行读进来只有第一行是研发部后面 4 行都是 NaN。这时候需要做前向填充def handle_merged_cells(df): # 对疑似合并的列做前向填充 df df.ffill() return df但ffill()不能无脑用。如果某列本来就有合法的空值比如备注列大部分为空前向填充会污染数据。我的做法是先检测哪些列存在连续 NaN 且上方有值的模式只对这些列做填充。def smart_ffill(df, threshold0.3): for col in df.columns: na_ratio df[col].isna().mean() # 缺失率在合理范围内且不是全空才考虑填充 if 0 na_ratio threshold: df[col] df[col].ffill() return df更严谨的做法是用openpyxl直接读取合并单元格信息from openpyxl import load_workbook wb load_workbook(file_path) ws wb.active merged_ranges ws.merged_cells.ranges for merged_range in merged_ranges: print(f合并区域: {merged_range})拿到合并区域后可以精确地知道哪些单元格需要填充避免误伤。这个方法麻烦一点但对数据质量要求高的场景值得。3.3 公式与计算值读哪个Excel 单元格里可能是公式比如SUM(B2:B10)。pandas 默认读的是公式的计算结果缓存值而不是公式本身。大多数情况下这是对的因为你要的是数据不是公式。但有个例外如果文件是用程序生成的公式还没被 Excel 计算过缓存值可能是空的或者过期的。这时候读出来就是 NaN。解决办法是用openpyxl的data_onlyFalse读公式或者用data_onlyTrue读缓存值根据需求选。# 读缓存值默认推荐 df pd.read_excel(file_path, engineopenpyxl) # 如果需要公式本身 wb load_workbook(file_path, data_onlyFalse)我一般建议业务方在提供 Excel 之前先选择性粘贴为数值把公式固化掉。这样能避免很多不确定性。3.4 日期与数字格式的陷阱Excel 的日期存储是个整数从 1900 年 1 月 1 日开始的天数pandas 读取时会尝试转换。但有时候会转错比如把20240315这种数字当成日期或者把日期读成时间戳。处理日期列时我习惯显式指定df pd.read_excel(file_path, parse_dates[入职日期, 签约日期])如果自动解析失败就手动转换df[入职日期] pd.to_datetime(df[入职日期], errorscoerce, format%Y-%m-%d)errorscoerce会把无法解析的值变成 NaT避免整个流程报错。数字格式同理有时候 Excel 里的12000读进来是12000.0需要转成整数df[薪资] pd.to_numeric(df[薪资], errorscoerce).fillna(0).astype(int)4. LlamaHub 连数据库DatabaseReader 的配置与查询设计4.1 LlamaHub 是什么为什么用它LlamaHub 是 LlamaIndex 生态里的数据连接器集合里面有几百个 Reader覆盖各种数据源。数据库这块主要用DatabaseReader它基于 SQLAlchemy理论上支持所有 SQLAlchemy 能连的数据库。为什么不用自己写 SQL 查询因为DatabaseReader帮你处理了结果集到 Document 的转换、元数据附加、分页这些琐事。而且它和 LlamaIndex 的索引、检索链路是无缝衔接的。安装很简单pip install llama-index-readers-database4.2 连接配置URI 与引擎选择DatabaseReader需要一个 SQLAlchemy 的 engine 或者连接 URI。以 PostgreSQL 为例from sqlalchemy import create_engine from llama_index.readers.database import DatabaseReader engine create_engine(postgresql://user:passwordlocalhost:5432/mydb) reader DatabaseReader(engineengine)MySQL 的话engine create_engine(mysqlpymysql://user:passwordlocalhost:3306/mydb)SQLite 最简单直接给文件路径engine create_engine(sqlite:///data.db)注意连接串里的密码如果包含特殊字符比如 、#、%需要做 URL 编码否则解析会出错。这个坑很隐蔽报错信息也不直观。我一般用urllib.parse.quote_plus处理一下。4.3 查询设计全表还是条件查询DatabaseReader的核心方法是load_data接受一个 SQL 查询字符串documents reader.load_data( querySELECT id, title, content, created_at FROM articles WHERE status published, metadata_cols[id, created_at] )metadata_cols指定哪些列作为元数据保留不参与文本内容。这个设计很实用因为像 id、时间戳这类字段放在正文里没意义但作为过滤条件很有用。查询设计上有几个原则第一只查需要的列。SELECT *在宽表上会拉回大量无用数据既慢又占空间。明确列出需要的列。第二控制返回行数。大表全量拉取会爆内存。用LIMIT分批或者加时间范围条件query SELECT id, title, content FROM articles WHERE created_at 2024-01-01 ORDER BY id LIMIT 1000 OFFSET 0 第三把多表关联的结果拍平。如果查询涉及 JOIN返回的列名可能重复需要加别名区分query SELECT a.id AS article_id, a.title, u.name AS author_name FROM articles a JOIN users u ON a.author_id u.id 4.4 结果集到 Document 的转换逻辑DatabaseReader默认把每一行转成一个 Document格式是列名: 值的拼接。这个默认行为对简单表够用但复杂场景需要自定义。我一般会继承DatabaseReader或者自己写转换逻辑把一行数据转成更自然的描述def row_to_document(row, table_desc): text f{table_desc}记录 text .join([f{k}为{v} for k, v in row.items() if v is not None]) return Document(texttext, metadata{source: database, table: table_desc})对于需要保留表结构说明的场景可以在每个 Document 前面加上表头解释或者单独建一个表结构说明的 Document检索时一起召回。4.5 增量更新与去重数据库数据是会变的不可能每次都全量重新导入。增量更新的关键是找到一个可靠的增量字段通常是自增 id 或者更新时间戳。# 记录上次导入的最大 id last_max_id get_last_imported_id() query fSELECT * FROM articles WHERE id {last_max_id} ORDER BY id LIMIT 1000 documents reader.load_data(queryquery) # 导入后更新记录 if documents: new_max_id max(doc.metadata[id] for doc in documents) save_last_imported_id(new_max_id)去重方面向量库一般支持按 id 覆盖。给每个 Document 设置稳定的 id比如farticle_{row_id}重复导入时就会覆盖而不是新增。5. 分块、元数据与检索质量的联动调优5.1 表格数据的分块粒度选择前面提到行级和块级两种粒度实际项目里往往需要混合。我的经验是数据类型推荐粒度理由记录型员工、产品行级每条记录独立查询精确汇总型报表、统计块级需要保留表格整体语境长文本字段描述、备注字段级单独切分避免被短字段稀释字段级切分是个进阶技巧。比如一个产品表有名称价格详细描述三列描述可能有几百字。如果整行转文本描述会淹没名称和价格。这时候可以把描述单独抽出来做一个 Document元数据里关联产品 id。5.2 元数据过滤的实战价值元数据不只是装饰它能大幅提升检索精度。举个真实例子一个知识库同时导入了 2023 和 2024 年的销售数据用户问今年销售额如果不做元数据过滤可能召回 2023 年的数据。解决办法是在导入时给每个块打上year标签检索时加过滤条件from llama_index.core.vector_stores import MetadataFilter, MetadataFilters filters MetadataFilters(filters[ MetadataFilter(keyyear, value2024) ]) retriever index.as_retriever(filtersfilters)这样检索范围就限定在 2024 年的数据里准确率提升明显。5.3 表头信息的处理策略表头是表格数据的灵魂但处理起来有讲究。有三种策略策略一每块重复表头。每个块前面都加上列名说明。优点是自包含缺点是冗余。策略二表头单独成块。把表头信息做成一个独立的 Document检索时可能被召回。优点是省空间缺点是可能召不回。策略三列名内联。每行转文本时带上列名薪资12000。这是我最推荐的方式兼顾了自包含和简洁。实际项目里我通常用策略三为主策略一为辅对特别重要的表。5.4 检索效果验证方法导入完了怎么知道效果好不好不能凭感觉。我一般准备一组测试问题覆盖精确查询、语义检索、跨表关联几种类型然后看召回率和准确率。test_queries [ 研发部有多少人, 最近一个月的投诉记录, 销售额最高的产品是什么 ] for q in test_queries: nodes retriever.retrieve(q) print(f问题: {q}) for node in nodes[:3]: print(f 召回: {node.text[:100]}...)如果召回的内容明显不相关就要回头检查分块策略和元数据设置。这个迭代过程通常要跑几轮才能调好。6. 几个我踩过的真实坑与应对6.1 大文件内存溢出有一次导入一个 200MB 的 Excelpandas 直接吃掉了 4GB 内存机器差点崩了。后来改成用openpyxl的只读模式逐行读from openpyxl import load_workbook wb load_workbook(file_path, read_onlyTrue) ws wb.active for row in ws.iter_rows(values_onlyTrue): # 逐行处理 passread_onlyTrue不会把整个文件加载到内存适合大文件。代价是不能随机访问只能顺序读。CSV 的话用chunksize分块读for chunk in pd.read_csv(file_path, chunksize10000): process(chunk)6.2 特殊字符导致的向量化异常表格数据里经常有各种特殊字符换行符、制表符、零宽字符、emoji。这些字符在向量化时可能引发问题或者污染检索结果。导入前做一次清洗是必要的import re def clean_text(text): if not isinstance(text, str): return str(text) # 去除零宽字符 text re.sub(r[\u200b-\u200f\ufeff], , text) # 统一换行符 text text.replace(\r\n, \n).replace(\r, \n) # 压缩多余空白 text re.sub(r\s, , text) return text.strip()零宽字符特别讨厌肉眼看不见但会让两个看起来一样的字符串不相等导致去重失效。6.3 数值精度丢失浮点数在转换过程中容易丢精度。比如 Excel 里的 0.1 0.2读出来可能是 0.30000000000000004。对于金额这类敏感数据我一般用Decimal处理或者导入前就格式化成字符串。from decimal import Decimal df[金额] df[金额].apply(lambda x: str(Decimal(str(x)).quantize(Decimal(0.01))))6.4 时间戳时区问题数据库里的时间戳通常带时区Excel 里的日期没有时区概念。混在一起处理时容易出错。我的做法是统一转成 UTC 或者统一用本地时间并在元数据里标注时区。from datetime import timezone def normalize_datetime(dt): if dt.tzinfo is None: # 假设是本地时间 return dt.replace(tzinfotimezone.utc) return dt.astimezone(timezone.utc)7. 把三条链路串起来的工程化建议7.1 统一的导入接口设计CSV、Excel、数据库三条链路最好抽象成一个统一的接口对外只暴露导入这一个动作class DataImporter: def import_csv(self, file_path, **kwargs): ... def import_excel(self, file_path, **kwargs): ... def import_database(self, connection_uri, query, **kwargs): ... def import(self, source_type, **kwargs): if source_type csv: return self.import_csv(**kwargs) elif source_type excel: return self.import_excel(**kwargs) elif source_type database: return self.import_database(**kwargs)这样上层调用不用关心底层细节也方便后续加新的数据源。7.2 导入日志与可追溯性每次导入都要记日志导入了什么文件、多少行、多少块、耗时多久、有没有异常。这些信息在排查问题时非常有用。import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) logger.info(f开始导入 {file_path}, 共 {len(df)} 行) logger.info(f生成 {len(chunks)} 个文本块) logger.info(f导入完成耗时 {elapsed:.2f} 秒)我还会把导入记录存到一张表里记录文件哈希、导入时间、块数量。这样重复导入同一个文件时能检测出来避免浪费。7.3 失败重试与断点续传大批量导入难免中途失败。设计上要支持断点续传记录已处理到的位置失败后从断点继续而不是从头再来。对于数据库导入用 id 或时间戳做断点。对于文件导入用行号或块索引做断点。关键是断点信息要持久化不能只存在内存里。7.4 数据质量校验导入前做一轮校验能省很多事。常见的校验项必填字段是否为空数值字段是否在合理范围日期字段是否可解析枚举字段值是否在预期集合内def validate_row(row, schema): errors [] for field, rules in schema.items(): value row.get(field) if rules.get(required) and pd.isna(value): errors.append(f{field} 为空) if rules.get(type) number and not isinstance(value, (int, float)): errors.append(f{field} 不是数字) return errors校验不通过的行可以跳过、可以修正、也可以记录到异常表里人工处理。具体策略看业务要求。7.5 性能优化的几个实用技巧最后分享几个性能上的经验。批量向量化比逐条快得多尽量攒一批再调 embedding 接口。数据库查询加索引尤其是增量字段上。Excel 读取用openpyxl的只读模式CSV 用chunksize分块。向量库写入用批量接口别一条一条插。还有一个容易被忽略的点embedding 调用是网络请求有延迟。如果数据量大考虑并发调用但要注意 API 的限流。我一般用concurrent.futures控制并发数在 5 到 10 之间既能提速又不会触发限流。from concurrent.futures import ThreadPoolExecutor def batch_embed(texts, batch_size100, max_workers5): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: futures [executor.submit(embed_batch, texts[i:ibatch_size]) for i in range(0, len(texts), batch_size)] for future in futures: results.extend(future.result()) return results这套流程跑下来一个几十万行的表格数据从导入到可检索通常能在半小时内完成。当然具体时间取决于数据复杂度和 embedding 服务的速度。我在实际项目里最大的体会是表格数据的 RAG 处理难点从来不在技术本身而在于对数据特性的理解和细节的把控。编码、合并单元格、公式、时区、精度每一个都是小问题但任何一个没处理好都会让最终的检索效果大打折扣。与其追求花哨的架构不如把这些基础环节做扎实。