当前位置：首页 > news >正文

智能CMDB构建：利用NLP从运维工单中自动提取配置项依赖关系的技术路径

news 2026/7/7 15:58:15

智能CMDB构建：利用NLP从运维工单中自动提取配置项依赖关系的技术路径

一、传统CMDB的熵增困境

CMDB（Configuration Management Database）是运维体系的"地图"。理论上，它应当精确描述每个配置项（CI）的属性、状态和依赖关系。但现实中，CMDB的数据质量在导入后的6个月内就会下降到不可信赖的水平。

根本原因不在于工具，而在于数据维护依赖人工。当一个DBA在工单中写下"扩容order-db主库后需同步更新payment-service的连接串"，这条依赖关系只会停留在工单的文本里，不会进入CMDB。下一个接手的人可能根本不知道payment-service依赖order-db。这种信息的割裂随着人员变动和系统演进不断加剧，最终导致CMDB沦为一台昂贵的电子表格。

NLP（自然语言处理）技术提供了另一种可能：将运维工单、变更记录和故障报告作为原始语料，自动从中抽取CI和CI间的关系，持续更新CMDB。这是一种"从旁路数据重建主线数据"的思路。

二、信息抽取的NLP技术栈选型

flowchart TD A[运维工单文本] --> B[文本预处理] B --> B1[中文分词 jieba] B --> B2[领域词典注入] B --> B3[格式化清洗] B3 --> C[命名实体识别 NER] C --> C1[基于BERT-BiLSTM-CRF模型] C --> C2[识别实体类型: 服务/数据库/主机/中间件/配置项] C2 --> D[关系抽取 RE] D --> D1[依存句法分析] D --> D2[关系模式匹配] D --> D3[远程监督学习] D3 --> E[知识融合与消歧] E --> E1[实体对齐 同服务别名归一化] E --> E2[关系去重与版本管理] E2 --> F[CMDB增量更新]

工单文本是典型的中文短文本，包含大量领域缩略语（如"RDS"、"SLB"、"POD"）和自定义命名（如"order-service-prod-v2"）。通用的NLP模型在这些文本上表现很差，必须进行领域适配。

命名实体识别（NER）的目标是从工单中识别出配置项实体及其类型。例如，从"因order-db主库CPU持续>90%，将payment-service的DB连接池从50调整到100"中，识别出：[order-db: 数据库实例]、[payment-service: 微服务]、[DB连接池: 配置项]。

BERT-BiLSTM-CRF是当前中文NER效果最好的模型架构。BERT负责提取上下文语义特征，BiLSTM捕捉文本的序列依赖，CRF层约束标签序列的合理性（如B-ENTITY后只能接I-ENTITY或O）。

