MGeo日志审计功能：记录每次推理请求的来源与时间戳

引言：为什么需要日志审计？

在实际生产环境中，模型服务不仅仅是“输入→推理→输出”的简单流程。随着系统复杂度提升，尤其是多用户、多业务方共用一个推理服务时，可追溯性和安全性成为不可忽视的问题。MGeo作为阿里开源的中文地址相似度识别模型，在实体对齐任务中表现出色，广泛应用于物流、电商、城市治理等场景。

然而，当多个团队或系统调用同一MGeo服务进行地址匹配时，若缺乏有效的访问追踪机制，一旦出现异常请求、性能瓶颈或数据泄露，将难以定位问题源头。因此，为MGeo推理服务添加日志审计功能——记录每一次请求的来源IP与时间戳——是保障系统可观测性和安全合规的关键一步。

本文将围绕如何在已部署的MGeo服务中实现这一功能，结合具体部署环境（如4090D单卡+Jupyter），从技术选型、代码实现到最佳实践，提供一套完整可落地的解决方案。

技术背景：MGeo 地址相似度识别简介

MGeo 是阿里巴巴开源的一款专注于中文地址语义理解与相似度计算的深度学习模型，特别适用于“实体对齐”任务。其核心能力在于：

• 理解不同表述方式下的同一地理位置（如“北京市朝阳区建国路88号” vs “北京朝阳建国路88号”）
• 输出两个地址之间的相似度分数（0~1）
• 支持高并发、低延迟的在线推理

该模型基于大规模真实地址数据训练，融合了BERT类预训练语言模型与地理编码先验知识，在多个行业 benchmark 上达到 SOTA 表现。

典型应用场景： - 快递面单地址去重 - 用户注册信息清洗 - 多源POI（兴趣点）合并 - 城市数字底座建设中的空间数据融合

当前部署环境下，MGeo以镜像形式运行于NVIDIA 4090D单卡服务器上，通过Python脚本/root/推理.py提供本地推理接口。用户可通过复制该脚本至工作区（/root/workspace）进行调试和可视化编辑。

但默认情况下，该脚本并未开启任何请求日志记录功能。这意味着我们无法知道： - 谁在什么时候调用了服务？ - 是否存在高频恶意试探？ - 某次错误结果是由哪个客户端触发的？

这正是我们需要引入日志审计模块的原因。

方案设计：构建轻量级请求审计系统

为了在不影响MGeo主推理性能的前提下实现审计功能，我们采用以下设计原则：

| 设计目标 | 实现策略 | |--------|---------| |低侵入性| 不修改原始推理逻辑，仅在入口处增加中间层 | |高性能写入| 使用异步日志记录，避免阻塞主线程 | |结构化存储| 日志包含timestamp,client_ip,request_id,input_text等字段 | |可扩展性| 支持后续接入ELK、Prometheus等监控体系 |

✅ 技术选型对比

| 方案 | 优点 | 缺点 | 适用性 | |------|------|------|--------| |print + 重定向 > log.txt| 简单直接 | 非结构化、无级别控制 | ❌ 不推荐用于生产 | |logging模块 + 文件处理器 | 标准库支持、灵活配置 | 同步写入可能影响性能 | ⭕ 基础可用 | |logging + QueueHandler + Threaded| 异步非阻塞 | 实现稍复杂 | ✅ 推荐 | | Flask/Werkzeug 自带日志 | 若使用Web API则天然支持 | 当前为本地脚本模式 | ❌ 不适用 |

最终选择：Python 内置logging模块 + 异步队列机制

实现步骤详解：为推理.py添加审计日志

我们将对原始推理.py进行改造，在不改变其核心推理逻辑的基础上，嵌入请求审计功能。

第一步：环境准备与依赖确认

确保当前 Conda 环境已激活并具备必要权限：

conda activate py37testmaas

检查 Python 版本是否兼容（建议 ≥3.7）：

import sys print(sys.version)

无需额外安装包，因logging和threading均为标准库组件。

第二步：定义结构化日志格式

创建日志记录器，并设置输出格式包含时间戳、来源IP（模拟）、请求ID等关键字段。

import logging import logging.handlers import threading import time import uuid from queue import Queue # 创建异步日志队列 log_queue = Queue() queue_handler = logging.handlers.QueueHandler(log_queue) logger = logging.getLogger('mgeo_audit') logger.addHandler(queue_handler) logger.setLevel(logging.INFO) # 定义日志格式 formatter = logging.Formatter( '{"timestamp": "%(asctime)s", ' '"level": "%(levelname)s", ' '"client_ip": "%(client_ip)s", ' '"request_id": "%(request_id)s", ' '"event": "%(message)s"}' ) # 后台线程处理日志写入 def log_writer(): file_handler = logging.FileHandler("/root/logs/mgeo_access.log", encoding='utf-8') file_handler.setFormatter(formatter) listener = logging.handlers.QueueListener(log_queue, file_handler) listener.start() # 启动异步写入线程 threading.Thread(target=log_writer, daemon=True).start()

💡说明：
- 使用QueueHandler将日志事件放入队列，由独立线程消费写入文件
-daemon=True表示主线程退出时自动终止
- 日志路径/root/logs/需提前创建：mkdir -p /root/logs

第三步：封装推理函数并注入审计逻辑

