安全分析技能路由器：用Python工程化约束大模型

之前在安全分析场景里接大模型时，我最大的感受是：AI 确实能看懂日志、能提建议，但很难稳定地按照一套标准流程走完整个安全扫描任务。同一个日志，换个时间问、换个模型问，过程和结论可能完全不同；换个人来调，可能又是一个新玩法。团队里每个人都在“调 AI”，但没有一套大家可以共用的、可复现、可审计的分析流水线。

后来我们逐步把“安全分析能力”从大模型的自由对话里剥离出来，做成一个个标准化的“技能”，再通过一个类似路由器的中间层，根据任务类型自动决定调用哪个技能、按什么顺序执行，最终输出结构化的扫描结果。这套设计，就是我们常说的“安全分析技能路由器”。

本文会围绕这个主题，拆解“技能路由器”是什么、解决什么问题、如何用 Python 实现一个最小可运行的版本，并给出一个基于 Nginx 日志分析的完整实战示例，帮助大家理解如何用工程化的方式约束 AI 的安全分析行为。

1. 背景与核心概念

1.1 为什么 AI 做安全分析容易“失控”

大模型接入安全分析流程，常见的做法是直接把日志丢给模型，让它“分析一下有没有风险”。这种方式在探索阶段非常高效，但一旦进入正式工作流，就会出现几个问题：

第一，输出不稳定。同一份日志，问两次可能得到两份侧重点完全不同的报告。安全分析要求的是可复现性，如果结论无法稳定复现，后续审计和处置就没法落地。

第二，流程不可控。大模型不会主动记住“先做日志清洗，再做异常统计，然后查询威胁情报，最后输出报告”这件事。用户少说一句，它就可能跳过某个关键步骤。

第三，工具能力不可复用。今天让 AI 查一个 IP 的威胁情报，明天换一个场景又要重新写提示词。查询逻辑、数据源、超时策略全部散落在对话里，无法沉淀成团队资产。

第四，安全责任边界模糊。安全问题不是聊天，扫描动作必须经过授权，结果必须有据可查。如果全靠大模型自由发挥，出了问题很难回溯。

1.2 什么是“技能路由器”

“技能路由器”这个概念借鉴了网络设备中的路由器思想：数据包进入路由器后，根据路由表决定下一跳走向。在安全分析场景中，我们把“任务”看作数据包，把“安全分析能力”看作技能，路由器则根据任务特征自动匹配一个或多个技能，并编排执行顺序。

更具体地说，一个技能是“完成一个安全分析步骤的最小可执行单元”，它应当具备：

• 明确的输入：比如日志文本、IP 地址、域名、配置片段。
• 明确的功能：比如日志异常检测、端口指纹识别、CVE 信息查询、报告生成。
• 标准化的输出：比如统一的数据结构，包含严重级别、描述、来源。
• 可独立测试：每个技能都能单独验证正确性。

路由器不对业务逻辑负责，它只做两件事：识别任务意图，决定技能执行链。这种设计让大模型从“执行者”变成“调度者”，从“凭感觉分析”变成“按标准流程走”，既保留了 AI 的理解能力，又限制了它的不可控性。

1.3 安全分析的典型应用场景

技能路由器的应用范围很广，安全团队在日常工作中遇到的重复性分析场景，基本都可以通过这个思路标准化：

2. 架构设计与核心模块

2.1 三层架构

一个完整的安全分析技能路由器，我习惯分成三层来看：

接入与任务理解层：接收任务描述和原始输入，负责理解用户想干什么。这一层可以是大模型，也可以是规则引擎，甚至可以是两者结合。

路由与编排层：这是整个系统的心脏。路由根据任务特征匹配候选技能，编排则确定技能的执行顺序，并传递上下文。

技能执行层：底层是一个个确定性的安全分析技能。它们不依赖大模型的“临场发挥”，执行逻辑固定、结果可复现。

这种分层的好处是：上层可以随时替换不同的大模型，下层可以持续沉淀新的安全分析技能，中间层保持稳定，团队协作边界清晰。

2.2 核心模块职责

技能路由器主要由四个核心模块组成：

