企业知识库的冷启动难题——文档质量差、数量少怎么办

适读人群：正在做企业知识库项目、面对乱七八糟文档的开发者 | 阅读时长：约 14 分钟 | 核心价值：真实案例中解决文档质量问题的工程实践

去年接了一个项目，客户是一家制造业企业，想把十几年积累的技术文档做成 AI 知识库，让工程师能快速查设备维修手册和工艺规范。

项目启动会上，客户负责人很自信地说："我们有三千多份文档，内容很全。"

等他们把文档发过来，我沉默了大概五分钟。

三千多份文档里：

• 扫描的纸质手册，PDF 里全是图片，没有可提取的文字
• 十年前 Word 2003 格式的规范文件，编码是 GBK，格式乱到没法看
• 工程师在 QQ 群里的聊天记录截图，保存成了 .jpg
• 一批 Excel 文件，每个 sheet 里都是没有表头的数据表
• 还有大量内容重复的版本：V1.0、V1.1、V1.2、V2.0……内容差异极小但不知道哪个是最新的

这就是企业知识库项目的现实。

---

先把问题分类

遇到这种情况，很多开发者第一反应是"用最好的OCR把所有文档处理一遍"，然后发现不够，再堆别的技术。

正确的做法是先分类，不同类型的文档问题，解决方案完全不同。

我把企业文档问题分成四大类：

1. 格式问题：扫描件、图片、编码异常——文字根本提取不出来
2. 结构问题：格式混乱、表格歪斜、标题层级错乱——提取出来了但语义支离破碎
3. 质量问题：口语化记录、信息不完整——内容本身有缺陷
4. 版本问题：大量重复和近似内容——检索结果充斥噪声

每类问题要单独处理，不能混为一谈。

---

问题一：扫描件处理

扫描件是最常见也最麻烦的。不是"用OCR处理一下"这么简单。

工业级扫描件的特点：

• 拍摄角度不正，文字倾斜
• 纸张泛黄，对比度低
• 有表格、有图表、有手写批注
• 偶尔有图章遮挡文字

直接用 Python tesseract 处理这类文档，识别率通常低于 70%，大量关键文字识别错误。

我们最后的方案是用 Azure Document Intelligence（原 Form Recognizer），它对工业文档的版面理解和 OCR 精度都好得多：

from azure.ai.formrecognizer import DocumentAnalysisClient
from azure.core.credentials import AzureKeyCredential
import re
from pathlib import Path

class IndustrialDocumentProcessor:
    def __init__(self, endpoint: str, api_key: str):
        self.client = DocumentAnalysisClient(
            endpoint=endpoint,
            credential=AzureKeyCredential(api_key)
        )
    
    def process_scanned_pdf(self, pdf_path: str) -> dict:
        """
        处理扫描PDF，返回结构化内容
        """
        with open(pdf_path, "rb") as f:
            poller = self.client.begin_analyze_document(
                "prebuilt-layout",  # 版面分析模型
                document=f
            )
        
        result = poller.result()
        
        # 提取结构化内容
        output = {
            "pages": [],
            "tables": [],
            "text_blocks": []
        }
        
        # 按页处理
        for page_num, page in enumerate(result.pages):
            page_content = {
                "page_number": page_num + 1,
                "lines": [],
                "confidence": []
            }
            
            for line in page.lines:
                page_content["lines"].append(line.content)
                # 计算行级别的置信度（所有词的平均）
                if line.words:
                    avg_conf = sum(w.confidence for w in line.words) / len(line.words)
                    page_content["confidence"].append(avg_conf)
            
            output["pages"].append(page_content)
        
        # 提取表格
        for table_idx, table in enumerate(result.tables):
            table_data = {
                "table_index": table_idx,
                "row_count": table.row_count,
                "column_count": table.column_count,
                "cells": []
            }
            
            for cell in table.cells:
                table_data["cells"].append({
                    "row": cell.row_index,
                    "col": cell.column_index,
                    "content": cell.content,
                    "is_header": cell.kind == "columnHeader"
                })
            
            output["tables"].append(table_data)
        
        return output
    
    def convert_to_markdown(self, processed: dict) -> str:
        """把处理结果转成 Markdown，用于后续向量化"""
        md_parts = []
        
        for page in processed["pages"]:
            if page["lines"]:
                # 低置信度的行标记出来（后续人工复核）
                lines_to_add = []
                for i, line in enumerate(page["lines"]):
                    conf = page["confidence"][i] if i < len(page["confidence"]) else 1.0
                    if conf < 0.7:
                        # 低置信度内容用特殊标记，方便后续审核
                        lines_to_add.append(f"[LOW_CONF:{conf:.2f}] {line}")
                    else:
                        lines_to_add.append(line)
                
                md_parts.append("\n".join(lines_to_add))
        
        # 表格转 Markdown 表格格式
        for table in processed["tables"]:
            md_table = self._table_to_markdown(table)
            md_parts.append(md_table)
        
        return "\n\n".join(md_parts)
    
    def _table_to_markdown(self, table: dict) -> str:
        """把结构化表格转成 Markdown 表格"""
        rows = {}
        headers = {}
        
        for cell in table["cells"]:
            row = cell["row"]
            col = cell["col"]
            
            if cell["is_header"]:
                headers[col] = cell["content"]
            else:
                if row not in rows:
                    rows[row] = {}
                rows[row][col] = cell["content"]
        
        if not rows:
            return ""
        
        max_col = max(cell["col"] for cell in table["cells"]) + 1
        
        lines = []
        # 表头
        header_row = [headers.get(i, f"列{i+1}") for i in range(max_col)]
        lines.append("| " + " | ".join(header_row) + " |")
        lines.append("| " + " | ".join(["---"] * max_col) + " |")
        
        # 数据行
        for row_idx in sorted(rows.keys()):
            row_data = [rows[row_idx].get(col, "") for col in range(max_col)]
            lines.append("| " + " | ".join(row_data) + " |")
        
        return "\n".join(lines)

