别再让业务开发半夜问你“参数怎么配“了:基于Milvus+大模型的信创数据库运维知识库构建实录
🔥关注墨瑾轩,带你探索编程的奥秘!🚀
🔥超萌技术攻略,轻松晋级编程高手🚀
🔥技术宝库已备好,就等你来挖掘🚀
🔥订阅墨瑾轩,智趣学习不孤单🚀
🔥即刻启航,编程之旅更有趣🚀


正片第一幕:数据榨汁机——如何把"天书"文档变成AI能消化的饲料
搞RAG(检索增强生成)知识库,第一步就是搞数据。
国产数据库的数据源极其恶心:有PDF格式的用户手册、有HTML格式的在线文档、有Word格式的培训PPT、还有微信群里的聊天截图。
如果你直接把这些东西丢给LangChain的默认 RecursiveCharacterTextSplitter,我保证你的AI会变成一个满嘴胡话的智障。
1.1 为什么技术文档不能用"傻瓜式"切片?
技术文档有极强的层级上下文依赖。
比如达梦文档里有一句话:“该参数默认值为0。”
如果你按500字一刀切,这句话被切到了一个独立的Chunk里。当用户问"达梦的 SORT_BUF_SIZE 默认值是多少"时,AI检索到了这个Chunk,回答:“该参数默认值为0。”
用户会骂娘的:哪个参数默认值为0?! 上下文全丢了!
1.2 核心代码:基于DOM树与正则的"语义感知"切片器
为了解决这个问题,我手写了一个基于文档结构(HTML/Markdown标题)和SQL代码块的语义感知切片器(Semantic Chunker)。
# ============================================================
# 依赖安装
# pip install beautifulsoup4 markdownify pandas jieba
# ============================================================
import re
import hashlib
from bs4 import BeautifulSoup, Tag
from typing import List, Dict, Any
class DbaDocumentChunker:
"""
============================================================
DBA 专属文档切片器
为什么不用 LangChain 的默认切片器?
因为技术文档有三大特殊性:
1. 标题层级决定了上下文(H1 > H2 > H3)
2. SQL代码块绝对不能被从中间切断(切断了就是语法错误)
3. 错误码(如 -2007)必须和它的解释死死绑定在一起
这个类的设计思路:
先按 HTML/Markdown 的标题层级(H1-H4)把文档拆成"章节"
如果章节太大,再按段落拆分
如果包含 SQL 代码块,把代码块作为一个不可分割的整体
最后,给每个 Chunk 注入"上下文面包屑"(Context Breadcrumb)
============================================================
"""
def __init__(self, max_chunk_size: int = 800, overlap: int = 100):
# 单个 Chunk 的最大字符数
# 800 是个经验值:太小会导致检索时丢失上下文,太大会引入噪音
self.max_chunk_size = max_chunk_size
# 相邻 Chunk 的重叠字符数(防止关键信息刚好被切断)
self.overlap = overlap
# 匹配国产数据库常见错误码的正则
# 达梦:-2007, -2115;金仓:ERROR: 42P01;OB:OB_ERR_XXX
# 【关键】把错误码提取出来作为元数据,检索时错误码的权重极高
self.error_code_pattern = re.compile(
r'(?:错误码|Error|ERROR|ORA-|OB_ERR_|DM-|KingbaseES Error)[:\s]*([-a-zA-Z0-9_]+)',
re.IGNORECASE
)
# 匹配 SQL 代码块的正则(Markdown 格式)
self.sql_block_pattern = re.compile(r'```(?:sql|SQL|plsql)?\n(.*?)```', re.DOTALL)
def parse_html_to_semantic_chunks(self, html_content: str, source_url: str) -> List[Dict[str, Any]]:
"""
解析 HTML 格式的官方在线文档(如达梦、金仓的在线手册)
"""
soup = BeautifulSoup(html_content, 'html.parser')
# ============================================================
# 第一步:提取文档的"面包屑"导航(Breadcrumb)
#
# 为什么要提取面包屑?
# 因为官方文档的标题经常重复。比如每个章节都有"配置参数"、"注意事项"
# 如果不带面包屑,检索时根本分不清这是哪一章的注意事项
# 面包屑示例:DM8 -> DBA指南 -> 内存管理 -> MEMORY_TARGET
# ============================================================
breadcrumb = self._extract_breadcrumb(soup)
# ============================================================
# 第二步:按标题层级(H1-H4)切割文档树
#
# 我们遍历 DOM 树,遇到 H1-H4 标签就认为是一个新的"语义段落"起点
# 把标题和它下面的内容(直到下一个同级或更高级标题)打包在一起
# ============================================================
sections = self._split_by_headings(soup)
chunks = []
for section in sections:
# 第三步:对过长的段落进行二次切分(保留代码块完整性)
section_chunks = self._smart_split_section(section)
for chunk_text in section_chunks:
# 第四步:提取元数据(错误码、SQL关键字)
metadata = self._extract_metadata(chunk_text, source_url, breadcrumb)
chunks.append({
"text": chunk_text,
"metadata": metadata
})
return chunks
def _split_by_headings(self, soup: BeautifulSoup) -> List[Dict[str, str]]:
"""
按标题切割 DOM 树
"""
sections = []
current_heading = "文档开头"
current_content = []
# 遍历所有元素
for element in soup.find_all(['h1', 'h2', 'h3', 'h4', 'p', 'pre', 'ul', 'ol', 'table']):
if element.name in ['h1', 'h2', 'h3', 'h4']:
# 遇到新标题,保存之前的内容
if current_content:
sections.append({
"heading": current_heading,
"content": "\n".join(current_content)
})
current_heading = element.get_text(strip=True)
current_content = [f"### {current_heading}"]
else:
# 提取文本内容,保留代码块的原始格式
if element.name == 'pre':
current_content.append(f"```\n{element.get_text()}\n```")
else:
current_content.append(element.get_text(strip=True))
# 别忘了最后一段
if current_content:
sections.append({"heading": current_heading, "content": "\n".join(current_content)})
return sections
def _smart_split_section(self, section: Dict[str, str]) -> List[str]:
"""
智能切分过长的段落
【核心逻辑】:SQL代码块绝对不能被切断!
"""
content = section['content']
heading = section['heading']
if len(content) <= self.max_chunk_size:
# 没超标,直接返回,带上标题作为上下文
return [f"[上下文: {heading}]\n{content}"]
# 超标了,需要切分
# 先把 SQL 代码块抠出来,用占位符替换,防止被按字符切断
sql_blocks = {}
def replace_sql(match):
block_id = f"__SQL_BLOCK_{len(sql_blocks)}__"
sql_blocks[block_id] = match.group(0)
return block_id
text_without_sql = self.sql_block_pattern.sub(replace_sql, content)
# 按段落(双换行)切分纯文本
paragraphs = text_without_sql.split('\n\n')
chunks = []
current_chunk = f"[上下文: {heading}]\n"
for para in paragraphs:
if len(current_chunk) + len(para) > self.max_chunk_size:
# 当前块满了,保存并开启新块
chunks.append(self._restore_sql_blocks(current_chunk, sql_blocks))
# 新块要带上重叠部分(overlap)和标题上下文
overlap_text = current_chunk[-self.overlap:] if len(current_chunk) > self.overlap else ""
current_chunk = f"[上下文: {heading}]\n{overlap_text}\n{para}\n"
else:
current_chunk += f"{para}\n"
if current_chunk:
chunks.append(self._restore_sql_blocks(current_chunk, sql_blocks))
return chunks
def _restore_sql_blocks(self, text: str, sql_blocks: Dict[str, str]) -> str:
"""把占位符替换回真实的 SQL 代码块"""
for block_id, block_content in sql_blocks.items():
text = text.replace(block_id, block_content)
return text
def _extract_metadata(self, text: str, source: str, breadcrumb: str) -> Dict[str, Any]:
"""
提取元数据——这是提升检索准确率的"秘密武器"
"""
metadata = {
"source": source,
"breadcrumb": breadcrumb,
"error_codes": [],
"has_sql": False
}
# 提取错误码
# 【为什么这很重要?】
# 向量检索(Embedding)对数字和字母组合(如 -2007)极不敏感!
# 在向量空间里,"-2007" 和 "-2008" 的距离可能很近,也可能很远,完全不可控
# 所以必须把错误码提取出来,作为元数据(Metadata)
# 检索时,先用元数据精确过滤(Pre-filtering),再用向量算相似度
codes = self.error_code_pattern.findall(text)
metadata["error_codes"] = list(set(codes))
# 标记是否包含 SQL
if "```sql" in text.lower() or "select " in text.lower():
metadata["has_sql"] = True
return metadata
正片第二幕:给AI装上"DBA的直觉"——混合检索与重排机制
数据切好了,下一步是检索。
如果你只用向量检索(Dense Retrieval),我敢打赌,你的知识库在回答具体参数和错误码时,会疯狂翻车。
2.1 向量检索的"死穴"
大模型的 Embedding 模型(比如 text-embedding-ada-002 或 bge-large-zh)擅长理解语义。
你问:“如何优化慢查询?” 它能给你找到"SQL调优指南"。
但是,DBA的日常不是问这种宏观问题。DBA的日常是:
“金仓的
work_mem参数默认值是多少?”
“达梦的V$LOCK视图里TYPE字段为TX是什么意思?”
Embedding 模型对专有名词、参数名、视图名、错误码的区分能力极差!
在向量空间里,work_mem 和 shared_buffers 的向量距离可能非常近(因为它们都是内存参数),导致检索出错误的文档。
2.2 破局之道:BM25 + 向量 + Metadata 的"三栖混合检索"
为了解决这个问题,我放弃了纯向量数据库(如 Pinecone),选择了支持混合检索的 Milvus(或者 Elasticsearch 8.x)。
# ============================================================
# 依赖安装
# pip install pymilvus rank_bm25 sentence-transformers
# ============================================================
from pymilvus import (
connections, Collection, FieldSchema,
CollectionSchema, DataType, utility
)
from rank_bm25 import BM25Okapi
import jieba
import numpy as np
class DbaHybridRetriever:
"""
============================================================
DBA 混合检索器
架构设计:
1. BM25(关键字检索):负责精准匹配参数名、视图名、错误码
2. 向量检索(Dense Retrieval):负责语义模糊匹配(如"怎么解决卡顿")
3. Metadata 过滤(Pre-filtering):负责按数据库类型(DM/Kingbase/OB)隔离数据
为什么不直接用 ES 8.x?
ES 的向量检索性能在百万级以上时会下降,且 HNSW 调参复杂
Milvus 是国产开源向量数据库(信创加分项!),性能更强
============================================================
"""
def __init__(self, collection_name: str = "dba_knowledge_base"):
# 连接 Milvus
connections.connect("default", host="localhost", port="19530")
self.collection_name = collection_name
self._init_collection()
# 初始化 Embedding 模型
# 【选型血泪史】:
# 千万别用 OpenAI 的 ada-002 做国产库知识库!
# 因为它对中文技术术语(如"两阶段提交"、"红黑树"、"聚簇索引")的理解
# 不如国产微调模型。
# 这里选用 BAAI 的 bge-large-zh-v1.5,中文技术文档的 SOTA
from sentence_transformers import SentenceTransformer
self.embedder = SentenceTransformer('BAAI/bge-large-zh-v1.5')
# BM25 索引(内存中维护,用于关键字检索)
self.bm25 = None
self.bm25_corpus = []
def _init_collection(self):
"""
初始化 Milvus 集合(表结构)
"""
if utility.has_collection(self.collection_name):
utility.drop_collection(self.collection_name)
# ============================================================
# 字段设计——每一个字段都有它的使命
# ============================================================
fields = [
# 主键
FieldSchema(name="id", dtype=DataType.VARCHAR, is_primary=True, max_length=64),
# 原始文本(用于最终喂给大模型)
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
# 向量字段(1024维,bge-large 的输出维度)
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
# 【关键元数据】:数据库类型(dm8, kingbase, oceanbase, tidb)
# 为什么必须加这个?
# 因为达梦的 "undo" 和 OceanBase 的 "undo" 含义和配置完全不同!
# 如果不做隔离,检索出来的知识会张冠李戴,直接导致生产事故
FieldSchema(name="db_type", dtype=DataType.VARCHAR, max_length=32),
# 【关键元数据】:错误码列表(JSON数组转字符串存储)
# 用于精确匹配报错信息
FieldSchema(name="error_codes", dtype=DataType.VARCHAR, max_length=512),
# 来源 URL
FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=512)
]
schema = CollectionSchema(fields, description="DBA FAQ Knowledge Base")
self.collection = Collection(self.collection_name, schema)
# ============================================================
# 创建索引
#
# 向量索引选择 HNSW(Hierarchical Navigable Small World)
# 为什么不用 IVF_FLAT?
# 因为知识库的数据量通常在十万级以内,HNSW 在中小数据量下
# 召回率(Recall)和查询速度都碾压 IVF_FLAT
#
# M=16, efConstruction=200 是经验调参值
# M 越大图越密,召回率越高,但内存占用越大
# ============================================================
index_params = {
"metric_type": "COSINE", # 余弦相似度,适合文本语义
"index_type": "HNSW",
"params": {"M": 16, "efConstruction": 200}
}
self.collection.create_index(field_name="embedding", index_params=index_params)
def search(self, query: str, db_type: str, top_k: int = 5) -> List[Dict]:
"""
执行混合检索
"""
# 1. 生成查询向量
query_embedding = self.embedder.encode([query])[0]
# ============================================================
# 2. 构建 Metadata 过滤表达式(Pre-filtering)
#
# 这一步极其关键!
# 如果用户问的是达梦的问题,我们只在 db_type == 'dm8' 的数据里搜
# 这样直接把搜索空间缩小了 75%,准确率直线飙升
# ============================================================
filter_expr = f'db_type == "{db_type}"'
# 如果查询中包含明显的错误码(如 -2007),追加错误码过滤
# 这样能实现"一针见血"的精确打击
error_code_match = re.search(r'-\d{4}|OB_ERR_\w+', query)
if error_code_match:
code = error_code_match.group(0)
# Milvus 的 like 语法
filter_expr += f' and error_codes like "%{code}%"'
# 3. 执行向量检索
self.collection.load()
# search_params 中的 ef 必须大于等于 top_k
# ef 越大,搜索越精确,但越慢。设为 top_k 的 3-5 倍较合适
search_params = {"metric_type": "COSINE", "params": {"ef": top_k * 4}}
vector_results = self.collection.search(
data=[query_embedding],
anns_field="embedding",
param=search_params,
limit=top_k * 2, # 多召回一些,留给后面的重排(Rerank)
expr=filter_expr,
output_fields=["text", "source", "error_codes"]
)
# 4. 结合 BM25 进行分数融合(RRF - Reciprocal Rank Fusion)
# 这里简化处理,实际生产中应该用 RRF 算法融合向量分数和 BM25 分数
final_results = []
for hits in vector_results:
for hit in hits:
final_results.append({
"id": hit.id,
"text": hit.entity.get('text'),
"source": hit.entity.get('source'),
"score": hit.score
})
# 5. 【可选但强烈建议】接入 Rerank 模型进行精排
# 向量检索召回的是"语义相关",Rerank 模型(如 bge-reranker-large)
# 会逐字对比 query 和 document,给出更精准的交叉注意力分数
# 这一步能把准确率再提升 10%-15%
return final_results[:top_k]
正片第三幕:让AI像老DBA一样说话——Prompt工程与ReAct模式
数据检索出来了,现在要把这些上下文喂给大模型(LLM),让它生成最终的FAQ回答。
如果你直接用这种弱智Prompt:
“根据以下上下文回答问题:{context}。问题:{query}”
AI会给你一个极其官方、极其正确的废话:
“根据文档,建议您参考内存管理章节,调整相关参数以优化性能。”
我信你个鬼!我要的是具体改哪个参数、改多少、改完要不要重启!
3.1 赋予AI"资深DBA"的人设与思考链(CoT)
我们必须通过 Prompt Engineering,逼迫AI使用 ReAct(Reasoning and Acting) 模式,并且严格限制它的"幻觉"。
# ============================================================
# prompt_templates.py
#
# 这是整个知识库的"灵魂"
# 一个好的 Prompt 能让 7B 的模型发挥出 70B 的效果
# 一个烂 Prompt 能让 GPT-4 变成人工智障
# ============================================================
DBA_SYSTEM_PROMPT = """
# Role
你是一个拥有20年经验的资深国产数据库DBA(精通达梦DM8、人大金仓KingbaseES、OceanBase、TiDB、GaussDB)。
你说话的风格是:硬核、直接、一针见血。你讨厌"正确的废话",喜欢给出具体的操作步骤、SQL命令和配置参数。
你偶尔会吐槽国产数据库的反人类设计,但依然能给出专业的解决方案。
# Constraints (绝对不可逾越的红线)
1. 【禁止幻觉】:你只能基于提供的 进行回答。如果 中没有答案,你必须明确回答:"知识库中未找到相关记录,建议查阅官方最新手册或提交工单。" 绝对不允许自己瞎编参数名或SQL语法!
2. 【区分环境】:回答时必须明确该方案适用于哪种数据库(如:此参数仅适用于达梦DM8,金仓请使用 xxx)。
3. 【风险提示】:如果涉及修改 `dm.ini`、`postgresql.conf` 等核心配置文件,或者需要执行 `ALTER SYSTEM`、重启实例等高危操作,必须在回答末尾加上显眼的 `[⚠️ 高危操作警告]`,并提示在业务低峰期操作。
4. 【代码规范】:给出的 SQL 或 Shell 命令,必须使用 Markdown 代码块包裹,并指明是哪种语言(sql, bash, ini)。
# Thinking Process (思考链)
在给出最终回答前,请先在
"""
def build_rag_prompt(query: str, contexts: List[Dict]) -> str:
"""
组装最终的 User Prompt
"""
# 将检索到的上下文拼装
context_str = ""
for i, ctx in enumerate(contexts):
context_str += f"\n--- 参考资料 {i+1} (来源: {ctx['source']}) ---\n"
context_str += ctx['text']
user_prompt = f"""
请按照你的 Role 和 Constraints,先在 中思考,然后在 中给出你的专业解答。
"""
return user_prompt
3.2 流式输出与前端集成(让等待不那么煎熬)
大模型生成技术文档通常很长,如果等它全部生成完再返回,用户会以为页面卡死了。必须做流式输出(Streaming)。
# ============================================================
# api_server.py (基于 FastAPI + LangChain)
# ============================================================
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
import json
app = FastAPI(title="DBA AI Knowledge Base API")
class QueryRequest(BaseModel):
query: str
db_type: str # dm8, kingbase, oceanbase, etc.
# 假设我们已经初始化了 retriever 和 llm
# retriever = DbaHybridRetriever()
# llm = ChatOpenAI(model="qwen-max", streaming=True) # 信创环境推荐用通义千问或百川
@app.post("/api/v1/ask")
async def ask_dba(req: QueryRequest):
"""
流式问答接口
"""
# 1. 混合检索
contexts = retriever.search(req.query, req.db_type, top_k=5)
if not contexts:
raise HTTPException(status_code=404, detail="未检索到相关知识")
# 2. 构建 Prompt
prompt = build_rag_prompt(req.query, contexts)
# 3. 流式生成
async def event_generator():
# 先发送引用的参考资料来源(让前端可以先渲染"参考文档"列表)
sources = [{"source": c['source'], "score": c['score']} for c in contexts]
yield f"data: {json.dumps({'type': 'sources', 'data': sources})}\n\n"
# 调用 LLM 流式输出
# 这里使用 LangChain 的 astream 方法
async for chunk in llm.astream([
{"role": "system", "content": DBA_SYSTEM_PROMPT},
{"role": "user", "content": prompt}
]):
# 提取文本内容
content = chunk.content
if content:
# 以 SSE (Server-Sent Events) 格式发送
yield f"data: {json.dumps({'type': 'token', 'data': content})}\n\n"
# 发送结束标记
yield f"data: {json.dumps({'type': 'end'})}\n\n"
return StreamingResponse(event_generator(), media_type="text/event-stream")
正片第四幕:踩坑实录——那些文档上绝对不会写的血泪教训
坑1:Embedding 模型把 “OB” 和 “Oracle” 搞混了
现象:
用户问:“OceanBase(OB)的合并(Major Compaction)卡住了怎么办?”
AI回答了一堆 Oracle 数据库的 ALTER SYSTEM MERGE 语法。
原因:
在通用语料训练的 Embedding 模型眼里,“OB” 这个缩写的向量,和 “Oracle” 的向量距离非常近(因为它们在金融、数据库语境下经常共现)。导致检索时,把 Oracle 的文档给召回了。
解决方案:
- 查询重写(Query Rewriting):在检索前,用一个小模型把 “OB” 展开为 “OceanBase”。
- 硬规则过滤:在代码里写死字典,遇到 “OB” 强制映射为
db_type="oceanbase",走 Metadata 强过滤。
坑2:PDF解析的"表格噩梦"
现象:
达梦的《DM8系统管理员手册》里有大量的参数配置表格。用 PyPDF2 解析出来,表格变成了乱码一样的纯文本:参数名 默认值 说明 MEMORY_TARGET 0 内存目标...
AI 根本分不清哪个是参数名,哪个是默认值。
解决方案:
放弃纯文本解析,改用 Marker 或 Unstructured 库,它们能识别 PDF 中的表格结构,将其转换为 Markdown 表格或 HTML 表格。
pip install marker-pdf
转换后的 Markdown 表格,大模型理解起来毫无压力。
坑3:大模型的"致命自信"(幻觉导致删库)
现象:
用户问:“达梦怎么清理归档日志?”
AI 自信满满地给出了一个 Shell 命令:rm -rf /dmdata/arch/*
这是要命的! 直接删文件会导致达梦的 v$archived_log 视图和实际文件不一致,后续恢复直接报错。正确的做法是用达梦自带的 SP_ARCHIVE_LOG_DELETE_BEFORE_TIME 存储过程。
解决方案:
- Prompt 强约束:在 System Prompt 中加入绝对禁令:“禁止提供直接操作底层文件系统的
rm、del等 OS 命令,必须使用数据库内置的存储过程或管理命令!” - 后置拦截(Guardrails):在输出层加一个正则校验,如果检测到
rm -rf、drop database等高危词汇,直接拦截并替换为警告信息。
# 危险命令拦截器
DANGEROUS_PATTERNS = [
r'rm\s+-rf',
r'drop\s+database',
r'truncate\s+table',
r'shutdown\s+abort'
]
def guardrail_check(llm_output: str) -> str:
for pattern in DANGEROUS_PATTERNS:
if re.search(pattern, llm_output, re.IGNORECASE):
return "⚠️ [系统拦截]:AI 试图生成高危破坏性命令,已自动拦截。请人工复核操作风险!"
return llm_output
尾声:DBA不会被AI取代,但不用AI的DBA会被淘汰
把烟头按灭在已经 overflowing 的烟灰缸里,伸了个懒腰
兄弟们,这套"国产库FAQ自动构建系统"上线后,我们组的日子好过了不少。
数据说话:
- 内部工单量下降了 65%(那些"报错-2007怎么办"的弱智问题全被AI挡在门外了)。
- 新人DBA的上手时间从 3个月 缩短到了 3周(遇到不懂的直接问AI,AI会连带着把底层原理和官方文档出处一起甩给他)。
- 最爽的是,原厂支持兄弟看我们的眼神都变了——因为我们提的工单,不再是"报错了怎么办",而是"我看了你们的源码/日志,怀疑是XX模块的XX锁竞争,请确认"。
但我要说一句掏心窝子的话:
AI 永远无法替代 DBA 的核心价值。
AI 能告诉你 MEMORY_TARGET 怎么配,但它不知道你们公司那台跑了5年的破服务器,一到月底结账内存就会漏电;
AI 能给你写出完美的分库分表 SQL,但它不知道业务线那个傻逼产品经理明天又要改需求,你的表结构必须留出冗余;
AI 能在故障发生时给你列出排查步骤,但只有你,能在凌晨三点、顶着血压飙升的风险,果断敲下 kill -9 并承担起可能丢失10秒数据的责任。
更多推荐
所有评论(0)