做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(text=text))这段逻辑很直白：遍历每一行，把列名和值拼成一段自然语言文本。customer_id: A1001、order_date: 2024-11-03、amount: 3500这样的格式，比纯CSV行好理解得多。实测下来5000行以内的表用这个方案，问答准确率比直接喂原始行文本高出一截。
但直接把df.iterrows()的结果逐个封装有个隐患：Pandas会把DataFrame的索引类型带出来，如果是整数索引那没问题，如果索引是字符串且列名也含特殊字符，拼接时容易出格式混乱。稳妥的做法是df = df.reset_index(drop=True)先重置索引，再构造文本。
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( text=text, metadata={ "source": "orders.csv", "row_index": idx, "table_name": "订单明细表", "data_date": "2024-11-01", "update_time": datetime.now().isoformat(), }, ) ) # 切分时给节点也带上元数据 splitter = SentenceSplitter(chunk_size=1024, chunk_overlap=64) 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", chunksize=2000) 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(text=block_text, metadata={"source": "big_orders.csv"}) splitter = SentenceSplitter(chunk_size=1024, chunk_overlap=128) 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_name=None把所有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_name=sheet_name, header=0) # 去掉完全为空的列和行 df.dropna(axis=1, how="all", inplace=True) df.dropna(axis=0, how="all", inplace=True) 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( text=row_text, metadata={ "source": "sales_report.xlsx", "sheet_name": sheet_name, }, ) )注意if pd.notna(row[col])这个条件。Excel表格里经常有某些行只有部分列有值，直接把NaN拼进文本会出现客户名称: nan这种垃圾内容，模型检索时容易误匹配。过滤掉空值之后，文本干净很多，实测问答准确率能提升两三个百分点。
3.3 公式单元格、日期和表头偏移的处理
公式问题用pd.read_excel(..., engine="openpyxl")解决不了根本，openpyxl读取公式单元格返回的是公式字符串，需要设置data_only=True才能拿到缓存的计算结果。
df = pd.read_excel( "sales_report.xlsx", sheet_name="Sheet1", engine="openpyxl", data_only=True, # 取计算后的值而不是公式 )但data_only=True有个副作用：如果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_name=sheet_name, header=None) # 找到第一个非全空行的索引作为表头 header_row_idx = raw.apply(lambda r: r.notna().sum() > 0, axis=1).idxmax() df = pd.read_excel(xls, sheet_name=sheet_name, header=header_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 # 连接串格式：mysql+pymysql://用户名:密码@主机:端口/库名?charset=utf8mb4 engine = create_engine( "mysql+pymysql://root:your_password@127.0.0.1:3306/business_db?charset=utf8mb4" ) reader = DatabaseReader( sqlalchemy_engine=engine, engine="mysql", ) # 方式一：全表读取 docs = reader.load_data(query="SELECT * FROM orders LIMIT 10000;") # 方式二：按条件过滤后读取 docs_filtered = reader.load_data( query="SELECT customer_id, order_date, amount FROM orders WHERE order_date >= '2024-01-01'" )charset=utf8mb4这个参数必须带上，不然后端存储中文时会报编码错误或者出现乱码。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( "mysql+pymysql://root:password@127.0.0.1:3306/business_db?charset=utf8mb4", pool_pre_ping=True, pool_recycle=3600, )pool_pre_ping=True会在每次取连接前先发一个SELECT 1探活，确保拿到的连接是可用的；pool_recycle=3600强制连接一小时后重建，防止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( query=f""" SELECT customer_id AS 客户ID, order_date AS 下单日期, amount AS 订单金额 FROM orders WHERE updated_at >= '{last_sync_time}' """ )5. 实战踩过的坑与能用到底的经验
5.1 五个高频问题：现象、根因和解决
把这个系列做过各种表格导入后，我把高频问题汇总成一张表，基本覆盖了常见的翻车现场。
| 问题现象 | 根本原因 | 处理办法 |
|---|---|---|
| 模型把数字列名当正文回答 | 读取时列名和值没有语义关联 | 行级文本化时拼成"列名：值"格式 |
| 中文乱码或问号 | 文件编码是GBK，Pandas默认UTF-8 | pd.read_csv(..., encoding="gbk")，不确定时用encoding_errors="ignore" |
| Excel公式列全是NaN | data_only未开启或文件无缓存 | data_only=True，必要时用LibreOffice转存 |
| 大表导入后检索极慢 | 节点数量爆炸、embedding耗时过长 | 分块读取+限制每块行数，或走Text-to-SQL连库路线 |
| 连库后查询偶发断连 | 连接池连接被MySQL端回收 | pool_pre_ping=True+pool_recycle=3600 |
其中编码问题我要多强调一句。很多业务系统导出的CSV是GBK或GB18030编码，直接用默认参数读，Pandas会抛UnicodeDecodeError。保险起见可以这样读：
for enc in ["utf-8", "gbk", "gb18030"]: try: df = pd.read_csv("file.csv", encoding=enc) break except UnicodeDecodeError: continue虽然笨一点，但能在编码不明确的情况下自动完成匹配，实测足够应对大多数场景。
5.2 判断标准：什么情况不该走表格导入
有一套判断逻辑值得在动手前走一遍，能帮你省下大量的无用功。
- 如果用户的查询都是"上个月总营收多少""A类客户占比"这类聚合统计问题，你其实不需要导入明细表，只需要把汇总结果做成一页PDF或一段Markdown文本导入就够，效果更好、成本更低。
- 如果用户的查询需要精确匹配某一行（比如"订单号A1001的收货地址是什么"），行级文本化方案能回答，但远不如直接连库查准确。此时应该连库。
- 如果表格里的列数超过20列，而且每列的值都有独立业务含义，行级文本化会让每行文本变得非常长且冗余，建议先做列筛选，只保留问答高频涉及的列。
- 如果表格会每天更新，不要每次全量重建索引，优先设计增量导入或连库实时读取。
这个判断标准不是拍脑袋想的，而是来自一次很惨痛的教训。当时我把一张50列的风险评估表全量导入，结果索引文件建了几百MB，查询时经常把不相关的列卷进来，准确率反而比只保留核心8列的时候还要低。后来学乖了，先做列筛选再做行级文本化，效果立刻好了两个档次。
表格导入这件事，表面看只是"把文件读进来转成Document"，实际上踩过的坑不比搭整个RAG管线少。CSV讲究清洗和切片，Excel讲究解构和语义注入，数据库连库讲究选型和安全。希望这篇能把你在表格数据导入路上可能遇到的问题提前挡掉一部分。最后再分享一个操作习惯：不管数据规模多大，导入完务必抽3到5条数据人工检查一下最终入库的Node文本，看看有没有乱码、空值、断行。这一步看起来费时间，但能避免你在后面问答调试时耗费更多的精力。
