import logging import os import langchain import tempfile import shutil import time from functools import wraps # 是否显示详细日志 log_verbose = False langchain.verbose = False # 通常情况下不需要更改以下内容 # 日志格式 LOG_FORMAT = "%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s" logger = logging.getLogger() logger.setLevel(logging.INFO) logging.basicConfig(format=LOG_FORMAT) # 日志存储路径 LOG_PATH = os.path.join(os.path.dirname(os.path.dirname(__file__)), "logs") if not os.path.exists(LOG_PATH): os.mkdir(LOG_PATH) # 临时文件目录,主要用于文件对话 BASE_TEMP_DIR = os.path.join(tempfile.gettempdir(), "chatchat") try: shutil.rmtree(BASE_TEMP_DIR) except Exception: pass os.makedirs(BASE_TEMP_DIR, exist_ok=True) # 计时器 import asyncio def timing_decorator(func): if asyncio.iscoroutinefunction(func): @wraps(func) async def async_wrapper(*args, **kwargs): start_time = time.time() logging.info(f"开始执行异步函数: {func.__name__}") result = await func(*args, **kwargs) end_time = time.time() elapsed = end_time - start_time logging.info(f"🕐异步函数 {func.__name__} 执行耗时: {elapsed:.2f} 秒") return result return async_wrapper else: @wraps(func) def sync_wrapper(*args, **kwargs): start_time = time.time() logging.info(f"开始执行函数: {func.__name__}") result = func(*args, **kwargs) end_time = time.time() elapsed = end_time - start_time logging.info(f"🕐函数 {func.__name__} 执行耗时: {elapsed:.2f} 秒") return result return sync_wrapper