Spaces:
Running
Running
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)) | |