Python скрипт для генерирования JSON примеров для fine tuning локальной модели Qwen2.5 (7B) для анализа динамической маршрутизации - анализа результатов выполнения команды Cisco "show ip route" Ниже представлен полноценный Python-скрипт, который генерирует синтетический датасет в формате .jsonl (JSON Lines), готовый для fine-tuning моделей семейства Qwen (через LLaMA-Factory, Axolotl или Unsloth). Скрипт создает как здоровые сценарии (чтобы модель не искала ошибки там, где их нет), так и сценарии с инжектированными ошибками (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") Как использовать этот скрипт и развивать датасет: Запуск: Просто сохраните код в файл 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, указывающий на этот файл. https://chat.qwen.ai/s/t_d4f70f5e-e775-42ad-b59f-bbcd5324cac9?fev=0.2.64 (c) by Valery Shmelev (Deutsche: Valery Shmeleff) https://oflameron.com https://ai-fine-tuning.blogspot.com/