企业内部AI知识库从0到1：制造业技术文档智能问答系统落地复盘

踩坑记录： 扫描时发现有些文件夹有权限问题，用

3.2 第二步：文档内容解析

文档解析主流程

class DocumentParser: def __init__(self): # 初始化PaddleOCR（中文模型） # 安装命令：pip install paddlepaddle paddleocr # 下载模型：paddleocr --show_log False self.ocr = PaddleOCR(lang='ch', use_angle_cls=True, use_gpu=True) def parse_pdf(self, file_path): """ PDF解析核心逻辑： 1. 如果PDF是文本型（文字可选中），直接提取文本 2. 如果PDF是扫描型（图片堆叠），用OCR识别 3. 处理表格（pdfplumber的table模式） """ all_text = [] with pdfplumber.open(file_path) as pdf: for page_num, page in enumerate(pdf.pages): # 优先尝试文本提取（又快又准） text = page.extract_text() if text and len(text.strip()) > 50: # 文本型PDF，直接用 all_text.append(f"[页{page_num+1}]\n{text}") else: # 扫描型PDF，转图片后OCR page_text = self._ocr_page(page, page_num) all_text.append(f"[页{page_num+1}]\n{page_text}") # 尝试提取表格（表格内容单独处理） tables = page.extract_tables() for table in tables: # 表格转CSV格式文本 table_text = self._table_to_text(table) all_text.append(f"[页{page_num+1}表格]\n{table_text}") return "\n\n".join(all_text) def _ocr_page(self, page, page_num): """将PDF页面转换为图片并OCR识别""" # PDF页面转图片 img_bytes = page.to_image().original img = Image.open(io.BytesIO(img_bytes)) # OCR识别 result = self.ocr.ocr(img, cls=True) text_lines = [] for line in result[0]: text_lines.append(line[1][0]) # line[1] = (文本, 置信度) return "\n".join(text_lines) def _table_to_text(self, table): """将表格二维数组转换为带分隔符的文本""" rows = [] for row in table: # 用 | 分隔各列，便于后续Embedding时保留表格结构信息 rows.append(" | ".join([str(cell) if cell else "" for cell in row])) return "\n".join(rows) def parse_docx(self, file_path): """解析Word文档（DOCX格式）""" doc = Document(file_path) paragraphs = [] for para in doc.paragraphs: # 过滤空白段落和样式标记 if para.text.strip(): # 保留标题样式信息（Heading1/2等） style = para.style.name text = para.text if "Heading" in style: paragraphs.append(f"## {text}") # 标记为标题 else: paragraphs.append(text) return "\n".join(paragraphs)

使用示例

OCR效果问题： 项目中发现，扫描质量差的PDF（有水印、字迹模糊）OCR准确率只有65%。解决方案： 1. 扫描前预处理：去噪、二值化（用OpenCV） 2. 建立"高风险文档"清单，后续人工审核

3.3 第三步：文本分块（Chunking）

分块策略直接影响检索质量，这是最体现经验的一步：

def chunk_documents(documents, chunk_size=500, chunk_overlap=50): """ 分块策略详解： chunk_size=500 tokens： - 太小：上下文不足，模型无法综合多个知识点回答 - 太大：向量检索时匹配度下降，且超过模型上下文窗口 - 实战经验：500对于技术问答类文档是最均衡的选择 chunk_overlap=50： - 保留块之间的重叠，避免关键信息被切断 - 重叠越大召回越高，但检索精度下降 - 50（约10%重叠）在效果和效率间平衡较好 """ text_splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, # 按token数分块 chunk_overlap=chunk_overlap, # 相邻块重叠token数 length_function=len, # 用字符数近似token数 separators=["\n\n", "\n", "。", "，", " "] # 优先按段落分 ) chunks = [] for doc in documents: # 按段落分割（\n\n是段落边界） split_texts = text_splitter.split_text(doc["content"]) for i, chunk_text in enumerate(split_texts): chunks.append({ "content": chunk_text, "metadata": { "source": doc["filename"], # 来源文件名 "filepath": doc["filepath"], # 完整路径 "chunk_index": i, # 块编号 "total_chunks": len(split_texts), # 总块数 "doc_type": doc["type"], # 文档类型（设备手册/工艺规程等） "last_modified": doc["modified"] } }) return chunks

特殊处理：表格不能按段落分割

对于表格类内容，强制整表为一个chunk（表格语义不可拆分）

3.4 第四步：向量化与入库

向量入库到Milvus

Milvus连接配置

def create_milvus_collection(): """创建Milvus Collection（如果不存在）""" # 定义Schema：每个向量对应一段文本 # 向量维度 = 1536（text-embedding-3-small的维度） fields = [ Field(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), Field(name="content", dtype=DataType.VARCHAR, max_length=65535), Field(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1536), Field(name="source", dtype=DataType.VARCHAR, max_length=512), Field(name="filepath", dtype=DataType.VARCHAR, max_length=1024), Field(name="doc_type", dtype=DataType.VARCHAR, max_length=128), Field(name="last_modified", dtype=DataType.VARCHAR, max_length=64), ] schema = CollectionSchema(fields=fields, description="技术文档知识库") if utility.has_collection(COLLECTION_NAME): utility.drop_collection(COLLECTION_NAME) collection = Collection(name=COLLECTION_NAME, schema=schema) # 创建索引（HNSW算法，高召回） # IVF_FLAT适合召回优先，HNSW适合精度+速度均衡 index_params = { "index_type": "HNSW", "metric_type": "IP", # Inner Product = 余弦相似度（归一化后等价） "params": {"M": 16, "efConstruction": 200} } collection.create_index(field_name="embedding", index_params=index_params) collection.load() return collection

