Simple logger (#15)

agentlite/logging/BaseLogger.py

@@ -0,0 +1,41 @@
+import os
+
+from agentlite.commons import AgentAct, TaskPackage
+from agentlite.logging.utils import *
+from agentlite.utils import bcolors
+
+
+class BaseAgentLogger:
+    def __init__(
+        self,
+        log_file_name: str = "agent.log",
+    ) -> None:
+        self.log_file_name = log_file_name
+
+    def receive_task(self, task: TaskPackage, agent_name: str):
+        """the agent receives a task and log it"""
+        pass
+
+    def execute_task(self, task: TaskPackage = None, agent_name: str = None, **kwargs):
+        """the agent starts to execute the task"""
+        pass
+
+    def end_execute(self, task: TaskPackage, agent_name: str = None):
+        """the agent finishes the task"""
+        pass
+
+    def take_action(self, action: AgentAct, agent_name: str, step_idx: int):
+        """the agent takes an action"""
+        pass
+
+    def get_obs(self, obs: str):
+        """get observation"""
+        pass
+
+    def get_prompt(self, prompt):
+        """get prompt"""
+        pass
+
+    def get_llm_output(self, output: str):
+        """get llm output"""
+        pass

agentlite/logging/multi_agent_log.py

@@ -4,16 +4,17 @@
 from agentlite.logging.utils import *
 from agentlite.utils import bcolors
 
+from .BaseLogger import BaseAgentLogger
 
-class AgentLogger:
+class AgentLogger(BaseAgentLogger):
     def __init__(
         self,
         log_file_name: str = "agent.log",
         FLAG_PRINT: bool = True,
         OBS_OFFSET: int = 100,
         PROMPT_DEBUG_FLAG: bool = False,
     ) -> None:
-        self.log_file_name = log_file_name
+        super().__init__(log_file_name=log_file_name)
         self.FLAG_PRINT = FLAG_PRINT  # whether print the log into terminal
         self.OBS_OFFSET = OBS_OFFSET
         self.PROMPT_DEBUG_FLAG = PROMPT_DEBUG_FLAG
@@ -90,17 +91,4 @@ def get_prompt(self, prompt):
     def get_llm_output(self, output: str):
         log_str = f"""LLM generates: {self.__color_prompt_str__(output)}"""
         if self.PROMPT_DEBUG_FLAG:
-            self.__save_log__(log_str)
-
-
-if __name__ == "__main__":
-    aloger = AgentLogger()
-    agent_name = "labor_agent"
-    task_pack = TaskPackage(
-        instruction="this is a instruction",
-        task_creator="agent_1",
-        task_executor=f"{agent_name}",
-        completion=False,
-    )
-
-    aloger.receive_task(task_pack, agent_name=agent_name)
+            self.__save_log__(log_str)

agentlite/logging/streamlit_log.py

@@ -1,12 +1,15 @@
 import os
+import chainlit as cl
 import streamlit as st
 
 
 from agentlite.commons import AgentAct, TaskPackage
 from agentlite.logging.utils import *
 from agentlite.utils import bcolors
 
-class AgentLogger:
+from .BaseLogger import BaseAgentLogger
+
+class AgentLogger(BaseAgentLogger):
     def __init__(
         self,
         log_file_name: str = "agent.log",
@@ -19,21 +22,6 @@ def __init__(
         self.OBS_OFFSET = OBS_OFFSET
         self.PROMPT_DEBUG_FLAG = PROMPT_DEBUG_FLAG
 
-    def __color_agent_name__(self, agent_name: str):
-        return f"""{bcolors.OKBLUE}{agent_name}{bcolors.ENDC}"""
-
-    def __color_task_str__(self, task_str: str):
-        return f"""{bcolors.OKCYAN}{task_str}{bcolors.ENDC}"""
-
-    def __color_act_str__(self, act_str: str):
-        return f"""{bcolors.OKBLUE}{act_str}{bcolors.ENDC}"""
-
-    def __color_obs_str__(self, act_str: str):
-        return f"""{bcolors.OKGREEN}{act_str}{bcolors.ENDC}"""
-
-    def __color_prompt_str__(self, prompt: str):
-        return f"""{bcolors.WARNING}{prompt}{bcolors.ENDC}"""
-
     def __check_log_file__(self):
         if os.path.isdir(self.log_file_name):
             return True
@@ -47,14 +35,9 @@ def __save_log__(self, log_str: str):
         st.session_state.messages.append({"role": "assistant", "content": log_str})
 
     def receive_task(self, task: TaskPackage, agent_name: str):
-        task_str = (
-            f"""[\n\tTask ID: {task.task_id}\n\tInstruction: {task.instruction}\n]"""
-        )
         log_str = f"""Agent {agent_name} """
         log_str += f"""receives the following TaskPackage:\n"""
-        # log_str += f"{self.__color_task_str__(task_str=task_str)}"
-        # self.__save_log__(log_str=log_str)
-
+
     def execute_task(self, task: TaskPackage = None, agent_name: str = None, **kwargs):
         log_str = f"""{agent_name} starts execution on TaskPackage {task.task_id}===="""
         # self.__save_log__(log_str=log_str)
@@ -73,10 +56,6 @@ def take_action(self, action: AgentAct, agent_name: str, step_idx: int):
         {act_str}```"""
         self.__save_log__(log_str)
 
-    def add_st_memory(self, agent_name: str):
-        log_str = f"""Action and Observation added to {agent_name} memory"""
-        self.__save_log__(log_str)
-
     def get_obs(self, obs: str):
         if len(obs) > self.OBS_OFFSET:
             obs = obs[: self.OBS_OFFSET] + "[TLDR]"
@@ -91,17 +70,4 @@ def get_prompt(self, prompt):
     def get_llm_output(self, output: str):
         log_str = f"""LLM generates: {output}"""
         if self.PROMPT_DEBUG_FLAG:
-            self.__save_log__(log_str)
-
-
-if __name__ == "__main__":
-    aloger = AgentLogger()
-    agent_name = "labor_agent"
-    task_pack = TaskPackage(
-        instruction="this is a instruction",
-        task_creator="agent_1",
-        task_executor=f"{agent_name}",
-        completion=False,
-    )
-
-    aloger.receive_task(task_pack, agent_name=agent_name)
+            self.__save_log__(log_str)