Make logger and config args optional

Author: rickyloynd-microsoft

python/packages/autogen-ext/src/autogen_ext/agentic_memory/_agentic_memory_bank.py

@@ -24,9 +24,13 @@ class AgenticMemoryBank:
     Stores task-completion insights in a vector DB for later retrieval.
 
     Args:
-        - settings: Settings for the memory bank.
         - reset: True to clear the DB before starting.
-        - logger: The PageLogger object to use for logging.
+        - config: An optional dict that can be used to override the following values:
+            - path: The path to the directory where the memory bank files are stored.
+            - relevance_conversion_threshold: The threshold used to normalize relevance.
+            - n_results: The maximum number of most relevant results to return for any given topic.
+            - distance_threshold: The maximum topic-insight distance for an insight to be retrieved.
+        - logger: An optional logger. If None, no logging will be performed.
 
     Methods:
         - reset: Forces immediate deletion of all contents, in memory and on disk.
@@ -37,15 +41,36 @@ class AgenticMemoryBank:
         - get_relevant_insights: Returns any insights from the memory bank that appear sufficiently relevant to the given task topics.
     """
 
-    def __init__(self, settings: Dict[str, Any], reset: bool, logger: PageLogger) -> None:
-        self.settings = settings
+    def __init__(
+        self,
+        reset: bool,
+        config: Dict[str, Any] | None = None,
+        logger: PageLogger | None = None,
+    ) -> None:
+        if logger is None:
+            logger = PageLogger()  # Nothing will be logged by this object.
         self.logger = logger
         self.logger.enter_function()
 
-        memory_dir_path = os.path.expanduser(self.settings["path"])
-        self.relevance_conversion_threshold = self.settings["relevance_conversion_threshold"]
-        self.n_results = self.settings["n_results"]
-        self.distance_threshold = self.settings["distance_threshold"]
+        # Assign default values that can be overridden by config.
+        memory_dir_path = os.path.expanduser("~/agentic_memory/temp")
+        self.relevance_conversion_threshold = 1.7
+        self.n_results = 25
+        self.distance_threshold = 100
+
+        if config is not None:
+            # Apply any overrides from the config.
+            for key in config:
+                if key == "path":
+                    memory_dir_path = os.path.expanduser(config[key])
+                elif key == "relevance_conversion_threshold":
+                    self.relevance_conversion_threshold = config[key]
+                elif key == "n_results":
+                    self.n_results = config[key]
+                elif key == "distance_threshold":
+                    self.distance_threshold = config[key]
+                else:
+                    self.logger.error('Unexpected item in config: ["{}"] = {}'.format(key, config[key]))
 
         path_to_db_dir = os.path.join(memory_dir_path, "string_map")
         self.path_to_dict = os.path.join(memory_dir_path, "uid_insight_dict.pkl")

python/packages/autogen-ext/src/autogen_ext/agentic_memory/_prompter.py

@@ -21,7 +21,7 @@ class Prompter:
 
     Args:
         client: The client to call the model.
-        logger: The logger to log the model calls.
+        logger: An optional logger. If None, no logging will be performed.
 
     Methods:
         call_model: Calls the model client with the given input and returns the response.
@@ -33,9 +33,12 @@ class Prompter:
         extract_advice: Returns advice from the given text, or None if not found.
     """
 
-    def __init__(self, client: ChatCompletionClient, logger: PageLogger):
-        self.client = client
+    def __init__(self, client: ChatCompletionClient, logger: PageLogger | None = None) -> None:
+        if logger is None:
+            logger = PageLogger()  # Nothing will be logged by this object.
         self.logger = logger
+
+        self.client = client
         self.default_system_message_content = "You are a helpful assistant."
         self.time_spent_in_model_calls = 0.0
         self.num_model_calls = 0

python/packages/autogen-ext/src/autogen_ext/agentic_memory/_string_similarity_map.py

@@ -22,7 +22,7 @@ class StringSimilarityMap:
     Args:
         - reset: True to clear the DB immediately after creation.
         - path_to_db_dir: Path to the directory where the DB is stored.
