基于大模型的数据库运维知识库构建：从日志到智能问答的排障助手

一、数据库运维的知识瓶颈：排障经验难以系统化传承

数据库运维的核心挑战不是缺少监控数据，而是缺少将告警、日志、指标与排障方案关联的知识体系。一个资深 DBA 看到"MySQL 1205 Lock wait timeout"就知道是死锁，看到"ClickHouse Too many parts"就知道合并策略需要调整——但这些经验散落在各人的笔记和记忆中。新人遇到问题只能翻文档、搜工单、问同事，平均排障时间远超预期。大模型的出现让构建智能排障知识库成为可能：将历史工单、排障记录、官方文档结构化入库，结合 RAG 技术实现自然语言问答。

graph TB
    A[运维知识来源] --> B[历史工单<br/>5000+ 条]
    A --> C[排障记录<br/>Wiki/Confluence]
    A --> D[官方文档<br/>MySQL/ClickHouse]
    A --> E[监控告警<br/>Prometheus 规则]

    B --> F[文档解析与切片]
    C --> F
    D --> F
    E --> F
    F --> G[向量化 Embedding]
    G --> H[向量数据库<br/>Milvus/Qdrant]

    I[用户提问] --> J[Query 改写与扩展]
    J --> K[向量检索 Top-K]
    H --> K
    K --> L[上下文组装 + LLM 推理]
    L --> M[排障方案输出]

    M --> N[用户反馈]
    N -->|有效| O[知识回流<br/>补充到知识库]
    N -->|无效| P[人工介入<br/>生成新工单]

二、RAG 架构与知识库构建的底层机制

2.1 文档切片策略对检索质量的影响

数据库运维文档的切片粒度直接影响检索精度。按固定长度切片会截断完整的排障步骤，按段落切片可能丢失上下文。最优策略是按"问题-方案"对切片：每个切片包含一个完整的故障描述和对应的排障步骤。

graph LR
    A[原始工单文档] --> B{切片策略}
    B -->|固定长度 512 token| C[截断风险高<br/>检索召回率 65%]
    B -->|按段落切片| D[上下文丢失<br/>检索召回率 72%]
    B -->|按问题-方案对| E[语义完整<br/>检索召回率 89%]

    E --> F[元数据标注]
    F --> G[数据库类型: MySQL/CK]
    F --> H[故障类别: 锁/性能/复制]
    F --> I[严重程度: P0-P3]

2.2 混合检索：向量检索 + 关键词检索的互补

纯向量检索擅长语义匹配但可能遗漏精确关键词（如错误码 1205、参数名 max_parts_in_total）。混合检索同时执行向量检索和 BM25 关键词检索，通过 Reciprocal Rank Fusion（RRF）融合结果。

2.3 Query 改写与多轮对话管理

用户提问往往信息不足——"数据库慢了"没有指明哪个库、什么操作慢。Query 改写模块利用 LLM 补全上下文，将模糊问题转为精确检索条件。

三、生产级代码实现与最佳实践

3.1 文档解析与切片管线

import re
from dataclasses import dataclass
from typing import List, Optional
from langchain.text_splitter import RecursiveCharacterTextSplitter


@dataclass
class TroubleshootChunk:
    """排障知识切片"""
    content: str
    metadata: dict  # 数据库类型、故障类别、严重程度
    chunk_id: str


