第320篇：LLM 辅助网络排障
关键词
LLM、大语言模型、网络排障、智能分析、日志分析、配置审查、根因定位、RAG
一、LLM 在网络排障中的应用
1.1 传统排障 vs LLM 辅助排障
传统排障方式：
┌─ 工程师 SSH 登录设备
├─ 逐条查看日志和状态
├─ 依赖个人经验和记忆
├─ 范围受限（只能查几台设备）
├─ 耗时：从分钟到小时
└─ 知识传承困难
LLM 辅助排障：
┌─ 收集多设备数据自动聚合
├─ 分析日志自动识别异常
├─ 基于知识库给出建议
├─ 生成排障报告
├─ 耗时：秒到分钟
└─ 知识可积累复用
LLM 在排障中的角色：
┌─ 不是替代工程师，而是"超级助手"
├─ 提供数据分析和模式识别
├─ 给出根因建议
└─ 最终决策由工程师确认
1.2 典型应用场景
LLM 辅助排障的十大场景：
1. 日志分析
┌─ 从海量日志中提取关键错误
├─ 识别日志模式（周期性错误/故障前兆）
└─ 关联多个设备的时间序列
2. 配置审查
┌─ 检查配置是否符合最佳实践
├─ 发现配置冲突（如 BGP/OSPF 参数不一致）
└─ 对比基线找差异
3. 故障根因分析
┌─ 基于告警和拓扑推断根因
├─ 给出排查路径
└─ 推荐修复方案
4. 命令建议
┌─ 根据问题描述推荐排障命令
├─ 解释命令输出
└─ 生成修复配置
5. 知识检索
┌─ 基于文档/案例检索解决方案
├─ 匹配相似历史故障
└─ 提供处理建议
二、RAG 知识库排障
2.1 RAG 架构
RAG（检索增强生成）排障架构：
| 用户问题：接口 GE0/0/1 flapping ↓ ┌────────────────────────────────────┐ └────────────────────────────────────┘ ↓ ┌────────────────────────────────────┐ └────────────────────────────────────┘ ↓ 回答：接口 flapping 可能原因列表 ┌─ 1. 检查光模块功率 ├─ 2. 查看 CRC 错误计数 ├─ 3. 检查对端接口状态 └─ 4. 建议更换光模块 | 向量检索（Embedding） ↓ 向量数据库 ┌──────────────────────────────┐ └──────────────────────────────┘ LLM 生成（GPT/文心/盘古） 基于检索到的相关知识生成回答 | 知识片段 1: interface flap 知识片段 2: CRC error check 知识片段 3: 光模块故障案例 | ||
|---|---|---|---|---|
2.2 知识库构建
#!/usr/bin/env python3
# llm_knowledge_base.py — 排障知识库构建
import json
import hashlib
from typing import List, Dict
class KnowledgeBase:
"""排障知识库"""
def __init__(self):
self.articles: List[Dict] = []
self.index: Dict[str, List[int]] = {}
def add_article(self, title: str, symptoms: List[str],
causes: List[str], solutions: List[str],
tags: List[str], source: str = ""):
"""添加知识文章"""
article = {
"id": hashlib.md5(title.encode()).hexdigest()[:8],
"title": title,
"symptoms": symptoms,
"causes": causes,
"solutions": solutions,
"tags": tags,
"source": source,
}
self.articles.append(article)
# 建立关键词索引
keywords = set()
for s in symptoms + tags + causes:
keywords.update(s.lower().split())
for kw in keywords:
if kw not in self.index:
self.index[kw] = []
self.index[kw].append(len(self.articles) - 1)
def search(self, query: str, top_k: int = 3) -> List[Dict]:
"""检索相关知识"""
query_words = set(query.lower().split())
# 简单关键词匹配检索
scores = {}
for word in query_words:
if word in self.index:
for idx in self.index[word]:
scores[idx] = scores.get(idx, 0) + 1
# 排序
ranked = sorted(scores.items(), key=lambda x: -x[1])
return [self.articles[idx] for idx, _ in ranked[:top_k]]
def to_dict(self) -> dict:
return {"articles": self.articles}
def save(self, path: str):
with open(path, "w", encoding="utf-8") as f:
json.dump(self.to_dict(), f, indent=2, ensure_ascii=False)
# ===== 构建排障知识库 =====
kb = KnowledgeBase()
# 添加常见故障知识
kb.add_article(
title="接口频繁 Flapping（震荡）",
symptoms=["interface flapping", "interface up/down",
"端口频繁UP DOWN", "端口震荡"],
causes=["光模块故障/老化", "光纤损坏/弯曲半径过小",
"对端设备接口故障", "配置协商不匹配"],
solutions=[
"display interface 检查 CRC 错误计数",
"display transceiver 检查光模块功率",
"检查两端双工/速率协商模式",
"更换光模块或光纤测试",
"如果 CRC 持续增长，建议更换光模块",
],
tags=["接口", "flapping", "光模块", "物理层"],
source="华为排障手册"
)
kb.add_article(
title="BGP 邻居不能建立",
symptoms=["BGP down", "bgp peer flapping",
"BGP 邻居无法建立", "BGP 状态 Idle",
"bgp active"],
causes=[
"TCP 连接不通（路由缺失/ACL 阻断）",
"AS 号配置错误",
"更新源地址不匹配",
"EBGP 多跳未配置",
"TTL 不足",
],
solutions=[
"ping 对端更新源地址验证连通性",
"display bgp peer 查看状态",
"检查 peer as-number 是否正确",
"检查 update-source 配置",
"EBGP 跨跳需配置 ebgp-max-hop",
],
tags=["BGP", "邻居", "路由", "TCP"],
source="华为BGP排障"
)
kb.add_article(
title="OSPF 邻居无法建立",
symptoms=["OSPF down", "ospf neighbor flapping",
"OSPF 邻居无法建立", "OSPF INIT/EXSTART",
"邻居卡在 INIT"],
causes=[
"Hello/Dead 间隔不一致",
"区域 ID 不匹配",
"网络类型不匹配",
"MTU 不匹配",
"认证配置不一致",
"二层链路故障",
],
solutions=[
"display ospf peer 查看邻居状态",
"检查两端 ospf timer hello/dead",
"检查 area 配置",
"检查 network-type (P2P/Broadcast)",
"检查 ip mtu 是否一致",
"检查 authentication-mode",
],
tags=["OSPF", "邻居", "IGP", "路由协议"],
source="华为OSPF排障"
)
kb.save("knowledge_base.json")
print(f"知识库已构建: {len(kb.articles)} 篇文章")
# 测试检索
query = "接口一直没有BGP邻居"
print(f"\n查询: {query}")
results = kb.search(query)
for r in results:
print(f"\n标题: {r['title']}")
print(f"可能原因: {', '.join(r['causes'][:3])}")
print(f"解决方案: {', '.join(r['solutions'][:3])}")
2.3 LLM 排障助手
#!/usr/bin/env python3
# llm_troubleshooter.py — LLM 排障助手
import json
from typing import Optional
class TroubleshootEngine:
"""排障引擎"""
def __init__(self, knowledge_path: str = "knowledge_base.json"):
with open(knowledge_path, "r", encoding="utf-8") as f:
self.kb = json.load(f)
def analyze_device_data(self, device_name: str,
logs: str, config: str,
commands: dict) -> str:
"""分析设备数据生成排障建议"""
findings = []
recommendations = []
# 1. 日志分析
error_patterns = {
"CRC": "接口 CRC 错误，可能光纤/光模块问题",
"flapping": "接口震荡，检查物理链路",
"down": "接口 DOWN，检查对端和物理连接",
"error": "出现错误日志",
"BGP.*Down": "BGP 邻居中断",
"OSPF.*Down": "OSPF 邻居中断",
"authenticat": "认证失败",
"timeout": "连接超时",
}
for line in logs.split("\n"):
for pattern, hint in error_patterns.items():
import re
if re.search(pattern, line, re.IGNORECASE):
findings.append(f"- {line.strip()[:80]}")
recommendations.append(f" → {hint}")
break
# 2. 接口状态分析
if "display interface brief" in commands:
intf_output = commands["display interface brief"]
down_interfaces = []
for line in intf_output.split("\n"):
if "down" in line.lower() and "GigabitEthernet" in line:
down_interfaces.append(line.split()[0])
if down_interfaces:
findings.append(f"DOWN 接口: {', '.join(down_interfaces)}")
recommendations.append("检查 DOWN 接口的物理连接和配置")
# 3. 配置分析
if "display current-configuration" in commands:
cfg = commands["display current-configuration"]
# 检查常见配置问题
if "ntp-service" not in cfg:
findings.append("未配置 NTP，可能导致时间不同步问题")
# 4. 生成报告
report = [
f"## {device_name} 排障分析",
f"### 发现的问题",
]
report.extend(findings if findings else ["未发现明显异常"])
report.append("\n### 建议措施")
report.extend(recommendations if recommendations else [
"1. 检查设备基本状态（CPU/内存/温度）",
"2. 确认链路物理状态",
"3. 查看系统日志寻找更多线索",
])
return "\n".join(report)
def suggest_commands(self, problem: str) -> list:
"""根据问题推荐排查命令"""
cmd_map = {
"接口": [
("接口状态", "display interface brief"),
("接口统计", "display interface GigabitEthernet0/0/1"),
("光模块信息", "display transceiver interface GigabitEthernet0/0/1"),
("接口 CRC", "display interface GigabitEthernet0/0/1 | include CRC"),
],
"BGP": [
("BGP 邻居摘要", "display bgp peer"),
("BGP 邻居详情", "display bgp peer 10.0.0.1 verbose"),
("BGP 路由", "display bgp routing-table"),
("BGP 统计", "display bgp statistics"),
],
"OSPF": [
("OSPF 邻居", "display ospf peer"),
("OSPF 接口", "display ospf interface"),
("OSPF LSDB", "display ospf lsdb"),
("OSPF 路由", "display ospf routing"),
],
"路由": [
("路由表", "display ip routing-table"),
("路由协议", "display ip routing-table protocol"),
("Fib 表", "display fib"),
],
"设备": [
("设备版本", "display version"),
("CPU 使用率", "display cpu-usage"),
("内存使用率", "display memory-usage"),
("环境状态", "display environment"),
("系统日志", "display logbuffer"),
],
}
suggestions = []
for keyword, cmds in cmd_map.items():
if keyword.lower() in problem.lower():
suggestions.extend(cmds)
if not suggestions:
for cmds in cmd_map.values():
suggestions.extend(cmds)
return suggestions[:8] # 最多返回 8 条
# 使用示例
if __name__ == "__main__":
engine = TroubleshootEngine()
# 模拟数据
logs = """
Jan 15 10:30:22 CORE-SW01 %%01IFNET/4/LINK_STATE(l)[0]:The line protocol
IP on the interface GigabitEthernet0/0/1 has entered the DOWN state.
Jan 15 10:30:25 CORE-SW01 %%01IFNET/4/LINK_STATE(l)[1]:The line protocol
IP on the interface GigabitEthernet0/0/1 has entered the UP state.
Jan 15 10:30:28 CORE-SW01 %%01IFNET/4/LINK_STATE(l)[2]:The line protocol
IP on the interface GigabitEthernet0/0/1 has entered the DOWN state.
"""
config = """
sysname CORE-SW01
interface GigabitEthernet0/0/1
description UPLINK
port link-type trunk
port trunk allow-pass vlan all
"""
commands = {
"display interface brief": "GigabitEthernet0/0/1 down",
}
report = engine.analyze_device_data("CORE-SW01", logs, config, commands)
print(report)
print("\n=== 推荐排查命令 ===")
problem = "接口 flapping，怀疑光模块问题"
for name, cmd in engine.suggest_commands(problem):
print(f" {name}: {cmd}")
三、LLM 配置审查
3.1 配置评审
#!/usr/bin/env python3
# llm_config_review.py — LLM 配置审查
class ConfigReview:
"""配置审查引擎"""
RULES = [
{
"name": "SSH 版本",
"check": lambda c: "ssh version 2" in c or "ssh server port" in c,
"severity": "error",
"message": "SSH v2 未配置，存在安全风险",
},
{
"name": "密码加密",
"check": lambda c: "cipher" in c or "hash" in c,
"severity": "warning",
"message": "密码未使用加密存储",
},
{
"name": "NTP 配置",
"check": lambda c: "ntp-service" in c or "ntp server" in c,
"severity": "warning",
"message": "未配置 NTP 服务器，时间可能不准确",
},
{
"name": "SNMP v3",
"check": lambda c: "snmp-agent" in c,
"severity": "info",
"message": "建议配置 SNMP v3 增强安全性",
},
{
"name": "空闲超时",
"check": lambda c: "idle-timeout" in c or "exec-timeout" in c,
"severity": "warning",
"message": "未配置空闲超时，SSH 会话可能长期挂起",
},
]
def review(self, device_name: str, config: str) -> list:
"""执行审查"""
findings = []
for rule in self.RULES:
passed = rule["check"](config)
if not passed:
findings.append({
"device": device_name,
"rule": rule["name"],
"severity": rule["severity"],
"message": rule["message"],
})
return findings
def generate_report(self, device_name: str, config: str) -> str:
"""生成审查报告"""
findings = self.review(device_name, config)
lines = [f"## 配置审查: {device_name}"]
lines.append(f"检查规则: {len(self.RULES)} 条")
errors = [f for f in findings if f["severity"] == "error"]
warnings = [f for f in findings if f["severity"] == "warning"]
infos = [f for f in findings if f["severity"] == "info"]
lines.append(f"错误: {len(errors)}, 警告: {len(warnings)}, 提示: {len(infos)}")
if findings:
lines.append("\n### 发现的问题")
for f in findings:
icon = {"error": "❌", "warning": "⚠", "info": "ℹ"}
lines.append(f"{icon[f['severity']]} {f['message']}")
else:
lines.append("\n✓ 所有检查通过")
return "\n".join(lines)
# 示例
review = ConfigReview()
sample_config = """
sysname CORE-SW01
interface GigabitEthernet0/0/1
description UPLINK
port link-type trunk
port trunk allow-pass vlan all
"""
print(review.generate_report("CORE-SW01", sample_config))
四、RAG 增强排障流程
4.1 完整排障流程
#!/usr/bin/env python3
# rag_troubleshoot_flow.py — RAG 增强排障
"""
RAG 排障流程：
1. 用户描述问题
2. 自动收集相关设备数据
3. RAG 检索相似案例
4. LLM 分析生成诊断
5. 给出排查建议
"""
class RAGTroubleshooter:
"""RAG 增强排障器"""
def __init__(self, knowledge_base: str = "knowledge_base.json"):
self.kb = TroubleshootEngine(knowledge_base)
def diagnose(self, device: str, problem: str,
auto_collect: bool = True) -> dict:
"""诊断流程"""
result = {
"device": device,
"problem": problem,
"steps": [],
}
# Step 1: 推荐排查命令
result["steps"].append({
"step": 1,
"action": "推荐排查命令",
"commands": self.kb.suggest_commands(problem),
})
# Step 2: 检索知识库
result["steps"].append({
"step": 2,
"action": "检索知识库",
"articles": [a["title"] for a in self.kb.kb.get("articles", [])
if any(k in problem for k in a.get("tags", []))],
})
# Step 3: 生成排查计划
result["steps"].append({
"step": 3,
"action": "排查计划",
"plan": [
"1. 检查物理层：接口状态、光模块功率",
"2. 检查链路层：CRC 错误、协商状态",
"3. 检查网络层：IP 连通性、路由表",
"4. 检查协议层：BGP/OSPF 邻居状态",
"5. 综合判断定位根因",
],
})
# Step 4: 预防建议
result["steps"].append({
"step": 4,
"action": "预防建议",
"suggestions": [
"配置接口联动检测（BFD）",
"启用 gRPC Telemetry 监控接口状态",
"配置日志告警推送",
"定期检查光模块健康状态",
],
})
return result
def print_diagnosis(self, result: dict):
"""打印诊断结果"""
print(f"=== 排障诊断报告 ===")
print(f"设备: {result['device']}")
print(f"问题: {result['problem']}\n")
for step in result["steps"]:
print(f"Step {step['step']}: {step['action']}")
if "commands" in step:
for name, cmd in step["commands"]:
print(f" {name}: {cmd}")
if "plan" in step:
for p in step["plan"]:
print(f" {p}")
if "suggestions" in step:
for s in step["suggestions"]:
print(f" • {s}")
print()
if __name__ == "__main__":
troubleshooter = RAGTroubleshooter()
result = troubleshooter.diagnose(
device="CORE-SW01",
problem="接口GE0/0/1频繁flapping并伴随CRC错误",
)
troubleshooter.print_diagnosis(result)
五、最佳实践
LLM 辅助排障最佳实践：
1. 数据质量
┌─ 确保输入数据的准确性和完整性
├─ 过滤敏感信息
├─ 标准化数据格式
└─ 标注数据来源和时效
2. 知识库维护
┌─ 持续更新排障案例
├─ 验证 LLM 输出的准确性
├─ 标注置信度等级
└─ 定期清理过期知识
3. 人机协作
┌─ LLM 提供建议，工程师决策
├─ 关键操作需要人工确认
├─ 反馈机制改进模型
└─ 保留人工干预通道
4. 安全
┌─ 不向 LLM 发送明文密码
├─ 控制 API 调用频率
├─ 审计所有 AI 辅助操作
└─ 敏感数据本地处理
下篇预告：第321篇 — 网络知识图谱与 AI Ops，将介绍如何构建网络知识图谱实现智能运维，以及 AI Ops 的实践路径。