任务规划器与验证器 — 技能指南

该技能为 AI Agent 提供一套安全、分步式的任务管理系统。

快速安装

# 克隆仓库
git clone https://github.com/cerbug45/task-planner-validator.git
cd task-planner-validator

# 完成！无需任何依赖 — 纯 Python 标准库实现

验证安装

# 运行测试
python test_basic.py

# 运行示例
python examples.py

基础用法

1. 导入并初始化

from task_planner import TaskPlanner

# 创建规划器实例
planner = TaskPlanner(auto_approve=False)

2. 定义执行器（Executor）

def my_executor(action: str, parameters: dict):
    """您自定义的执行逻辑"""
    if action == "fetch_data":
        # 从 API、数据库等获取数据
        return {"data": [1, 2, 3]}
    elif action == "process_data":
        # 处理数据
        return {"processed": True}
    else:
        return {"status": "completed"}

3. 创建计划

steps = [
    {
        "description": "获取用户数据",
        "action": "fetch_data",
        "parameters": {"source": "database"},
        "expected_output": "用户列表"
    },
    {
        "description": "处理用户数据",
        "action": "process_data",
        "parameters": {"validation": True},
        "expected_output": "已处理的数据"
    }
]

plan = planner.create_plan(
    title="数据处理流水线",
    description="获取并处理用户数据",
    steps=steps
)

4. 验证并执行

# 验证计划
is_valid, warnings = planner.validate_plan(plan)
if warnings:
    print("警告:", warnings)

# 批准计划
planner.approve_plan(plan, approved_by="admin")

# 执行计划
success, results = planner.execute_plan(plan, my_executor)

# 获取执行摘要
summary = planner.get_execution_summary(plan)
print(f"进度: {summary['progress_percentage']}%")

核心特性

安全验证

自动识别危险操作：

steps = [
    {
        "description": "删除旧文件",
        "action": "delete_files",  # ⚠️ 危险操作！
        "parameters": {"path": "/data/old"},
        "safety_check": True,  # 系统将发出警告
        "rollback_possible": False  # 不可回滚
    }
]

试运行模式（Dry Run Mode）

仅模拟执行，不真实触发操作：

success, results = planner.execute_plan(
    plan, 
    my_executor, 
    dry_run=True  # 仅模拟
)

保存与加载计划

持久化存储计划以便复用：

# 保存计划
planner.save_plan(plan, "my_plan.json")

# 后续加载
loaded_plan = planner.load_plan("my_plan.json")

# 验证完整性
if loaded_plan.verify_integrity():
    planner.execute_plan(loaded_plan, my_executor)

错误处理

灵活控制错误行为：

success, results = planner.execute_plan(
    plan,
    my_executor,
    stop_on_error=False  # 出错后继续执行后续步骤
)

# 检查结果
for result in results:
    if not result['success']:
        print(f"步骤 {result['order']} 执行失败: {result['error']}")

步骤配置项

每个步骤支持以下参数：

{
    "description": str,          # 必填：人类可读的描述
    "action": str,               # 必填：操作标识符
    "parameters": dict,          # 必填：操作所需参数
    "expected_output": str,      # 必填：预期输出结果
    "safety_check": bool,        # 可选：启用安全验证（默认为 True）
    "rollback_possible": bool,   # 可选：是否支持回滚（默认为 True）
    "max_retries": int          # 可选：最大重试次数（默认为 3）
}

常见使用场景

API 编排

steps = [
    {
        "description": "身份认证",
        "action": "api_auth",
        "parameters": {"service": "github"},
        "expected_output": "认证令牌"
    },
    {
        "description": "获取数据",
        "action": "api_fetch",
        "parameters": {"endpoint": "/repos"},
        "expected_output": "仓库列表"
    }
]

数据流水线

steps = [
    {
        "description": "抽取数据",
        "action": "extract",
        "parameters": {"source": "database"},
        "expected_output": "原始数据"
    },
    {
        "description": "转换数据",
        "action": "transform",
        "parameters": {"rules": ["normalize", "validate"]},
        "expected_output": "清洗后的数据"
    },
    {
        "description": "加载数据",
        "action": "load",
        "parameters": {"destination": "warehouse"},
        "expected_output": "成功确认信息"
    }
]

系统自动化

