diff --git a/src/serena/tools/__init__.py b/src/serena/tools/__init__.py index bf0ed5d3..bc37b17c 100644 --- a/src/serena/tools/__init__.py +++ b/src/serena/tools/__init__.py @@ -2,3 +2,4 @@ from .tools_base import * from .file_tools import * from .ls_tools import * +from .memory_tools import * diff --git a/src/serena/tools/memory_tools.py b/src/serena/tools/memory_tools.py index 6e2bd1ac..63bba4fe 100644 --- a/src/serena/tools/memory_tools.py +++ b/src/serena/tools/memory_tools.py @@ -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()}") diff --git a/src/serena/tools/tools_base.py b/src/serena/tools/tools_base.py index 6e2bd1ac..f76730de 100644 --- a/src/serena/tools/tools_base.py +++ b/src/serena/tools/tools_base.py @@ -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.