企业大模型落地：从脏数据到AI-ready数据底座的五步工程化实践

企业里做大模型落地，十有八九卡在同一个地方：模型选好了，算力也到位了，结果一问业务问题就胡言乱语，或者干脆答非所问。折腾几轮之后大家才反应过来，问题根本不在模型本身，而在喂进去的数据。原始业务数据里塞满了重复记录、格式错乱、字段缺失、口径打架，直接丢给大模型，它只能一本正经地编。所谓 AI-ready 数据底座，说白了就是把企业里那些"能用但不好用"的脏数据，经过一套工程化流程，变成大模型和 RAG 系统真正吃得下、消化得了的干净燃料。这套流程不是调几个 API 就完事，它涉及数据接入、清洗、切分、向量化、检索增强五个环节，每个环节都有坑。下面我按自己实际做过几个企业知识库项目的经验，把这五步拆开讲透，顺带把 pgvector、RAG 框架选型、Python 处理链路这些实操细节一并交代清楚，适合正在做企业私有化大模型落地、RAG 知识库搭建的工程师和数据同学参考。

1. 先搞清楚"脏数据"到底脏在哪，别急着上工具

很多人一上来就装 pgvector、拉 RAG 框架，结果数据没理清楚，后面全是返工。我建议第一步先做数据体检，把问题摸清楚再动手。

1.1 企业数据的四类典型病症

我接触过的企业数据，脏法基本逃不出这四类：

• 结构脏：同一个客户名，在 CRM 里叫"某某科技有限公司"，在合同系统里叫"某某科技"，在工单系统里又变成"某某科技（上海）"。字段类型也不统一，有的日期是2024-01-01，有的是2024/1/1，还有的是时间戳。
• 内容脏：PDF 扫描件里的文字带 OCR 噪声，Excel 单元格里混着换行符和全角空格，HTML 抓下来的正文裹着一堆导航栏和广告。
• 语义脏：同一个指标，销售部门叫"成交额"，财务部门叫"确认收入"，口径还不一样。这种脏最要命，因为它不是格式问题，是业务定义冲突。
• 冗余脏：同一份制度文档存了五个版本，内容 90% 重复，只是修订日期不同。全喂进去，检索时召回一堆近似结果，反而稀释了有效信息。

提示：数据体检阶段不要急着写清洗代码，先抽样。每个数据源抽 50 到 100 条，人工过一遍，把问题归类记下来。这份清单后面会直接决定你的清洗规则和切分策略。

1.2 为什么"先清洗后切分"这个顺序不能反

有个反直觉的点：很多人觉得切分是最后一步，其实切分策略会反过来影响清洗规则。比如你打算按段落切，那清洗时就要保留段落结构，不能把所有换行都干掉；如果你打算按固定长度切，那清洗时可以更激进地压平文本。

我的做法是先定切分粒度，再倒推清洗要保留哪些结构信息。企业文档一般按语义段落切效果最好，那就意味着清洗阶段必须保住标题层级、列表结构、表格边界。这个顺序想清楚了，后面能省大量返工。

1.3 数据源盘点表怎么建

动手前先建一张数据源盘点表，把每个源的关键信息列清楚，这张表后面会贯穿整个工程链路：

这张表的价值在于，它让你在写清洗脚本时心里有数：哪些源值得投入重资源做精细处理，哪些源量大但价值低可以先粗处理。别平均用力，企业项目里时间永远不够。

2. 清洗不是"删删改改"，而是一套可复现的流水线

清洗环节最容易犯的错，是写成一次性脚本，跑完就扔。下次数据更新，又得重来。正确的做法是把它做成可配置、可复现的流水线。

2.1 用 Python 搭一条分层清洗链路

我习惯把清洗拆成三层，每层职责单一，方便单独调试和替换：

import re import unicodedata def normalize_text(text: str) -> str: # 第一层：字符级归一化 text = unicodedata.normalize("NFKC", text) # 全角转半角 text = re.sub(r"\s+", " ", text) # 压缩连续空白 text = re.sub(r"[\u200b-\u200f]", "", text) # 去掉零宽字符 return text.strip() def clean_structure(text: str) -> str: # 第二层：结构级清洗，保留段落和列表 lines = text.split("\n") cleaned = [] for line in lines: line = line.strip() if not line: continue # 去掉页眉页脚这类重复噪声 if re.match(r"^第\s*\d+\s*页", line): continue cleaned.append(line) return "\n".join(cleaned) def dedup_semantic(chunks: list, threshold: float = 0.95) -> list: # 第三层：语义级去重，后面向量化阶段会用到 seen = [] result = [] for c in chunks: if not any(similarity(c, s) > threshold for s in seen): result.append(c) seen.append(c) return result

