Coverage for prompt.py: 23%

1""" 

2System Prompt 构建模块。 

3参考 OpenClaw 的分段设计，每个模块独立，最后组合。 

4""" 

5import os 

6import platform 

7from pathlib import Path 

8from qrclaw.config import AGENT_NAME 

9from qrclaw.memory import LongTermMemory 

10from qrclaw.skills.registry import SkillRegistry 

11from qrclaw.tools.registry import get_schemas 

12from qrclaw.logger import get_logger 

13 

14logger = get_logger("qrclaw.prompt") 

15 

16_TOOL_DESCRIPTIONS = { 

17 "read_file": "读取本地文件内容，需要查看文件时使用", 

18 "write_file": "写入或创建本地文件，用户要求保存代码/文档时直接使用", 

19 "list_directory": "列出目录下的文件和子目录，不知道目录结构时先用它探索", 

20 "web_search": "联网搜索获取最新信息，查找文档、新闻、技术资料时使用", 

21 "web_fetch": "访问指定网页并提取纯净的 Markdown 正文，适合阅读文章、文档", 

22 "run_shell": "执行 shell 命令，需要运行程序、安装依赖、操作系统时使用", 

23 "write_memory": "写入中期记忆，仅用于记录用户偏好、项目配置等需要跨会话复用的信息，任务结果、调研报告等不要写入", 

24 "read_memory": "读取中期记忆，查看之前记录的用户偏好或配置信息", 

25 "review_memory": "审查中期记忆，识别过时、重复内容，支持分析和清理操作。建议定期调用维护记忆质量", 

26 "use_skill": "使用指定的技能（Skill）来完成复杂任务，技能是预定义的工作流", 

27 "create_plan": "为复杂任务创建执行计划，拆解步骤时标注依赖关系，无依赖的步骤可用 spawn_agent 并行执行", 

28 "spawn_agent": "在后台启动子 agent 并行执行独立子任务，任务可拆分时批量调用，子 agent 完成后结果自动打印", 

29 "wait_agents": "等待所有后台子 agent 完成，返回各子 agent 的结构化摘要，再根据摘要决定是否用 read_file 查看详情", 

30} 

31 

32 

33def _build_tooling_section(tool_names: list[str]) -> str: 

34 """ 

35 动态构建工具列表。 

36 优先从 registry 获取实时的 schema 描述， 

37 如果 registry 里没有（理论上不应该），再回退到 _TOOL_DESCRIPTIONS。 

38 """ 

39 if not tool_names: 

40 return "" 

41 

42 # 1. 动态获取所有已注册工具的描述 

43 # 格式: {'read_file': '读取本地文件...', 'web_search': '联网搜索...'} 

44 dynamic_descriptions = {} 

45 try: 

46 schemas = get_schemas() 

47 for schema in schemas: 

48 name = schema.get("function", {}).get("name") 

49 desc = schema.get("function", {}).get("description") 

50 if name and desc: 

51 dynamic_descriptions[name] = desc 

52 except Exception as e: 

53 logger.warning(f"获取工具 schema 失败: {e}") 

54 

55 lines = ["## 可用工具", "工具名称区分大小写，按名称精确调用："] 

56 

57 for name in tool_names: 

58 # 2. 优先用动态描述，没有则用硬编码的备用 

59 desc = dynamic_descriptions.get(name) or _TOOL_DESCRIPTIONS.get(name, "") 

60 

61 if desc: 

62 lines.append(f"- {name}: {desc}") 

63 else: 

64 # 如果真的都没有，只显示名字 

65 lines.append(f"- {name}") 

66 

67 return "\n".join(lines) 

68 

69 

70def _build_safety_section() -> str: 

71 return "\n".join([ 

72 "## 安全边界", 

73 "- 不执行破坏性命令，例如 rm -rf /、格式化磁盘、删除系统文件", 

74 "- 不读取敏感文件，例如 ~/.ssh/id_rsa、密钥文件、密码文件", 

75 "- 执行危险操作前，必须先询问用户确认", 

76 "- 不尝试绕过任何安全限制", 

77 ]) 

78 

79 

80def _build_workspace_section() -> str: 

81 cwd = os.getcwd() 

