From ca8f65fe3574dc44d72c7042b4e8cccc7196537d Mon Sep 17 00:00:00 2001 From: JOJO <1498581755@qq.com> Date: Sun, 14 Dec 2025 17:56:18 +0800 Subject: [PATCH] feat: index-based memory tool with append/replace/delete --- core/main_terminal.py | 42 +++++++++--------- modules/memory_manager.py | 80 ++++++++++++++++++++++++++++++++-- utils/tool_result_formatter.py | 13 +++++- 3 files changed, 108 insertions(+), 27 deletions(-) diff --git a/core/main_terminal.py b/core/main_terminal.py index 20b3837..50cc869 100644 --- a/core/main_terminal.py +++ b/core/main_terminal.py @@ -1474,15 +1474,16 @@ class MainTerminal: "type": "function", "function": { "name": "update_memory", - "description": "更新记忆文件", + "description": "按条目更新记忆列表（自动编号）。append 追加新条目；replace 用序号替换；delete 用序号删除。", "parameters": { "type": "object", "properties": { "memory_type": {"type": "string", "enum": ["main", "task"], "description": "记忆类型"}, - "content": {"type": "string", "description": "要添加的内容"}, - "operation": {"type": "string", "enum": ["append", "replace"], "description": "操作类型"} + "content": {"type": "string", "description": "条目内容。append/replace 时必填"}, + "operation": {"type": "string", "enum": ["append", "replace", "delete"], "description": "操作类型"}, + "index": {"type": "integer", "description": "要替换/删除的序号（从1开始）"} }, - "required": ["memory_type", "content", "operation"] + "required": ["memory_type", "operation"] } } }, @@ -2097,25 +2098,24 @@ class MainTerminal: elif tool_name == "update_memory": memory_type = arguments["memory_type"] - content = arguments["content"] operation = arguments["operation"] - - if memory_type == "main": - if operation == "append": - success = self.memory_manager.append_main_memory(content) - else: - success = self.memory_manager.write_main_memory(content) + content = arguments.get("content") + index = arguments.get("index") + + # 参数校验 + if operation == "append" and (not content or not str(content).strip()): + result = {"success": False, "error": "append 操作需要 content"} + elif operation == "replace" and (index is None or index <= 0 or not content or not str(content).strip()): + result = {"success": False, "error": "replace 操作需要有效的 index 和 content"} + elif operation == "delete" and (index is None or index <= 0): + result = {"success": False, "error": "delete 操作需要有效的 index"} else: - if operation == "append": - success = self.memory_manager.append_task_memory(content) - else: - success = self.memory_manager.write_task_memory(content) - - result = { - "success": success, - "memory_type": memory_type, - "operation": operation - } + result = self.memory_manager.update_entries( + memory_type=memory_type, + operation=operation, + content=content, + index=index + ) elif tool_name == "todo_create": result = self.todo_manager.create_todo_list( diff --git a/modules/memory_manager.py b/modules/memory_manager.py index f2f7034..f98a49f 100644 --- a/modules/memory_manager.py +++ b/modules/memory_manager.py @@ -5,6 +5,7 @@ import json from datetime import datetime from pathlib import Path from typing import Dict, List, Optional +import re try: from config import MAIN_MEMORY_FILE, TASK_MEMORY_FILE, DATA_DIR, OUTPUT_FORMATS except ImportError: @@ -261,9 +262,8 @@ class MemoryManager: # 主记忆统计 if stats["main_memory"]["exists"]: stat = self.main_memory_path.stat() - content = self.read_main_memory() stats["main_memory"]["size"] = stat.st_size - stats["main_memory"]["lines"] = len(content.split('\n')) + stats["main_memory"]["lines"] = len(self._read_entries("main")) stats["main_memory"]["last_modified"] = datetime.fromtimestamp( stat.st_mtime ).isoformat() @@ -271,9 +271,8 @@ class MemoryManager: # 任务记忆统计 if stats["task_memory"]["exists"]: stat = self.task_memory_path.stat() - content = self.read_task_memory() stats["task_memory"]["size"] = stat.st_size - stats["task_memory"]["lines"] = len(content.split('\n')) + stats["task_memory"]["lines"] = len(self._read_entries("task")) stats["task_memory"]["last_modified"] = datetime.fromtimestamp( stat.st_mtime ).isoformat() @@ -305,3 +304,76 @@ class MemoryManager: except Exception as e: print(f"{OUTPUT_FORMATS['error']} 合并记忆失败: {e}") return False + + # ===================== 新增：条目化记忆操作 ===================== + + def _get_path(self, memory_type: str) -> Path: + return self.main_memory_path if memory_type == "main" else self.task_memory_path + + def _read_entries(self, memory_type: str) -> List[str]: + """读取记忆条目列表（形如 1. 内容）。忽略非编号行。""" + path = self._get_path(memory_type) + if not path.exists(): + return [] + try: + with open(path, 'r', encoding='utf-8') as f: + lines = f.read().splitlines() + except Exception: + return [] + entries: List[str] = [] + for line in lines: + match = re.match(r'^\s*(\d+)\.\s*(.*\S)?', line) + if match: + entries.append(match.group(2) or '') + return entries + + def _write_entries(self, memory_type: str, entries: List[str]) -> bool: + path = self._get_path(memory_type) + try: + path.parent.mkdir(parents=True, exist_ok=True) + with open(path, 'w', encoding='utf-8') as f: + for idx, text in enumerate(entries, start=1): + f.write(f"{idx}. {text.strip()}\n") + return True + except Exception as e: + print(f"{OUTPUT_FORMATS['error']} 写入记忆条目失败: {e}") + return False + + def update_entries(self, memory_type: str, operation: str, content: Optional[str] = None, index: Optional[int] = None) -> Dict: + """ + 条目式更新：append/replace/delete + append: 只需 content，自动在末尾追加编号 + replace: 需要 index + content + delete: 只需 index + """ + entries = self._read_entries(memory_type) + op = operation.lower() + + if op == "append": + if not content or not str(content).strip(): + return {"success": False, "error": "append 需要 content"} + entries.append(str(content).strip()) + success = self._write_entries(memory_type, entries) + return {"success": success, "operation": op, "memory_type": memory_type, "count": len(entries)} + + if op == "replace": + if index is None or index <= 0: + return {"success": False, "error": "replace 需要有效的 index（从1开始）"} + if not content or not str(content).strip(): + return {"success": False, "error": "replace 需要 content"} + if index > len(entries): + return {"success": False, "error": f"序号 {index} 超出当前记忆条目数 {len(entries)}"} + entries[index - 1] = str(content).strip() + success = self._write_entries(memory_type, entries) + return {"success": success, "operation": op, "memory_type": memory_type, "index": index, "count": len(entries)} + + if op == "delete": + if index is None or index <= 0: + return {"success": False, "error": "delete 需要有效的 index（从1开始）"} + if index > len(entries): + return {"success": False, "error": f"序号 {index} 超出当前记忆条目数 {len(entries)}"} + entries.pop(index - 1) + success = self._write_entries(memory_type, entries) + return {"success": success, "operation": op, "memory_type": memory_type, "index": index, "count": len(entries)} + + return {"success": False, "error": f"未知操作: {operation}"} diff --git a/utils/tool_result_formatter.py b/utils/tool_result_formatter.py index 8bcb07e..07373cd 100644 --- a/utils/tool_result_formatter.py +++ b/utils/tool_result_formatter.py @@ -389,9 +389,18 @@ def _format_update_memory(result_data: Dict[str, Any]) -> str: return _format_failure("update_memory", result_data) mem_type = result_data.get("memory_type") or "main" operation = result_data.get("operation") or "write" - verb = "追加" if operation == "append" else "覆盖" label = "主记忆" if mem_type == "main" else "任务记忆" - return f"{label}已{verb}完成。" + idx = result_data.get("index") + count = result_data.get("count") + if operation == "append": + suffix = f"（共 {count} 条）" if count is not None else "" + return f"{label}已追加新条目{suffix}" + if operation == "replace": + return f"{label}第 {idx} 条已替换。" + if operation == "delete": + suffix = f"（剩余 {count} 条）" if count is not None else "" + return f"{label}第 {idx} 条已删除{suffix}" + return f"{label}已更新。" def _format_create_sub_agent(result_data: Dict[str, Any]) -> str: