企业里做大模型落地十有八九卡在同一个地方模型选好了算力也到位了结果一问业务问题就胡言乱语或者干脆答非所问。折腾几轮之后大家才反应过来问题根本不在模型本身而在喂进去的数据。原始业务数据里塞满了重复记录、格式错乱、字段缺失、口径打架直接丢给大模型它只能一本正经地编。所谓 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 数据源盘点表怎么建动手前先建一张数据源盘点表把每个源的关键信息列清楚这张表后面会贯穿整个工程链路数据源类型预估量级更新频率主要脏点优先级产品手册PDF约 800 份季度OCR 噪声、图表混排高客服工单数据库约 50 万条实时口语化、错别字高内部制度Word约 300 份不定期多版本冗余中技术文档Markdown约 2000 篇周代码块与正文混杂高这张表的价值在于它让你在写清洗脚本时心里有数哪些源值得投入重资源做精细处理哪些源量大但价值低可以先粗处理。别平均用力企业项目里时间永远不够。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:imax_len] for i in range(0, len(text), max_len)]3.2 块大小和重叠怎么定块大小没有万能值但有个经验区间中文文档一般 300 到 800 字比较合适。太小了语义不完整太大了检索精度下降。重叠部分一般设 10% 到 20%防止关键信息正好卡在切分边界上被切断。文档类型建议块大小重叠比例理由技术文档500-800 字15%段落长逻辑连贯客服工单200-400 字10%单条短独立性强制度文件400-600 字20%条款间有引用关系产品手册300-500 字15%图文混排需保结构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越大建索引越慢但质量越好。一般m16、ef_construction64是个不错的起点。4.3 索引参数调优的实测经验我做过一组对比测试在 50 万条数据上mef_construction建索引耗时召回率内存占用832快一般低1664中好中32128慢很好高实测下来m16是性价比最高的选择。除非你的召回率要求极高否则没必要上m32内存吃不消。查询时的ef_search参数也要调默认值偏小我一般设成 100 左右召回和延迟比较平衡。提示pgvector 在 Windows 上编译安装比较麻烦很多人会去找预编译包。我的建议是生产环境直接用 LinuxWindows 只用来做开发验证。如果非要在 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(), keylambda 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[资料{i1}] 来源{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 模块数据接入、清洗、切分、向量化、检索各管各的用配置文件串起来。新项目来了换配置就行代码基本不用动。最后分享一个实用的小技巧清洗和切分阶段一定要把中间结果落盘保存。别每次都从头跑那样调试一次要等半天。把清洗后的文本、切分后的块都存成文件调试检索策略时直接读文件效率能提升好几倍。这个习惯我坚持了好几年每次都能省下大量等待时间。