refactor(memory): restructure memory tools with new identity and meta memory features

This commit is contained in:
jinli.yl 2026-01-23 00:33:16 +08:00
parent 3c272ac859
commit c927d264e1
25 changed files with 260 additions and 495 deletions

View file

@ -1,16 +1,24 @@
"""memory tools"""
from .add_history import AddHistory
from .base_memory_tool import BaseMemoryTool
from .read_history import ReadHistory
from .read_user_profile import ReadUserProfile
from .update_user_profile import UpdateUserProfile
from .history.add_history import AddHistory
from .history.read_history import ReadHistory
from .identity.add_identity import AddIdentity
from .identity.read_identity import ReadIdentity
from .meta.add_meta_memory import AddMetaMemory
from .meta.read_meta_memory import ReadMetaMemory
from .user_profile.read_user_profile import ReadUserProfile
from .user_profile.update_user_profile import UpdateUserProfile
from ...core import R
__all__ = [
"AddHistory",
"BaseMemoryTool",
"AddHistory",
"ReadHistory",
"AddIdentity",
"ReadIdentity",
"AddMetaMemory",
"ReadMetaMemory",
"ReadUserProfile",
"UpdateUserProfile",
]

View file

@ -2,10 +2,10 @@
from loguru import logger
from .base_memory_tool import BaseMemoryTool
from ...core.enumeration import MemoryType
from ...core.schema import ToolCall, MemoryNode, Message
from ...core.utils import format_messages
from ..base_memory_tool import BaseMemoryTool
from ....core.enumeration import MemoryType
from ....core.schema import ToolCall, MemoryNode, Message
from ....core.utils import format_messages
class AddHistory(BaseMemoryTool):
@ -25,7 +25,7 @@ class AddHistory(BaseMemoryTool):
"properties": {},
"required": [],
},
}
},
)
async def execute(self):

View file

@ -2,8 +2,8 @@
from loguru import logger
from .base_memory_tool import BaseMemoryTool
from ...core.schema import MemoryNode, ToolCall
from ..base_memory_tool import BaseMemoryTool
from ....core.schema import MemoryNode, ToolCall
class ReadHistory(BaseMemoryTool):

View file

@ -0,0 +1,42 @@
"""Add identity memory tool"""
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ....core.schema import ToolCall
class AddIdentity(BaseMemoryTool):
"""Tool to add or update agent identity memory"""
def __init__(self, **kwargs):
kwargs["enable_multiple"] = False
super().__init__(**kwargs)
def _build_tool_call(self) -> ToolCall:
return ToolCall(
**{
"description": "add or update agent identity memory.",
"parameters": {
"type": "object",
"properties": {
"identity_memory": {
"type": "string",
"description": "Agent identity content, such as role, personality, or current state.",
},
},
"required": ["identity_memory"],
},
},
)
async def execute(self):
identity_memory = self.context.get("identity_memory", "")
if not identity_memory:
logger.warning("No valid identity memory provided")
return "No valid identity memory provided for update."
self.local_memory.save("identity_memory", identity_memory)
logger.info(f"Successfully updated identity memory: {identity_memory}")
return "Successfully updated identity memory."

View file

@ -0,0 +1,36 @@
"""Read identity memory tool"""
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ....core.schema import ToolCall
class ReadIdentity(BaseMemoryTool):
"""Tool to read agent identity memory"""
def __init__(self, **kwargs):
kwargs["enable_multiple"] = False
super().__init__(**kwargs)
def _build_tool_call(self) -> ToolCall:
return ToolCall(
**{
"description": "read agent identity memory.",
"parameters": {
"type": "object",
"properties": {},
"required": [],
},
},
)
async def execute(self):
identity_memory = self.local_memory.load("identity_memory")
if not identity_memory:
logger.info("No identity memory found")
return "No identity memory found."
logger.info(f"Read identity memory: {identity_memory}")
return f"Identity\n{identity_memory}"

View file

