企业官网建设流程全解析

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

做过 RAG 项目的人都有一个共识：PDF 和纯文本的解析虽然烦，但至少路子是清晰的，大不了上 OCR、调分段策略。真正让人头疼的是表格和数据库这类结构化数据。你拿到的 CSV 可能有几十列、编码格式五花八门、缺失值满天飞；Excel 更离谱，一个文件里塞了七八个 sheet，每个 sheet 的表头还不一样，有的合并单元格，有的前几行是说明文字。数据库那边呢，表结构复杂、字段类型不统一、还有各种关联查询的需求。

我前后在三个 RAG 项目里处理过结构化数据导入，踩的坑足够写一本小册子。这一篇就把 CSV、Excel 和通过 LlamaHub 连接数据库这三条路完整走一遍，从数据探查、清洗、分块策略到最终灌入向量库，每一步都给出可直接复现的操作和参数。

这篇文章适合已经在做 RAG 项目、需要把结构化数据接入知识库的开发者。如果你还没搭好基础的 RAG 框架，建议先看前两篇关于文档解析和向量化的内容，这篇默认你已经有了基本的 RAG 管线。

提示：结构化数据导入 RAG 的核心矛盾在于——表格的强结构性和向量检索的语义模糊性之间存在天然张力。后面所有的策略选择，本质上都是在调和这个矛盾。

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

2.1 三种数据源的特性对比与选型逻辑

在动手写代码之前，先把 CSV、Excel、数据库这三种数据源的核心差异理清楚。很多人一上来就写 loader，结果写到一半发现数据格式跟自己预想的完全不一样，返工成本极高。

选型的核心判断依据是：数据是否频繁更新、结构是否稳定、是否需要行级检索。如果数据每周更新一次且结构固定，CSV 批量导入是最省事的；如果数据由业务人员手工维护、格式经常变，Excel 解析要做得足够健壮；如果需要实时查询或做聚合分析，那必须走数据库连接的路子。

2.2 分块策略：表格数据怎么切才不丢语义

这是整个流程里最关键的决策点。文本分块可以按 token 数切，但表格不行——你把一行数据切成两半，语义就碎了。

我的经验是采用行级序列化 + 语义增强的策略。具体来说分三步：

第一步，把每一行数据转成一段自然语言描述。比如一行销售数据{日期: 2024-01-15, 产品: A, 销量: 1200, 区域: 华东}，转成"2024年1月15日，产品A在华东区域的销量为1200件"。这样做的目的是让向量模型能理解这行数据的语义，而不是把它当成一堆无意义的数字。

第二步，给每段描述加上表名和字段说明作为前缀。比如"【销售表】日期：2024-01-15，产品：A……"。这个前缀在检索时能提供关键的上下文信息，尤其是当你有多个表的时候。

第三步，根据行数和字段数决定是否合并。如果一行数据字段很少（比如只有三四个字段），可以把相邻的几行合并成一段，避免碎片化。一般控制在每段 200-500 字比较合适。

注意：不要试图把整个表格塞进一个 chunk。我试过把一张 500 行的表直接转成一段文本，结果向量化之后检索效果极差，因为语义被稀释得太厉害了。

2.3 元数据设计：让检索能精确定位

结构化数据导入 RAG 有一个纯文本没有的优势——你可以给每个 chunk 附加丰富的元数据。这些元数据在检索阶段可以做过滤，大幅提升准确率。

我通常会给每个 chunk 附加以下元数据：

• source_type: csv / excel / database
• table_name: 来源表名或 sheet 名
• row_index: 原始行号（方便溯源）
• columns: 涉及的字段列表
• import_time: 导入时间戳

这些元数据在 LangChain 或 LlamaIndex 里都可以通过Document对象的metadata字段传入。检索时可以用filter参数做精确过滤，比如只检索"销售表"里"华东区域"的数据。

3. CSV 导入实战：从编码探测到批量入库

3.1 编码探测与分隔符识别

CSV 最坑的地方就是编码。国内的数据源大量使用 GBK 或 GB2312，而 Python 默认按 UTF-8 读取，直接报UnicodeDecodeError。更隐蔽的是 UTF-8-BOM，文件头多了\xef\xbb\xbf三个字节，导致第一列列名读出来带个不可见字符，后面按列名取值全部失败。

我的处理方案是用chardet做编码探测，然后统一转成 UTF-8：