-        - logger: The PageLogger object to use for logging.
+        - logger: An optional logger. If None, no logging will be performed.
 
     Methods:
         - add_input_output_pair: Adds one input-output string pair to the DB.
@@ -31,7 +31,9 @@ class StringSimilarityMap:
         - save_string_pairs: Saves the string-pair dict to disk.
     """
 
-    def __init__(self, reset: bool, path_to_db_dir: str, logger: PageLogger) -> None:
+    def __init__(self, reset: bool, path_to_db_dir: str, logger: PageLogger | None = None) -> None:
+        if logger is None:
+            logger = PageLogger()  # Nothing will be logged by this object.
         self.logger = logger
         self.path_to_db_dir = path_to_db_dir
 

python/packages/autogen-ext/src/autogen_ext/agentic_memory/agentic_memory_controller.py

@@ -15,11 +15,14 @@ class AgenticMemoryController:
     Manages memory-based learning, testing, and the flow of information to and from the memory bank.
 
     Args:
-        settings: Settings for the memory controller.
         reset: True to clear the memory bank before starting.
         client: The client to call the model.
         task_assignment_callback: The callback to assign a task to the agent.
-        logger: The logger to log the model calls.
+        - config: An optional dict that can be used to override the following values:
+            - max_train_trials: The maximum number of trials to attempt when training on a task.
+            - max_test_trials: The maximum number of trials to attempt when testing on a task.
+            - AgenticMemoryBank: A config dict passed to AgenticMemoryBank.
+        logger: An optional logger. If None, a default logger will be created.
 
     Methods:
         reset_memory: Resets the memory bank.
@@ -34,19 +37,38 @@ class AgenticMemoryController:
 
     def __init__(
         self,
-        settings: Dict[str, Any],
         reset: bool,
         client: ChatCompletionClient,
         task_assignment_callback: Callable[[str], Awaitable[Tuple[str, str]]],
-        logger: PageLogger,
+        config: Dict[str, Any] | None = None,
+        logger: PageLogger | None = None,
     ) -> None:
+        if logger is None:
+            logger = PageLogger({"level": "INFO"})
         self.logger = logger
         self.logger.enter_function()
-        self.settings = settings
+
+        # Assign default values that can be overridden by config.
+        self.max_train_trials = 10
+        self.max_test_trials = 3
+        agentic_memory_bank_config = None
+
+        if config is not None:
+            # Apply any overrides from the config.
+            for key in config:
+                if key == "max_train_trials":
+                    self.max_train_trials = config[key]
+                elif key == "max_test_trials":
+                    self.max_test_trials = config[key]
+                elif key == "AgenticMemoryBank":
+                    agentic_memory_bank_config = config[key]
+                else:
+                    self.logger.error('Unexpected item in config: ["{}"] = {}'.format(key, config[key]))
+
         self.client = client
         self.task_assignment_callback = task_assignment_callback
         self.prompter = Prompter(client, logger)
-        self.memory_bank = AgenticMemoryBank(self.settings["AgenticMemoryBank"], reset=reset, logger=logger)
+        self.memory_bank = AgenticMemoryBank(reset=reset, config=agentic_memory_bank_config, logger=logger)
         self.grader = Grader(client, logger)
         self.logger.leave_function()
 
@@ -62,9 +84,7 @@ async def train_on_task(self, task: str, expected_answer: str) -> None:
         """
         self.logger.enter_function()
         self.logger.info("Iterate on the task, possibly discovering a useful new insight.\n")
-        _, insight = await self._iterate_on_task(
-            task, expected_answer, self.settings["max_train_trials"], self.settings["max_test_trials"]
-        )
+        _, insight = await self._iterate_on_task(task, expected_answer)
         if insight is None:
             self.logger.info("No useful insight was discovered.\n")
         else:
