企业官网建设流程全解析

1. 为什么表格和数据库是 RAG 数据导入的硬骨头

做过 RAG 项目的人都有一个共识：PDF 和网页文本的解析虽然烦，但至少路子是通的，无非是版面分析、分块策略、元数据抽取这些活儿。真正让人头疼的，是那些结构化的数据源——CSV、Excel 和业务数据库。这些东西看起来规规矩矩，但一旦要喂给向量库做检索增强，问题就全冒出来了。

我最近在做一个企业知识库项目，数据源里有一大半是运营同事维护的 Excel 报表和 MySQL 里的业务表。刚开始我以为这部分最简单，毕竟结构化数据嘛，读出来转成文本不就完了。结果踩了一圈坑才发现，表格数据的 RAG 导入有一套完全不同的逻辑：你面对的不是"文本"，而是"关系"。一行数据里每个字段的含义、字段之间的关联、表与表之间的外键关系，这些信息如果处理不好，检索出来的结果就是一堆没有上下文的数字，模型根本没法用。

这篇是 RAG 数据导入与解析系列的第三篇，专门讲 CSV、Excel 和 LlamaHub 连库这三条路径。我会把每种数据源的解析思路、工具选型、实操代码和踩坑经验都摊开讲。适合正在搭建 RAG 知识库、需要把业务数据接进来的开发者，也适合想搞清楚"结构化数据到底怎么变成向量"这个问题的朋友。不管你是用 LangChain、LlamaIndex 还是自己手搓流程，这里面的思路都能直接套用。

2. 结构化数据接入 RAG 的整体设计思路

2.1 结构化数据与文本数据的本质差异

先说清楚一个核心认知：文本数据的分块逻辑是"语义完整"，结构化数据的分块逻辑是"记录完整"。一篇文档你切成 500 字一段，每段还能独立表达意思；但一张表你按行切，单独一行数据脱离表头就是天书。

举个例子，一张销售表里有这么一行：

2024-03-15 | 华东区 | A产品 | 1280 | 35600.00

你把这行直接扔进向量库，检索的时候用户问"华东区A产品卖了多少"，向量匹配可能命中，但模型拿到这行数据根本不知道 1280 是销量还是库存，35600 是金额还是成本。所以结构化数据导入 RAG 的第一原则是：每一块数据都必须自带完整的上下文描述。

我的做法是把每行数据转换成一个"自解释"的文本块，格式大概是这样：

销售记录：日期2024年3月15日，区域为华东区，产品为A产品，销售数量1280件，销售金额35600元。

这样每个块都是完整的语义单元，检索和生成都不会出问题。代价是存储膨胀，但这是值得的。

2.2 三种数据源的选型考量

CSV、Excel、数据库这三类数据源，处理难度和适用场景差别很大，我整理了一张对照表：

选型的核心判断标准是：数据是否还在被人持续维护。如果 Excel 是运营每天在更新的，那你要考虑增量导入；如果 CSV 是一次性导出的，全量处理就行；如果是数据库，那基本都要做增量同步，否则每次全量拉取既慢又浪费 embedding 成本。

2.3 从原始数据到向量块的完整链路

不管哪种数据源，最终都要走同一条链路：

数据读取 → 结构解析 → 行级/记录级转换 → 上下文补全 → 分块 → Embedding → 向量库

这里面最容易被忽略的是"上下文补全"这一步。很多人直接把 DataFrame 转成字符串就完事了，结果检索效果很差。上下文补全要做的事情包括：把表头信息嵌入每行、把编码值翻译成可读文本（比如状态码 1 变成"已激活"）、把关联表的信息 join 进来。这一步做得好不好，直接决定了 RAG 的上限。

3. CSV 导入实战：从编码坑到百万行处理

3.1 CSV 读取的编码与分隔符陷阱

CSV 看起来是最简单的格式，但实际项目里，CSV 的坑一点不比 PDF 少。最常见的就是编码问题。国内业务系统导出的 CSV，十有八九是 GBK 或 GB2312 编码，你用默认的 UTF-8 去读，直接报UnicodeDecodeError。

