基于RAG的数据库知识库构建：让LLM读懂内部运维文档与故障案例

2026/7/9 6:34:34

基于RAG的数据库知识库构建：让LLM读懂内部运维文档与故障案例

一、知识孤岛困境：运维文档写了三年，但故障时还是靠人肉检索

一家公司在三年内积累了超过 800 篇运维文档、1200 份故障复盘报告、50 个 SOP 手册。这些文档散落在 Confluence、GitLab Wiki、飞书文档和每个人的本地笔记中。当凌晨 3 点 MySQL 主从延迟告警响起时，值班工程师要同时翻阅四五个系统来查找类似问题的处理方案——而正确答案往往隐藏在去年季度复盘会议上某个同事分享的 PDF 里。

传统的检索方式——全文搜索——无法理解"主从延迟"和"Seconds_Behind_Master 突增"指的是同一个问题的不同表述。这让运维知识库成为一个"图书馆有所有答案但你永远找不到"的悖论。

基于 RAG（Retrieval-Augmented Generation）的解决方案重新定义了知识检索的方式：不是让用户搜索关键词，而是让用户用自然语言描述问题，系统自动从知识库中检索最相关的文档，并由 LLM 整合成可操作的答案。

二、RAG知识库的文档处理与检索增强管线

flowchart TB subgraph Ingestion["文档摄入"] A1[Confluence Wiki] --> C[文档解析与分块] A2[GitLab Wiki] --> C A3[飞书文档] --> C A4[PDF报告] --> C A5[Markdown笔记] --> C end C --> D[文本分块<br/>Chunking] D --> E[Embedding生成] E --> F[(向量数据库<br/>ChromaDB)] subgraph Query["查询流程"] G[用户提问] --> H[问题Embedding] H --> I[向量检索 Top-K] F --> I I --> J[重排序<br/>Reranker] J --> K[上下文组装] end K --> L[LLM回答生成] L --> M{质量检查} M -->|有引用| N[返回答案+源文档链接] M -->|无引用/不确定| O[提示信息不足] N --> P[(查询日志)] P --> Q[反馈循环] Q -->|人工标注| F

管道分为两个独立阶段：

离线摄入阶段：将各种格式的运维文档统一解析为纯文本，按照语义边界分块（Chunking），生成 Embedding 向量并存入向量数据库。这一阶段需要处理中文文本的分词边界、表格内容的保留、Markdown 代码块的语义完整性等特殊问题。

在线查询阶段：用户的自然语言问题被转化为向量，从向量数据库中检索最相关的 Top-K 文档块，经重排序模型优化后组装为 LLM 的上下文，最终生成带有明确引用来源的答案。

三、代码实现

3.1 文档摄入管道

import os import re from typing import List, Dict, Tuple from dataclasses import dataclass import hashlib from chromadb import PersistentClient from chromadb.utils import embedding_functions @dataclass class DocumentChunk: """文档分块""" doc_id: str title: str content: str chunk_index: int source: str # 来源系统 source_url: str metadata: Dict class DocumentIngestionPipeline: """运维文档摄入管道""" CHUNK_SIZE = 500 # 每个分块的字数 CHUNK_OVERLAP = 100 # 分块之间的重叠字数 def __init__(self, persist_dir: str = "./ops_kb"): self.client = PersistentClient(path=persist_dir) self.ef = embedding_functions.SentenceTransformerEmbeddingFunction( model_name="BAAI/bge-large-zh-v1.5" ) # 确保集合存在 try: self.collection = self.client.get_collection("ops_docs") except Exception: self.collection = self.client.create_collection( name="ops_docs", embedding_function=self.ef, metadata={"description": "运维知识库"} ) def ingest_from_markdown(self, file_path: str, source: str): """摄入 Markdown 文档""" with open(file_path, 'r', encoding='utf-8') as f: content = f.read() title = self._extract_title(content) doc_id = hashlib.md5(file_path.encode()).hexdigest()[:12] chunks = self._chunk_text(content) ids = [] documents = [] metadatas = [] for i, chunk in enumerate(chunks): chunk_id = f"{doc_id}_{i}" ids.append(chunk_id) documents.append(chunk) metadatas.append({ "doc_id": doc_id, "title": title, "chunk_index": i, "source": source, "source_path": file_path, "word_count": len(chunk) }) # 批量插入 self.collection.add( ids=ids, documents=documents, metadatas=metadatas ) print(f"摄入完成: {title} -> {len(chunks)} 个分块") def _chunk_text(self, text: str) -> List[str]: """智能文本分块，保持语义边界""" # 方法选择的多策略尝试： # 优先按 Markdown 标题分块，再按段落分块，最后按字数分块 chunks = [] # 移除标题行用于分块 body = re.sub(r'^#\s+.*\n', '', text) # 按双换行符（段落边界）初次分块 paragraphs = body.split('\n\n') current_chunk = "" for para in paragraphs: para = para.strip() if not para: continue # 代码块不拆开 if para.startswith('```') and current_chunk: current_chunk += '\n\n' + para continue if len(current_chunk) + len(para) < self.CHUNK_SIZE: current_chunk += '\n\n' + para if current_chunk else para else: if current_chunk.strip(): chunks.append(current_chunk.strip()) current_chunk = para if current_chunk.strip(): chunks.append(current_chunk.strip()) return chunks def _extract_title(self, text: str) -> str: """提取文档标题""" match = re.search(r'^#\s+(.+)', text, re.MULTILINE) return match.group(1) if match else "无标题"

