kg_agent/lib/my_trace.py

from datetime import datetime
import logging

def get_current_time():
    import uuid
    random_uuid = str(uuid.uuid4())
    return datetime.now().strftime("%Y%m%d/%H%M%S") + "_" + random_uuid[:2]

def set_trace_logfire():
    from agents.tracing.setup import GLOBAL_TRACE_PROVIDER
    GLOBAL_TRACE_PROVIDER.shutdown()
    import logfire
    current_time = get_current_time()
    logfire.configure(service_name=f'{current_time}')
    logfire.instrument_openai_agents()
    import urllib.parse
    current_time_encoded = urllib.parse.quote(current_time)
    import logging
    LOG_LEVEL = "WARNING"
    # 设置日志
    logging.basicConfig(
        level=getattr(logging, LOG_LEVEL),
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    logger = logging.getLogger(__name__)
    log_url = f'https://logfire-us.pydantic.dev/semsevens/test?q=service_name+%3D+%27{current_time_encoded}%27&last=30d'
    logger.warning(f"任务日志链接: {log_url}")
    return current_time, log_url

def set_trace():
    # 设置全局logging级别，覆盖所有子模块
    logging.basicConfig(level=logging.WARNING, force=True)
    # 确保根logger级别生效
    logging.getLogger().setLevel(logging.WARNING)

    # 临时绕过 logfire
    try:
        return set_trace_logfire()
    except Exception as e:
        current_time = get_current_time()
        logger = logging.getLogger(__name__)
        logger.warning(f"Logfire 不可用 ({e})，使用简单模式（时间戳: {current_time}）")
        return current_time, ""


def set_trace_smith():
    from agents.tracing import GLOBAL_TRACE_PROVIDER
    GLOBAL_TRACE_PROVIDER.shutdown()
    from agents import set_trace_processors
    from langsmith.wrappers import OpenAIAgentsTracingProcessor
    import logging
    current_time = get_current_time()
    set_trace_processors([OpenAIAgentsTracingProcessor(name=f'{current_time}')])
    import urllib.parse
    LOG_LEVEL = "WARNING"
    # 设置日志
    logging.basicConfig(
        level=getattr(logging, LOG_LEVEL),
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    logger = logging.getLogger(__name__)
    current_time_encoded = urllib.parse.quote(current_time)
    log_url = f'https://smith.langchain.com/o/3ebe0715-9709-4594-a0aa-40a77a4e10bd/projects/p/611fa0d6-5510-4f60-b693-87e2ccc2ea5f?timeModel=%7B%22duration%22%3A%227d%22%7D&searchModel=%7B%22filter%22%3A%22and%28eq%28is_root%2C+true%29%2C+eq%28name%2C+%5C%22{current_time_encoded}%5C%22%29%29%22%2C%22searchFilter%22%3A%22eq%28is_root%2C+true%29%22%7D'
    LOG_LEVEL = "WARNING"
    logger.warning(f"任务日志链接: {log_url}")
        
def set_debug():
    import logging
    # 设置全局日志级别为DEBUG，确保所有模块生效
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    # 确保根日志记录器也设置为DEBUG级别
    logging.getLogger().setLevel(logging.DEBUG)

def set_info():
    import logging
    # 设置全局日志级别为INFO，确保所有模块生效
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    # 确保根日志记录器也设置为INFO级别
    logging.getLogger().setLevel(logging.INFO)