做RAG项目的人多半都遇到过这个场面信心满满地把公司那张核心业务表导入知识库结果模型回答时要么把列名当成正文念出来要么把不同行的数据串在一起胡编更离谱的是问它上个月A类客户总数是多少它回答出一串完全对不上的数字。问题基本都出在导入环节——表格数据跟PDF、Word那种连续文本完全不是一个物种它的信息藏在二维结构里按文本切片的方式一拆语义就碎了。这篇文章是系列第三篇重点解决RAG中表格与数据库的导入问题主要包括三块CSV文件怎么导、Excel多Sheet工作簿怎么解析、以及借助LlamaHub生态直接连数据库取数的完整链路。适合正在搭本地知识库、或者做企业级RAG应用但卡在结构化数据导入这步的开发者阅读。1. 为什么表格数据在RAG里这么难搞1.1 表格与纯文本的本质差异先想清楚一个问题为什么PDF、Markdown、txt这类格式可以直接丢给解析器切分表格却不行因为文本是线性结构句子和段落天然有前后文关系切片后哪怕丢了上下文模型也能靠残缺语义猜个大概。但表格是二维结构每一行的含义依赖于列头每一列的含义又依赖表头分组。你把一个10列的表按行切片每片变成独立的一行文本模型根本不知道销售额成本利润这几个词到底对应什么数据。举个例子一张订单表里有customer_id、order_date、amount三列。切片后某一段是A1001, 2024-11-03, 3500模型看到这行字只会当成一个普通句子它无法理解A1001是人名代号还是订单号3500是金额还是数量。但如果把表头语义注入到每一行里变成客户A1001在2024-11-03下单金额3500元模型就能正确理解这个数据点。这就是表格导入的核心矛盾原始形态的信息密度高但模型读不懂扁平化之后模型读得懂但信息被稀释。所以方案设计的重点不是怎么把表格塞进去而是怎么在保留结构语义的前提下让模型能理解。1.2 三条技术路线的取舍表格导入业界主流做法大致分三类各有明确的适用场景方案原理适合场景缺点行级文本化把每行转成自然语言描述再交给文档切分器行数少、列数多、需要精确查询数据量大时Token消耗高摘要切片混合先做整体摘要再把明细分块摘要作为公共上下文表很大但问答集中在统计汇总场景查询具体明细时摘要没用结构化索引表格原样入库查询时通过代码把自然语言转成查询语句数据量大、更新频繁、需要精确计算实现成本高依赖LLM转SQL能力这篇里CSV和Excel走的是第一类和第二类的结合LlamaHub连库走的是第三类。你不需要一开始就选最重的方案小表用行级文本化完全够用大表再考虑连库查询。2. CSV导入最省事的格式也有讲究2.1 十分钟跑通的基座方案CSV是所有表格格式里结构最简单的没有多Sheet、没有公式、没有合并单元格读出来就是规整的行列。在LlamaIndex里导入CSV最直接的方式是用Pandas读出来再封装成Document不需要走文件解析器。import pandas as pd from llama_index.core import Document, VectorStoreIndex from llama_index.core.node_parser import SentenceSplitter df pd.read_csv(orders.csv) print(df.shape, df.columns.tolist()) # 输出示例(5000, 8) [customer_id, order_date, amount, ...] docs [] for _, row in df.iterrows(): text 、.join([f{col}: {row[col]} for col in df.columns]) docs.append(Document(texttext))这段逻辑很直白遍历每一行把列名和值拼成一段自然语言文本。customer_id: A1001、order_date: 2024-11-03、amount: 3500这样的格式比纯CSV行好理解得多。实测下来5000行以内的表用这个方案问答准确率比直接喂原始行文本高出一截。但直接把df.iterrows()的结果逐个封装有个隐患Pandas会把DataFrame的索引类型带出来如果是整数索引那没问题如果索引是字符串且列名也含特殊字符拼接时容易出格式混乱。稳妥的做法是df df.reset_index(dropTrue)先重置索引再构造文本。2.2 元数据注入从能用变成好用光把每行转成文本还不够。实际使用中你会发现模型经常问这个表是什么时候的数据数据从哪来的而原始行文本里没有这些信息。解决方案是给Document加元数据让这些公共信息随着每个节点一起进入向量库。from datetime import datetime documents [] for idx, row in df.iterrows(): text 、.join([f{col}: {row[col]} for col in df.columns]) documents.append( Document( texttext, metadata{ source: orders.csv, row_index: idx, table_name: 订单明细表, data_date: 2024-11-01, update_time: datetime.now().isoformat(), }, ) ) # 切分时给节点也带上元数据 splitter SentenceSplitter(chunk_size1024, chunk_overlap64) nodes splitter.get_nodes_from_documents(documents)加了source和row_index之后你可以在问答结果里直接回溯这条数据来自CSV的哪一行做引用溯源非常方便。data_date这个字段尤其重要——RAG最怕模型拿旧数据回答新问题有了日期元数据后续可以做时间过滤比如只检索某个月份之后的数据。2.3 数据量大时的切分策略CSV到了几万行甚至几十万行逐行封装Document会让节点数量爆炸检索速度和内存双双告急。此时不能再用一行一个Document的思路而应走分块读取列语义保持的路线。from llama_index.core.node_parser import SentenceSplitter # 每2000行为一个Document chunk_df pd.read_csv(big_orders.csv, chunksize2000) all_nodes [] for chunk in chunk_df: header 、.join([f{col}: {col} for col in chunk.columns]) text_lines [header] for _, row in chunk.iterrows(): text_lines.append(、.join([f{col}: {row[col]} for col in chunk.columns])) block_text \n.join(text_lines[:50]) # 每块最多50行防止超长 doc Document(textblock_text, metadata{source: big_orders.csv}) splitter SentenceSplitter(chunk_size1024, chunk_overlap128) all_nodes.extend(splitter.get_nodes_from_documents([doc]))这个做法的关键是text_lines[:50]这个上限。为什么不一次性把2000行全部拼进去因为一个Document超过一定长度后切分器会把中间的行拦腰截断一行数据可能被切成两半语义就断了。限制在50行以内既能保证每块都有足够上下文又不会让切分器把行拆碎。有一个经验值可以参考单行文本长度 × 行数 ≤ chunk_size × 0.7。如果每行平均80个字符chunk_size设1024那一块最多放8到9行。放多了必然被截断。3. Excel导入多Sheet、公式和脏数据的组合拳3.1 为什么不能把xlsx当CSV处理你可能觉得Excel不就是带格式的CSV吗用Pandas照样read_excel一把梭。实际项目里Excel比CSV麻烦得多主要坑在三处第一一个工作簿有多个Sheet每个Sheet结构可能完全不同有的表头在第一行有的在第三行有的压根没表头。第二单元格里可能是公式Pandas默认读出来是公式字符串而不是计算后的值比如SUM(B2:B10)你拿去喂模型它看到的是一串公式而不是一个数字。第三合并单元格、空行、空列、日期格式混乱这些脏数据在CSV里不会出现但在Excel里几乎必现。所以Excel导入的核心不是读取而是清洗和解构。3.2 多Sheet工作簿的结构化导入先说读取层。推荐用pd.read_excel加sheet_nameNone把所有Sheet一次性读进来然后针对每个Sheet单独处理。千万不要循环里反复打开同一个文件性能差且容易因句柄问题报错。import pandas as pd from llama_index.core import Document xls pd.ExcelFile(sales_report.xlsx) all_docs [] for sheet_name in xls.sheet_names: df pd.read_excel(xls, sheet_namesheet_name, header0) # 去掉完全为空的列和行 df.dropna(axis1, howall, inplaceTrue) df.dropna(axis0, howall, inplaceTrue) for _, row in df.iterrows(): row_text 、.join( f{col}: {row[col]} for col in df.columns if pd.notna(row[col]) ) all_docs.append( Document( textrow_text, metadata{ source: sales_report.xlsx, sheet_name: sheet_name, }, ) )注意if pd.notna(row[col])这个条件。Excel表格里经常有某些行只有部分列有值直接把NaN拼进文本会出现客户名称: nan这种垃圾内容模型检索时容易误匹配。过滤掉空值之后文本干净很多实测问答准确率能提升两三个百分点。3.3 公式单元格、日期和表头偏移的处理公式问题用pd.read_excel(..., engineopenpyxl)解决不了根本openpyxl读取公式单元格返回的是公式字符串需要设置data_onlyTrue才能拿到缓存的计算结果。df pd.read_excel( sales_report.xlsx, sheet_nameSheet1, engineopenpyxl, data_onlyTrue, # 取计算后的值而不是公式 )但data_onlyTrue有个副作用如果Excel文件是从没被Excel程序打开过的纯代码生成文件缓存结果可能不存在读出来依然是空值。这种情况的处理办法是保留两路读取公式字符串和计算值都拿到手优先用计算值缺失时再把公式字符串去掉号和函数名尝试提取常量部分。日期列的问题也很隐蔽。Excel里的日期本质是序列号Pandas读出来可能是Timestamp对象也可能是字符串还可能是整数。统一格式化是必须的from datetime import datetime def normalize_cell_value(val): if isinstance(val, (datetime, pd.Timestamp)): return val.strftime(%Y-%m-%d) return val表头偏移处理也列一下。有的Excel为了美观在真正表头上方多了一两行标题文字读进来之后第一行成了垃圾数据。做法是先手动设置header参数或者读原始数据后自己找表头行raw pd.read_excel(xls, sheet_namesheet_name, headerNone) # 找到第一个非全空行的索引作为表头 header_row_idx raw.apply(lambda r: r.notna().sum() 0, axis1).idxmax() df pd.read_excel(xls, sheet_namesheet_name, headerheader_row_idx)这样虽然多一些代码但能覆盖绝大多数从业务系统导出的带抬头Excel。3.4 表格描述注入让模型知道自己在看什么多Sheet场景下有一个容易被忽视的问题模型只知道自己在看一堆文本片段不知道这些片段属于什么业务模块。同一列名叫金额在订单表里是订单金额在退款表里是退款金额语义完全不同。所以每个Sheet在组织文本时应该在块的首部注入一段该Sheet的结构描述sheet_meta { 订单表: 本表为订单明细每行代表一笔客户订单包含客户ID、下单日期、订单金额、商品类别等信息, 退款表: 本表为退款记录每行代表一笔退款申请包含订单号、退款金额、审批状态等信息, } block_text f【表格说明】{sheet_meta.get(sheet_name, )}\n block_text \n.join( 、.join(f{col}: {normalize_cell_value(row[col])} for col in df.columns if pd.notna(row[col])) for _, row in df.iterrows() )这段描述会跟着每个节点一起被向量化模型检索到具体行时能同时看到上下文。我测试过加入表格描述后涉及跨表比较的问题比如退款金额超过订单金额的有哪些准确率提升明显。4. LlamaHub连库实战从数据库到知识库的完整链路4.1 为什么需要连库而不是导CSV如果你的数据存在MySQL、PostgreSQL或者SQL Server里最省事的方式当然是导出CSV再走前面两节的路子。但导出方案有三个绕不开的问题数据实时性差——每次导出都要手动操作导完数据可能已经过期权限管理缺失——导出的CSV是全量裸数据落在本地后没法做行级或列级权限隔离规模受限——千万级数据表导出成CSV再向量化节点数量大到索引建不起来。连库取数的核心思路是不把数据全量搬到向量库而是把查询能力留给数据库RAG负责理解和编排。具体来说通过文本到SQL的方式让模型根据用户问题生成查询语句去数据库里精确取数再把计算结果作为上下文回答用户。4.2 LlamaHub上连库工具的选型逻辑LlamaHub是LlamaIndex官方的集成仓库里面已经有不少数据库连接器。操作数据库相关的reader主要看两个各自定位不同llama_index.readers.database.DatabaseReader负责把数据库表内容读成Document适合把整表或查询结果导入向量库llama_index.readers.database.SQLDatabaseNodeMapping则负责根据自然语言查询去数据库取数并映射成节点适合做Text-to-SQL的实时查询链路。选型逻辑很简单数据量小、查询频率低用DatabaseReader全量导入数据量大、查询频率高、对实时性要求高用SQLDatabaseNodeMapping走实时查询。大多数企业内部场景是后者。4.3 一次MySQL连库的完整配置过程下面是一套可用性很高的MySQL连库实现SQLAlchemy负责连接DatabaseReader负责读表。这段代码我跑过多次依赖项为llama-index-readers-database、sqlalchemy、pymysql。from sqlalchemy import create_engine from llama_index.readers.database import DatabaseReader # 连接串格式mysqlpymysql://用户名:密码主机:端口/库名?charsetutf8mb4 engine create_engine( mysqlpymysql://root:your_password127.0.0.1:3306/business_db?charsetutf8mb4 ) reader DatabaseReader( sqlalchemy_engineengine, enginemysql, ) # 方式一全表读取 docs reader.load_data(querySELECT * FROM orders LIMIT 10000;) # 方式二按条件过滤后读取 docs_filtered reader.load_data( querySELECT customer_id, order_date, amount FROM orders WHERE order_date 2024-01-01 )charsetutf8mb4这个参数必须带上不然后端存储中文时会报编码错误或者出现乱码。load_data接收原生的SQL字符串意味着你完全可以按需查询不用把整个表都导出来。读出来的docs是标准的Document列表可以直接交给向量索引。但这里有个坑数据库里的列名如果含下划线或者大写字母模型可能会误解语义比如create_time被理解成创建时间没问题但CUST_NO这种缩写就容易懵。建议读表时在SQL里直接用AS给列起别名SELECT customer_id AS 客户ID, order_date AS 下单日期, amount AS 订单金额 FROM orders中文别名在向量化时效果远好于英文缩写因为模型的语义空间里下单日期比order_date更容易被检索命中。4.4 增量更新与连接安全连库方案跑起来之后最常遇到的问题不是连不上而是长时间运行后连接断掉。SQLAlchemy的连接池默认会回收空闲连接但这在长驻服务里反而容易出问题连接被数据库端断开但应用侧不知道下一次查询直接报MySQL Connection Not Available。解决办法是显式配置连接池的预检参数engine create_engine( mysqlpymysql://root:password127.0.0.1:3306/business_db?charsetutf8mb4, pool_pre_pingTrue, pool_recycle3600, )pool_pre_pingTrue会在每次取连接前先发一个SELECT 1探活确保拿到的连接是可用的pool_recycle3600强制连接一小时后重建防止MySQL端主动断开。这两个参数加上之后长驻服务基本不会再出现断连问题。增量更新的设计上不要在每次启动任务时全表扫描而是用WHERE updated_at last_sync_time这样的条件增量取数配合元数据里的sync_time字段一起入库。import time last_sync_time time.strftime(%Y-%m-%d %H:%M:%S, time.localtime(time.time() - 86400)) docs reader.load_data( queryf SELECT customer_id AS 客户ID, order_date AS 下单日期, amount AS 订单金额 FROM orders WHERE updated_at {last_sync_time} )5. 实战踩过的坑与能用到底的经验5.1 五个高频问题现象、根因和解决把这个系列做过各种表格导入后我把高频问题汇总成一张表基本覆盖了常见的翻车现场。问题现象根本原因处理办法模型把数字列名当正文回答读取时列名和值没有语义关联行级文本化时拼成列名值格式中文乱码或问号文件编码是GBKPandas默认UTF-8pd.read_csv(..., encodinggbk)不确定时用encoding_errorsignoreExcel公式列全是NaNdata_only未开启或文件无缓存data_onlyTrue必要时用LibreOffice转存大表导入后检索极慢节点数量爆炸、embedding耗时过长分块读取限制每块行数或走Text-to-SQL连库路线连库后查询偶发断连连接池连接被MySQL端回收pool_pre_pingTruepool_recycle3600其中编码问题我要多强调一句。很多业务系统导出的CSV是GBK或GB18030编码直接用默认参数读Pandas会抛UnicodeDecodeError。保险起见可以这样读for enc in [utf-8, gbk, gb18030]: try: df pd.read_csv(file.csv, encodingenc) break except UnicodeDecodeError: continue虽然笨一点但能在编码不明确的情况下自动完成匹配实测足够应对大多数场景。5.2 判断标准什么情况不该走表格导入有一套判断逻辑值得在动手前走一遍能帮你省下大量的无用功。如果用户的查询都是上个月总营收多少A类客户占比这类聚合统计问题你其实不需要导入明细表只需要把汇总结果做成一页PDF或一段Markdown文本导入就够效果更好、成本更低。如果用户的查询需要精确匹配某一行比如订单号A1001的收货地址是什么行级文本化方案能回答但远不如直接连库查准确。此时应该连库。如果表格里的列数超过20列而且每列的值都有独立业务含义行级文本化会让每行文本变得非常长且冗余建议先做列筛选只保留问答高频涉及的列。如果表格会每天更新不要每次全量重建索引优先设计增量导入或连库实时读取。这个判断标准不是拍脑袋想的而是来自一次很惨痛的教训。当时我把一张50列的风险评估表全量导入结果索引文件建了几百MB查询时经常把不相关的列卷进来准确率反而比只保留核心8列的时候还要低。后来学乖了先做列筛选再做行级文本化效果立刻好了两个档次。表格导入这件事表面看只是把文件读进来转成Document实际上踩过的坑不比搭整个RAG管线少。CSV讲究清洗和切片Excel讲究解构和语义注入数据库连库讲究选型和安全。希望这篇能把你在表格数据导入路上可能遇到的问题提前挡掉一部分。最后再分享一个操作习惯不管数据规模多大导入完务必抽3到5条数据人工检查一下最终入库的Node文本看看有没有乱码、空值、断行。这一步看起来费时间但能避免你在后面问答调试时耗费更多的精力。