# ADK 2.0の便利だけど知られていない機能 🌍 ワークフローの構造を実行時に動的に決めたい — そんな要求に応えるのが動的ワークフローです。 ADK 2.0の動的ワークフロー機能では、`@node`デコレータと` 📌 タイトル：動的ワークフロー 🔗 URL： 🧩 概要 動的ワークフローは、`@node`デコレータで関数をノードとしてマークし、` 🛠 使い方 `@node`デコレータで関数をラップし、` ```python from adk import node, WorkflowContext import asyncio @node async def process_items(ctx: WorkflowContext): items = ctx.input.output["items"] # ループ処理 for item in items: await input=item) # 並列実行 tasks = [ input=item) for item in items] results = await asyncio.gather(*tasks) return Event(output={"results": results}) @node(rerun_on_resume=True) async def orchestrator(ctx: WorkflowContext): # このノードは再開時にも再実行される data = await processed = await input=data) return Event(output=processed) ``` RequestInputと組み合わせれば、動的ワークフロー内でもHuman-in-the-Loopを実現できます。 🏗 本番システムへの組み込み方 ・チェックポイントを活用し、長時間実行ワークフローの耐障害性を確保する ・`rerun_on_resume=True`は状態に依存する親ノードにのみ適用し、冪等性を意識する ・並列実行数を制御し、外部APIのレートリミットに配慮する ・再帰の深さに上限を設け、無限ループを防止する 💡 ユースケース 🔁 可変長のデータリストに対するループ処理（件数が事前に分からないケース） ⚡ 複数の独立したタスクを`asyncio.gather()`で並列実行 🌳 ツリー構造のデータを再帰的に処理するワークフロー 🔄 処理結果に応じて次のステップを動的に決定するアダプティブパイプライン ⚠️ 注意点 動的ワークフローは柔軟性が高い反面、静的なグラフ定義に比べてフローの全体像が把握しにくくなります。再帰やループの終了条件を明示的に定義し、無限ループに陥らないよう注意してください。また、`asyncio.gather()`での並列実行時は、各タスクのエラーハンドリングを個別に行う必要があります。 ✨ 静的なグラフでは表現できない複雑なロジックを、Pythonの自然な制御構造で記述できるのが動的ワークフローの強みです。チェックポイント機能と組み合わせて、堅牢な本番ワークフローを構築しましょう。 #ADK# #AIAgent#