| import random | |
| from typing import Optional | |
| from termcolor import COLORS, colored | |
| from lagent.utils import get_logger | |
| from .hook import Hook | |
| class MessageLogger(Hook): | |
| def __init__(self, name: str = 'lagent'): | |
| self.logger = get_logger( | |
| name, 'info', '%(asctime)s %(levelname)8s %(name)8s - %(message)s') | |
| self.sender2color = {} | |
| def before_agent(self, agent, messages, session_id): | |
| for message in messages: | |
| self._process_message(message, session_id) | |
| def after_agent(self, agent, message, session_id): | |
| self._process_message(message, session_id) | |
| def before_action(self, executor, message, session_id): | |
| self._process_message(message, session_id) | |
| def after_action(self, executor, message, session_id): | |
| self._process_message(message, session_id) | |
| def _process_message(self, message, session_id): | |
| sender = message.sender | |
| color = self.sender2color.setdefault(sender, | |
| random.choice(list(COLORS))) | |
| self.logger.info( | |
| colored( | |
| f'session id: {session_id}, message sender: {sender}\n' | |
| f'{message.content}', color)) | |