@ -0,0 +1,89 @@
"""Add meta memory tool"""
import json
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ....core.enumeration import MemoryType
from ....core.schema import ToolCall
class AddMetaMemory(BaseMemoryTool):
"""Tool to add memory metadata entries to meta storage"""
def __init__(self, **kwargs):
kwargs["enable_multiple"] = True
super().__init__(**kwargs)
def _build_multiple_tool_call(self) -> ToolCall:
"""Build and return the multiple tool call schema"""
return ToolCall(
**{
"description": "add memory metadata entries to register memory types and targets. "
"Before using, verify Main Agent's Meta Memory doesn't already contain the "
"same memory_type(memory_target) combinations.",
"parameters": {
"type": "object",
"properties": {
"meta_memories": {
"type": "array",
"description": "List of memory metadata entries to add",
"items": {
"type": "object",
"properties": {
"memory_type": {
"type": "string",
"description": "Type of memory: 'personal' for person-specific preferences, "
"'procedural' for how-to knowledge",
"enum": [MemoryType.PERSONAL.value, MemoryType.PROCEDURAL.value],
},
"memory_target": {
"type": "string",
"description": "Target identifier, "
"e.g., person's name ('John') or domain ('deployment')",
},
},
"required": ["memory_type", "memory_target"],
},
},
},
"required": ["meta_memories"],
},
},
)
async def execute(self):
existing_memories: list[dict] = self.local_memory.load("meta_memories") or []
existing_set = {(m["memory_type"], m["memory_target"]) for m in existing_memories}
# Filter and build new memories to add
new_memories: list[dict] = []
meta_memories: list[dict] = self.context.get("meta_memories", [])
for mem in meta_memories:
memory_type = mem.get("memory_type", "")
memory_target = mem.get("memory_target", "")
# Check if valid and not duplicate
if (
memory_type in [MemoryType.PERSONAL.value, MemoryType.PROCEDURAL.value]
and memory_target
and (memory_type, memory_target) not in existing_set
):
new_memories.append({"memory_type": memory_type, "memory_target": memory_target})
existing_set.add((memory_type, memory_target))
if not new_memories:
output = "No new meta memories to add (all entries already exist or invalid)."
logger.info(output)
return output
# Merge, sort and save
all_memories = sorted(existing_memories + new_memories, key=lambda m: (m["memory_type"], m["memory_target"]))
self.local_memory.save("meta_memories", all_memories)
# Format output
output = f"Successfully update meta memory entries: {json.dumps(new_memories, ensure_ascii=False)}"
logger.info(output)
return output

View file

@ -0,0 +1,66 @@
"""Read meta memory tool"""
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ....core.enumeration import MemoryType
from ....core.schema import ToolCall
class ReadMetaMemory(BaseMemoryTool):
"""Tool to read memory metadata from meta storage"""
TYPE_DESC_DICT = {
MemoryType.IDENTITY.value: "self-cognition memory storing agent's identity and state",
MemoryType.PERSONAL.value: "person-specific memory storing preferences and context",
MemoryType.PROCEDURAL.value: "procedural memory storing how-to knowledge and processes",
}
def __init__(self, enable_identity_memory: bool = False, **kwargs):
kwargs["enable_multiple"] = False
super().__init__(**kwargs)
self.enable_identity_memory = enable_identity_memory
def _build_tool_call(self) -> ToolCall:
return ToolCall(
**{
"description": "read memory metadata registry to see what types of memories are being tracked.",
"parameters": {
"type": "object",
"properties": {},
"required": [],
},
},
)
async def execute(self):
# Load and filter meta memories
result = self.local_memory.load("meta_memories")
all_memories = result if result is not None else []
memories = [
m for m in all_memories if m.get("memory_type") in [MemoryType.PERSONAL.value, MemoryType.PROCEDURAL.value]
]
if self.enable_identity_memory:
memories.append(
{
"memory_type": MemoryType.IDENTITY.value,
"memory_target": "self",
},
)
# Format output
if memories:
lines = [
f"- {m['memory_type']}({m['memory_target']}): {self.TYPE_DESC_DICT.get(m['memory_type'], '')}"
for m in memories
]
output = "\n".join(lines)
logger.info(f"Retrieved {len(memories)} meta memory entries")
else:
output = "No memory metadata found."
logger.info(output)
return output

View file

@ -4,9 +4,9 @@ from typing import Literal
from loguru import logger
from .base_memory_tool import BaseMemoryTool
from ...core.schema import ToolCall
from ...core.schema.memory_node import MemoryNode
from ..base_memory_tool import BaseMemoryTool
from ....core.schema import ToolCall
from ....core.schema.memory_node import MemoryNode
class ReadUserProfile(BaseMemoryTool):

View file

@ -2,10 +2,10 @@
from loguru import logger
from .base_memory_tool import BaseMemoryTool
from ...core.schema import ToolCall
from ...core.schema.memory_node import MemoryNode
from ...core.utils import deduplicate_memories
from ..base_memory_tool import BaseMemoryTool
from ....core.schema import ToolCall
from ....core.schema.memory_node import MemoryNode
from ....core.utils import deduplicate_memories
class UpdateUserProfile(BaseMemoryTool):