82 os_name = platform.system() 

83 return "\n".join([ 

84 "## 工作环境", 

85 f"- 操作系统：{os_name}", 

86 f"- 当前工作目录：{cwd}", 

87 "- 文件操作默认相对于当前工作目录", 

88 ]) 

89 

90 

91def _build_behavior_section(is_sub_agent: bool = False) -> str: 

92 lines = [ 

93 "## 行为准则", 

94 "- 优先用工具完成任务，不要凭空猜测文件内容或命令结果", 

95 "- 每次只调用一个工具，观察结果后再决定下一步", 

96 "- 任务完成后，用简洁的语言告诉用户做了什么、结果是什么", 

97 "- 遇到错误时，说明原因并给出解决建议", 

98 "- 语言简洁，不废话", 

99 "", 

100 "## 工具调用风格", 

101 "- 常规、低风险的工具调用直接执行，不要先解释再询问用户", 

102 "- 只在以下情况才先说明：删除等敏感操作、用户明确要求时", 

103 "- 当有专用工具可以完成某个操作时，直接用工具，不要让用户自己去跑命令", 

104 ] 

105 

106 if is_sub_agent: 

107 # 子 agent 专用：汇报压缩规则 

108 lines.extend([ 

109 "", 

110 "## ⚠️ 你是子 Agent", 

111 "你由主 Agent 派生，完成任务后必须向主 Agent 汇报。", 

112 "", 

113 "**汇报规则（极重要）：**", 

114 "1. 只返回结构化摘要，禁止输出完整执行过程", 

115 "2. 摘要格式：", 

116 "```", 

117 "## 任务结果", 

118 "- 状态：成功/失败/部分完成", 

119 "- 完成内容：[一段话概括做了什么]", 

120 "- 关键产出：[文件路径/关键数据，如有]", 

121 "- 遇到问题：[如有，简述]", 

122 "```", 

123 "3. 禁止包含：详细日志、代码片段、长篇解释", 

124 "4. 控制在 500 字以内", 

125 ]) 

126 else: 

127 # 主 agent 专用：plan 触发规则 + 多 agent 协作规则 

128 lines.extend([ 

129 "", 

130 "## 🚨 强制规则：何时必须使用 create_plan", 

131 "", 

132 "遇到以下情况，**第一反应必须调用 create_plan**，禁止直接开始执行：", 

133 "1. 任务包含 3 个及以上步骤", 

134 "2. 任务涉及多个文件或目录", 

135 "3. 任务需要探索未知环境（如分析新项目）", 

136 "4. 任务可能需要并行处理（如同时处理多个模块）", 

137 "5. 任务失败需要回滚或重试", 

138 "", 

139 "**错误示例**：用户说「分析这个项目」，直接 list_directory 然后读文件", 

140 "**正确示例**：用户说「分析这个项目」，先调用 create_plan 创建分析计划", 

141 "", 

142 "## 多 Agent 协作", 

143 "- 遇到以下情况，主动使用 spawn_agent 创建子 agent 并行处理：", 

144 " 1. 任务可以拆分成多个独立子任务（互不依赖）", 

145 " 2. 需要同时处理多个文件、模块或方向", 

146 " 3. 任务预计耗时较长，可以并行加速", 

147 "- 使用方式：先批量调用 spawn_agent 启动所有子 agent，再调用 wait_agents 等待结果", 

148 "- 子 agent 只返回摘要，如需详情用 read_file 查看日志文件", 

149 "- 简单任务、单一任务不需要子 agent，直接自己完成", 

150 ]) 

151 

152 return "\n".join(lines) 

153 

154 

155def _build_memory_section(memory: LongTermMemory = None) -> str: 

156 """构建中期记忆部分""" 

157 if memory is None: 

158 memory = LongTermMemory() 

159 

160 content = memory.load() 

161 

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

163 # 记忆为空，不注入 

164 return "" 

165 

166 logger.info("注入中期记忆到 system prompt") 

167 return "\n".join([ 

168 "## 中期记忆", 

169 "以下是你之前记录的重要信息，请在回答时参考：", 

170 "", 

171 content, 

172 ]) 

173 

174 

175def _build_heartbeat_section(heartbeat_file: Path = None) -> str: 

