feat: index-based memory tool with append/replace/delete

This commit is contained in:
JOJO 2025-12-14 17:56:18 +08:00
parent 8755688c8e
commit ca8f65fe35
3 changed files with 108 additions and 27 deletions

View File

@ -1474,15 +1474,16 @@ class MainTerminal:
"type": "function", "type": "function",
"function": { "function": {
"name": "update_memory", "name": "update_memory",
"description": "更新记忆文件", "description": "按条目更新记忆列表自动编号。append 追加新条目replace 用序号替换delete 用序号删除。",
"parameters": { "parameters": {
"type": "object", "type": "object",
"properties": { "properties": {
"memory_type": {"type": "string", "enum": ["main", "task"], "description": "记忆类型"}, "memory_type": {"type": "string", "enum": ["main", "task"], "description": "记忆类型"},
"content": {"type": "string", "description": "要添加的内容"}, "content": {"type": "string", "description": "条目内容。append/replace 时必填"},
"operation": {"type": "string", "enum": ["append", "replace"], "description": "操作类型"} "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": elif tool_name == "update_memory":
memory_type = arguments["memory_type"] memory_type = arguments["memory_type"]
content = arguments["content"]
operation = arguments["operation"] operation = arguments["operation"]
content = arguments.get("content")
index = arguments.get("index")
if memory_type == "main": # 参数校验
if operation == "append": if operation == "append" and (not content or not str(content).strip()):
success = self.memory_manager.append_main_memory(content) 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: else:
success = self.memory_manager.write_main_memory(content) result = self.memory_manager.update_entries(
else: memory_type=memory_type,
if operation == "append": operation=operation,
success = self.memory_manager.append_task_memory(content) content=content,
else: index=index
success = self.memory_manager.write_task_memory(content) )
result = {
"success": success,
"memory_type": memory_type,
"operation": operation
}
elif tool_name == "todo_create": elif tool_name == "todo_create":
result = self.todo_manager.create_todo_list( result = self.todo_manager.create_todo_list(

View File

@ -5,6 +5,7 @@ import json
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Dict, List, Optional from typing import Dict, List, Optional
import re
try: try:
from config import MAIN_MEMORY_FILE, TASK_MEMORY_FILE, DATA_DIR, OUTPUT_FORMATS from config import MAIN_MEMORY_FILE, TASK_MEMORY_FILE, DATA_DIR, OUTPUT_FORMATS
except ImportError: except ImportError:
@ -261,9 +262,8 @@ class MemoryManager:
# 主记忆统计 # 主记忆统计
if stats["main_memory"]["exists"]: if stats["main_memory"]["exists"]:
stat = self.main_memory_path.stat() stat = self.main_memory_path.stat()
content = self.read_main_memory()
stats["main_memory"]["size"] = stat.st_size 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( stats["main_memory"]["last_modified"] = datetime.fromtimestamp(
stat.st_mtime stat.st_mtime
).isoformat() ).isoformat()
@ -271,9 +271,8 @@ class MemoryManager:
# 任务记忆统计 # 任务记忆统计
if stats["task_memory"]["exists"]: if stats["task_memory"]["exists"]:
stat = self.task_memory_path.stat() stat = self.task_memory_path.stat()
content = self.read_task_memory()
stats["task_memory"]["size"] = stat.st_size 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( stats["task_memory"]["last_modified"] = datetime.fromtimestamp(
stat.st_mtime stat.st_mtime
).isoformat() ).isoformat()
@ -305,3 +304,76 @@ class MemoryManager:
except Exception as e: except Exception as e:
print(f"{OUTPUT_FORMATS['error']} 合并记忆失败: {e}") print(f"{OUTPUT_FORMATS['error']} 合并记忆失败: {e}")
return False 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}"}

View File

@ -389,9 +389,18 @@ def _format_update_memory(result_data: Dict[str, Any]) -> str:
return _format_failure("update_memory", result_data) return _format_failure("update_memory", result_data)
mem_type = result_data.get("memory_type") or "main" mem_type = result_data.get("memory_type") or "main"
operation = result_data.get("operation") or "write" operation = result_data.get("operation") or "write"
verb = "追加" if operation == "append" else "覆盖"
label = "主记忆" if mem_type == "main" 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: def _format_create_sub_agent(result_data: Dict[str, Any]) -> str: