Task Memory

Task Memory#

Overview#

TaskMem is a memory-based capability enhancement feature provided by the Agent framework. It can store the conclusions of operations and the evaluation module at each step of agent reasoning. When executed again, it retrieves similar scenarios, dynamically summarizes experience to improve the task success rate.

Characteristic#

  • Memory-based Evolution: During multi-round execution, TaskMem utilizes trajectory memory and the evaluation results at that time (optional) to form an experience optimization context.

  • Accessible Evaluation Functions: TaskMem includes an instant evaluation module entry point that can generate correct/incorrect, scores, and evaluations. It can also leave some fields blank and store them directly, analyzing them uniformly during rule generation.

  • Experience Generation Guidance Interface: Supports users to pass in guidance descriptions for experience summarization in the form of a Prompt, to guide the experience generation for new tasks.

Best practice#

Build TaskMem and use it directly (can also be integrated with Agent).

import os
from loguru import logger
from evofabric.core.vectorstore import ChromaDB
from evofabric.core.mem import TaskMem, TASK_SUMMARY_PROMPT_EN
from evofabric.core.typing import UserMessage, AssistantMessage, LLMChatResponse, StateMessage
from evofabric.core.clients import OpenAIChatClient, SentenceTransformerEmbed


# 1. create a chat client
chat_client = OpenAIChatClient(
    model=os.getenv("MODEL_NAME"),
    stream=False,
    client_kwargs={
        "api_key": os.getenv("OPENAI_API_KEY"),
        "base_url": os.getenv("OPENAI_BASE_URL")
    }
    )

# 2. create a vectorstore
embed_client = SentenceTransformerEmbed(
    device="cpu",
    model="sentence-transformers/all-MiniLM-L6-v2",
)

vectorstore = ChromaDB(
    collection_name="chroma_db",
    persist_directory="your db path",
    embedding=embed_client,
    top_k=2
)

# 3. define your critic function – demo implementation
async def demo_critic(messages: list[StateMessage]) -> tuple[bool, float, str]:
    try:
        system_prompt = f"""
        The agent is helping a user troubleshoot mobile-phone malfunctions.
        The correct actions for some common faults are:
        - When the phone loses network: check the wireless-network configuration.
        - When the phone overheats: inspect and clean up unnecessary background apps.
        ...

        Message history:
        {messages}

        Judge whether the agent's action is reasonable in the current context.
        Reply with the following fields:
        <correctness>True/False</correctness>
        <comment>Analyse the agent's policy: explain why it is right or wrong.</comment>
        """
        analysis_messages = [UserMessage(content=system_prompt)]
        analysis = ""
        async for msg in chat_client.create_on_stream(analysis_messages):
            if isinstance(msg, LLMChatResponse):
                analysis = msg.content
        logger.info(f"Evaluation result: {analysis}")

        correctness_str = analysis.split("<correctness>")[1].split("</correctness>")[0]
        correctness = "True" in correctness_str
        score = 1.0 if correctness else 0.0
        comment = analysis.split("<comment>")[1].split("</comment>")[0]
        return correctness, score, comment
    except Exception as e:
        return True, 0.5, str(e)


# Create an English-task memory instance
task_mem = TaskMem(
    zh_mode=False,  # English mode
    vectorstore=vectorstore,
    chat_client=chat_client,
    user_summary_prompt=TASK_SUMMARY_PROMPT_EN,  # English prompt constant example
    eval_func=demo_critic
)

# Example conversation turns
state_messages = [
    UserMessage(content="My phone lost network connection."),
    AssistantMessage(content="I comforted you."),
]
await task_mem.add_messages(state_messages)

state_messages = [
    UserMessage(content="My phone lost network connection."),
    AssistantMessage(content="Let me help you check the wireless network configuration."),
]
await task_mem.add_messages(state_messages)

# System prompt generation via retrieval
retrieval_messages = [UserMessage(content="My phone lost network again.")]
retrieved_messages = await task_mem.retrieval_update(retrieval_messages)
logger.info(f"Memory retrieved: {retrieved_messages[0].content}")
Contents