RAG 在工业设备维修手册中的应用：技术术语的向量化检索挑战 RAG 在工业设备维修手册中的应用技术术语的向量化检索挑战一、深度引言与场景痛点大家好我是赵咕咕。去年我们帮一家重型机械厂做一个智能维修助手。数据源是过去三十年积累的维修手册总计 1.2 万份 PDF覆盖了铣床、磨床、数控车床等 200 多种设备。需求很简单维修工输入设备故障现象系统检索相关维修方案。原型第一版用bge-large-zh做了向量化准确率只有 42%。排查后发现一个致命问题通用 embedding 模型根本理解不了离子氮化、静压导轨、电液伺服阀这种专业术语。它们被当成普通词语做了平均池化和离子、氮化、静压这些日常词混在一起语义全丢了。更糟糕的是——设备维修领域有大量同义词和缩略语。比例阀和伺服比例阀在手册里是同一个东西CNC 系统和计算机数字控制系统也是同一个东西。通用 embedding 模型把它们当成完全不同的概念。这篇文章我会分享如何在 RAG 系统中处理工业术语带来的向量化挑战。二、底层机制与原理深度剖析2.1 为什么通用 Embedding 模型在工业领域失效通用 embedding 模型的训练语料主要是新闻、百科、网页文本。当它遇到静压导轨时它会用静压的导轨这种组合语义去理解。但静压导轨是一个不可拆分的专有名词——它的意思完全不同于静止压力的导轨。这跟 NLP 里的专用名词实体识别NER是同一个问题通用模型不知道哪些词组是不可分割的术语单元。2.2 术语感知的向量化流水线这个流水线和普通 RAG 最大的区别在于术语词典的介入。术语词典在整个流程中扮演了三个角色分词锚点在文本分块时以术语为锚点不在术语内部切断。加权信号术语所在的句子或段落在向量化时获得额外权重。召回扩展检索时自动扩展同义词和缩略语提高召回率。2.3 BM25 向量的混合检索工业维修场景的一个特点是术语匹配比语义匹配更重要。维修工说润滑不良目标是找到润滑故障相关的手册条目——这是关键词匹配的强项。但如果维修工说这台机器跑起来声音很大像金属摩擦那就需要语义理解了。所以我们用 BM25 做第一路召回精确术语匹配向量做第二路召回语义模糊匹配最后由重排序模型做融合。三、生产级代码实现import asyncio import re import logging from dataclasses import dataclass, field from typing import Any import numpy as np import jieba logger logging.getLogger(__name__) dataclass class TermEntry: 术语词典条目。 term: str # 标准术语 synonyms: list[str] field(default_factorylist) # 同义词 abbreviations: list[str] field(default_factorylist) # 缩略语 category: str # 分类液压/电气/机械/数控 weight: float 2.0 # 向量化时的加权系数 class IndustrialTermManager: 工业术语管理器。 def __init__(self, terms: list[TermEntry] | None None): self._terms: dict[str, TermEntry] {} self._alias_to_term: dict[str, str] {} # 所有别名→标准术语 if terms: for t in terms: self.add_term(t) def add_term(self, term: TermEntry) - None: 添加术语并建立倒排索引。 self._terms[term.term] term self._alias_to_term[term.term] term.term for s in term.synonyms: self._alias_to_term[s] term.term for a in term.abbreviations: self._alias_to_term[a] term.term def normalize(self, text: str) - str: 将文本中的别名归一化为标准术语。 # 按长度降序排列优先匹配长术语避免比例匹配了比例阀 sorted_aliases sorted( self._alias_to_term.keys(), keylen, reverseTrue, ) result text for alias in sorted_aliases: standard self._alias_to_term[alias] if alias ! standard: # 别名才需要替换 result result.replace(alias, standard) return result def extract_terms(self, text: str) - list[tuple[str, float]]: 从文本中提取术语及其权重。 使用正向最长匹配算法精确识别术语边界。 found: dict[str, tuple[int, float]] {} # 按术语长度降序最长匹配优先 sorted_terms sorted( self._terms.items(), keylambda x: len(x[0]), reverseTrue, ) pos 0 while pos len(text): matched False for term, entry in sorted_terms: if text[pos:pos len(term)] term: # 检查是否为完整词边界 end pos len(term) is_word_boundary True if pos 0 and text[pos - 1].isalnum(): is_word_boundary False if end len(text) and text[end].isalnum(): is_word_boundary False if is_word_boundary: found[term] (pos, entry.weight) pos end matched True break if not matched: pos 1 return [(t, w) for t, (_, w) in found.items()] def expand_query(self, query: str) - str: 查询扩展加入同义词以提高召回率。 expanded_parts [query] terms self.extract_terms(query) for term, _ in terms: entry self._terms.get(term) if entry: # 加入同义词 for syn in entry.synonyms: expanded_parts.append(syn) return .join(expanded_parts) class IndustrialRAG: 工业设备维修手册 RAG 系统。 def __init__( self, term_manager: IndustrialTermManager, embedding_model: Any, vector_store: Any, bm25_index: Any, llm_client: Any, ): self._terms term_manager self._embedder embedding_model self._vector_store vector_store self._bm25 bm25_index self._llm llm_client async def search( self, query: str, top_k: int 10, timeout: float 30.0 ) - list[dict[str, Any]]: 混合检索主入口。 try: return await asyncio.wait_for( self._search_impl(query, top_k), timeouttimeout, ) except asyncio.TimeoutError: logger.error(检索超时: query%.50s, query) return [] async def _search_impl( self, query: str, top_k: int ) - list[dict[str, Any]]: # ── 第一步术语归一化 ── normalized_query self._terms.normalize(query) # ── 第二步查询扩展同义词 ── expanded_query self._terms.expand_query(normalized_query) # ── 第三步BM25 关键词召回 ── try: bm25_results self._bm25.search( expanded_query, ktop_k ) except Exception as e: logger.error(BM25 检索失败: %s, e) bm25_results [] # ── 第四步向量语义召回 ── try: # 提取术语并加权 weighted_query self._build_weighted_query(normalized_query) query_emb await self._embedder.embed_query(weighted_query) vec_results self._vector_store.similarity_search_with_score( query_emb, ktop_k ) except Exception as e: logger.error(向量检索失败: %s, e) vec_results [] # ── 第五步RRF 融合 ── merged self._rrf_fusion(bm25_results, vec_results, k60) # ── 第六步LLM 重排序 ── reranked await self._rerank(query, merged[:top_k * 2]) return reranked[:top_k] def _build_weighted_query(self, text: str) - str: 构建术语加权查询文本。 terms self._terms.extract_terms(text) if not terms: return text # 术语重复加权简单策略重复术语来增加 embedding 权重 parts [text] for term, weight in terms: repeat_count int(weight) for _ in range(repeat_count - 1): parts.append(term) return .join(parts) staticmethod def _rrf_fusion( bm25_results: list[tuple[str, float]], vec_results: list[tuple[str, float]], k: int 60, ) - list[tuple[str, float]]: RRF (Reciprocal Rank Fusion) 融合两路检索结果。 scores: dict[str, float] {} for rank, (doc_id, _) in enumerate(bm25_results, 1): scores[doc_id] scores.get(doc_id, 0) 1.0 / (k rank) for rank, (doc_id, _) in enumerate(vec_results, 1): scores[doc_id] scores.get(doc_id, 0) 1.0 / (k rank) return sorted(scores.items(), keylambda x: x[1], reverseTrue) async def _rerank( self, query: str, candidates: list[tuple[str, float]] ) - list[dict[str, Any]]: LLM 重排序候选文档。 if not candidates: return [] doc_list \n---\n.join( f[文档{i}]: {doc[:200]} for i, (doc, _) in enumerate(candidates) ) prompt f以下是设备维修查询{query}的候选检索结果。 请按相关性从高到低排序返回排序后的文档编号列表。 {doc_list} 返回格式: JSON数组如 [3, 1, 5, 2, 4] try: response await self._llm.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}], temperature0, response_format{type: json_object}, ) import json data json.loads( response.choices[0].message.content or [] ) ranks data if isinstance(data, list) else data.get(ranking, []) result [] for idx in ranks: if 0 idx len(candidates): doc, score candidates[idx] result.append({content: doc, score: float(score)}) return result except Exception as e: logger.error(LLM 重排序失败: %s, e) return [ {content: doc, score: float(score)} for doc, score in candidates ] # ─── 术语词典示例 ─── async def main(): # 构建工业术语词典 terms [ TermEntry( term静压导轨, synonyms[静压滑动导轨, 液体静压导轨], category机械, weight2.0, ), TermEntry( term比例阀, synonyms[伺服比例阀, 电液比例阀], abbreviations[PPV], category液压, weight2.5, ), TermEntry( termCNC系统, synonyms[计算机数字控制系统, 数控系统], abbreviations[CNC], category数控, weight2.0, ), TermEntry( term离子氮化, synonyms[等离子氮化, 辉光离子氮化], category热处理, weight2.0, ), ] manager IndustrialTermManager(terms) # 测试术语提取 text CNC系统的静压导轨需要定期检查比例阀的工作状态 extracted manager.extract_terms(text) print(提取术语:, extracted) # 测试归一化 normalized manager.normalize(text) print(归一化后:, normalized) # 测试查询扩展 expanded manager.expand_query(比例阀故障) print(扩展查询:, expanded) if __name__ __main__: asyncio.run(main())代码中的关键设计正向最长匹配按术语长度降序排列先匹配长的。这样伺服比例阀不会先被比例匹配掉。词边界检查术语匹配时检查前后字符避免离子氮化匹配到高离子氮化设备中的部分虽然在这个例子里是合理的但防止误匹配。术语加权策略在查询 embedding 时将术语重复插入查询文本中让 embedding 模型对术语部分加权。RRF 融合BM25 和向量检索结果通过 RRF 融合避免任一路的分数差异过大。LLM 重排序兜底LLM 失败时降级为原始融合排序不阻塞检索。四、边界分析与架构权衡4.1 术语词典的维护成本工业术语词典是这个方案的核心但也是维护成本最高的部分。2000 术语不可能一次性建完一个月后还会新增。建议的流程初始化从已有维修手册中自动提取高频专有名词TF-IDF 人工审核。增量更新每次人工标注的检索不满意案例检查是否因为术语缺失如果是则补充入词典。版本控制术语词典用 Git 管理每次变更做 Code Review。4.2 BM25 还是 Dense在工业维修场景BM25 的精确关键词匹配比向量语义匹配更重要。维修工查询的目标不是找到意思相近的文档而是找到包含这个故障名的维修方案。建议BM25 路权重0.6向量路权重0.4如果全用向量一个查询主轴发热可能会返回发动机过热的文档——语义上很接近但在设备维修场景里毫无用处。4.3 什么时候需要术语词典场景是否需要术语词典问答通用百科内容不需要通用 embedding 足够某个垂直行业领域知识必须有否则准确率低于 50%多语言混合场景需要多语言术语对齐词典缩写/行业黑话频繁出现必须有缩略语→标准术语映射纯技术 API 文档embedding 通常够用API 命名本来就有规律4.4 同义词扩展的风险查询扩展是一把双刃剑。加入同义词能提高召回率但可能降低精确率。比如刀架的同义词是工具架如果在CNC 刀架故障查询中加入工具架可能会召回一批仓库工具架管理的无关文档。缓解方案同义词扩展只在初次检索返回结果太少时触发。先做一轮精确检索结果数 3 时才扩展。五、总结工业维修 RAG 的核心挑战不是向量检索算法本身而是让 embedding 模型认识那些对通用模型来说完全陌生的工业术语。解决方案的关键是术语词典驱动的多层处理分块时保护术语完整性不在术语中间切断。向量化时加权术语让术语在 embedding 中占据更大比重。检索时扩展同义词提高不同说法下的召回率。融合时平衡精确与召回BM25 精确匹配 向量语义补充 RRF 融合。通用 embedding 模型像是一个能读所有书的通才但在工业领域你需要教它认识这个领域的方言。术语词典就是那本方言字典。有意思的是这个词典本身也随着系统使用越来越有价值——每次维修工反馈搜不到可能就是一个新术语的发现。日积月累词典成为企业知识体系的核心资产。下一篇预告Redis 灾备方案异地多活场景下向量索引的同步延迟和一致性取舍。