帮助用户为 hybrid_exec 系统(脚本/LLM/Agent 混合执行,`src/mini_agent/hybrid_exec/`)构建、测试、调试、改进一个符合规范的"混合任务"(`TaskSpec`)——从零写出 `task_id`/`description`/`input_data`/`output_validator`,用 `mini-agent hybrid-exec run`(或工具 `run_hybrid_exec_task`/真实 `HybridExecutor`)跑一次验证,看 attempts 决策轨迹排查探索/修复/降级为什么没按预期走,用 `mini-agent hybrid-exec list/show`(或工具 `list_hybrid_exec_tasks`/`show_hybrid_exec_task`)检查/手工调整 `.agent/hybrid_exec/scripts/<task_id>/` 里落盘的脚本版本,最终决定是留作独立调用、命令行触发、Agent 工具调用,还是接入 workflow 的 `hybrid_step`。当用户说"帮我做一个hybrid_exec任务"、"这个任务用脚本/LLM/Agent混合执行"、"给xxx写个TaskSpec"、"这个hybrid task一直探索失败"、"脚本总是被修复失败降级"、"hybrid_step怎么配"、"用命令行跑一下这个hybrid任务"时使用。
Resources
1Install
npx skillscat add onewaymyway/mini-agent/hybrid-exec-task-generator Install via the SkillsCat registry.
hybrid_exec 任务生成器(构建 / 测试 / 调试 / 改进 TaskSpec)
用于 next_doc/hybrid_exec_design_plan.md 描述的脚本/LLM/Agent 混合执行
系统。核心问题是把用户的一个具体需求("从文本抽取实体"、"把日志汇总成
摘要"……)表达成一个规范的 TaskSpec,跑通"脚本优先、坏了先修脚本、
修不好再降级 LLM/Agent"这条决策链路,并在效果不理想时知道去哪一层排查。
完整字段/流程/存储细节见 docs/hybrid-exec-guide.md(本 skill 只摘录
生成/调试时必需的部分,字段语义有分歧时以该文档为准)。不用于生成workflow.yaml 里普通的 agent/python_step 类型 step——那是
workflow-generator skill 的场景;hybrid_exec 任务只有明确要用"脚本优先、
自动降级"这套机制时才用本 skill。
第一步:判断这个需求适不适合做成 hybrid_exec 任务
先问自己(或直接问用户):
- 会被重复调用吗? 只跑一次的一次性任务不值得,直接让 Agent 做或写
个一次性脚本更划算——hybrid_exec 的核心价值是"探索一次、固化成脚本、
以后重复调用时低成本命中",摊销的是重复调用的成本。 - 输入结构是否基本稳定? 每次调用形状差异很大(比如任务描述本身随
每次调用大幅变化)会导致脚本反复失配、频繁走修复/降级,收益不明显。
这种更适合直接用allow_tiers=(LLM,)或(AGENT,)长期走 Fallback,
不强求产出脚本。 - 能不能写出
output_validator? 能明确"什么样的产出算对"(哪怕只是
"是个 dict 且包含某几个 key"这种弱校验)时,探索/修复阶段的自动判定会
可靠很多;完全写不出校验标准的任务,dry-run 只能退化成"不抛异常就算
过",脚本质量没有兜底,需要更依赖人工抽查前几次产出。
都满足或至少满足前两条,才继续下面的构建流程。
第二步:起草 TaskSpec
关键字段(完整定义见 src/mini_agent/hybrid_exec/spec.py):
| 字段 | 怎么定 |
|---|---|
task_id |
稳定标识,脚本仓库按它归档。用蛇形命名 + 版本后缀习惯,如 extract_entities_v1;同一个 task_id 复用会命中已有脚本,改需求想重新探索要么换新 task_id,要么用 force_reexplore=True |
description |
自然语言目标描述,直接进 Explorer/Repairer 的 prompt。要具体:输入从哪个字段读、输出要是什么结构、边界条件怎么处理,写得越模糊,探索出来的脚本质量越随缘 |
input_data |
本次调用的真实输入样例(不是 schema 描述)——探索阶段会拿这份数据做 dry-run,必须是脚本 run(ctx) 里 ctx.params 能读到的真实值,不能是占位符 |
output_validator |
Callable[[Any], tuple[bool, str]],返回 (是否通过, 原因)。不传则"不抛异常即算成功"。优先写这个,哪怕只是弱校验(类型 + 必需 key),比完全不校验可靠得多 |
allow_tiers |
三层子集 (script, llm, agent)。先用 (script, llm) 控制成本,观察 LLMExplorer 是否够用;确实需要多轮探查环境/数据形状再逐步开 agent(成本明显更高) |
max_script_repair_attempts |
默认 2,脚本报错后最多修几轮(最后一轮若预算够且允许 AGENT 会升级到 AgentRepairer) |
force_reexplore |
调试/迭代 description 时常用,强制忽略仓库里已有脚本重新探索一次 |
agent_fs_write_enabled |
默认 False,只读沙箱;确实需要 Explorer/Repairer 拉起的 Agent 读写项目文件时才显式打开 |
不确定 description 怎么措辞、output_validator 怎么写、allow_tiers
该配哪几档时,先看 reference/task_templates.md——里面按"结构化信息
抽取 / 文本摘要 / 格式转换 / 简单分类判断 / 网页内容解析"五类常见任务
类型整理了可以直接抄改的措辞模板 + 校验函数工厂 + 建议档位,比每次都
从零现场设计快得多;当前任务不属于这五类之一时才需要完全现场设计。
第三步:跑一次验证(不要凭空猜结果)
优先用现成的接入方式,而不是每次都手写驱动脚本——这三条路径底层都是同
一个 HybridExecutor,结果和落盘的脚本仓库完全一致,选哪个只取决于
"Agent 在对话里操作"还是"人在终端里操作":
(a) 在当前 Agent 对话里直接调用工具(最省事,Agent 自己就能做,不用
落盘任何驱动脚本):
调用工具 run_hybrid_exec_task(
task_id="extract_entities_v1",
description="从输入文本中抽取人名/机构名,返回 {\"entities\": [...]}",
input_json='{"text": "张三在腾讯工作,李四在阿里巴巴。"}',
allow_tiers="script,llm",
)返回文本里直接带 ok=/tier_used=/script_version=/output=;ok=False
时还会带上 attempts 决策轨迹的简要串。想先看仓库里有没有这个 task_id、
现状如何,先调用 list_hybrid_exec_tasks() / show_hybrid_exec_task(task_id)。
(b) 用命令行(人在终端里手动验证,或要写进调试记录方便复现):
mini-agent hybrid-exec run extract_entities_v1 \
--field 'text=张三在腾讯工作,李四在阿里巴巴。' \
--desc '从输入文本中抽取人名/机构名,返回 {"entities": [...]}' \
--allow-tiers script,llm \
-v # 打印完整 attempts 决策轨迹,见第四步input_data 支持 --field key=value(可重复)/--input-file <path>/
位置参数 JSON 字符串/stdin 管道四种传法,Windows PowerShell 下优先用--field 或 --input-file(位置参数里的 JSON 双引号容易被 PowerShell
转义丢失,报 unrecognized arguments: xxx};见 docs/hybrid-exec-guide.md
§十四)。mini-agent hybrid-exec list / show <task_id> 对应第五步的仓库
检查,不用再手动 read_file 拼路径。
(c) 需要自定义 output_validator 时,才手写驱动脚本((a)/(b) 都不支持
传自定义校验函数,只能"不抛异常即算成功";确实需要精确校验产出结构时用
这条路径,用 bash 工具跑):
import sys
sys.path.insert(0, "src") # 若尚未 pip install -e .
from mini_agent.hybrid_exec import default_executor, TaskSpec, ExecutionTier
executor = default_executor(project_root=".") # 自动读该项目的 providers.json
result = executor.run(TaskSpec(
task_id="extract_entities_v1",
description="从输入文本中抽取人名/机构名,返回 {\"entities\": [...]}",
input_data={"text": "张三在腾讯工作,李四在阿里巴巴。"},
allow_tiers=(ExecutionTier.SCRIPT, ExecutionTier.LLM),
output_validator=lambda out: (
isinstance(out, dict) and isinstance(out.get("entities"), list),
f"期望 dict 且含 entities 列表,实际 {out!r}",
),
))
print("ok=", result.ok, "tier=", result.tier_used.value, "version=", result.script_version)
print("output=", result.output)
print("attempts=")
for a in result.attempts:
print(f" [{'OK' if a.ok else 'FAIL'}] {a.stage} ({a.tier.value}) {a.reason}")跑之前确认 providers.json 已配置好((a)/(b)/(c) 三条路径都会像主 Agent
一样自动加载,不需要手动传 model/provider/api_key;没配好会在真正发起
调用时报错,报错信息里能看出是 provider 没配还是 model 名不对)。若这个
任务将来要嵌入某个 python_step 脚本内部调用,改用default_executor(ctx.project_root, llm=ctx.llm)——见 docs/hybrid-exec-guide.md
§一.1,此时不需要(也不应该)用 (a)/(b)/(c) 独立测;应直接在目标python_step 脚本里用 ctx.llm 现场验证,复用 workflow 当次已解析好的
provider/模型。
第四步:看 attempts 定位问题出在哪一层
ExecutionResult.attempts(CLI 用 -v 打印,工具 run_hybrid_exec_task
失败时自动带上)是完整决策轨迹,stage 命名规律:
stage 前缀 |
含义 | 没通过时该看什么 |
|---|---|---|
explore_llm / explore_llm_dryrun |
LLMExplorer 产出脚本 / 该脚本用真实 input_data 跑通过没 |
产出报错通常是 prompt 里 description 不够具体,或模型没理解 run(ctx) 协议(可以直接打印 LLMExplorer.explore() 产出的源码看哪里错);dryrun 报错看 reason 里的 traceback,多是 ctx.params key 名理解错了——回头改 description,别急着开 AGENT |
explore_agent / explore_agent_dryrun |
LLMExplorer 没通过后升级 AgentExplorer 的结果 | 只有 allow_tiers 含 AGENT 才会走到;仍失败通常是任务本身对"输入形状"依赖太强,input_data 样例不够有代表性 |
script_run / script_run_validate |
命中已有脚本后的真实执行 / 校验结果 | script_run 失败进修复阶段;script_run_validate 失败说明脚本能跑但结果不对,同样按失败处理进修复——先看是不是 output_validator 本身写太严格 |
repair_llm#N / repair_llm#N_dryrun |
第 N 轮 LLM 修复 / 该轮修复后 dry-run 结果 | 反复修不好,通常是原脚本的错误定位方式对 LLM 不友好——检查报错信息本身是否清晰(比如裸 Exception 而不是具体异常类型),必要时先人工把脚本改成报更明确的错,再让它继续修 |
repair_agent#N... |
最后一轮升级到 AgentRepairer | 同上,且只在 max_script_repair_attempts > 1 且允许 AGENT 时出现 |
fallback_llm / fallback_llm_validate |
脚本这条路彻底放弃后,LLM 直接给答案 | 走到这里说明脚本层面已经不可用,先看前面 explore_*/repair_* 为什么全失败,而不是只在这一层调 prompt |
fallback_agent / fallback_agent_validate |
最高层级兜底,没有更弱的降级空间 | 这里还是不通过,ok 会如实反映校验失败——回头重新审视 output_validator 是不是定义得不合理,或任务本身超出当前模型能力 |
proactive_reexplore_check |
仅 ReexplorePolicy 启用时出现,判断是否要机会主义重探索 |
与本次是否成功无关,不用于排查失败 |
调试口诀:先看最早失败的那个 stage,大概率问题根源就在那一层,不要
跳过它直接去改更下游(比如脚本 dry-run 就没过,却跑去调output_validator,治标不治本)。
第五步:检查/手工干预脚本仓库(有需要时)
脚本真实落盘在 .agent/hybrid_exec/scripts/<task_id>/(vN.py +meta.json)。查看用命令行 mini-agent hybrid-exec list(列举所有
task_id 及当前状态)/show <task_id>(某个 task_id 的 meta.json +vN.py 源码一起打印),或对话里让 Agent 调用工具list_hybrid_exec_tasks()/show_hybrid_exec_task(task_id),都不需要手动read_file/list_dir 拼路径了:
- 想看当前 active 的是哪个版本:
show_hybrid_exec_task(task_id)或mini-agent hybrid-exec show <task_id>直接给出active_version、每个
版本的success_count/fail_count/consecutive_fail/status。 - 脚本写得不理想,想直接手改:可以用
write_file/patch_file改vN.py本身(比如探索出来的脚本能跑但风格不好),改完必须用第三步
的方式((a)/(b)/(c) 任一)重新跑一次(走script_run/script_run_validate路径)验证没改坏;不要只改文件就假设它还能用。 - 想清空重来:删掉整个
<task_id>/目录,或调用时传force_reexplore=True(CLI 对应--force-reexplore,工具对应同名参数)
更安全(不用手动清目录,且失败了旧脚本还在)。 - 连续失败太多次被自动退役(
meta.json里status: retired):下次run()会自动重新走探索,不需要手动处理;如果想立刻验证退役后的重探索
行为,直接再跑一次第三步的方式即可。
第六步:定下来之后怎么用
四种落地形态,按场景选,不互斥(同一个 task_id 可以同时被多条路径
调用,共享同一份脚本仓库):
主 Agent 对话里直接用:确定好的
task_id/description直接让主
Agent 以后调用run_hybrid_exec_task(task_id=..., description=..., input_json=...)
即可,不用每次都重新走这个 skill 的构建流程——仓库里已有 active 脚本
时会直接命中。命令行触发(cron/systemd/CI,或人工手动跑一次):
mini-agent hybrid-exec run extract_entities_v1 --field 'text=...'独立调用(daemon 自主循环、其它 workflow 的
python_step
内部当库调用):把第三步 (c) 的驱动脚本思路整理成正式代码里的一次default_executor(project_root).run(TaskSpec(...))调用;嵌入某个python_step脚本内部时改用llm=ctx.llm(见第三步末尾)。接入 workflow(作为某个 workflow 的一个 step):写成
hybrid_step
类型(需要myplugins/hybrid_step.py插件已启用,删除该文件即禁用):- id: extract_entities type: hybrid_step depends_on: [fetch_text] params: task_id: extract_entities_v1 # 与前面验证时用的保持一致,直接复用已探索出的脚本 description: "从输入文本中抽取人名/机构名,返回 JSON" allow_tiers: [script, llm] max_script_repair_attempts: 2 result_required_keys: [entities] # 等价于给一个"必须含这些 key"的 output_validator这种写法下
hybrid_step会按step.model → wf.defaults.model → 全局 cfg.model解析出模型、构造一次共享的LLMHelper给探索/修复/兜底
共用,不需要在 YAML 里额外指定 provider——和主 Agent、独立调用路径的
模型解析规则一致。想先在沙箱里验证这个 step 而不是接入整条 DAG,用
workflow-debugger skill 的test_workflow_step。
常见坑
input_data传成了 schema 描述而不是真实样例:探索阶段的 dry-run
是拿这份数据真跑一次,传占位符/说明文字会导致 dry-run 通不过、误判为
"脚本写得不好",其实是输入给错了。output_validator抛异常:run_validator内部会捕获,但排查时容易
忘记检查校验函数本身是否对某些边界输入(比如output是None)会
抛异常——先确认output_validator自己够健壮。- 同一个
task_id换了完全不同的需求:会直接命中旧脚本、拿旧逻辑跑
新需求,产出通常是"能跑但结果不对"→ 触发修复循环、越修越怪。换需求要
么换新task_id,要么显式force_reexplore=True。 - 一上来就开
allow_tiers含AGENT:成本明显更高,多数任务(SCRIPT, LLM)就够,先跑通再按需升级。 - 独立执行场景下重复 new 出多个
default_executor():每次default_executor(project_root)内部只构造一次共享的LLMHelper(与
主 Agent 的LLMClientPool构造方式一致,含多 key 轮转/故障转移),但
这个共享是"每个HybridExecutor实例内部"的——如果在一个循环里反复default_executor(),也会反复重建。同一批调用应该复用同一个HybridExecutor实例(executor = default_executor(...)建一次,executor.run(task)多次调用),不要每次都重新default_executor()。 - 命令行位置参数传 JSON 在 Windows PowerShell 下报
unrecognized arguments: xxx}:PowerShell 把含双引号的字符串传给
原生 exe 时容易丢失内层引号转义,导致 JSON 被空格拆散。改用--field key=value(可重复)或--input-file <path>即可绕开,不要在
这个问题上排查 hybrid_exec 本身的逻辑(详见docs/hybrid-exec-guide.md§十四)。