agently-dynamic-task

Agently 动态任务

当问题属于动态任务图时使用此技能：模型或应用提交必须经过规划、验证、剪枝、解析到处理器并执行的 DAG 数据。

TaskDAG 是 Agently 的 DAG 基础能力。动态任务是当前基于 TaskDAG 规划与执行的兼容性与便捷外观，不是 TriggerFlow 的子 API，也不是战略性 Agent 任务生命周期。TriggerFlow 是 TaskDAGExecutor 之下的执行基底，而 Agently.create_dynamic_task(...) 是当前面向应用的外观入口；agent.create_dynamic_task(...) 仍是面向旧版提示词快照调用方的兼容外观，不是默认指导。

当更底层的证据或运行时可视化很重要时，请使用 TaskDAGExecutor.compile_blocks(...) 或 async_run_blocks(...)：TaskDAG 仍然掌管图验证、依赖语义和语义输出；Blocks 仅将已验证的 DAG 片段降级为基于 TriggerFlow 的 ExecutionBlockGraph，并映射 EvidenceEnvelope/ResultAdapter 输出。默认的 TaskDAGExecutor.async_run(...) 路径不经过 Blocks；它直接将已验证的 DAG 编译到 TriggerFlow。

原生优先规则

• 当普通应用代码需要紧凑门面时，使用 Agently.create_dynamic_task(...)
• 当 DAG 是更广泛业务任务中的一个阶段时，独立运行该 DAG， 并将其快照作为证据传给 AgentExecution；不要 将 DynamicTask 暴露为 AgentExecution 路由候选，也不要将 Dynamic Task 门面变成长期运行的 AgentTask 生命周期所有者
• 仅当需要分阶段控制时，才拆分为 AgentlyTaskDAGPlanner、TaskDAGValidator、TaskDAGResolver 和 TaskDAGExecutor
• 当调用方需要 Blocks 生命周期证据路径时，使用 TaskDAGExecutor.compile_blocks(...) 或 async_run_blocks(...)；不要绕过 TaskDAGValidator，也不要将 Blocks 视为 DAG 所有者
• 当调用方已拥有 DAG 且应跳过模型规划时，使用 plan=<TaskDAG>
• 当提交的 DAG 是经过审查的配置产物时，使用 TaskDAG.from_yaml(...) 或 TaskDAG.from_json(...)
• 使用提交的 DAG inputs 占位符进行小型运行时接线： ${INIT} / ${INIT.foo} 用于初始图输入， ${DEPS.task_id.path} 用于依赖结果，${STATE...} 用于 执行本地状态，${TRIGGER...} 用于原始 TriggerFlow 触发 载荷。槽位名称不区分大小写，但示例应使用大写。 整字符串占位符保留原始值类型；嵌入式 占位符会字符串化到周围文本中。缺失的运行时路径在执行期间 失败关闭
• 当提交的 DAG 通过 create_dynamic_task(...) 或 TaskDAGExecutor 运行时，保持 ${INIT...} 连接到该门面/执行器拥有的提交图输入。 不要在新示例中使用旧的 AgentExecution 本地 DynamicTask 钩子；AgentExecution 通过 TaskGraph 和 ExecutionGraph 规划普通工作。
• 当模型必须生成 DAG 时，使用 planner=<agent-or-provider-settings>
• 使用 model=<agent-or-provider-settings> 作为模型任务执行资源
• 对于本地/自定义任务，使用 handlers={"risk_check_handler": handler}；处理程序名称应显式，通常以 _handler 结尾
• 仅当规划器应被允许使用这些任务类型时，才传递 actions=... 或 skills=...；不要默认暴露它们
• 对于模型拥有的语义输出，使用 Agently 输出模式，而不是用示例本地 JSON/代码块解析器解析模型文本
• 在示例中保持后端业务系统调用可模拟，但保持模型拥有的规划、分类、起草、评估、修订和最终响应生成由模型驱动
• 要求每个重要的模型任务都有清晰的输出模式，可以通过门面级 output_schema 或节点级 inputs.output_schema 实现
• 当生成或提交计划时，按结果类型设置模型任务 inputs.output_format： 省略它以让请求读取 prompt.default_output_format（全局默认 json）；对于紧凑的机器控制数据、操作 参数、路由标志、独立布尔值/数字、严格提取、 模型评判以及密集的全类型数组/对象，使用 json；对于扁平字符串长 文本/代码/HTML/SVG/Markdown/SQL/模板，使用 xml_field；当可接受重试延迟时，对于与类型化列表/对象/布尔值/数字字段混合的长 散文/代码字段，显式使用 hybrid；仅为了旧版 节标题兼容性，显式使用 flat_markdown；仅当有意需要 YAML 目标文档时，显式使用 yaml_literal；仅当 目标模型通过代表性稳定性检查后，可接受结构模式驱动选择和重试延迟时，使用 auto。qwen2.5:7b 检查 发现 hybrid 可能省略节标题或将旧脚手架注释回显到文本 字段中
• 让 TaskDAGValidator 在执行前拒绝重复 ID、缺失依赖、循环、不支持的必要任务类型、模式版本不匹配、不安全副作用策略以及未知的必要处理程序
• 仅当未知可选处理程序不影响必要的语义输出、必要的下游节点、审批或副作用策略时，才允许修剪它们
• 将生成的计划数据保留在执行输入或执行状态中；不要使用共享 TriggerFlow 流数据作为每次运行的 DAG 状态
• 对于小型声明式输入 接线，使用 ${INIT...} / ${DEPS...} 占位符；对于较大的转换，将逻辑保留在处理程序或模型任务中
• 不要教规划器将内部解析器条目（如 validate 或 emit）用作普通任务类型
• 不要仅仅因为普通只读模型任务调用 LLM 提供商就将其标记为网络副作用
• 除非用户明确要求审批，否则不要为分析、综合、起草或最终响应模型任务添加审批门

