SecGPT-14B代码实例：Python调用OpenAI API实现批量日志安全评分

1. 项目背景与价值

在网络安全运维中，每天都会产生大量系统日志和安全事件记录。传统的人工分析方式效率低下，难以应对海量日志的实时检测需求。SecGPT-14B作为专业的网络安全大模型，能够快速理解日志内容并给出安全评分，帮助安全团队高效识别潜在威胁。

核心价值：

• 自动化分析：批量处理日志文件，无需人工逐条检查
• 智能评分：基于上下文理解给出0-100的安全风险评分
• 快速响应：单条日志分析仅需0.5-2秒
• 可集成：标准OpenAI API接口，轻松嵌入现有工作流

2. 环境准备与API配置

2.1 安装必要依赖

pip install openai pandas tqdm

2.2 API基础配置

import openai
import pandas as pd
from tqdm import tqdm

# 配置API连接
openai.api_base = "http://127.0.0.1:8000/v1"  # 本地部署地址
openai.api_key = "EMPTY"  # 本地部署无需密钥

# 指定模型
MODEL_NAME = "SecGPT-14B"

3. 核心功能实现

3.1 单条日志评分函数

def score_log_entry(log_entry):
    """对单条日志进行安全评分"""
    prompt = f"""请对以下系统日志进行安全风险评估，给出0-100的风险评分(越高越危险)，
并简要说明理由。直接返回JSON格式：
{{
    "score": 评分值,
    "reason": "风险原因"
}}

日志内容：{log_entry}"""

    response = openai.ChatCompletion.create(
        model=MODEL_NAME,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,  # 降低随机性
        max_tokens=256
    )
    
    try:
        return eval(response.choices[0].message.content)
    except:
        return {"score": -1, "reason": "解析失败"}

3.2 批量日志处理实现

def batch_score_logs(log_file, output_file="results.csv"):
    """批量处理日志文件"""
    # 读取日志文件
    with open(log_file) as f:
        logs = [line.strip() for line in f if line.strip()]
    
    results = []
    for log in tqdm(logs, desc="分析进度"):
        result = score_log_entry(log)
        results.append({
            "log": log,
            "score": result["score"],
            "reason": result["reason"]
        })
    
    # 保存结果
    pd.DataFrame(results).to_csv(output_file, index=False)
    print(f"分析完成，结果已保存到 {output_file}")

4. 实战案例演示

4.1 示例日志文件

创建测试日志文件sample_logs.txt：

用户admin从192.168.1.100成功登录
检测到对/api/user/list的异常高频访问(127次/分钟)
数据库备份任务执行失败：权限不足
防火墙规则被临时禁用
检测到可疑SQL注入尝试：SELECT * FROM users WHERE 1=1

4.2 执行批量分析

batch_score_logs("sample_logs.txt", "security_scores.csv")

4.3 预期输出结果

生成的security_scores.csv示例：

log	score	reason
用户admin从192.168.1.100成功登录	10	正常登录行为
检测到对/api/user/list的异常高频访问(127次/分钟)	75	可能为暴力破解尝试
数据库备份任务执行失败：权限不足	40	需检查权限配置
防火墙规则被临时禁用	90	高危操作，需立即核查
检测到可疑SQL注入尝试：SELECT * FROM users WHERE 1=1	100	明确的SQL注入攻击

5. 高级功能扩展

5.1 实时监控集成方案

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class LogHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith(".log"):
            print(f"检测到日志变更: {event.src_path}")
            batch_score_logs(event.src_path)

# 启动监控
observer = Observer()
observer.schedule(LogHandler(), path="/var/log/")
observer.start()

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

5.2 告警阈值设置

def analyze_with_alert(log_file, threshold=70):
    """带告警功能的日志分析"""
    df = pd.read_csv(batch_score_logs(log_file))
    
    high_risk = df[df["score"] >= threshold]
    if not high_risk.empty:
        send_alert(high_risk)
    
    return df

def send_alert(high_risk_logs):
    """发送告警通知"""
    alert_msg = "发现高风险日志：\n" + "\n".join(
        f"[评分 {row['score']}] {row['log']}" 
        for _, row in high_risk_logs.iterrows()
    )
    print(alert_msg)
    # 这里可集成邮件/短信等告警方式

6. 性能优化建议

1. 批量请求优化：

   # 将多条日志合并为一个请求
   def batch_score(logs, batch_size=5):
       prompt = "请分别评估以下日志的安全风险...\n" + "\n".join(
           f"{i+1}. {log}" for i, log in enumerate(logs)
       )
       response = openai.ChatCompletion.create(...)
       # 解析批量响应
   

2. 缓存机制：

   from functools import lru_cache
   
   @lru_cache(maxsize=1000)
   def cached_score(log_entry):
       return score_log_entry(log_entry)
   

3. 连接池配置：

   import httpx
   
   client = httpx.Client(base_url=openai.api_base)
   openai.requestssession = client
   

7. 总结与建议

通过本文介绍的方法，您可以快速构建一个基于SecGPT-14B的日志安全分析系统。在实际应用中建议：

1. 日志预处理：清洗无关日志，提高分析效率
2. 评分校准：根据业务特点调整评分标准
3. 结果复核：对高风险日志进行人工确认
4. 持续优化：收集误报案例优化提示词

对于企业级应用，可以考虑：

• 与SIEM系统集成
• 建立历史评分基线
• 开发可视化监控面板

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。