Agentic memory

python/.gitignore

@@ -157,7 +157,7 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/
 
 .ruff_cache/
 

python/packages/autogen-ext/pyproject.toml

@@ -66,6 +66,8 @@ jupyter-executor = [
     "nbclient>=0.10.2",
 ]
 
+agentic-memory = ["chromadb>=0.6.3"]
+
 semantic-kernel-core = [
     "semantic-kernel>=1.17.1",
 ]
@@ -126,7 +128,7 @@ dev = [
 [tool.ruff]
 extend = "../../pyproject.toml"
 include = ["src/**", "tests/*.py"]
-exclude = ["src/autogen_ext/agents/web_surfer/*.js", "src/autogen_ext/runtimes/grpc/protos", "tests/protos"]
+exclude = ["src/autogen_ext/agents/web_surfer/*.js", "src/autogen_ext/runtimes/grpc/protos", "tests/protos", "README.md"]
 
 [tool.pyright]
 extends = "../../pyproject.toml"

python/packages/autogen-ext/src/autogen_ext/agentic_memory/README.md

@@ -0,0 +1,81 @@
+# Agentic Memory
+
+This AutoGen extension provides an implementation of agentic memory, which we define as a
+broad ability for AI agents to accomplish tasks more effectively by learning quickly and continually (over the long term).
+This is distinct from what RAG or long context windows can provide.
+While still under active research and development, this implementation of agentic memory
+can be attached to virtually any unmodified AI agent, and is designed to enable agents that:
+
+* Remember guidance, corrections, and demonstrations provided by users.
+* Succeed more frequently on tasks after finding successful solutions to similar tasks.
+* Learn and adapt quickly to changing circumstances to enable workflows that are dynamic and self-healing.
+
+The implementation is also intended to:
+
+* Be general purpose, unconstrained by types and schemas required by standard databases.
+* Augment rather than interfere with an agent’s special capabilities, such as powerful reasoning, long-horizon autonomy, and tool handling.
+* Operate in both foreground and background modes, so that an agent can discuss tasks with a user (in the foreground)
+then work productively on those tasks (in the background) while the user does other things.
+* Allow for fine-grained transparency and auditing of individual memories by human users or other agents.
+* Allow agents to be personalized (to a single user) as well as specialized (to a subject, domain or project).
+The benefits of personalization scale linearly with the number of users, but the benefits of domain specialization
+can scale quadratically with the number of users working in that domain, as insights gained from interactions with one user
+can benefit other users in similar situations.
+* Support multiple memory banks dynamically attached to an agent at runtime.
+* Enable enforcement of security boundaries at the level of individual memory banks.
+* Allow users to download and port memory banks between agents and systems.
+
+![agentic_memory.png](../../../imgs/agentic_memory.png)
+
+The block diagram above outlines the key components of our baseline agentic memory architecture,
+which augments an agent or team with agentic memory mechanisms.
+
+The **Agentic Memory Controller** implements the fast-learning methods described below,
+and manages communication with an **Agentic Memory Bank** containing a vector DB and associated structures.
+
+The **Apprentice** is a minimal reference implementation that wraps the combination of agentic memory plus some agent or team.
+Certain applications will use the Apprentice,
+while others will directly instantiate and call the Agentic Memory Controller.
+
+We’ve successfully tested agentic memory with a simple AssistantAgent and MagenticOneGroupChat.
+
+## Memory Creation and Storage
+
+Each stored memory is an insight (in text form) crafted to help the agent accomplish future tasks that are similar
+to some task encountered in the past. If the user provides advice for solving a given task,
+the advice is extracted and stored as an insight. If the user demonstrates how to perform a task,
+the task and demonstration are stored together as an insight that could be applied to similar but different tasks.
+If the agent is given a task (free of side-effects) and some means of determining success or failure,
+the memory controller repeats the following learning loop in the background some number of times:
+
+1. Test the agent on the task a few times to check for a failure.
+2. If a failure is found, analyze the agent’s response in order to:
+   1. Diagnose the failure of reasoning or missing information,
+   2. Phrase a general piece of advice, such as what a teacher might give to a student,
+   3. Temporarily append this advice to the task description,
+   4. Return to step 1.
+   5. If some piece of advice succeeds in helping the agent solve the task a number of times, add the advice as an insight to memory.
+3. For each insight to be stored in memory, an LLM is prompted to generate a set of free-form, multi-word topics related to the insight. Each topic is embedded to a fixed-length vector and stored in a vector DB mapping it to the topic’s related insight.
+
+## Memory Retrieval and Usage
+
+When the agent is given a task, the following steps are performed by the memory controller:
+1. The task is rephrased into a generalized form.
+2. A set of free-form, multi-word query topics are generated from the generalized task.
+3. A potentially large number of previously stored topics, those most similar to each query topic, are retrieved from the vector DB along with the insights they map to.
+4. These candidate insights are filtered by the aggregate similarity of their stored topics to the query topics.
+5. In the final filtering stage, an LLM is prompted to return only those insights that seem potentially useful in solving the task at hand.
+
+Retrieved insights that pass the filtering steps are listed under a heading like
+“Important insights that may help solve tasks like this”, then appended to the task description before it is passed to the agent as usual.
+
+## Setup and Usage
+
+Install AutoGen and its extension package as follows:
+
+`pip install "autogen-ext[agentic-memory]"`
+
+We provide [sample code](../../../../../samples/agentic_memory) to illustrate the following forms of memory-based fast learning:
+* Agent learning from user advice and corrections
+* Agent learning from user demonstrations
+* Agent learning from its own experience

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

@@ -0,0 +1,6 @@
+from .agentic_memory_controller import AgenticMemoryController
+from .apprentice import Apprentice
+from .grader import Grader
+from .page_logger import PageLogger
+
+__all__ = ["Apprentice", "PageLogger", "Grader", "AgenticMemoryController"]

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

@@ -0,0 +1,181 @@
+import os
+import pickle
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Tuple
+
+from ._string_similarity_map import StringSimilarityMap
+from .page_logger import PageLogger
+
+
+@dataclass
+class Insight:
+    """
+    Represents a task-completion insight, which is a string that may help solve a task.
+    """
+
+    id: str
+    insight_str: str
+    task_str: str | None
+    topics: List[str]
+
+
+class AgenticMemoryBank:
+    """
+    Stores task-completion insights in a vector DB for later retrieval.
+
+    Args:
+        - reset: True to clear the DB before starting.
+        - 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.
+        - save_insights: Saves the current insight structures (possibly empty) to disk.
+        - contains_insights: Returns True if the memory bank contains any insights.
+        - add_insight: Adds an insight to the memory bank, given topics related to the insight, and optionally the task.
+        - add_task_with_solution: Adds a task-insight pair to the memory bank, to be retrieved together later.
+        - get_relevant_insights: Returns any insights from the memory bank that appear sufficiently relevant to the given task topics.
+    """
+
+    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()
+
+        # 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")
+
+        self.string_map = StringSimilarityMap(reset=reset, path_to_db_dir=path_to_db_dir, logger=self.logger)
+
+        # Load or create the associated insight dict on disk.
+        self.uid_insight_dict: Dict[str, Insight] = {}
+        self.last_insight_id = 0
+        if (not reset) and os.path.exists(self.path_to_dict):
+            self.logger.info("\nLOADING INSIGHTS FROM DISK  {}".format(self.path_to_dict))
+            self.logger.info("    Location = {}".format(self.path_to_dict))
+            with open(self.path_to_dict, "rb") as f:
+                self.uid_insight_dict = pickle.load(f)
+                self.last_insight_id = len(self.uid_insight_dict)
+                self.logger.info("\n{} INSIGHTS LOADED".format(len(self.uid_insight_dict)))
+
+        # Clear the DB if requested.
+        if reset:
+            self._reset_insights()
+
+        self.logger.leave_function()
+
+    def reset(self) -> None:
+        """
+        Forces immediate deletion of all contents, in memory and on disk.
+        """
+        self.string_map.reset_db()
+        self._reset_insights()
+
+    def _reset_insights(self) -> None:
+        """
+        Forces immediate deletion of the insights, in memory and on disk.
+        """
+        self.uid_insight_dict = {}
+        self.save_insights()
+
+    def save_insights(self) -> None:
+        """
+        Saves the current insight structures (possibly empty) to disk.
+        """
+        self.string_map.save_string_pairs()
+        with open(self.path_to_dict, "wb") as file:
+            pickle.dump(self.uid_insight_dict, file)
+
+    def contains_insights(self) -> bool:
+        """
+        Returns True if the memory bank contains any insights.
+        """
+        return len(self.uid_insight_dict) > 0
+
+    def _map_topics_to_insight(self, topics: List[str], insight_id: str, insight: Insight) -> None:
+        """
+        Adds a mapping in the vec DB from each topic to the insight.
+        """
+        self.logger.enter_function()
+        self.logger.info("\nINSIGHT\n{}".format(insight.insight_str))
+        for topic in topics:
+            self.logger.info("\n TOPIC = {}".format(topic))
+            self.string_map.add_input_output_pair(topic, insight_id)
+        self.uid_insight_dict[insight_id] = insight
+        self.logger.leave_function()
+
+    def add_insight(self, insight_str: str, topics: List[str], task_str: Optional[str] = None) -> None:
+        """
+        Adds an insight to the memory bank, given topics related to the insight, and optionally the task.
+        """
+        self.last_insight_id += 1
+        id_str = str(self.last_insight_id)
+        insight = Insight(id=id_str, insight_str=insight_str, task_str=task_str, topics=topics)
+        self._map_topics_to_insight(topics, id_str, insight)
+
+    def add_task_with_solution(self, task: str, solution: str, topics: List[str]) -> None:
+        """
+        Adds a task-solution pair to the memory bank, to be retrieved together later as a combined insight.
+        This is useful when the insight is a demonstration of how to solve a given type of task.
+        """
+        self.last_insight_id += 1
+        id_str = str(self.last_insight_id)
+        # Prepend the insight to the task description for context.
+        insight_str = "Example task:\n\n{}\n\nExample solution:\n\n{}".format(task, solution)
+        insight = Insight(id=id_str, insight_str=insight_str, task_str=task, topics=topics)
+        self._map_topics_to_insight(topics, id_str, insight)
+
+    def get_relevant_insights(self, task_topics: List[str]) -> Dict[str, float]:
+        """
+        Returns any insights from the memory bank that appear sufficiently relevant to the given task topics.
+        """
+        # Process the matching topics to build a dict of insight-relevance pairs.
+        matches: List[Tuple[str, str, float]] = []  # Each match is a tuple: (topic, insight, distance)
+        insight_relevance_dict: Dict[str, float] = {}
+        for topic in task_topics:
+            matches.extend(self.string_map.get_related_string_pairs(topic, self.n_results, self.distance_threshold))
+        for match in matches:
+            relevance = self.relevance_conversion_threshold - match[2]
+            insight_id = match[1]
+            insight_str = self.uid_insight_dict[insight_id].insight_str
+            if insight_str in insight_relevance_dict:
+                insight_relevance_dict[insight_str] += relevance
+            else:
+                insight_relevance_dict[insight_str] = relevance
+
+        # Filter out insights with overall relevance below zero.
+        for insight in list(insight_relevance_dict.keys()):
+            if insight_relevance_dict[insight] < 0:
+                del insight_relevance_dict[insight]
+
+        return insight_relevance_dict