@@ -219,7 +239,7 @@ def _format_memory_section(self, memories: List[str]) -> str:
         return memory_section
 
     async def _test_for_failure(
-        self, task: str, task_plus_insights: str, expected_answer: str, num_trials: int
+        self, task: str, task_plus_insights: str, expected_answer: str
     ) -> Tuple[bool, str, str]:
         """
         Attempts to solve the given task multiple times to find a failure case to learn from.
@@ -231,7 +251,7 @@ async def _test_for_failure(
         failure_found = False
         response, work_history = "", ""
 
-        for trial in range(num_trials):
+        for trial in range(self.max_test_trials):
             self.logger.info("\n-----  TRIAL {}  -----\n".format(trial + 1))
 
             # Attempt to solve the task.
@@ -252,9 +272,7 @@ async def _test_for_failure(
         self.logger.leave_function()
         return failure_found, response, work_history
 
-    async def _iterate_on_task(
-        self, task: str, expected_answer: str, max_train_trials: int, max_test_trials: int
-    ) -> Tuple[str, None | str]:
+    async def _iterate_on_task(self, task: str, expected_answer: str) -> Tuple[str, None | str]:
         """
         Repeatedly assigns a task to the agent, and tries to learn from failures by creating useful insights as memories.
         """
@@ -270,7 +288,7 @@ async def _iterate_on_task(
         successful_insight = None
 
         # Loop until success (or timeout) while learning from failures.
-        for trial in range(1, max_train_trials + 1):
+        for trial in range(1, self.max_train_trials + 1):
             self.logger.info("\n-----  TRAIN TRIAL {}  -----\n".format(trial))
             task_plus_insights = task
 
@@ -284,7 +302,7 @@ async def _iterate_on_task(
 
             # Can we find a failure case to learn from?
             failure_found, response, work_history = await self._test_for_failure(
-                task, task_plus_insights, expected_answer, max_test_trials
+                task, task_plus_insights, expected_answer
             )
             if not failure_found:
                 # No. Time to exit the loop.
@@ -299,7 +317,7 @@ async def _iterate_on_task(
                 break
 
             # Will we try again?
-            if trial == max_train_trials:
+            if trial == self.max_train_trials:
                 # No. We're out of training trials.
                 self.logger.info("\nNo more trials will be attempted.\n")
                 break

python/packages/autogen-ext/src/autogen_ext/agentic_memory/apprentice.py

@@ -26,9 +26,12 @@ class Apprentice:
     and call the Agentic Memory Controller using this class as an example.
 
     Args:
-        settings: The settings for the apprentice.
         client: The client to call the model.
-        logger: The logger to log the model calls.
+        - config: An optional dict that can be used to override the following values:
+            - name_of_agent_or_team: The name of the target agent or team for assigning tasks to.
+            - disable_prefix_caching: True to disable prefix caching by prepending random ints to the first message.
+            - AgenticMemoryController: A config dict passed to AgenticMemoryController.
+        logger: An optional logger. If None, a default logger will be created.
 
     Methods:
         reset_memory: Resets the memory bank.
@@ -38,22 +41,44 @@ class Apprentice:
         train_on_task: Repeatedly assigns a task to the completion agent, and tries to learn from failures by creating useful insights as memories.
     """
 
-    def __init__(self, settings: Dict[str, Any], client: ChatCompletionClient, logger: PageLogger) -> None:
-        self.client = client
+    def __init__(
+        self,
+        client: ChatCompletionClient,
+        config: Dict[str, Any] | None = None,
+        logger: PageLogger | None = None,
+    ) -> None:
+        if logger is None:
+            logger = PageLogger({"level": "INFO"})
         self.logger = logger
-        self.name_of_agent_or_team = settings["name_of_agent_or_team"]
-        self.disable_prefix_caching = settings["disable_prefix_caching"]
 
+        # Assign default values that can be overridden by config.
+        self.name_of_agent_or_team = "SimpleAgent"
+        self.disable_prefix_caching = False
+        agentic_memory_controller_config = None
+
+        if config is not None:
+            # Apply any overrides from the config.
+            for key in config:
+                if key == "name_of_agent_or_team":
+                    self.name_of_agent_or_team = config[key]
+                elif key == "disable_prefix_caching":
+                    self.disable_prefix_caching = config[key]
+                elif key == "AgenticMemoryController":
+                    agentic_memory_controller_config = config[key]
+                else:
+                    self.logger.error('Unexpected item in config: ["{}"] = {}'.format(key, config[key]))
+
+        self.client = client
         if self.disable_prefix_caching:
             self.rand = random.Random()
             self.rand.seed(int(time.time() * 1000))
 
         # Create the AgenticMemoryController, which creates the AgenticMemoryBank.
         self.memory_controller = AgenticMemoryController(
-            settings=settings["AgenticMemoryController"],
             reset=True,
             client=self.client,
             task_assignment_callback=self.assign_task_to_agent_or_team,
+            config=agentic_memory_controller_config,
             logger=self.logger,
         )
 

