MemBase

MemBase#

Overview#

MemBase defines the basic protocol for interaction between Memory and agents. When using the Evofabric framework, custom memory usage can be implemented by inheriting MemBase and implementing its basic methods.

Characteristics#

  • Unified Interface: All memory systems inherit from MemBase, providing a consistent calling method.

  • Memory Expansion: Users can inherit MemBase and define their own methods.

  • Example Memory: Based on MemBase, EvoFabric implements three commonly used memory types—RetrievalMem, ChatMem, and TaskMem—for direct user use.

Open Source Memory Adaptation#

Implement the retrieval_update, add_messages, and clear methods (all asynchronous).

from typing import Dict, Optional, Union, List
from pydantic import Field, PrivateAttr
from mem0 import AsyncMemory
from evofabric.core.mem import MemBase
from evofabric.core.typing import StateMessage, UserMessage


# Example implementation of Mem0
class Mem0Module(MemBase):
        message_rounds: int = Field(default=20,
                                    description="Messages rounds considered in retrieval and save")
        mem_config: Dict[str, Union[str, bool, Dict]] = Field(description="Mem0 initialization config")
        user_id: str = Field(default="default", description="mem0 user_id")
        _mem0_client: Optional[AsyncMemory] = PrivateAttr(default=None)

        async def get_client(self) -> AsyncMemory:
            """Initialize AsyncMem when first call"""
            if self._mem0_client is None:
                self._mem0_client = await AsyncMemory.from_config(self.mem_config)
            return self._mem0_client

        async def retrieval_update(self, messages: List[StateMessage], **kwargs) -> List[StateMessage]:
            """Implement the retrieval_update method"""
            context = "context: "
            for msg in messages[-self.message_rounds:]:
                context += f"\n{msg.role}: {msg.content}"
            _mem0_client = await self.get_client()
            items = await _mem0_client.search(context, user_id=self.user_id)
            contents = [item['memory'] for item in items['results']]
            messages = [UserMessage(content="Historical memory: " + str(contents))] + messages
            return messages

        # Implement memory update functionality
        async def add_messages(self, messages: List[StateMessage], **kwargs) -> None:
            """Implement the add_messages method"""
            _mem0_client = await self.get_client()
            message_dict = [
                {"role": msg.role, "content": msg.content} for msg in messages[-self.message_rounds:]
            ]
            await _mem0_client.add(message_dict, user_id=self.user_id)
            return

        async def clear(self):
            """Implement the clear method"""
            _mem0_client = await self.get_client()
            await _mem0_client.reset()

Access Agent#

The current Agent supports passing in one or more Memory systems. Retrieval: Before the Agent’s reasoning, it will retrieve from each memory system based on context information and concatenate UserMessage information into the message list. The entry method is retrieval_update. Update: After the Agent’s reasoning is complete, the context information will be updated to the memory database one by one. The entry method is add_messages. The above execution process is implemented within the Agent, and it is only necessary to connect the memory module.

from evofabric.core.agent import AgentNode

# Create an English chat mem
chat_mem = ChatMem(
    zh_mode=True,
    vectorstore=vectorstore,
    chat_client=test_client_common,
    feat_define_prompt="Please extract the user's professional identity and personal habits."
)

# Create an agent node using memory
agent = AgentNode(client=chat_client, memory=[retrieval_mem])
Contents