class DBTroubleshootSplitter:
    """按问题-方案对切片的分割器"""

    # 匹配工单中的故障描述和排障步骤
    PROBLEM_PATTERN = re.compile(
        r'(?:故障描述|问题现象|问题描述)[:：]\s*(.+?)(?=排障|解决|方案|原因|$)',
        re.DOTALL
    )
    SOLUTION_PATTERN = re.compile(
        r'(?:排障过程|解决方案|处理步骤|根因分析)[:：]\s*(.+?)(?=故障|问题|总结|$)',
        re.DOTALL
    )

    def split_ticket(self, ticket_text: str, metadata: dict) -> List[TroubleshootChunk]:
        """将工单拆分为问题-方案对"""
        problems = self.PROBLEM_PATTERN.findall(ticket_text)
        solutions = self.SOLUTION_PATTERN.findall(ticket_text)

        chunks = []
        # 将问题与方案配对
        for i, (problem, solution) in enumerate(
            zip(problems, solutions)
        ):
            combined = f"故障现象: {problem.strip()}\n排障方案: {solution.strip()}"
            chunk = TroubleshootChunk(
                content=combined,
                metadata={
                    **metadata,
                    "chunk_type": "problem_solution_pair",
                    "pair_index": i,
                },
                chunk_id=f"{metadata['ticket_id']}_pair_{i}"
            )
            chunks.append(chunk)

        # 未配对的剩余内容用通用切片器处理
        remaining = self._extract_remaining(ticket_text, problems, solutions)
        if remaining.strip():
            fallback_splitter = RecursiveCharacterTextSplitter(
                chunk_size=800,
                chunk_overlap=100,
                separators=["\n\n", "\n", "。", "；"]
            )
            for j, text in enumerate(fallback_splitter.split_text(remaining)):
                chunks.append(TroubleshootChunk(
                    content=text,
                    metadata={**metadata, "chunk_type": "fallback", "chunk_index": j},
                    chunk_id=f"{metadata['ticket_id']}_fallback_{j}"
                ))

        return chunks

    def _extract_remaining(self, text: str, problems, solutions) -> str:
        """提取未被配对匹配到的文本"""
        used_spans = []
        for pattern, items in [(self.PROBLEM_PATTERN, problems), (self.SOLUTION_PATTERN, solutions)]:
            for match in pattern.finditer(text):
                used_spans.append((match.start(), match.end()))

        remaining_parts = []
        prev_end = 0
        for start, end in sorted(used_spans):
            if start > prev_end:
                remaining_parts.append(text[prev_end:start])
            prev_end = end
        if prev_end < len(text):
            remaining_parts.append(text[prev_end:])

        return "\n".join(remaining_parts)

3.2 混合检索与 RRF 融合

from typing import List, Dict, Tuple
import numpy as np


class HybridRetriever:
    """向量检索 + BM25 关键词检索的混合检索器"""

    def __init__(self, vector_store, bm25_index, embedding_model):
        self.vector_store = vector_store
        self.bm25_index = bm25_index
        self.embedding_model = embedding_model

    def search(
        self,
        query: str,
        top_k: int = 10,
        vector_weight: float = 0.7,
        bm25_weight: float = 0.3,
        filters: Optional[Dict] = None
    ) -> List[Tuple[TroubleshootChunk, float]]:
        """混合检索并融合排序"""

        # 1. 向量检索
        query_embedding = self.embedding_model.embed_query(query)
        vector_results = self.vector_store.similarity_search_with_score(
            query_embedding, top_k=top_k * 2, filters=filters
        )

        # 2. BM25 关键词检索
        bm25_results = self.bm25_index.search(query, top_k=top_k * 2)

        # 3. Reciprocal Rank Fusion 融合
        fused_scores = self._rrf_fusion(
            vector_results, bm25_results,
            vector_weight, bm25_weight
        )

        # 4. 按融合分数排序返回 Top-K
        sorted_results = sorted(
            fused_scores.items(),
            key=lambda x: x[1],
            reverse=True
        )[:top_k]

        return [(chunk_id_to_chunk[cid], score) for cid, score in sorted_results]

    def _rrf_fusion(
        self,
        vector_results: List,
        bm25_results: List,
        v_weight: float,
        b_weight: float,
        k: int = 60  # RRF 平滑参数
    ) -> Dict[str, float]:
        """Reciprocal Rank Fusion 融合算法"""
        fused = {}

        # 向量检索结果贡献
        for rank, (chunk_id, _) in enumerate(vector_results):
            if chunk_id not in fused:
                fused[chunk_id] = 0.0
            fused[chunk_id] += v_weight / (k + rank + 1)

        # BM25 检索结果贡献
        for rank, (chunk_id, _) in enumerate(bm25_results):
            if chunk_id not in fused:
                fused[chunk_id] = 0.0
            fused[chunk_id] += b_weight / (k + rank + 1)

        return fused

