Merge pull request emrgnt-cmplxty#56 from maks-ivanov/feature/beautif…

…y-logging

Feature/beautify logging

automata/core/agents/automata_agent.py

@@ -22,7 +22,7 @@
 
  agent.run()
 
- TODO - Add error checking to ensure that we don't terminate when 
+ TODO - Add error checking to ensure that we don't terminate when
  our previous result returned an error
 """
 import logging
@@ -33,6 +33,7 @@
 import openai
 import yaml
 from pydantic import BaseModel
+from termcolor import colored
 from transformers import GPT2Tokenizer
 
 from automata.config import * # noqa F403
@@ -193,6 +194,7 @@ def iter_task(
  stream=self.stream,
  )
  if self.stream:
+ print(colored("\n>>>", "green", attrs=["blink"]) + colored(" Agent:", "green"))
  accumulated_output = ""
  separator = " "
  response_text = ""
@@ -206,26 +208,25 @@ def iter_task(
  words = accumulated_output.split(separator)
  # Print all words except the last one, as it may be an incomplete word
  for word in words[:-1]:
- print(word, end=" ", flush=True)
+ print(colored(str(word), "green"), end=" ", flush=True)
  # Keep the last (potentially incomplete) word for the next iteration
  accumulated_output = words[-1]
+ # Print the last word
+ print(colored(str(accumulated_output), "green"))
  else:
  response_text = response_summary["choices"][0]["message"]["content"]
-
- if self.verbose:
- logger.info("OpenAI Response:\n%s\n" % response_text)
+ logger.debug("OpenAI Response:\n%s\n" % response_text)
  processed_inputs = self._process_input(response_text)
  self._save_interaction({"role": "assistant", "content": response_text})
 
  # If there are processed inputs, return here
  if len(processed_inputs) > 0:
- message = "Observation:\n{" + "\n"
+ message = "{" + "\n"
  for i, output in enumerate(processed_inputs):
- message += f'"output_{i}": "{(output)}", \n'
+ message += f'"output_{i}": {(output)}, \n'
  message += "}"
  self._save_interaction({"role": "user", "content": message})
- if self.verbose:
- logger.info("Synthetic User Message:\n%s\n" % message)
+ logger.debug("Synthetic User Message:\n%s\n" % message)
  return processed_inputs
 
  # If there are no outputs, then the user has must respond to continue
@@ -237,7 +238,6 @@ def iter_task(
 
  def run(self) -> str:
  """Run until the initial instruction terminates."""
-
  while True:
  self.iter_task()
  # Check the previous previous agent message to see if it is a completion message
@@ -250,18 +250,16 @@ def run(self) -> str:
  def replay_messages(self) -> str:
  """Replay the messages in the conversation."""
  if len(self.messages) == 0:
- if self.verbose:
- logger.info("No messages to replay.")
+ logger.debug("No messages to replay.")
  return "No messages to replay."
  for message in self.messages[1:]:
  if AutomataAgent.is_completion_message(message["content"]):
  return message["content"]
  processed_outputs = self._process_input(message["content"])
- if self.verbose:
- logger.info("Role:\n%s\n\nMessage:\n%s\n" % (message["role"], message["content"]))
- logger.info("Processing message content = %s" % (message["content"]))
- logger.info("\nProcessed Outputs:\n%s\n" % processed_outputs)
- logger.info("-" * 100)
+ logger.debug("Role:\n%s\n\nMessage:\n%s\n" % (message["role"], message["content"]))
+ logger.debug("Processing message content = %s" % (message["content"]))
+ logger.debug("\nProcessed Outputs:\n%s\n" % processed_outputs)
+ logger.debug("-" * 60)
  return "No completion message found."
 
  def modify_last_instruction(self, new_instruction: str) -> None:
@@ -332,7 +330,7 @@ def _load_prompt(self) -> str:
  def _process_input(self, response_text: str):
  """Process the messages in the conversation."""
  tool_calls = AutomataAgent._parse_input_string(response_text)
- logger.info("Tool Calls: %s" % tool_calls)
+ logger.debug("Tool Calls: %s" % tool_calls)
  outputs = []
  for tool_request in tool_calls:
  requested_tool, requested_tool_input = (

automata/core/utils.py

@@ -5,6 +5,7 @@
 import os
 from typing import Any, List
 
+import colorlog
 import numpy as np
 import openai
 import yaml
@@ -66,16 +67,27 @@ def format_config_path(config_dir: str, config_path: str) -> str:
 
 def get_logging_config(log_level=logging.INFO) -> dict:
  """Returns logging configuration."""
+ color_scheme = {
+ "DEBUG": "cyan",
+ "INFO": "green",
+ "WARNING": "yellow",
+ "ERROR": "red",
+ "CRITICAL": "bold_red",
+ }
  logging_config = {
  "version": 1,
  "disable_existing_loggers": False,
  "formatters": {
- "default": {"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"}
+ "colored": {
+ "()": colorlog.ColoredFormatter,
+ "format": "%(log_color)s%(levelname)s:%(name)s:%(message)s",
+ "log_colors": color_scheme,
+ }
  },
  "handlers": {
  "console": {
  "class": "logging.StreamHandler",
- "formatter": "default",
+ "formatter": "colored",
  "level": log_level,
  }
  },

automata/scripts/main_automata.py

@@ -3,6 +3,8 @@
 import logging.config
 from typing import Dict
 
+from termcolor import colored
+
 from automata.configs.agent_configs import AutomataConfigVersion
 from automata.core import Toolkit, ToolkitType, load_llm_toolkits
 from automata.core.agents.automata_agent import AutomataAgentBuilder, AutomataAgentConfig
@@ -51,7 +53,21 @@ def main():
  help="Should the instruction prompt include an overview?",
  )
 
+ parser.add_argument("-v", "--verbose", action="store_true", help="increase output verbosity")
+
  args = parser.parse_args()
+
+ logging_config = get_logging_config(log_level=logging.DEBUG if args.verbose else logging.INFO)
+ logging.config.dictConfig(logging_config)
+ logger = logging.getLogger(__name__)
+
+ requests_logger = logging.getLogger("urllib3")
+
+ # Set the logging level for the requests logger to WARNING
+ requests_logger.setLevel(logging.INFO)
+ openai_logger = logging.getLogger("openai")
+ openai_logger.setLevel(logging.INFO)
+
  assert not (
  args.instructions is None and args.session_id is None
  ), "You must provide instructions for the agent if you are not providing a session_id."
@@ -71,8 +87,10 @@ def main():
  }
  else:
  initial_payload = {}
- logger.info("Passing in instructions: %s", args.instructions)
- logger.info("-" * 100)
+ logger.info(
+ f"Passing in instructions:\n{colored(args.instructions, color='white', on_color='on_green')}"
+ )
+ logger.info("-" * 60)
 
  agent_config_version = AutomataConfigVersion(args.config_version)
  agent_config = AutomataAgentConfig.load(agent_config_version)
@@ -90,7 +108,6 @@ def main():
 
  logger.info("Running the agent now...")
  if args.session_id is None:
- logger.info("Running...")
  result = agent.run()
  logger.info("Result: %s", result)
  else:

automata/tools/documentation_tools/documentation_gpt.py

@@ -39,7 +39,9 @@ def __init__(self, url, model="gpt-3.5-turbo", temperature=0.7, verbose=False):
  retriever=self.vector_store.as_retriever(),
  return_source_documents=self.verbose,
  )
- logging_config = get_logging_config()
+ logging_config = get_logging_config(
+ log_level=logging.DEBUG if self.verbose else logging.INFO
+ )
  logging.config.dictConfig(logging_config)
  self.logger = logging.getLogger(__name__)
 

automata/tools/tool_management/python_writer_tool_manager.py

@@ -7,7 +7,7 @@
 Attributes:
 - writer (PythonWriter): A PythonWriter object for manipulating local pythonf iles.
 
-Example - 
+Example -
  python_indexer = PythonIndexer(root_py_path())
  python_writer = PythonWriter(python_indexer)
  python_writer_tool_manager = PythonWriterToolManager(python_writer)

automata/tools/tool_management/tests/test_python_writer_tool_manager.py

@@ -121,6 +121,7 @@ def test_extend_module_with_documented_new_class(python_writer_tool_builder):
  class_str = textwrap.dedent(
  '''from typing import List
 
+
 from automata.core.base.tool import Tool
 
 from automata.tools.python_tools.python_agent import PythonAgent