steps = [
    {
        "description": "备份数据库",
        "action": "backup",
        "parameters": {"target": "postgres"},
        "expected_output": "备份文件路径",
        "rollback_possible": True
    },
    {
        "description": "更新数据库结构",
        "action": "migrate",
        "parameters": {"version": "2.0"},
        "expected_output": "迁移完成",
        "rollback_possible": True
    },
    {
        "description": "验证完整性",
        "action": "verify",
        "parameters": {"checks": ["all"]},
        "expected_output": "所有检查通过"
    }
]

最佳实践

1. 始终先执行验证

is_valid, warnings = planner.validate_plan(plan)
if not is_valid:
    print("计划验证失败！")
    for warning in warnings:
        print(f"  - {warning}")
    exit(1)

2. 使用描述性名称

# 推荐 ✅
{
    "description": "从 PostgreSQL 生产数据库中获取活跃用户",
    "action": "fetch_active_users_postgres_prod",
    ...
}

# 不推荐 ❌
{
    "description": "获取数据",
    "action": "get",
    ...
}

3. 显式标记危险操作

{
    "description": "删除超过 30 天的临时文件",
    "action": "cleanup_temp_files",
    "parameters": {"age_days": 30, "path": "/tmp"},
    "safety_check": True,      # ⚠️ 将触发安全警告
    "rollback_possible": False  # ⚠️ 不可撤销！
}

4. 使用试运行模式先行测试

# 总是先进行试运行
success, results = planner.execute_plan(plan, my_executor, dry_run=True)

if success:
    # 再正式执行
    success, results = planner.execute_plan(plan, my_executor, dry_run=False)

5. 优雅地处理错误

def safe_executor(action: str, parameters: dict):
    try:
        result = execute_action(action, parameters)
        return result
    except Exception as e:
        logging.error(f"执行 {action} 失败: {e}")
        raise  # 重新抛出异常，交由规划器统一处理

高级特性

自动化流程启用自动批准

# 对自动化工作流跳过人工审批环节
planner = TaskPlanner(auto_approve=True)

检查点系统（Checkpoint System）

# 对支持回滚的步骤，系统将自动创建检查点
# 查看检查点历史记录
checkpoints = planner.executor.checkpoint_stack

执行历史记录

# 查看执行历史
history = planner.executor.execution_history
for entry in history:
    print(f"{entry['timestamp']}: {entry['step_id']} - {entry['status']}")

自定义验证规则

# 向 SafetyValidator 添加自定义验证项
planner.safety_validator.dangerous_operations.append('my_dangerous_op')
planner.safety_validator.sensitive_paths.append('/my/sensitive/path')

故障排查

“执行前必须批准计划”

# 解决方案：先批准计划
planner.approve_plan(plan, approved_by="admin")
# 或启用自动批准模式
planner = TaskPlanner(auto_approve=True)

安全验证警告

# 审阅警告内容，确保相关操作确属有意为之
is_valid, warnings = planner.validate_plan(plan)
for warning in warnings:
    print(warning)

# 若操作本身安全，仍可手动批准
if is_valid:  # 计划仍有效，仅存在警告
    planner.approve_plan(plan)

步骤执行顺序异常

# 确保 order 字段值为连续整数
steps[0]['order'] = 1
steps[1]['order'] = 2
steps[2]['order'] = 3

文件结构

task-planner-validator/
├── task_planner.py      # 主库文件
├── examples.py          # 使用示例
├── test_basic.py        # 测试套件
├── README.md            # 完整文档
├── QUICKSTART.md        # 快速入门指南
├── API.md              # API 参考手册
├── SKILL.md            # 当前文件
└── LICENSE              # MIT 许可证

运行要求

• Python 3.8 或更高版本
• 无外部依赖！

测试

# 运行基础测试
python test_basic.py

# 运行示例脚本
python examples.py

# 两者均应显示 “✅ ALL TESTS PASSED”

获取帮助

• 📖 查阅 README.md 中的完整文档
• 🚀 参考 QUICKSTART.md 获取快速上手示例
• 📚 查看 API.md 获取完整 API 参考
• 💡 浏览 examples.py 查看真实可用代码
• 🐛 在 GitHub 上提交问题反馈

许可证

MIT 许可证 — 详见 LICENSE 文件

作者

cerbug45

• GitHub: @cerbug45

⭐ 若您觉得此项目有用，请在 GitHub 上为其点亮 Star！