""" Goal 工具 - 计划管理 提供 goal 工具供 LLM 管理执行计划。 """ from typing import Optional, List, TYPE_CHECKING from agent.tools import tool if TYPE_CHECKING: from .goal_models import GoalTree from .protocols import TraceStore # ===== 全局 GoalTree 状态管理 ===== _current_goal_tree = None def set_goal_tree(tree): """设置当前 GoalTree（由 AgentRunner 调用）""" global _current_goal_tree _current_goal_tree = tree def get_goal_tree(): """获取当前 GoalTree""" return _current_goal_tree # ===== LLM 可调用的 goal 工具 ===== @tool(description="管理执行计划，添加/完成/放弃目标，切换焦点") async def goal( add: Optional[str] = None, reason: Optional[str] = None, after: Optional[str] = None, under: Optional[str] = None, done: Optional[str] = None, abandon: Optional[str] = None, focus: Optional[str] = None, context: Optional[dict] = None ) -> str: """ 管理执行计划，添加/完成/放弃目标，切换焦点。 Args: add: 添加目标（逗号分隔多个） reason: 创建理由（逗号分隔多个，与 add 一一对应） after: 在指定目标后面添加（同层级） under: 为指定目标添加子目标 done: 完成当前目标，值为 summary abandon: 放弃当前目标，值为原因 focus: 切换焦点到指定 ID context: 工具执行上下文（包含 store 和 trace_id） Returns: str: 更新后的计划状态文本 """ tree = get_goal_tree() if tree is None: return "错误：GoalTree 未初始化" # 从 context 获取 store 和 trace_id store = context.get("store") if context else None trace_id = context.get("trace_id") if context else None return await goal_tool( tree=tree, store=store, trace_id=trace_id, add=add, reason=reason, after=after, under=under, done=done, abandon=abandon, focus=focus ) # ===== 核心逻辑函数 ===== async def goal_tool( tree: "GoalTree", store: Optional["TraceStore"] = None, trace_id: Optional[str] = None, add: Optional[str] = None, reason: Optional[str] = None, after: Optional[str] = None, under: Optional[str] = None, done: Optional[str] = None, abandon: Optional[str] = None, focus: Optional[str] = None, ) -> str: """ 管理执行计划。 Args: tree: GoalTree 实例 store: TraceStore 实例（用于推送事件） trace_id: 当前 Trace ID add: 添加目标（逗号分隔多个） reason: 创建理由（逗号分隔多个，与 add 一一对应） after: 在指定目标后面添加（同层级） under: 为指定目标添加子目标 done: 完成当前目标，值为 summary abandon: 放弃当前目标，值为原因 focus: 切换焦点到指定 ID Returns: 更新后的计划状态文本 """ changes = [] # 1. 处理 done（完成当前目标） if done is not None: if not tree.current_id: return f"错误：没有当前目标可以完成。当前焦点为空，请先使用 focus 参数切换到要完成的目标。\n\n当前计划：\n{tree.to_prompt()}" # 完成当前目标 # 如果同时指定了 focus，则不清空焦点（后面会切换到新目标） # 如果只有 done，则清空焦点 clear_focus = (focus is None) goal = tree.complete(tree.current_id, done, clear_focus=clear_focus) display_id = tree._generate_display_id(goal) changes.append(f"已完成: {display_id}. {goal.description}") # 推送事件 if store and trace_id: print(f"[DEBUG] goal_tool: calling store.update_goal for done: goal_id={goal.id}") await store.update_goal(trace_id, goal.id, status="completed", summary=done) else: print(f"[DEBUG] goal_tool: skip event push (store={store}, trace_id={trace_id})") # 检查是否有级联完成的父目标（complete方法已经处理，这里只需要记录） if goal.parent_id: parent = tree.find(goal.parent_id) if parent and parent.status == "completed": parent_display_id = tree._generate_display_id(parent) changes.append(f"自动完成: {parent_display_id}. {parent.description}（所有子目标已完成）") # 2. 处理 focus（切换焦点到新目标） if focus is not None: goal = tree.find_by_display_id(focus) if not goal: return f"错误：找不到目标 {focus}\n\n当前计划：\n{tree.to_prompt()}" tree.focus(goal.id) display_id = tree._generate_display_id(goal) changes.append(f"切换焦点: {display_id}. {goal.description}") # 3. 处理 abandon（放弃当前目标） if abandon is not None: if not tree.current_id: return f"错误：没有当前目标可以放弃。当前焦点为空。\n\n当前计划：\n{tree.to_prompt()}" goal = tree.abandon(tree.current_id, abandon) display_id = tree._generate_display_id(goal) changes.append(f"已放弃: {display_id}. {goal.description}") # 推送事件 if store and trace_id: print(f"[DEBUG] goal_tool: calling store.update_goal for abandon: goal_id={goal.id}") await store.update_goal(trace_id, goal.id, status="abandoned", summary=abandon) else: print(f"[DEBUG] goal_tool: skip event push (store={store}, trace_id={trace_id})") # 4. 处理 add if add is not None: # 检查 after 和 under 互斥 if after is not None and under is not None: return "错误：after 和 under 参数不能同时指定" descriptions = [d.strip() for d in add.split(",") if d.strip()] if descriptions: # 解析 reasons（与 descriptions 一一对应） reasons = None if reason: reasons = [r.strip() for r in reason.split(",")] # 如果 reasons 数量少于 descriptions，补空字符串 while len(reasons) < len(descriptions): reasons.append("") # 确定添加位置 if after is not None: # 在指定 goal 后面添加（同层级） target_goal = tree.find_by_display_id(after) if not target_goal: return f"错误：找不到目标 {after}\n\n当前计划：\n{tree.to_prompt()}" new_goals = tree.add_goals_after(target_goal.id, descriptions, reasons=reasons) changes.append(f"在 {tree._generate_display_id(target_goal)} 后面添加 {len(new_goals)} 个同级目标") elif under is not None: # 为指定 goal 添加子目标 parent_goal = tree.find_by_display_id(under) if not parent_goal: return f"错误：找不到目标 {under}\n\n当前计划：\n{tree.to_prompt()}" new_goals = tree.add_goals(descriptions, reasons=reasons, parent_id=parent_goal.id) changes.append(f"在 {tree._generate_display_id(parent_goal)} 下添加 {len(new_goals)} 个子目标") else: # 默认行为：添加到当前焦点下（如果有焦点），否则添加到顶层 parent_id = tree.current_id new_goals = tree.add_goals(descriptions, reasons=reasons, parent_id=parent_id) if parent_id: parent_display_id = tree._generate_display_id(tree.find(parent_id)) changes.append(f"在 {parent_display_id} 下添加 {len(new_goals)} 个子目标") else: changes.append(f"添加 {len(new_goals)} 个顶层目标") # 推送事件 if store and trace_id: print(f"[DEBUG] goal_tool: calling store.add_goal for {len(new_goals)} new goals") for goal in new_goals: await store.add_goal(trace_id, goal) else: print(f"[DEBUG] goal_tool: skip event push (store={store}, trace_id={trace_id})") # 如果没有焦点且添加了目标，自动 focus 到第一个新目标 if not tree.current_id and new_goals: tree.focus(new_goals[0].id) display_id = tree._generate_display_id(new_goals[0]) changes.append(f"自动切换焦点: {display_id}") # 返回当前状态 result = [] if changes: result.append("## 更新") result.extend(f"- {c}" for c in changes) result.append("") result.append("## Current Plan") result.append(tree.to_prompt()) return "\n".join(result) def create_goal_tool_schema() -> dict: """创建 goal 工具的 JSON Schema""" return { "name": "goal", "description": """管理执行计划。目标工具是灵活的支持系统，帮助你组织和追踪工作进度。 使用策略（按需选择）： - 全局规划：先规划所有目标，再逐个执行 - 渐进规划：走一步看一步，每次只创建下一个目标 - 动态调整：行动中随时 abandon 不可行的目标，创建新目标 参数： - add: 添加目标（逗号分隔多个） - reason: 创建理由（逗号分隔，与 add 一一对应） - after: 在指定目标后面添加同级目标。使用目标 ID。 - under: 为指定目标添加子目标。使用目标 ID。如已有子目标，追加到最后。 - done: 完成当前目标，值为 summary（记录关键结论） - abandon: 放弃当前目标，值为原因 - focus: 切换焦点到指定目标。使用目标 ID。 位置控制（优先使用 after）： - 不指定 after/under: 添加到当前 focus 下作为子目标（无 focus 时添加到顶层） - after="X": 在目标 X 后面添加兄弟节点（同层级） - under="X": 为目标 X 添加子目标 - after 和 under 不能同时指定 执行顺序： - done → focus → abandon → add - 如果同时指定 done 和 focus，会先完成当前目标，再切换焦点到新目标 示例： - goal(add="分析代码, 实现功能, 测试") - 添加顶层目标 - goal(add="设计接口, 实现代码", under="2") - 为目标2添加子目标 - goal(add="编写文档", after="3") - 在目标3后面添加同级任务 - goal(add="集成测试", after="2.2") - 在目标2.2后面添加同级任务 - goal(done="发现用户模型在 models/user.py") - 完成当前目标 - goal(done="已完成调研", focus="2") - 完成当前目标，切换到目标2 - goal(abandon="方案A需要Redis，环境没有") - 放弃当前目标 注意： - 目标 ID 的格式为 "1", "2", "2.1", "2.2" 等，在计划视图中可以看到 - reason 应该与 add 的目标数量一致，如果数量不一致，缺少的 reason 将为空 """, "parameters": { "type": "object", "properties": { "add": { "type": "string", "description": "添加目标（逗号分隔多个）" }, "reason": { "type": "string", "description": "创建理由（逗号分隔多个，与 add 一一对应）。说明为什么要做这些目标。" }, "after": { "type": "string", "description": "在指定目标后面添加（同层级）。使用目标的 ID，如 \"2\" 或 \"2.1\"。" }, "under": { "type": "string", "description": "为指定目标添加子目标。使用目标的 ID，如 \"2\" 或 \"2.1\"。" }, "done": { "type": "string", "description": "完成当前目标，值为 summary" }, "abandon": { "type": "string", "description": "放弃当前目标，值为原因" }, "focus": { "type": "string", "description": "切换焦点到指定目标。使用目标的 ID，如 \"2\" 或 \"2.1\"。" } }, "required": [] } }