def ingest_chunks(chunks, batch_size=100): """批量向量化并入库""" embedding_model = OpenAIEmbeddings(model="text-embedding-3-small") collection = create_milvus_collection() total = len(chunks) for i in range(0, total, batch_size): batch = chunks[i:i+batch_size] # 批量Embedding texts = [chunk["content"] for chunk in batch] embeddings = embedding_model.embed_documents(texts) # 准备插入数据 entities = [ [chunk["content"] for chunk in batch], embeddings, [chunk["metadata"]["source"] for chunk in batch], [chunk["metadata"]["filepath"] for chunk in batch], [chunk["metadata"]["doc_type"] for chunk in batch], [chunk["metadata"]["last_modified"] for chunk in batch], ] # 批量插入 collection.insert(entities) if (i + batch_size) % 1000 == 0: print(f"已入库 {min(i + batch_size, total)}/{total}") collection.flush() print(f"入库完成，共 {total} 个chunks")

性能数据：

• 5万份文档，解析后约 120万 个chunks
• Embedding入库速度：约 800 chunks/秒（batch_size=100，并行调用API）
• 总耗时：约 25 分钟（含解析）
• Milvus存储占用：约 4.5GB（向量）+ 500MB（文本元数据）

3.5 第五步：检索与生成

检索+生成流程

class TechDocQA: def __init__(self, milvus_collection, reranker_model): self.collection = milvus_collection self.reranker = reranker_model # BGE-reranker self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.1) # 混合检索：向量+关键词（互补） # 向量检索：语义相似度好，但精确术语可能漏检 # BM25：关键词精确匹配，补充向量检索 self.hybrid_retriever = self._build_hybrid_retriever() def _build_hybrid_retriever(self): """构建混合检索器：向量检索 + BM25关键词检索，加权合并""" # 向量检索器（LangChain包装） vector_retriever = ... # 连接到Milvus # BM25检索器（基于关键词） bm25_retriever = BM25Retriever.from_texts( texts=[...], # 所有chunk文本 metadatas=[...] ) # 权重：向量70%，BM25 30% ensemble = EnsembleRetriever( retrievers=[vector_retriever, bm25_retriever], weights=[0.7, 0.3] ) return ensemble def query(self, question, top_k=10, rerank_top_n=5): """ 查询流程： 1. 混合检索Top-K（检索广） 2. BGE-reranker重排（排准） 3. 取Top-N送LLM生成 """ # 步骤1：混合检索 retrieved_docs = self.hybrid_retriever.get_relevant_documents( question, k=top_k ) # 步骤2：重排序 # 把检索到的文档和原问题一起送入reranker # reranker会输出每个文档与问题的相关性分数 doc_contents = [doc.page_content for doc in retrieved_docs] rerank_results = self.reranker rerank( query=question, documents=doc_contents, top_n=rerank_top_n ) # 步骤3：取Top-N构建上下文 context_docs = [retrieved_docs[r["index"]] for r in rerank_results] context_text = "\n\n---\n\n".join([ f"[来源：{doc.metadata['source']}]\n{doc.page_content}" for doc in context_docs ]) # 步骤4：LLM生成（带约束的Prompt） prompt = f""" 你是一个制造业技术文档问答助手。请基于提供的参考资料回答用户问题。

规则： 1. 只使用参考资料中的信息，不要编造 2. 每个事实陈述后用[来源X]标注来自哪个文档 3. 如果资料不足以回答，明确说"资料不足以回答此问题" 4. 如果多个文档的信息有矛盾，指出这一点

参考资料： {context_text}

用户问题：{question}

回答： """ response = self.llm.invoke(prompt) return { "answer": response.content, "sources": [ { "source": doc.metadata["source"], "filepath": doc.metadata["filepath"], "relevance_score": r["score"] } for doc, r in zip(context_docs, rerank_results) ] }

4. 部署与效果数据

4.1 部署架构

4.2 上线后效果数据

| 指标 | 上线前 | 上线3个月后 | |------|--------|------------| | 问题解决率（工程师自评） | 25% | 89% | | 平均回答时间 | N/A（找不到答案） | 8.3秒 | | 答案来源可追溯 | 0% | 100% | | 日活用户数 | N/A | 213人/天 | | 工程师满意度 | 2.1/5 | 4.3/5 |

4.3 分类型效果

| 问题类型 | 准确率 | 说明 | |----------|--------|------| | 设备操作类（操作步骤） | 94% | 文档结构好，有明确步骤 | | 工艺参数类（温度/压力/时间） | 87% | 表格解析有时不准 | | 故障诊断类（原因+解决方案） | 82% | 老文档描述不完整 | | 安全规范类 | 96% | 文档质量高且格式标准 | | 人员/历史类（谁负责/什么时候） | 61% | 这类知识本就不在文档里 |

5. 踩坑全记录

坑1：扫描PDF的OCR质量（最大坑）

问题： 20%的PDF是扫描件，OCR识别率只有65%，很多关键参数被识别错误。

解决： 1. 先用

坑2：表格解析丢失结构

问题： 工艺参数文档中大量表格，解析后变成一维文本，参数名和值对不上。

解决：

在表格解析时强制保留行列对应关系

坑3：检索"设备型号"时召回不准

问题： 工程师问"A100设备怎么校准"，检索召回的是其他设备文档。

原因： A100是专有名词，但Embedding模型对设备型号的语义捕捉不够精确。

解决： 在检索前对Query做"设备型号提取+扩展"，生成同义词：