starmen1265/kb-skills
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(dev-workflow): Phase 1 - 技能框架搭建

Browse Source 
- 创建 SKILL.md 定义9阶段流程
- 定义触发词: '开始开发...'
- 创建状态文件结构 state.json
- 创建需求收集模板 requirements.json
- 创建状态管理工具 state_manager.py
- 支持跨会话状态恢复
This commit is contained in:
starmen1265
2026-03-09 18:57:32 +08:00
parent e23b88c3fa
commit d19062fc1e
4 changed files with 580 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.planning/dev-workflow/requirements.json
+22
View File
@@ -0,0 +1,22 @@
{
"schema": "dev-workflow-requirements-v1",
"collected_at": null,
"project_type": null,
"summary": {
"name": null,
"description": null,
"core_value": null
},
"functional_requirements": [],
"technical_requirements": {
"language": null,
"framework": null,
"database": null,
"dependencies": []
},
"constraints": [],
"acceptance_criteria": [],
"out_of_scope": [],
"questions_asked": [],
"user_answers": []
}
.planning/dev-workflow/state.json
+88
View File
@@ -0,0 +1,88 @@
{
"schema": "dev-workflow-state-v1",
"project_name": null,
"current_phase": 0,
"current_stage": "initialized",
"started_at": null,
"updated_at": null,
"retry_count": 0,
"max_retries": 3,
"phases": {
"1": {
"name": "头脑风暴",
"status": "pending",
"started_at": null,
"completed_at": null,
"data": null
},
"2": {
"name": "方案设计",
"status": "pending",
"started_at": null,
"completed_at": null,
"agents": [],
"data": null
},
"3": {
"name": "方案审核",
"status": "pending",
"started_at": null,
"completed_at": null,
"data": null
},
"4": {
"name": "方案汇报",
"status": "pending",
"started_at": null,
"completed_at": null,
"user_decision": null,
"data": null
},
"5": {
"name": "Spec编排",
"status": "pending",
"started_at": null,
"completed_at": null,
"data": null
},
"6": {
"name": "并行开发",
"status": "pending",
"started_at": null,
"completed_at": null,
"waves": [],
"data": null
},
"7": {
"name": "测试",
"status": "pending",
"started_at": null,
"completed_at": null,
"test_results": null,
"data": null
},
"8": {
"name": "代码审查",
"status": "pending",
"started_at": null,
"completed_at": null,
"agents": [],
"issues": [],
"data": null
},
"9": {
"name": "修复验证",
"status": "pending",
"started_at": null,
"completed_at": null,
"fixes": [],
"verification": null,
"data": null
}
},
"context": {
"trigger_message": null,
"working_directory": null,
"project_type": null
}
}
dev-workflow/SKILL.md
+205
View File
@@ -0,0 +1,205 @@
---
name: dev-workflow
description: "开始开发..." 时使用 - 多代理协作开发编排技能，解决单Agent开发的对话卡住和开发不完整问题
---
# dev-workflow - 多代理协作开发编排
## 什么时候使用
当用户发送包含以下模式的消息时，此技能自动触发：
| 触发词 | 行为 | 示例 |
|--------|------|------|
| `开始开发` + 描述 | 启动完整开发流程 | "开始开发一个用户登录功能" |
| `开始开发` + 项目类型 | 启动项目开发 | "开始开发一个新的 kb-skill" |
## 核心价值
通过多代理协作和分阶段编排，解决单Agent开发的两大痛点：
1. **对话卡住** — 长时间开发填满上下文窗口，质量下降
2. **开发不完整** — 缺乏系统性流程，遗漏边界情况或测试
## 9阶段流程
```
┌─────────────────────────────────────────────────────────────────┐
│ DEV-WORKFLOW PIPELINE │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 阶段1: 头脑风暴 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 主Agent 通过问答澄清需求，直到足够进行完整开发 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段2: 方案设计 (3个子代理并行) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Agent A │ │ Agent B │ │ Agent C │ │
│ │ 方案1 │ │ 方案2 │ │ 方案3 │ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ ↓ │
│ 阶段3: 方案审核 (交叉审核) │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ A审BC → B审AC → C审AB → 汇总 → 综合方案 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段4: 方案汇报 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 主Agent 汇报综合方案 → 用户决策 (批准/修改/重新设计) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段5: Spec编排 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 生成开发规格 → 分析任务依赖 → 分波执行计划 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段6: 并行开发 (按波次执行) │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ Wave 1: [Task1, Task2] 并行 → Wave 2: [Task3] → ... │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段7: 测试 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 运行现有测试 + 运行生成测试 → 通过/失败(重试) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段8: 代码审查 (3个子代理并行) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ Agent X │ │ Agent Y │ │ Agent Z │ │
│ │ 完整审查│ │ 完整审查│ │ 完整审查│ │
│ └─────────┘ └─────────┘ └─────────┘ │
│ ↓ │
│ 阶段9: 修复验证 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 汇总问题 → 安排修复 → 验证子代理确认 → 汇报完成 │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
```
## 状态持久化
每个阶段完成后保存状态到 `.planning/dev-workflow/state.json`：
```json
{
"current_phase": 3,
"current_stage": "方案设计",
"started_at": "2026-03-09T18:00:00+08:00",
"updated_at": "2026-03-09T19:30:00+08:00",
"retry_count": 0,
"phases": {
"1": { "status": "completed", "data": {...} },
"2": { "status": "completed", "data": {...} },
"3": { "status": "in_progress", "agents": [...] }
}
}
```
支持从任意阶段恢复继续。
## 失败处理
| 场景 | 处理方式 |
|------|----------|
| 子代理失败 | 自动重试，最多3次 |
| 测试失败 | 自动重试，最多3次 |
| 审查发现问题 | 汇总后安排修复 |
| 超过3次重试 | 汇报用户等待决策 |
## 子代理配置
| 阶段 | 子代理数量 | 职责 |
|------|-----------|------|
| 方案设计 | 3 | 自由探索不同方案 |
| 方案审核 | 3 | 交叉审核其他方案 |
| 并行开发 | N (按任务数) | 执行开发任务 |
| 代码审查 | 3 | 独立完整审查 |
| 修复 | N (按问题数) | 修复发现的问题 |
| 验证 | 1 | 确认修复完成 |
## 汇报机制
每个阶段完成后向用户汇报：
```
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
DEV-WORKFLOW ► 阶段 X 完成
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
**阶段名称**: [阶段名称]
**状态**: ✓ 完成 / ⚠ 需要关注
**耗时**: [X分钟]
**成果摘要**:
- [成果1]
- [成果2]
**下一阶段**: [阶段名称]
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
```
## 目录结构
```
.planning/dev-workflow/
├── state.json # 当前状态
├── requirements.json # 收集的需求
├── solutions/ # 方案设计
│ ├── solution-A.md
│ ├── solution-B.md
│ └── solution-C.md
├── reviews/ # 方案审核
│ ├── review-A.md
│ ├── review-B.md
│ └── review-C.md
├── final-solution.md # 综合方案
├── spec.md # 开发规格
├── execution-plan.md # 执行计划
├── development/ # 开发过程
│ └── wave-1/
│ ├── task-1.json
│ └── task-2.json
├── test-results.json # 测试结果
├── code-review/ # 代码审查
│ ├── review-X.md
│ ├── review-Y.md
│ └── review-Z.md
└── final-report.md # 最终汇报
```
## 适用项目类型
| 类型 | 支持情况 | 说明 |
|------|----------|------|
| 通用软件项目 | ✓ 完全支持 | Python/JS/Go 等任意语言 |
| kb-skills 技能 | ✓ 完全支持 | Python + SKILL.md 规范 |
| Web应用 | ✓ 完全支持 | 前后端分离项目 |
| CLI工具 | ✓ 完全支持 | 命令行工具 |
## 执行流程
主Agent检测到触发词后：
1. **初始化状态文件** — 创建 `.planning/dev-workflow/` 目录和 `state.json`
2. **进入阶段1** — 开始头脑风暴问答循环
3. **按流程执行** — 依次完成9个阶段
4. **每阶段汇报** — 完成后向用户汇报进度
5. **最终汇报** — 所有阶段完成后生成完整报告
## 恢复机制
如果会话中断：
1. 主Agent读取 `state.json` 获取当前状态
2. 根据状态恢复到对应阶段
3. 继续执行未完成的阶段
4. 保持所有已完成的成果
## 注意事项
- **主Agent不参与实际开发** — 保持编排者角色，避免上下文膨胀
- **方案必须用户确认** — 阶段4汇报后等待用户决策
- **测试必须通过** — 阶段7测试失败需修复后才能继续
- **审查问题必须处理** — 阶段9汇总所有问题并修复
dev-workflow/state_manager.py
+265
View File
@@ -0,0 +1,265 @@
#!/usr/bin/env python3
"""
dev-workflow 状态管理工具
用于读写和更新 dev-workflow 的状态文件。
Usage:
python3 state_manager.py --init
python3 state_manager.py --status
python3 state_manager.py --phase 3
python3 state_manager.py --complete-phase 2
python3 state_manager.py --update-phase 3 --data '{"agents": ["A", "B", "C"]}'
"""
import argparse
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Any, Optional
STATE_FILE = Path(".planning/dev-workflow/state.json")
REQUIREMENTS_FILE = Path(".planning/dev-workflow/requirements.json")
def get_state_path() -> Path:
"""获取状态文件路径"""
# 从当前目录向上查找项目根目录
cwd = Path.cwd()
while cwd != cwd.parent:
if (cwd / ".planning").exists():
return cwd / STATE_FILE
cwd = cwd.parent
return Path.cwd() / STATE_FILE
def load_state() -> dict:
"""加载状态文件"""
state_path = get_state_path()
if not state_path.exists():
print(f"❌ 状态文件不存在: {state_path}", file=sys.stderr)
sys.exit(1)
with open(state_path, "r", encoding="utf-8") as f:
return json.load(f)
def save_state(state: dict) -> None:
"""保存状态文件"""
state_path = get_state_path()
state["updated_at"] = datetime.now().isoformat()
state_path.parent.mkdir(parents=True, exist_ok=True)
with open(state_path, "w", encoding="utf-8") as f:
json.dump(state, f, ensure_ascii=False, indent=2)
print(f"✅ 状态已保存: {state_path}")
def init_state(project_name: str = None, trigger_message: str = None) -> dict:
"""初始化状态文件"""
state_path = get_state_path()
state = {
"schema": "dev-workflow-state-v1",
"project_name": project_name,
"current_phase": 1,
"current_stage": "头脑风暴",
"started_at": datetime.now().isoformat(),
"updated_at": datetime.now().isoformat(),
"retry_count": 0,
"max_retries": 3,
"phases": {
"1": {"name": "头脑风暴", "status": "in_progress", "started_at": datetime.now().isoformat()},
"2": {"name": "方案设计", "status": "pending"},
"3": {"name": "方案审核", "status": "pending"},
"4": {"name": "方案汇报", "status": "pending"},
"5": {"name": "Spec编排", "status": "pending"},
"6": {"name": "并行开发", "status": "pending"},
"7": {"name": "测试", "status": "pending"},
"8": {"name": "代码审查", "status": "pending"},
"9": {"name": "修复验证", "status": "pending"}
},
"context": {
"trigger_message": trigger_message,
"working_directory": str(Path.cwd()),
"project_type": None
}
}
save_state(state)
return state
def get_status() -> str:
"""获取当前状态摘要"""
state = load_state()
phase = state["current_phase"]
stage = state["current_stage"]
started = state.get("started_at", "N/A")
completed = sum(1 for p in state["phases"].values() if p.get("status") == "completed")
total = len(state["phases"])
status = f"""
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
DEV-WORKFLOW STATUS
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Project: {state.get("project_name", "未命名")}
Phase: {phase}/9 - {stage}
Progress: {completed}/{total} phases completed
Started: {started}
Retries: {state.get("retry_count", 0)}/{state.get("max_retries", 3)}
Phases:
"""
for num, phase_data in state["phases"].items():
status_icon = {
"completed": "✓",
"in_progress": "→",
"pending": "○",
"failed": "✗"
}.get(phase_data.get("status", "pending"), "?")
status += f" {status_icon} Phase {num}: {phase_data['name']}\n"
status += "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
return status
def set_phase(phase_num: int) -> None:
"""设置当前阶段"""
state = load_state()
if str(phase_num) not in state["phases"]:
print(f"❌ 无效的阶段: {phase_num}", file=sys.stderr)
sys.exit(1)
# 标记前一个阶段为完成
current = state["current_phase"]
if current > 0 and state["phases"][str(current)].get("status") == "in_progress":
state["phases"][str(current)]["status"] = "completed"
state["phases"][str(current)]["completed_at"] = datetime.now().isoformat()
# 设置新阶段
state["current_phase"] = phase_num
state["current_stage"] = state["phases"][str(phase_num)]["name"]
state["phases"][str(phase_num)]["status"] = "in_progress"
state["phases"][str(phase_num)]["started_at"] = datetime.now().isoformat()
save_state(state)
print(f"✅ 已进入阶段 {phase_num}: {state['current_stage']}")
def complete_phase(phase_num: int) -> None:
"""标记阶段完成"""
state = load_state()
if str(phase_num) not in state["phases"]:
print(f"❌ 无效的阶段: {phase_num}", file=sys.stderr)
sys.exit(1)
state["phases"][str(phase_num)]["status"] = "completed"
state["phases"][str(phase_num)]["completed_at"] = datetime.now().isoformat()
# 自动进入下一阶段
next_phase = phase_num + 1
if str(next_phase) in state["phases"]:
state["current_phase"] = next_phase
state["current_stage"] = state["phases"][str(next_phase)]["name"]
state["phases"][str(next_phase)]["status"] = "pending"
save_state(state)
print(f"✅ 阶段 {phase_num} 已完成")
def update_phase_data(phase_num: int, data: dict) -> None:
"""更新阶段数据"""
state = load_state()
if str(phase_num) not in state["phases"]:
print(f"❌ 无效的阶段: {phase_num}", file=sys.stderr)
sys.exit(1)
if "data" not in state["phases"][str(phase_num)]:
state["phases"][str(phase_num)]["data"] = {}
state["phases"][str(phase_num)]["data"].update(data)
save_state(state)
print(f"✅ 阶段 {phase_num} 数据已更新")
def increment_retry() -> bool:
"""增加重试计数，返回是否还可以继续重试"""
state = load_state()
state["retry_count"] = state.get("retry_count", 0) + 1
max_retries = state.get("max_retries", 3)
can_retry = state["retry_count"] < max_retries
save_state(state)
if can_retry:
print(f"⚠️ 重试 {state['retry_count']}/{max_retries}")
else:
print(f"❌ 已达到最大重试次数 ({max_retries})")
return can_retry
def reset_retry() -> None:
"""重置重试计数"""
state = load_state()
state["retry_count"] = 0
save_state(state)
print("✅ 重试计数已重置")
def main():
parser = argparse.ArgumentParser(description="dev-workflow 状态管理工具")
parser.add_argument("--init", action="store_true", help="初始化状态文件")
parser.add_argument("--status", action="store_true", help="显示当前状态")
parser.add_argument("--phase", type=int, metavar="N", help="设置当前阶段")
parser.add_argument("--complete-phase", type=int, metavar="N", help="标记阶段完成")
parser.add_argument("--update-phase", type=int, metavar="N", help="更新阶段数据")
parser.add_argument("--data", type=str, help="JSON 数据 (与 --update-phase 配合)")
parser.add_argument("--retry", action="store_true", help="增加重试计数")
parser.add_argument("--reset-retry", action="store_true", help="重置重试计数")
parser.add_argument("--project-name", type=str, help="项目名称 (与 --init 配合)")
parser.add_argument("--trigger", type=str, help="触发消息 (与 --init 配合)")
args = parser.parse_args()
if args.init:
init_state(args.project_name, args.trigger)
elif args.status:
print(get_status())
elif args.phase:
set_phase(args.phase)
elif args.complete_phase:
complete_phase(args.complete_phase)
elif args.update_phase:
if not args.data:
print("❌ 需要提供 --data 参数", file=sys.stderr)
sys.exit(1)
update_phase_data(args.update_phase, json.loads(args.data))
elif args.retry:
increment_retry()
elif args.reset_retry:
reset_retry()
else:
parser.print_help()
if __name__ == "__main__":
main()