• 技能注册中心（SkillRegistry）：统一管理所有技能实例，支持注册、查询、去重和校验。
• 任务上下文（ScanContext）：贯穿整个扫描任务的数据对象，保存任务描述、原始输入、中间参数、技能结果和元数据。
• 路由器（SkillRouter）：基于规则或大模型，将任务映射到技能列表。
• 编排器（ScanOrchestrator）：按技能列表顺序执行，把每个技能的产出追加进上下文，最终形成完整扫描结果。

2.3 标准化扫描流程设计

标准化的核心在于把流程节点固定下来。本文采用的是一个简洁的五步流程：

输入规范化 -> 技能选择 -> 技能链执行 -> 结果汇总 -> 报告输出

输入规范化统一处理日志文本、参数、来源信息，避免不同技能对输入格式有额外要求。技能选择由路由器完成，技能链执行由编排器完成。结果汇总是把所有技能的SkillResult合并，最终报告输出可以是一个自动生成的 Markdown 文档，也可以是 JSON 结构化数据。

这套流程的好处是：任何一个环节出现问题，都能快速定位；任何一个新技能加入，只需要注册到注册中心，不需要改动其他模块。

3. 环境准备与项目结构

3.1 技术选型与版本说明

本文示例使用 Python 实现，重点演示设计思路。版本方面不绑定某一个固定版本，实际开发请以你的项目环境为准。建议使用 Python 3.10 及以上版本，以便完整使用类型注解和dataclass特性。

依赖方面，示例代码只会用到 Python 标准库，不需要额外安装第三方包。如果你要接入大模型做智能路由，需要准备一个 OpenAI 兼容接口的 API Key，具体库和版本要看你的模型供应商。

3.2 项目结构

security_skill_router/ ├── main.py ├── models.py ├── skill_base.py ├── registry.py ├── router.py ├── orchestrator.py ├── llm_client.py └── skills/ ├── __init__.py ├── log_scan_skill.py ├── threat_lookup_skill.py └── report_skill.py

每个文件的职责如下：

• models.py：数据模型定义。
• skill_base.py：技能抽象基类。
• registry.py：技能注册中心。
• router.py：任务路由与技能选择。
• orchestrator.py：扫描编排器。
• llm_client.py：大模型客户端抽象。
• skills/：各类安全分析技能的具体实现。

4. 核心代码实现

4.1 定义数据模型

数据模型是整个系统的“通用语言”。如果每个技能各自定义一套输出格式，路由器和报告模块就没法统一处理。

# models.py from dataclasses import dataclass, field from typing import List, Optional from enum import Enum class SkillStatus(str, Enum): """技能执行状态""" PENDING = "pending" RUNNING = "running" SUCCESS = "success" FAILED = "failed" @dataclass class Finding: """一条具体的分析发现""" severity: str title: str detail: str source: str = "" @dataclass class SkillResult: """单个技能的执行结果""" skill_name: str status: SkillStatus = SkillStatus.PENDING findings: List[Finding] = field(default_factory=list) summary: str = "" error: str = "" @dataclass class ScanContext: """一次安全扫描任务的完整上下文""" task: str input_data: Optional[str] = None params: dict = field(default_factory=dict) results: List[SkillResult] = field(default_factory=list) metadata: dict = field(default_factory=dict) def add_result(self, result: SkillResult) -> None: """追加一个技能的执行结果""" self.results.append(result)

Finding是统一的发现结构，severity用简单字符串表示高中低等级，方便后续做过滤排序。ScanContext贯穿整个任务，技能之间不直接通信，所有中间数据都放在上下文里，这样每个技能都是无状态的，也方便审计回溯。

4.2 技能基类与注册中心

技能基类规定了一个技能必须提供哪些信息、必须实现什么方法：

# skill_base.py from abc import ABC, abstractmethod from models import ScanContext, SkillResult class BaseSkill(ABC): """所有安全分析技能的抽象基类""" name: str = "base" description: str = "" @abstractmethod def run(self, context: ScanContext) -> SkillResult: """执行技能，接收扫描上下文，返回技能结果""" raise NotImplementedError

有了这个基类，新增技能只需要继承BaseSkill，实现run方法即可。注册中心负责把这些技能统一管理起来：