View file

@ -1,50 +0,0 @@
"""Add history memory operation."""
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ...core.context import C
from ...core.enumeration import MemoryType
from ...core.schema import ToolCall, Message
from ...core.utils import format_messages
@C.register_op()
class AddHistoryMemory(BaseMemoryTool):
"""Add history memory from conversation messages."""
def _build_tool_call(self) -> ToolCall:
return ToolCall(
**{
"description": self.get_prompt("tool"),
"parameters": {
"type": "object",
"properties": {
"messages": {
"type": "array",
"description": self.get_prompt("messages"),
"items": {"type": "object"},
},
},
"required": ["messages"],
},
},
)
async def execute(self):
messages: list[Message | dict] = self.context.get("messages", [])
if not messages:
self.output = "No messages provided for addition."
return
messages = [Message(**m) if isinstance(m, dict) else m for m in messages]
memory_content = format_messages(messages)
memory_node = self._build_memory_node(memory_content=memory_content, memory_type=MemoryType.HISTORY)
vector_node = memory_node.to_vector_node()
await self.vector_store.delete(vector_ids=[vector_node.vector_id])
await self.vector_store.insert(nodes=[vector_node])
self.memory_nodes.append(memory_node)
self.output = "Successfully added history memory to vector_store."
logger.info(self.output)

View file

@ -1,14 +0,0 @@
tool: |
Add history memory from conversation messages.
tool_multiple: |
Add multiple history memories in a single operation.
messages: |
List of message objects with 'role' and 'content' fields.
metadata: |
Optional metadata (time, session_id, topic, etc.).
histories: |
List of history objects, each with messages and optional metadata.

View file

@ -1,65 +0,0 @@
"""Read history memory operation."""
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ...core.context import C
from ...core.schema import MemoryNode
@C.register_op()
class ReadHistoryMemory(BaseMemoryTool):
"""Read history memories by IDs."""
def _build_parameters(self) -> dict:
return {
"type": "object",
"properties": {
"ref_memory_id": {
"type": "string",
"description": self.get_prompt("ref_memory_id"),
},
},
"required": ["ref_memory_id"],
}
def _build_multiple_parameters(self) -> dict:
return {
"type": "object",
"properties": {
"ref_memory_ids": {
"type": "array",
"description": self.get_prompt("ref_memory_ids"),
"items": {"type": "string"},
},
},
"required": ["ref_memory_ids"],
}
async def execute(self):
if self.enable_multiple:
ref_memory_ids: list[str] = self.context.get("ref_memory_ids", [])
else:
ref_memory_id = self.context.get("ref_memory_id", "")
ref_memory_ids: list[str] = [ref_memory_id] if ref_memory_id else []
# Remove empty IDs and duplicates
ref_memory_ids = [mid for mid in ref_memory_ids if mid]
ref_memory_ids = list(dict.fromkeys(ref_memory_ids)) # Remove duplicates while preserving order
if not ref_memory_ids:
self.output = "No valid reference memory IDs provided for reading."
logger.warning(self.output)
return
# Query original history dialogues by ref_memory_id
nodes = await self.vector_store.get(vector_ids=ref_memory_ids)
if not nodes:
self.output = "No history memories found with the provided reference IDs."
logger.warning(self.output)
return
memories: list[MemoryNode] = [MemoryNode.from_vector_node(n) for n in nodes]
self.output = "---\n".join([m.content for m in memories])
logger.info(f"Successfully read {len(memories)} history memories by reference IDs.")

View file

@ -1,11 +0,0 @@
tool: |
Read original history dialogue by reference memory ID.
tool_multiple: |
Read multiple original history dialogues by reference memory IDs.
ref_memory_id: |
Reference memory ID to query the original history dialogue.
ref_memory_ids: |
List of reference memory IDs to query the original history dialogues. Please provide unique IDs without duplicates.

View file

@ -1,27 +0,0 @@
"""Read identity memory operation."""
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ...core.context import C
@C.register_op()
class ReadIdentityMemory(BaseMemoryTool):
"""Read identity memory for agent self-cognition."""
def __init__(self, **kwargs):
kwargs["enable_multiple"] = False
super().__init__(**kwargs)
def _build_parameters(self) -> dict:
return {
"type": "object",
"properties": {},
"required": [],
}
async def execute(self):
identity_memory = self.meta_memory.load("identity_memory") or ""
self.output = identity_memory or "No identity memory found."
logger.info(self.output)

View file

@ -1,3 +0,0 @@
tool: |
Read the identity memory for the agent.
Retrieve self-cognition information such as identity, role, personality, or current state.

