Upload 5 files

aworld/memory/README.md

@@ -0,0 +1,62 @@
+## multi-agents memory
+![](../../readme_assets/framework_memory_example.png)
+
+### Short-Term Memory
+
+Short-term memory (InMemory) is suitable for lightweight, temporary multi-agent memory scenarios. Data is only stored in memory, making it ideal for testing and small-scale experiments.
+
+**Usage Example:**
+
+```python
+from aworld.core.memory import MemoryConfig, MemoryItem
+from aworld.memory.main import MemoryFactory
+
+# Create InMemory config
+memory_config = MemoryConfig(provider="inmemory", enable_summary=False)
+# Initialize Memory
+memory = MemoryFactory.from_config(memory_config)
+
+# Add a memory item
+memory.add(MemoryItem(content="Hello, world!", metadata={"user_id": "u1"}, tags=["greeting"]))
+
+# Get all memory items
+all_memories = memory.get_all()
+for item in all_memories:
+    print(item.content)
+```
+
+### Long-Term Memory
+
+Long-term memory (Mem0) is suitable for persistent, vectorized retrieval and summarization in multi-agent scenarios. It supports LLM-based summarization and vector storage.
+
+**Usage Example:**
+
+```python
+from aworld.core.memory import MemoryConfig, MemoryItem
+from aworld.memory.main import MemoryFactory
+
+# Create Mem0 config (requires mem0 and related dependencies)
+memory_config = MemoryConfig(
+    provider="mem0",
+    enable_summary=True,           # Enable summarization
+    summary_rounds=5,              # Generate a summary every 5 rounds
+    embedder_provider="huggingface", # Embedding model provider
+    embedder_model="all-MiniLM-L6-v2", # Embedding model name
+    embedder_dims=384
+)
+# Initialize Memory
+memory = MemoryFactory.from_config(memory_config)
+
+# Add a memory item
+memory.add(MemoryItem(content="The agent visited Hangzhou.", metadata={"user_id": "u1"}, tags=["travel"]))
+
+# Get all memory items
+all_memories = memory.get_all()
+for item in all_memories:
+    print(item.content)
+```
+
+> Note: To use mem0, you must install `mem0` and `sentence-transformers` in advance, and configure the required LLM environment variables.
+
+### CheckPoint
+TODO

aworld/memory/__init__.py

@@ -0,0 +1,2 @@
+
+

aworld/memory/main.py