# registry.py from typing import List from skill_base import BaseSkill class SkillRegistry: """技能注册中心：负责技能注册、查询和去重""" def __init__(self): self._skills = {} def register(self, skill: BaseSkill) -> None: if skill.name in self._skills: raise ValueError(f"技能重复注册: {skill.name}") self._skills[skill.name] = skill def get(self, name: str) -> BaseSkill: return self._skills.get(name) def names(self) -> List[str]: return list(self._skills.keys())

注册中心看起来很简单，但它解决了两个问题：一是所有技能集中可见，团队可以快速了解系统已具备哪些能力；二是可以拦截重复注册，避免同名技能互相覆盖。

4.3 实现路由器

路由器是“技能路由器”这个名字的直接体现。它接收任务描述，输出一个有序的技能名称列表。为了保证基础可用性，建议先用规则路由做兜底，再尝试用大模型增强路由。

# router.py from models import ScanContext from registry import SkillRegistry class SkillRouter: """任务分诊与技能路由器""" def __init__(self, registry: SkillRegistry, llm_client=None, rules: dict = None): self.registry = registry self.llm_client = llm_client self.rules = rules or {} def route(self, context: ScanContext) -> list: # 第一优先：规则路由，保证基础可用性 skill_names = self._route_by_rules(context.task) # 第二优先：大模型增强路由 if self.llm_client: try: llm_names = self._route_by_llm(context.task) if llm_names: skill_names = llm_names except Exception as exc: print(f"[router] LLM 路由失败，使用规则兜底: {exc}") return skill_names def _route_by_rules(self, task: str) -> list: task_lower = task.lower() matched = [] for skill_name, keywords in self.rules.items(): if any(kw in task_lower for kw in keywords): matched.append(skill_name) if not matched: # 默认技能，保证任何任务都有可执行的流程 matched = ["log_scan"] return matched def _route_by_llm(self, task: str) -> list: available_names = self.registry.names() prompt = f"""你是一个安全分析任务调度器，请从技能列表中选择最合适的技能，并按执行顺序返回 JSON 数组。 技能列表: {available_names} 任务描述: {task} 输出要求: 只返回 JSON 数组，例如 ["log_scan", "report"]。 """ response = self.llm_client.chat(prompt) # 这里需要根据实际返回格式做健壮性解析 import json try: names = json.loads(response) if isinstance(names, list): return [n for n in names if n in available_names] except json.JSONDecodeError: return [] return []

规则路由的可维护性很高，适合关键词明确的任务；大模型路由则能处理更自然的描述。两者结合，即使大模型接口暂时不可用，系统仍然能按默认流程运行。

4.4 实现流程编排器

编排器负责把路由器选出的技能链串起来执行，并把结果逐步放入上下文：

# orchestrator.py from models import ScanContext from registry import SkillRegistry from router import SkillRouter class ScanOrchestrator: """扫描编排器：将任务拆解为技能链并依次执行""" def __init__(self, registry: SkillRegistry, router: SkillRouter): self.registry = registry self.router = router def run(self, task: str, input_data: str = None, params: dict = None) -> ScanContext: context = ScanContext(task=task, input_data=input_data, params=params or {}) skill_names = self.router.route(context) context.metadata["skill_chain"] = skill_names print(f"[orchestrator] 技能链: {skill_names}") for skill_name in skill_names: skill = self.registry.get(skill_name) if not skill: print(f"[orchestrator] 技能不存在，跳过: {skill_name}") continue print(f"[orchestrator] 执行技能: {skill_name}") result = skill.run(context) context.add_result(result) if result.error: print(f"[orchestrator] 技能 {skill_name} 失败: {result.error}") return context

到这里，一个最小可运行的框架已经成型。接下来用 Nginx 日志分析场景做一个完整实战，让整个流程跑起来。

5. 完整实战案例：Nginx 日志安全分析

5.1 场景设定

假设安全团队收到一份 Nginx 访问日志，需要完成三个任务：

1. 检测日志中的异常访问行为。
2. 对可疑 IP 进行威胁情报查询。
3. 生成标准化的安全分析报告。

我们以一小段示例日志作为输入：

