AI智能实体侦测服务日志记录功能：操作审计追踪实现教程
1. 引言
1.1 业务场景描述
在企业级AI应用中，命名实体识别（NER）服务常用于新闻分析、舆情监控、客户信息提取等关键场景。随着系统规模扩大，多个用户或API调用方并行使用AI实体侦测服务时，缺乏操作日志记录将导致无法追溯谁在何时执行了哪些文本分析任务，给安全审计、问题排查和资源管理带来巨大挑战。
当前部署的AI智能实体侦测服务虽已具备高性能中文实体抽取与WebUI交互能力，但默认未开启详细的操作日志追踪。为满足合规性要求和运维可观察性需求，亟需实现一套轻量级、可扩展的日志记录机制。
1.2 痛点分析
- 操作不可追溯：无法知道某次实体识别请求来自哪个用户或IP。
- 调试困难：当识别结果异常时，缺少输入文本与时间戳的关联记录。
- 无访问统计：难以评估服务使用频率、高峰时段及负载分布。
1.3 方案预告
本文将基于已部署的RaNER模型Web服务，手把手实现一个完整的操作审计追踪系统，包含： - 请求日志自动记录（时间、IP、输入文本、识别结果） - 日志结构化存储（JSON格式文件） - Web界面查看最近操作记录 - 安全过滤敏感信息（如脱敏长文本）
最终实现“谁、何时、输入什么、得到什么”的全流程可审计。
2. 技术方案选型
2.1 可行方案对比
| 方案 | 优点 | 缺点 | 适用性 |
|---|---|---|---|
| Flask 内置 logging + 文件写入 | 轻量、无需依赖、易集成 | 功能简单，难查询 | ✅ 适合本项目 |
| SQLite 记录操作日志 | 结构清晰，支持查询 | 增加数据库依赖 | ⚠️ 过重 |
| ELK 栈（Elasticsearch+Logstash+Kibana） | 强大分析能力 | 部署复杂，资源消耗高 | ❌ 不适用 |
结论：选择Flask logging + JSON日志文件方案。理由如下： - 服务本身基于Flask构建，天然支持日志中间件 - 日志量预计不大（每日<1万条），文件存储足够 - 易于后续对接日志收集工具（如Filebeat）
2.2 架构设计思路
[用户提交文本] ↓ [Flask路由接收请求] ↓ [调用RaNER模型进行实体识别] ↓ [中间件：记录日志到 audit.log] ↓ [返回高亮HTML结果] ↓ [前端展示 + 提供日志查看入口]核心组件： -audit_logger.py：日志处理器，负责格式化与写入 -logs/audit.log：结构化JSON日志文件 -/history路由：提供最近10条操作记录API
3. 实现步骤详解
3.1 环境准备
确保服务运行目录结构如下：
/ner-webui/ ├── app.py # 主应用入口 ├── models/ # RaNER模型文件 ├── static/ # 静态资源 ├── templates/ # 前端模板 └── logs/ # 新建日志目录 └── audit.log # 操作审计日志创建日志目录：
mkdir -p logs touch logs/audit.log3.2 核心代码实现
修改主应用文件app.py
import json import os from datetime import datetime from flask import Flask, request, jsonify, render_template import logging # 初始化Flask应用 app = Flask(__name__) # 配置日志 LOG_FILE = 'logs/audit.log' logging.basicConfig(level=logging.INFO) def write_audit_log(client_ip, input_text, entities): """写入审计日志""" log_entry = { "timestamp": datetime.now().isoformat(), "client_ip": client_ip, "input_length": len(input_text), "input_preview": input_text[:50] + "..." if len(input_text) > 50 else input_text, "entity_count": len(entities), "entities": entities } try: with open(LOG_FILE, 'a', encoding='utf-8') as f: f.write(json.dumps(log_entry, ensure_ascii=False) + '\n') except Exception as e: print(f"日志写入失败: {e}") @app.route('/detect', methods=['POST']) def detect_entities(): data = request.get_json() text = data.get('text', '') # --- 此处调用RaNER模型进行实体识别 --- # 模拟返回结果（实际应替换为真实模型输出） from mock_model import mock_ner_predict entities = mock_ner_predict(text) # ---------------------------------------- # 写入审计日志 client_ip = request.headers.get('X-Forwarded-For', request.remote_addr) write_audit_log(client_ip, text, entities) return jsonify({"success": True, "entities": entities}) @app.route('/history', methods=['GET']) def get_history(): """获取最近10条操作记录""" if not os.path.exists(LOG_FILE): return jsonify([]) logs = [] try: with open(LOG_FILE, 'r', encoding='utf-8') as f: lines = [line.strip() for line in f.readlines() if line.strip()] # 取最后10条 recent = lines[-10:] for line in recent: logs.append(json.loads(line)) except Exception as e: print(f"读取日志失败: {e}") return jsonify(logs[::-1]) # 倒序：最新在前 @app.route('/') def index(): return render_template('index.html') if __name__ == '__main__': app.run(host='0.0.0.0', port=7860)创建模拟模型文件mock_model.py（用于测试）
def mock_ner_predict(text): """模拟RaNER模型输出（实际应加载真实模型）""" import re entities = [] # 简单规则匹配（仅演示用） persons = re.finditer(r'([赵钱孙李周吴郑王][小大伟强丽静涛])', text) for m in persons: entities.append({ "text": m.group(1), "type": "PER", "start": m.start(), "end": m.end() }) locations = re.finditer(r'(北京|上海|广州|深圳|杭州)', text) for m in locations: entities.append({ "text": m.group(), "type": "LOC", "start": m.start(), "end": m.end() }) organizations = re.finditer(r'(公司|集团|大学|医院)', text) for m in organizations: entities.append({ "text": m.group(), "type": "ORG", "start": m.start(), "end": m.end() }) return entities更新前端页面templates/index.html添加历史记录查看
<!DOCTYPE html> <html> <head> <title>🔍 AI 智能实体侦测服务</title> <style> body { font-family: 'Courier New', monospace; background: #0f0f1a; color: #00ffcc; padding: 20px; } .container { max-width: 900px; margin: 0 auto; } textarea { width: 100%; height: 150px; background: #1a1a2e; color: #fff; border: 1px solid #00ffcc; padding: 10px; } button { background: #00bfff; color: white; padding: 10px 20px; border: none; cursor: pointer; font-size: 16px; } #result { margin-top: 20px; border: 1px dashed #00ffcc; padding: 15px; background: rgba(0,255,204,0.1); } .history-btn { background: #8a2be2; margin-left: 10px; } .history-modal { display: none; position: fixed; top: 10%; left: 10%; right: 10%; background: #1a1a2e; border: 1px solid #00ffcc; padding: 20px; z-index: 100; } .close-btn { float: right; cursor: pointer; color: #ff0055; } </style> </head> <body> <div class="container"> <h1>🚀 AI 智能实体侦测服务</h1> <p>粘贴任意中文文本，自动识别并高亮人名、地名、机构名。</p> <textarea id="inputText" placeholder="请输入待分析的文本..."></textarea> <div style="margin: 15px 0;"> <button onclick="detect()">🚀 开始侦测</button> <button class="history-btn" onclick="showHistory()">📋 查看操作记录</button> </div> <div id="result"></div> <!-- 历史记录弹窗 --> <div id="historyModal" class="history-modal"> <span class="close-btn" onclick="document.getElementById('historyModal').style.display='none'">×</span> <h3>📜 最近操作记录（最近10条）</h3> <div id="historyList"></div> </div> </div> <script> async function detect() { const text = document.getElementById('inputText').value; const res = await fetch('/detect', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ text }) }); const data = await res.json(); let html = text; // 按位置倒序插入标签，避免索引偏移 data.entities.sort((a,b)=>b.start-a.start).forEach(ent=>{ const color = ent.type === 'PER' ? 'red' : ent.type === 'LOC' ? 'cyan' : 'yellow'; const tag = `<span style="color:${color};font-weight:bold">${ent.text}</span>`; html = html.slice(0, ent.start) + tag + html.slice(ent.end); }); document.getElementById('result').innerHTML = html; } async function showHistory() { const modal = document.getElementById('historyModal'); const list = document.getElementById('historyList'); list.innerHTML = '加载中...'; modal.style.display = 'block'; try { const res = await fetch('/history'); const logs = await res.json(); if (logs.length === 0) { list.innerHTML = '暂无操作记录。'; return; } list.innerHTML = logs.map(log => ` <div style="margin:10px 0;padding:10px;border-bottom:1px solid #333"> <strong>⏱ ${new Date(log.timestamp).toLocaleString()}</strong><br> 🖥 IP: ${log.client_ip}<br> 📄 输入: ${log.input_preview}<br> 🔍 识别出 ${log.entity_count} 个实体 </div> `).join(''); } catch (e) { list.innerHTML = '加载失败: ' + e.message; } } </script> </body> </html>3.3 关键代码解析
日志写入函数write_audit_log
- 使用追加模式 (
'a') 写入，保证每次请求独立成行 - 字段脱敏：长文本只保留前50字符预览
- 时间戳采用ISO标准格式，便于后续解析
路由/history
- 返回JSON数组，兼容前后端分离架构
- 逆序返回（最新在前），提升用户体验
- 异常捕获防止因日志损坏导致接口崩溃
前端历史查看功能
- 模态框设计不打断主流程
- 自动格式化时间显示
- 支持快速关闭
4. 实践问题与优化
4.1 实际遇到的问题
问题1：并发写入日志冲突
多用户同时请求可能导致文件写入竞争。
✅解决方案： 使用线程锁保护写入操作：
import threading log_lock = threading.Lock() def write_audit_log(...): with log_lock: with open(...) as f: f.write(...)问题2：日志文件无限增长
长期运行后日志文件过大，影响性能。
✅解决方案： 添加日志轮转机制（按天分割）：
from datetime import datetime LOG_FILE = f"logs/audit_{datetime.now().strftime('%Y%m%d')}.log"问题3：敏感信息泄露风险
原始文本可能包含隐私内容。
✅解决方案： 配置可选脱敏策略：
# 配置项 REDACT_FULL_TEXT = False # 是否完全不记录原文 # 日志中替换为占位符 "input_preview": "[REDACTED]" if REDACT_FULL_TEXT else text[:50]+"..."5. 性能优化建议
5.1 异步日志写入（推荐）
避免阻塞主线程，提升响应速度：
import asyncio import aiofiles async def async_write_log(entry): async with aiofiles.open(LOG_FILE, 'a') as f: await f.write(json.dumps(entry, ensure_ascii=False) + '\n') # 在视图中使用： # asyncio.create_task(async_write_log(log_entry))5.2 日志压缩归档
定期压缩旧日志节省空间：
# 每日凌晨执行 find logs/ -name "audit_*.log" -mtime +7 -exec gzip {} \;5.3 添加索引字段
若未来需快速检索，可在日志中增加：
"keywords": ["张三", "北京", "科技公司"]6. 总结
6.1 实践经验总结
通过本次改造，我们成功为AI智能实体侦测服务增加了完整的操作审计能力。核心收获包括： -轻量级设计优于重型框架：对于中小规模服务，原生文件日志足以胜任 -结构化日志是关键：JSON格式便于机器解析和后续分析 -前端集成提升可用性：无需SSH即可查看操作历史
6.2 最佳实践建议
- 始终记录客户端IP：即使在内网环境，也有助于定位问题来源
- 控制日志粒度：避免记录完整长文本，平衡审计与隐私
- 定期清理机制：设置自动删除超过30天的日志文件
💡获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