3.2 查询与生成引擎

from typing import List, Dict, Optional import json class RAGQueryEngine: """基于 RAG 的知识库查询引擎""" SYSTEM_PROMPT = """你是一个运维知识助手。请基于提供的运维文档片段回答问题。 规则： 1. 只使用提供的文档内容回答，不得编造信息 2. 如果文档片段信息不足，明确告知用户"当前知识库中未找到相关信息" 3. 回答时引用具体的文档标题和来源 4. 对于故障排查类问题，按步骤给出排查方案 5. 使用中文回答""" def __init__(self, collection, llm_client): self.collection = collection self.llm = llm_client self.conversation_history: List[Dict] = [] def ask(self, question: str, top_k: int = 5) -> Dict: """处理用户问题并返回答案""" # 1. 检索相关文档 retrieved_docs = self._retrieve(question, top_k) if not retrieved_docs: return { "answer": "抱歉，在当前知识库中未找到与您问题相关的文档。建议检查问题描述是否准确，或联系 DBA 团队补充相关文档。", "sources": [] } # 2. 构建上下文 context = self._build_context(retrieved_docs) # 3. 生成回答 answer = self._generate(question, context) # 4. 提取引用来源 sources = self._extract_sources(retrieved_docs) return { "answer": answer, "sources": sources, "context_docs": [ {"title": d["title"], "content_preview": d["content"][:200]} for d in retrieved_docs ] } def _retrieve(self, question: str, top_k: int) -> List[Dict]: """向量检索相关文档""" results = self.collection.query( query_texts=[question], n_results=top_k ) docs = [] if results and results['ids'] and results['ids'][0]: for i, doc_id in enumerate(results['ids'][0]): docs.append({ "id": doc_id, "content": results['documents'][0][i], "metadata": results['metadatas'][0][i], "distance": results.get('distances', [[0]*top_k])[0][i] }) return docs def _build_context(self, docs: List[Dict]) -> str: """组装检索到的文档作为 LLM 上下文""" context_parts = [] for i, doc in enumerate(docs): meta = doc['metadata'] title = meta.get('title', '未知文档') source = meta.get('source', '未知来源') context_parts.append( f"【文档 {i+1}】{title}（来源：{source}）\n{doc['content']}" ) return "\n\n---\n\n".join(context_parts) def _generate(self, question: str, context: str) -> str: """调用 LLM 生成答案""" user_message = f"""用户问题：{question} 相关运维文档： {context} 请基于以上文档内容回答问题。""" response = self.llm.chat( system=self.SYSTEM_PROMPT, user_message=user_message, temperature=0.3 ) return response def _extract_sources(self, docs: List[Dict]) -> List[Dict]: """提取引用来源""" sources = [] seen = set() for doc in docs: meta = doc['metadata'] title = meta.get('title', '') if title not in seen: seen.add(title) sources.append({ "title": title, "source": meta.get('source', ''), "relevance": f"{1 - doc.get('distance', 0):.2%}" }) return sources

3.3 反馈循环与质量改进

class FeedbackManager: """收集用户反馈，持续改进检索质量""" def __init__(self, feedback_file: str = "./feedback.jsonl"): self.feedback_file = feedback_file def record(self, question: str, answer: str, rating: int, comment: str = ""): """记录用户反馈""" record = { "question": question, "answer_preview": answer[:200], "rating": rating, # 1-5 "comment": comment, "timestamp": datetime.now().isoformat() } with open(self.feedback_file, 'a') as f: f.write(json.dumps(record, ensure_ascii=False) + '\n') def get_low_quality_queries(self, threshold: int = 3) -> List[Dict]: """获取低质量反馈的问题，用于人工优化""" low_quality = [] if not os.path.exists(self.feedback_file): return low_quality with open(self.feedback_file) as f: for line in f: record = json.loads(line) if record['rating'] < threshold: low_quality.append(record) return low_quality