Analysis Agent¶
Analysis Agent 是纯计算引擎,用于路由分析、What-If 模拟和拓扑可视化。完全在隔离沙箱中运行 — 无网络访问,无设备连接。
功能声明
| ID | 声明 | 状态 |
|---|---|---|
| C-NE-22 | What-If 模拟:禁用链路后重新计算连通性和影响范围 | 🔶 需 LLM |
| C-NE-23 | Analysis Agent 仅使用 run_python_simulation,不访问实时设备 |
⬜ 待验证 |
单工具,全能力¶
Analysis 只有一个工具:run_python_simulation。沙箱内 agent 编写 Python 代码,可访问:
| 全局变量 | 类型 | 用途 |
|---|---|---|
db |
DatabaseProxy | db.query(sql) — 只读访问生产快照数据 |
sim |
SimulationProxy | sim.clone(tables) + sim.execute(sql) — 内存可写副本 |
nx |
networkx | 图引擎:DiGraph、shortest_path、connected_components |
netutils |
netutils | IP 计算:子网重叠、接口规范化、前缀计算 |
json, math, itertools, collections |
stdlib | 标准库 |
沙箱强制 network_isolation=True — 无 HTTP 调用,无 socket 连接。
使用场景¶
路由分析¶
"R1 上哪些 BGP 邻居未建立?"
Agent 通过 db.query() 查询 v_bgp_neighbors_auto 并返回过滤结果。
What-If 模拟¶
"如果 R2 丢失所有链路会怎样?"
Agent 执行:
- 克隆拓扑数据:
sim.clone(['topology_links', 'ospf_neighbors', 'bgp_neighbors']) - 变异:
sim.execute("UPDATE sim_topology_links SET link_status = 'down' WHERE source_device = 'R2'") - 从变异状态构建 networkx 图
- 通过
nx.connected_components()计算影响范围 - 返回受影响设备、丢失路径和协议影响
路径分析¶
"R1 到 R4 的最短 OSPF 路径是什么?"
Agent 从 OSPF cost 构建加权图,运行 nx.shortest_path() 计算。
拓扑可视化¶
"画出网络拓扑"
Agent 查询 netops.topology_links,构建 networkx 图,渲染 Mermaid 图表(含设备状态和接口标签)。
模拟工作流¶
每次模拟遵循此模式:
# 第 1 步 — 检查数据覆盖
coverage = db.query("""
SELECT 'topology_links' tbl, count(*) rows FROM netops.topology_links
UNION ALL SELECT 'bgp_neighbors', count(*) FROM v_bgp_neighbors_auto
""")
# 第 2 步 — 克隆所需表
sim.clone(['topology_links', 'ospf_neighbors', 'routes', 'bgp_neighbors'])
# 第 3 步 — What-If 变异
sim.execute("UPDATE sim_topology_links SET link_status = 'down' WHERE source_device = ?", ['R2'])
# 第 4 步 — 构建 networkx 图
links = sim.execute("SELECT source_device, destination_device, link_status FROM sim_topology_links").fetchall()
G = nx.DiGraph()
for src, dst, status in links:
G.add_edge(src, dst, link_status=status)
# 第 5 步 — 仅活跃链路子图分析
active = [(u, v) for u, v, d in G.edges(data=True) if d['link_status'] == 'active']
G_active = G.edge_subgraph(active).copy()
# 第 6 步 — 返回结果
_result = {
"blast_radius": [list(c) for c in nx.connected_components(G_active.to_undirected())],
"isolated_devices": [n for n in G.nodes() if n not in G_active],
}
Agent 始终输出 Markdown 报告,包含:当前状态、模拟范围、影响分析、变更计划、验证命令和风险分级。