
我是张大鹏做了十多年人工智能带过不少项目。说实话最难的不是写业务代码是线上出了 Bug 你却查不出来。最近在完善 InsMatrixAutomation 项目时我花了一整周时间重构日志系统现在终于敢说任何请求进来我都能追踪到它的一生。今天给大家详细介绍一下这套日志系统的设计思路和实现方案。一、为什么中小项目需要结构化日志1.1 我踩过的那些日志坑做开发这么多年我见过太多团队在日志上吃亏痛点后果日志分散在多处排查一个问题要翻十几个文件格式不统一AI 解析困难无法批量查询缺少 request_id同一个请求的日志串不到一起日志过多或过少要么淹没在噪音里要么关键信息被遗漏我的感受是很多中小项目根本不重视日志觉得能打印就行。等到线上出问题才发现日志根本没法用这时候已经晚了。1.2 结构化日志的核心价值为什么我坚持用结构化日志三个原因可追溯每个请求有唯一 ID串联所有相关日志可查询JSON 格式可以直接用 jq、pandas 分析甚至丢给 AI 处理可持久化同时落文件和入库既方便实时查看又便于历史追溯二、Loguru vs 标准 logging vs structlog为什么我选 Loguru2.1 三大框架横评在设计日志系统之初我调研了 Python 生态里最常见的三种日志方案对比项标准 loggingLogurustructlog依赖无需安装需安装配置复杂度高低中JSON 序列化手动自带自带轮转/压缩需配合自带需配合代码可读性一般极佳一般Flask 集成一般好一般我的结论是对于中小型 Flask 应用Loguru 是最佳选择。零配置就能用logger.add()搞定一切没有道理拒绝它。2.2 快速安装uvaddloguru是的就这一行。没有了。三、Loguru 实战零配置搞定 JSON 日志3.1 基础配置Loguru 的精髓在于logger.add()这一行# logging/config.pyfromloguruimportloggerfrompathlibimportPathdefsetup_logging(log_level:strINFO,log_dir:strlogs):初始化日志系统# 移除默认处理器避免重复输出logger.remove()# 确保日志目录存在log_pathPath(log_dir)log_path.mkdir(parentsTrue,exist_okTrue)# 文件输出 - JSON Lines 格式logger.add(f{log_dir}/app_{{time:YYYYMMDD}}.jsonl,format{time:ISO}|{level}|{message},levellog_level,rotation00:00,# 每天零点轮转retention30 days,# 保留30天compressionzip,# 压缩旧日志serializeTrue,# JSON 序列化enqueueTrue,# 线程安全)# 控制台输出 - 可读格式开发用logger.add(sinklambdamsg:print(msg),formatgreen{time:HH:mm:ss}/green | level{level}/level | {message},levelDEBUG,)returnlogger我的设计思路生产和开发用不同的格式——生产环境用 JSON 便于分析开发环境用彩色格式便于阅读。3.2 JSON Lines 文件格式日志输出是这样的{datetime:2026-05-08T16:30:00.123Z,level:INFO,message:request_started,request_id:a1b2c3d4-e5f6-7890-abcd-ef1234567890,method:POST,path:/add_record,ip:127.0.0.1}{datetime:2026-05-08T16:30:00.200Z,level:INFO,message:request_completed,request_id:a1b2c3d4-e5f6-7890-abcd-ef1234567890,status_code:302,duration_ms:77.45}每行都是独立的 JSON 对象这就是 JSON Lines 格式。好处是什么可以直接用cat logs/app_20260508.jsonl | jq查询甚至直接丢给 pandas 分析。四、请求链路追踪request_id 如何串联所有日志4.1 为什么需要 request_id一个常见的场景用户反馈我提交的订单没反应。你要查日志发现有 20 条相关记录但你不知道哪条是哪条请求的。这就是没有 request_id 的痛苦。我的解决方案每个请求进来自动生成一个 UUID 作为 request_id挂在 Flask 的g对象上全程传递。4.2 中间件实现# logging/middleware.pyfromflaskimportg,requestimportuuidfromdatetimeimportdatetimefrom.configimportget_loggerdefinit_middleware(app):注册请求中间件到 Flask appapp.before_requestdefbefore_request():# 生成请求IDg.request_idstr(uuid.uuid4())g.start_timedatetime.now()loggerget_logger()logger.info(request_started,extra{request_id:g.request_id,method:request.method,path:request.path,ip:request.remote_addr,user_agent:request.user_agent.string,endpoint:request.endpoint,})app.after_requestdefafter_request(response):# 计算耗时duration_ms(datetime.now()-g.start_time).total_seconds()*1000loggerget_logger()logger.info(request_completed,extra{request_id:g.request_id,status_code:response.status_code,duration_ms:round(duration_ms,2),method:request.method,path:request.path,})# 把 request_id 返回给客户端方便排查response.headers[X-Request-ID]g.request_idreturnresponseapp.errorhandler(Exception)defhandle_exception(e):duration_ms(datetime.now()-g.start_time).total_seconds()*1000loggerget_logger()logger.error(request_error,extra{request_id:g.request_id,error_type:type(e).__name__,error_message:str(e),duration_ms:round(duration_ms,2),method:request.method,path:request.path,})return{error:str(e)},500关键点before_request生成 request_id 并记录请求开始after_request计算耗时并记录请求结束异常被errorhandler捕获保证不会漏记X-Request-IDheader 返回给客户端用户可以提供这个 ID 给我们排查4.3 效果验证现在查日志超简单# 查询某个请求的所有日志catlogs/app_20260508.jsonl|jqselect(.request_id a1b2c3d4-e5f6-7890-abcd-ef1234567890)# 统计错误率catlogs/app_20260508.jsonl|jq-smap(select(.level ERROR)) | length五、装饰器模式用 log_operation 简化业务日志5.1 为什么需要业务日志装饰器很多团队只在 HTTP 层记录日志但这样不够。比如add_record()函数内部发生了什么参数是什么返回值是什么耗时多少这些信息 HTTP 层日志是给不了的。我的解决方案写一个log_operation装饰器给任何函数加上日志能力。5.2 装饰器实现# logging/decorator.pyfromfunctoolsimportwrapsfromflaskimportgimporttimeimporttracebackfrom.configimportget_loggerdeflog_operation(operation:str,module:str,log_input:boolTrue,log_output:boolTrue): 装饰器为关键操作添加日志 用法 log_operation(add_record, SampleTable) def add_record(): ... defdecorator(func):wraps(func)defwrapper(*args,**kwargs):loggerget_logger()start_timetime.time()# 获取 request_id如果存在request_idgetattr(g,request_id,no-request)# 构造日志基础信息log_extra{request_id:request_id,operation:operation,module:module,}# 记录输入参数iflog_input:log_extra[input]{args:str(args)[1:-1],kwargs:{k:str(v)fork,vinkwargs.items()}}try:# 执行原函数resultfunc(*args,**kwargs)# 记录输出iflog_output:log_extra[output]str(result)[:500]log_extra[status]successlog_extra[level]INFOduration_ms(time.time()-start_time)*1000log_extra[duration_ms]round(duration_ms,2)logger.info(foperation_success:{operation},extralog_extra)returnresultexceptExceptionase:# 记录错误duration_ms(time.time()-start_time)*1000log_extra.update({status:error,level:ERROR,duration_ms:round(duration_ms,2),error_type:type(e).__name__,error_detail:traceback.format_exc(),})logger.error(foperation_error:{operation},extralog_extra)raisereturnwrapperreturndecorator5.3 使用示例在业务代码中这样用# views.pyfromapplication.loggingimportlog_operationapp.route(/add_record,methods[GET,POST])log_operation(add_record,SampleTable)defadd_record():formRecordForm()ifform.validate_on_submit():modelSampleTable()model.add_data(titleform.title.data,descriptionform.description.data)flash(记录已添加,success)returnredirect(url_for(index))returnrender_template(add_record.html,formform)我的感受是这种装饰器模式的好处在于关注点分离——业务逻辑只管业务逻辑日志自动加上非常干净。六、双写策略文件 数据库各自的优势6.1 为什么需要双写单一存储有两个问题存储方式优点缺点文件简单、可靠、写入快查询不便难以聚合数据库查询方便、可聚合占用数据库资源、需要清理我的方案两者都要。文件用于实时查看和归档数据库用于查询和统计。6.2 OperationLog 数据模型# models.pyclassOperationLog(db.Model):操作日志模型__tablename__operation_logsiddb.Column(db.Integer,primary_keyTrue)request_iddb.Column(db.String(36),nullableFalse,indexTrue)operationdb.Column(db.String(100),nullableFalse,indexTrue)moduledb.Column(db.String(100),nullableFalse,indexTrue)leveldb.Column(db.String(10),nullableFalse)statusdb.Column(db.String(10),nullableFalse)duration_msdb.Column(db.Float)input_datadb.Column(db.Text)# JSONoutput_datadb.Column(db.Text)# JSONerror_detaildb.Column(db.Text)ip_addressdb.Column(db.String(45))user_agentdb.Column(db.String(500))endpointdb.Column(db.String(200))methoddb.Column(db.String(10))created_atdb.Column(db.DateTime,defaultdatetime.datetime.now,indexTrue)关键索引设计CREATEINDEXidx_operation_logs_request_idONoperation_logs(request_id);CREATEINDEXidx_operation_logs_operationONoperation_logs(operation);CREATEINDEXidx_operation_logs_created_atONoperation_logs(created_at);6.3 AI 排查场景示例有了结构化日志排查问题变得超级简单# 查询某操作近1小时错误日志logsOperationLog.query.filter(OperationLog.operationadd_record,OperationLog.statuserror,OperationLog.created_atdatetime.datetime.now()-timedelta(hours1)).all()# 输出报告{summary:过去1小时 add_record 操作成功 45 次失败 3 次,error_rate:6.5%,errors:[{request_id:a1b2c3d4-...,time:2026-05-08T16:30:00,error_type:IntegrityError,error:UNIQUE constraint failed,endpoint:/add_record}]}七、日志轮转与自动清理7.1 文件日志轮转Loguru 自带轮转功能配置超简单logger.add(f{log_dir}/app_{{time:YYYYMMDD}}.jsonl,rotation00:00,# 每天零点创建新文件retention30 days,# 保留30天compressionzip,# 旧日志压缩)7.2 数据库日志清理# 清理旧日志defcleanup_old_logs(days:int30):清理超过指定天数的日志fromdatetimeimportdatetime,timedelta cutoffdatetime.now()-timedelta(daysdays)OperationLog.query.filter(OperationLog.created_atcutoff).delete()db.session.commit()建议通过 Flask CLI command 或定时任务如 APScheduler执行。7.3 日志清理计划存储清理策略执行方式文件日志30天自动删除 zip压缩Loguru 自动数据库日志30天自动清理Flask CLI 每日执行八、生产环境避坑指南8.1 敏感信息泄露这是最容易踩的坑。很多团队日志里直接写密码、token、身份证号等于把敏感信息公开了。我的脱敏方案# decorator.py 中增加脱敏处理SENSITIVE_FIELDS{password,token,secret,api_key,authorization}defsanitize_kwargs(kwargs):脱敏敏感字段sanitized{}fork,vinkwargs.items():ifk.lower()inSENSITIVE_FIELDS:sanitized[k]***REDACTED***else:sanitized[k]vreturnsanitized8.2 日志级别配置生产环境和开发环境要区分# config.pyclassConfig:# 开发环境LOG_LEVELDEBUG# 生产环境classProduction(Config):LOG_LEVELINFO# 减少日志量8.3 性能注意事项问题解决方案高并发写入使用enqueueTrue异步写入日志写入阻塞Loguru 默认异步不用担心大日志体输出截断我的装饰器里限制了 500 字符8.4 磁盘满的容错磁盘满了怎么办不能直接崩溃吧try:logger.info(operation,extra{...})exceptOSError:# 优雅降级不阻断业务pass九、完整目录结构InsMatrixAutomation/ ├── application/ │ ├── __init__.py # 初始化日志模块 │ ├── logging/ # 【新】日志模块 │ │ ├── __init__.py │ │ ├── config.py # 日志配置 │ │ ├── middleware.py # 请求中间件 │ │ └── decorator.py # log_operation 装饰器 │ ├── models.py # OperationLog 模型 │ └── views.py # log_operation │ ├── logs/ # 【新】日志输出目录 │ └── app_20260508.jsonl │ ├── tests/ # 【新】测试目录 │ ├── unit/ │ │ ├── test_config.py │ │ ├── test_decorator.py │ │ └── test_models.py │ └── integration/ │ ├── test_middleware.py │ └── test_json_output.py │ └── run.py十、总结维度内容核心思路Loguru 零配置 request_id 链路追踪 log_operation 业务装饰器日志格式JSON Lines兼顾可读性和可分析性持久化文件 数据库双写各取所长轮转策略文件每天轮转、30天保留数据库每日清理30天前数据安全考虑敏感字段脱敏、日志级别区分、磁盘满容错写在最后日志系统看起来不起眼但它是线上问题的黑匣子。我的经验是宁可少写一个功能也要先把日志做好。一个好的日志系统让你在凌晨三点被叫醒时能在 5 分钟内定位问题而不是对着日志发呆到天亮。我是张大鹏专注 AI 全栈教育培训。如果你对日志系统设计有任何问题欢迎评论区交流。参考资料Loguru 官方文档Flask Logging Best PracticesOWASP API Security Top 10作者张大鹏团队大鹏 AI 教育日期2026-05-09