API 形态

提交的 DAG：

from agently import Agently

async def risk_check_handler(context):
    return {
        "task_id": context.task.id,
        "deps": dict(context.dependency_results),
    }

task = Agently.create_dynamic_task(
    target="review policy",
    plan={
        "graph_id": "review",
        "task_schema_version": "task_dag/v1",
        "tasks": [
            {"id": "extract", "kind": "local", "binding": "risk_check_handler"},
            {
                "id": "final",
                "kind": "local",
                "binding": "risk_check_handler",
                "depends_on": ["extract"],
                "inputs": {
                    "source": "${INIT}",
                    "extract_result": "${DEPS.extract}",
                },
            },
        ],
        "semantic_outputs": {"final": "final"},
    },
    handlers={"risk_check_handler": risk_check_handler},
)

snapshot = await task.async_start(timeout=10)
result = snapshot["semantic_outputs"]["final"]

由配置支持的提交 DAG：

from agently import Agently
from agently.core import TaskDAG

graph = TaskDAG.from_yaml("plans/review.yaml")
task = Agently.create_dynamic_task(
    target="review policy",
    plan=graph,
    handlers={"risk_check_handler": risk_check_handler},
)
snapshot = await task.async_run(graph_input={"doc": "policy"}, timeout=10)

对 JSON/JSON5 文件或原始内容使用 TaskDAG.from_json(...)。两个加载器都支持 task_dag_key_path="plans.review"，用于从更大的配置中选择一个 DAG。使用 graph.get_yaml(path) 或 graph.get_json(path) 导出规范化后的图。

自动规划的 DAG：

from agently import Agently

task = Agently.create_dynamic_task(
    target=user_goal,
    planner=planner_agent,
    model=worker_agent,
    output_schema={
        "answer": (str, "final user-facing answer", True),
    },
)

graph = await task.async_plan(max_retries=3)
task.validate(graph, strict_schema_version=True)
snapshot = await task.async_run(graph_input={"goal": user_goal}, timeout=30)

更底层的集成：

from agently.builtins.plugins import AgentlyTaskDAGPlanner
from agently.core import TaskDAGExecutor, TaskDAGResolver, TaskDAGValidator

resolver = TaskDAGResolver({"risk_check_handler": risk_check_handler})
validator = TaskDAGValidator(resolver)
planner = AgentlyTaskDAGPlanner(validator=validator)

graph = await planner.async_plan(planner_agent, {"target": "review policy"})
validation = validator.validate(graph, strict_schema_version=True)
snapshot = await TaskDAGExecutor(resolver, validator=validator).async_run(graph)

规划器约束

规划器插件和规划器提示词必须声明：

• 任务 schema 版本，通常为 task_dag/v1
• 允许的任务类型和可用的绑定名称
• 必需的语义输出叶子节点
• 当场景存在实际限制时的最大任务数量
• 会成为可重试的模型输出失败的校验检查
• 禁止行为：不稳定的任务 id、将依赖结果嵌入静态任务输入、隐式副作用、隐藏的 action/skill 使用，以及内部 resolver 任务类型

facade 规划器应接入 .output(planner.output_schema(), format="json")、必需的 planner.ensure_keys() 以及 .validate(planner.validate_output)，以便结构性问题在执行前成为模型输出校验失败。编译/运行仍会重新校验图。

层级归属：

• TaskDAG / TaskDAGNode 数据契约和 YAML/JSON 图配置属于 agently.types.data
• TaskDAGExecutor、TaskDAGValidator 和 TaskDAGResolver 属于核心层
• AgentlyTaskDAGPlanner 作为插件关注点，负责规划器输出 schema、ensure keys 和规划器指令
• Agently.create_dynamic_task(...) 是面向应用的 facade 入口； agent.create_dynamic_task(...) 仅为兼容性指引

反模式

• 不要把 Dynamic Task 指南放在 TriggerFlow 下，就好像它是 TriggerFlow 语法一样
• 不要为提交的 DAG 数据手写调度器
• 当 Agently 输出 schema 可以负责形状时，不要用 json.loads(...)、正则标记分割或代码块提取来解析模型文本
• 默认不要向规划器暴露 actions 或 skills
• 在推荐示例中，不要把模型负责的分类、起草、质量检查或最终响应生成隐藏在确定性的本地替代方案背后
• 当 Agently.create_dynamic_task(...) 已足够时，不要在普通应用代码中调用底层的 TaskDAGExecutor
• 不要把 Blocks 描述为 TaskDAG 执行的必需项；普通的 async_run(...) 路径是直接的，而 Blocks 需要显式的 compile_blocks(...) / async_run_blocks(...)
• 不要使用诸如 risk_check 这样含糊的 handler 名称；请使用 risk_check_handler 或其他能描述可调用角色名称的名字
• 不要在此处重复 TriggerFlow 文档；只需说明 Dynamic Task 在内部编译为 TriggerFlow

接下来阅读

• references/overview.md
• examples/minimal.py