我的处理策略是写一个自动探测编码的函数，用chardet库先检测，再决定用什么编码读：

import chardet import pandas as pd def read_csv_smart(file_path): with open(file_path, 'rb') as f: raw = f.read(100000) result = chardet.detect(raw) encoding = result['encoding'] # GB2312 和 GBK 兼容，统一用 GBK 读更稳 if encoding and encoding.upper() in ['GB2312', 'GB18030']: encoding = 'GBK' return pd.read_csv(file_path, encoding=encoding)

分隔符也是个大坑。标准 CSV 用逗号，但有些系统导出用分号、制表符甚至竖线。更麻烦的是，字段内容里本身可能包含逗号，这时候就要靠引号包裹。pandas 的read_csv默认能处理引号包裹的情况，但如果分隔符本身就不对，读出来的列数会错乱。我的经验是先用csv.Sniffer探测一下：

import csv def detect_delimiter(file_path, encoding='utf-8'): with open(file_path, 'r', encoding=encoding) as f: sample = f.read(10000) sniffer = csv.Sniffer() dialect = sniffer.sniff(sample) return dialect.delimiter

提示：如果 CSV 文件里有大量中文，建议读取时显式指定encoding参数，不要依赖自动检测，因为 chardet 对短样本的检测准确率并不高。

3.2 大文件分块读取与内存控制

百万行级别的 CSV 直接pd.read_csv会把内存吃满。我试过读一个 200 万行的销售明细，16G 内存的机器直接卡死。解决办法是用chunksize参数分块读取：

def process_large_csv(file_path, chunk_size=10000): results = [] for chunk in pd.read_csv(file_path, chunksize=chunk_size, encoding='utf-8'): # 对每个 chunk 做处理 processed = transform_chunk(chunk) results.extend(processed) return results

分块读取的关键是transform_chunk这个函数要设计成无状态的，每个 chunk 独立处理，不依赖其他 chunk 的数据。如果你的处理逻辑需要全局统计（比如归一化），那就得先扫一遍拿到统计量，再扫第二遍做转换。

3.3 行转文本的模板设计

CSV 转文本块的核心是模板设计。我一般用两种模板：键值对模板和自然语言模板。

键值对模板适合字段多、结构固定的场景：

def row_to_kv_text(row, columns): parts = [f"{col}：{row[col]}" for col in columns if pd.notna(row[col])] return "；".join(parts)

自然语言模板适合需要语义连贯的场景：

def row_to_nl_text(row): return (f"这是一条{row['类型']}记录，" f"发生在{row['日期']}，" f"涉及{row['对象']}，" f"金额为{row['金额']}元，" f"当前状态为{row['状态']}。")

实测下来，自然语言模板的检索效果普遍更好，因为 embedding 模型对完整句子的语义捕捉能力更强。但键值对模板的信息密度更高，适合字段特别多的宽表。我的建议是：字段少于 10 个用自然语言模板，超过 10 个用键值对模板。

3.4 元数据保留与增量导入策略

CSV 导入时一定要保留元数据，至少包括：源文件名、导入时间、行号。这些信息在排查问题和做数据溯源时非常有用。我在每个文本块的 metadata 里都会塞这些字段：

metadata = { "source": file_path, "import_time": datetime.now().isoformat(), "row_index": idx, "file_type": "csv" }

增量导入的话，CSV 本身没有天然的增量标识，需要靠业务字段来判断。常见做法是用时间戳字段做水位线，每次只导入比上次水位线新的数据。如果 CSV 里没有时间字段，那就只能靠文件级别的去重——记录已导入文件的 hash 值，避免重复导入同一个文件。

4. Excel 导入实战：多 Sheet 与合并单元格的破解之道

4.1 Excel 与 CSV 的处理差异