---

问题二：乱格式 Word 文档

老 Word 文档的问题是格式混乱：用空格做缩进、标题不用样式而是手动加粗、章节编号是手打的。

用 python-docx 直接读出来就是一坨没有结构的文字。

我的处理方案：用启发式规则重建结构：

from docx import Document
import re
from typing import List, Tuple

class WordDocumentNormalizer:
    """对乱格式Word文档做结构重建"""
    
    # 常见标题模式（中文技术文档）
    HEADING_PATTERNS = [
        # "一、xxx" "二、xxx" 这种
        (r'^[一二三四五六七八九十]+[、.．]\s*(.+)', 1),
        # "1. xxx" "2. xxx"
        (r'^(\d+)[.．]\s*(.+)', 1),
        # "1.1 xxx" "1.1.1 xxx"
        (r'^(\d+\.\d+(?:\.\d+)?)\s+(.+)', 2),
        # "第X章" "第X节"
        (r'^第[一二三四五六七八九十\d]+[章节条款]\s*(.+)', 1),
    ]
    
    def normalize(self, docx_path: str) -> str:
        """返回重建结构后的 Markdown"""
        doc = Document(docx_path)
        
        paragraphs = []
        for para in doc.paragraphs:
            text = para.text.strip()
            if not text:
                continue
            
            # 判断这个段落是否是标题
            heading_level, cleaned_text = self._detect_heading(para)
            
            if heading_level > 0:
                paragraphs.append(("#" * heading_level + " " + cleaned_text, "heading"))
            else:
                paragraphs.append((text, "body"))
        
        # 同时处理表格
        for table in doc.tables:
            table_md = self._table_to_markdown(table)
            paragraphs.append((table_md, "table"))
        
        # 组合成 Markdown
        return "\n\n".join(text for text, _ in paragraphs)
    
    def _detect_heading(self, para) -> Tuple[int, str]:
        """
        判断段落是否是标题，返回 (层级, 文本)
        层级0表示不是标题
        """
        text = para.text.strip()
        
        # 检查Word自带的样式
        style_name = para.style.name.lower()
        if "heading" in style_name:
            level_match = re.search(r'\d+', style_name)
            if level_match:
                return int(level_match.group()), text
        
        # 检查字体特征：大字号 + 加粗 通常是标题
        if para.runs:
            is_bold = any(run.bold for run in para.runs if run.text.strip())
            font_size = max(
                (run.font.size.pt if run.font.size else 0) 
                for run in para.runs if run.text.strip()
            ) if para.runs else 0
            
            if is_bold and font_size >= 14 and len(text) < 50:
                return 2, text
        
        # 检查文本模式
        for pattern, level in self.HEADING_PATTERNS:
            match = re.match(pattern, text)
            if match:
                # 取最后一个捕获组作为标题文字
                clean = match.groups()[-1] if match.groups() else text
                # 如果文字太长，不认为是标题
                if len(clean) < 80:
                    return level, clean
        
        return 0, text
    
    def _table_to_markdown(self, table) -> str:
        rows = []
        for row in table.rows:
            cells = [cell.text.strip().replace("\n", " ") for cell in row.cells]
            rows.append(cells)
        
        if not rows:
            return ""
        
        max_cols = max(len(row) for row in rows)
        
        lines = []
        lines.append("| " + " | ".join(rows[0]) + " |")
        lines.append("| " + " | ".join(["---"] * max_cols) + " |")
        for row in rows[1:]:
            # 补齐列数
            while len(row) < max_cols:
                row.append("")
            lines.append("| " + " | ".join(row) + " |")
        
        return "\n".join(lines)

