mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-12 22:14:54 +00:00
Add memory facade (mem) including onboarding
Introduce MemoryApi, exposing the memory operations list_memories, read_memory, write_memory, edit_memory, rename_memory and delete_memory (the latter four marked as editing operations) as well as onboarding, which provides the onboarding instructions. list_memories returns a MemoryList, which exposes the (writable and read-only) memory names to code and renders as JSON. The memory tools and the onboarding tool delegate to the API via MemoryApiMixin; the onboarding tool retains its check for the availability of the memory writing tool. The mem facade is always part of the REPL. The modes no-onboarding, no-memories and benchmark exclude the corresponding APIs (mem.onboarding, or the mem facade as a whole), mirroring their tool exclusions.
This commit is contained in:
1 parent
d9cda96725
commit
8e4d68cd40
8 files changed
+273
-39
No files matched your search
+2
-1
@@ -50,6 +50,7 @@ from serena.prompt_factory import SerenaPromptFactory
|
||||
from serena.repl.api.edit_api import EditApi
|
||||
from serena.repl.api.jb_api import JetBrainsApi
|
||||
from serena.repl.api.lsp_api import LspApi
|
||||
from serena.repl.api.mem_api import MemoryApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
from serena.repl.repl import SerenaRepl
|
||||
from serena.task_executor import TaskExecutor
|
||||
@@ -1184,7 +1185,7 @@ class SerenaAgent:
|
||||
api_scope.exclude_editing()
|
||||
|
||||
# gather facades
|
||||
facades = [Facade.from_api(EditApi(self), api_scope)]
|
||||
facades = [Facade.from_api(EditApi(self), api_scope), Facade.from_api(MemoryApi(self), api_scope)]
|
||||
if self._language_backend.is_lsp():
|
||||
facades.append(Facade.from_api(LspApi(self), api_scope))
|
||||
elif self._language_backend.is_jetbrains():
|
||||
|
||||
@@ -0,0 +1,171 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
"""
|
||||
The implementation of memory operations (and onboarding, which creates the initial memories).
|
||||
"""
|
||||
|
||||
import logging
|
||||
import platform
|
||||
from typing import TYPE_CHECKING, Literal
|
||||
|
||||
from serena.memories.memory_manager import MemoryManager
|
||||
|
||||
from ..facade import FacadeApi, facade_method
|
||||
from ..representable import Renderer, RepresentableViaRenderer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from serena.agent import SerenaAgent
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MemoryList(RepresentableViaRenderer):
|
||||
"""
|
||||
The available memories: `memories` (writable) and `read_only_memories` (e.g. global memories), each a list of names.
|
||||
"""
|
||||
|
||||
def __init__(self, memories_list: MemoryManager.MemoriesList, renderer: "MemoryListRenderer"):
|
||||
"""
|
||||
:param memories_list: the list of memories
|
||||
:param renderer: the renderer to use for representing the list
|
||||
"""
|
||||
super().__init__(renderer)
|
||||
self.memories_list_ = memories_list
|
||||
|
||||
@property
|
||||
def memories(self) -> list[str]:
|
||||
return sorted(self.memories_list_.memories)
|
||||
|
||||
@property
|
||||
def read_only_memories(self) -> list[str]:
|
||||
return sorted(self.memories_list_.read_only_memories)
|
||||
|
||||
|
||||
class MemoryListRenderer(Renderer[MemoryList]):
|
||||
def render(self, obj: MemoryList) -> str:
|
||||
return self._limit_length(self._to_json(obj.memories_list_.to_dict()))
|
||||
|
||||
|
||||
class MemoryApi(FacadeApi):
|
||||
def __init__(self, agent: "SerenaAgent") -> None:
|
||||
super().__init__(
|
||||
agent,
|
||||
name="mem",
|
||||
description="project memories (persistent notes for future tasks): listing, reading, writing, editing; onboarding",
|
||||
)
|
||||
|
||||
def _get_memory_manager(self) -> MemoryManager:
|
||||
return self._get_project().memory_manager
|
||||
|
||||
@facade_method()
|
||||
def list_memories(self, topic: str = "") -> MemoryList:
|
||||
"""
|
||||
Lists the available memories, optionally filtered by topic.
|
||||
|
||||
:param topic: the topic (prefix of the memory name, e.g. "frontend") to restrict the listing to; empty for all memories
|
||||
:return: the memories
|
||||
"""
|
||||
return MemoryList(self._get_memory_manager().list_memories(topic), MemoryListRenderer(self._agent, -1))
|
||||
|
||||
@facade_method()
|
||||
def read_memory(self, memory_name: str) -> str:
|
||||
"""
|
||||
Reads a memory that is likely to be relevant to the current task, inferring relevance e.g. from the name.
|
||||
|
||||
:param memory_name: the name of the memory
|
||||
:return: the memory's content
|
||||
"""
|
||||
return self._get_memory_manager().load_memory(memory_name)
|
||||
|
||||
@facade_method(can_edit=True)
|
||||
def write_memory(self, memory_name: str, content: str, max_chars: int = -1) -> str:
|
||||
"""
|
||||
Writes information about this project that can be useful for future tasks in md format.
|
||||
The name should be meaningful and can include "/" to organize into topics.
|
||||
If explicitly instructed, use the "global/" prefix for writing a memory that is shared across projects.
|
||||
References to other memories should be inside backticks and prefixed with mem:,
|
||||
e.g., `mem:auth`.
|
||||
|
||||
:param memory_name: the memory name
|
||||
:param content: the memory content (utf-8-encoded markdown)
|
||||
:param max_chars: the maximum content length; -1 for the configured default
|
||||
:return: a message indicating the result
|
||||
"""
|
||||
if max_chars == -1:
|
||||
max_chars = self._agent.serena_config.default_max_tool_answer_chars
|
||||
if len(content) > max_chars:
|
||||
raise ValueError(
|
||||
f"Content for {memory_name} is too long. Max length is {max_chars} characters. Please make the content shorter."
|
||||
)
|
||||
return self._get_memory_manager().save_memory(memory_name, content, is_tool_context=True)
|
||||
|
||||
@facade_method(can_edit=True)
|
||||
def edit_memory(
|
||||
self,
|
||||
memory_name: str,
|
||||
needle: str,
|
||||
repl: str,
|
||||
mode: Literal["literal", "regex"],
|
||||
allow_multiple_occurrences: bool = False,
|
||||
) -> str:
|
||||
"""
|
||||
Replaces content matching a pattern in a memory.
|
||||
|
||||
:param memory_name: the name of the memory
|
||||
:param needle: the string or regex pattern to search for. In regex mode, be careful to not replace too much!
|
||||
If `mode` is "literal", this string will be matched exactly.
|
||||
If `mode` is "regex", this string will be treated as a regular expression (syntax of Python's `re` module,
|
||||
with the MULTILINE and DOTALL flags enabled).
|
||||
:param repl: the replacement string (verbatim).
|
||||
:param mode: either "literal" or "regex", specifying how the `needle` parameter is to be interpreted.
|
||||
:param allow_multiple_occurrences: whether to allow matching and replacing multiple occurrences.
|
||||
If false and multiple occurrences are found, an error will be raised.
|
||||
:return: a message indicating the result
|
||||
"""
|
||||
return self._get_memory_manager().edit_memory(
|
||||
memory_name, needle, repl, mode, allow_multiple_occurrences, is_tool_context=True, regex_multiline=True
|
||||
)
|
||||
|
||||
@facade_method(can_edit=True)
|
||||
def rename_memory(self, old_name: str, new_name: str) -> str:
|
||||
"""
|
||||
Renames or moves a memory; use "/" in the name to organize into topics.
|
||||
The "global" topic should only be used if explicitly instructed.
|
||||
References to other memories that are marked with the `mem:` prefix will be updated accordingly.
|
||||
References in read-only memories are not affected.
|
||||
|
||||
:param old_name: the current name of the memory
|
||||
:param new_name: the new name of the memory
|
||||
:return: a message indicating the result
|
||||
"""
|
||||
renaming_message, n_references_updated = self._get_memory_manager().rename_memory_and_propagate_references(
|
||||
old_name, new_name, is_tool_context=True
|
||||
)
|
||||
if n_references_updated > 0:
|
||||
log.info(f"Updated {n_references_updated} references to memory {old_name} to {new_name}")
|
||||
return renaming_message
|
||||
|
||||
@facade_method(can_edit=True)
|
||||
def delete_memory(self, memory_name: str) -> str:
|
||||
"""
|
||||
Deletes a memory; only call this if instructed explicitly or permission was granted by the user.
|
||||
|
||||
:param memory_name: the name of the memory
|
||||
:return: a message indicating the result
|
||||
"""
|
||||
return self._get_memory_manager().delete_memory(memory_name, is_tool_context=True)
|
||||
|
||||
@facade_method()
|
||||
def onboarding(self) -> str:
|
||||
"""
|
||||
Provides the instructions for performing onboarding (identifying the project structure and essential tasks,
|
||||
e.g. for testing or building, and recording the findings in memories).
|
||||
Call this if onboarding was not performed yet, at most once per conversation.
|
||||
|
||||
:return: the instructions on how to create the onboarding information
|
||||
"""
|
||||
# seed the project-local memory-maintenance memory (or detect a global override) so
|
||||
# the prompt can point the agent at the conventions before it writes anything
|
||||
memory_maintenance_name = self._get_memory_manager().ensure_memory_maintenance_memory()
|
||||
return self._agent.prompt_factory.create_onboarding_prompt(
|
||||
system=platform.system(), memory_maintenance_name=memory_maintenance_name
|
||||
)
|
||||
@@ -18,3 +18,5 @@ excluded_tools:
|
||||
- delete_memory
|
||||
- rename_memory
|
||||
- onboarding
|
||||
excluded_apis:
|
||||
- mem
|
||||
@@ -9,3 +9,5 @@ excluded_tools:
|
||||
- rename_memory
|
||||
- list_memories
|
||||
- onboarding
|
||||
excluded_apis:
|
||||
- mem
|
||||
@@ -3,3 +3,5 @@ prompt: |
|
||||
The onboarding process is not applied.
|
||||
excluded_tools:
|
||||
- onboarding
|
||||
excluded_apis:
|
||||
- mem.onboarding
|
||||
@@ -1,14 +1,22 @@
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
import logging
|
||||
from typing import Literal
|
||||
from typing import Literal, cast
|
||||
|
||||
from serena.repl.api.mem_api import MemoryApi
|
||||
from serena.tools import Tool, ToolMarkerCanEdit
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
class MemoryApiMixin:
|
||||
"""
|
||||
Mixin for tools which delegate to the memory API
|
||||
"""
|
||||
|
||||
def _api(self) -> MemoryApi:
|
||||
tool = cast(Tool, cast(object, self))
|
||||
return MemoryApi(tool.agent)
|
||||
|
||||
|
||||
class WriteMemoryTool(Tool, ToolMarkerCanEdit):
|
||||
class WriteMemoryTool(Tool, ToolMarkerCanEdit, MemoryApiMixin):
|
||||
"""
|
||||
Write some information (utf-8-encoded) about this project that can be useful for future tasks to a memory in md format.
|
||||
The memory name should be meaningful.
|
||||
@@ -26,18 +34,10 @@ class WriteMemoryTool(Tool, ToolMarkerCanEdit):
|
||||
:param content: memory content, utf8-encoded
|
||||
:param max_chars: see other tools
|
||||
"""
|
||||
# NOTE: utf-8 encoding is configured in the MemoriesManager
|
||||
if max_chars == -1:
|
||||
max_chars = self.agent.serena_config.default_max_tool_answer_chars
|
||||
if len(content) > max_chars:
|
||||
raise ValueError(
|
||||
f"Content for {memory_name} is too long. Max length is {max_chars} characters. " + "Please make the content shorter."
|
||||
)
|
||||
|
||||
return self.memory_manager.save_memory(memory_name, content, is_tool_context=True)
|
||||
return self._api().write_memory(memory_name, content, max_chars)
|
||||
|
||||
|
||||
class ReadMemoryTool(Tool):
|
||||
class ReadMemoryTool(Tool, MemoryApiMixin):
|
||||
"""
|
||||
Reads the content of a memory file.
|
||||
"""
|
||||
@@ -46,10 +46,10 @@ class ReadMemoryTool(Tool):
|
||||
"""
|
||||
Use to read a memory that is likely to be relevant to the current task, inferring relevance e.g. from the name.
|
||||
"""
|
||||
return self.memory_manager.load_memory(memory_name)
|
||||
return self._api().read_memory(memory_name)
|
||||
|
||||
|
||||
class ListMemoriesTool(Tool):
|
||||
class ListMemoriesTool(Tool, MemoryApiMixin):
|
||||
"""
|
||||
Lists available memories.
|
||||
"""
|
||||
@@ -58,10 +58,10 @@ class ListMemoriesTool(Tool):
|
||||
"""
|
||||
Lists available memories, optionally filtered by topic.
|
||||
"""
|
||||
return self._to_json(self.memory_manager.list_memories(topic).to_dict())
|
||||
return self._api().list_memories(topic).represent()
|
||||
|
||||
|
||||
class DeleteMemoryTool(Tool, ToolMarkerCanEdit):
|
||||
class DeleteMemoryTool(Tool, ToolMarkerCanEdit, MemoryApiMixin):
|
||||
"""
|
||||
Delete a memory file.
|
||||
"""
|
||||
@@ -70,10 +70,10 @@ class DeleteMemoryTool(Tool, ToolMarkerCanEdit):
|
||||
"""
|
||||
Delete a memory, only call if instructed explicitly or permission was granted by the user.
|
||||
"""
|
||||
return self.memory_manager.delete_memory(memory_name, is_tool_context=True)
|
||||
return self._api().delete_memory(memory_name)
|
||||
|
||||
|
||||
class RenameMemoryTool(Tool, ToolMarkerCanEdit):
|
||||
class RenameMemoryTool(Tool, ToolMarkerCanEdit, MemoryApiMixin):
|
||||
"""
|
||||
Renames or moves a memory, updating references that are marked with the `mem:` prefix.
|
||||
"""
|
||||
@@ -85,15 +85,10 @@ class RenameMemoryTool(Tool, ToolMarkerCanEdit):
|
||||
References to other memories that are marked with the `mem:` prefix will be updated accordingly.
|
||||
References in read-only memories are not affected.
|
||||
"""
|
||||
renaming_message, n_references_updated = self.memory_manager.rename_memory_and_propagate_references(
|
||||
old_name, new_name, is_tool_context=True
|
||||
)
|
||||
if n_references_updated > 0:
|
||||
log.info(f"Updated {n_references_updated} references to memory {old_name} to {new_name}")
|
||||
return renaming_message
|
||||
return self._api().rename_memory(old_name, new_name)
|
||||
|
||||
|
||||
class EditMemoryTool(Tool, ToolMarkerCanEdit):
|
||||
class EditMemoryTool(Tool, ToolMarkerCanEdit, MemoryApiMixin):
|
||||
"""
|
||||
Replaces content matching a regular expression in a memory.
|
||||
"""
|
||||
@@ -119,6 +114,4 @@ class EditMemoryTool(Tool, ToolMarkerCanEdit):
|
||||
:param allow_multiple_occurrences: whether to allow matching and replacing multiple occurrences.
|
||||
If false and multiple occurrences are found, an error will be returned.
|
||||
"""
|
||||
return self.memory_manager.edit_memory(
|
||||
memory_name, needle, repl, mode, allow_multiple_occurrences, is_tool_context=True, regex_multiline=True
|
||||
)
|
||||
return self._api().edit_memory(memory_name, needle, repl, mode, allow_multiple_occurrences)
|
||||
@@ -3,12 +3,11 @@ Tools supporting the general workflow of the agent
|
||||
"""
|
||||
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||
|
||||
import platform
|
||||
|
||||
from serena.tools import Tool, ToolMarkerDoesNotRequireActiveProject, ToolMarkerOptional, WriteMemoryTool
|
||||
from serena.tools.memory_tools import MemoryApiMixin
|
||||
|
||||
|
||||
class OnboardingTool(Tool):
|
||||
class OnboardingTool(Tool, MemoryApiMixin):
|
||||
"""
|
||||
Performs onboarding (identifying the project structure and essential tasks, e.g. for testing or building).
|
||||
"""
|
||||
@@ -23,11 +22,7 @@ class OnboardingTool(Tool):
|
||||
write_memory_tool_available = self.agent.tool_is_exposed(WriteMemoryTool.get_name_from_cls())
|
||||
if not write_memory_tool_available:
|
||||
return "Memory writing tool not activated, skipping onboarding."
|
||||
system = platform.system()
|
||||
# seed the project-local memory-maintenance memory (or detect a global override) so
|
||||
# the prompt can point the agent at the conventions before it writes anything
|
||||
memory_maintenance_name = self.memory_manager.ensure_memory_maintenance_memory()
|
||||
return self.prompt_factory.create_onboarding_prompt(system=system, memory_maintenance_name=memory_maintenance_name)
|
||||
return self._api().onboarding()
|
||||
|
||||
|
||||
class InitialInstructionsTool(Tool, ToolMarkerDoesNotRequireActiveProject):
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
"""
|
||||
Tests for the memory facade API.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.project import Project
|
||||
from serena.repl.api.mem_api import MemoryApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(tmp_path: Path) -> MemoryApi:
|
||||
project = Project.load(str(tmp_path), serena_config=SerenaConfig(gui_log_window=False, web_dashboard=False))
|
||||
agent = MagicMock()
|
||||
agent.get_active_project_or_raise.return_value = project
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return MemoryApi(agent)
|
||||
|
||||
|
||||
def test_facade_exposes_memory_operations(api: MemoryApi) -> None:
|
||||
facade = Facade.from_api(api, ApiScope())
|
||||
assert facade.name == "mem"
|
||||
assert set(facade.enabled_method_names) == {
|
||||
"list_memories",
|
||||
"read_memory",
|
||||
"write_memory",
|
||||
"edit_memory",
|
||||
"rename_memory",
|
||||
"delete_memory",
|
||||
"onboarding",
|
||||
}
|
||||
assert {name for name in facade.enabled_method_names if facade.get_method(name).info.can_edit} == {
|
||||
"write_memory",
|
||||
"edit_memory",
|
||||
"rename_memory",
|
||||
"delete_memory",
|
||||
}
|
||||
|
||||
|
||||
def test_memory_lifecycle(api: MemoryApi) -> None:
|
||||
api.write_memory("topic/first", "# First\nhello")
|
||||
api.write_memory("second", "see `mem:topic/first`")
|
||||
|
||||
# listing exposes the names to code and renders as JSON (global memories of the machine may be present, too)
|
||||
memory_list = api.list_memories()
|
||||
assert {"second", "topic/first"} <= set(memory_list.memories)
|
||||
assert api.list_memories("topic").memories == ["topic/first"]
|
||||
assert '"memories"' in memory_list.represent()
|
||||
|
||||
# reading, editing, renaming (with reference propagation) and deleting
|
||||
assert api.read_memory("topic/first") == "# First\nhello"
|
||||
api.edit_memory("topic/first", "hello", "world", mode="literal")
|
||||
assert api.read_memory("topic/first") == "# First\nworld"
|
||||
api.rename_memory("topic/first", "topic/renamed")
|
||||
assert "mem:topic/renamed" in api.read_memory("second")
|
||||
api.delete_memory("second")
|
||||
assert "second" not in api.list_memories().memories
|
||||
assert api.list_memories("topic").memories == ["topic/renamed"]
|
||||
|
||||
|
||||
def test_write_memory_rejects_overlong_content(api: MemoryApi) -> None:
|
||||
with pytest.raises(ValueError, match="too long"):
|
||||
api.write_memory("big", "x" * 100, max_chars=10)
|
||||
Reference in new issue
Block a user