autogen_ext.experimental.task_centric_memory#

class MemoryController(reset: bool, client: ChatCompletionClient, task_assignment_callback: Callable[[str], Awaitable[Tuple[str, str]]] | None = None, config: MemoryControllerConfig | None = None, logger: PageLogger | None = None)[source]#

Bases: object

(EXPERIMENTAL, RESEARCH IN PROGRESS)

Implements fast, memory-based learning, and manages the flow of information to and from a memory bank.

Parameters:

  • reset – True to empty the memory bank before starting.

  • client – The model client to use internally.

  • task_assignment_callback – An optional callback used to assign a task to any agent managed by the caller.

  • config –

    An optional dict that can be used to override the following values:

    • max_train_trials: The maximum number of learning iterations to attempt when training on a task.

    • max_test_trials: The total number of attempts made when testing for failure on a task.

    • MemoryBank: A config dict passed to MemoryBank.

  • logger – An optional logger. If None, a default logger will be created.

Example

The task-centric-memory extra first needs to be installed:

pip install "autogen-ext[task-centric-memory]"

The following code snippet shows how to use this class for the most basic storage and retrieval of memories.:

import asyncio
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.experimental.task_centric_memory import MemoryController
from autogen_ext.experimental.task_centric_memory.utils import PageLogger


async def main() -> None:
    client = OpenAIChatCompletionClient(model="gpt-4o")
    logger = PageLogger(config={"level": "DEBUG", "path": "./pagelogs/quickstart"})  # Optional, but very useful.
    memory_controller = MemoryController(reset=True, client=client, logger=logger)

    # Add a few task-insight pairs as memories, where an insight can be any string that may help solve the task.
    await memory_controller.add_memo(task="What color do I like?", insight="Deep blue is my favorite color")
    await memory_controller.add_memo(task="What's another color I like?", insight="I really like cyan")
    await memory_controller.add_memo(task="What's my favorite food?", insight="Halibut is my favorite")

    # Retrieve memories for a new task that's related to only two of the stored memories.
    memos = await memory_controller.retrieve_relevant_memos(task="What colors do I like most?")
    print("{} memories retrieved".format(len(memos)))
    for memo in memos:
        print("- " + memo.insight)


asyncio.run(main())
async add_memo(insight: str, task: None | str = None, index_on_both: bool = True) → None[source]#

Adds one insight to the memory bank, using the task (if provided) as context.

async add_task_solution_pair_to_memory(task: str, solution: str) → None[source]#

Adds a task-solution pair to the memory bank, to be retrieved together later as a combined insight. This is useful when the task-solution pair is an exemplar of solving a task related to some other task.

async assign_task(task: str, use_memory: bool = True, should_await: bool = True) → str[source]#

Assigns a task to some agent through the task_assignment_callback, along with any relevant memories.

async consider_memo_storage(text: str) → str | None[source]#

Tries to extract any advice from the given text and add it to memory.

async handle_user_message(text: str, should_await: bool = True) → str[source]#

Handles a user message by extracting any advice as an insight to be stored in memory, and then calling assign_task().

reset_memory() → None[source]#

Empties the memory bank in RAM and on disk.

async retrieve_relevant_memos(task: str) → List[Memo][source]#

Retrieves any memos from memory that seem relevant to the task.

async test_on_task(task: str, expected_answer: str, num_trials: int = 1) → Tuple[str, int, int][source]#

Assigns a task to the agent, along with any relevant memos retrieved from memory.

async train_on_task(task: str, expected_answer: str) → None[source]#

Repeatedly assigns a task to the agent, and tries to learn from failures by creating useful insights as memories.