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]#

基类: object

(实验性，研究中)

实现基于内存的快速学习，并管理信息进出内存库的流动。

参数:

reset – 启动前清空内存库。
client – 内部使用的模型客户端。
task_assignment_callback – 一个可选的回调函数，用于将任务分配给调用方管理的任何代理。
config –
一个可选的字典，可用于覆盖以下值：
- generalize_task: 是否以更通用的术语重写任务。
- revise_generalized_task: 是否批判性地重写通用任务。
- generate_topics: 是否直接基于任务或从任务中提取的主题进行检索。
- validate_memos: 是否对检索到的备忘录应用最终验证阶段。
- max_memos_to_retrieve: 从 retrieve_relevant_memos() 返回的最大备忘录数量。
- max_train_trials: 训练任务时尝试的最大学习迭代次数。
- max_test_trials: 测试任务失败时尝试的总次数。
- MemoryBank: 传递给 MemoryBank 的配置字典。
logger – 一个可选的日志记录器。如果为 None，则会创建一个默认的日志记录器。

示例

首先需要安装 task-centric-memory 额外功能

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

以下代码片段展示了如何使用此类别进行最基本的内存存储和检索。

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())

reset_memory() → None[source]#: 清空 RAM 和磁盘中的内存库。

async train_on_task(task: str, expected_answer: str) → None[source]#: 反复将任务分配给代理，并通过创建有用的洞察力作为记忆来尝试从失败中学习。

async test_on_task(task: str, expected_answer: str, num_trials: int = 1) → Tuple[str, int, int][source]#: 将任务以及从内存中检索到的任何相关备忘录分配给代理。

async add_memo(insight: str, task: None | str = None, index_on_both: bool = True) → None[source]#: 将一条洞察力添加到内存库，并使用任务（如果提供）作为上下文。

async add_task_solution_pair_to_memory(task: str, solution: str) → None[source]#: 将任务-解决方案对添加到内存库，以便稍后作为组合洞察力一起检索。这在任务-解决方案对是解决与某些其他任务相关的任务的典范时非常有用。

async retrieve_relevant_memos(task: str) → List[Memo][source]#: 从内存中检索与任务相关的任何备忘录。

async assign_task(task: str, use_memory: bool = True, should_await: bool = True) → str[source]#: 通过 task_assignment_callback 将任务以及任何相关记忆分配给某个代理。

async consider_memo_storage(text: str) → str | None[source]#: 尝试从给定文本中提取任何建议并将其添加到内存中。

async handle_user_message(text: str, should_await: bool = True) → str[source]#: 通过提取任何建议作为洞察力存储在内存中，然后调用 assign_task() 来处理用户消息。

class MemoryControllerConfig[source]#

基类: TypedDict

generalize_task: bool#

revise_generalized_task: bool#

generate_topics: bool#

validate_memos: bool#

max_memos_to_retrieve: int#

max_train_trials: int#

max_test_trials: int#

MemoryBank: MemoryBankConfig#

class MemoryBankConfig[source]#

基类: TypedDict

path: str#

relevance_conversion_threshold: float#

n_results: int#

distance_threshold: int#