1. RAG 数据导入与解析全攻略表格与数据库导入的整体设计思路做 RAG 项目的人迟早会撞上一堵墙PDF 和网页文本还没处理利索业务方已经甩过来一堆 Excel 报表和 MySQL 连接串了。我见过太多团队在 demo 阶段用几个 txt 文件跑通了检索问答一到真实场景就卡在“数据进不来”这一步。表格和数据库这两类数据源恰恰是企业知识库里占比最大、价值密度最高的部分——财务报表、产品目录、客户清单、订单记录几乎全是结构化的行列数据。这一篇要聊的就是怎么把 CSV、Excel 和数据库里的数据干净利落地喂进 RAG 管道。核心工具链是LlamaIndex 生态下的 LlamaHub配合pandas做预处理用DatabaseReader连库最终产出带元数据的Document对象直接对接后续的向量化和索引构建。整套方案解决的核心问题是结构化数据如何保留语义、如何分块、如何带上可检索的元信息。适合谁看如果你已经跑通过基础的 RAG 流程知道Document、Node、VectorStoreIndex这些概念但面对表格数据不知道怎么下手这篇就是给你写的。零基础也能跟我会把每一步的参数和意图都讲清楚。先说整体思路。表格数据进 RAG最大的坑在于“直接转文本”。很多人图省事df.to_string()一扔就完事结果检索出来的全是密密麻麻的数字语义完全丢失。正确的做法是按行或按业务实体切分每一行或每一组相关行变成一个独立的Document同时把列名作为字段标签嵌入文本把关键列抽出来放进metadata。这样检索时既能命中内容又能通过元数据过滤。数据库这边思路类似但多了一层 SQL 查询的灵活性。DatabaseReader允许你自定义查询语句把多表关联的结果直接拉成扁平结构再按行转Document。这里的关键决策是在 SQL 层做聚合还是拉平后在 Python 层做。我的经验是能在 SQL 里JOIN和WHERE的绝不拖到内存里处理尤其是数据量上百万行的时候内存会教你做人。LlamaHub 在这里扮演的角色是“连接器市场”。它提供了大量现成的ReaderCSV、Excel、数据库、Notion、Airtable 应有尽有。你不需要自己从零写解析逻辑直接download_loader拉下来用就行。但要注意LlamaHub 的 loader 质量参差不齐有些年久失修依赖版本对不上。所以我的习惯是优先用 pandas 自己读读完之后手动构造 Document只有在处理特殊格式比如复杂 Excel 多 sheet、数据库方言时才动用 LlamaHub 的专用 Reader。提示LlamaHub 的 loader 安装后会放在llama_index.readers命名空间下但不同版本的导入路径可能不同。建议锁定llama-index的版本避免今天能跑的代码明天就ImportError。2. CSV 与 Excel 导入的核心细节与实操要点2.1 CSV 读取的编码与分隔符陷阱CSV 看起来最简单实际上坑最多。第一个坑是编码。国内业务系统导出的 CSV十有八九是GBK或GB2312你直接用pd.read_csv(file.csv)会报UnicodeDecodeError。我的做法是先探测编码import chardet with open(data.csv, rb) as f: raw f.read(10000) encoding chardet.detect(raw)[encoding] print(f检测到编码: {encoding}) df pd.read_csv(data.csv, encodingencoding)chardet不是万能的短文件可能误判。更稳的方案是写个循环依次尝试[utf-8, gbk, gb2312, latin1]哪个不报错用哪个。latin1是兜底它永远不会报错但中文会变乱码所以只用来确认文件能读进来。第二个坑是分隔符。有些系统导出的是制表符分隔的.tsv或者用分号;分隔。pd.read_csv的sep参数支持正则可以写成sepr\s来兼容空格和制表符混用的情况。但要注意如果字段内容里本身包含分隔符就需要quoting参数配合。第三个坑是大文件内存溢出。一个 500MB 的 CSV 读进 pandas 可能占 2GB 内存。解决方案是分块读取chunk_size 10000 documents [] for chunk in pd.read_csv(big_file.csv, chunksizechunk_size, encodingutf-8): for _, row in chunk.iterrows(): text row_to_text(row) doc Document(texttext, metadataextract_metadata(row)) documents.append(doc)分块的好处不只是省内存还能边读边处理避免一次性加载完再转换的峰值压力。chunk_size设多少合适我的经验是 5000 到 20000 之间太小了 I/O 次数多太大了内存优势不明显。2.2 行转 Document 的文本构造策略这是整个流程里最需要动脑子的地方。一行数据怎么变成一段有语义的文本直接str(row)肯定不行。我的标准做法是构造“字段名字段值”的拼接串def row_to_text(row, columns): parts [] for col in columns: value row[col] if pd.notna(value) and str(value).strip(): parts.append(f{col}{value}) return .join(parts)这样一行“张三28北京工程师”就变成了“姓名张三年龄28城市北京职位工程师”。检索时用户问“北京的工程师有哪些”向量模型能匹配到“城市北京”和“职位工程师”这两个语义片段。但这里有个细节数值型字段要不要带单位。如果列名是“销售额”值是“12000”那文本里最好写成“销售额12000元”。单位信息如果不在列名里就要在构造时补上。我一般会维护一个column_unit_map字典专门处理这种情况。另一个细节是空值处理。pd.notna()过滤掉 NaN但有些系统用“-”或“N/A”表示空。这些要在读取阶段就用na_values参数统一成 NaNdf pd.read_csv(data.csv, na_values[-, N/A, null, NULL, ])2.3 Excel 多 Sheet 与合并单元格的处理Excel 比 CSV 复杂得多。第一个问题是多 sheet。pd.read_excel默认只读第一个 sheet要读全部得用sheet_nameNone返回一个字典sheets_dict pd.read_excel(workbook.xlsx, sheet_nameNone, engineopenpyxl) for sheet_name, df in sheets_dict.items(): for _, row in df.iterrows(): text row_to_text(row, df.columns) doc Document( texttext, metadata{source: workbook.xlsx, sheet: sheet_name} ) documents.append(doc)把 sheet 名放进 metadata 很关键检索时可以按 sheet 过滤比如只搜“2024年Q1”这个 sheet 的数据。第二个问题是合并单元格。Excel 里常见的表头跨列合并pandas 读进来后合并区域只有第一个单元格有值其余是 NaN。这会导致列名变成Unnamed: 1、Unnamed: 2。解决方案是读完后做前向填充df pd.read_excel(file.xlsx, header0) df.columns pd.Series(df.columns).ffill()但更稳妥的做法是跳过表头手动指定列名。如果表格结构固定直接headerNone然后df.columns [列1, 列2, ...]避免自动解析带来的意外。第三个问题是公式单元格。openpyxl默认读公式的计算结果但如果文件是用其他工具生成的可能读到公式本身。这时候要加data_onlyTruedf pd.read_excel(file.xlsx, engineopenpyxl, engine_kwargs{data_only: True})注意data_onlyTrue只在文件被 Excel 打开并保存过、缓存了计算结果时才有效。如果文件是程序生成的且从未用 Excel 打开读到的还是 None。这种情况只能换xlrd引擎或者让业务方重新导出。2.4 元数据抽取的取舍原则不是所有列都适合放进 metadata。metadata 会被存进向量库每条都占空间而且过滤时是精确匹配。我的原则是只放高基数、可能用于过滤的列。比如“部门”“年份”“地区”适合放 metadata“备注”这种长文本就不适合。def extract_metadata(row, meta_columns): meta {} for col in meta_columns: if col in row and pd.notna(row[col]): meta[col] str(row[col]) return metameta_columns怎么定看业务查询习惯。如果用户经常问“华东区的销售数据”那“地区”就必须进 metadata。如果从来没人按“创建时间”过滤那就别放省空间。3. LlamaHub 连库实战DatabaseReader 的配置与查询3.1 DatabaseReader 的安装与连接配置LlamaHub 的DatabaseReader在llama-index-readers-database这个包里。安装pip install llama-index-readers-database pip install sqlalchemy pip install pymysql # MySQL 驱动连接配置用 SQLAlchemy 的 URI 格式from llama_index.readers.database import DatabaseReader reader DatabaseReader( schememysql, hostlocalhost, port3306, userreadonly_user, passwordyour_password, dbnamebusiness_db )这里有个安全要点永远用只读账号。RAG 系统只需要SELECT权限给个readonly账号避免误操作写库。我见过有人用 root 连库结果查询语句写错把生产数据改了这种事一次就够记一辈子。3.2 自定义 SQL 查询与结果扁平化DatabaseReader的核心方法是load_data接受一个 SQL 查询字符串query SELECT o.order_id, o.order_date, c.customer_name, c.city, p.product_name, oi.quantity, oi.unit_price, oi.quantity * oi.unit_price AS total_amount FROM orders o JOIN customers c ON o.customer_id c.customer_id JOIN order_items oi ON o.order_id oi.order_id JOIN products p ON oi.product_id p.product_id WHERE o.order_date 2024-01-01 documents reader.load_data(queryquery)load_data返回的Document列表每个 Document 对应查询结果的一行。默认情况下它会把整行拼成文本格式是col1: val1\ncol2: val2。这个格式其实还不错但如果你想自定义可以传text_template参数。这里的关键决策是SQL 里做多少Python 里做多少。我的原则是关联查询JOIN必须在 SQL 里做Python 处理多表关联效率极低聚合计算SUM、COUNT尽量在 SQL 里做减少传输数据量文本拼接、元数据抽取在 Python 里做SQL 不擅长字符串处理但有个例外如果聚合后的结果行数很少比如按地区汇总只有几十行那在 SQL 里聚合反而不好因为聚合后丢失了明细语义。用户问“华东区销售额多少”聚合后的“华东区120万”能回答但用户问“华东区哪些订单超过1万”聚合数据就答不了。所以聚合与否取决于检索问题的粒度。3.3 大表分页与增量导入直接SELECT * FROM huge_table是自杀行为。几百万行数据一次性拉进内存Python 进程直接 OOM。正确做法是分页def load_in_batches(reader, base_query, batch_size5000): offset 0 all_docs [] while True: query f{base_query} LIMIT {batch_size} OFFSET {offset} docs reader.load_data(queryquery) if not docs: break all_docs.extend(docs) offset batch_size print(f已加载 {offset} 行) return all_docs但OFFSET在数据量大时性能很差因为数据库要扫描前 offset 行再丢弃。更好的方案是用游标分页基于主键或时间戳last_id 0 while True: query fSELECT * FROM orders WHERE order_id {last_id} ORDER BY order_id LIMIT 5000 docs reader.load_data(queryquery) if not docs: break last_id extract_last_id(docs) all_docs.extend(docs)这种方式每次查询都走索引效率稳定。前提是表有自增主键或单调递增的字段。增量导入则是记录上次导入的最大 ID 或时间戳下次只拉新增部分。这对每天更新的业务库特别有用避免全量重跑。3.4 数据库字段类型与文本转换的坑数据库里的字段类型五花八门转文本时要注意DATETIME类型转成字符串时格式要统一建议strftime(%Y-%m-%d %H:%M:%S)DECIMAL类型在 Python 里是Decimal对象直接拼字符串没问题但要注意精度BLOB或JSON类型要特殊处理JSON 可以json.dumps后拼入BLOB 一般跳过NULL值要过滤不要拼成“字段名None”def safe_str(value): if value is None: return if isinstance(value, datetime): return value.strftime(%Y-%m-%d %H:%M:%S) if isinstance(value, Decimal): return f{value:.2f} return str(value)4. 常见问题与排查技巧实录4.1 编码与格式类问题速查问题现象可能原因解决方案UnicodeDecodeError: utf-8 codec cant decode文件是 GBK 编码用chardet探测或尝试encodinggbk中文显示为乱码编码不匹配确认文件实际编码用对应编码读取列名变成Unnamed: 0表头行有合并单元格或空列手动指定names参数或做前向填充Excel 读取报InvalidFileException引擎不匹配.xlsx用openpyxl.xls用xlrd日期列读成数字Excel 日期序列号用pd.to_datetime转换或指定parse_dates大文件读取内存溢出一次性加载用chunksize分块读取4.2 数据库连接与查询类问题连接超时是最常见的。原因通常是防火墙或数据库配置了wait_timeout。解决方案是在 SQLAlchemy 的 URI 里加连接池参数reader DatabaseReader( schememysql, hostlocalhost, port3306, userreadonly, passwordpass, dbnamedb, engine_kwargs{ pool_size: 5, max_overflow: 10, pool_recycle: 3600, pool_pre_ping: True } )pool_pre_pingTrue特别重要它会在每次从连接池取连接时先 ping 一下避免用到已经断开的连接。查询结果为空但 SQL 在客户端能跑出数据通常是权限问题。只读账号可能没有某些表的SELECT权限或者WHERE条件里的时间格式不对。建议先在数据库客户端用同样的账号跑一遍查询。字段名大小写问题。MySQL 在 Linux 下默认区分大小写Windows 下不区分。如果代码里写row[OrderID]但实际列名是orderid就会 KeyError。解决方案是统一转小写或用row._asdict()后做大小写不敏感匹配。4.3 元数据与检索效果类问题检索不到表格数据最常见的原因是文本构造太差。如果一行数据转成文本后只有数字没有语义向量模型根本匹配不到。检查方法把几个 Document 的text打印出来人工读一遍看能不能理解这行数据在说什么。元数据过滤失效通常是类型问题。metadata 里的值必须是字符串或数字不能是datetime对象或Decimal。存进去之前统一str()转换。重复数据。数据库增量导入时如果没做好去重同一行可能被多次导入。解决方案是用doc_id做唯一标识LlamaIndex 的Document支持doc_id参数重复的会被覆盖。实操心得我习惯在导入完成后随机抽 10 条 Document把text和metadata打印出来人工检查。这一步花 5 分钟能省掉后面几小时的调试。4.4 性能优化与批量处理技巧批量嵌入比逐条嵌入快得多。LlamaIndex 的VectorStoreIndex支持批量插入但如果你用的是自定义嵌入模型要确保embed_batch_size设置合理。太大容易超时太小效率低。我的经验值是 100 到 500 之间。并行读取。多个 CSV 文件可以并行处理from concurrent.futures import ThreadPoolExecutor def process_file(filepath): df pd.read_csv(filepath) return [row_to_document(row) for _, row in df.iterrows()] with ThreadPoolExecutor(max_workers4) as executor: results executor.map(process_file, file_list) all_docs [doc for docs in results for doc in docs]但要注意Python 的 GIL 限制了 CPU 密集型任务的并行效果。如果是 I/O 密集读文件、连数据库多线程有效如果是 CPU 密集文本处理、嵌入计算多进程更合适。索引持久化。每次跑都重新嵌入太浪费时间。LlamaIndex 支持把索引存到磁盘index.storage_context.persist(persist_dir./storage)下次直接load_index_from_storage省掉嵌入步骤。但要注意如果数据更新了需要重建索引或做增量更新。5. 从导入到检索的完整链路串联5.1 数据管道的目录结构设计一个可维护的 RAG 数据管道目录结构应该清晰rag_pipeline/ ├── data/ │ ├── raw/ # 原始 CSV、Excel 文件 │ ├── processed/ # 清洗后的中间文件 │ └── storage/ # 索引持久化目录 ├── readers/ │ ├── csv_reader.py │ ├── excel_reader.py │ └── db_reader.py ├── config/ │ └── settings.yaml # 数据库连接、列映射配置 └── main.py把连接信息和列映射抽到配置文件里避免硬编码。换一个数据库或换一张表只改配置不改代码。5.2 导入后的验证与质量检查数据导入不是终点验证才是。我一般做三层检查第一层数量检查。导入的 Document 数量是否和源数据行数一致如果不一致是过滤了空行还是漏读了 sheet第二层内容抽样。随机抽 20 条人工看文本是否通顺、元数据是否完整。第三层检索测试。构造几个典型问题跑一遍检索看返回的结果是否相关。这一步最能暴露问题——有时候文本看起来没问题但检索就是不准说明文本构造策略需要调整。# 检索测试示例 test_queries [ 2024年华东区的销售额是多少, 张三的订单有哪些, 哪些产品的单价超过1000 ] for q in test_queries: response query_engine.query(q) print(f问题: {q}) print(f回答: {response}) print(---)如果检索结果不理想优先检查文本里有没有包含问题中的关键词元数据过滤条件是否太严格分块大小是否合适5.3 增量更新与版本管理业务数据每天都在变RAG 知识库不能只建一次。增量更新的策略取决于数据源CSV/Excel 文件用文件修改时间判断只处理新增或修改的文件数据库用时间戳或自增 ID 做增量拉取混合场景维护一个last_sync记录每次只拉变化部分版本管理方面我建议每次全量重建时保留旧索引的备份。如果新索引效果变差可以快速回滚。索引文件不大存几个版本的成本可以接受。提示增量更新时要注意删除操作。如果源数据删了一行RAG 索引里对应的 Document 也要删。LlamaIndex 支持delete操作但需要知道doc_id。所以导入时一定要记录doc_id和源数据的映射关系。5.4 实际项目中的取舍经验做了几个 RAG 项目后我总结出几条取舍原则不要追求全量导入。业务库里 80% 的数据可能从来没人查。先导入最近一年的、活跃度高的数据跑通流程后再逐步扩展。全量导入不仅慢还会稀释检索质量——太多无关文档会干扰向量匹配。元数据宁少勿多。每加一个 metadata 字段就多一份存储和过滤开销。只加真正会用于过滤的字段。文本构造要贴近用户提问习惯。用户怎么问文本就怎么组织。如果用户习惯说“帮我查一下北京的订单”那文本里“城市北京”就比“北京”单独出现更容易匹配。定期重建索引。向量模型会更新数据分布会变化索引跑久了效果会下降。我一般每季度做一次全量重建中间用增量更新维持。这套流程跑下来CSV、Excel 和数据库的数据就能稳定地进入 RAG 管道了。核心就三件事读得进来、转得合理、查得准确。每一步的细节都值得反复打磨尤其是文本构造和元数据设计这两个地方偷懒后面检索效果一定打折扣。