mirror of
https://github.com/OpenHands/OpenHands.git
synced 2025-12-26 13:52:43 +08:00
349 lines
11 KiB
Python
349 lines
11 KiB
Python
import copy
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
import traceback
|
|
from datetime import datetime
|
|
from typing import Literal, Mapping
|
|
|
|
from termcolor import colored
|
|
|
|
LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO').upper()
|
|
DEBUG = os.getenv('DEBUG', 'False').lower() in ['true', '1', 'yes']
|
|
if DEBUG:
|
|
LOG_LEVEL = 'DEBUG'
|
|
|
|
LOG_TO_FILE = os.getenv('LOG_TO_FILE', 'False').lower() in ['true', '1', 'yes']
|
|
DISABLE_COLOR_PRINTING = False
|
|
|
|
ColorType = Literal[
|
|
'red',
|
|
'green',
|
|
'yellow',
|
|
'blue',
|
|
'magenta',
|
|
'cyan',
|
|
'light_grey',
|
|
'dark_grey',
|
|
'light_red',
|
|
'light_green',
|
|
'light_yellow',
|
|
'light_blue',
|
|
'light_magenta',
|
|
'light_cyan',
|
|
'white',
|
|
]
|
|
|
|
LOG_COLORS: Mapping[str, ColorType] = {
|
|
'ACTION': 'green',
|
|
'USER_ACTION': 'light_red',
|
|
'OBSERVATION': 'yellow',
|
|
'USER_OBSERVATION': 'light_green',
|
|
'DETAIL': 'cyan',
|
|
'ERROR': 'red',
|
|
'PLAN': 'light_magenta',
|
|
}
|
|
|
|
|
|
class NoColorFormatter(logging.Formatter):
|
|
"""Formatter for non-colored logging in files."""
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
# Create a deep copy of the record to avoid modifying the original
|
|
new_record: logging.LogRecord = copy.deepcopy(record)
|
|
# Strip ANSI color codes from the message
|
|
new_record.msg = strip_ansi(new_record.msg)
|
|
|
|
return super().format(new_record)
|
|
|
|
|
|
def strip_ansi(s: str) -> str:
|
|
"""
|
|
Removes ANSI escape sequences from str, as defined by ECMA-048 in
|
|
http://www.ecma-international.org/publications/files/ECMA-ST/Ecma-048.pdf
|
|
# https://github.com/ewen-lbh/python-strip-ansi/blob/master/strip_ansi/__init__.py
|
|
"""
|
|
pattern = re.compile(r'\x1B\[\d+(;\d+){0,2}m')
|
|
stripped = pattern.sub('', s)
|
|
return stripped
|
|
|
|
|
|
class ColoredFormatter(logging.Formatter):
|
|
def format(self, record):
|
|
msg_type = record.__dict__.get('msg_type')
|
|
event_source = record.__dict__.get('event_source')
|
|
if event_source:
|
|
new_msg_type = f'{event_source.upper()}_{msg_type}'
|
|
if new_msg_type in LOG_COLORS:
|
|
msg_type = new_msg_type
|
|
if msg_type in LOG_COLORS and not DISABLE_COLOR_PRINTING:
|
|
msg_type_color = colored(msg_type, LOG_COLORS[msg_type])
|
|
msg = colored(record.msg, LOG_COLORS[msg_type])
|
|
time_str = colored(
|
|
self.formatTime(record, self.datefmt), LOG_COLORS[msg_type]
|
|
)
|
|
name_str = colored(record.name, LOG_COLORS[msg_type])
|
|
level_str = colored(record.levelname, LOG_COLORS[msg_type])
|
|
if msg_type in ['ERROR'] or DEBUG:
|
|
return f'{time_str} - {name_str}:{level_str}: {record.filename}:{record.lineno}\n{msg_type_color}\n{msg}'
|
|
return f'{time_str} - {msg_type_color}\n{msg}'
|
|
elif msg_type == 'STEP':
|
|
msg = '\n\n==============\n' + record.msg + '\n'
|
|
return f'{msg}'
|
|
return super().format(record)
|
|
|
|
|
|
file_formatter = NoColorFormatter(
|
|
'%(asctime)s - %(name)s:%(levelname)s: %(filename)s:%(lineno)s - %(message)s',
|
|
datefmt='%H:%M:%S',
|
|
)
|
|
llm_formatter = logging.Formatter('%(message)s')
|
|
|
|
|
|
class RollingLogger:
|
|
max_lines: int
|
|
char_limit: int
|
|
log_lines: list[str]
|
|
|
|
def __init__(self, max_lines=10, char_limit=80):
|
|
self.max_lines = max_lines
|
|
self.char_limit = char_limit
|
|
self.log_lines = [''] * self.max_lines
|
|
|
|
def is_enabled(self):
|
|
return DEBUG and sys.stdout.isatty()
|
|
|
|
def start(self, message=''):
|
|
if message:
|
|
print(message)
|
|
self._write('\n' * self.max_lines)
|
|
self._flush()
|
|
|
|
def add_line(self, line):
|
|
self.log_lines.pop(0)
|
|
self.log_lines.append(line[: self.char_limit])
|
|
self.print_lines()
|
|
|
|
def write_immediately(self, line):
|
|
self._write(line)
|
|
self._flush()
|
|
|
|
def print_lines(self):
|
|
"""Display the last n log_lines in the console (not for file logging).
|
|
This will create the effect of a rolling display in the console.
|
|
"""
|
|
self.move_back()
|
|
for line in self.log_lines:
|
|
self.replace_current_line(line)
|
|
|
|
def move_back(self, amount=-1):
|
|
"""
|
|
'\033[F' moves the cursor up one line.
|
|
"""
|
|
if amount == -1:
|
|
amount = self.max_lines
|
|
self._write('\033[F' * (self.max_lines))
|
|
self._flush()
|
|
|
|
def replace_current_line(self, line=''):
|
|
"""
|
|
'\033[2K\r' clears the line and moves the cursor to the beginning of the line.
|
|
"""
|
|
self._write('\033[2K' + line + '\n')
|
|
self._flush()
|
|
|
|
def _write(self, line):
|
|
if not self.is_enabled():
|
|
return
|
|
sys.stdout.write(line)
|
|
|
|
def _flush(self):
|
|
if not self.is_enabled():
|
|
return
|
|
sys.stdout.flush()
|
|
|
|
|
|
class SensitiveDataFilter(logging.Filter):
|
|
def filter(self, record):
|
|
# start with attributes
|
|
sensitive_patterns = [
|
|
'api_key',
|
|
'aws_access_key_id',
|
|
'aws_secret_access_key',
|
|
'e2b_api_key',
|
|
'github_token',
|
|
'jwt_secret',
|
|
'modal_api_token_id',
|
|
'modal_api_token_secret',
|
|
]
|
|
|
|
# add env var names
|
|
env_vars = [attr.upper() for attr in sensitive_patterns]
|
|
sensitive_patterns.extend(env_vars)
|
|
|
|
# and some special cases
|
|
sensitive_patterns.append('JWT_SECRET')
|
|
sensitive_patterns.append('LLM_API_KEY')
|
|
sensitive_patterns.append('GITHUB_TOKEN')
|
|
sensitive_patterns.append('SANDBOX_ENV_GITHUB_TOKEN')
|
|
|
|
# this also formats the message with % args
|
|
msg = record.getMessage()
|
|
record.args = ()
|
|
|
|
for attr in sensitive_patterns:
|
|
pattern = rf"{attr}='?([\w-]+)'?"
|
|
msg = re.sub(pattern, f"{attr}='******'", msg)
|
|
|
|
# passed with msg
|
|
record.msg = msg
|
|
return True
|
|
|
|
|
|
def get_console_handler(log_level: int = logging.INFO, extra_info: str | None = None):
|
|
"""Returns a console handler for logging."""
|
|
console_handler = logging.StreamHandler()
|
|
console_handler.setLevel(log_level)
|
|
formatter_str = '\033[92m%(asctime)s - %(name)s:%(levelname)s\033[0m: %(filename)s:%(lineno)s - %(message)s'
|
|
if extra_info:
|
|
formatter_str = f'{extra_info} - ' + formatter_str
|
|
console_handler.setFormatter(ColoredFormatter(formatter_str, datefmt='%H:%M:%S'))
|
|
return console_handler
|
|
|
|
|
|
def get_file_handler(log_dir: str, log_level: int = logging.INFO):
|
|
"""Returns a file handler for logging."""
|
|
os.makedirs(log_dir, exist_ok=True)
|
|
timestamp = datetime.now().strftime('%Y-%m-%d')
|
|
file_name = f'openhands_{timestamp}.log'
|
|
file_handler = logging.FileHandler(os.path.join(log_dir, file_name))
|
|
file_handler.setLevel(log_level)
|
|
file_handler.setFormatter(file_formatter)
|
|
return file_handler
|
|
|
|
|
|
# Set up logging
|
|
logging.basicConfig(level=logging.ERROR)
|
|
|
|
|
|
def log_uncaught_exceptions(ex_cls, ex, tb):
|
|
"""Logs uncaught exceptions along with the traceback.
|
|
|
|
Args:
|
|
ex_cls (type): The type of the exception.
|
|
ex (Exception): The exception instance.
|
|
tb (traceback): The traceback object.
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
logging.error(''.join(traceback.format_tb(tb)))
|
|
logging.error('{0}: {1}'.format(ex_cls, ex))
|
|
|
|
|
|
sys.excepthook = log_uncaught_exceptions
|
|
openhands_logger = logging.getLogger('openhands')
|
|
current_log_level = logging.INFO
|
|
|
|
if LOG_LEVEL in logging.getLevelNamesMapping():
|
|
current_log_level = logging.getLevelNamesMapping()[LOG_LEVEL]
|
|
openhands_logger.setLevel(current_log_level)
|
|
|
|
if current_log_level == logging.DEBUG:
|
|
LOG_TO_FILE = True
|
|
openhands_logger.debug('DEBUG mode enabled.')
|
|
|
|
openhands_logger.addHandler(get_console_handler(current_log_level))
|
|
openhands_logger.addFilter(SensitiveDataFilter(openhands_logger.name))
|
|
openhands_logger.propagate = False
|
|
openhands_logger.debug('Logging initialized')
|
|
|
|
LOG_DIR = os.path.join(
|
|
# parent dir of openhands/core (i.e., root of the repo)
|
|
os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
|
|
'logs',
|
|
)
|
|
|
|
if LOG_TO_FILE:
|
|
openhands_logger.addHandler(
|
|
get_file_handler(LOG_DIR, current_log_level)
|
|
) # default log to project root
|
|
openhands_logger.debug(f'Logging to file in: {LOG_DIR}')
|
|
|
|
# Exclude LiteLLM from logging output
|
|
logging.getLogger('LiteLLM').disabled = True
|
|
logging.getLogger('LiteLLM Router').disabled = True
|
|
logging.getLogger('LiteLLM Proxy').disabled = True
|
|
|
|
|
|
class LlmFileHandler(logging.FileHandler):
|
|
"""# LLM prompt and response logging"""
|
|
|
|
def __init__(self, filename, mode='a', encoding='utf-8', delay=False):
|
|
"""Initializes an instance of LlmFileHandler.
|
|
|
|
Args:
|
|
filename (str): The name of the log file.
|
|
mode (str, optional): The file mode. Defaults to 'a'.
|
|
encoding (str, optional): The file encoding. Defaults to None.
|
|
delay (bool, optional): Whether to delay file opening. Defaults to False.
|
|
"""
|
|
self.filename = filename
|
|
self.message_counter = 1
|
|
if DEBUG:
|
|
self.session = datetime.now().strftime('%y-%m-%d_%H-%M')
|
|
else:
|
|
self.session = 'default'
|
|
self.log_directory = os.path.join(LOG_DIR, 'llm', self.session)
|
|
os.makedirs(self.log_directory, exist_ok=True)
|
|
if not DEBUG:
|
|
# Clear the log directory if not in debug mode
|
|
for file in os.listdir(self.log_directory):
|
|
file_path = os.path.join(self.log_directory, file)
|
|
try:
|
|
os.unlink(file_path)
|
|
except Exception as e:
|
|
openhands_logger.error(
|
|
'Failed to delete %s. Reason: %s', file_path, e
|
|
)
|
|
filename = f'{self.filename}_{self.message_counter:03}.log'
|
|
self.baseFilename = os.path.join(self.log_directory, filename)
|
|
super().__init__(self.baseFilename, mode, encoding, delay)
|
|
|
|
def emit(self, record):
|
|
"""Emits a log record.
|
|
|
|
Args:
|
|
record (logging.LogRecord): The log record to emit.
|
|
"""
|
|
filename = f'{self.filename}_{self.message_counter:03}.log'
|
|
self.baseFilename = os.path.join(self.log_directory, filename)
|
|
self.stream = self._open()
|
|
super().emit(record)
|
|
self.stream.close()
|
|
openhands_logger.debug('Logging to %s', self.baseFilename)
|
|
self.message_counter += 1
|
|
|
|
|
|
def _get_llm_file_handler(name: str, log_level: int):
|
|
# The 'delay' parameter, when set to True, postpones the opening of the log file
|
|
# until the first log message is emitted.
|
|
llm_file_handler = LlmFileHandler(name, delay=True)
|
|
llm_file_handler.setFormatter(llm_formatter)
|
|
llm_file_handler.setLevel(log_level)
|
|
return llm_file_handler
|
|
|
|
|
|
def _setup_llm_logger(name: str, log_level: int):
|
|
logger = logging.getLogger(name)
|
|
logger.propagate = False
|
|
logger.setLevel(log_level)
|
|
if LOG_TO_FILE:
|
|
logger.addHandler(_get_llm_file_handler(name, log_level))
|
|
return logger
|
|
|
|
|
|
llm_prompt_logger = _setup_llm_logger('prompt', current_log_level)
|
|
llm_response_logger = _setup_llm_logger('response', current_log_level)
|