假设原推理.py中存在如下核心函数：

def match_addresses(addr1: str, addr2: str) -> float: # 模拟MGeo推理过程 import random time.sleep(0.1) # 模拟模型加载延迟 return round(random.uniform(0.7, 1.0), 4)

我们将其包装成带审计的日志版本：

def audit_and_match(addr1: str, addr2: str, client_ip: str = "127.0.0.1"): # 生成唯一请求ID request_id = str(uuid.uuid4())[:8] # 记录开始时间 start_time = time.time() # 审计日志上下文 extra = { 'client_ip': client_ip, 'request_id': request_id } # 写入进入日志 logger.info(f"received_request addr1='{addr1}' addr2='{addr2}'", extra=extra) try: # 执行真实推理 result = match_addresses(addr1, addr2) latency = time.time() - start_time # 记录成功响应 logger.info(f"inference_success similarity={result} latency_ms={int(latency*1000)}", extra=extra) return result except Exception as e: # 记录异常 logger.error(f"inference_failed error='{str(e)}'", extra=extra) raise

第四步：模拟客户端调用并验证日志输出

添加测试代码段以验证功能：

if __name__ == "__main__": # 测试几组地址匹配 test_cases = [ ("北京市海淀区中关村大街1号", "北京海淀中关村大街1号"), ("上海市浦东新区张江高科园区", "上海浦东张江科技园"), ("广州市天河区体育东路", "广州天河体育东") ] for a1, a2 in test_cases: res = audit_and_match(a1, a2, client_ip="192.168.1.100") print(f"Similarity: {res}")

运行后查看日志文件：

tail -f /root/logs/mgeo_access.log

预期输出示例：

{"timestamp": "2025-04-05 10:23:15,123", "level": "INFO", "client_ip": "192.168.1.100", "request_id": "a1b2c3d4", "event": "received_request addr1='北京市海淀区中关村大街1号' addr2='北京海淀中关村大街1号'"} {"timestamp": "2025-04-05 10:23:15,246", "level": "INFO", "client_ip": "192.168.1.100", "request_id": "a1b2c3d4", "event": "inference_success similarity=0.9321 latency_ms=123"}

实践难点与优化建议

🔧 难点一：如何获取真实客户端 IP？

当前示例中client_ip为手动传入参数。但在真实部署中，需根据调用方式动态提取：

| 调用方式 | 获取IP方法 | |--------|-----------| | 本地脚本调用 | 可固定为localhost或主机内网IP | | REST API（Flask/FastAPI） |request.remote_addr| | gRPC 服务 |context.peer()| | Jupyter Notebook 调用 | 可通过%env或元数据获取 notebook 客户端IP |

示例（FastAPI）：python @app.post("/match") async def match_endpoint(item: AddressPair, request: Request): client_ip = request.client.host return {"similarity": audit_and_match(item.a1, item.a2, client_ip)}

🚀 优化建议一：日志轮转防爆盘

长期运行可能导致日志文件过大。使用RotatingFileHandler实现按大小切分：

file_handler = logging.handlers.RotatingFileHandler( "/root/logs/mgeo_access.log", maxBytes=100*1024*1024, # 100MB backupCount=10 )

📊 优化建议二：结构化日志便于分析

建议将日志输出为 JSON 格式，方便后续接入 ELK 或阿里云 SLS：

import json_log_formatter class CustomJsonFormatter(json_log_formatter.JSONFormatter): def json_record(self, message, extra, record): return { 'timestamp': record.asctime, 'level': record.levelname, 'client_ip': extra.get('client_ip', 'unknown'), 'request_id': extra.get('request_id'), 'event': message, 'module': record.module, }

🔐 安全建议：敏感信息脱敏

地址本身可能含用户隐私（如门牌号、姓名），建议在日志中做部分掩码：

def mask_address(addr: str) -> str: if len(addr) <= 6: return addr return addr[:3] + "*" * (len(addr)-6) + addr[-3:] # 使用 masked_a1 = mask_address(addr1) logger.info(f"received_request addr1='{masked_a1}' ...", extra=extra)

总结：打造可审计的MGeo推理服务

通过对推理.py脚本的轻量级改造，我们成功实现了对每一次MGeo地址相似度推理请求的完整审计追踪，包括：

✅ 请求时间戳
✅ 客户端来源IP
✅ 唯一请求ID
✅ 推理耗时与结果状态
✅ 结构化JSON日志输出

这套方案具有以下优势：

• 零依赖：仅使用Python标准库，无需安装第三方包
• 高性能：异步写入不影响主推理流程
• 易集成：可无缝嵌入现有脚本或未来Web服务
• 可扩展：支持对接日志平台、告警系统、数据分析管道

下一步建议：迈向生产级可观测性

完成基础日志审计后，可进一步升级为完整的可观测体系：

1. 接入日志服务：将/root/logs/mgeo_access.log推送至阿里云SLS或ELK
2. 建立仪表盘：统计QPS、平均延迟、异常率等指标
3. 设置告警规则：如单位时间内失败请求超过阈值自动通知
4. 结合Trace系统：与OpenTelemetry集成，实现全链路追踪

🎯一句话总结：
日志不是附加品，而是AI服务的“黑匣子”。只有看得见请求从哪来、往哪去，才能真正掌控模型的每一次心跳。