""" 基于BERT-BiLSTM-CRF的运维领域NER模型训练与推理 使用transformers + pytorch-crf """ import torch import torch.nn as nn from torch.utils.data import Dataset, DataLoader from transformers import BertTokenizer, BertModel, AdamW from torchcrf import CRF from typing import List, Dict, Tuple, Optional import json import os class OpsNERDataset(Dataset): """运维工单NER数据集""" # 实体标签定义（BIO标注体系） LABEL_MAP = { "O": 0, # 非实体 "B-SERVICE": 1, # 微服务实体-开始 "I-SERVICE": 2, # 微服务实体-内部 "B-DATABASE": 3, # 数据库实体-开始 "I-DATABASE": 4, # 数据库实体-内部 "B-HOST": 5, # 主机/容器实体-开始 "I-HOST": 6, # 主机/容器实体-内部 "B-MIDDLEWARE": 7, # 中间件实体-开始 "I-MIDDLEWARE": 8, # 中间件实体-内部 "B-CONFIG": 9, # 配置项实体-开始 "I-CONFIG": 10, # 配置项实体-内部 } # 反向映射：ID -> 标签名 ID_TO_LABEL = {v: k for k, v in LABEL_MAP.items()} def __init__( self, texts: List[str], labels: List[List[str]], tokenizer: BertTokenizer, max_length: int = 256 ): """ 初始化数据集 :param texts: 工单文本列表 :param labels: 对应每个文本的BIO标签序列 :param tokenizer: BERT分词器 :param max_length: 最大序列长度 """ if len(texts) != len(labels): raise ValueError( f"文本和标签列表长度不一致: {len(texts)} vs {len(labels)}" ) self.texts = texts self.labels = labels self.tokenizer = tokenizer self.max_length = max_length def __len__(self) -> int: return len(self.texts) def __getitem__(self, idx: int) -> Dict: text = self.texts[idx] labels = self.labels[idx] # BERT分词，返回input_ids、attention_mask等 encoding = self.tokenizer( text, max_length=self.max_length, padding="max_length", truncation=True, return_tensors="pt" ) # 构造标签序列 # CLS和SEP位置用O标签，PAD位置用-100（损失函数忽略） label_ids = [self.LABEL_MAP.get(l, 0) for l in labels] label_ids = label_ids[:self.max_length - 2] # 减2位给CLS和SEP label_ids = [self.LABEL_MAP["O"]] + label_ids + [self.LABEL_MAP["O"]] # 填充到max_length padding_length = self.max_length - len(label_ids) label_ids += [-100] * padding_length return { "input_ids": encoding["input_ids"].squeeze(0), "attention_mask": encoding["attention_mask"].squeeze(0), "labels": torch.tensor(label_ids, dtype=torch.long) } class BERT_BiLSTM_CRF(nn.Module): """BERT + BiLSTM + CRF NER模型""" def __init__( self, num_labels: int, bert_model_name: str = "bert-base-chinese", lstm_hidden_dim: int = 256, dropout_rate: float = 0.1 ): """ 初始化模型 :param num_labels: 标签总数（含O标签） :param bert_model_name: 预训练BERT模型名称 :param lstm_hidden_dim: BiLSTM隐藏层维度 :param dropout_rate: Dropout比例 """ super().__init__() # BERT编码器（冻结下层以加速微调） self.bert = BertModel.from_pretrained(bert_model_name) bert_hidden = self.bert.config.hidden_size # 通常为768 # BiLSTM层 self.lstm = nn.LSTM( input_size=bert_hidden, hidden_size=lstm_hidden_dim, num_layers=2, batch_first=True, bidirectional=True, dropout=dropout_rate if 2 > 1 else 0 ) # 全连接层：将BiLSTM输出映射到标签空间 self.fc = nn.Linear(lstm_hidden_dim * 2, num_labels) self.dropout = nn.Dropout(dropout_rate) # CRF层：对标签序列进行全局约束 self.crf = CRF(num_labels, batch_first=True) def forward( self, input_ids: torch.Tensor, attention_mask: torch.Tensor, labels: Optional[torch.Tensor] = None ) -> Tuple[torch.Tensor, Optional[torch.Tensor]]: """ 前向传播 :param input_ids: token ID序列 [batch, seq_len] :param attention_mask: 注意力掩码 [batch, seq_len] :param labels: 真实标签（训练时提供，推理时为None） [batch, seq_len] :return: (损失, 预测标签序列) """ # BERT编码 bert_output = self.bert( input_ids=input_ids, attention_mask=attention_mask ) sequence_output = bert_output.last_hidden_state # [batch, seq_len, 768] # BiLSTM编码 lstm_output, _ = self.lstm(sequence_output) lstm_output = self.dropout(lstm_output) # 全连接投影 emissions = self.fc(lstm_output) # [batch, seq_len, num_labels] # 计算损失 loss = None if labels is not None: # CRF负对数似然损失 loss = -self.crf( emissions, labels, mask=attention_mask.bool(), reduction="mean" ) # 解码最优标签序列（Viterbi算法） with torch.no_grad(): predictions = self.crf.decode( emissions, mask=attention_mask.bool() ) return loss, predictions def predict(self, text: str, tokenizer) -> List[Dict]: """ 单条文本推理，返回识别的实体列表 :param text: 输入工单文本 :param tokenizer: BERT分词器 :return: [{entity, type, start, end}] 实体列表 """ self.eval() encoding = tokenizer( text, max_length=256, padding="max_length", truncation=True, return_tensors="pt" ) input_ids = encoding["input_ids"] attention_mask = encoding["attention_mask"] with torch.no_grad(): _, predictions = self.forward(input_ids, attention_mask) # 将预测标签转换回实体列表 entities = [] pred_labels = predictions[0] current_entity = None for i, label_id in enumerate(pred_labels): label = self.ID_TO_LABEL.get(label_id, "O") if label_id != -100 else "O" if label.startswith("B-"): # 保存上一个实体 if current_entity: entities.append(current_entity) # 开始新实体 current_entity = { "entity": tokenizer.decode([input_ids[0][i].item()]), "type": label[2:], "start": i, "end": i } elif label.startswith("I-") and current_entity: # 扩展当前实体 current_entity["entity"] += tokenizer.decode( [input_ids[0][i].item()] ) current_entity["end"] = i else: if current_entity: entities.append(current_entity) current_entity = None if current_entity: entities.append(current_entity) # 清理特殊token for e in entities: e["entity"] = e["entity"].replace("##", "") \ .replace("[CLS]", "") \ .replace("[SEP]", "").strip() return entities

