之前在安全分析场景里接大模型时我最大的感受是AI 确实能看懂日志、能提建议但很难稳定地按照一套标准流程走完整个安全扫描任务。同一个日志换个时间问、换个模型问过程和结论可能完全不同换个人来调可能又是一个新玩法。团队里每个人都在“调 AI”但没有一套大家可以共用的、可复现、可审计的分析流水线。后来我们逐步把“安全分析能力”从大模型的自由对话里剥离出来做成一个个标准化的“技能”再通过一个类似路由器的中间层根据任务类型自动决定调用哪个技能、按什么顺序执行最终输出结构化的扫描结果。这套设计就是我们常说的“安全分析技能路由器”。本文会围绕这个主题拆解“技能路由器”是什么、解决什么问题、如何用 Python 实现一个最小可运行的版本并给出一个基于 Nginx 日志分析的完整实战示例帮助大家理解如何用工程化的方式约束 AI 的安全分析行为。1. 背景与核心概念1.1 为什么 AI 做安全分析容易“失控”大模型接入安全分析流程常见的做法是直接把日志丢给模型让它“分析一下有没有风险”。这种方式在探索阶段非常高效但一旦进入正式工作流就会出现几个问题第一输出不稳定。同一份日志问两次可能得到两份侧重点完全不同的报告。安全分析要求的是可复现性如果结论无法稳定复现后续审计和处置就没法落地。第二流程不可控。大模型不会主动记住“先做日志清洗再做异常统计然后查询威胁情报最后输出报告”这件事。用户少说一句它就可能跳过某个关键步骤。第三工具能力不可复用。今天让 AI 查一个 IP 的威胁情报明天换一个场景又要重新写提示词。查询逻辑、数据源、超时策略全部散落在对话里无法沉淀成团队资产。第四安全责任边界模糊。安全问题不是聊天扫描动作必须经过授权结果必须有据可查。如果全靠大模型自由发挥出了问题很难回溯。1.2 什么是“技能路由器”“技能路由器”这个概念借鉴了网络设备中的路由器思想数据包进入路由器后根据路由表决定下一跳走向。在安全分析场景中我们把“任务”看作数据包把“安全分析能力”看作技能路由器则根据任务特征自动匹配一个或多个技能并编排执行顺序。更具体地说一个技能是“完成一个安全分析步骤的最小可执行单元”它应当具备明确的输入比如日志文本、IP 地址、域名、配置片段。明确的功能比如日志异常检测、端口指纹识别、CVE 信息查询、报告生成。标准化的输出比如统一的数据结构包含严重级别、描述、来源。可独立测试每个技能都能单独验证正确性。路由器不对业务逻辑负责它只做两件事识别任务意图决定技能执行链。这种设计让大模型从“执行者”变成“调度者”从“凭感觉分析”变成“按标准流程走”既保留了 AI 的理解能力又限制了它的不可控性。1.3 安全分析的典型应用场景技能路由器的应用范围很广安全团队在日常工作中遇到的重复性分析场景基本都可以通过这个思路标准化场景技能链示例产出物访问日志异常分析日志清洗 - 异常统计 - 威胁情报查询 - 报告生成异常 IP 清单、风险等级配置基线核查配置采集 - 规则比对 - 差异说明 - 整改建议不合规配置列表漏洞情报跟进组件版本提取 - CVE 查询 - 影响面分析 - 修复建议漏洞影响评估报告告警事件研判告警归一化 - 关联分析 - 威胁打分 - 处置建议事件研判结论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_factorylist) summary: str error: str dataclass class ScanContext: 一次安全扫描任务的完整上下文 task: str input_data: Optional[str] None params: dict field(default_factorydict) results: List[SkillResult] field(default_factorylist) metadata: dict field(default_factorydict) 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_clientNone, 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(tasktask, input_datainput_data, paramsparams 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 访问日志需要完成三个任务检测日志中的异常访问行为。对可疑 IP 进行威胁情报查询。生成标准化的安全分析报告。我们以一小段示例日志作为输入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_nameself.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( severitymedium, title日志异常, detailitem, sourceself.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_nameself.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( severityhigh, titlefIP 威胁情报命中: {ip_to_check}, detailf情报标签{info[tag]}风险分{info[score]}说明{info[desc]}, sourceself.name )) result.status SkillStatus.SUCCESS result.summary fIP {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_nameself.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(registryregistry, rulesrules) orchestrator ScanOrchestrator(registryregistry, routerrouter) 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_datanginx_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] 日志异常: 高频访问 IP192.168.1.1次数3可能存在扫描行为 - [medium] 日志异常: 越权访问尝试IP192.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. 常见问题与排查思路以下是我在实际使用技能路由器过程中遇到的高频问题及排查建议问题现象常见原因解决思路LLM 路由失败导致流程中断大模型接口超时、返回格式不规范增加规则路由兜底大模型路由失败时直接使用规则结果技能重复注册启动时误加载了多个同名技能在注册中心增加重复名称告警统一技能命名规范日志解析不出任何记录日志格式与正则不匹配先做样本日志调试确认正则覆盖实际日志格式技能链执行顺序不符合预期规则关键词覆盖了场景导致误匹配细化关键词列表增加优先级字段上下文数据过大大日志文件全部塞进内存增加输入限制大文件先做采样或分段处理报告中的风险等级不一致不同技能对 severity 的定义不一致在 Finding 模型层统一枚举禁止自由字符串排查思路可以按照这四步走先确认输入数据是否正常再确认路由结果是否符合预期然后逐个技能单独测试最后检查编排器的上下文传递是否正确。把问题定位到模块层级多数故障都能在十分钟内找到根因。7. 最佳实践与工程建议7.1 合法授权是前提安全扫描必须是合法授权的行为。在系统入口增加“任务是否已授权”的标记并在报告中保留操作人、时间、授权凭证等审计字段。未授权任务一律拒绝执行。7.2 技能要做最小化设计一个技能只做一件事。把“日志分析”、“情报查询”、“报告生成”拆开而不是做一个大而全的“分析技能”。技能越小越容易测试、越容易复用也越容易排查问题。7.3 保留可追溯性每个 SkillResult 都记录了技能名称、状态、摘要、错误信息。运营一段时间后这些数据是宝贵的资产可以用来分析哪些技能最常被调用、哪些技能容易失败、哪些规则需要优化。7.4 引入降级策略大模型路由是增强能力不是必要能力。当大模型不可用时规则路由必须能保证基本流程运行。同样威胁情报查询超时时不能阻断整个扫描流程可以降级为“暂缺情报”继续执行。7.5 数据脱敏与最小权限分析日志时可能涉及敏感数据。技能执行前应先做字段级脱敏比如隐藏用户 ID、Token、手机号。同时遵循最小权限原则技能只能访问完成任务所需的数据不得无条件读取全量日志。7.6 建立技能回归测试每个技能都应该有自己的测试用例路由器的规则变化也要纳入回归范围。安全分析系统本身如果不可靠它输出的结论就更不可靠。8. 总结与后续学习方向本文从安全分析场景中 AI 输出不稳定、流程不可控的问题出发介绍了“安全分析技能路由器”的设计思路把安全分析能力抽象为标准化的技能通过路由器选择技能链编排器固定流程最终输出可复现、可审计的分析结论。掌握这个模式之后你可以继续往几个方向深入接入更丰富的数据源和技能比如 CVE 查询、组件指纹识别、配置基线核查。把规则路由替换为基于向量检索的语义路由提高任务分诊准确率。为技能增加并发执行能力缩短扫描时间。把结果接入工单系统或即时通知形成完整的响应闭环。实际项目中不要一开始就追求大而全的智能系统。先把一两个高频安全分析场景跑通沉淀出稳定的技能库再逐步扩展路由规则和 AI 能力这种演进方式更稳妥也更容易获得团队信任。 安全分析技能路由器：用Python工程化约束大模型
之前在安全分析场景里接大模型时我最大的感受是AI 确实能看懂日志、能提建议但很难稳定地按照一套标准流程走完整个安全扫描任务。同一个日志换个时间问、换个模型问过程和结论可能完全不同换个人来调可能又是一个新玩法。团队里每个人都在“调 AI”但没有一套大家可以共用的、可复现、可审计的分析流水线。后来我们逐步把“安全分析能力”从大模型的自由对话里剥离出来做成一个个标准化的“技能”再通过一个类似路由器的中间层根据任务类型自动决定调用哪个技能、按什么顺序执行最终输出结构化的扫描结果。这套设计就是我们常说的“安全分析技能路由器”。本文会围绕这个主题拆解“技能路由器”是什么、解决什么问题、如何用 Python 实现一个最小可运行的版本并给出一个基于 Nginx 日志分析的完整实战示例帮助大家理解如何用工程化的方式约束 AI 的安全分析行为。1. 背景与核心概念1.1 为什么 AI 做安全分析容易“失控”大模型接入安全分析流程常见的做法是直接把日志丢给模型让它“分析一下有没有风险”。这种方式在探索阶段非常高效但一旦进入正式工作流就会出现几个问题第一输出不稳定。同一份日志问两次可能得到两份侧重点完全不同的报告。安全分析要求的是可复现性如果结论无法稳定复现后续审计和处置就没法落地。第二流程不可控。大模型不会主动记住“先做日志清洗再做异常统计然后查询威胁情报最后输出报告”这件事。用户少说一句它就可能跳过某个关键步骤。第三工具能力不可复用。今天让 AI 查一个 IP 的威胁情报明天换一个场景又要重新写提示词。查询逻辑、数据源、超时策略全部散落在对话里无法沉淀成团队资产。第四安全责任边界模糊。安全问题不是聊天扫描动作必须经过授权结果必须有据可查。如果全靠大模型自由发挥出了问题很难回溯。1.2 什么是“技能路由器”“技能路由器”这个概念借鉴了网络设备中的路由器思想数据包进入路由器后根据路由表决定下一跳走向。在安全分析场景中我们把“任务”看作数据包把“安全分析能力”看作技能路由器则根据任务特征自动匹配一个或多个技能并编排执行顺序。更具体地说一个技能是“完成一个安全分析步骤的最小可执行单元”它应当具备明确的输入比如日志文本、IP 地址、域名、配置片段。明确的功能比如日志异常检测、端口指纹识别、CVE 信息查询、报告生成。标准化的输出比如统一的数据结构包含严重级别、描述、来源。可独立测试每个技能都能单独验证正确性。路由器不对业务逻辑负责它只做两件事识别任务意图决定技能执行链。这种设计让大模型从“执行者”变成“调度者”从“凭感觉分析”变成“按标准流程走”既保留了 AI 的理解能力又限制了它的不可控性。1.3 安全分析的典型应用场景技能路由器的应用范围很广安全团队在日常工作中遇到的重复性分析场景基本都可以通过这个思路标准化场景技能链示例产出物访问日志异常分析日志清洗 - 异常统计 - 威胁情报查询 - 报告生成异常 IP 清单、风险等级配置基线核查配置采集 - 规则比对 - 差异说明 - 整改建议不合规配置列表漏洞情报跟进组件版本提取 - CVE 查询 - 影响面分析 - 修复建议漏洞影响评估报告告警事件研判告警归一化 - 关联分析 - 威胁打分 - 处置建议事件研判结论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_factorylist) summary: str error: str dataclass class ScanContext: 一次安全扫描任务的完整上下文 task: str input_data: Optional[str] None params: dict field(default_factorydict) results: List[SkillResult] field(default_factorylist) metadata: dict field(default_factorydict) 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_clientNone, 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(tasktask, input_datainput_data, paramsparams 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 访问日志需要完成三个任务检测日志中的异常访问行为。对可疑 IP 进行威胁情报查询。生成标准化的安全分析报告。我们以一小段示例日志作为输入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_nameself.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( severitymedium, title日志异常, detailitem, sourceself.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_nameself.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( severityhigh, titlefIP 威胁情报命中: {ip_to_check}, detailf情报标签{info[tag]}风险分{info[score]}说明{info[desc]}, sourceself.name )) result.status SkillStatus.SUCCESS result.summary fIP {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_nameself.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(registryregistry, rulesrules) orchestrator ScanOrchestrator(registryregistry, routerrouter) 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_datanginx_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] 日志异常: 高频访问 IP192.168.1.1次数3可能存在扫描行为 - [medium] 日志异常: 越权访问尝试IP192.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. 常见问题与排查思路以下是我在实际使用技能路由器过程中遇到的高频问题及排查建议问题现象常见原因解决思路LLM 路由失败导致流程中断大模型接口超时、返回格式不规范增加规则路由兜底大模型路由失败时直接使用规则结果技能重复注册启动时误加载了多个同名技能在注册中心增加重复名称告警统一技能命名规范日志解析不出任何记录日志格式与正则不匹配先做样本日志调试确认正则覆盖实际日志格式技能链执行顺序不符合预期规则关键词覆盖了场景导致误匹配细化关键词列表增加优先级字段上下文数据过大大日志文件全部塞进内存增加输入限制大文件先做采样或分段处理报告中的风险等级不一致不同技能对 severity 的定义不一致在 Finding 模型层统一枚举禁止自由字符串排查思路可以按照这四步走先确认输入数据是否正常再确认路由结果是否符合预期然后逐个技能单独测试最后检查编排器的上下文传递是否正确。把问题定位到模块层级多数故障都能在十分钟内找到根因。7. 最佳实践与工程建议7.1 合法授权是前提安全扫描必须是合法授权的行为。在系统入口增加“任务是否已授权”的标记并在报告中保留操作人、时间、授权凭证等审计字段。未授权任务一律拒绝执行。7.2 技能要做最小化设计一个技能只做一件事。把“日志分析”、“情报查询”、“报告生成”拆开而不是做一个大而全的“分析技能”。技能越小越容易测试、越容易复用也越容易排查问题。7.3 保留可追溯性每个 SkillResult 都记录了技能名称、状态、摘要、错误信息。运营一段时间后这些数据是宝贵的资产可以用来分析哪些技能最常被调用、哪些技能容易失败、哪些规则需要优化。7.4 引入降级策略大模型路由是增强能力不是必要能力。当大模型不可用时规则路由必须能保证基本流程运行。同样威胁情报查询超时时不能阻断整个扫描流程可以降级为“暂缺情报”继续执行。7.5 数据脱敏与最小权限分析日志时可能涉及敏感数据。技能执行前应先做字段级脱敏比如隐藏用户 ID、Token、手机号。同时遵循最小权限原则技能只能访问完成任务所需的数据不得无条件读取全量日志。7.6 建立技能回归测试每个技能都应该有自己的测试用例路由器的规则变化也要纳入回归范围。安全分析系统本身如果不可靠它输出的结论就更不可靠。8. 总结与后续学习方向本文从安全分析场景中 AI 输出不稳定、流程不可控的问题出发介绍了“安全分析技能路由器”的设计思路把安全分析能力抽象为标准化的技能通过路由器选择技能链编排器固定流程最终输出可复现、可审计的分析结论。掌握这个模式之后你可以继续往几个方向深入接入更丰富的数据源和技能比如 CVE 查询、组件指纹识别、配置基线核查。把规则路由替换为基于向量检索的语义路由提高任务分诊准确率。为技能增加并发执行能力缩短扫描时间。把结果接入工单系统或即时通知形成完整的响应闭环。实际项目中不要一开始就追求大而全的智能系统。先把一两个高频安全分析场景跑通沉淀出稳定的技能库再逐步扩展路由规则和 AI 能力这种演进方式更稳妥也更容易获得团队信任。