python/packages/autogen-ext/src/autogen_ext/agentic_memory/grader.py

@@ -25,17 +25,19 @@ class Grader:
 
     Args:
         client: The client to call the model.
-        logger: The logger to log the model calls.
+        logger: An optional logger. If None, no logging will be performed.
 
     Methods:
         test_apprentice: Tests the apprentice on the given task.
         call_model: Calls the model with the given input and returns the response.
         is_response_correct: Determines whether the response is equivalent to the task's correct answer.
     """
 
-    def __init__(self, client: ChatCompletionClient, logger: PageLogger):
-        self.client = client
+    def __init__(self, client: ChatCompletionClient, logger: PageLogger | None = None) -> None:
+        if logger is None:
+            logger = PageLogger()  # Nothing will be logged by this object.
         self.logger = logger
+        self.client = client
 
         # Check whether to report results to the client.
         self.report_results = hasattr(self.client, "report_result")
@@ -51,9 +53,8 @@ async def test_apprentice(
         num_trials: int,
         use_memory: bool,
         client: ChatCompletionClient,
-        logger: PageLogger,
     ) -> Tuple[int, int]:
-        logger.enter_function()
+        self.logger.enter_function()
 
         self.logger.info("Testing the apprentice on the given task.\n")
 
@@ -74,7 +75,7 @@ async def test_apprentice(
                 self.logger.info("Answer is INCORRECT.\n")
 
         self.logger.info("\nSuccess rate:  {}%\n".format(round((num_successes / num_trials) * 100)))
-        logger.leave_function()
+        self.logger.leave_function()
         return num_successes, num_trials
 
     async def call_model(

python/packages/autogen-ext/src/autogen_ext/agentic_memory/page_logger.py

@@ -71,9 +71,9 @@ class PageLogger:
     Logs text and images to a set of HTML pages, one per function/method, linked to each other in a call tree.
 
     Args:
-        settings: A dictionary containing the following keys:
+        - config: An optional dict that can be used to override the following values:
             - level: The logging level, one of DEBUG, INFO, WARNING, ERROR, CRITICAL, or NONE.
-            - path: The path to the directory where the log files will be saved.
+            - path: The path to the directory where the log files will be written.
 
     Methods:
         debug: Adds DEBUG text to the current page if debugging level <= DEBUG.
@@ -91,7 +91,7 @@ class PageLogger:
         leave_function: Finishes the page corresponding to the current function call.
     """
 
-    def __init__(self, settings: Dict[str, Any]) -> None:
+    def __init__(self, config: Dict[str, Any] | None = None) -> None:
         self.levels = {
             "DEBUG": 10,
             "INFO": 20,
@@ -100,10 +100,25 @@ def __init__(self, settings: Dict[str, Any]) -> None:
             "CRITICAL": 50,
             "NONE": 100,
         }
-        self.level = self.levels[settings["level"]]
+
+        # Assign default values that can be overridden by config.
+        self.level = self.levels["NONE"]  # Default to no logging at all.
+        self.log_dir = os.path.expanduser("~/pagelogs/temp")
+
+        if config is not None:
+            # Apply any overrides from the config.
+            for key in config:
+                if key == "level":
+                    self.level = self.levels[config[key]]
+                elif key == "path":
+                    self.log_dir = os.path.expanduser(config[key])
+                else:
+                    raise ValueError(f"Unknown key in PageLogger config: {key}")
+
+        # If the log level is set to NONE or higher, don't log anything.
         if self.level >= self.levels["NONE"]:
             return
-        self.log_dir = os.path.expanduser(settings["path"])
+
         self.page_stack = PageStack()
         self.pages: List[Page] = []
         self.last_page_id = 0