import chardet import pandas as pd def detect_encoding(file_path, sample_size=100000): with open(file_path, 'rb') as f: raw = f.read(sample_size) result = chardet.detect(raw) return result['encoding'], result['confidence'] def read_csv_robust(file_path): encoding, confidence = detect_encoding(file_path) if confidence < 0.7: # 置信度低时，尝试常见编码 for enc in ['utf-8', 'gbk', 'gb2312', 'latin-1']: try: df = pd.read_csv(file_path, encoding=enc, nrows=5) encoding = enc break except (UnicodeDecodeError, pd.errors.ParserError): continue df = pd.read_csv(file_path, encoding=encoding) # 去除列名中的 BOM 和空白 df.columns = df.columns.str.replace('\ufeff', '').str.strip() return df

分隔符也不能想当然。虽然叫 CSV（逗号分隔），但实际项目中我遇到过用分号、制表符、甚至竖线分隔的。pandas 的sep=None配合engine='python'可以自动嗅探分隔符，但大文件下性能很差。我的做法是先读前几行手动判断，确定分隔符后再全量读取。

3.2 缺失值与异常值处理

CSV 里的缺失值有无数种表示方式：空字符串、NULL、N/A、-、nan、None。如果不统一处理，转成自然语言描述时会出现"销量为N/A"这种毫无意义的句子。

import numpy as np def clean_dataframe(df): # 统一缺失值标识 null_markers = ['', 'NULL', 'N/A', 'n/a', '-', 'nan', 'None', 'null', 'NA'] df = df.replace(null_markers, np.nan) # 数值列：填充或标记 for col in df.select_dtypes(include=[np.number]).columns: if df[col].isna().sum() / len(df) > 0.5: df = df.drop(columns=[col]) # 缺失超过50%直接丢弃 else: df[col] = df[col].fillna(0) # 文本列：用"未知"填充 for col in df.select_dtypes(include=['object']).columns: df[col] = df[col].fillna('未知') return df

这里有个经验：缺失率超过 50% 的列直接丢掉。我见过一张表 30 列里有 18 列全是空的，这些列不仅没有信息量，还会在序列化时产生大量噪音，拖累检索效果。

3.3 行级序列化与批量向量化

清洗完之后就是核心的序列化环节。我的做法是给每一行生成一段描述文本，然后批量送入 embedding 模型：

def row_to_text(row, table_name, columns): parts = [f"【{table_name}】"] for col in columns: val = row[col] if isinstance(val, float) and val == int(val): val = int(val) # 避免出现 1200.0 这种 parts.append(f"{col}：{val}") return "，".join(parts) def dataframe_to_documents(df, table_name): documents = [] columns = df.columns.tolist() for idx, row in df.iterrows(): text = row_to_text(row, table_name, columns) doc = Document( page_content=text, metadata={ "source_type": "csv", "table_name": table_name, "row_index": int(idx), "columns": ",".join(columns) } ) documents.append(doc) return documents

批量向量化的时候注意控制 batch size。我实测下来，OpenAI 的 embedding 接口 batch size 设在 100-500 之间比较稳，太大容易超时，太小则吞吐量上不去。如果用本地模型（比如 BGE 系列），batch size 可以设到 64-128，取决于显存大小。

实操心得：百万行级别的 CSV 不要一次性全部加载到内存。用 pandas 的chunksize参数分块读取，每块处理完就写入向量库，处理完释放内存。我处理过一个 200 万行的 CSV，16GB 内存的机器如果不分块直接 OOM。

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

4.1 多 Sheet 遍历与表头识别

Excel 文件最大的特点就是"一个文件多个 sheet"，而且每个 sheet 的结构可能完全不同。有的 sheet 是数据表，有的是说明页，有的是图表。你不能无脑把所有 sheet 都当数据表处理。

我的策略是先做一轮 sheet 扫描，判断每个 sheet 是否是有效数据表：

import openpyxl def scan_excel_sheets(file_path): wb = openpyxl.load_workbook(file_path, read_only=True, data_only=True) sheet_info = [] for sheet_name in wb.sheetnames: ws = wb[sheet_name] # 读取前10行做判断 rows = [] for i, row in enumerate(ws.iter_rows(max_row=10, values_only=True)): rows.append(row) if i >= 9: break # 判断依据：非空单元格比例、是否有表头特征 non_empty = sum(1 for row in rows for cell in row if cell is not None) total = sum(len(row) for row in rows) density = non_empty / total if total > 0 else 0 sheet_info.append({ "name": sheet_name, "density": density, "is_data_sheet": density > 0.3 and len(rows) > 2 }) wb.close() return sheet_info