关系抽取在NER之后执行。对已识别出实体的句子，使用依存句法分析（HanLP或LTP）找出实体间的依存弧。然后利用预定义的关系模式（如"nsubj→dobj"对应"依赖"关系、"nmod:range→case"对应"部署于"关系）进行匹配。对于模式无法覆盖的情况，使用远程监督（distant supervision）利用已有CMDB中的已知关系自动标注训练数据。

三、工单到CMDB的端到端抽取管道

sequenceDiagram participant Ticket as 工单系统 participant Pipeline as 抽取管道 participant NER as NER模型 participant RE as 关系抽取 participant CMDB as CMDB API Ticket->>Pipeline: 新工单/变更单webhook Pipeline->>Pipeline: 文本预处理+分词 Pipeline->>NER: 请求NER识别 NER-->>Pipeline: [{entity, type, position}] Pipeline->>Pipeline: 实体对配对 Pipeline->>RE: 请求关系抽取 RE-->>Pipeline: [{subject, relation, object, confidence}] alt confidence > 0.8 Pipeline->>CMDB: 自动写入依赖关系 else 0.5 < confidence <= 0.8 Pipeline->>CMDB: 标记为待确认，加入审核队列 else confidence <= 0.5 Pipeline->>Pipeline: 丢弃低置信度结果 end CMDB-->>Pipeline: 返回写入结果 Pipeline-->>Ticket: 更新工单状态

置信度分层是关键设计。高于0.8的抽取结果自动入库，0.5到0.8之间的标记为"待确认"进入人工审核队列，低于0.5的直接丢弃。这个阈值不是固定的——系统上线初期可以将自动入库阈值设高（如0.9），随着模型在反馈数据上的持续训练逐步下调。

实体对齐是另一个必须解决的工程问题。同一条数据库实例在工单中可能被写成order-db、order-db-prod、rds-order-master、订单主库。需要通过别名映射表加嵌入向量相似度来做归一化：

""" 配置项实体对齐与消歧模块 """ def entity_alignment( entity: str, entity_type: str, cmdb_index: dict, alias_map: dict, similarity_threshold: float = 0.85 ) -> Optional[str]: """ 将NER识别的实体对齐到CMDB中的标准CI标识 :param entity: NER识别出的实体名称 :param entity_type: 实体类型（SERVICE/DATABASE/HOST等） :param cmdb_index: CMDB中已有CI的索引 {standard_name: ci_object} :param alias_map: 别名映射 {alias: standard_name} :param similarity_threshold: 相似度阈值 :return: 匹配到的标准CI名称，未匹配返回None """ # 优先级1：精确匹配别名 normalized = entity.lower().replace("-", "").replace("_", "") for alias, standard_name in alias_map.items(): alias_norm = alias.lower().replace("-", "").replace("_", "") if alias_norm == normalized: return standard_name # 优先级2：精确匹配标准名 if entity in cmdb_index: return entity # 优先级3：模糊匹配（使用编辑距离或Jaccard相似度） best_match = None best_score = 0.0 for standard_name in cmdb_index: # 计算Jaccard相似度（基于字符级bigram） entity_bigrams = set( zip(entity, entity[1:]) ) if len(entity) >= 2 else set() name_bigrams = set( zip(standard_name, standard_name[1:]) ) if len(standard_name) >= 2 else set() if not entity_bigrams or not name_bigrams: continue intersection = entity_bigrams & name_bigrams union = entity_bigrams | name_bigrams jaccard_sim = len(intersection) / len(union) if union else 0.0 if jaccard_sim > best_score and jaccard_sim >= similarity_threshold: best_score = jaccard_sim best_match = standard_name return best_match

四、CMDB增量更新的一致性保障

flowchart TD A[抽取结果] --> B{关系是否已存在} B -->|已存在且一致| C[跳过，更新last_seen时间戳] B -->|已存在但冲突| D[记录冲突，通知CMDB owner] B -->|不存在| E{置信度判断} E -->|>0.8 自动写入| F[创建CI或关系] E -->|0.5-0.8 待审核| G[加入人工审核队列] E -->|<0.5| H[丢弃] F --> I[触发变更通知] G --> J[CMDB Owner审核] J -->|确认| F J -->|驳回| H

CMDB增量更新有两条铁律。第一，永不删除，只做软标记。即使NLP模型推断某条依赖关系已不存在，也只能标记为status=archived而非物理删除。第二，冲突必须上报。当抽取出的关系与现有关系矛盾时（如模型认为A依赖C，但CMDB已记录A依赖B且B和C是不同实体），系统不自行判断对错，而是创建冲突工单通知人工介入。