ad

l2mac/core.py

@@ -10,7 +10,13 @@
 from l2mac.llm_providers.general import setup_chat_rate_limiter
 from l2mac.llm_providers.rate_limiter import ChatRateLimiter
 from l2mac.utils.logging import create_logger_in_process, generate_log_file_path
-from l2mac.utils.run import DebuggingLevel, Domain, seed_all, to_dotdict
+from l2mac.utils.run import (
+    DebuggingLevel,
+    Domain,
+    load_prompt_program,
+    seed_all,
+    to_dotdict,
+)
 
 app = typer.Typer(help="Generate based on the prompt with LLM-automatic Computer")
 
@@ -47,12 +53,15 @@ def run_l2mac(
     init_config: Annotated[bool, typer.Option(help="Initialize the configuration file for L2MAC.")] = False,
 ):
     """
-    Generate based on the prompt with LLM-automatic Computer
+    Generate based on the input prompt with LLM-automatic Computer (L2MAC).
     """
     if init_config:
         print("Initializing configuration file...")
         copy_config_to_home()
         return None
+    # Process inputs
+    if prompt_program is not None:
+        prompt_program = load_prompt_program(prompt_program)
     config = load_config()
     log_path = generate_log_file_path(__file__, log_folder=config.setup.log_dir, config=config)
     config.setup.log_path = log_path

l2mac/l2mac.py

@@ -51,6 +51,14 @@ def __init__(
         self.logger = logger
         self.rate_limiter = rate_limiter
         self.name = "L2MAC"
+        self.run_tests = run_tests
+        self.project_name = project_name
+        self.steps = steps
+        self.prompt_program = prompt_program
+        self.prompts_file_path = prompts_file_path
+        self.tools_enabled = tools_enabled
+        self.debugging_level = debugging_level
+        self.wandb = wandb
         self.reset()
 
     def seed(self, seed_value):
@@ -92,7 +100,7 @@ def reset(self):
             self.message_hash = hash_messages([])
 
         self.log_folder_path = f"{self.config.setup.log_path.split('.txt')[0]}_{self.env.env_name}_{self.env.seed}/"
-        self.folder_path = f"workspace/{self.config.setup.log_path.split('_')[0].split('/')[1]}/"
+        self.folder_path = f"workspace/{self.project_name}_{self.config.setup.log_path.split('_')[0].split('/')[1]}/"
         Path(self.folder_path).mkdir(parents=True, exist_ok=True)
         Path(self.log_folder_path).mkdir(parents=True, exist_ok=True)
         write_files_from_dict(self.file_dict, base_dir=f"{self.folder_path}")
@@ -274,6 +282,13 @@ def _run(self, steps: int = 10):
         max_reflections = 1
         current_reflection = 0
         current_dialog = deepcopy(self.meta_messages)
+        if self.prompt_program is not None:
+            # Determine if the prompt program is a list of strings or a path to a file.
+            if isinstance(self.prompt_program, list):
+                self.steps = self.prompt_program
+            elif isinstance(self.prompt_program, str):
+                with open(self.prompt_program, "r") as f:
+                    self.steps = json.load(f)
         while len(steps) <= 50 and current_reflection < max_reflections:
             current_reflection += 1
             initial_response_message = self.get_llm_response(
@@ -393,7 +408,10 @@ def _run(self, steps: int = 10):
                 self.sub_messages.append(response_message)
                 if response_message.get("tool_calls"):
                     function_return_messages, self.file_dict = process_function_call_and_return_message(
-                        response_message["tool_calls"], self.file_dict, tools=self.functions
+                        response_message["tool_calls"],
+                        self.file_dict,
+                        tools=self.functions,
+                        enable_tests=self.run_tests,
                     )
                     for function_return_message in function_return_messages:
                         self.sub_messages.append(function_return_message)

l2mac/utils/run.py

@@ -1,3 +1,5 @@
+import ast
+import os
 import random
 import time
 from enum import Enum
@@ -56,3 +58,48 @@ def to_dotdict(obj):
         return [to_dotdict(item) for item in obj]
     else:
         return obj
+
+
+def load_prompt_program(input_string: str):
+    """
+    Loads a prompt program from a given file path or parses it from a string in list format.
+
+    Args:
+    input_string (str): The path to the prompt program file or the prompt program as a string in a list format.
+
+    Returns:
+    list: The loaded or parsed prompt program as a list, or None if not found or invalid.
+    """
+    # check input_string is a list type
+    if isinstance(input_string, list) and len(input_string) >= 1:
+        return input_string
+    # Check if the input string is a path to a file
+    if os.path.isfile(input_string):
+        try:
+            with open(input_string, "r", encoding="utf-8") as file:
+                # Read the file content and attempt to parse it
+                file_content = file.read()
+                return ast.literal_eval(file_content)
+        except (SyntaxError, ValueError, IOError) as e:
+            print(f"Error reading or parsing `prompt_program` file path of {input_string} | Error: {e}")
+            raise e
+    else:
+        # Try to parse it directly as a list from the string
+        try:
+            result = ast.literal_eval(input_string)
+            if isinstance(result, list):
+                return result
+            else:
+                raise ValueError("Input of `prompt_program` is not a list.")
+        except (SyntaxError, ValueError) as e:
+            print(f"Error reading or parsing `prompt_program` string encoded of {input_string} | Error: {e}")
+            raise e
+
+
+# Example usage
+user_input = input("Enter the path to the prompt program or the prompt program as a list string: ")
+prompt_program = load_prompt_program(user_input)
+if prompt_program is not None:
+    print("Loaded prompt program:", prompt_program)
+else:
+    print("No valid prompt program found, returned None.")