tiennm99/serena
1
0 Fork 0
mirror of https://github.com/tiennm99/serena.git synced 2026-10-13 06:15:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Extract module memory_tools

Browse Source
This commit is contained in:
Dominik Jain authored and Dominik Jain committed 2025-07-05 18:31:45 +02:00
1 parent 87255ff5b7
commit 1a6245afa2
3 files changed
+3 -684
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/serena/tools/__init__.py
+1
View File
@@ -2,3 +2,4 @@
from .tools_base import *
from .file_tools import *
from .ls_tools import *
from .memory_tools import *
src/serena/tools/memory_tools.py
+1 -618
View File
@@ -1,362 +1,6 @@
import inspect
import json
import os
import platform
import traceback
from abc import ABC, abstractmethod
from collections.abc import Callable, Generator, Iterable
from copy import copy
from types import TracebackType
from typing import TYPE_CHECKING, Any, Self, TypeVar
from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metadata
from sensai.util import logging
from sensai.util.string import dict_string
from serena.config.context_mode import SerenaAgentMode
from serena.prompt_factory import PromptFactory
from serena.symbol import SymbolManager
from serena.util.inspection import iter_subclasses
from serena.util.shell import execute_shell_command
from solidlsp import SolidLanguageServer
if TYPE_CHECKING:
from serena.agent import LinesRead, MemoriesManager, SerenaAgent
log = logging.getLogger(__name__)
T = TypeVar("T")
SUCCESS_RESULT = "OK"
class Component(ABC):
def __init__(self, agent: "SerenaAgent"):
self.agent = agent
@property
def language_server(self) -> SolidLanguageServer:
assert self.agent.language_server is not None
return self.agent.language_server
def get_project_root(self) -> str:
"""
:return: the root directory of the active project, raises a ValueError if no active project configuration is set
"""
return self.agent.get_project_root()
@property
def prompt_factory(self) -> PromptFactory:
return self.agent.prompt_factory
@property
def memories_manager(self) -> "MemoriesManager":
assert self.agent.memories_manager is not None
return self.agent.memories_manager
@property
def symbol_manager(self) -> SymbolManager:
assert self.agent.symbol_manager is not None
return self.agent.symbol_manager
@property
def lines_read(self) -> "LinesRead":
assert self.agent.lines_read is not None
return self.agent.lines_read
TOOL_DEFAULT_MAX_ANSWER_LENGTH = int(2e5)
class ToolMarkerCanEdit:
"""
Marker class for all tools that can perform editing operations on files.
"""
class ToolMarkerDoesNotRequireActiveProject:
pass
class ToolInterface(ABC):
"""Protocol defining the complete interface that make_tool() expects from a tool."""
@abstractmethod
def get_name(self) -> str:
"""Get the tool name."""
...
@abstractmethod
def get_apply_docstring(self) -> str:
"""Get the docstring for the tool application, used by the MCP server."""
...
@abstractmethod
def get_apply_fn_metadata(self) -> FuncMetadata:
"""Get the metadata for the tool application function, used by the MCP server."""
...
@abstractmethod
def apply_ex(self, log_call: bool = True, catch_exceptions: bool = True, **kwargs: Any) -> str:
"""Apply the tool with logging and exception handling."""
...
class Tool(Component, ToolInterface):
# NOTE: each tool should implement the apply method, which is then used in
# the central method of the Tool class `apply_ex`.
# Failure to do so will result in a RuntimeError at tool execution time.
# The apply method is not declared as part of the base Tool interface since we cannot
# know the signature of the (input parameters of the) method in advance.
#
# The docstring and types of the apply method are used to generate the tool description
# (which is use by the LLM, so a good description is important)
# and to validate the tool call arguments.
@classmethod
def get_name_from_cls(cls) -> str:
name = cls.__name__
if name.endswith("Tool"):
name = name[:-4]
# convert to snake_case
name = "".join(["_" + c.lower() if c.isupper() else c for c in name]).lstrip("_")
return name
def get_name(self) -> str:
return self.get_name_from_cls()
def get_apply_fn(self) -> Callable:
apply_fn = getattr(self, "apply")
if apply_fn is None:
raise RuntimeError(f"apply not defined in {self}. Did you forget to implement it?")
return apply_fn
@classmethod
def can_edit(cls) -> bool:
"""
Returns whether this tool can perform editing operations on code.
:return: True if the tool can edit code, False otherwise
"""
return issubclass(cls, ToolMarkerCanEdit)
@classmethod
def get_tool_description(cls) -> str:
docstring = cls.__doc__
if docstring is None:
return ""
return docstring.strip()
@classmethod
def get_apply_docstring_from_cls(cls) -> str:
"""Get the docstring for the apply method from the class (static metadata).
Needed for creating MCP tools in a separate process without running into serialization issues.
"""
# First try to get from __dict__ to handle dynamic docstring changes
if "apply" in cls.__dict__:
apply_fn = cls.__dict__["apply"]
else:
# Fall back to getattr for inherited methods
apply_fn = getattr(cls, "apply", None)
if apply_fn is None:
raise AttributeError(f"apply method not defined in {cls}. Did you forget to implement it?")
docstring = apply_fn.__doc__
if not docstring:
raise AttributeError(f"apply method has no (or empty) docstring in {cls}. Did you forget to implement it?")
return docstring.strip()
def get_apply_docstring(self) -> str:
"""Get the docstring for the apply method (instance method implementing ToolProtocol)."""
return self.get_apply_docstring_from_cls()
def get_apply_fn_metadata(self) -> FuncMetadata:
"""Get the metadata for the apply method (instance method implementing ToolProtocol)."""
return self.get_apply_fn_metadata_from_cls()
@classmethod
def get_apply_fn_metadata_from_cls(cls) -> FuncMetadata:
"""Get the metadata for the apply method from the class (static metadata).
Needed for creating MCP tools in a separate process without running into serialization issues.
"""
# First try to get from __dict__ to handle dynamic docstring changes
if "apply" in cls.__dict__:
apply_fn = cls.__dict__["apply"]
else:
# Fall back to getattr for inherited methods
apply_fn = getattr(cls, "apply", None)
if apply_fn is None:
raise AttributeError(f"apply method not defined in {cls}. Did you forget to implement it?")
return func_metadata(apply_fn, skip_names=["self", "cls"])
def _log_tool_application(self, frame: Any) -> None:
params = {}
ignored_params = {"self", "log_call", "catch_exceptions", "args", "apply_fn"}
for param, value in frame.f_locals.items():
if param in ignored_params:
continue
if param == "kwargs":
params.update(value)
else:
params[param] = value
log.info(f"{self.get_name_from_cls()}: {dict_string(params)}")
@staticmethod
def _limit_length(result: str, max_answer_chars: int) -> str:
if (n_chars := len(result)) > max_answer_chars:
result = (
f"The answer is too long ({n_chars} characters). "
+ "Please try a more specific tool query or raise the max_answer_chars parameter."
)
return result
def is_active(self) -> bool:
return self.agent.tool_is_active(self.__class__)
def apply_ex(self, log_call: bool = True, catch_exceptions: bool = True, **kwargs) -> str: # type: ignore
"""
Applies the tool with the given arguments
"""
def task() -> str:
apply_fn = self.get_apply_fn()
try:
if not self.is_active():
return f"Error: Tool '{self.get_name_from_cls()}' is not active. Active tools: {self.agent.get_active_tool_names()}"
except Exception as e:
return f"RuntimeError while checking if tool {self.get_name_from_cls()} is active: {e}"
if log_call:
self._log_tool_application(inspect.currentframe())
try:
# check whether the tool requires an active project and language server
if not isinstance(self, ToolMarkerDoesNotRequireActiveProject):
if self.agent._active_project is None:
return (
"Error: No active project. Ask to user to select a project from this list: "
+ f"{self.agent.serena_config.project_names}"
)
if not self.agent.is_language_server_running():
log.info("Language server is not running. Starting it ...")
self.agent.reset_language_server()
# apply the actual tool
result = apply_fn(**kwargs)
except Exception as e:
if not catch_exceptions:
raise
msg = f"Error executing tool: {e}\n{traceback.format_exc()}"
log.error(
f"Error executing tool: {e}. "
f"Consider restarting the language server to solve this (especially, if it's a timeout of a symbolic operation)",
exc_info=e,
)
result = msg
if log_call:
log.info(f"Result: {result}")
try:
self.language_server.save_cache()
except Exception as e:
log.error(f"Error saving language server cache: {e}")
return result
future = self.agent.issue_task(task, name=self.__class__.__name__)
return future.result(timeout=self.agent.serena_config.tool_timeout)
class EditedFileContext:
"""
Context manager for file editing.
Create the context, then use `set_updated_content` to set the new content, the original content
being provided in `original_content`.
When exiting the context without an exception, the updated content will be written back to the file.
"""
def __init__(self, relative_path: str, agent: "SerenaAgent"):
self._project = agent.get_active_project()
assert self._project is not None
self._abs_path = os.path.join(self._project.project_root, relative_path)
if not os.path.isfile(self._abs_path):
raise FileNotFoundError(f"File {self._abs_path} does not exist.")
with open(self._abs_path, encoding=self._project.project_config.encoding) as f:
self._original_content = f.read()
self._updated_content: str | None = None
def __enter__(self) -> Self:
return self
def get_original_content(self) -> str:
"""
:return: the original content of the file before any modifications.
"""
return self._original_content
def set_updated_content(self, content: str) -> None:
"""
Sets the updated content of the file, which will be written back to the file
when the context is exited without an exception.
:param content: the updated content of the file
"""
self._updated_content = content
def __exit__(self, exc_type: type[BaseException] | None, exc_value: BaseException | None, traceback: TracebackType | None) -> None:
if self._updated_content is not None and exc_type is None:
assert self._project is not None
with open(self._abs_path, "w", encoding=self._project.project_config.encoding) as f:
f.write(self._updated_content)
log.info(f"Updated content written to {self._abs_path}")
# Language servers should automatically detect the change and update its state accordingly.
# If they do not, we may have to add a call to notify it.
class CheckOnboardingPerformedTool(Tool):
"""
Checks whether project onboarding was already performed.
"""
def apply(self) -> str:
"""
Checks whether project onboarding was already performed.
You should always call this tool before beginning to actually work on the project/after activating a project,
but after calling the initial instructions tool.
"""
list_memories_tool = self.agent.get_tool(ListMemoriesTool)
memories = json.loads(list_memories_tool.apply())
if len(memories) == 0:
return (
"Onboarding not performed yet (no memories available). "
+ "You should perform onboarding by calling the `onboarding` tool before proceeding with the task."
)
else:
return f"""The onboarding was already performed, below is the list of available memories.
Do not read them immediately, just remember that they exist and that you can read them later, if it is necessary
for the current task.
Some memories may be based on previous conversations, others may be general for the current project.
You should be able to tell which one you need based on the name of the memory.
{memories}"""
class OnboardingTool(Tool):
"""
Performs onboarding (identifying the project structure and essential tasks, e.g. for testing or building).
"""
def apply(self) -> str:
"""
Call this tool if onboarding was not performed yet.
You will call this tool at most once per conversation.
:return: instructions on how to create the onboarding information
"""
system = platform.system()
return self.prompt_factory.create_onboarding_prompt(system=system)
from serena.tools import TOOL_DEFAULT_MAX_ANSWER_LENGTH, Tool
class WriteMemoryTool(Tool):
@@ -423,264 +67,3 @@ class DeleteMemoryTool(Tool):
or no longer relevant for the project.
"""
return self.memories_manager.delete_memory(memory_file_name)
class ThinkAboutCollectedInformationTool(Tool):
"""
Thinking tool for pondering the completeness of collected information.
"""
def apply(self) -> str:
"""
Think about the collected information and whether it is sufficient and relevant.
This tool should ALWAYS be called after you have completed a non-trivial sequence of searching steps like
find_symbol, find_referencing_symbols, search_files_for_pattern, read_file, etc.
"""
return self.prompt_factory.create_think_about_collected_information()
class ThinkAboutTaskAdherenceTool(Tool):
"""
Thinking tool for determining whether the agent is still on track with the current task.
"""
def apply(self) -> str:
"""
Think about the task at hand and whether you are still on track.
Especially important if the conversation has been going on for a while and there
has been a lot of back and forth.
This tool should ALWAYS be called before you insert, replace, or delete code.
"""
return self.prompt_factory.create_think_about_task_adherence()
class ThinkAboutWhetherYouAreDoneTool(Tool):
"""
Thinking tool for determining whether the task is truly completed.
"""
def apply(self) -> str:
"""
Whenever you feel that you are done with what the user has asked for, it is important to call this tool.
"""
return self.prompt_factory.create_think_about_whether_you_are_done()
class SummarizeChangesTool(Tool):
"""
Provides instructions for summarizing the changes made to the codebase.
"""
def apply(self) -> str:
"""
Summarize the changes you have made to the codebase.
This tool should always be called after you have fully completed any non-trivial coding task,
but only after the think_about_whether_you_are_done call.
"""
return self.prompt_factory.create_summarize_changes()
class PrepareForNewConversationTool(Tool):
"""
Provides instructions for preparing for a new conversation (in order to continue with the necessary context).
"""
def apply(self) -> str:
"""
Instructions for preparing for a new conversation. This tool should only be called on explicit user request.
"""
return self.prompt_factory.create_prepare_for_new_conversation()
class ExecuteShellCommandTool(Tool, ToolMarkerCanEdit):
"""
Executes a shell command.
"""
def apply(
self,
command: str,
cwd: str | None = None,
capture_stderr: bool = True,
max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH,
) -> str:
"""
Execute a shell command and return its output.
IMPORTANT: you should always consider the memory about suggested shell commands before using this tool.
If this memory was not loaded in the current conversation, you should load it using the `read_memory` tool
before using this tool.
You should have at least once looked at the suggested shell commands from the corresponding memory
created during the onboarding process before using this tool.
Never execute unsafe shell commands like `rm -rf /` or similar! Generally be very careful with deletions.
:param command: the shell command to execute
:param cwd: the working directory to execute the command in. If None, the project root will be used.
:param capture_stderr: whether to capture and return stderr output
:param max_answer_chars: if the output is longer than this number of characters,
no content will be returned. Don't adjust unless there is really no other way to get the content
required for the task.
:return: a JSON object containing the command's stdout and optionally stderr output
"""
_cwd = cwd or self.get_project_root()
result = execute_shell_command(command, cwd=_cwd, capture_stderr=capture_stderr)
result = result.json()
return self._limit_length(result, max_answer_chars)
class ActivateProjectTool(Tool, ToolMarkerDoesNotRequireActiveProject):
"""
Activates a project by name.
"""
def apply(self, project: str) -> str:
"""
Activates the project with the given name.
:param project: the name of a registered project to activate or a path to a project directory
"""
from ..config.serena_config import ProjectConfig
active_project, new_project_generated, new_project_config_generated = self.agent.activate_project_from_path_or_name(project)
if new_project_generated:
result_str = (
f"Created and activated a new project with name {active_project.project_name} at {active_project.project_root}, language: {active_project.project_config.language.value}. "
+ "You can activate this project later by name."
)
else:
result_str = f"Activated existing project with name {active_project.project_name} at {active_project.project_root}, language: {active_project.project_config.language.value}"
if new_project_config_generated:
result_str += (
f"\nNote: A new project configuration was autogenerated because the given path did not contain a {ProjectConfig.SERENA_DEFAULT_PROJECT_FILE} file."
+ f"You can now edit the project configuration in the file {active_project.path_to_project_yml()}. In particular, you may want to edit the project name and the initial prompt."
)
if active_project.project_config.initial_prompt:
result_str += f"\nAdditional project information:\n {active_project.project_config.initial_prompt}"
result_str += (
f"\nAvailable memories:\n {json.dumps(list(self.memories_manager.list_memories()))}"
+ "You should not read these memories directly, but rather use the `read_memory` tool to read them later if needed for the task."
)
result_str += f"\nAvailable tools:\n {json.dumps(self.agent.get_active_tool_names())}"
return result_str
class RemoveProjectTool(Tool, ToolMarkerDoesNotRequireActiveProject):
"""
Removes a project from the Serena configuration.
"""
def apply(self, project_name: str) -> str:
"""
Removes a project from the Serena configuration.
:param project_name: Name of the project to remove
"""
self.agent.serena_config.remove_project(project_name)
return f"Successfully removed project '{project_name}' from configuration."
class SwitchModesTool(Tool):
"""
Activates modes by providing a list of their names
"""
def apply(self, modes: list[str]) -> str:
"""
Activates the desired modes, like ["editing", "interactive"] or ["planning", "one-shot"]
:param modes: the names of the modes to activate
"""
mode_instances = [SerenaAgentMode.load(mode) for mode in modes]
self.agent.set_modes(mode_instances)
# Inform the Agent about the activated modes and the currently active tools
result_str = f"Successfully activated modes: {', '.join([mode.name for mode in mode_instances])}" + "\n"
result_str += "\n".join([mode_instance.prompt for mode_instance in mode_instances]) + "\n"
result_str += f"Currently active tools: {', '.join(self.agent.get_active_tool_names())}"
return result_str
class GetCurrentConfigTool(Tool):
"""
Prints the current configuration of the agent, including the active and available projects, tools, contexts, and modes.
"""
def apply(self) -> str:
"""
Print the current configuration of the agent, including the active and available projects, tools, contexts, and modes.
"""
return self.agent.get_current_config_overview()
class InitialInstructionsTool(Tool, ToolMarkerDoesNotRequireActiveProject):
"""
Gets the initial instructions for the current project.
Should only be used in settings where the system prompt cannot be set,
e.g. in clients you have no control over, like Claude Desktop.
"""
def apply(self) -> str:
"""
Get the initial instructions for the current coding project.
You should always call this tool before starting to work (including using any other tool) on any programming task!
The only exception is when a user asks you to activate a project, in which case you should call the `activate_project` first
instead and then call this tool.
"""
return self.agent.create_system_prompt()
class ToolRegistry:
_tool_dict: dict[str, type[Tool]] | None = None
"""maps tool name to the corresponding tool class"""
@staticmethod
def _iter_tool_classes() -> Generator[type[Tool], None, None]:
"""
Iterate over Tool subclasses.
"""
yield from iter_subclasses(Tool)
@classmethod
def _get_tool_dict(cls) -> dict[str, type[Tool]]:
if cls._tool_dict is None:
cls._tool_dict = {tool_class.get_name_from_cls(): tool_class for tool_class in cls._iter_tool_classes()}
return cls._tool_dict
@classmethod
def get_tool_class_by_name(cls, tool_name: str) -> type[Tool]:
try:
return cls._get_tool_dict()[tool_name]
except KeyError as e:
available_tools = "\n".join(ToolRegistry.get_tool_names())
raise ValueError(f"Tool with name {tool_name} not found. Available tools:\n{available_tools}") from e
@classmethod
def get_all_tool_classes(cls) -> list[type[Tool]]:
return list(cls._get_tool_dict().values())
@classmethod
def get_tool_names(cls) -> list[str]:
return list(cls._get_tool_dict().keys())
@classmethod
def tool_dict(cls) -> dict[str, type[Tool]]:
"""Maps tool name to the corresponding tool class"""
return copy(cls._get_tool_dict())
@classmethod
def print_tool_overview(cls, tools: Iterable[type[Tool] | Tool] | None = None) -> None:
"""
Print a summary of the tools. If no tools are passed, a summary of all tools is printed.
"""
if tools is None:
tools = cls._get_tool_dict().values()
tool_dict: dict[str, type[Tool] | Tool] = {}
for tool_class in tools:
tool_dict[tool_class.get_name_from_cls()] = tool_class
for tool_name in sorted(tool_dict.keys()):
tool_class = tool_dict[tool_name]
print(f" * `{tool_name}`: {tool_class.get_tool_description().strip()}")
src/serena/tools/tools_base.py
+1 -66
View File
@@ -326,6 +326,7 @@ class CheckOnboardingPerformedTool(Tool):
You should always call this tool before beginning to actually work on the project/after activating a project,
but after calling the initial instructions tool.
"""
from .memory_tools import ListMemoriesTool
list_memories_tool = self.agent.get_tool(ListMemoriesTool)
memories = json.loads(list_memories_tool.apply())
if len(memories) == 0:
@@ -359,72 +360,6 @@ class OnboardingTool(Tool):
return self.prompt_factory.create_onboarding_prompt(system=system)
class WriteMemoryTool(Tool):
"""
Writes a named memory (for future reference) to Serena's project-specific memory store.
"""
def apply(self, memory_name: str, content: str, max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH) -> str:
"""
Write some information about this project that can be useful for future tasks to a memory.
Use markdown formatting for the content.
The information should be short and to the point.
The memory name should be meaningful, such that from the name you can infer what the information is about.
It is better to have multiple small memories than to have a single large one because
memories will be read one by one and we only ever want to read relevant memories.
This tool is either called during the onboarding process or when you have identified
something worth remembering about the project from the past conversation.
"""
if len(content) > max_answer_chars:
raise ValueError(
f"Content for {memory_name} is too long. Max length is {max_answer_chars} characters. " + "Please make the content shorter."
)
return self.memories_manager.save_memory(memory_name, content)
class ReadMemoryTool(Tool):
"""
Reads the memory with the given name from Serena's project-specific memory store.
"""
def apply(self, memory_file_name: str, max_answer_chars: int = TOOL_DEFAULT_MAX_ANSWER_LENGTH) -> str:
"""
Read the content of a memory file. This tool should only be used if the information
is relevant to the current task. You can infer whether the information
is relevant from the memory file name.
You should not read the same memory file multiple times in the same conversation.
"""
return self.memories_manager.load_memory(memory_file_name)
class ListMemoriesTool(Tool):
"""
Lists memories in Serena's project-specific memory store.
"""
def apply(self) -> str:
"""
List available memories. Any memory can be read using the `read_memory` tool.
"""
return json.dumps(self.memories_manager.list_memories())
class DeleteMemoryTool(Tool):
"""
Deletes a memory from Serena's project-specific memory store.
"""
def apply(self, memory_file_name: str) -> str:
"""
Delete a memory file. Should only happen if a user asks for it explicitly,
for example by saying that the information retrieved from a memory file is no longer correct
or no longer relevant for the project.
"""
return self.memories_manager.delete_memory(memory_file_name)
class ThinkAboutCollectedInformationTool(Tool):
"""
Thinking tool for pondering the completeness of collected information.