这三层分别解决字符、结构、语义三个层面的问题。字符层用unicodedata做 NFKC 归一化，能把全角标点、全角数字统一成半角，这一步对中文文档特别关键，因为很多 PDF 提取出来的文字全角半角混着来，不统一后面检索会漏召回。

2.2 中文文档清洗的几个特殊坑

中文企业文档有几个英文资料里很少提的坑，我踩过：

• 全角空格和全角标点：\u3000是全角空格，普通\s在部分正则引擎里匹配不到，得单独处理。
• PDF 换行错位：PDF 提取时经常在句子中间插入换行，导致一句话被拆成两行。简单粗暴地删换行会把段落也删掉，正确做法是判断行尾是否有标点，没有标点才合并。
• 表格提取：PDF 里的表格提取出来经常是错位的，我一般用pdfplumber单独抽表格，转成 Markdown 表格再拼回正文，比纯文本提取靠谱得多。

import pdfplumber def extract_pdf(path: str) -> str: parts = [] with pdfplumber.open(path) as pdf: for page in pdf.pages: text = page.extract_text() or "" tables = page.extract_tables() for table in tables: md = "\n".join( "| " + " | ".join(cell or "" for cell in row) + " |" for row in table ) text += "\n\n" + md parts.append(text) return "\n\n".join(parts)

2.3 清洗规则要配置化，别硬编码

我见过太多项目把清洗规则写死在代码里，结果业务方说"这个字段不能删"，就得改代码重新部署。正确做法是把规则抽成配置文件：

rules: - name: remove_page_header pattern: "^第\\s*\\d+\\s*页" action: drop_line - name: normalize_fullwidth action: nfkc - name: merge_broken_lines condition: "line_end_no_punctuation" action: merge_next

这样业务方调整规则时，改 YAML 就行，不用碰代码。这个设计在企业环境里特别重要，因为数据治理规则经常变，配置化能让你少加很多班。

注意：清洗日志一定要留。每条规则命中了多少条数据、删了什么，都要记录。出了问题能追溯，这是企业项目的基本要求，也是你后面跟业务方对齐口径的依据。

3. 切分策略决定检索质量，这一步最容易被低估

清洗完的数据是一大坨文本，直接向量化效果很差，因为一个向量塞不下太多信息。切分就是把它拆成合适大小的块，每块语义相对完整。

3.1 固定长度切分为什么不够用

最简单的切法是按固定字符数切，比如每 500 字一块。但这样切出来的块经常在句子中间断开，语义不完整。检索时召回一个"半句话"，大模型拿到也没法用。

更好的做法是按语义边界切：优先在段落、标题、列表项这些自然边界处切，实在没有边界再按长度兜底。我一般用递归切分，先按标题切，再按段落切，最后按句子切，逐级降级。

def recursive_split(text: str, max_len: int = 500) -> list: separators = ["\n## ", "\n### ", "\n\n", "\n", "。", "！", "？"] for sep in separators: if sep in text: parts = text.split(sep) chunks, buf = [], "" for p in parts: if len(buf) + len(p) <= max_len: buf += sep + p if buf else p else: if buf: chunks.append(buf) buf = p if buf: chunks.append(buf) return chunks return [text[i:i+max_len] for i in range(0, len(text), max_len)]

3.2 块大小和重叠怎么定

块大小没有万能值，但有个经验区间：中文文档一般 300 到 800 字比较合适。太小了语义不完整，太大了检索精度下降。重叠部分一般设 10% 到 20%，防止关键信息正好卡在切分边界上被切断。

3.3 元数据一定要跟着块走

切分时千万别只留文本，元数据必须一起带上。元数据包括：来源文件、章节标题、页码、更新时间、权限标签。这些信息在检索阶段能大幅提升效果，比如可以按权限过滤、按时间排序、按来源加权。

chunk = { "text": "...", "metadata": { "source": "产品手册_v3.pdf", "section": "第三章 安装说明", "page": 12, "updated_at": "2024-06-01", "access_level": "internal" } }

我做过一个项目，就是因为切分时没带权限标签，结果检索时把高管薪酬文档召回给了普通员工，差点出大事。元数据不是可选项，是必选项。

4. 向量化与 pgvector 落地：选型、建库、调优

数据切好之后，要转成向量存进向量数据库。这一步涉及模型选型、数据库选型、索引调优，每个都有讲究。

4.1 嵌入模型怎么选

