# Stp > 结构化任务规划与分步执行 V2（异步子代理架构）。支持将每个步骤通过 session_spawn 创建子代理异步执行，主会话保持非阻塞。功能包括：步骤分解、子代理执行、子代理检验（LLM判断）、状态跟踪、Heartbeat 监控、任务中断。触发词：/stp、任务规划、步骤执行。 - Skill: `scotthuang/stp` (Agent Skill, multi-file: 5 files) - Install (CLI): `npx skillmds@latest add scotthuang/stp` - Raw SKILL.md: https://api.skillmd.com/api/skills/scotthuang/stp/raw - Safety review: pending (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - Author: scotthuang (https://skillmd.com/u/scotthuang) - Updated: 2026-09-10 - Page: https://skillmd.com/skills/scotthuang/stp --- > **路径变量说明**（本文档通用）： > - `` = `~/.openclaw/workspace/skills/stp` > - `` = `/scripts` > - `` = `~/.openclaw/workspace/task-list` > - `` = `~/.openclaw/workspace/tasks` --- # STP V2（异步子代理架构） ## 背景 **重大更新**：V2 采用子代理异步架构，每个步骤通过 `session_spawn` 创建独立的子代理执行，主会话保持非阻塞，彻底解决了 V1 长任务阻塞主会话的问题。 - **V1**：在主会话运行，长任务会阻塞整个会话 - **V2**：每个步骤通过 `session_spawn` 创建子代理执行，主会话非阻塞 ## 核心概念 ``` 主会话 (非阻塞) │ ├── session_spawn (步骤执行) → 执行子代理 │ │ │ └──→ announce 完成/失败 │ ├── session_spawn (步骤检验) → 检验子代理 (LLM) │ │ │ └──→ 返回通过/不通过 │ └── heartbeat (定时检查) → 监控子代理状态 ``` --- ## 一、计划书格式 ### 1.1 标准格式（含检验标准） ```markdown # 任务名称 ## 任务描述 用户原始需求的简要描述 ## 技术方案 - 使用的工具/库/API - 关键技术约束 ## 全局设置 - 步骤超时时间: [无超时 / N 分钟] - 任务完成后删除目录: [是 / 否]（默认否，保留任务记录） ## 核心执行步骤 - [ ] 步骤 1：具体描述 - **执行 Prompt**: 给执行子代理的具体指令 - **检验标准**: 给检验子代理（LLM）的验证条件，用于判断步骤是否成功完成 - [ ] 步骤 2：具体描述 - **执行 Prompt**: xxx - **检验标准**: xxx ... ## 预期产出 - 输出文件/结果说明 ``` ### 1.2 用户确认流程 ```markdown 📋 任务计划书已生成 任务名称：xxx 文件位置：~/.openclaw/workspace/tasks/task-xxx/task_steps.md 步骤数：3 全局设置： - 步骤超时时间：无超时 - 任务完成后删除目录：否（默认保留） 核心执行步骤： - [ ] 步骤 1：编写股票查询脚本 - 执行 Prompt：编写 Python 脚本，使用 AkShare 查询股票数据 - 检验标准：脚本存在于 temp/scripts/stock_query.py，可执行无报错 - [ ] 步骤 2：查询贵州茅台收盘价 - 执行 Prompt：运行脚本查询 600519.SH - 检验标准：输出包含"600519.SH"和收盘价数值 ... ======================================== 确认执行： 输入 "ok" 或 "确认" → 开始执行 输入 "取消" → 放弃此任务 输入 "修改" → 调整计划 输入 "超时 30" → 设置每个步骤超时 30 分钟 输入 "删除" 或 "清理" → 任务完成后删除目录 ======================================== ``` --- ## 二、任务目录结构 ``` ~/.openclaw/workspace/tasks/ └── task-{ID}/ ├── stp-plan-{ID}.md # 计划书 ├── task_steps.md # 步骤文档（含状态、子代理ID、超时计数） └── temp/ ├── scripts/ └── downloads/ ``` ### task_steps.md 格式（含子代理追踪） ```markdown ## 任务基础信息 - 任务名称：xxx - 任务ID：task-1 - 创建时间：2026-02-28 22:00:00 - 步骤超时时间：无超时 ## 核心执行步骤 ### 步骤 1：编写股票查询脚本 - **状态**: 执行中 - **执行子代理**: subagent:abc (runId: xyz) - **检验子代理**: 待创建 - **超时计数**: 执行(0/2) | 检验(0/2) - **执行 Prompt**: 编写 Python 脚本... - **检验标准**: 脚本存在于 temp/scripts/stock_query.py ### 步骤 2：查询贵州茅台 - **状态**: 待执行 ... ``` --- ## 三、执行流程 ### 3.1 状态机 ``` 待执行 → 执行中 → 待检验 → 检验中 → (通过) → 待执行(下一步) ↓ (不通过) 等待用户决策（调整/重试/终止） ``` **⚠️ 注意：执行中 → 待检验 → 检验中 是必经步骤，禁止跳过！** ### 3.2 主会话编排逻辑 #### 启动任务 **⚠️ 重要：必须先展示计划书并确认，才能执行！** **完整流程：** 1. **用户给任务** → AI 生成计划书内容（内存中） 2. **创建任务目录**：在 `~/.openclaw/workspace/tasks/` 下创建新的 `task-{ID}` 目录（ID 自增） 3. **保存计划书**：将计划书保存到 `task-{ID}/stp-plan-{ID}.md` 4. **展示计划书给用户**（包含文件位置） 5. **等待用户确认**（输入 "ok" / "确认" 等） 6. 用户确认后： - 调用 `stp_orchestrator.py start ` 初始化任务 - **读取 task_steps.md 获取步骤 1 的执行 Prompt** - **使用 sessions_spawn 启动执行子代理**： ```python sessions_spawn( task="<步骤 1 的执行 Prompt>", label="task-{ID}-step-1-exec", cleanup="keep" ) ``` - 更新 task_steps.md 中步骤 1 的状态为"执行中"，记录 exec_subagent - **回复用户时必须说明如何中断任务**，例如："如需终止请输入：中断 task-{ID}" 7. 用户取消 → **保留 task-{ID} 目录和计划书**（不删除） **禁止跳过确认步骤！** **⚠️ 重要：每个任务必须独立思考！** - 生成计划书时，**禁止**读取或参考已有的任务计划书（如 tasks/ 目录下的 .md 文件） - 即使任务内容相似，也必须从用户需求出发重新思考 - 不要复用旧计划的思路，每个任务都是全新的 #### 执行步骤 1. 主会话更新步骤状态为"执行中" 2. 创建执行子代理： ```bash sessions_spawn( task="<步骤的执行 Prompt>", label="task-{ID}-step-{N}-exec", cleanup="keep" ) ``` 3. 记录子代理信息到 task_steps.md： - 步骤状态改为"执行中" - 记录执行子代理 ID (subagent:xxx) - 记录 runId - 记录执行时间 4. 返回非阻塞响应 #### 检验步骤 **⚠️ 重要：执行子代理完成后，必须先检验才能执行下一步！禁止跳过检验步骤！禁止让主会话 LLM 直接判断检验结果！** 1. 收到执行子代理的 announce 2. **立即启动检验子代理**，不允许 LLM 自行判断 3. 创建检验子代理（LLM）： ```bash sessions_spawn( task="请根据以下检验标准判断步骤是否成功完成。 检验标准：{步骤的检验标准} 执行结果：{执行子代理的输出} 请返回：通过 / 不通过，并说明原因", label="task-{ID}-step-{N}-verify", cleanup="keep" ) ``` 3. 记录检验子代理信息到 task_steps.md： - 步骤状态改为"检验中" - 记录检验子代理 ID - 记录 runId 4. 检验子代理返回结果 5. 检验通过 → 更新状态为"已完成"，执行下一步 6. 检验不通过 → 询问用户：调整方案 / 重试 / 终止 #### 失败处理 - 执行失败 → 询问用户：调整方案 / 重试 / 终止 - 检验不通过 → 询问用户：调整方案 / 重试 / 终止 --- ## 四、Heartbeat 监控 ### 4.1 工作流程 1. **启动任务时**：`start` 命令自动创建 cron job（`stp-heartbeat-{task_id}`） 2. **Cron 触发**：每 10 分钟触发 isolated session，执行 `heartbeat ` 3. **检查状态**：对每个已知子代理调用 `sessions_history_sync` 获取实际状态 4. **基于实际状态判断**： - `tool_count == 0`：pending（等待开始） - `is_running == true`（正在等待工具返回或最近 5 分钟有活动）：running（执行中） - `is_running == false` 且超过 5 分钟无活动：completed（已完成） - 超过 30 分钟仍在工作中：stuck（卡住） ### 4.2 sessions_history_sync 返回状态 | 字段 | 说明 | |------|------| | tool_count | 工具调用次数 | | tool_call_count | toolCall 数量 | | tool_result_count | toolResult 数量 | | is_waiting | 等待工具返回中（toolCall > toolResult） | | is_running | 正在执行（is_waiting 或最近 5 分钟有活动） | | is_recent | 最近 5 分钟有活动 | ### 4.2 状态判断规则 | 状态 | 条件 | 处理 | |------|------|------| | pending | tool_count == 0 | 等待 | | running | 最近 5 分钟有活动 | 正常 | | completed | 超过 5 分钟无活动 | 通知用户，更新状态 | | stuck | 超过 30 分钟无活动 | 增加超时计数，>= 2 则告知用户 | ### 4.3 通知用户 - 如果 `completed_subagents` 有内容：通知用户子任务完成，需要继续检验 - 如果 `stuck_count > 0` 且超时计数 >= 2：提示用户决定是否重试或终止 ### 4.4 挂起处理示例 ``` ⚠️ 步骤 2 可能已挂死 执行子代理已等待 30+ 分钟无响应 Tool: exec (git clone ...) 请选择： - 继续等待 → 再次等待 10 分钟 - 重试 → 终止当前子代理，重新执行 - 终止 → 结束整个任务 ``` --- ## 五、任务中断 ### 5.1 触发方式 用户输入：`中断 {任务名称}` 或 `中断 task-{ID}` ### 5.2 中断流程 1. 解析中断命令，获取任务 ID（如 `task-23` → `23`） 2. 运行命令：`python3 /stp_orchestrator.py interrupt ` 3. 解析 JSON 输出，获取 `subagent_ids_for_kill` 列表 4. **在主会话中直接调用 subagents 工具杀掉每个子代理**： ```python subagents(action="kill", target="agent:main:subagent:xxx") ``` （注意：target 需要完整的 session key，如 `agent:main:subagent:xxx`） 5. **杀掉子代理后，检查并杀掉残留进程**： - 对每个被杀的子代理，调用 `sessions_history` 获取其执行历史 - 从历史中解析 exec 命令的返回结果，提取 `details.pid`（进程 PID） - 用 `kill ` 杀掉进程 - 如果解析不到 PID，再用关键词匹配作为后备方案 6. 自动删除对应的 cron job 7. 通知用户任务已中断（包括杀掉的残留进程 PID） --- ## 六、脚本说明 ### 6.1 stp_orchestrator.py（核心编排） ```bash # 启动任务（自动创建 cron job 用于 heartbeat） python3 /stp_orchestrator.py start # 查看任务状态 python3 /stp_orchestrator.py status # 检查 heartbeat（需要传入 task_id） python3 /stp_orchestrator.py heartbeat # 中断任务（自动删除对应的 cron job） python3 /stp_orchestrator.py interrupt ``` ### Cron Job 自动管理 - **启动任务时**：`start` 命令自动创建 cron job（`stp-heartbeat-{task_id}`），每 10 分钟检查一次 - **Heartbeat 检查**：cron job 触发 isolated session，执行 `heartbeat ` - **任务中断时**：`interrupt` 命令自动删除对应的 cron job - **任务完成时**：heartbeat 检测到以下情况会自动清理 cron： - 没有活跃子代理 - 所有子代理不在工作中（completed/idle） - 所有子代理会话不存在 - 检验通过后，cron 会收到清理信号并删除自己 **无需配置 HEARTBEAT.md**，完全自动化。 --- ## 七、使用示例 ### 7.1 自然语言模式 ```bash # 用户：帮我查三支股票价格 # AI 自动生成计划书，用户确认后： python3 /stp_orchestrator.py start ~/.openclaw/workspace/tasks/task-xxx/stp-plan-xxx.md ``` ### 7.3 主会话交互 ``` 用户: 帮我查三支股票价格 [AI 根据用户需求动态生成计划书] 📋 任务计划书已生成 文件：~/.openclaw/workspace/tasks/task-xxx/stp-plan-xxx.md 步骤数：3 ... 用户: ok ✅ 任务已启动 (task-1) 你可以继续做其他事，我会定期汇报进度 ⚠️ **如需终止任务**，请输入：中断 task-{ID} --- ### 7.4 主会话中断任务示例 ``` 用户: 中断 task-23 [主会话执行:] 1. 运行: python3 /stp_orchestrator.py interrupt 23 2. 解析输出获取 subagent_ids_for_kill: ["agent:main:subagent:xxx", ...] 3. 对每个 ID 调用 subagents 工具杀掉子代理 4. 对每个被杀的子代理： - 调用 sessions_history 获取执行历史 - 解析 exec 返回结果中的 details.pid - 用 kill 杀掉进程 - 如果没有 PID，用关键词匹配作为后备 5. 通知用户任务已中断 ``` --- ## 八、注意事项 1. **严格串行**：必须等上一步检验通过才能执行下一步 2. **子代理通信**：通过 announce 链通信，不使用 sessions_send 3. **状态持久化**：所有状态保存在 task_steps.md 4. **Heartbeat**：默认 10 分钟检查一次 5. **超时判定**：单个 tool 执行 30 分钟算超时，给 2 次机会（总共 60 分钟） --- ## 意见反馈 欢迎提交 Issue 或 Pull Request！ 🔗 GitHub：https://github.com/scotthuang/openclawSkills/tree/main/stp --- ## Changelog ### 2026-03-02 #### 修复 - Cron Job 添加 `--channel webchat` 参数，避免执行时报错 - task_step 输出清理信息：任务完成或中断时，在 `task_steps.md` 中记录清理信息（时间、终止的子代理、删除的 cron、终止的进程） ### 2026-03-01 #### 新增 - 每个 STP 任务创建独立目录 `task-{ID}/`，用户取消后也保留 - 计划书保存在 `stp-plan-{ID}.md` - 中断任务时自动杀掉子代理的残留进程（通过解析 sessions_history 获取 PID） - 强制使用检验子代理验证结果，禁止主会话 LLM 直接判断 #### 优化 - 目录结构简化，移除 `result.txt` 和 `task_execution.log` - 修正脚本命令说明，移除不存在的 `execute`、`verify`、`retry` 命令 - 添加"每个任务必须独立思考"的规则，禁止参考已有计划书 - 修正章节编号（7.3 交互、7.4 中断示例） #### 修复 - 修复中断任务时只杀子代理、不杀残留进程的 bug - 修复执行步骤后跳过检验子代理、直接启动下一步的 bug