Coverage for tools / review_memory.py: 15%

1""" 

2review_memory 工具 

3 

4让 LLM 审查中期记忆，智能判断哪些该保留、删除、合并。 

5""" 

6from pydantic import BaseModel, Field 

7from qrclaw.tools.registry import register 

8from qrclaw.logger import get_logger 

9 

10logger = get_logger("qrclaw.tools.review_memory") 

11 

12 

13class ReviewMemoryArgs(BaseModel): 

14 action: str = Field( 

15 default="analyze", 

16 description="操作类型：analyze（分析并返回报告）、cleanup（执行清理）、archive（归档旧记忆）" 

17 ) 

18 keep_recent: int = Field( 

19 default=10, 

20 description="保留最近 N 条记忆（用于 archive 操作）" 

21 ) 

22 

23 

24def _call_llm(prompt: str) -> str: 

25 """调用 LLM""" 

26 from qrclaw.config import OPENAI_API_KEY, OPENAI_MODEL, OPENAI_BASE_URL 

27 from openai import OpenAI 

28 

29 client = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL or None) 

30 

31 response = client.chat.completions.create( 

32 model=OPENAI_MODEL, 

33 messages=[{"role": "user", "content": prompt}], 

34 ) 

35 

36 return response.choices[0].message.content 

37 

38 

39@register( 

40 description="审查中期记忆，由 LLM 智能判断哪些该保留、删除、合并。建议定期在 heartbeat 中调用。", 

41 args_model=ReviewMemoryArgs, 

42) 

43def review_memory(action: str = "analyze", keep_recent: int = 10) -> str: 

44 """ 

45 审查中期记忆。 

46 

47 Args: 

48 action: 操作类型 

49 - analyze: 分析记忆状态，返回报告（默认） 

50 - cleanup: 让 LLM 智能清理记忆 

51 - archive: 归档旧记忆（保留最近 N 条） 

52 keep_recent: 保留最近 N 条记忆（仅 archive 操作使用） 

53 

54 Returns: 

55 str: 分析报告或操作结果 

56 """ 

57 from qrclaw.agent import get_workspace 

58 from qrclaw.memory import LongTermMemory 

59 from qrclaw.workspace import Workspace 

60 from datetime import datetime 

61 import re 

62 

63 logger.info(f"审查中期记忆: action={action}, keep_recent={keep_recent}") 

64 

65 # 获取工作空间 

66 ws = get_workspace() or Workspace("default") 

67 memory = LongTermMemory(ws.memory_file) 

68 

69 # 读取记忆 

70 content = memory.load() 

71 if not content or content.strip() == "# QRClaw 中期记忆": 

72 return "中期记忆为空，无需审查" 

73 

74 if action == "analyze": 

75 # 让 LLM 分析记忆状态 

76 prompt = f"""请分析以下中期记忆，回答： 

77 

781. 总体评估：记忆质量如何？是否需要清理？ 

792. 过时信息：哪些内容已经过时（已完成的任务、旧的项目状态）？ 

803. 重复信息：是否有重复或相似的内容？ 

814. 重要信息：哪些内容必须保留（用户偏好、重要知识、长期配置）？ 

825. 清理建议：建议执行什么操作？ 

83 

84--- 

85{content} 

86---""" 

87 

88 logger.info("调用 LLM 分析记忆") 

89 result = _call_llm(prompt) 

90 logger.info("记忆分析完成") 

91 return result 

92 

93 elif action == "cleanup": 

94 # 让 LLM 清理记忆 

95 prompt = f"""你是记忆管理助手。请审查并清理以下中期记忆。 

96 

97规则： 

981. 删除过时信息：已完成的任务、旧的项目状态、不再相关的内容 

992. 合并重复信息：相似的内容合并为一条 

1003. 保留重要信息：用户偏好、工作习惯、重要知识、长期有效的配置 

1014. 保持格式：输出仍为 Markdown 格式，以 "# QRClaw 中期记忆" 开头 

102 

103直接输出清理后的记忆内容，不要解释。 

104 

105--- 

106{content} 

107---""" 

108 

109 logger.info("调用 LLM 清理记忆") 

110 new_content = _call_llm(prompt) 

111 

112 # 确保格式正确 

113 if not new_content.strip().startswith("# QRClaw 中期记忆"): 

114 new_content = "# QRClaw 中期记忆\n\n" + new_content 

115 

116 # 计算变化 

117 old_len = len(content) 

118 new_len = len(new_content) 

119 saved = old_len - new_len 

120 

121 # 保存 

122 memory.save(new_content) 

123 logger.info(f"记忆清理完成: {old_len} -> {new_len} 字符, 节省 {saved} 字符") 

124 

125 return f"记忆清理完成：\n- 原大小：{old_len} 字符\n- 新大小：{new_len} 字符\n- 节省：{saved} 字符\n\n清理后的记忆已保存。" 

126 

127 elif action == "archive": 

128 # 归档：按条目分割，保留最近 N 条 

129 parts = content.split("---\n\n") 

130 

131 # 过滤掉标题 

132 entries = [p.strip() for p in parts if p.strip() and not p.strip().startswith("# QRClaw")] 

133 

134 if len(entries) <= keep_recent: 

135 return f"条目数（{len(entries)}）未超过保留数量（{keep_recent}），无需归档" 

136 

137 to_archive = entries[:-keep_recent] 

138 to_keep = entries[-keep_recent:] 

139 

140 # 写入归档文件 

141 archive_file = ws.root / "ARCHIVE.md" 

142 archive_lines = ["# 归档记忆\n", f"\n归档时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n"] 

143 for e in to_archive: 

144 archive_lines.append("\n---\n\n") 

145 archive_lines.append(e) 

146 archive_file.write_text("".join(archive_lines), encoding="utf-8") 

147 

148 # 更新 MEMORY.md 

149 memory_lines = ["# QRClaw 中期记忆\n"] 

150 for e in to_keep: 

151 memory_lines.append("\n---\n\n") 

152 memory_lines.append(e) 

153 memory.save("".join(memory_lines)) 

154 

155 logger.info(f"记忆归档完成: 归档 {len(to_archive)} 条, 保留 {len(to_keep)} 条") 

156 

157 return f"归档完成：\n- 归档 {len(to_archive)} 条旧记忆到 ARCHIVE.md\n- 保留 {len(to_keep)} 条最近记忆" 

158 

159 else: 

160 return f"未知操作: {action}。可用操作: analyze, cleanup, archive"