192.168.1.1 - - [20/Feb/2025:10:12:33 +0800] "GET /admin/login HTTP/1.1" 200 5423 "-" "Mozilla/5.0" 10.0.0.5 - - [20/Feb/2025:10:12:35 +0800] "GET /api/v1/order HTTP/1.1" 200 1200 "-" "curl/7.68.0" 192.168.1.1 - - [20/Feb/2025:10:12:36 +0800] "POST /login HTTP/1.1" 302 300 "-" "python-requests/2.25.1" 10.0.0.5 - - [20/Feb/2025:10:12:40 +0800] "GET /api/v1/user/list HTTP/1.1" 200 4310 "-" "curl/7.68.0" 192.168.1.1 - - [20/Feb/2025:10:12:41 +0800] "GET /admin/config HTTP/1.1" 403 200 "-" "Mozilla/5.0" 10.0.0.5 - - [20/Feb/2025:10:12:45 +0800] "GET /api/v1/user/list HTTP/1.1" 200 4310 "-" "curl/7.68.0"

场景中的 IP 均为示例数据，不要在生产环境直接照搬。

5.2 实现日志扫描技能

日志扫描技能负责解析每一行日志，提取 IP、请求方法和状态码，并应用简单的异常规则：

# skills/log_scan_skill.py import re from collections import Counter from models import Finding, SkillResult, SkillStatus from skill_base import BaseSkill class LogScanSkill(BaseSkill): name = "log_scan" description = "对访问日志进行异常检测与统计分析" def run(self, context) -> SkillResult: result = SkillResult(skill_name=self.name) try: lines = (context.input_data or "").strip().splitlines() if not lines: result.status = SkillStatus.SUCCESS result.summary = "日志为空，未发现异常。" return result anomalies = [] ip_counter = Counter() for line in lines: match = re.search(r'^(\S+).*?"(\S+) (\S+) [^"]*" (\d{3})', line) if not match: continue ip, method, path = match.group(1), match.group(2), match.group(3) status = int(match.group(4)) ip_counter[ip] += 1 if status >= 500: anomalies.append(f"服务端异常 {status}，IP={ip}，请求={method} {path}") if status == 403 and ("/admin" in path or "/config" in path): anomalies.append(f"越权访问尝试，IP={ip}，请求={method} {path}") for ip, count in ip_counter.items(): if count > 3: anomalies.append(f"高频访问 IP={ip}，次数={count}，可能存在扫描行为") result.status = SkillStatus.SUCCESS result.summary = f"共解析 {len(lines)} 条日志，发现 {len(anomalies)} 条异常。" for item in anomalies[:10]: result.findings.append(Finding( severity="medium", title="日志异常", detail=item, source=self.name )) except Exception as exc: result.status = SkillStatus.FAILED result.error = str(exc) return result

这个技能把异常规则写死在代码里，保证每次执行结果一致，不会因为“大模型心情不好”而漏判。

5.3 实现威胁情报查询技能

威胁情报查询技能从上下文中提取可疑 IP，并查询情报数据。这里的模拟数据只是为了演示流程，生产环境请替换为真实威胁情报 API：

# skills/threat_lookup_skill.py import re from models import Finding, SkillResult, SkillStatus from skill_base import BaseSkill class ThreatLookupSkill(BaseSkill): name = "threat_lookup" description = "查询 IP 维度的威胁情报（演示用模拟数据）" def run(self, context) -> SkillResult: result = SkillResult(skill_name=self.name) ip_to_check = context.params.get("ip") if not ip_to_check: match = re.search(r'\b(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\b', context.input_data or "") ip_to_check = match.group(1) if match else None if not ip_to_check: result.status = SkillStatus.SUCCESS result.summary = "未提取到需要查询的 IP。" return result mock_db = { "192.168.1.1": {"score": 80, "tag": "扫描器", "desc": "历史存在主动扫描行为"}, "10.0.0.5": {"score": 20, "tag": "正常", "desc": "暂无负面情报"}, } info = mock_db.get(ip_to_check, {"score": 0, "tag": "未知", "desc": "暂无情报"}) if info["score"] >= 50: result.findings.append(Finding( severity="high", title=f"IP 威胁情报命中: {ip_to_check}", detail=f"情报标签={info['tag']}，风险分={info['score']}，说明={info['desc']}", source=self.name )) result.status = SkillStatus.SUCCESS result.summary = f"IP {ip_to_check} 查询完成，风险分 {info['score']}" return result

