From 6d5630c820ed9c18790d20530ad547f71facd557 Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Sun, 31 May 2026 13:15:37 +0800 Subject: [PATCH] feat(file-edit): support batched replacements --- cli/src/eventMapper.ts | 4 +- core/main_terminal_parts/tools_definition.py | 39 ++-- core/main_terminal_parts/tools_execution.py | 24 +- modules/file_manager.py | 217 +++++++++++++++++- server/chat_flow_tool_loop.py | 38 ++- .../components/chat/actions/ToolAction.vue | 180 +++++++++++---- .../components/chat/actions/toolRenderers.ts | 150 ++++++++---- utils/tool_result_formatter.py | 52 ++++- 8 files changed, 548 insertions(+), 156 deletions(-) diff --git a/cli/src/eventMapper.ts b/cli/src/eventMapper.ts index deec0c0..0b172c9 100644 --- a/cli/src/eventMapper.ts +++ b/cli/src/eventMapper.ts @@ -254,9 +254,7 @@ function editLines(args: any): string[] { newLines.filter((line) => line.length > 0).forEach((line, idx) => result.push(`${idx + 1} +${line}`)); }; if (replacements.length) { - for (const rep of replacements) emitPair(String(rep.old_text || rep.old || ''), String(rep.new_text || rep.new || '')); - } else { - emitPair(String(args.old_string || ''), String(args.new_string || '')); + for (const rep of replacements) emitPair(String(rep.old_string || ''), String(rep.new_string || '')); } return result; } diff --git a/core/main_terminal_parts/tools_definition.py b/core/main_terminal_parts/tools_definition.py index 5d58c68..41921d0 100644 --- a/core/main_terminal_parts/tools_definition.py +++ b/core/main_terminal_parts/tools_definition.py @@ -576,7 +576,7 @@ class MainTerminalToolsDefinitionMixin: "type": "function", "function": { "name": "edit_file", - "description": "在文件中执行精确的字符串替换;建议先使用 read_file 获取最新内容以确保精确匹配。", + "description": "在文件中按顺序执行一组或多组精确字符串替换;建议先使用 read_file 获取最新内容以确保精确匹配。任意一组失败时不会写入文件。", "parameters": { "type": "object", "properties": self._inject_intent({ @@ -584,21 +584,32 @@ class MainTerminalToolsDefinitionMixin: "type": "string", "description": "要修改文件的相对路径" }, - "old_string": { - "type": "string", - "description": "要替换的文本(需与文件内容精确匹配,保留缩进;建议提供至少3行提升定位稳定性。需要批量替换的场景可以单行或不足一行)" - }, - "new_string": { - "type": "string", - "description": "用于替换的新文本(必须不同于 old_string)" - }, - "replace_all": { - "type": "boolean", - "enum": [True, False], - "description": "是否替换所有匹配内容(必填)。false=仅替换首个匹配,true=替换全部匹配。" + "replacements": { + "type": "array", + "description": "替换项数组,按数组顺序依次执行。每一项包含 old_string/new_string,可选 replace_all;replace_all 未提供时默认为 false。", + "minItems": 1, + "items": { + "type": "object", + "properties": { + "old_string": { + "type": "string", + "description": "要替换的文本(需与文件内容精确匹配,保留缩进;建议提供至少3行提升定位稳定性。需要批量替换的场景可以单行或不足一行)" + }, + "new_string": { + "type": "string", + "description": "用于替换的新文本(必须不同于 old_string)" + }, + "replace_all": { + "type": "boolean", + "enum": [True, False], + "description": "是否替换该 old_string 的所有匹配内容。false=仅替换首个匹配,true=替换全部匹配;默认 false。" + } + }, + "required": ["old_string", "new_string"] + } } }), - "required": ["file_path", "old_string", "new_string", "replace_all"] + "required": ["file_path", "replacements"] } } }, diff --git a/core/main_terminal_parts/tools_execution.py b/core/main_terminal_parts/tools_execution.py index a614498..fbd7fb3 100644 --- a/core/main_terminal_parts/tools_execution.py +++ b/core/main_terminal_parts/tools_execution.py @@ -1226,29 +1226,13 @@ class MainTerminalToolsExecutionMixin: if read_guard_error: return json.dumps(read_guard_error, ensure_ascii=False) path = arguments.get("file_path") - old_text = arguments.get("old_string") - new_text = arguments.get("new_string") - replace_all_provided = "replace_all" in arguments - replace_all = arguments.get("replace_all") if replace_all_provided else None + replacements = arguments.get("replacements") if not path: result = {"success": False, "error": "缺少必要参数: file_path"} - elif old_text is None or new_text is None: - result = {"success": False, "error": "缺少必要参数: old_string/new_string"} - elif not replace_all_provided: - result = {"success": False, "error": "缺少必要参数: replace_all(必须显式传入 true 或 false)"} - elif not isinstance(replace_all, bool): - result = {"success": False, "error": "replace_all 必须是 true 或 false"} - elif old_text == new_text: - result = {"success": False, "error": "old_string 与 new_string 相同,无法执行替换"} - elif not old_text: - result = {"success": False, "error": "old_string 不能为空,请从 read_file 内容中精确复制"} + elif not isinstance(replacements, list) or not replacements: + result = {"success": False, "error": "缺少必要参数: replacements(必须是非空数组)"} else: - result = self.file_manager.replace_in_file( - path, - old_text, - new_text, - replace_all=replace_all - ) + result = self.file_manager.replace_many_in_file(path, replacements) elif tool_name == "create_folder": result = self.file_manager.create_folder(arguments["path"]) diff --git a/modules/file_manager.py b/modules/file_manager.py index a4d56d5..226a6da 100644 --- a/modules/file_manager.py +++ b/modules/file_manager.py @@ -5,7 +5,7 @@ import shutil from pathlib import Path import re from bisect import bisect_right -from typing import Optional, Dict, List, Set, Tuple, TYPE_CHECKING +from typing import Any, Optional, Dict, List, Set, Tuple, TYPE_CHECKING from datetime import datetime try: from config import ( @@ -1311,6 +1311,177 @@ class FileManager: return result + def replace_many_in_file(self, path: str, replacements: List[Dict[str, Any]]) -> Dict: + """按顺序执行多组精确字符串替换;任意一组失败时不写入文件。""" + result = self.read_file(path) + if not result["success"]: + return result + + if not isinstance(replacements, list) or not replacements: + return {"success": False, "error": "replacements 必须是非空数组"} + + if len(replacements) > 100: + return { + "success": False, + "error": "replacements 数量过多,最多支持 100 组", + "suggestion": "请拆分为多次 edit_file 调用" + } + + original_content = result["content"] + current_content = original_content + details: List[Dict[str, Any]] = [] + failed_details: List[Dict[str, Any]] = [] + total_replacements = 0 + total_found_matches = 0 + short_old_text_indices: List[int] = [] + + for zero_based_index, item in enumerate(replacements): + index = zero_based_index + 1 + detail: Dict[str, Any] = { + "index": index, + "status": "pending", + "replace_all": False, + "found_matches": 0, + "replacements": 0, + "matched_lines": [], + } + + if not isinstance(item, dict): + detail.update({ + "status": "error", + "reason": "替换项必须是对象,包含 old_string 和 new_string" + }) + failed_details.append(detail.copy()) + details.append(detail) + break + + old_text = item.get("old_string") + new_text = item.get("new_string") + replace_all = item.get("replace_all", False) + detail["replace_all"] = replace_all + + if old_text is None or new_text is None: + detail.update({ + "status": "error", + "reason": "缺少 old_string/new_string" + }) + failed_details.append(detail.copy()) + details.append(detail) + break + if not isinstance(old_text, str) or not isinstance(new_text, str): + detail.update({ + "status": "error", + "reason": "old_string/new_string 必须是字符串" + }) + failed_details.append(detail.copy()) + details.append(detail) + break + if not isinstance(replace_all, bool): + detail.update({ + "status": "error", + "reason": "replace_all 必须是 true 或 false;不提供时默认为 false" + }) + failed_details.append(detail.copy()) + details.append(detail) + break + if old_text == new_text: + detail.update({ + "status": "error", + "reason": "old_string 与 new_string 相同,无法执行替换" + }) + failed_details.append(detail.copy()) + details.append(detail) + break + if not old_text: + detail.update({ + "status": "error", + "reason": "old_string 不能为空,请从 read_file 内容中精确复制" + }) + failed_details.append(detail.copy()) + details.append(detail) + break + + if len(old_text) > 9999999999: + detail.update({ + "status": "error", + "reason": "要替换的文本过长,可能导致性能问题" + }) + failed_details.append(detail.copy()) + details.append(detail) + break + if len(new_text) > 9999999999: + detail.update({ + "status": "error", + "reason": "替换的新文本过长,建议分块处理" + }) + failed_details.append(detail.copy()) + details.append(detail) + break + + if len(old_text.splitlines()) < 3: + short_old_text_indices.append(index) + + matched_lines = self._find_match_line_numbers(current_content, old_text) + found_count = len(matched_lines) + if found_count == 0: + detail.update({ + "status": "not_found", + "reason": "未找到要替换的内容" + }) + failed_details.append(detail.copy()) + details.append(detail) + break + + replacement_count = found_count if replace_all else 1 + contexts = self._find_match_contexts( + current_content, + old_text, + max_matches=replacement_count + ) + + if replace_all: + current_content = current_content.replace(old_text, new_text) + else: + current_content = current_content.replace(old_text, new_text, 1) + + detail.update({ + "status": "success", + "found_matches": found_count, + "replacements": replacement_count, + "matched_lines": matched_lines, + "contexts": contexts, + }) + details.append(detail) + total_found_matches += found_count + total_replacements += replacement_count + + if failed_details: + return { + "success": False, + "path": result.get("path") or path, + "error": f"第 {failed_details[0].get('index')} 组替换失败:{failed_details[0].get('reason')}", + "failed_replacement_index": failed_details[0].get("index"), + "failed_details": failed_details, + "details": details, + "write_performed": False, + "completed_replacements": len([item for item in details if item.get("status") == "success"]), + } + + write_result = self.write_file(path, current_content) + if write_result["success"]: + write_result["replacements"] = total_replacements + write_result["found_matches"] = total_found_matches + write_result["replacement_groups"] = len(replacements) + write_result["details"] = details + write_result["write_performed"] = True + message_parts: List[str] = [f"共 {len(replacements)} 组替换,替换 {total_replacements} 处"] + if short_old_text_indices: + indices_text = ",".join(str(item) for item in short_old_text_indices) + message_parts.append(f"提示:第 {indices_text} 组 old_string 少于3行,已继续执行") + write_result["message"] = ";".join(message_parts) + print(f"{OUTPUT_FORMATS['file']} 批量替换了 {total_replacements} 处内容") + return write_result + @staticmethod def _find_match_line_numbers(content: str, target: str) -> List[int]: """返回 target 在 content 中每个匹配起始位置对应的行号(1-based)。""" @@ -1330,6 +1501,50 @@ class FileManager: search_start = idx + max(1, target_len) return line_numbers + + @staticmethod + def _find_match_contexts(content: str, target: str, max_matches: Optional[int] = None) -> List[Dict[str, Any]]: + """返回每个实际替换匹配位置的上下文行,仅用于结果 metadata 展示。""" + if not target: + return [] + + lines = content.splitlines() + newline_positions = [idx for idx, ch in enumerate(content) if ch == "\n"] + contexts: List[Dict[str, Any]] = [] + search_start = 0 + target_len = len(target) + old_line_count = max(1, len(target.splitlines())) + + while True: + if max_matches is not None and len(contexts) >= max_matches: + break + idx = content.find(target, search_start) + if idx < 0: + break + + start_line = bisect_right(newline_positions, idx) + 1 + end_line = start_line + old_line_count - 1 + before_start = max(1, start_line - 2) + before_lines = [ + {"line": line_no, "text": lines[line_no - 1]} + for line_no in range(before_start, start_line) + if 0 <= line_no - 1 < len(lines) + ] + after_end = min(len(lines), end_line + 2) + after_lines = [ + {"line": line_no, "text": lines[line_no - 1]} + for line_no in range(end_line + 1, after_end + 1) + if 0 <= line_no - 1 < len(lines) + ] + contexts.append({ + "old_start_line": start_line, + "old_end_line": end_line, + "before_lines": before_lines, + "after_lines": after_lines, + }) + search_start = idx + max(1, target_len) + + return contexts def clear_file(self, path: str) -> Dict: """清空文件内容""" diff --git a/server/chat_flow_tool_loop.py b/server/chat_flow_tool_loop.py index 95b864f..2e748b8 100644 --- a/server/chat_flow_tool_loop.py +++ b/server/chat_flow_tool_loop.py @@ -47,28 +47,15 @@ def _build_tool_approval_preview(web_terminal, function_name: str, arguments: Di if function_name == "edit_file": file_path = args.get("file_path") - old_string = args.get("old_string") - new_string = args.get("new_string") - replace_all_provided = "replace_all" in args - replace_all = args.get("replace_all") if replace_all_provided else None + replacements = args.get("replacements") preview["file_path"] = file_path - preview["replace_all"] = replace_all if not file_path: preview["summary"] = "缺少 file_path" return preview - if old_string is None or new_string is None: - preview["summary"] = "缺少 old_string/new_string" + if not isinstance(replacements, list) or not replacements: + preview["summary"] = "缺少 replacements(必须是非空数组)" return preview - if not replace_all_provided: - preview["summary"] = "缺少 replace_all(必须显式传入 true 或 false)" - return preview - if not isinstance(replace_all, bool): - preview["summary"] = "replace_all 必须是 true 或 false" - return preview - old_text = str(old_string or "") - short_old_text_notice = len(old_text.splitlines()) < 3 - if short_old_text_notice: - preview["notice"] = "提示:old_string 少于3行,允许继续执行;需要批量替换的场景可以单行或不足一行" + preview["replacement_groups"] = len(replacements) try: valid, err, full_path = web_terminal.file_manager._validate_path(str(file_path)) if not valid or full_path is None: @@ -80,7 +67,16 @@ def _build_tool_approval_preview(web_terminal, function_name: str, arguments: Di preview["summary"] = "目标文件不存在,无法生成上下文预览" return preview content = full_path.read_text(encoding="utf-8", errors="ignore") - new_text = str(new_string or "") + first = replacements[0] if isinstance(replacements[0], dict) else {} + old_text = str(first.get("old_string") or "") + new_text = str(first.get("new_string") or "") + replace_all = first.get("replace_all", False) + if not isinstance(replace_all, bool): + preview["summary"] = "第 1 组 replace_all 必须是 true 或 false" + return preview + short_old_text_notice = bool(old_text and len(old_text.splitlines()) < 3) + if short_old_text_notice: + preview["notice"] = "提示:第 1 组 old_string 少于3行,允许继续执行;需要批量替换的场景可以单行或不足一行" old_lines = old_text.splitlines() new_lines = new_text.splitlines() idx = content.find(old_text) if old_text else -1 @@ -102,7 +98,7 @@ def _build_tool_approval_preview(web_terminal, function_name: str, arguments: Di "old_end_line": end_line_no, } mode_text = "全部匹配" if replace_all is True else "首个匹配" - preview["summary"] = f"编辑 {file_path} 第 {start_line_no}-{end_line_no} 行({mode_text})" + preview["summary"] = f"编辑 {file_path},共 {len(replacements)} 组;预览第 1 组第 {start_line_no}-{end_line_no} 行({mode_text})" else: preview["edit_context"] = { "before": [], @@ -112,9 +108,9 @@ def _build_tool_approval_preview(web_terminal, function_name: str, arguments: Di "old_start_line": None, "old_end_line": None, } - preview["summary"] = "未在文件中定位到 old_string,显示原始替换内容" + preview["summary"] = f"共 {len(replacements)} 组替换;未在文件中定位到第 1 组 old_string,显示原始替换内容" if short_old_text_notice: - preview["summary"] = f"{preview['summary']}(old_string 少于3行,已告知并继续)" + preview["summary"] = f"{preview['summary']}(第 1 组 old_string 少于3行,已告知并继续)" except Exception as exc: preview["summary"] = f"生成编辑预览失败: {exc}" return preview diff --git a/static/src/components/chat/actions/ToolAction.vue b/static/src/components/chat/actions/ToolAction.vue index cf67aa9..0c86a32 100644 --- a/static/src/components/chat/actions/ToolAction.vue +++ b/static/src/components/chat/actions/ToolAction.vue @@ -318,8 +318,6 @@ function renderWriteFile(result: any, args: any): string { function renderEditFile(result: any, args: any): string { const path = args.file_path || args.path || result.path || ''; const status = formatToolStatusLabel(result, '✓ 已编辑'); - const replaceAllValue = - typeof args.replace_all === 'boolean' ? (args.replace_all ? '开' : '关') : '无'; const foundCount = typeof result.found_matches === 'number' ? result.found_matches @@ -328,17 +326,12 @@ function renderEditFile(result: any, args: any): string { : null; const replacedCount = typeof result.replacements === 'number' ? result.replacements : null; - // 兼容两种数据格式: - // 1. 新格式:args.replacements 数组 - // 2. 旧格式:args.old_string 和 args.new_string - const replacements = args.replacements || []; - let oldString = args.old_string || ''; - let newString = args.new_string || ''; + const replacements = Array.isArray(args.replacements) ? args.replacements : []; let html = '