---

问题三：口语化聊天记录

这是最让人头疼的问题。工程师在群里说的那些东西，往往是宝贵的隐性知识，但格式太乱、信息太碎。

比如：

小王：那个问题解决了吗
老陈：解决了，把那个超时参数调大就行，redis的
小王：多大
老陈：300秒，我试了200不够

这种内容，直接向量化进知识库，检索效果很差。

我的处理方案是：用 LLM 对这类内容做结构化提炼，再存入知识库：

from openai import OpenAI

client = OpenAI()

CHAT_NORMALIZATION_PROMPT = """
以下是一段技术讨论的聊天记录。请将其中有价值的技术信息提炼成结构化的知识条目。

要求：
1. 只保留有技术价值的信息，过滤掉闲聊
2. 补全省略的上下文（如"那个参数"要说清楚是什么参数）
3. 输出格式：
   - 问题/场景：[描述遇到的问题]
   - 解决方案：[具体的解决步骤]
   - 关键参数/配置：[如有]
   - 注意事项：[如有]

如果聊天记录没有有价值的技术信息，返回：{"has_value": false}

聊天记录：
{chat_content}

以JSON格式返回。
"""

def normalize_chat_log(chat_content: str) -> dict | None:
    """把聊天记录转化为结构化知识条目"""
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",  # 用便宜的模型，这是批量处理任务
        messages=[
            {"role": "user", "content": CHAT_NORMALIZATION_PROMPT.format(
                chat_content=chat_content
            )}
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    
    result = json.loads(response.choices[0].message.content)
    
    if not result.get("has_value", True):
        return None
    
    # 转成可以存入知识库的格式
    parts = []
    if "问题/场景" in result:
        parts.append(f"## 问题场景\n{result['问题/场景']}")
    if "解决方案" in result:
        parts.append(f"## 解决方案\n{result['解决方案']}")
    if "关键参数/配置" in result:
        parts.append(f"## 关键配置\n{result['关键参数/配置']}")
    if "注意事项" in result:
        parts.append(f"## 注意事项\n{result['注意事项']}")
    
    return {
        "content": "\n\n".join(parts),
        "source_type": "chat_log",
        "raw": chat_content
    }

---

问题四：版本去重

大量重复文档会严重污染检索结果。

我的方案是两步走：

1. 用文件名 + 修改时间初步筛选，只保留最新版本
2. 对内容相似度超过 95% 的文档，只保留一份（或合并差异）

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

def deduplicate_documents(docs: list[dict], similarity_threshold: float = 0.95) -> list[dict]:
    """
    基于内容相似度去重
    docs: [{"id": ..., "content": ..., "modified_time": ...}, ...]
    """
    if len(docs) <= 1:
        return docs
    
    contents = [doc["content"] for doc in docs]
    
    vectorizer = TfidfVectorizer(max_features=5000)
    tf_matrix = vectorizer.fit_transform(contents)
    sim_matrix = cosine_similarity(tf_matrix)
    
    # 找出重复组
    n = len(docs)
    to_remove = set()
    
    for i in range(n):
        if i in to_remove:
            continue
        for j in range(i + 1, n):
            if j in to_remove:
                continue
            if sim_matrix[i, j] >= similarity_threshold:
                # 保留修改时间更新的那个
                if docs[i].get("modified_time", 0) >= docs[j].get("modified_time", 0):
                    to_remove.add(j)
                else:
                    to_remove.add(i)
                    break
    
    kept = [doc for idx, doc in enumerate(docs) if idx not in to_remove]
    print(f"去重前: {n} 篇，去重后: {len(kept)} 篇，移除: {len(to_remove)} 篇")
    
    return kept

---

最重要的一个认知

做了这个项目之后，我对"企业AI知识库"这件事的认知发生了根本变化。

文档治理是工程问题，不是AI问题。

很多甲方觉得"我们买了AI系统，把文档丢进去就能用"。实际情况是，AI能处理的是格式转换和质量提升，但它没法凭空变出那些从来没被记录下来的知识，也没法判断哪个版本的文档是准确的。

在这个项目里，我花在"文档治理"上的时间，大概是花在"AI系统"上的两倍。

这不是AI的失败，这是知识管理本来就该做的事情。只是以前没有压力去做，现在做AI知识库了，欠的债都要还。