嵌入模型决定了检索质量的上限。企业场景下我一般考虑三个维度：中文效果、部署方式、成本。

• 中文效果：通用多语言模型在中文上往往不如专门优化的中文模型。选型时一定要用自己业务数据做小规模评测，别只看榜单。
• 部署方式：企业私有化场景下，能本地部署的模型优先，数据不出内网是硬要求。
• 成本：如果数据量大，调用外部 API 的成本会很高，本地部署虽然前期投入大，但长期更划算。

评测方法很简单：准备 50 到 100 个业务问题，每个问题标注正确答案所在的文档块，然后看模型能不能把这些块排进前 5。这个指标比任何榜单都真实。

4.2 pgvector 建库实操

pgvector 是我在企业项目里用得最多的向量存储方案，因为它跟 PostgreSQL 无缝集成，事务、权限、备份这些企业级能力直接复用，不用额外维护一套系统。

-- 启用扩展 CREATE EXTENSION IF NOT EXISTS vector; -- 建表 CREATE TABLE knowledge_chunks ( id BIGSERIAL PRIMARY KEY, content TEXT NOT NULL, embedding vector(1024), source VARCHAR(255), section VARCHAR(255), access_level VARCHAR(50), updated_at TIMESTAMP DEFAULT NOW() ); -- 建 HNSW 索引 CREATE INDEX ON knowledge_chunks USING hnsw (embedding vector_cosine_ops) WITH (m = 16, ef_construction = 64);

这里vector(1024)的维度要跟你选的嵌入模型输出维度一致，选错了插不进去。HNSW 索引的m和ef_construction是两个关键参数：m越大召回率越高但内存占用越大，ef_construction越大建索引越慢但质量越好。一般m=16、ef_construction=64是个不错的起点。

4.3 索引参数调优的实测经验

我做过一组对比测试，在 50 万条数据上：

实测下来，m=16是性价比最高的选择。除非你的召回率要求极高，否则没必要上m=32，内存吃不消。查询时的ef_search参数也要调，默认值偏小，我一般设成 100 左右，召回和延迟比较平衡。

提示：pgvector 在 Windows 上编译安装比较麻烦，很多人会去找预编译包。我的建议是生产环境直接用 Linux，Windows 只用来做开发验证。如果非要在 Windows 上跑，用 Docker 起一个 PostgreSQL 加 pgvector 的镜像最省事。

5. RAG 检索增强：把数据底座真正用起来

数据存好了，最后一步是让大模型能用上。RAG 的核心思路是：用户提问时，先从向量库检索相关块，再把块和问题一起喂给大模型，让它基于这些块回答。

5.1 基础 RAG 链路长什么样

一条最简 RAG 链路包含四步：问题向量化、向量检索、上下文拼接、大模型生成。

def rag_query(question: str, top_k: int = 5) -> str: q_vec = embed(question) chunks = vector_search(q_vec, top_k) context = "\n\n".join(c["content"] for c in chunks) prompt = f"""基于以下资料回答问题，资料中没有的信息不要编造。 资料： {context} 问题：{question} """ return llm_generate(prompt)

这个链路能跑通，但效果往往一般。问题出在检索环节：纯向量检索对关键词不敏感，用户问"某某型号的参数"，向量检索可能召回一堆语义相近但型号不对的块。

5.2 混合检索：向量加关键词才是企业场景正解

企业数据里大量存在专有名词、型号、编号，这些靠向量检索很难精准命中。我的做法是混合检索：向量检索和关键词检索各召回一批，再用 RRF 算法融合排序。

def hybrid_search(question: str, top_k: int = 5): vec_results = vector_search(embed(question), top_k * 2) kw_results = keyword_search(question, top_k * 2) # RRF 融合 scores = {} for rank, r in enumerate(vec_results): scores[r["id"]] = scores.get(r["id"], 0) + 1 / (60 + rank) for rank, r in enumerate(kw_results): scores[r["id"]] = scores.get(r["id"], 0) + 1 / (60 + rank) ranked = sorted(scores.items(), key=lambda x: -x[1]) return [get_chunk(i) for i, _ in ranked[:top_k]]

RRF 里的 60 是个经验常数，来自原始论文，实测效果稳定。混合检索在企业场景下比纯向量检索的召回率能提升 20% 以上，尤其是涉及型号、编号、专有名词的查询。

5.3 重排序：最后一公里的精度提升

检索召回一批块之后，还可以加一个重排序模型，对候选块做精细打分。重排序模型比嵌入模型慢，但精度高，一般只对前 20 到 50 个候选做重排，再取前 5 个给大模型。

