Console outputs improved (but still room for improvement)

golemgpt/__main__.py

@@ -30,11 +30,14 @@ def main():
     memory.load(job_key)
     goals = memory.goals
 
-    # while not goals:
-    #     goal = input("Enter a goal for the Golem-GPT:\n?> ").strip()
-    #     if goal:
-    #         goals.append(goal)
-    goals = ["Get weather in Batumi, Georgia."]
+    while not goals:
+        goal = input("Enter a goal for the Golem-GPT:\n?> ").strip()
+        if goal:
+            goals.append(goal)
+    # goals = [
+    #     "Get weather in Batumi, Georgia. "
+    #     "Write results in human readable format."
+    # ]
 
     golem = GeneralGolem(
         goals=goals,

golemgpt/actions/ask_human_input.py

@@ -6,4 +6,5 @@ def ask_human_input_action(query: str, **kwargs):
         result = input(f"{query}\n?> ")
     except KeyboardInterrupt:
         raise JobFinished()
-    return f"{query}\nAnswer is: {result}"
+    # return f"{query}\nAnswer is: {result}"
+    return result

golemgpt/actions/summarize_file.py

@@ -15,7 +15,7 @@ def summarize_file_action(
     with path.open("r") as file:
         content = file.read()
     content = content[:MAX_RAW_SIZE]
-    cognitron = golem.spawn_cognitron()
+    cognitron = golem.cognitron()
     prompt = f"Summarize the text, use the hint '{hint}':\n\n{content}"
     reply = cognitron.communicate(prompt)
     return write_file_action(out_filename, reply)

golemgpt/codex/reasonable.py

@@ -5,7 +5,10 @@
 
 class ReasonableCodex(BaseCodex):
     check_actions_depth = 3
-    name = "reasonable-codex"
+
+    @property
+    def name(self):
+        return self.cognitron.name
 
     def is_job_finished(self, actions: list[dict]) -> bool:
         if len(actions) == 1:
@@ -32,7 +35,7 @@ def align_actions(self, actions: list[dict]) -> bool:
 
         check_actions = actions[: self.check_actions_depth]
         prompt, answer_expected = self.align_actions_prompt(check_actions)
-        console.message(self.name, prompt)
+        console.message(self.name, prompt, tags=["prompt"])
         answer_real = self.cognitron.ask_yesno(prompt)
         if answer_expected == answer_real:
             return True

golemgpt/cognitron/openai.py

@@ -68,12 +68,14 @@ def communicate(self, message: str, **options) -> Reply:
         )
         reply = result["choices"][0]["message"]
 
+        # Update history
         messages.append(reply)
         self.memory.messages = messages
         self.memory.save()
 
+        # Print console output
         reply_text = self.get_last_message()
-        console.message(self.name, reply_text)
+        console.message(self.name, reply_text, tags=["reply"])
 
         return Reply(text=reply_text)
 
@@ -86,7 +88,7 @@ def plan_actions(self, prompt: str, attempt: int = 0) -> list[dict]:
         we try to detect and truncate.
 
         """
-        console.message(self.name, prompt)
+        console.message(self.name, prompt, tags=["prompt"])
 
         reply = self.communicate(prompt)
         reply_text = reply.text
@@ -96,7 +98,7 @@ def plan_actions(self, prompt: str, attempt: int = 0) -> list[dict]:
         # prepended with some text, like "Here is your JSON:"
         preamble = self.lexicon.find_preamble(reply_text)
         if preamble:
-            reply_text = reply_text[len(preamble):]
+            reply_text = reply_text[len(preamble) :]
             console.debug(f"Parse plan (trunc.):\n{reply_text}\n")
 
         reply_text = reply_text.strip()
@@ -123,7 +125,7 @@ def initializer_prompt(self) -> str:
 
     def goal_prompt(self, goal: str) -> str:
         tools_hint = (
-            "NOTES: Use tools provided. "
+            "NOTES: Use tools provided. Prefer public APIs (no credentials). "
             "Ask credentials from user if needed, be specific for which "
             "API endpoint you'll use it (exact address needed)."
         )
@@ -252,7 +254,7 @@ def plan_actions(self, prompt: str, attempt: int = 0) -> list[dict]:
         list because that's how OpenAI's tools feature works currently.
 
         """
-        console.message(self.name, prompt)
+        console.message(self.name, prompt, tags=["prompt"])
         reply = self.communicate(prompt)
         actions: list[dict] = []
 
@@ -308,12 +310,21 @@ def communicate(self, message: str, **options) -> Reply:
             json=payload,
         )
         reply = result["choices"][0]["message"]
