You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何实现仅需初始化一次的CustomLogger,类同原生logging?

解决CustomLogger全局复用问题

问题现状

当前实现的CustomLogger需要在每个文件中重复实例化(传入tracker ID等参数)才能使用,无法像原生logging一样在一个入口文件初始化后,其他文件直接调用。

解决方案思路

利用Python模块的单例特性,将Logger实例封装为模块级别的全局对象,提供初始化函数和获取函数,确保整个程序中只有一个Logger实例,且只需在入口文件初始化一次。

修改后的代码实现

1. 重构CustomLogger.py

调整为全局实例模式,增加初始化和获取方法:

import json
import logging
import os

class StreamLogger(logging.Logger):
    def __init__(self, name, level=logging.DEBUG):
        super().__init__(name, level)
        streamHandler = logging.StreamHandler()
        formatter = logging.Formatter('[%(asctime)s] [%(funcName)s] %(message)s')
        streamHandler.setFormatter(formatter)
        self.addHandler(streamHandler)

# 全局Logger实例,初始为None
_global_logger = None

def setup(name, k_dict, level=logging.DEBUG):
    """初始化全局Logger,仅第一次调用有效"""
    global _global_logger
    if _global_logger is None:
        class Logger(logging.Logger):
            def __init__(self, name, k_dict, level=logging.DEBUG):
                super().__init__(name, level)
                self.k_dict = k_dict.copy()  # 复制字典避免外部修改影响日志
                self.json_logging = 'TRANSFORM_TO_JSON' in os.environ
                self.stream_logger = StreamLogger(name, level)

            def _transform_to_json(self, msg):
                msg_dict = self.k_dict.copy()
                msg_dict['message'] = msg
                return json.dumps(msg_dict)

            def info(self, msg, *args, **kwargs):
                if self.json_logging:
                    self.stream_logger.info(self._transform_to_json(msg), *args, **kwargs)
                else:
                    self.stream_logger.info(msg, *args, **kwargs)
        _global_logger = Logger(name, k_dict, level)
    return _global_logger

def get_logger():
    """获取已初始化的全局Logger"""
    if _global_logger is None:
        raise RuntimeError("Logger未初始化,请先调用setup()方法")
    return _global_logger

2. 修改入口文件test-executor.py

仅在入口处调用setup初始化Logger:

import CustomLogger
from CustomExecutor import iterative_executor

def main():
    # 初始化全局Logger
    logging = CustomLogger.setup('sample-run', {'cnt': 10})
    logging.info("Starting run")
    iterative_executor(1)
    logging.info("Done with execution")

if __name__ == '__main__':
    main()

3. 修改CustomExecutor.py

直接获取全局Logger实例,无需重复初始化:

from CustomLogger import get_logger

# 获取已初始化的全局Logger
logging = get_logger()

def iterative_executor(iter_cnt=2):
    for i in range(iter_cnt):
        logging.info(f"Current iteration cnt {i}")

关键优化点

  • 模块级变量_global_logger确保全局只有一个Logger实例,避免重复初始化
  • setup函数仅在第一次调用时创建实例,后续调用直接返回已有实例
  • 对k_dict进行复制,防止外部字典修改影响日志内容
  • 修复了原代码中json_transform函数名与调用不匹配的问题

内容的提问来源于stack exchange,提问作者madmatrix

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.21 11:00:10