agently-dynamic-task

Agently Dynamic Task

このスキルは、問題が動的なタスクグラフである場合に使用します。モデルまたはアプリケーションが DAG データを送信し、それを計画、検証、プルーニング、ハンドラーへの解決、および実行する必要があります。

TaskDAG は Agently の DAG 基盤機能です。Dynamic Task は、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 がより広範なビジネスタスク内の 1 フェーズである場合は、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=... を渡します。デフォルトで公開しないでください
• モデルが所有する意味的出力には、例ローカルの JSON/コードブロックパーサーでモデルのテキストを解析するのではなく、Agently の出力スキーマを使用してください
• 例ではバックエンドのビジネスシステム呼び出しをモック可能に保ちつつ、モデルが所有する計画、分類、起草、評価、修正、および最終応答生成はモデル駆動のままにしてください
• 重要な各モデルタスクには、ファサードレベルの output_schema またはノードレベルの inputs.output_schema のいずれかを通じて、明確な出力スキーマを必須にしてください
• 計画が生成または送信される際に、結果の型に応じてモデルタスクの inputs.output_format を設定します: 省略すると、リクエストは prompt.default_output_format を読み取ります (グローバルのデフォルトは json)。コンパクトなマシン制御データ、アクション 引数、ルーティングフラグ、単独のブール値/数値、厳密な抽出、 モデル判定、および密な全型付き配列/オブジェクトには json を使用します。フラットな文字列の長文 テキスト/コード/HTML/SVG/Markdown/SQL/テンプレートには xml_field を使用します。長い 散文/コードフィールドと型付きのリスト/オブジェクト/ブール値/数値フィールドが混在し、 リトライ遅延が許容できる場合は明示的な hybrid を使用します。明示的な flat_markdown は レガシーのセクションヘッダー互換性のみに使用します。明示的な yaml_literal は YAML ターゲットドキュメントが意図的に必要な場合にのみ使用します。auto は、 ターゲットモデルが代表的な安定性チェックを通過した後、構造的スキーマ駆動の選択と リトライ遅延が許容できる場合にのみ使用します。qwen2.5:7b のチェックでは、 hybrid がセクションヘッダーを省略したり、古いスキャフォールドのコメントをテキスト フィールドにエコーしたりする可能性があることが確認されています
• TaskDAGValidator に、実行前に重複 ID、欠落した依存関係、循環、サポートされていない必須タスク種別、スキーマバージョンの不一致、安全でない副作用ポリシー、および不明な必須ハンドラーを拒否させてください
• 不明なオプションハンドラーは、必須の意味的出力、必須のダウンストリームノード、承認、または副作用ポリシーに影響しない場合にのみ、プルーニングを許可してください
• 生成された計画データは実行入力または実行状態に保持してください。実行ごとの DAG 状態に共有 TriggerFlow フローデータを使用しないでください
• 小さな宣言的入力配線には ${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(...) を使用します。どちらのローダーも、より大きな設定から 1 つの DAG を選択するための task_dag_key_path="plans.review" をサポートしています。正規化されたグラフをエクスポートするには、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)

プランナーの制約

プランナープラグインとプランナープロンプトは、次のものを宣言する必要があります:

• タスクスキーマのバージョン。通常は task_dag/v1
• 許可されるタスクの種類と利用可能なバインディング名
• 必須のセマンティック出力リーフ
• シナリオに実用的な制限がある場合の最大タスク数
• 再試行可能なモデル出力の失敗となる検証チェック
• 禁止される振る舞い: 不安定なタスク ID、静的なタスク入力に埋め込まれた依存関係の結果、暗黙の副作用、隠れたアクション/スキルの使用、および内部リゾルバータスクの種類

ファサードプランナーは、.output(planner.output_schema(), format="json")、必須の planner.ensure_keys()、および .validate(planner.validate_output) を結線して、構造上の問題が実行前にモデル出力の検証失敗となるようにする必要があります。コンパイル/実行の際にも、グラフは再度検証されます。

レイヤーの所有権:

• TaskDAG / TaskDAGNode のデータ契約と YAML/JSON グラフ設定は agently.types.data に属します
• TaskDAGExecutor、TaskDAGValidator、および TaskDAGResolver はコアに属します
• AgentlyTaskDAGPlanner は、プラグインの関心事としてプランナー出力スキーマ、ensure keys、およびプランナー指示を所有します
• Agently.create_dynamic_task(...) はアプリ向けのファサードエントリポイントです; agent.create_dynamic_task(...) は互換性のみのガイダンスです

アンチパターン

• Dynamic Task のガイダンスを、あたかも TriggerFlow の構文であるかのように TriggerFlow の下に置かないでください
• 送信された DAG データ用にスケジューラーを自作しないでください
• Agently の出力スキーマが形状を所有できる場合に、json.loads(...)、正規表現によるマーカー分割、またはコードブロック抽出でモデルのテキストを解析しないでください
• デフォルトで actions や skills をプランナーに公開しないでください
• 推奨例において、モデルが所有する分類、ドラフト作成、品質チェック、または最終応答生成を、決定論的なローカル代替物の背後に隠さないでください
• Agently.create_dynamic_task(...) で十分な場合に、通常のアプリコードから低レベルの TaskDAGExecutor を呼び出さないでください
• Blocks を TaskDAG の実行に必須であるかのように説明しないでください; 通常の async_run(...) パスは直接的であり、一方 Blocks には明示的な compile_blocks(...) / async_run_blocks(...) が必要です
• risk_check のような曖昧なハンドラー名を使用しないでください; risk_check_handler または呼び出し可能な役割を説明する別の名前を使用してください
• ここで TriggerFlow のドキュメントを重複させないでください; Dynamic Task が内部的に TriggerFlow にコンパイルされることだけを説明してください

次に読む

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