|
Разработка Датасетов
Head Hunter >>>>
Тюнинг Локальных ИИ
Локальные модели ИИ для изолированных сетей требуют специализированного дообучения (fine tuning). Для fine tuning нужны специализированное наборы данных (наборы примеров) в формате JSON - Data Sets.
Примеров нужно минимум 2000-3000 даже для компактных моделей ИИ. Подготовка дата сетов такого размера и хорошего качества довольно сложный процесс. Для многих задач большую часть дата сета можно сгенерировать с помощью специально подготовленных Python (например) скриптов.
Один из вариантов использования локального ИИ - анализ работы Ethernet-сети предприятия. В частности - анализ динамической маршрутизации. ИИ просматривает результат выполнения команды Cisco "show ip route" и ищет ошибки маршрутизации.
Для этого была выбрана модель ИИ, которую установили на HP-сервер с видеокартой. Для реальной работы ИИ его неообходимо дополнительно тренировать (fine tuning)
Для этого нужны обучающие наборы данных - Data Sets в формате JSON
Ниже представлен полноценный Python-скрипт, который генерирует синтетический датасет в формате .jsonl (JSON Lines), готовый для fine-tuning моделей семейства Qwen (через LLaMA-Factory, Axolotl или Unsloth).
Python скрипт создает как сценарии без ошибок маршрутизации (чтобы модель не искала ошибки там, где их нет), так и сценарии с некоторыми инжектированными ошибками (Null0 Blackhole, Missing Default Route, Unreachable Next-hop).
Python-скрипт для генерации датасета - generate_dataset.py
PYTHON
import json
import random
import ipaddress
from typing import List, Dict, Tuple
# ==========================================
# 1. Базовые шаблоны и пулы данных
# ==========================================
INTERNAL_SUBNETS = ["10.10.0.0/16", "10.20.0.0/16", "172.16.1.0/24", "192.168.10.0/24"]
WAN_SUBNETS = ["203.0.113.0/30", "198.51.100.0/30"]
PROTOCOLS = ["C", "L", "S", "O", "D", "B"]
def get_ip_from_subnet(subnet_str: str, is_gateway: bool = False) -> str:
"""Получает IP-адрес из подсети (первый или последний для шлюза)"""
net = ipaddress.IPv4Network(subnet_str)
if is_gateway:
return str(net.broadcast_address - 1)
return str(net.network_address + 1)
def generate_interface_brief(interfaces: List[Dict]) -> str:
"""Генерирует вывод show ip interface brief"""
lines = ["Interface IP-Address OK? Method Status Protocol"]
for intf in interfaces:
lines.append(f"{intf['name']:<22} {intf['ip']:<15} YES manual up up")
return "\n".join(lines)
def generate_route_table(routes: List[Dict]) -> str:
"""Генерирует вывод show ip route"""
header = (
"Codes: L - local, C - connected, S - static, R - RIP, M - mobile, B - BGP\n"
" D - EIGRP, EX - EIGRP external, O - OSPF, IA - OSPF inter area\n"
" N1 - OSPF NSSA external type 1, N2 - OSPF NSSA external type 2\n"
" O E2 - OSPF external type 2\n\n"
)
lines = [header]
for route in routes:
lines.append(route['line'])
return "\n".join(lines)
# ==========================================
# 2. Генераторы сценариев (Scenario Generators)
# ==========================================
def generate_healthy_edge_router() -> Dict:
"""Генерирует здоровый пограничный маршрутизатор"""
wan_subnet = random.choice(WAN_SUBNETS)
wan_ip = get_ip_from_subnet(wan_subnet, is_gateway=False)
wan_gw = get_ip_from_subnet(wan_subnet, is_gateway=True)
lan_subnet = random.choice(INTERNAL_SUBNETS)
lan_ip = get_ip_from_subnet(lan_subnet, is_gateway=False)
interfaces = [
{"name": "GigabitEthernet0/0", "ip": lan_ip},
{"name": "GigabitEthernet0/1", "ip": wan_ip}
]
routes = [
{"line": f"C {lan_subnet} is directly connected, GigabitEthernet0/0"},
{"line": f"L {lan_ip}/32 is directly connected, GigabitEthernet0/0"},
{"line": f"C {wan_subnet} is directly connected, GigabitEthernet0/1"},
{"line": f"L {wan_ip}/32 is directly connected, GigabitEthernet0/1"},
{"line": f"S* 0.0.0.0/0 [1/0] via {wan_gw}"},
{"line": f"O IA 10.0.0.0/8 [110/50] via {lan_ip}, 01:20:00, GigabitEthernet0/0"}
]
return {
"scenario": "healthy",
"interfaces": interfaces,
"routes": routes,
"expected_analysis": "Конфигурация маршрутизатора корректна. Присутствуют напрямую подключенные
сети, маршрут по умолчанию (0.0.0.0/0) указывает на корректный шлюз провайдера, и отсутствуют признаки
черных дыр или недостижимых next-hop адресов.",
"error_type": "None",
"severity": "Normal"
}
def generate_null0_blackhole() -> Dict:
"""Генерирует ошибку: суммарный маршрут в Null0 без конкретных подсетей"""
lan_subnet = "10.10.0.0/16"
lan_ip = get_ip_from_subnet(lan_subnet, is_gateway=False)
interfaces = [{"name": "GigabitEthernet0/0", "ip": lan_ip}]
# ОШИБКА: Есть суммарный маршрут, но НЕТ конкретных маршрутов внутри него
routes = [
{"line": f"C {lan_subnet} is directly connected, GigabitEthernet0/0"},
{"line": f"L {lan_ip}/32 is directly connected, GigabitEthernet0/0"},
{"line": f"O 10.0.0.0/8 is a summary, 00:15:23, Null0"}
]
return {
"scenario": "null0_blackhole",
"interfaces": interfaces,
"routes": routes,
"expected_analysis": f"В таблице маршрутизации присутствует суммарный маршрут '10.0.0.0/8' с
интерфейсом вывода Null0. Однако в таблице отсутствуют более специфичные маршруты (например, /16 или /24),
входящие в этот диапазон (кроме напрямую подключенной {lan_subnet}). Любой трафик, предназначенный для
других подсетей диапазона 10.x.x.x, будет немедленно отброшен интерфейсом Null0.",
"error_type": "Null0 Blackhole / Missing Specific Routes",
"severity": "High"
}
def generate_missing_default_route() -> Dict:
"""Генерирует ошибку: пограничный роутер без маршрута по умолчанию"""
wan_subnet = random.choice(WAN_SUBNETS)
wan_ip = get_ip_from_subnet(wan_subnet, is_gateway=False)
lan_subnet = random.choice(INTERNAL_SUBNETS)
lan_ip = get_ip_from_subnet(lan_subnet, is_gateway=False)
interfaces = [
{"name": "GigabitEthernet0/0", "ip": lan_ip},
{"name": "GigabitEthernet0/1", "ip": wan_ip} # Внешний интерфейс активен
]
# ОШИБКА: Отсутствует строка S* 0.0.0.0/0
routes = [
{"line": f"C {lan_subnet} is directly connected, GigabitEthernet0/0"},
{"line": f"L {lan_ip}/32 is directly connected, GigabitEthernet0/0"},
{"line": f"C {wan_subnet} is directly connected, GigabitEthernet0/1"},
{"line": f"L {wan_ip}/32 is directly connected, GigabitEthernet0/1"},
{"line": f"O IA 10.0.0.0/8 [110/50] via 10.10.0.2, 01:20:00, GigabitEthernet0/0"}
]
return {
"scenario": "missing_default_route",
"interfaces": interfaces,
"routes": routes,
"expected_analysis": f"Маршрутизатор имеет активный внешний интерфейс
GigabitEthernet0/1 с IP-адресом {wan_ip}, что указывает на его роль пограничного
устройства. Однако в таблице маршрутизации полностью отсутствует маршрут
по умолчанию (0.0.0.0/0). Без этого маршрута устройство не знает, куда отправлять
трафик во внешние сети, что приведет к потере внешней связности.",
"error_type": "Missing Default Route",
"severity": "Critical"
}
def generate_unreachable_nexthop() -> Dict:
"""Генерирует ошибку: статический маршрут с недостижимым Next-hop"""
lan_subnet = "172.16.1.0/24"
lan_ip = get_ip_from_subnet(lan_subnet, is_gateway=False)
# ОШИБКА: Next-hop 172.16.2.254, но сети 172.16.2.0/24 нет в таблице
bad_nexthop = "172.16.2.254"
dest_network = "10.10.0.0/16"
interfaces = [{"name": "GigabitEthernet0/0", "ip": lan_ip}]
routes = [
{"line": f"C {lan_subnet} is directly connected, GigabitEthernet0/0"},
{"line": f"L {lan_ip}/32 is directly connected, GigabitEthernet0/0"},
{"line": f"S {dest_network} [1/0] via {bad_nexthop}"}
]
return {
"scenario": "unreachable_nexthop",
"interfaces": interfaces,
"routes": routes,
"expected_analysis": f"В таблице маршрутизации присутствует статический
маршрут к сети {dest_network} с Next-hop адресом {bad_nexthop}. Однако на
данном маршрутизаторе нет ни одного интерфейса или маршрута, обеспечивающего
связность с сетью 172.16.2.0/24. Маршрутизатор не может отправить пакет, так как не
знает, как достичь указанного Next-hop (проблема рекурсивного поиска или опечатка в IP-адресе).",
"error_type": "Unreachable Next-hop / Invalid Static Route",
"severity": "High"
}
# ==========================================
# 3. Форматирование в ChatML / JSONL
# ==========================================
def format_to_chatml(data: Dict) -> Dict:
"""Форматирует сценарий в структуру messages для fine-tuning"""
prompt_text = (
f"Проанализируй следующие данные с маршрутизатора:\n\n"
f"# show ip interface brief\n{generate_interface_brief(data['interfaces'])}\n\n"
f"# show ip route\n{generate_route_table(data['routes'])}"
)
if data['scenario'] == 'healthy':
response_dict = {
"analysis": data['expected_analysis'],
"error_type": data['error_type'],
"severity": data['severity'],
"affected_prefixes": [],
"recommendation": "Действий не требуется. Конфигурация соответствует лучшим практикам."
}
else:
# Извлекаем префиксы из анализа для структурированного ответа
affected = []
if "10.0.0.0/8" in data['expected_analysis']: affected.append("10.0.0.0/8")
if "0.0.0.0/0" in data['expected_analysis']: affected.append("0.0.0.0/0")
if "10.10.0.0/16" in data['expected_analysis']: affected.append("10.10.0.0/16")
recommendation = "Проверьте конфигурацию маршрутизации и исправьте выявленную несогласованность."
if data['scenario'] == 'null0_blackhole':
recommendation = "Проверьте конфигурацию агрегации маршрутов. Убедитесь, что конкретные
подсети существуют и анонсируются, либо удалите команду суммаризации."
elif data['scenario'] == 'missing_default_route':
recommendation = "Добавьте статический маршрут по умолчанию (ip route 0.0.0.0 0.0.0.0 )
или проверьте работу динамического протокола."
elif data['scenario'] == 'unreachable_nexthop':
recommendation = "Проверьте правильность указания IP-адреса Next-hop в статическом маршруте
на предмет опечаток, либо добавьте маршрут до сети самого next-hop."
response_dict = {
"analysis": data['expected_analysis'],
"error_type": data['error_type'],
"severity": data['severity'],
"affected_prefixes": affected,
"recommendation": recommendation
}
return {
"messages": [
{
"role": "system",
"content": "Ты — эксперт по сетевой инженерии Cisco. Твоя задача - проанализировать вывод
команд 'show ip interface brief' и 'show ip route', найти конфигурационные ошибки или аномалии
и вернуть результат в строгом формате JSON."
},
{
"role": "user",
"content": prompt_text
},
{
"role": "assistant",
"content": json.dumps(response_dict, ensure_ascii=False)
}
]
}
# ==========================================
# 4. Главная функция генерации
# ==========================================
def generate_dataset(num_samples: int, output_file: str):
"""Генерирует датасет и сохраняет в JSONL"""
generators = [
(generate_healthy_edge_router, 0.30), # 30% здоровых данных
(generate_null0_blackhole, 0.25), # 25% Null0 Blackhole
(generate_missing_default_route, 0.25), # 25% Missing Default Route
(generate_unreachable_nexthop, 0.20) # 20% Unreachable Next-hop
]
with open(output_file, 'w', encoding='utf-8') as f:
for _ in range(num_samples):
# Выбор генератора на основе весов
r = random.random()
cumulative = 0
chosen_generator = generators[0][0]
for gen, weight in generators:
cumulative += weight
if r <= cumulative:
chosen_generator = gen
break
# Генерация и запись
sample_data = chosen_generator()
chatml_sample = format_to_chatml(sample_data)
# Запись в одну строку (стандарт JSONL)
f.write(json.dumps(chatml_sample, ensure_ascii=False) + '\n')
print(f" Датасет успешно сгенерирован: {num_samples} примеров сохранено в {output_file}")
if __name__ == "__main__":
# Генерируем 1000 примеров для старта (можно увеличить до 5000+)
generate_dataset(num_samples=1000, output_file="qwen_cisco_routing_dataset.jsonl")
AI Fine Tuning
Dataset Development
arrayphotocintrol 
Как использовать этот скрипт и развивать Data Set:
Запуск: Просто сохраните код в файл generate_dataset.py и запустите: python generate_dataset.py. На выходе вы получите файл qwen_cisco_routing_dataset.jsonl.
Проверка формата: Откройте полученный файл и проверьте несколько строк. Каждая строка должна быть валидным JSON-объектом с ключом "messages".
Масштабирование (Следующие шаги):
Добавьте больше вариаций: В функции generate_healthy_edge_router можно добавить рандомизацию количества интерфейсов (от 2 до 8), чтобы модель училась игнорировать "шум" в больших таблицах.
Добавьте новые ошибки: Напишите функции по аналогии для:
Routing Loop / Recursive Routing (Next-hop указывает на собственный IP роутера или на интерфейс туннеля, через который должен идти трафик).
Suboptimal ECMP (Два пути к одной сети, но один через GigabitEthernet, другой через медленный Serial/Tunnel).
Administrative Distance Conflict (Маршрут RIP перекрывает маршрут OSPF из-за ручной настройки distance).
Интеграция с LLaMA-Factory: Этот .jsonl файл полностью совместим с форматом alpaca или sharegpt, который используется в популярных фреймворках для fine-tuning (например, в hiyouga/LLaMA-Factory). Вам нужно будет только создать файл dataset_info.json, указывающий на этот файл.
AI Fine Tuning >>
Datasets >>
Датасеты >>
Разработка Датасетов для настройки ИИ
|
|
AI Fine Tuning Настройка (дообучение) локальных моделей ИИ. Разработка JSON примеров для обучения (data sets)