06 从技能目录到调度任务:动态技能系统的最小闭环
结论先说:动态技能系统的难点不是“发现技能”,而是“把发现到的技能变成受控能力” #
很多人第一次做技能系统,都会被 importlib.import_module() 吸引,觉得动态加载最难。其实不是。
真正难的是后半段:
- 哪些目录算有效技能
- 哪些技能允许手动触发,哪些要自动调度
- 一个技能出错时,为什么不能拖垮整个系统
- 技能的调度元数据该放哪里,运行时又该信任到什么程度
一、最小闭环是什么 #
先给出这篇最重要的模型:
磁盘目录
-> SkillLoader 发现并导入
-> Runner 分类与注册
-> Scheduler 负责时间触发
-> dispatch 负责手动触发
-> Skill.run(context) 真正执行
对应伪代码:
class SkillLoader:
def discover(self, skill_root: str) -> list[Skill]:
skills = []
for subdir in os.listdir(skill_root):
skill = self._load_skill(subdir)
if skill:
skills.append(skill)
return skills
class Runner:
def setup(self, skills: list[Skill]):
for skill in skills:
if skill.interval or skill.cron:
self.scheduler.register(skill)
else:
self.manual_registry[skill.name] = skill
def dispatch(self, name: str, context: dict):
return self.manual_registry[name].run(context)
这个闭环的价值在于:发现、分类、调度、执行四步被拆开了。
二、加载器的职责:目录是不是一个合法技能 #
SkillLoader 不应该关心业务语义,它只该判断这个目录能不能变成可运行能力。
2.1 最小判定条件 #
从当前项目的设计看,一个目录至少要满足:
- 有
logic.py logic.py能被导入- 模块里暴露
run
# 伪代码:加载器判断技能是否合法
mod = importlib.import_module(f"skills.{dir_name}.logic")
if not hasattr(mod, "run"):
return None
return Skill(name=dir_name, run=mod.run, instruction=load_instruction(...))
2.2 为什么要宽容失败 #
如果某个技能因为:
- 语法错误
- 缺依赖
- 没有
run - instruction 不完整
就把整个加载过程炸掉,那这套系统根本不适合长期运行。
所以合理的策略是:
一个技能坏了,只跳过这一个技能,不影响其他技能注册。
这不是“放松约束”,而是隔离故障域。
三、为什么调度逻辑不应该写进 Loader #
这是这套设计里最值得讲的 tradeoff 之一。
方案 A:Loader 负责加载 + 调度 #
优点:简单,文件少 缺点:
- 职责混在一起
- 很难单独测试“调度分类是否正确”
- 加载器开始依赖时间系统和调度框架
方案 B:Loader 只负责发现,Runner 负责分类,Scheduler 负责时间触发 #
优点:
- 责任边界清楚
- 手动技能和自动技能共用同一个 Skill 抽象
- 测试容易拆开 缺点:
- 模块数变多
- 读代码时需要跨三层跳转
SecurityClaw 走的是 B。这是正确的。因为“动态加载”本质是文件系统问题,而“定时执行”本质是运行时编排问题,它们不是一件事。
四、调度元数据应该放在哪 #
当前设计把调度信息放进 instruction.md 或相邻元数据里,而不是写死在代码里。
这件事的意义很大:
4.1 让技能本身携带运行方式 #
一个技能如果带了:
intervalcronrun_on_first_startup
那它就不只是“一个函数”,而是“一个带运行契约的能力单元”。
4.2 让运行时可以统一解释 #
Runner 只看这些声明,不需要知道业务内容:
if skill.interval:
scheduler.add_interval_job(skill.run, skill.interval)
elif skill.cron:
scheduler.add_cron_job(skill.run, skill.cron)
else:
manual_registry[skill.name] = skill
这就是“声明式能力系统”的核心:能力由文件定义,运行方式由元数据声明,框架只负责解释。
五、完整时序:从目录到调度 #
sequenceDiagram
participant Disk as 技能目录
participant Loader as SkillLoader
participant Runner as Runner
participant Scheduler as Scheduler
Disk->>Loader: os.listdir(skills/)
Loader->>Loader: 导入 logic.py / 检查 run
alt skill 非法
Loader-->>Runner: 跳过
else skill 合法
Loader-->>Runner: 返回 Skill 对象
Runner->>Runner: 解析 instruction / manifest
alt 有 interval 或 cron
Runner->>Scheduler: 注册定时任务
else 无调度元数据
Runner->>Runner: 注册为手动技能
end
end
这张图里最值得注意的一点是:Scheduler 不需要知道技能来自哪里。它只接收一个已经被 Runner 分类过的“可执行能力”。
这正是分层价值所在。
六、失败路径:动态技能系统最容易炸的地方 #
6.1 空目录 #
应该跳过,不报 fatal
6.2 logic.py 缺失 #
应该跳过,不注册
6.3 run 不存在 #
应该跳过,并留下结构化 warning
6.4 元数据冲突 #
例如同时声明 interval 和 cron
这时必须有优先级规则,否则运行时语义不确定。
6.5 技能导入阶段连外部依赖 #
如果技能在 import 时就去连数据库或打网络,那 Loader 会变得非常脆弱。
正确做法应该是:
- import 阶段只加载定义
- 真正外部依赖放进
run(context)里
这是保证“发现”和“执行”分层的关键。
七、测试为什么要围绕 tmp_path 写 #
动态加载最难测的不是业务逻辑,而是目录形态。最好的办法就是:
- 临时建目录
- 临时写
logic.py - 临时写元数据
- 调
discover(tmp_path) - 验证结果
def test_skip_no_run(tmp_path):
skill_dir = tmp_path / "bad_skill"
skill_dir.mkdir()
(skill_dir / "logic.py").write_text("x = 1")
skills = loader.discover(tmp_path)
assert "bad_skill" not in [s.name for s in skills]
这种测试之所以重要,是因为它证明的不是“技能业务对不对”,而是“框架能不能在脏环境下保持稳定”。
八、这一篇要你真正记住的框架 #
动态技能系统不是“会 import 的插件机制”,而是“把目录、元数据、时间调度和手动触发整合成一个可控能力系统”。
你可以用这四层记它:
发现(Loader)
-> 分类(Runner)
-> 触发(Scheduler / dispatch)
-> 执行(Skill.run)
只要这四层混了两层,系统就会开始难测、难扩展、难恢复。
九、验证命令 #
pytest tests/test_skill_loader.py -v --tb=short
pytest tests/test_runner.py tests/test_scheduler.py -v