coding-alt/AutoGPT
0
1"""Logging module for Auto-GPT."""2import json3import logging4import os5import random6import re7import time8import traceback9from logging import LogRecord10 11from colorama import Fore, Style12 13from autogpt.config import Config, Singleton14from autogpt.speech import say_text15 16CFG = Config()17 18 19class Logger(metaclass=Singleton):20 """21 Logger that handle titles in different colors.22 Outputs logs in console, activity.log, and errors.log23 For console handler: simulates typing24 """25 26 def __init__(self):27 # create log directory if it doesn't exist28 this_files_dir_path = os.path.dirname(__file__)29 log_dir = os.path.join(this_files_dir_path, "../logs")30 if not os.path.exists(log_dir):31 os.makedirs(log_dir)32 33 log_file = "activity.log"34 error_file = "error.log"35 36 console_formatter = AutoGptFormatter("%(title_color)s %(message)s")37 38 # Create a handler for console which simulate typing39 self.typing_console_handler = TypingConsoleHandler()40 self.typing_console_handler.setLevel(logging.INFO)41 self.typing_console_handler.setFormatter(console_formatter)42 43 # Create a handler for console without typing simulation44 self.console_handler = ConsoleHandler()45 self.console_handler.setLevel(logging.DEBUG)46 self.console_handler.setFormatter(console_formatter)47 48 # Info handler in activity.log49 self.file_handler = logging.FileHandler(50 os.path.join(log_dir, log_file), "a", "utf-8"51 )52 self.file_handler.setLevel(logging.DEBUG)53 info_formatter = AutoGptFormatter(54 "%(asctime)s %(levelname)s %(title)s %(message_no_color)s"55 )56 self.file_handler.setFormatter(info_formatter)57 58 # Error handler error.log59 error_handler = logging.FileHandler(60 os.path.join(log_dir, error_file), "a", "utf-8"61 )62 error_handler.setLevel(logging.ERROR)63 error_formatter = AutoGptFormatter(64 "%(asctime)s %(levelname)s %(module)s:%(funcName)s:%(lineno)d %(title)s"65 " %(message_no_color)s"66 )67 error_handler.setFormatter(error_formatter)68 69 self.typing_logger = logging.getLogger("TYPER")70 self.typing_logger.addHandler(self.typing_console_handler)71 self.typing_logger.addHandler(self.file_handler)72 self.typing_logger.addHandler(error_handler)73 self.typing_logger.setLevel(logging.DEBUG)74 75 self.logger = logging.getLogger("LOGGER")76 self.logger.addHandler(self.console_handler)77 self.logger.addHandler(self.file_handler)78 self.logger.addHandler(error_handler)79 self.logger.setLevel(logging.DEBUG)80 81 def typewriter_log(82 self, title="", title_color="", content="", speak_text=False, level=logging.INFO83 ):84 if speak_text and CFG.speak_mode:85 say_text(f"{title}. {content}")86 87 if content:88 if isinstance(content, list):89 content = " ".join(content)90 else:91 content = ""92 93 self.typing_logger.log(94 level, content, extra={"title": title, "color": title_color}95 )96 97 def debug(98 self,99 message,100 title="",101 title_color="",102 ):103 self._log(title, title_color, message, logging.DEBUG)104 105 def warn(106 self,107 message,108 title="",109 title_color="",110 ):111 self._log(title, title_color, message, logging.WARN)112 113 def error(self, title, message=""):114 self._log(title, Fore.RED, message, logging.ERROR)115 116 def _log(self, title="", title_color="", message="", level=logging.INFO):117 if message:118 if isinstance(message, list):119 message = " ".join(message)120 self.logger.log(level, message, extra={"title": title, "color": title_color})121 122 def set_level(self, level):123 self.logger.setLevel(level)124 self.typing_logger.setLevel(level)125 126 def double_check(self, additionalText=None):127 if not additionalText:128 additionalText = (129 "Please ensure you've setup and configured everything"130 " correctly. Read https://github.com/Torantulino/Auto-GPT#readme to "131 "double check. You can also create a github issue or join the discord"132 " and ask there!"133 )134 135 self.typewriter_log("DOUBLE CHECK CONFIGURATION", Fore.YELLOW, additionalText)136 137 138"""139Output stream to console using simulated typing140"""141 142 143class TypingConsoleHandler(logging.StreamHandler):144 def emit(self, record):145 min_typing_speed = 0.05146 max_typing_speed = 0.01147 148 msg = self.format(record)149 try:150 words = msg.split()151 for i, word in enumerate(words):152 print(word, end="", flush=True)153 if i < len(words) - 1:154 print(" ", end="", flush=True)155 typing_speed = random.uniform(min_typing_speed, max_typing_speed)156 time.sleep(typing_speed)157 # type faster after each word158 min_typing_speed = min_typing_speed * 0.95159 max_typing_speed = max_typing_speed * 0.95160 print()161 except Exception:162 self.handleError(record)163 164 165class ConsoleHandler(logging.StreamHandler):166 def emit(self, record) -> None:167 msg = self.format(record)168 try:169 print(msg)170 except Exception:171 self.handleError(record)172 173 174class AutoGptFormatter(logging.Formatter):175 """176 Allows to handle custom placeholders 'title_color' and 'message_no_color'.177 To use this formatter, make sure to pass 'color', 'title' as log extras.178 """179 180 def format(self, record: LogRecord) -> str:181 if hasattr(record, "color"):182 record.title_color = (183 getattr(record, "color")184 + getattr(record, "title")185 + " "186 + Style.RESET_ALL187 )188 else:189 record.title_color = getattr(record, "title")190 if hasattr(record, "msg"):191 record.message_no_color = remove_color_codes(getattr(record, "msg"))192 else:193 record.message_no_color = ""194 return super().format(record)195 196 197def remove_color_codes(s: str) -> str:198 ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")199 return ansi_escape.sub("", s)200 201 202logger = Logger()203 204 205def print_assistant_thoughts(ai_name, assistant_reply):206 """Prints the assistant's thoughts to the console"""207 from autogpt.json_utils.json_fix_llm import (208 attempt_to_fix_json_by_finding_outermost_brackets,209 fix_and_parse_json,210 )211 212 try:213 try:214 # Parse and print Assistant response215 assistant_reply_json = fix_and_parse_json(assistant_reply)216 except json.JSONDecodeError:217 logger.error("Error: Invalid JSON in assistant thoughts\n", assistant_reply)218 assistant_reply_json = attempt_to_fix_json_by_finding_outermost_brackets(219 assistant_reply220 )221 if isinstance(assistant_reply_json, str):222 assistant_reply_json = fix_and_parse_json(assistant_reply_json)223 224 # Check if assistant_reply_json is a string and attempt to parse225 # it into a JSON object226 if isinstance(assistant_reply_json, str):227 try:228 assistant_reply_json = json.loads(assistant_reply_json)229 except json.JSONDecodeError:230 logger.error("Error: Invalid JSON\n", assistant_reply)231 assistant_reply_json = (232 attempt_to_fix_json_by_finding_outermost_brackets(233 assistant_reply_json234 )235 )236 237 assistant_thoughts_reasoning = None238 assistant_thoughts_plan = None239 assistant_thoughts_speak = None240 assistant_thoughts_criticism = None241 if not isinstance(assistant_reply_json, dict):242 assistant_reply_json = {}243 assistant_thoughts = assistant_reply_json.get("thoughts", {})244 assistant_thoughts_text = assistant_thoughts.get("text")245 246 if assistant_thoughts:247 assistant_thoughts_reasoning = assistant_thoughts.get("reasoning")248 assistant_thoughts_plan = assistant_thoughts.get("plan")249 assistant_thoughts_criticism = assistant_thoughts.get("criticism")250 assistant_thoughts_speak = assistant_thoughts.get("speak")251 252 logger.typewriter_log(253 f"{ai_name.upper()} THOUGHTS:", Fore.YELLOW, f"{assistant_thoughts_text}"254 )255 logger.typewriter_log(256 "REASONING:", Fore.YELLOW, f"{assistant_thoughts_reasoning}"257 )258 259 if assistant_thoughts_plan:260 logger.typewriter_log("PLAN:", Fore.YELLOW, "")261 # If it's a list, join it into a string262 if isinstance(assistant_thoughts_plan, list):263 assistant_thoughts_plan = "\n".join(assistant_thoughts_plan)264 elif isinstance(assistant_thoughts_plan, dict):265 assistant_thoughts_plan = str(assistant_thoughts_plan)266 267 # Split the input_string using the newline character and dashes268 lines = assistant_thoughts_plan.split("\n")269 for line in lines:270 line = line.lstrip("- ")271 logger.typewriter_log("- ", Fore.GREEN, line.strip())272 273 logger.typewriter_log(274 "CRITICISM:", Fore.YELLOW, f"{assistant_thoughts_criticism}"275 )276 # Speak the assistant's thoughts277 if CFG.speak_mode and assistant_thoughts_speak:278 say_text(assistant_thoughts_speak)279 else:280 logger.typewriter_log("SPEAK:", Fore.YELLOW, f"{assistant_thoughts_speak}")281 282 return assistant_reply_json283 except json.decoder.JSONDecodeError:284 logger.error("Error: Invalid JSON\n", assistant_reply)285 if CFG.speak_mode:286 say_text(287 "I have received an invalid JSON response from the OpenAI API."288 " I cannot ignore this response."289 )290 291 # All other errors, return "Error: + error message"292 except Exception:293 call_stack = traceback.format_exc()294 logger.error("Error: \n", call_stack)295 296 297def print_assistant_thoughts(298 ai_name: object, assistant_reply_json_valid: object299) -> None:300 assistant_thoughts_reasoning = None301 assistant_thoughts_plan = None302 assistant_thoughts_speak = None303 assistant_thoughts_criticism = None304 305 assistant_thoughts = assistant_reply_json_valid.get("thoughts", {})306 assistant_thoughts_text = assistant_thoughts.get("text")307 if assistant_thoughts:308 assistant_thoughts_reasoning = assistant_thoughts.get("reasoning")309 assistant_thoughts_plan = assistant_thoughts.get("plan")310 assistant_thoughts_criticism = assistant_thoughts.get("criticism")311 assistant_thoughts_speak = assistant_thoughts.get("speak")312 logger.typewriter_log(313 f"{ai_name.upper()} THOUGHTS:", Fore.YELLOW, f"{assistant_thoughts_text}"314 )315 logger.typewriter_log("REASONING:", Fore.YELLOW, f"{assistant_thoughts_reasoning}")316 if assistant_thoughts_plan:317 logger.typewriter_log("PLAN:", Fore.YELLOW, "")318 # If it's a list, join it into a string319 if isinstance(assistant_thoughts_plan, list):320 assistant_thoughts_plan = "\n".join(assistant_thoughts_plan)321 elif isinstance(assistant_thoughts_plan, dict):322 assistant_thoughts_plan = str(assistant_thoughts_plan)323 324 # Split the input_string using the newline character and dashes325 lines = assistant_thoughts_plan.split("\n")326 for line in lines:327 line = line.lstrip("- ")328 logger.typewriter_log("- ", Fore.GREEN, line.strip())329 logger.typewriter_log("CRITICISM:", Fore.YELLOW, f"{assistant_thoughts_criticism}")330 # Speak the assistant's thoughts331 if CFG.speak_mode and assistant_thoughts_speak:332 say_text(assistant_thoughts_speak)333