Excel 比 CSV 复杂的地方在于：一个文件可以有多个 sheet，单元格可以有格式、公式、合并单元格，还有隐藏行列。这些特性让 Excel 的解析难度上了一个台阶。

我用得最多的库是openpyxl和pandas。pandas的read_excel底层也是调openpyxl，但封装得更友好。不过pandas有个问题：它默认只读第一个 sheet，多 sheet 场景需要手动指定sheet_name=None来读取所有 sheet：

import pandas as pd def read_all_sheets(file_path): sheets = pd.read_excel(file_path, sheet_name=None, engine='openpyxl') # sheets 是一个 dict，key 是 sheet 名，value 是 DataFrame return sheets

4.2 多 Sheet 批量处理与命名规范

多 sheet 的 Excel 通常有两种组织方式：一种是每个 sheet 结构相同，只是数据分片（比如按月份分 sheet）；另一种是每个 sheet 结构不同，代表不同的业务实体。

对于第一种，处理逻辑很简单，把所有 sheet 合并成一个 DataFrame 就行：

def merge_same_structure_sheets(sheets): dfs = [] for sheet_name, df in sheets.items(): df['_sheet_name'] = sheet_name dfs.append(df) return pd.concat(dfs, ignore_index=True)

对于第二种，每个 sheet 要单独设计转换模板。这时候 sheet 名本身就是重要的元数据，要保留到文本块里。我的做法是在文本块开头加上 sheet 名作为上下文：

def sheet_to_text(sheet_name, df): texts = [] for idx, row in df.iterrows(): text = f"【{sheet_name}】" + row_to_nl_text(row) texts.append(text) return texts

4.3 合并单元格的数据填充

合并单元格是 Excel 解析里最恶心的问题。比如一个表里"部门"这一列，第一行是"销售部"，合并了 5 行，pandas读出来只有第一行有值，后面 4 行都是 NaN。如果你直接处理，后面 4 行就丢了部门信息。

解决办法是用前向填充（forward fill）：

def fill_merged_cells(df): # 只对object类型的列做前向填充，避免影响数值列 for col in df.columns: if df[col].dtype == 'object': df[col] = df[col].ffill() return df

但这里有个坑：如果某一列本来就有合法的空值，前向填充会把空值也填上，导致数据失真。所以更稳妥的做法是先用openpyxl读取合并单元格的范围，精确知道哪些单元格是合并的，再针对性填充：

from openpyxl import load_workbook def get_merged_ranges(file_path, sheet_name): wb = load_workbook(file_path) ws = wb[sheet_name] return [str(rng) for rng in ws.merged_cells.ranges]

拿到合并范围后，就能精确地把合并区域的值填充到每个单元格。这个方法比无脑 ffill 靠谱得多，尤其是在数据本身允许空值的情况下。

4.4 公式单元格的处理

Excel 里的公式单元格，pandas默认读出来的是公式的计算结果（如果文件被 Excel 打开保存过），但如果文件是程序生成的、从未被 Excel 打开，读出来的可能是公式本身或者 None。

处理策略是：优先读缓存值，读不到就跳过。openpyxl有个data_only=True参数，可以读取公式的缓存值：

wb = load_workbook(file_path, data_only=True)

但要注意，如果文件从未被 Excel 计算过，缓存值是 None。这种情况下，公式单元格的数据是拿不到的，只能提示用户先用 Excel 打开保存一次。这是 Excel 格式本身的限制，没有完美的解决方案。

4.5 Excel 转文本的实操模板

综合上面的处理，一个完整的 Excel 转文本流程大概是这样：

def excel_to_text_blocks(file_path): sheets = pd.read_excel(file_path, sheet_name=None, engine='openpyxl') all_blocks = [] for sheet_name, df in sheets.items(): df = fill_merged_cells(df) for idx, row in df.iterrows(): text = f"工作表：{sheet_name}。" + row_to_nl_text(row) metadata = { "source": file_path, "sheet": sheet_name, "row_index": idx, "file_type": "excel" } all_blocks.append({"text": text, "metadata": metadata}) return all_blocks