+        actions = reply["tool_calls"]
 
+        # Update history
         messages.append(reply)
         self.memory.messages = messages
         self.memory.save()
 
-        reply_text = self.get_last_message()
-        console.message(self.name, reply_text)
+        # Print console output
+        reply_text = ""
+        for item in actions:
+            arguments = json_loads(item["function"]["arguments"])
+            arguments_text = " | ".join(
+                [f"{k}={v}" for k, v in arguments.items()]
+            )
+            reply_text += f'- {item["function"]["name"]}({arguments_text})'
+        console.message(self.name, reply_text.strip(), tags=["reply"])
 
-        return Reply(actions=reply["tool_calls"])
+        return Reply(actions=actions)

golemgpt/golems/general.py

@@ -54,24 +54,21 @@ def __init__(
     def __str__(self) -> str:
         return f"{self.__class__.__name__}({self.job_key})"
 
-    @property
-    def lexicon(self) -> BaseLexicon:
-        return self.core.lexicon
-
     def start_job(self) -> None:
         """Start the job."""
         goals_text = "\n".join(self.goals)
         console.info(f"Starting job: {self.job_key}")
         console.info(f"Goals:\n{goals_text}")
 
         self.initialize()
-        outcome = self.lexicon.goal_prompt(self.goals[-1])
+
+        outcome = self.lexicon().goal_prompt(self.goals[-1])
         while True:
             try:
                 if outcome:
-                    action_plan = self.core.plan_actions(outcome)
-                    if self.codex().align_actions(action_plan):
-                        self.action_plan = action_plan
+                    new_action_plan = self.plan_actions(outcome)
+                    self.align_actions(new_action_plan)
+                    self.action_plan = new_action_plan
                 outcome = self.run_action()
             except JobFinished:
                 break
@@ -95,7 +92,7 @@ def initialize(self) -> None:
 
         if self.memory.is_history_empty:
             self.memory.goals = self.goals
-            iniital_history = self.lexicon.initializer_history()
+            iniital_history = self.lexicon().initializer_history()
             for message in iniital_history:
                 self.memory.messages.append(message)
         else:
@@ -104,11 +101,11 @@ def initialize(self) -> None:
 
         self.memory.save()
 
-    def spawn_cognitron(self, **options) -> BaseCognitron:
-        """Return a Cognitron instance."""
+    def cognitron(self, **options) -> BaseCognitron:
+        """Return a new Cognitron instance."""
         assert self.settings, "Error: settings must be initialized first."
         assert self.memory, "Error: memory must be initialized first."
-        key = f"{self.job_key}.{genkey()}"
+        key = f"{self.job_key}/{genkey()}"
         memory = self.memory.spawn(key)
         return self.cognitron_class(
             settings=self.settings,
@@ -118,10 +115,13 @@ def spawn_cognitron(self, **options) -> BaseCognitron:
         )
 
     def codex(self, **options) -> BaseCodex:
-        """Return a Codex instance."""
-        cognitron = self.spawn_cognitron(name="Codex", **options)
+        """Return a new Codex instance."""
+        cognitron = self.cognitron(name="Codex", **options)
         return self.codex_class(cognitron)
 
+    def lexicon(self) -> BaseLexicon:
+        return self.core.lexicon
+
     def run_action(self) -> str:
         """Run the next action in the plan."""
         if not self.action_plan:
@@ -130,38 +130,18 @@ def run_action(self) -> str:
         action, result = self.runner(action_item, golem=self)
         if not result:
             return ""
-        return self.lexicon.action_result_prompt(action, result)
+        return self.lexicon().action_result_prompt(action, result)
 
-    def plan_actions(self, prompt: str, attempt: int = 0) -> None:
-        """Ask to update the plan based on the prompt."""
-        console.message(self.name, prompt)
-        reply = self.core.communicate(prompt)
+    def plan_actions(self, prompt: str, attempt: int = 0) -> list[dict]:
+        """Generate an action plan based on the prompt."""
         try:
-            self.action_plan = self.core.plan_actions(prompt)
+            action_plan = self.core.plan_actions(prompt)
         except ParseActionsError:
-            self.try_restore_plan(reply, attempt + 1)
+            # TODO: Ingest format reminder message to the memory and retry,
+            # check `lexicon.remind_format_prompt()`.
+            raise
+        return action_plan
 
-    def align_actions(self, prompt: str) -> list[str]:
+    def align_actions(self, action_plan: list[dict]) -> list[str]:
         """Align the action plan with the codex."""
-        return self.codex().align_actions(self.action_plan)
-
-    def try_restore_plan(self, reply: str, attempt: int = 0) -> None:
-        """Try restore the plan after malformed reply."""
-        # Finish if too many failed attempts:
-        if attempt > DEFAULT_RETRY_PLAN_ATTEMPTS:
-            raise JobFinished()
-
-        # Ask in a helper dialog, if job is finished:
-        question = self.lexicon.guess_finish_prompt(reply)
-        if self.helper_yesno(question):
-            raise JobFinished()
-
-        # Try to plan again after remainder about the format:
-        remainder = self.lexicon.remind_format_prompt()
-        self.plan_actions(remainder, attempt + 1)
-
-    def helper_yesno(self, question: str) -> bool:
-        """Guess if the reply is a yes or no."""
-        prompt = self.lexicon.yesno_prompt(question)
-        cognitron = self.spawn_cognitron(name="Helper")
-        return cognitron.ask_yesno(prompt)
+        return self.codex().align_actions(action_plan)

golemgpt/memory/localfiles.py

@@ -12,6 +12,13 @@ def __init__(self, root_dir: Path) -> None:
     def root_dir(self) -> Path:
         return self.config['root_dir']
 
+    def get_path(self, filename: str) -> Path:
+        assert self.key
+        path = self.root_dir
+        for key_part in self.key.split('/'):
+            path = path / key_part
+        return path / filename
+
     def spawn(self, key: str) -> 'LocalFilesMemory':
         """Spawn a fresh memory instance."""
         isinstance = LocalFilesMemory(self.root_dir)
@@ -20,17 +27,15 @@ def spawn(self, key: str) -> 'LocalFilesMemory':
 
     def load_file(self, filename: str, default: object) -> list:
         """Load JSON data from a local file."""
-        assert self.key
-        path = self.root_dir / self.key / filename
+        path = self.get_path(filename)
         if path.exists():
             with path.open() as file:
                 return json_load(file)
         return default
 
     def save_file(self, filename: str, data: object) -> None:
         """Save JSON data to a local file."""
-        assert self.key
-        path = self.root_dir / self.key / filename
+        path = self.get_path(filename)
         path.parent.mkdir(parents=True, exist_ok=True)
         with path.open('w') as file:
             json_dump(data, file, indent=2, ensure_ascii=False)

golemgpt/utils/console.py

@@ -2,28 +2,29 @@
 
 _DEBUG = False
 
-_COLORS = ['light_red', 'green', 'yellow', 'blue', 'light_magenta']
+_COLORS = ["light_red", "green", "yellow", "blue", "light_magenta"]
 
 
 def set_debug(enabled: bool) -> None:
     global _DEBUG
     _DEBUG = enabled
 
 
-def message(author: str, text: str) -> None:
+def message(author: str, text: str, tags: list[str] | None = None) -> None:
     author = author.upper()
     idx = sum(ord(char) for char in author) % len(_COLORS)
     author_color = _COLORS[idx]
-    cprint(colored(author, author_color, None, []))
+    title = author + (f" [{', '.join(tags)}]" if tags else "")
+    cprint(colored(title, author_color, None, []))
     print(text)
-    print('')
+    print("")
 
 
 def info(text: str) -> None:
-    cprint(text, 'cyan')
+    cprint(text, "cyan")
 
 
-def debug(text: str, indent='    ') -> None:
+def debug(text: str, indent="    ") -> None:
     if _DEBUG:
-        indented = '\n'.join([f'{indent}{x}' for x in text.splitlines()])
-        cprint(indented, 'grey')
+        indented = "\n".join([f"{indent}{x}" for x in text.splitlines()])
+        cprint(indented, "grey")