@@ -0,0 +1,390 @@
+# coding: utf-8
+# Copyright (c) 2025 inclusionAI.
+import abc
+import asyncio
+import json
+import os
+from typing import Optional
+
+from aworld.config import ConfigDict
+from aworld.core.memory import MemoryBase, MemoryItem, MemoryStore, MemoryConfig
+from aworld.logs.util import logger
+from aworld.models.llm import get_llm_model, acall_llm_model
+
+
+class InMemoryMemoryStore(MemoryStore):
+    def __init__(self):
+        self.memory_items = []
+
+    def add(self, memory_item: MemoryItem):
+        self.memory_items.append(memory_item)
+
+    def get(self, memory_id) -> Optional[MemoryItem]:
+        return next((item for item in self.memory_items if item.id == memory_id), None)
+
+    def get_first(self, filters: dict = None) -> Optional[MemoryItem]:
+        """Get the first memory item."""
+        filtered_items = self.get_all(filters)
+        if len(filtered_items) == 0:
+            return None
+        return filtered_items[0]
+
+    def total_rounds(self, filters: dict = None) -> int:
+        """Get the total number of rounds."""
+        return len(self.get_all(filters))
+
+    def get_all(self, filters: dict = None) -> list[MemoryItem]:
+        """Filter memory items based on filters."""
+        filtered_items = [item for item in self.memory_items if self._filter_memory_item(item, filters)]
+        return filtered_items
+
+    def _filter_memory_item(self, memory_item: MemoryItem, filters: dict = None) -> bool:
+        if memory_item.deleted:
+            return False
+        if filters is None:
+            return True
+        if filters.get('user_id') is not None:
+            if memory_item.metadata.get('user_id') is None:
+                return False
+            if memory_item.metadata.get('user_id') != filters['user_id']:
+                return False
+        if filters.get('agent_id') is not None:
+            if memory_item.metadata.get('agent_id') is None:
+                return False
+            if memory_item.metadata.get('agent_id') != filters['agent_id']:
+                return False
+        if filters.get('task_id') is not None:
+            if memory_item.metadata.get('task_id') is None:
+                return False
+            if memory_item.metadata.get('task_id') != filters['task_id']:
+                return False
+        if filters.get('session_id') is not None:
+            if memory_item.metadata.get('session_id') is None:
+                return False
+            if memory_item.metadata.get('session_id') != filters['session_id']:
+                return False
+        if filters.get('memory_type') is not None:
+            if memory_item.memory_type is None:
+                return False
+            if memory_item.memory_type != filters['memory_type']:
+                return False
+        return True
+
+    def get_last_n(self, last_rounds, filters: dict = None) -> list[MemoryItem]:
+        return self.memory_items[-last_rounds:]  # Get the last n items
+
+    def update(self, memory_item: MemoryItem):
+        for index, item in enumerate(self.memory_items):
+            if item.id == memory_item.id:
+                self.memory_items[index] = memory_item  # Update the item in the list
+                break
+
+    def delete(self, memory_id):
+        exists = self.get(memory_id)
+        if exists:
+            exists.deleted = True
+
+    def history(self, memory_id) -> list[MemoryItem] | None:
+        exists = self.get(memory_id)
+        if exists:
+            return exists.histories
+        return None
+
+
+class MemoryFactory:
+
+    @classmethod
+    def from_config(cls, config: MemoryConfig) -> "MemoryBase":
+        """
+        Initialize a Memory instance from a configuration dictionary.
+
+        Args:
+            config (dict): Configuration dictionary.
+
+        Returns:
+            InMemoryStorageMemory: Memory instance.
+        """
+        if config.provider == "inmemory":
+            return InMemoryStorageMemory(
+                memory_store=InMemoryMemoryStore(),
+                config=config,
+                enable_summary=config.enable_summary,
+                summary_rounds=config.summary_rounds
+            )
+        elif config.provider == "mem0":
+            from aworld.memory.mem0.mem0_memory import Mem0Memory
+            return Mem0Memory(
+                memory_store=InMemoryMemoryStore(),
+                config=config
+            )
+        else:
+            raise ValueError(f"Invalid memory store type: {config.get('memory_store')}")
+
+
+class Memory(MemoryBase):
+    __metaclass__ = abc.ABCMeta
+
+    def __init__(self, memory_store: MemoryStore, config: MemoryConfig, **kwargs):
+        self.memory_store = memory_store
+        self.config = config
+        self._llm_instance = None
+
+    @property
+    def default_llm_instance(self):
+        def get_env(key: str, default_key: str, default_val: object=None):
+            return os.getenv(key) if os.getenv(key) else os.getenv(default_key, default_val)
+
+        if not self._llm_instance:
+            self._llm_instance = get_llm_model(conf=ConfigDict({
+                "llm_model_name": get_env("MEM_LLM_MODEL_NAME", "LLM_MODEL_NAME"),
+                "llm_api_key": get_env("MEM_LLM_API_KEY", "LLM_MODEL_NAME") ,
+                "llm_base_url": get_env("MEM_LLM_BASE_URL", 'LLM_BASE_URL'),
+                "temperature": get_env("MEM_LLM_TEMPERATURE", "MEM_LLM_TEMPERATURE", 1.0),
+                "streaming": 'False'
+            }))
+        return self._llm_instance
+
+    def _build_history_context(self, messages) -> str:
+        """Build the history context string from a list of messages.
+
+        Args:
+            messages: List of message objects with 'role', 'content', and optional 'tool_calls'.
+        Returns:
+            Concatenated context string.
+        """
+        history_context = ""
+        for item in messages:
+            history_context += (f"\n\n{item['role']}: {item['content']}, "
+                                f"{'tool_calls:' + json.dumps(item['tool_calls']) if 'tool_calls' in item and item['tool_calls'] else ''}")
+        return history_context
+
+    async def _call_llm_summary(self, summary_messages: list) -> str:
+        """Call LLM to generate summary and log the process.
+
+        Args:
+            summary_messages: List of messages to send to LLM.
+        Returns:
+            Summary content string.
+        """
+        logger.info(f"🤔 [Summary] Creating summary memory, history messages: {summary_messages}")
+        llm_response = await acall_llm_model(
+            self.default_llm_instance,
+            messages=summary_messages,
+            stream=False
+        )
+        logger.info(f'🤔 [Summary] summary_content: result is {llm_response.content[:400] + "...truncated"} ')
+        return llm_response.content
+
+    def _get_parsed_history_messages(self, history_items: list[MemoryItem]) -> list[dict]:
+        """Get and format history messages for summary.
+
+        Args:
+            history_items: list[MemoryItem]
+        Returns:
+            List of parsed message dicts
+        """
+        parsed_messages = [
+            {
+                'role': message.metadata['role'],
+                'content': message.content,
+                'tool_calls': message.metadata.get('tool_calls') if message.metadata.get('tool_calls') else None
+            }
+            for message in history_items]
+        return parsed_messages
+
+    async def async_gen_multi_rounds_summary(self, to_be_summary: list[MemoryItem]) -> str:
+        logger.info(
+            f"🤔 [Summary] Creating summary memory, history messages")
+        if len(to_be_summary) == 0:
+            return ""
+        parsed_messages = self._get_parsed_history_messages(to_be_summary)
+        history_context = self._build_history_context(parsed_messages)
+
+        summary_messages = [
+            {"role": "user", "content": self.config.summary_prompt.format(context=history_context)}
+        ]
+
+        return await self._call_llm_summary(summary_messages)
+
+    async def async_gen_summary(self, filters: dict, last_rounds: int) -> str:
+        """A tool for summarizing the conversation history."""
+
+        logger.info(f"🤔 [Summary] Creating summary memory, history messages [filters -> {filters}, "
+                    f"last_rounds -> {last_rounds}]")
+        history_items = self.memory_store.get_last_n(last_rounds, filters=filters)
+        if len(history_items) == 0:
+            return ""
+        parsed_messages = self._get_parsed_history_messages(history_items)
+        history_context = self._build_history_context(parsed_messages)
+
+        summary_messages = [
+            {"role": "user", "content": self.config.summary_prompt.format(context=history_context)}
+        ]
+
+        return await self._call_llm_summary(summary_messages)
+
+    async def async_gen_cur_round_summary(self, to_be_summary: MemoryItem, filters: dict, last_rounds: int) -> str:
+        if self.config.enable_summary and len(to_be_summary.content) < self.config.summary_single_context_length:
+            return to_be_summary.content
+
+        logger.info(f"🤔 [Summary] Creating summary memory, history messages [filters -> {filters}, "
+                    f"last_rounds -> {last_rounds}]: to be summary content is {to_be_summary.content}")
+        history_items = self.memory_store.get_last_n(last_rounds, filters=filters)
+        if len(history_items) == 0:
+            return ""
+        parsed_messages = self._get_parsed_history_messages(history_items)
+
+        # Append the to_be_summary
+        parsed_messages.append({
+            "role": to_be_summary.metadata['role'],
+            "content": f"{to_be_summary.content}",
+            'tool_call_id': to_be_summary.metadata['tool_call_id'],
+        })
+        history_context = self._build_history_context(parsed_messages)
+
+        summary_messages = [
+            {"role": "user", "content": self.config.summary_prompt.format(context=history_context)}
+        ]
+
+        return await self._call_llm_summary(summary_messages)
+
+    def search(self, query, limit=100, filters=None) -> Optional[list[MemoryItem]]:
+        pass
+
+
+class InMemoryStorageMemory(Memory):
+    def __init__(self, memory_store: MemoryStore, config: MemoryConfig, enable_summary: bool = True, **kwargs):
+        super().__init__(memory_store=memory_store, config=config)
+        self.summary = {}
+        self.summary_rounds = self.config.summary_rounds
+        self.enable_summary = self.config.enable_summary
+
+    def add(self, memory_item: MemoryItem, filters: dict = None):
+        self.memory_store.add(memory_item)
+
+        # Check if we need to create or update summary
+        if self.enable_summary:
+            total_rounds = len(self.memory_store.get_all())
+            if total_rounds > self.summary_rounds:
+                self._create_or_update_summary(total_rounds)
+
+    def _create_or_update_summary(self, total_rounds: int):
+        """Create or update summary based on current total rounds.
+
+        Args:
+            total_rounds (int): Total number of rounds.
+        """
+        summary_index = int(total_rounds / self.summary_rounds)
+        start = (summary_index - 1) * self.summary_rounds
+        end = total_rounds - self.summary_rounds
+
+        # Ensure we have valid start and end indices
+        start = max(0, start)
+        end = max(start, end)
+
+        # Get the memory items to summarize
+        items_to_summarize = self.memory_store.get_all()[start:end + 1]
+        print(f"{total_rounds}start: {start}, end: {end},")
+
+        # Create summary content
+        summary_content = self._summarize_items(items_to_summarize, summary_index)
+
+        # Create the range key
+        range_key = f"{start}_{end}"
+
+        # Check if summary for this range already exists
+        if range_key in self.summary:
+            # Update existing summary
+            self.summary[range_key].content = summary_content
+            self.summary[range_key].updated_at = None  # This will update the timestamp
+        else:
+            # Create new summary
+            summary_item = MemoryItem(
+                content=summary_content,
+                metadata={
+                    "summary_index": summary_index,
+                    "start_round": start,
+                    "end_round": end,
+                    "role": "system"
+                },
+                tags=["summary"]
+            )
+            self.summary[range_key] = summary_item
+
+    def _summarize_items(self, items: list[MemoryItem], summary_index: int) -> str:
+        """Summarize a list of memory items.
+
+        Args:
+            items (list[MemoryItem]): List of memory items to summarize.
+            summary_index (int): Summary index.
+
+        Returns:
+            str: Summary content.
+        """
+        # This is a placeholder. In a real implementation, you might use an LLM or other method
+        # to create a meaningful summary of the content
+        return asyncio.run(self.async_gen_multi_rounds_summary(items))
+
+    def update(self, memory_item: MemoryItem):
+        self.memory_store.update(memory_item)
+
+    def delete(self, memory_id):
+        self.memory_store.delete(memory_id)
+
+    def get(self, memory_id) -> Optional[MemoryItem]:
+        return self.memory_store.get(memory_id)
+
+    def get_all(self, filters: dict = None) -> list[MemoryItem]:
+        return self.memory_store.get_all()
+
+    def get_last_n(self, last_rounds, add_first_message=True, filters: dict = None) -> list[MemoryItem]:
+        """Get last n memories.
+
+        Args:
+            last_rounds (int): Number of memories to retrieve.
+            add_first_message (bool):
+
+        Returns:
+            list[MemoryItem]: List of latest memories.
+        """
+        memory_items = self.memory_store.get_last_n(last_rounds)
+        while len(memory_items) > 0 and memory_items[0].metadata and "tool_call_id" in memory_items[0].metadata and \
+                memory_items[0].metadata["tool_call_id"]:
+            last_rounds = last_rounds + 1
+            memory_items = self.memory_store.get_last_n(last_rounds)
+
+        # If summary is disabled or no summaries exist, return just the last_n_items
+        if not self.enable_summary or not self.summary:
+            return memory_items
+
+        # Calculate the range for relevant summaries
+        all_items = self.memory_store.get_all()
+        total_items = len(all_items)
+        end_index = total_items - last_rounds
+
+        # Get complete summaries
+        result = []
+        complete_summary_count = end_index // self.summary_rounds
+
+        # Get complete summaries
+        for i in range(complete_summary_count):
+            range_key = f"{i * self.summary_rounds}_{(i + 1) * self.summary_rounds - 1}"
+            if range_key in self.summary:
+                result.append(self.summary[range_key])
+
+        # Get the last incomplete summary if exists
+        remaining_items = end_index % self.summary_rounds
+        if remaining_items > 0:
+            start = complete_summary_count * self.summary_rounds
+            range_key = f"{start}_{end_index - 1}"
+            if range_key in self.summary:
+                result.append(self.summary[range_key])
+
+        # Add the last n items
+        result.extend(memory_items)
+
+        # Add first user input
+        if add_first_message and last_rounds < self.memory_store.total_rounds():
+            memory_items.insert(0, self.memory_store.get_first())
+
+        return result