View file

@ -1,39 +0,0 @@
"""Update identity memory operation."""
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ...core.context import C
@C.register_op()
class UpdateIdentityMemory(BaseMemoryTool):
"""Update identity memory for agent self-cognition."""
def __init__(self, **kwargs):
kwargs["enable_multiple"] = False
super().__init__(**kwargs)
def _build_parameters(self) -> dict:
return {
"type": "object",
"properties": {
"identity_memory": {
"type": "string",
"description": self.get_prompt("identity_memory"),
},
},
"required": ["identity_memory"],
}
async def execute(self):
identity_memory = self.context.get("identity_memory", "")
if not identity_memory:
self.output = "No valid identity memory provided for update."
logger.warning(self.output)
return
self.meta_memory.save("identity_memory", identity_memory)
self.output = "Successfully updated identity memory."
logger.info(self.output)

View file

@ -1,7 +0,0 @@
tool: |
Update the identity memory for the agent.
Store self-cognition information such as identity, role, personality, or current state.
identity_memory: |
The identity memory content to store.
Should be a clear statement capturing the agent's self-cognition or current state.

View file

@ -1,121 +0,0 @@
"""Add meta memory operation for adding memory metadata."""
import json
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ...core.context import C
from ...core.enumeration import MemoryType
@C.register_op()
class AddMetaMemory(BaseMemoryTool):
"""Add memory metadata (memory_type and memory_target) to meta storage.
Supports single/multiple addition modes via `enable_multiple` parameter.
"""
def _build_item_schema(self) -> tuple[dict, list[str]]:
"""Build shared schema properties and required fields for meta memory items.
Returns:
Tuple of (properties dict, required fields list).
"""
properties = {
"memory_type": {
"type": "string",
"description": self.get_prompt("memory_type"),
"enum": [MemoryType.PERSONAL.value, MemoryType.PROCEDURAL.value],
},
"memory_target": {
"type": "string",
"description": self.get_prompt("memory_target"),
},
}
required = ["memory_type", "memory_target"]
return properties, required
def _build_parameters(self) -> dict:
"""Build input schema for single meta memory addition."""
properties, required = self._build_item_schema()
return {
"type": "object",
"properties": properties,
"required": required,
}
def _build_multiple_parameters(self) -> dict:
"""Build input schema for multiple meta memory addition."""
item_properties, required_fields = self._build_item_schema()
return {
"type": "object",
"properties": {
"meta_memories": {
"type": "array",
"description": self.get_prompt("meta_memories"),
"items": {
"type": "object",
"properties": item_properties,
"required": required_fields,
},
},
},
"required": ["meta_memories"],
}
def _load_meta_memories(self) -> list[dict]:
"""Load existing meta memories from cache."""
return self.meta_memory.load("meta_memories") or []
def _save_meta_memories(self, memories: list[dict]) -> bool:
"""Save meta memories to cache."""
return self.meta_memory.save("meta_memories", memories)
@staticmethod
def _filter_memory_type_target(memory_type: str, memory_target: str, existing_set: set) -> bool:
result = (
memory_type in [MemoryType.PERSONAL.value, MemoryType.PROCEDURAL.value]
and memory_target
and (memory_type, memory_target) not in existing_set
)
if result:
existing_set.add((memory_type, memory_target))
return result
async def execute(self):
"""Execute addition: load existing, merge with new, and save.
Duplicates (same memory_type and memory_target) are skipped.
"""
existing_memories: list[dict] = self._load_meta_memories()
existing_set = {(m["memory_type"], m["memory_target"]) for m in existing_memories}
# Build new memories to add based on mode
new_memories: list[dict] = []
if self.enable_multiple:
meta_memories: list[dict] = self.context.get("meta_memories", [])
for mem in meta_memories:
memory_type = mem.get("memory_type", "")
memory_target = mem.get("memory_target", "")
if self._filter_memory_type_target(memory_type, memory_target, existing_set):
new_memories.append({"memory_type": memory_type, "memory_target": memory_target})
else:
memory_type = self.context.get("memory_type", "")
memory_target = self.context.get("memory_target", "")
if self._filter_memory_type_target(memory_type, memory_target, existing_set):
new_memories.append({"memory_type": memory_type, "memory_target": memory_target})
if not new_memories:
self.output = "No new meta memories to add (all entries already exist or invalid)."
return
# Merge and save
all_memories = existing_memories + new_memories
self._save_meta_memories(all_memories)
# Format output
added_str = json.dumps(new_memories, ensure_ascii=False)
self.output = f"Successfully added {len(new_memories)} meta memory entries: {added_str}"
logger.info(self.output)