data_only=True这个参数很关键。Excel 里的公式单元格如果不加这个参数，读出来的是公式本身（比如=SUM(A1:A10)），而不是计算结果。加上之后读到的才是值。但要注意，如果这个 Excel 文件从来没有被 Excel 软件打开保存过，缓存的计算值可能不存在，读出来是 None。这种情况就需要用formulas库或者手动计算。

4.2 合并单元格的向下填充

合并单元格是 Excel 解析的头号杀手。比如"部门"这一列，第一行是"销售部"，合并了下面 5 行，openpyxl 读出来只有第一行有值，后面 5 行都是 None。如果不处理，序列化出来就是"部门：未知"。

处理方案是检测合并区域并向下填充：

def fill_merged_cells(ws): merged_ranges = list(ws.merged_cells.ranges) for merge_range in merged_ranges: min_row, min_col = merge_range.min_row, merge_range.min_col max_row, max_col = merge_range.max_row, merge_range.max_col top_left_value = ws.cell(row=min_row, column=min_col).value for row in range(min_row, max_row + 1): for col in range(min_col, max_col + 1): ws.cell(row=row, column=col).value = top_left_value return ws

这个操作要在读取数据之前做。注意read_only模式下不能修改单元格，需要先以普通模式加载，填充完再读。大文件下这会比较慢，但合并单元格的处理没有更好的办法。

4.3 表头行自动定位

很多业务人员做的 Excel 前几行是标题、说明、日期之类的，真正的表头在第 3 行甚至第 5 行。如果直接用header=0读，列名全是"Unnamed: 1"这种。

我的做法是扫描前 10 行，找到"最像表头"的那一行。判断标准是：该行非空单元格数量最多、且该行以下的数据行类型一致性最高。

def find_header_row(df_raw): best_row = 0 best_score = 0 for i in range(min(10, len(df_raw))): row = df_raw.iloc[i] non_null = row.notna().sum() # 表头行通常非空值多，且都是字符串 str_count = sum(1 for v in row if isinstance(v, str)) score = non_null + str_count * 0.5 if score > best_score: best_score = score best_row = i return best_row

找到表头行之后，用pd.read_excel(file_path, sheet_name=sheet, header=best_row)重新读取即可。

踩坑记录：有一次处理一个财务报表，表头在第 4 行，但第 3 行有个合并单元格写着"2024年度"，我一开始没处理，结果这一行被当成了数据行，序列化出来变成"2024年度：未知，未知，未知……"，污染了好几个 chunk。后来加了表头定位逻辑才解决。

5. 通过 LlamaHub 连接数据库的完整流程

5.1 DatabaseReader 的配置与连接

LlamaHub 提供了DatabaseReader这个工具，可以直接把数据库查询结果转成 Document 对象。它底层用的是 SQLAlchemy，所以理论上支持所有 SQLAlchemy 兼容的数据库——MySQL、PostgreSQL、SQLite、Oracle 都行。

先安装依赖：

pip install llama-index-readers-database sqlalchemy pymysql

连接配置的核心是构造连接字符串。以 MySQL 为例：

from llama_index.readers.database import DatabaseReader connection_string = "mysql+pymysql://user:password@host:3306/dbname?charset=utf8mb4" reader = DatabaseReader(uri=connection_string) documents = reader.load_data( query="SELECT id, product_name, category, price, stock FROM products WHERE status = 'active'", metadata={"table": "products", "source": "mysql"} )

DatabaseReader默认会把查询结果的每一行转成一段文本，格式是column: value的拼接。这个默认行为对于简单的表够用，但对于字段多、需要语义增强的场景，建议自己写转换逻辑。

5.2 SQL 查询转自然语言的策略

直接用SELECT *把整张表拉出来是最粗暴的做法，效果往往不好。更好的方式是根据 RAG 的检索需求，定制查询语句，并且在查询层面就做好语义增强。

我的做法是写一个 SQL 模板，用CONCAT在数据库层面就把行数据拼成自然语言：

