☰
企业大模型落地:从脏数据到AI-ready数据底座的五步工程化实践
2026/10/26 11:06:49 网站建设 项目流程

企业里做大模型落地,十有八九卡在同一个地方:模型选好了,算力也到位了,结果一问业务问题就胡言乱语,或者干脆答非所问。折腾几轮之后大家才反应过来,问题根本不在模型本身,而在喂进去的数据。原始业务数据里塞满了重复记录、格式错乱、字段缺失、口径打架,直接丢给大模型,它只能一本正经地编。所谓 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:i+max_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越大建索引越慢但质量越好。一般m=16、ef_construction=64是个不错的起点。

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

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

mef_construction建索引耗时召回率内存占用
832快一般低
1664中好中
32128慢很好高

实测下来,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 模块,数据接入、清洗、切分、向量化、检索各管各的,用配置文件串起来。新项目来了,换配置就行,代码基本不用动。

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

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询