View file

@ -1,26 +0,0 @@
tool: |
Add a memory metadata entry to register a new memory type and target.
IMPORTANT: Before using this tool, verify that the Main Agent's Meta Memory does NOT already contain the same <memory_type>(<memory_target>) combination. Only create new entries if they don't exist.
Use this tool to define what types of memories should be tracked, such as:
- Personal memories: "John", "Alice" (person-specific preferences and context)
- Procedural memories: "deployment_process", "code_review_steps" (how-to knowledge)
tool_multiple: |
Add multiple memory metadata entries to register multiple memory types and targets at once.
Before using this tool, verify that the Main Agent's Meta Memory does NOT already contain the same <memory_type>(<memory_target>) combinations. Only create new entries for those that don't exist.
Use this tool to define multiple memory tracking categories in a single operation.
Each entry specifies a memory_type and memory_target for organizing different memory domains.
meta_memories: |
A list of memory metadata entries to add. Each entry contains memory_type and memory_target.
memory_type: |
The type of memory to register. Valid values are: personal, procedural.
- personal: Person-specific memory storing preferences and context about specific individuals
- procedural: Procedural memory storing how-to knowledge and step-by-step processes
memory_target: |
The target identifier for this memory category.
Examples:
- For personal memory: person's name (e.g., "John", "Alice")
- For procedural memory: domain or topic name (e.g., "deployment", "code_review")

View file

@ -1,97 +0,0 @@
"""Read meta memory operation for retrieving memory metadata."""
from loguru import logger
from ..base_memory_tool import BaseMemoryTool
from ...core.context import C
from ...core.enumeration import MemoryType
@C.register_op()
class ReadMetaMemory(BaseMemoryTool):
"""Read memory metadata (memory_type and memory_target) from meta storage.
This operation reads stored memory metadata and optionally includes
TOOL and IDENTITY type memories.
"""
def __init__(
self,
enable_identity_memory: bool = False,
**kwargs,
):
"""Initialize ReadMetaMemory.
Args:
enable_identity_memory: Include IDENTITY type meta memory. Defaults to False.
**kwargs: Additional arguments for BaseMemoryTool.
"""
kwargs["enable_multiple"] = False
super().__init__(**kwargs)
self.enable_identity_memory = enable_identity_memory
def _build_parameters(self) -> dict:
"""Build input schema for reading meta memory.
No input parameters required for reading.
"""
return {
"type": "object",
"properties": {},
"required": [],
}
def _load_meta_memories(self) -> list[dict[str, str]]:
"""Load meta memories from cache and apply filters."""
result = self.meta_memory.load("meta_memories")
all_memories = result if result is not None else []
filtered_memories = []
for m in all_memories:
if m.get("memory_type") in [MemoryType.PERSONAL.value, MemoryType.PROCEDURAL.value]:
filtered_memories.append(m)
if self.enable_identity_memory:
filtered_memories.append(
{
"memory_type": MemoryType.IDENTITY.value,
"memory_target": "self",
},
)
return filtered_memories
def format_memory_metadata(self, memories: list[dict[str, str]]) -> str:
"""Format memory metadata into a readable string.
Args:
memories: List of memory metadata entries.
Returns:
str: Formatted memory metadata string.
"""
if not memories:
return ""
lines = []
for memory in memories:
memory_type = memory["memory_type"]
memory_target = memory["memory_target"]
description = self.get_prompt(f"type_{memory_type}")
lines.append(f"- {memory_type}({memory_target}): {description}")
return "\n".join(lines)
async def execute(self):
"""Execute the read meta memory operation.
Reads memory metadata from cache storage and formats output.
"""
memories = self._load_meta_memories()
if memories:
self.output = self.format_memory_metadata(memories)
logger.info(f"Retrieved {len(memories)} meta memory entries")
else:
self.output = "No memory metadata found."
logger.info(self.output)

View file

@ -1,16 +0,0 @@
tool: |
Read the memory metadata registry to see what types of memories are being tracked.
Use this tool to retrieve all registered memory types and their targets.
This helps understand what memory categories are available for storing and retrieving information.
type_identity: |
Self-cognition memory storing agent's identity, personality, and current state.
type_personal: |
Person-specific memory storing preferences and context about specific individuals.
type_procedural: |
Procedural memory storing how-to knowledge and step-by-step processes.
type_tool: |
Tool memory storing tool usage patterns, success rates, token consumption, and latency.