这一步的收益很明显：我做过对比，加了重排序之后，答案准确率能再提升 10% 到 15%。代价是延迟增加几百毫秒，看业务能不能接受。对准确性要求高的场景，这个投入值得。

5.4 上下文拼接的几个细节

拼接上下文时，有几个细节直接影响效果：

• 块之间加分隔符：用---或明确的分隔标记，让大模型知道这是不同的块。
• 带上来源信息：每个块前面标注来源和章节，大模型回答时能引用出处，用户也更信任。
• 控制总长度：别超过大模型的上下文窗口，一般留 20% 余量给问题和回答。
• 按相关性排序：最相关的块放最前面，大模型对开头的内容注意力更高。

def build_context(chunks: list) -> str: parts = [] for i, c in enumerate(chunks): parts.append( f"[资料{i+1}] 来源：{c['metadata']['source']} " f"章节：{c['metadata']['section']}\n{c['content']}" ) return "\n\n---\n\n".join(parts)

5.5 评估与迭代：没有评估就没有优化

RAG 系统上线不是终点，得持续评估。我一般建一个小型评测集，包含 100 到 200 个真实业务问题，每个问题标注标准答案和应召回的文档块。每次调整检索策略或换模型，都跑一遍评测集，看指标变化。

核心指标有三个：召回率（相关块有没有被召回）、准确率（召回的块是不是真的相关）、答案正确率（最终回答对不对）。这三个指标要分开看，因为问题可能出在任何一个环节。召回率低说明检索策略有问题，召回率高但答案错说明生成环节有问题。

注意：评测集一定要用真实业务问题，别自己编。我见过有人用自己编的问题做评测，结果上线后真实用户的问题一个都答不好。真实问题的表达方式、用词习惯，跟你想象的完全不一样。

6. 几个我踩过的坑和对应的解法

前面讲了完整链路，这里单独拎几个坑出来，都是真金白银换来的经验。

6.1 数据更新了，向量库没更新

这是最常见的坑。源文档更新了，但向量库还是旧的，用户问新内容答不上来。解法是建增量更新机制：给每个块记录源文件的哈希值，定期扫描源文件，哈希变了就重新切分、重新向量化、更新对应块。

import hashlib def file_hash(path: str) -> str: with open(path, "rb") as f: return hashlib.md5(f.read()).hexdigest()

这个机制不复杂，但一定要有。否则数据底座会慢慢腐烂，几个月后就没法用了。

6.2 权限控制没做，检索越权

前面提过，元数据里必须带权限标签，检索时按用户权限过滤。pgvector 支持在查询时加 WHERE 条件，把权限过滤和向量检索合在一起：

SELECT content, source FROM knowledge_chunks WHERE access_level = ANY($1) ORDER BY embedding <=> $2 LIMIT 5;

这样既保证了检索效率，又保证了权限安全。千万别在应用层做过滤，那样会把无权访问的块也召回出来，只是最后没展示，存在泄露风险。

6.3 大模型上下文塞太满，反而答不准

很多人觉得上下文给得越多越好，其实不然。塞太多不相关的块，会稀释有效信息，大模型反而抓不住重点。我的经验是 top_k 控制在 3 到 5 个块，每个块 500 字左右，总共 2000 到 3000 字，效果最好。超过这个量，准确率反而下降。

6.4 中文分词的坑

关键词检索环节要用到中文分词。Python 里常用 jieba，但 jieba 对专业术语的分词效果一般。我的做法是加自定义词典，把企业专有名词、产品型号都加进去：

import jieba jieba.load_userdict("custom_dict.txt")

这个词典要持续维护，每次发现分词错误就加进去。看着麻烦，但对检索效果的提升很实在。

7. 从脏数据到 AI-ready，本质是一次数据治理

做完几个项目之后我有个体会：所谓 AI-ready 数据底座，技术含量其实没那么高，难的是数据治理的耐心。清洗规则要一条条调，切分参数要一次次试，评测集要一个个标，这些活没有捷径。

但好消息是，这套流程一旦跑通，就是可复用的。下一个数据源接进来，改改配置、调调参数就能用。我现在的做法是把整条链路封装成几个独立的 Python 模块，数据接入、清洗、切分、向量化、检索各管各的，用配置文件串起来。新项目来了，换配置就行，代码基本不用动。

最后分享一个实用的小技巧：清洗和切分阶段，一定要把中间结果落盘保存。别每次都从头跑，那样调试一次要等半天。把清洗后的文本、切分后的块都存成文件，调试检索策略时直接读文件，效率能提升好几倍。这个习惯我坚持了好几年，每次都能省下大量等待时间。