3.3 Query 改写与排障问答

from openai import OpenAI


class DBTroubleshootQA:
    """数据库排障智能问答"""

    SYSTEM_PROMPT = """你是一个数据库运维排障助手。根据检索到的历史排障知识，回答用户的问题。

规则：
1. 优先引用检索到的知识片段，标注来源工单编号
2. 如果检索结果不足以回答问题，明确说明并建议人工排查
3. 给出排障步骤时按优先级排序：先止血、再定位、后修复
4. 涉及数据库参数修改时，给出修改前后的值和回滚方案
5. 不要编造不存在的错误码或参数名"""

    def __init__(self, retriever: HybridRetriever, llm_client: OpenAI):
        self.retriever = retriever
        self.llm = llm_client

    def rewrite_query(self, raw_query: str, chat_history: List[dict]) -> str:
        """利用 LLM 改写用户查询，补全上下文"""
        rewrite_prompt = f"""根据对话历史，将用户的模糊提问改写为精确的检索查询。
原始提问: {raw_query}
对话历史: {chat_history[-3:] if chat_history else '无'}

改写要求:
- 补全缺失的数据库类型（MySQL/ClickHouse/PostgreSQL）
- 提取错误码和关键参数名
- 明确故障类别（锁/性能/复制/存储）

只输出改写后的查询，不要解释。"""

        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": rewrite_prompt}],
            temperature=0.1
        )
        return response.choices[0].message.content.strip()

    def answer(self, question: str, chat_history: List[dict] = None) -> dict:
        """回答排障问题"""
        # 1. Query 改写
        rewritten = self.rewrite_query(question, chat_history or [])

        # 2. 混合检索
        retrieved = self.retriever.search(rewritten, top_k=5)

        # 3. 组装上下文
        context_parts = []
        for chunk, score in retrieved:
            source = chunk.metadata.get('ticket_id', '未知来源')
            context_parts.append(
                f"[来源: {source} | 相关度: {score:.3f}]\n{chunk.content}"
            )
        context = "\n\n---\n\n".join(context_parts)

        # 4. LLM 生成回答
        response = self.llm.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": self.SYSTEM_PROMPT},
                {"role": "user", "content": f"检索到的排障知识:\n{context}\n\n用户问题: {question}"}
            ],
            temperature=0.2
        )

        return {
            "answer": response.choices[0].message.content,
            "rewritten_query": rewritten,
            "sources": [chunk.metadata for chunk, _ in retrieved],
            "confidence": min(score for _, score in retrieved[:3]) if retrieved else 0.0
        }

四、知识库构建的架构权衡

4.1 检索质量 vs 构建成本

4.2 切片粒度 vs 上下文完整性

过细的切片（200 token）检索精度高但上下文不完整，过粗的切片（2000 token）上下文完整但检索噪声大。按"问题-方案对"切片是运维场景的最优平衡点，平均切片长度 500-800 token。

4.3 适用边界与禁用场景

适用场景：

• 历史工单超过 1000 条的中大型运维团队
• 故障类型可归类、排障步骤可标准化的场景
• 新人培训与排障知识传承

禁用场景：

• 工单数据不足 100 条，RAG 检索无有效召回
• 故障高度定制化，无法从历史经验推导
• 涉及敏感数据（如生产密码、密钥），不适合入库

五、总结

数据库运维知识库的核心价值不是替代 DBA，而是将散落的排障经验系统化、可检索化。RAG 架构的关键在于检索质量——混合检索比纯向量检索在运维场景更可靠，因为错误码、参数名等精确关键词的匹配至关重要。按"问题-方案对"切片比固定长度切片召回率提升约 20%，这是运维文档结构化程度高的天然优势。知识库需要持续运营：用户反馈回流、新工单自动入库、过期方案定期清理，否则知识库会逐渐失效。