aworld/memory/models.py

@@ -0,0 +1,197 @@
+from pydantic import BaseModel, ConfigDict, Field
+from aworld.core.memory import MemoryItem
+from typing import Any, Dict, List, Optional, Literal
+
+from aworld.models.model_response import ToolCall
+
+class MessageMetadata(BaseModel):
+    """
+    Metadata for memory messages, including user, session, task, and agent information.
+    Args:
+        user_id (str): The ID of the user.
+        session_id (str): The ID of the session.
+        task_id (str): The ID of the task.
+        agent_id (str): The ID of the agent.
+    """
+    user_id: str = Field(description="The ID of the user")
+    session_id: str = Field(description="The ID of the session")
+    task_id: str = Field(description="The ID of the task")
+    agent_id: str = Field(description="The ID of the agent")
+    agent_name: str = Field(description="The name of the agent")
+
+    model_config = ConfigDict(extra="allow")
+
+    @property
+    def to_dict(self) -> Dict[str, Any]:
+        return self.model_dump()
+
+class AgentExperienceItem(BaseModel):
+    skill: str = Field(description="The skill demonstrated in the experience")
+    actions: List[str] = Field(description="The actions taken by the agent")
+
+class AgentExperience(MemoryItem):
+    """
+    Represents an agent's experience, including skills and actions.
+    All custom attributes are stored in content and metadata.
+    Args:
+        agent_id (str): The ID of the agent.
+        skill (str): The skill demonstrated in the experience.
+        actions (List[str]): The actions taken by the agent.
+        metadata (Optional[Dict[str, Any]]): Additional metadata.
+    """
+    def __init__(self, agent_id: str, skill: str, actions: List[str], metadata: Optional[Dict[str, Any]] = None) -> None:
+        meta = metadata.copy() if metadata else {}
+        meta['agent_id'] = agent_id
+        agent_experience = AgentExperienceItem(skill=skill, actions=actions)
+        super().__init__(content=agent_experience, metadata=meta, memory_type="agent_experience")
+
+    @property
+    def agent_id(self) -> str:
+        return self.metadata['agent_id']
+
+    @property
+    def skill(self) -> str:
+        return self.content.skill
+
+    @property
+    def actions(self) -> List[str]:
+        return self.content.actions
+
+class UserProfileItem(BaseModel):
+    key: str = Field(description="The key of the profile")
+    value: Any = Field(description="The value of the profile")
+
+class UserProfile(MemoryItem):
+    """
+    Represents a user profile key-value pair.
+    All custom attributes are stored in content and metadata.
+    Args:
+        user_id (str): The ID of the user.
+        key (str): The profile key.
+        value (Any): The profile value.
+        metadata (Optional[Dict[str, Any]]): Additional metadata.
+    """
+    def __init__(self, user_id: str, key: str, value: Any, metadata: Optional[Dict[str, Any]] = None) -> None:
+        meta = metadata.copy() if metadata else {}
+        meta['user_id'] = user_id
+        user_profile = UserProfileItem(key=key, value=value)
+        super().__init__(content=user_profile, metadata=meta, memory_type="user_profile")
+
+    @property
+    def user_id(self) -> str:
+        return self.metadata['user_id']
+
+    @property
+    def key(self) -> str:
+        return self.content.key
+
+    @property
+    def value(self) -> Any:
+        return self.content.value
+
+class MemoryMessage(MemoryItem):
+    """
+    Represents a memory message with role, user, session, task, and agent information.
+    Args:
+        role (str): The role of the message sender.
+        metadata (MessageMetadata): Metadata object containing user, session, task, and agent IDs.
+        content (Optional[Any]): Content of the message.
+    """
+    def __init__(self, role: str, metadata: MessageMetadata, content: Optional[Any] = None) -> None:
+        meta = metadata.to_dict
+        meta['role'] = role
+        super().__init__(content=content, metadata=meta, memory_type="message")
+
+    @property
+    def role(self) -> str:
+        return self.metadata['role']
+
+    @property
+    def user_id(self) -> str:
+        return self.metadata['user_id']
+
+    @property
+    def session_id(self) -> str:
+        return self.metadata['session_id']
+
+    @property
+    def task_id(self) -> str:
+        return self.metadata['task_id']
+
+    @property
+    def agent_id(self) -> str:
+        return self.metadata['agent_id']
+
+class SystemMessage(MemoryMessage):
+    """
+    Represents a system message with role and content.
+    Args:
+        metadata (MessageMetadata): Metadata object containing user, session, task, and agent IDs.
+        content (str): The content of the message.
+    """
+    def __init__(self, content: str, metadata: MessageMetadata) -> None:
+        super().__init__(role="system", metadata=metadata, content=content)
+
+    @property
+    def content(self) -> str:
+        return self._content
+
+class HumanMessage(MemoryMessage):
+    """
+    Represents a human message with role and content.
+    Args:
+        metadata (MessageMetadata): Metadata object containing user, session, task, and agent IDs.
+        content (str): The content of the message.
+    """
+    def __init__(self, metadata: MessageMetadata, content: str) -> None:
+        super().__init__(role="human", metadata=metadata, content=content)
+
+    @property
+    def content(self) -> str:
+        return self._content
+
+class AIMessage(MemoryMessage):
+    """
+    Represents an AI message with role and content.
+    Args:
+        metadata (MessageMetadata): Metadata object containing user, session, task, and agent IDs.
+        content (str): The content of the message.
+    """
+    def __init__(self, content: str, tool_calls: List[ToolCall], metadata: MessageMetadata) -> None:
+        meta = metadata.to_dict
+        meta['tool_calls'] = [tool_call.to_dict() for tool_call in tool_calls]
+        super().__init__(role="assistant", metadata=MessageMetadata(**meta), content=content)
+
+    @property
+    def content(self) -> str:
+        return self._content
+    
+    @property
+    def tool_calls(self) -> List[ToolCall]:
+        return [ToolCall(**tool_call) for tool_call in self.metadata['tool_calls']]
+
+class ToolMessage(MemoryMessage):
+    """
+    Represents a tool message with role, content, tool_call_id, and status.
+    Args:
+        metadata (MessageMetadata): Metadata object containing user, session, task, and agent IDs.
+        tool_call_id (str): The ID of the tool call.
+        status (Literal["success", "error"]): The status of the tool call.
+        content (str): The content of the message.
+    """
+    def __init__(self, tool_call_id: str, content: str, status: Literal["success", "error"] = "success", metadata: MessageMetadata = None) -> None:
+        metadata.tool_call_id = tool_call_id
+        metadata.status = status
+        super().__init__(role="tool", metadata=metadata, content=content)
+
+    @property
+    def tool_call_id(self) -> str:
+        return self.metadata['tool_call_id']
+
+    @property
+    def status(self) -> str:
+        return self.metadata['status']
+
+    @property
+    def content(self) -> str:
+        return self._content

aworld/memory/utils.py

@@ -0,0 +1,33 @@
+import tiktoken
+from aworld.logs.util import logger
+
+# TODO: merge to `models` package
+
+MODEL_TO_ENCODING = {
+    "gpt-3.5-turbo": "cl100k_base",
+    "gpt-4": "cl100k_base",
+    "text-davinci-003": "p50k_base",
+    "text-embedding-ada-002": "cl100k_base",
+    "text-curie-001": "r50k_base",
+    "text-babbage-001": "r50k_base",
+    "text-ada-001": "r50k_base",
+}
+
+def get_encoding_for_model(model_name: str) -> tiktoken.Encoding:
+    """
+    Automatically select the corresponding encoder based on the model name.
+    """
+    encoding_name = MODEL_TO_ENCODING.get(model_name)
+    if encoding_name is None:
+        logger.warning(f"model '{model_name}' not found in mapping table.")
+        return "cl100k_base"
+    return encoding_name
+
+def count_tokens(model_name: str, content: str):
+    encoding = tiktoken.get_encoding(get_encoding_for_model(model_name))
+
+    tokens = encoding.encode(content)
+
+    token_count = len(tokens)
+
+    return token_count