#!/usr/bin/env python3 """This is an adaptation of the Anthropic Text Editor tool from https://github.com/anthropics/anthropic-quickstarts/blob/main/computer-use-demo/computer_use_demo/tools/edit.py However, we made it python 3.6 compatible and stateless (all state is saved in a json file) """ import argparse import json import re import subprocess import sys from collections import defaultdict from pathlib import Path from typing import List, Optional, Tuple import io from registry import registry as REGISTRY # There are some super strange "ascii can't decode x" errors, # that can be solved with setting the default encoding for stdout # (note that python3.6 doesn't have the reconfigure method) sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8") # Type annotation comments for Python 3.5 compatibility TRUNCATED_MESSAGE = "To save on context only part of this file has been shown to you. You should retry this tool after you have searched inside the file with `grep -n` in order to find the line numbers of what you are looking for." # type: str MAX_RESPONSE_LEN = 16000 # type: int MAX_WINDOW_EXPANSION_VIEW = int(REGISTRY.get("MAX_WINDOW_EXPANSION_VIEW", 0)) MAX_WINDOW_EXPANSION_EDIT_CONFIRM = int(REGISTRY.get("MAX_WINDOW_EXPANSION_EDIT_CONFIRM", 0)) USE_FILEMAP = REGISTRY.get("USE_FILEMAP", "false").lower() == "true" USE_LINTER = REGISTRY.get("USE_LINTER", "false").lower() == "true" Command = str SNIPPET_LINES = 4 # type: int LINT_WARNING_TEMPLATE = """ Your edits have been applied, but the linter has found syntax errors. {errors} Please review the changes and make sure they are correct. In addition to the above errors, please also check the following: 1. The edited file is correctly indented 2. The edited file does not contain duplicate lines 3. The edit does not break existing functionality In rare cases, the linter errors might not actually be errors or caused by your edit. Please use your own judgement. Edit the file again if necessary. """ def maybe_truncate(content: str, truncate_after: Optional[int] = MAX_RESPONSE_LEN): """Truncate content and append a notice if content exceeds the specified length.""" return ( content if not truncate_after or len(content) <= truncate_after else content[:truncate_after] + TRUNCATED_MESSAGE ) class Flake8Error: """A class to represent a single flake8 error""" def __init__(self, filename: str, line_number: int, col_number: int, problem: str): self.filename = filename self.line_number = line_number self.col_number = col_number self.problem = problem @classmethod def from_line(cls, line: str): try: prefix, _sep, problem = line.partition(": ") filename, line_number, col_number = prefix.split(":") except (ValueError, IndexError) as e: msg = "Invalid flake8 error line: {}".format(line) raise ValueError(msg) from e return cls(filename, int(line_number), int(col_number), problem) def __eq__(self, other): if not isinstance(other, Flake8Error): return NotImplemented return ( self.filename == other.filename and self.line_number == other.line_number and self.col_number == other.col_number and self.problem == other.problem ) def __repr__(self): return "Flake8Error(filename={}, line_number={}, col_number={}, problem={})".format( self.filename, self.line_number, self.col_number, self.problem) def _update_previous_errors( previous_errors: List[Flake8Error], replacement_window: Tuple[int, int], replacement_n_lines: int ) -> List[Flake8Error]: """Update the line numbers of the previous errors to what they would be after the edit window. This is a helper function for `_filter_previous_errors`. All previous errors that are inside of the edit window should not be ignored, so they are removed from the previous errors list. Args: previous_errors: list of errors with old line numbers replacement_window: the window of the edit/lines that will be replaced replacement_n_lines: the number of lines that will be used to replace the text Returns: list of errors with updated line numbers """ updated = [] lines_added = replacement_n_lines - (replacement_window[1] - replacement_window[0] + 1) for error in previous_errors: if error.line_number < replacement_window[0]: # no need to adjust the line number updated.append(error) continue if replacement_window[0] <= error.line_number <= replacement_window[1]: # The error is within the edit window, so let's not ignore it # either way (we wouldn't know how to adjust the line number anyway) continue # We're out of the edit window, so we need to adjust the line number updated.append(Flake8Error(error.filename, error.line_number + lines_added, error.col_number, error.problem)) return updated def format_flake8_output( input_string: str, show_line_numbers: bool = False, *, previous_errors_string: str = "", replacement_window: Optional[Tuple[int, int]] = None, replacement_n_lines: Optional[int] = None, ) -> str: """Filter flake8 output for previous errors and print it for a given file. Args: input_string: The flake8 output as a string show_line_numbers: Whether to show line numbers in the output previous_errors_string: The previous errors as a string replacement_window: The window of the edit (lines that will be replaced) replacement_n_lines: The number of lines used to replace the text Returns: The filtered flake8 output as a string """ # print("Replacement window: {}".format(replacement_window)) # print("Replacement n lines:", replacement_n_lines) # print("Previous errors string:", previous_errors_string) # print("Input string:", input_string) errors = [Flake8Error.from_line(line.strip()) for line in input_string.split("\n") if line.strip()] # print("New errors before filtering: errors={}".format(errors)) lines = [] if previous_errors_string: assert replacement_window is not None assert replacement_n_lines is not None previous_errors = [ Flake8Error.from_line(line.strip()) for line in previous_errors_string.split("\n") if line.strip() ] # print("Previous errors before updating: previous_errors={}".format(previous_errors)) previous_errors = _update_previous_errors(previous_errors, replacement_window, replacement_n_lines) # print("Previous errors after updating: previous_errors={}".format(previous_errors)) errors = [error for error in errors if error not in previous_errors] # Sometimes new errors appear above the replacement window that were 'shadowed' by the previous errors # they still clearly aren't caused by the edit. errors = [error for error in errors if error.line_number >= replacement_window[0]] # print("New errors after filtering: errors={}".format(errors)) for error in errors: if not show_line_numbers: lines.append("- {}".format(error.problem)) else: lines.append("- line {} col {}: {}".format(error.line_number, error.col_number, error.problem)) return "\n".join(lines) def flake8(file_path: str) -> str: """Run flake8 on a given file and return the output as a string""" if Path(file_path).suffix != ".py": return "" cmd = REGISTRY.get("LINT_COMMAND", "flake8 --isolated --select=F821,F822,F831,E111,E112,E113,E999,E902 {file_path}") # don't use capture_output because it's not compatible with python3.6 out = subprocess.run(cmd.format(file_path=file_path), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) return out.stdout.decode() class Filemap: def show_filemap(self, file_contents: str, encoding: str = "utf8"): import warnings from tree_sitter_languages import get_language, get_parser warnings.simplefilter("ignore", category=FutureWarning) parser = get_parser("python") language = get_language("python") tree = parser.parse(bytes(file_contents.encode(encoding, errors="replace"))) # See https://tree-sitter.github.io/tree-sitter/using-parsers#pattern-matching-with-queries. query = language.query(""" (function_definition body: (_) @body) """) # TODO: consider special casing docstrings such that they are not elided. This # could be accomplished by checking whether `body.text.decode('utf8')` starts # with `"""` or `'''`. elide_line_ranges = [ (node.start_point[0], node.end_point[0]) for node, _ in query.captures(tree.root_node) # Only elide if it's sufficiently long if node.end_point[0] - node.start_point[0] >= 5 ] # Note that tree-sitter line numbers are 0-indexed, but we display 1-indexed. elide_lines = {line for start, end in elide_line_ranges for line in range(start, end + 1)} elide_messages = [(start, "... eliding lines {}-{} ...".format(start+1, end+1)) for start, end in elide_line_ranges] out = [] for i, line in sorted( elide_messages + [(i, line) for i, line in enumerate(file_contents.splitlines()) if i not in elide_lines] ): out.append("{:6d} {}".format(i+1, line)) return "\n".join(out) class WindowExpander: def __init__(self, suffix: str = ""): """Try to expand viewports to include whole functions, classes, etc. rather than using fixed line windows. Args: suffix: Filename suffix """ self.suffix = suffix if self.suffix: assert self.suffix.startswith(".") def _find_breakpoints(self, lines: List[str], current_line: int, direction=1, max_added_lines: int = 30) -> int: """Returns 1-based line number of breakpoint. This line is meant to still be included in the viewport. Args: lines: List of lines of the file current_line: 1-based line number of the current viewport direction: 1 for down, -1 for up max_added_lines: Maximum number of lines to extend Returns: 1-based line number of breakpoint. This line is meant to still be included in the viewport. """ assert 1 <= current_line <= len(lines) assert 0 <= max_added_lines # 1. Find line range that we want to search for breakpoints in if direction == 1: # down if current_line == len(lines): # already last line, can't extend down return current_line iter_lines = range(current_line, 1 + min(current_line + max_added_lines, len(lines))) elif direction == -1: # up if current_line == 1: # already first line, can't extend up return current_line iter_lines = range(current_line, -1 + max(current_line - max_added_lines, 1), -1) else: msg = "Invalid direction {}".format(direction) raise ValueError(msg) # 2. Find the best breakpoint in the line range # Every condition gives a score, the best score is the best breakpoint best_score = 0 best_breakpoint = current_line for i_line in iter_lines: next_line = None line = lines[i_line - 1] if i_line + direction in iter_lines: next_line = lines[i_line + direction - 1] score = 0 if line == "": score = 1 if next_line == "": # Double new blank line: score = 2 if self.suffix == ".py" and any( re.match(regex, line) for regex in [r"^\s*def\s+", r"^\s*class\s+", r"^\s*@"] ): # We include decorators here, because they are always on top of the function/class definition score = 3 if score > best_score: best_score = score best_breakpoint = i_line if direction == 1 and i_line != current_line: best_breakpoint -= 1 if i_line == 1 or i_line == len(lines): score = 3 if score > best_score: best_score = score best_breakpoint = i_line # print("Score {} for line {} ({})".format(score, i_line, line)) # print("Best score {} for line {} ({})".format(best_score, best_breakpoint, lines[best_breakpoint-1])) if direction == 1 and best_breakpoint < current_line or direction == -1 and best_breakpoint > current_line: # We don't want to shrink the view port, so we return the current line return current_line return best_breakpoint def expand_window(self, lines: List[str], start: int, stop: int, max_added_lines: int) -> Tuple[int, int]: """ Args: lines: All lines of the file start: 1-based line number of the start of the viewport stop: 1-based line number of the end of the viewport max_added_lines: Maximum number of lines to extend (separately for each side) Returns: Tuple of 1-based line numbers of the start and end of the viewport. Both inclusive. """ # print("Input:", start, stop) assert 1 <= start <= stop <= len(lines), (start, stop, len(lines)) if max_added_lines <= 0: # Already at max range, no expansion return start, stop new_start = self._find_breakpoints(lines, start, direction=-1, max_added_lines=max_added_lines) new_stop = self._find_breakpoints(lines, stop, direction=1, max_added_lines=max_added_lines) # print("Expanded window is {} to {}".format(new_start, new_stop)) assert new_start <= new_stop, (new_start, new_stop) assert new_start <= start, (new_start, start) assert start - new_start <= max_added_lines, (start, new_start) assert new_stop >= stop, (new_stop, stop) assert new_stop - stop <= max_added_lines, (new_stop, stop) return new_start, new_stop class EditTool: """ An filesystem editor tool that allows the agent to view, create, and edit files. The tool parameters are defined by Anthropic and are not editable. """ name = "str_replace_editor" def __init__(self): super().__init__() self._encoding = None @property def _file_history(self): return defaultdict(list, json.loads(REGISTRY.get("file_history", "{}"))) @_file_history.setter def _file_history(self, value: dict): REGISTRY["file_history"] = json.dumps(value) def __call__( self, *, command: Command, path: str, file_text: Optional[str] = None, view_range: Optional[List[int]] = None, old_str: Optional[str] = None, new_str: Optional[str] = None, insert_line: Optional[int] = None, **kwargs, ): _path = Path(path) self.validate_path(command, _path) if command == "view": return self.view(_path, view_range) elif command == "create": if file_text is None: print("Parameter `file_text` is required for command: create") sys.exit(1) self.create_file(_path, file_text) return None elif command == "str_replace": if old_str is None: print("Parameter `old_str` is required for command: str_replace") sys.exit(2) return self.str_replace(_path, old_str, new_str) elif command == "insert": if insert_line is None: print("Parameter `insert_line` is required for command: insert") sys.exit(3) if new_str is None: print("Parameter `new_str` is required for command: insert") sys.exit(4) return self.insert(_path, insert_line, new_str) elif command == "undo_edit": return self.undo_edit(_path) print( 'Unrecognized command {}. The allowed commands for the {} tool are: "view", "create", "str_replace", "insert", "undo_edit"'.format(command, self.name) ) sys.exit(5) def validate_path(self, command: str, path: Path): """ Check that the path/command combination is valid. """ # Check if its an absolute path if not path.is_absolute(): suggested_path = Path.cwd() / path print( "The path {} is not an absolute path, it should start with `/`. Maybe you meant {}?".format(path, suggested_path) ) sys.exit(6) # Check if path exists if not path.exists() and command != "create": print("The path {} does not exist. Please provide a valid path.".format(path)) sys.exit(7) if path.exists() and command == "create": print("File already exists at: {}. Cannot overwrite files using command `create`.".format(path)) sys.exit(8) # Check if the path points to a directory if path.is_dir(): if command != "view": print("The path {} is a directory and only the `view` command can be used on directories".format(path)) sys.exit(9) def create_file(self, path: Path, file_text: str): if not path.parent.exists(): print("The parent directory {} does not exist. Please create it first.".format(path.parent)) sys.exit(21) self.write_file(path, file_text) self._file_history[path].append(file_text) print("File created successfully at: {}".format(path)) def view(self, path: Path, view_range: Optional[List[int]] = None): """Implement the view command""" if path.is_dir(): if view_range: print("The `view_range` parameter is not allowed when `path` points to a directory.") sys.exit(10) out = subprocess.run( r"find {} -maxdepth 2 -not -path '*/\.*'".format(path), shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) stdout = out.stdout.decode() stderr = out.stderr.decode() if not stderr: stdout = "Here's the files and directories up to 2 levels deep in {}, excluding hidden items:\n{}\n".format(path, stdout) print(stdout) return file_content = self.read_file(path) if view_range: if len(view_range) != 2 or not all(isinstance(i, int) for i in view_range): print("Invalid `view_range`. It should be a list of two integers.") sys.exit(11) file_lines = file_content.split("\n") n_lines_file = len(file_lines) init_line, final_line = view_range if init_line < 1 or init_line > n_lines_file: print( "Invalid `view_range`: {}. Its first element `{}` should be within the range of lines of the file: {}".format(view_range, init_line, [1, n_lines_file]) ) sys.exit(12) if final_line > n_lines_file: print( "Invalid `view_range`: {}. Its second element `{}` should be smaller than the number of lines in the file: `{}`".format(view_range, final_line, n_lines_file) ) sys.exit(13) if final_line != -1 and final_line < init_line: print( "Invalid `view_range`: {}. Its second element `{}` should be larger or equal than its first `{}`".format(view_range, final_line, init_line) ) sys.exit(14) if final_line == -1: final_line = n_lines_file # Expand the viewport to include the whole function or class init_line, final_line = WindowExpander(suffix=path.suffix).expand_window( file_lines, init_line, final_line, max_added_lines=MAX_WINDOW_EXPANSION_VIEW ) file_content = "\n".join(file_lines[init_line - 1 : final_line]) else: if path.suffix == ".py" and len(file_content) > MAX_RESPONSE_LEN and USE_FILEMAP: try: filemap = Filemap().show_filemap(file_content, encoding=self._encoding or "utf-8") except Exception: # If we fail to show the filemap, just show the truncated file content pass else: print( "This file is too large to display entirely. Showing abbreviated version. " "Please use `str_replace_editor view` with the `view_range` parameter to show selected lines next." ) filemap = maybe_truncate(filemap.expandtabs()) print(filemap) print( "The above file has been abbreviated. Please use `str_replace editor view` with `view_range` to look at relevant files in detail." ) return # Else just show init_line = 1 # init_line is 1-based print(self._make_output(file_content, str(path), init_line=init_line)) def str_replace(self, path: Path, old_str: str, new_str: Optional[str]): """Implement the str_replace command, which replaces old_str with new_str in the file content""" # Read the file content file_content = self.read_file(path).expandtabs() old_str = old_str.expandtabs() new_str = new_str.expandtabs() if new_str is not None else "" # Check if old_str is unique in the file occurrences = file_content.count(old_str) if occurrences == 0: print("No replacement was performed, old_str `{}` did not appear verbatim in {}.".format(old_str, path)) sys.exit(15) elif occurrences > 1: file_content_lines = file_content.split("\n") lines = [idx + 1 for idx, line in enumerate(file_content_lines) if old_str in line] print( "No replacement was performed. Multiple occurrences of old_str `{}` in lines {}. Please ensure it is unique".format(old_str, lines) ) sys.exit(16) if new_str == old_str: print("No replacement was performed, old_str `{}` is the same as new_str `{}`.".format(old_str, new_str)) sys.exit(161) pre_edit_lint = "" if USE_LINTER: try: pre_edit_lint = flake8(str(path)) except Exception as e: print("Warning: Failed to run pre-edit linter on {}: {}".format(path, e)) # Replace old_str with new_str new_file_content = file_content.replace(old_str, new_str) # Write the new content to the file self.write_file(path, new_file_content) post_edit_lint = "" if USE_LINTER: try: post_edit_lint = flake8(str(path)) except Exception as e: print("Warning: Failed to run post-edit linter on {}: {}".format(path, e)) epilogue = "" if post_edit_lint: ... replacement_window_start_line = file_content.split(old_str)[0].count("\n") + 1 replacement_lines = len(new_str.split("\n")) replacement_window_end_line = replacement_window_start_line + replacement_lines - 1 replacement_window = (replacement_window_start_line, replacement_window_end_line) errors = format_flake8_output( post_edit_lint, previous_errors_string=pre_edit_lint, replacement_window=replacement_window, replacement_n_lines=replacement_lines, ) if errors.strip(): epilogue = LINT_WARNING_TEMPLATE.format(errors=errors) # Save the content to history self._file_history[path].append(file_content) # Create a snippet of the edited section replacement_line = file_content.split(old_str)[0].count("\n") start_line = max(1, replacement_line - SNIPPET_LINES) end_line = min(replacement_line + SNIPPET_LINES + new_str.count("\n"), len(new_file_content.splitlines())) start_line, end_line = WindowExpander(suffix=path.suffix).expand_window( new_file_content.split("\n"), start_line, end_line, max_added_lines=MAX_WINDOW_EXPANSION_EDIT_CONFIRM ) snippet = "\n".join(new_file_content.split("\n")[start_line - 1 : end_line]) # Prepare the success message success_msg = "The file {} has been edited. ".format(path) success_msg += self._make_output(snippet, "a snippet of {}".format(path), start_line) success_msg += "Review the changes and make sure they are as expected. Edit the file again if necessary." success_msg += epilogue print(success_msg) def insert(self, path: Path, insert_line: int, new_str: str): """Implement the insert command, which inserts new_str at the specified line in the file content.""" file_text = self.read_file(path).expandtabs() new_str = new_str.expandtabs() file_text_lines = file_text.split("\n") n_lines_file = len(file_text_lines) if insert_line < 0 or insert_line > n_lines_file: print( "Invalid `insert_line` parameter: {}. It should be within the range of lines of the file: {}".format(insert_line, [0, n_lines_file]) ) sys.exit(17) new_str_lines = new_str.split("\n") new_file_text_lines = file_text_lines[:insert_line] + new_str_lines + file_text_lines[insert_line:] snippet_lines = ( file_text_lines[max(0, insert_line - SNIPPET_LINES) : insert_line] + new_str_lines + file_text_lines[insert_line : insert_line + SNIPPET_LINES] ) new_file_text = "\n".join(new_file_text_lines) snippet = "\n".join(snippet_lines) self.write_file(path, new_file_text) self._file_history[path].append(file_text) # todo: Also expand these windows success_msg = "The file {} has been edited. ".format(path) success_msg += self._make_output( snippet, "a snippet of the edited file", max(1, insert_line - SNIPPET_LINES + 1), ) success_msg += "Review the changes and make sure they are as expected (correct indentation, no duplicate lines, etc). Edit the file again if necessary." print(success_msg) def undo_edit(self, path: Path): """Implement the undo_edit command.""" if not self._file_history[path]: print("No edit history found for {}.".format(path)) sys.exit(18) old_text = self._file_history[path].pop() self.write_file(path, old_text) print("Last edit to {} undone successfully. {}".format(path, self._make_output(old_text, str(path)))) def read_file(self, path: Path): """Read the content of a file from a given path; raise a ToolError if an error occurs.""" encodings = [ (None, None), ("utf-8", None), ("latin-1", None), ("utf-8", "replace"), ] exception = None for self._encoding, errors in encodings: try: text = path.read_text(encoding=self._encoding, errors=errors) except UnicodeDecodeError as e: exception = e else: break else: print("Ran into UnicodeDecodeError {} while trying to read {}".format(exception, path)) sys.exit(19) return text def write_file(self, path: Path, file: str): """Write the content of a file to a given path; raise a ToolError if an error occurs.""" try: path.write_text(file, encoding=self._encoding or "utf-8") except Exception as e: print("Ran into {} while trying to write to {}".format(e, path)) sys.exit(20) def _make_output( self, file_content: str, file_descriptor: str, init_line: int = 1, expand_tabs: bool = True, ): """Generate output for the CLI based on the content of a file.""" file_content = maybe_truncate(file_content) if expand_tabs: file_content = file_content.expandtabs() file_content = "\n".join(["{:6}\t{}".format(i + init_line, line) for i, line in enumerate(file_content.split("\n"))]) return "Here's the result of running `cat -n` on {}:\n".format(file_descriptor) + file_content + "\n" def main(): parser = argparse.ArgumentParser() parser.add_argument("command", type=str) parser.add_argument("path", type=str) parser.add_argument("--file_text", type=str) parser.add_argument("--view_range", type=int, nargs=2) parser.add_argument("--old_str", type=str) parser.add_argument("--new_str", type=str) parser.add_argument("--insert_line", type=int) args = parser.parse_args() tool = EditTool() tool( command=args.command, path=args.path, file_text=args.file_text, view_range=args.view_range, old_str=args.old_str, new_str=args.new_str, insert_line=args.insert_line, ) if __name__ == "__main__": main()