注意：Excel 文件如果很大（超过 10MB），openpyxl的读取速度会明显变慢。这时候可以考虑用read_only=True模式，或者先把 Excel 转成 CSV 再处理。

5. LlamaHub 连库实战：DatabaseReader 的正确打开方式

5.1 LlamaHub 与 DatabaseReader 简介

LlamaHub 是 LlamaIndex 生态里的数据连接器集合，里面有几百个 Reader，覆盖各种数据源。做数据库导入，核心用的是DatabaseReader，它底层基于 SQLAlchemy，支持 MySQL、PostgreSQL、SQLite、SQL Server 等主流数据库。

安装很简单：

pip install llama-index-readers-database sqlalchemy pymysql

DatabaseReader的基本用法是传入一个 SQLAlchemy 的 engine 和查询语句，它会执行查询并把每行结果转成 LlamaIndex 的 Document 对象：

from llama_index.readers.database import DatabaseReader from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://user:password@host:3306/dbname") reader = DatabaseReader(engine=engine) documents = reader.load_data(query="SELECT * FROM sales_records LIMIT 1000")

5.2 连接配置与查询设计

连接配置这块，有几个参数必须注意。首先是字符集，MySQL 连接一定要指定charset=utf8mb4，否则中文会乱码：

engine = create_engine( "mysql+pymysql://user:password@host:3306/dbname?charset=utf8mb4", pool_size=5, max_overflow=10, pool_recycle=3600 )

pool_recycle这个参数很关键，MySQL 默认 8 小时会断开空闲连接，如果不设置，长时间运行的程序会报MySQL server has gone away。设成 3600 秒（1 小时）比较稳妥。

查询设计上，千万不要用SELECT *。原因有两个：一是宽表字段太多，转成文本后信息冗余严重；二是有些字段是二进制或者超长文本，读出来会拖慢速度。正确的做法是显式列出需要的字段，并且用JOIN把关联表的信息带进来：

SELECT s.id, s.sale_date, s.quantity, s.amount, p.product_name, p.category, r.region_name FROM sales_records s JOIN products p ON s.product_id = p.id JOIN regions r ON s.region_id = r.id WHERE s.sale_date >= '2024-01-01'

这样查出来的每一行都是自包含的，不需要额外的关联查询就能生成完整的文本块。

5.3 表关系处理与上下文补全

数据库最大的优势是有明确的外键关系，这让我们可以做更丰富的上下文补全。比如一张订单表，关联了用户表、商品表、地址表，我可以在查询阶段就把这些信息 join 进来，生成这样的文本块：

订单记录：订单号 SO20240315001，下单时间 2024年3月15日 10:23:45， 客户为张三（VIP等级：金卡），商品为 iPhone 15 Pro（类别：数码）， 数量 1 台，金额 8999 元，收货地址为上海市浦东新区。

这种文本块的检索效果非常好，因为用户的问题往往涉及多个维度的信息，而 join 后的文本块天然包含了这些维度。

5.4 增量同步与水位线机制

数据库导入一定要做增量，否则每次全量拉取，几百万行数据光 embedding 就要花不少钱。增量同步的核心是水位线（watermark）机制：记录上次同步的最大 ID 或最大时间戳，下次只拉比这个值新的数据。

def incremental_sync(engine, last_sync_id): query = f""" SELECT * FROM sales_records WHERE id > {last_sync_id} ORDER BY id ASC LIMIT 10000 """ reader = DatabaseReader(engine=engine) documents = reader.load_data(query=query) if documents: new_last_id = documents[-1].metadata['id'] save_watermark(new_last_id) return documents

水位线要持久化存储，可以放 Redis、SQLite 或者一个简单的 JSON 文件。我一般用 SQLite，轻量且可靠。

提示：如果表的主键不是自增 ID，而是 UUID，那就不能用水位线了。这时候可以用时间戳字段做水位线，但要注意时间戳可能有重复，需要用>=而不是>，并且配合去重逻辑。