SELECT CONCAT( '产品', product_name, '属于', category, '类目，当前售价', price, '元，库存', stock, '件' ) AS description FROM products WHERE status = 'active';

这样查出来的每一行本身就是一段完整的自然语言描述，直接就能转成 Document，省去了 Python 层的转换。而且数据库的字符串拼接性能通常比 Python 循环处理要快。

对于需要多表关联的场景，用 JOIN 把关联字段拼进来：

SELECT CONCAT( '订单', o.order_id, '由客户', c.customer_name, '于', o.order_date, '购买，总金额', o.total_amount, '元' ) AS description FROM orders o JOIN customers c ON o.customer_id = c.id WHERE o.order_date >= '2024-01-01';

5.3 增量同步与去重机制

数据库数据是动态变化的，不可能每次全量重新导入。需要设计增量同步机制。

最简单的方案是用时间戳字段做增量：

from datetime import datetime, timedelta last_sync_time = get_last_sync_time() # 从配置或数据库读取 query = f""" SELECT CONCAT(...) AS description, id, updated_at FROM products WHERE updated_at > '{last_sync_time}' """

去重的话，用id字段作为唯一标识，在写入向量库时先删除相同id的旧记录再插入新的。ChromaDB 和 Milvus 都支持按 metadata 过滤删除。

注意：增量同步要考虑时区问题。如果数据库存的是 UTC 时间，而你的同步脚本用的是本地时间，会出现漏同步或重复同步。统一用 UTC 时间戳做比较最稳妥。

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

6.1 编码与格式类问题速查

6.2 向量化与检索效果差的自查清单

结构化数据导入后检索效果不好，通常不是 embedding 模型的问题，而是数据预处理的问题。按以下顺序排查：

1. 检查序列化文本是否可读：随便抽几个 Document 的page_content打印出来，如果读起来像天书，向量模型也理解不了。
2. 检查 chunk 长度分布：如果大量 chunk 只有十几个字，说明分块太碎；如果大量超过 1000 字，说明合并过度。
3. 检查元数据是否完整：用filter检索时能不能精确命中目标表。
4. 检查是否有重复数据：同一行数据被导入了多次，检索时全是重复结果。
5. 检查数值列的处理：像"1200.0"这种浮点表示，在语义上和"1200"有细微差异，统一转成整数。

6.3 性能优化：百万行数据的处理策略

处理百万行级别的数据，性能瓶颈通常在三个地方：文件读取、序列化转换、向量化写入。

文件读取用 pandas 的chunksize分块，每块 5 万行左右。序列化转换用df.apply比逐行iterrows快 10 倍以上。向量化写入用批量接口，并且开多线程并发。

from concurrent.futures import ThreadPoolExecutor def process_chunk(chunk_df, table_name): docs = dataframe_to_documents(chunk_df, table_name) # 批量向量化并写入 vector_store.add_documents(docs) return len(docs) with pd.read_csv(file_path, chunksize=50000, encoding='utf-8') as reader: with ThreadPoolExecutor(max_workers=4) as executor: futures = [] for chunk in reader: chunk = clean_dataframe(chunk) futures.append(executor.submit(process_chunk, chunk, 'sales')) for f in futures: print(f"处理了 {f.result()} 行")

我实测下来，200 万行数据用这个方案，从读取到全部写入向量库，大约 40 分钟。瓶颈在 embedding API 的调用上，如果换成本地 GPU 推理，可以压缩到 10 分钟以内。

6.4 几个容易忽略的细节

日期格式统一。CSV 里的日期可能是2024/1/15、2024-01-15、15/01/2024各种格式。序列化之前统一转成YYYY-MM-DD格式，否则检索"2024年1月"的时候会漏掉很多。

大数字的可读性。1200000和1,200,000在语义上是一样的，但向量模型可能处理不同。我通常会在序列化时加上千分位，或者转成"120万"这种中文习惯的表达。

布尔值的处理。数据库里的0/1、true/false、Y/N要统一转成"是/否"，否则检索"是否有效"的时候匹配不上。

空表的处理。有些 sheet 或查询结果可能是空的，要在代码里加判断，避免生成空的 Document 污染向量库。

最后分享一个排查技巧：当你觉得检索效果不对劲的时候，把用户的 query 和检索到的 top-5 Document 一起打印出来，逐条看为什么这条被召回了、那条没被召回。十有八九问题出在序列化文本的表述方式上，而不是向量模型本身。