实际项目中，威胁情报查询应该加上缓存、超时和限流策略，避免拖慢整个扫描流程。

5.4 实现报告生成技能

报告生成技能不负责分析，它只汇总前面技能的结果，输出一份结构化的报告：

# skills/report_skill.py from datetime import datetime from models import SkillResult, SkillStatus from skill_base import BaseSkill class ReportSkill(BaseSkill): name = "report" description = "汇总所有技能结果，生成标准化的安全分析报告" def run(self, context) -> SkillResult: result = SkillResult(skill_name=self.name) lines = [] lines.append("# 安全分析报告") lines.append(f"- 任务: {context.task}") lines.append(f"- 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") lines.append(f"- 技能链: {' -> '.join(context.metadata.get('skill_chain', []))}") lines.append("") total_findings = 0 for skill_result in context.results: lines.append(f"## {skill_result.skill_name}") lines.append(f"- 状态: {skill_result.status.value}") lines.append(f"- 摘要: {skill_result.summary}") if skill_result.error: lines.append(f"- 错误: {skill_result.error}") if skill_result.findings: total_findings += len(skill_result.findings) lines.append(f"- 发现数量: {len(skill_result.findings)}") for finding in skill_result.findings: lines.append(f" - [{finding.severity}] {finding.title}: {finding.detail}") lines.append("") lines.append("## 总结") lines.append(f"本次扫描共发现 {total_findings} 条风险，请结合业务实际判断处置优先级。") context.metadata["report"] = "\n".join(lines) result.status = SkillStatus.SUCCESS result.summary = f"报告生成完成，共 {total_findings} 条发现。" return result

这里的context.metadata["report"]就是最终报告文本，后续可以继续接工单系统、邮件通知或者其他下游流程。

5.5 组装并运行

万事俱备，最后在main.py中组装整个系统：

# main.py from registry import SkillRegistry from router import SkillRouter from orchestrator import ScanOrchestrator from skills.log_scan_skill import LogScanSkill from skills.threat_lookup_skill import ThreatLookupSkill from skills.report_skill import ReportSkill def build_pipeline(): registry = SkillRegistry() registry.register(LogScanSkill()) registry.register(ThreatLookupSkill()) registry.register(ReportSkill()) rules = { "log_scan": ["日志", "log", "nginx", "访问"], "threat_lookup": ["ip", "威胁情报", "threat", "查询"], "report": ["报告", "report", "总结"], } router = SkillRouter(registry=registry, rules=rules) orchestrator = ScanOrchestrator(registry=registry, router=router) return orchestrator if __name__ == "__main__": nginx_log = """192.168.1.1 - - [20/Feb/2025:10:12:33 +0800] "GET /admin/login HTTP/1.1" 200 5423 "-" "Mozilla/5.0" 10.0.0.5 - - [20/Feb/2025:10:12:35 +0800] "GET /api/v1/order HTTP/1.1" 200 1200 "-" "curl/7.68.0" 192.168.1.1 - - [20/Feb/2025:10:12:36 +0800] "POST /login HTTP/1.1" 302 300 "-" "python-requests/2.25.1" 10.0.0.5 - - [20/Feb/2025:10:12:40 +0800] "GET /api/v1/user/list HTTP/1.1" 200 4310 "-" "curl/7.68.0" 192.168.1.1 - - [20/Feb/2025:10:12:41 +0800] "GET /admin/config HTTP/1.1" 403 200 "-" "Mozilla/5.0" 10.0.0.5 - - [20/Feb/2025:10:12:45 +0800] "GET /api/v1/user/list HTTP/1.1" 200 4310 "-" "curl/7.68.0" """ orchestrator = build_pipeline() context = orchestrator.run( task="分析这份 Nginx 日志中的异常访问，并生成报告", input_data=nginx_log, ) print("\n========== 最终报告 ==========") print(context.metadata.get("report", ""))

运行主程序前，请确认项目结构完整，skills目录下有__init__.py文件（可以是空文件），保证包可以被正确导入。

5.6 运行结果说明