5.5 数据库导入的性能优化

数据库导入的性能瓶颈通常在两个地方：查询速度和 embedding 速度。查询这块，确保 WHERE 条件和 ORDER BY 字段上有索引，否则大表扫描会非常慢。embedding 这块，建议批量处理，不要一行一行调 embedding API：

def batch_embed(texts, batch_size=100): all_embeddings = [] for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] embeddings = embed_model.get_text_embedding_batch(batch) all_embeddings.extend(embeddings) return all_embeddings

批量 embedding 比单条调用快 5 到 10 倍，这是实测数据。另外，如果数据量特别大，可以考虑用本地 embedding 模型（比如 BGE 系列），避免 API 调用的网络开销和费用。

6. 常见问题与排查技巧实录

6.1 编码与格式类问题速查

6.2 数据质量类问题排查

结构化数据导入 RAG 最怕的是数据质量问题。我遇到过几种典型情况：

第一种是空值泛滥。某张表 30% 的字段是 NULL，转成文本后就是一堆"字段X：None"，这种块检索出来毫无价值。处理办法是在转换时过滤掉空值字段，如果一行里有效字段少于 3 个，直接跳过这行。

第二种是脏数据。比如日期字段里混了"未知"、"待定"这种文本，数值字段里有"约100"这种带修饰的值。这类数据要么在 SQL 层面用 CASE WHEN 清洗，要么在 Python 层面用正则处理。

第三种是重复数据。同一行数据在表里出现多次，导入后向量库里有一堆重复块。解决办法是在导入前用drop_duplicates去重，或者用业务主键做唯一性校验。

6.3 检索效果差的排查思路

如果导入完成后检索效果不好，按这个顺序排查：

1. 检查文本块质量：随机抽 10 个块，看内容是否完整、是否有意义。如果块本身是垃圾，检索不可能好。
2. 检查 embedding 模型：中文场景一定要用支持中文的模型，用纯英文模型效果会差很多。
3. 检查分块粒度：结构化数据的块不宜太大也不宜太小。一行一块通常合适，但如果一行内容特别多，可以考虑按字段组拆分。
4. 检查元数据过滤：如果检索时加了元数据过滤条件，确认过滤条件是否正确，有时候是过滤太严导致召回不足。
5. 检查相似度阈值：阈值设太高会漏召回，设太低会引入噪声。一般从 0.7 开始调。

6.4 我的独家避坑经验

分享几个文档里不会写、但实际项目中非常管用的经验：

第一，永远保留原始数据的一份快照。导入过程中如果发现转换逻辑有问题，需要重新导入，没有原始快照就得重新找数据源，非常麻烦。

第二，给每个文本块加一个唯一 ID。这个 ID 可以是源文件+行号的组合，方便后续更新和删除。向量库的更新操作都是基于 ID 的，没有 ID 就只能全量重建。

第三，小批量试跑再全量导入。先用 100 行数据跑通整个流程，确认检索效果没问题，再全量导入。我见过太多人直接全量导入，结果发现模板设计有问题，几百万条数据白导了。

第四，监控 embedding 成本。结构化数据量大，embedding 费用容易失控。导入前先估算一下：总行数 × 平均 token 数 × 单价。如果成本太高，考虑用本地模型或者做数据抽样。

第五，数据库连接要加超时和重试。网络抖动、数据库重启都会导致连接中断，加个重试机制能省很多事：

from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10)) def safe_query(engine, query): reader = DatabaseReader(engine=engine) return reader.load_data(query=query)

这套流程我在三个项目里跑过，从几万行的 Excel 到千万行的数据库表都处理过，稳定性没问题。核心还是那句话：结构化数据导入 RAG，难点不在技术，在于对数据本身的理解。你得知道每个字段是什么意思、字段之间什么关系、用户会怎么问问题，才能设计出好的转换模板。工具只是手段，数据理解才是根本。