176 """构建心跳任务部分""" 

177 if heartbeat_file is None or not heartbeat_file.exists(): 

178 return "" 

179 

180 try: 

181 content = heartbeat_file.read_text(encoding="utf-8") 

182 if not content.strip(): 

183 return "" 

184 

185 logger.info("注入心跳任务到 system prompt") 

186 return "\n".join([ 

187 "## 心跳任务", 

188 "以下是定期执行的维护任务，请在每次心跳触发时执行：", 

189 "", 

190 content, 

191 ]) 

192 except Exception as e: 

193 logger.warning(f"读取心跳任务文件失败: {e}") 

194 return "" 

195 

196 

197def _build_skills_section(skill_registry: SkillRegistry) -> str: 

198 """构建技能部分（轻量级描述）""" 

199 skills_list = skill_registry.get_skills_list() 

200 

201 if not skills_list: 

202 # 没有技能，不注入 

203 return "" 

204 

205 logger.info(f"注入 {len(skills_list)} 个技能到 system prompt") 

206 

207 lines = [ 

208 "## 可用技能", 

209 "以下是你可用的技能（Skills），用于完成复杂任务：", 

210 "", 

211 ] 

212 

213 for i, skill_info in enumerate(skills_list, 1): 

214 lines.append(f"{i}. {skill_info}") 

215 

216 lines.extend([ 

217 "", 

218 "使用建议：", 

219 "- 当用户要求完成某个任务时，检查是否有匹配的技能", 

220 "- 如果有匹配的技能，调用 use_skill 工具来执行", 

221 "- 例如：use_skill(skill_name='analyze-project', args={'project_path': '.'})", 

222 ]) 

223 

224 return "\n".join(lines) 

225 

226 

227def _build_plan_section(active_plan: dict | None) -> str: 

228 """构建当前执行计划部分""" 

229 if not active_plan: 

230 return "" 

231 lines = [ 

232 "## 当前执行计划", 

233 f"目标：{active_plan['goal']}", 

234 "", 

235 "步骤状态：", 

236 ] 

237 for step in active_plan["steps"]: 

238 status = "✅ 已完成" if step["done"] else "⬜ 待执行" 

239 lines.append(f"- Step {step['id']}: {step['description']} [{status}]") 

240 lines.extend([ 

241 "", 

242 "执行规则：", 

243 "1. 有依赖关系的步骤必须串行执行（前置步骤完成后再执行下一步）", 

244 "2. 无依赖关系的步骤可以用 spawn_agent 并行执行，再用 wait_agents 等待结果，效率更高", 

245 "3. 每完成一步调用 complete_step 标记完成后再继续", 

246 "4. 最终整合步骤必须等所有子任务完成后再执行", 

247 ]) 

248 return "\n".join(lines) 

249 

250 

251def build_system_prompt( 

252 tool_names: list[str] | None = None, 

253 memory: LongTermMemory = None, 

254 skill_registry: SkillRegistry | None = None, 

255 active_plan: dict | None = None, 

256 heartbeat_file: Path | None = None, 

257 is_sub_agent: bool = False, 

258) -> str: 

259 """构建完整的 system prompt 

260  

261 Args: 

262 tool_names: 可用工具列表 

263 memory: 长期记忆实例 

264 skill_registry: 技能注册表 

265 active_plan: 当前执行计划 

266 heartbeat_file: 心跳任务文件路径 

267 is_sub_agent: 是否是子 agent（影响行为准则） 

268 """ 

269 sections = [ 

270 f"你是 {AGENT_NAME}，一个运行在用户本地的自主 AI Agent。", 

271 "", 

272 _build_tooling_section(tool_names or []), 

273 "", 

274 _build_skills_section(skill_registry or SkillRegistry()), 

275 "", 

276 _build_behavior_section(is_sub_agent=is_sub_agent), 

277 "", 

278 _build_safety_section(), 

279 "", 

280 _build_workspace_section(), 

281 "", 

282 _build_memory_section(memory), 

283 "", 

284 _build_heartbeat_section(heartbeat_file), 

285 "", 

286 _build_plan_section(active_plan), 

287 ] 

288 return "\n".join(sections)