预期的运行过程大致如下：

[orchestrator] 技能链: ['log_scan', 'threat_lookup', 'report'] [orchestrator] 执行技能: log_scan [orchestrator] 执行技能: threat_lookup [orchestrator] 执行技能: report ========== 最终报告 ========== # 安全分析报告 - 任务: 分析这份 Nginx 日志中的异常访问，并生成报告 - 时间: 2025-02-20 10:15:00 - 技能链: log_scan -> threat_lookup -> report ## log_scan - 状态: success - 摘要: 共解析 6 条日志，发现 4 条异常。 - 发现数量: 4 - [medium] 日志异常: 高频访问 IP=192.168.1.1，次数=3，可能存在扫描行为 - [medium] 日志异常: 越权访问尝试，IP=192.168.1.1，请求=GET /admin/config ... ## threat_lookup - 状态: success - 摘要: IP 192.168.1.1 查询完成，风险分 80 - 发现数量: 1 - [high] IP 威胁情报命中: 192.168.1.1 - 情报标签=扫描器，风险分=80，说明=历史存在主动扫描行为 ## report - 状态: success - 摘要: 报告生成完成，共 5 条发现。 ## 总结 本次扫描共发现 5 条风险，请结合业务实际判断处置优先级。

这个结果的核心在于：同样的输入、同样的技能链，无论谁运行、什么时间运行，结果都是一致的。这就是标准化流程带给安全分析的最大价值。

6. 常见问题与排查思路

以下是我在实际使用技能路由器过程中遇到的高频问题及排查建议：

排查思路可以按照这四步走：先确认输入数据是否正常，再确认路由结果是否符合预期，然后逐个技能单独测试，最后检查编排器的上下文传递是否正确。把问题定位到模块层级，多数故障都能在十分钟内找到根因。

7. 最佳实践与工程建议

7.1 合法授权是前提

安全扫描必须是合法授权的行为。在系统入口增加“任务是否已授权”的标记，并在报告中保留操作人、时间、授权凭证等审计字段。未授权任务一律拒绝执行。

7.2 技能要做最小化设计

一个技能只做一件事。把“日志分析”、“情报查询”、“报告生成”拆开，而不是做一个大而全的“分析技能”。技能越小，越容易测试、越容易复用，也越容易排查问题。

7.3 保留可追溯性

每个 SkillResult 都记录了技能名称、状态、摘要、错误信息。运营一段时间后，这些数据是宝贵的资产，可以用来分析哪些技能最常被调用、哪些技能容易失败、哪些规则需要优化。

7.4 引入降级策略

大模型路由是增强能力，不是必要能力。当大模型不可用时，规则路由必须能保证基本流程运行。同样，威胁情报查询超时时，不能阻断整个扫描流程，可以降级为“暂缺情报”继续执行。

7.5 数据脱敏与最小权限

分析日志时可能涉及敏感数据。技能执行前应先做字段级脱敏，比如隐藏用户 ID、Token、手机号。同时遵循最小权限原则，技能只能访问完成任务所需的数据，不得无条件读取全量日志。

7.6 建立技能回归测试

每个技能都应该有自己的测试用例，路由器的规则变化也要纳入回归范围。安全分析系统本身如果不可靠，它输出的结论就更不可靠。

8. 总结与后续学习方向

本文从安全分析场景中 AI 输出不稳定、流程不可控的问题出发，介绍了“安全分析技能路由器”的设计思路：把安全分析能力抽象为标准化的技能，通过路由器选择技能链，编排器固定流程，最终输出可复现、可审计的分析结论。

掌握这个模式之后，你可以继续往几个方向深入：

• 接入更丰富的数据源和技能，比如 CVE 查询、组件指纹识别、配置基线核查。
• 把规则路由替换为基于向量检索的语义路由，提高任务分诊准确率。
• 为技能增加并发执行能力，缩短扫描时间。
• 把结果接入工单系统或即时通知，形成完整的响应闭环。

实际项目中，不要一开始就追求大而全的智能系统。先把一两个高频安全分析场景跑通，沉淀出稳定的技能库，再逐步扩展路由规则和 AI 能力，这种演进方式更稳妥，也更容易获得团队信任。