# ADK 2.0の便利だけど知られていない機能 🌍 毎回同じシステムプロンプトやツール定義をLLMに送信するのは、コストもレイテンシも無駄だと感じませんか？ ADK 2.0のコンテキストキャッシュ（ContextCacheConfig）は、繰り返し送信されるコンテキストデータをキャッシュし、LLMの呼び出しコストとレイテンシを削減する機能です。Gemini 2.0以降、Python v1.15.0以降、Java v0.1.0以降で利用可能です。 📌 タイトル：コンテキストキャッシュ (ContextCacheConfig) 🔗 URL： 🧩 概要 ContextCacheConfigは、LLMに送信するコンテキスト（システムプロンプト、ツール定義、会話履歴の固定部分など）をキャッシュすることで、トークン消費を削減します。3つの主要パラメータがあります。min_tokensはキャッシュを有効にするための最小トークン数のしきい値（デフォルト0）、ttl_secondsはキャッシュの有効期限（デフォルト1800秒＝30分）、cache_intervalsはキャッシュの最大再利用回数（デフォルト10回）です。これらをAppオブジェクトに設定することで、自動的にキャッシュが適用されます。 🛠 使い方 ContextCacheConfigを作成し、Appに設定します。 ```python from import App from google.adk.context import ContextCacheConfig cache_config = ContextCacheConfig( min_tokens=1000, # 1000トークン以上でキャッシュ有効 ttl_seconds=3600, # 1時間キャッシュを保持 cache_intervals=20, # 最大20回再利用 ) app = App( agent=my_agent, context_cache_config=cache_config, ) ``` min_tokensを適切に設定することで、小さなコンテキストでは通常送信し、大きなコンテキストのみキャッシュするように制御できます。 🏗 本番システムへの組み込み方 ・大きなシステムプロンプトや多数のツール定義を持つエージェントで特にコスト効果が高い ・ttl_secondsをワークロードのパターンに合わせて調整する（短い会話→短いTTL、長い会話→長いTTL） ・cache_intervalsをリクエスト頻度に応じて設定し、キャッシュの鮮度とコスト削減のバランスを取る ・コスト削減効果をモニタリングし、パラメータを継続的に最適化する 💡 ユースケース 💰 大規模なシステムプロンプトを持つエージェントのAPI呼び出しコストを削減 ⚡ 繰り返しのツール定義送信を省略してレスポンスレイテンシを改善 🔁 高頻度のリクエストが発生するチャットボットでトークン消費を最適化 📋 固定的なコンテキスト（ルール、ガイドライン等）の再送信を効率化 ⚠️ 注意点 Gemini 2.0以降のモデルでのみ利用可能です。キャッシュが有効な間はコンテキストの変更が反映されないため、頻繁にシステムプロンプトを変更する場合はttl_secondsを短く設定してください。また、cache_intervalsを超えると新しいキャッシュが作成されるため、コスト最適化の効果が変動する可能性があります。 ✨ コンテキストキャッシュは、特にコンテキストが大きく頻繁にリクエストされるシナリオで、コストとパフォーマンスの両面で大きな改善をもたらします。 #ADK# #AIAgent#

# ADKの便利で実践的な使い方 ✋ 「本当にこのメールを送信しますか？」— ADKのAction Confirmationsは、不可逆な操作の前にユーザー確認を挟む、安全なエージェント構築のための仕組みです。 📌 タイトル：Action Confirmations — 不可逆操作の事前確認 🔗 URL： 🧩 概要 ADKのAction Confirmationsは、メール送信、データ削除、決済処理など、取り消しが困難な操作の実行前にユーザーの明示的な承認を求める機能です。エージェントが「送信してよいですか？」と確認し、ユーザーが承認した場合のみ実行されます。これにより、自律的なエージェントでありながら、重要な判断ポイントでは人間のコントロールを維持できます。 🛠 使い方 確認付きツールの定義方法です。 確認付きツールを定義するには、`google.adk` から `Agent` と `ToolContext` をインポートします。`send_email(to: str, subject: str, body: str, tool_context: ToolContext)` 関数では、実際の送信前に `tool_context.actions.request_confirmation(message=...)` を呼び出し、宛先・件名・本文のプレビューを含む確認メッセージを表示します。ユーザーが承認した場合のみ、メール送信処理が実行されます。同様に `delete_records(table: str, condition: str, tool_context: ToolContext)` では、削除対象の件数を事前にカウントし、確認メッセージに件数を含めてユーザーに承認を求めます。これらのツールを `Agent` の `name="admin_assistant"`、`model="gemini-2.5-flash"` に `tools` として渡します。 🏗 実践的な使い方 **確認が必要な操作の判断基準**: 以下の操作には確認を付けることを推奨します。 - 外部への送信（メール、メッセージ、API呼び出し） - データの変更・削除 - 課金が発生する操作 - 権限変更やアクセス制御の変更 決済処理の例として `process_payment(amount: float, currency: str, recipient: str, tool_context: ToolContext)` を定義します。この関数では `tool_context.actions.request_confirmation(message=...)` で送金先・通貨・金額を含む確認メッセージを表示し、ユーザーの承認後に `payment_gateway.charge(amount=..., currency=..., recipient=...)` を実行して決済を処理します。 **確認メッセージの設計**: 確認メッセージには、操作の対象・影響範囲・不可逆性を明確に記載します。ユーザーが判断に必要な情報を過不足なく提供しましょう。 **段階的な確認**: 複数のステップがある場合、各ステップで確認を取るか、最終ステップでまとめて確認するかを設計します。ユーザー体験とのバランスを考慮してください。 💡 ユースケース 📧 メール・メッセージの送信確認 🗑️ データベースレコードの削除確認 💳 決済・送金の実行確認 🔐 権限変更・アクセス制御の変更確認 ⚠️ 注意点 - 確認が多すぎるとユーザー体験が悪化します。本当に不可逆な操作に絞って確認を設定してください。 - 確認メッセージが不十分だと、ユーザーが適切な判断を下せません。操作の影響を具体的に記載しましょう。 - バッチ処理や自動化パイプラインでは確認がボトルネックになります。自動化が必要な場面では確認をスキップする設計も検討してください。 ✨ Action Confirmationsを適切に設定することで、エージェントの自律性と人間の安全管理を両立できます。「取り返しのつかない操作」にだけ確認を入れるのがコツです！ #ADK# #AIAgent#

# ADK 2.0の便利だけど知られていない機能 🌍 エージェントが「先週の会話の内容」を覚えていてくれたら、もっと自然なやり取りができると思いませんか？ ADK 2.0のメモリ機能は、セッションを超えた長期的な知識をエージェントに提供するための仕組みです。過去の会話や学習した情報を記憶し、必要に応じて呼び出すことができます。 📌 タイトル：メモリ 🔗 URL： 🧩 概要 メモリはStateとは異なり、セッション横断の長期的な知識を管理します。3つのメモリサービス実装が用意されています。InMemoryMemoryServiceは開発・テスト用のインメモリ実装、VertexAiMemoryBankServiceはセマンティック検索ベースの本番向け実装、VertexAiRagMemoryServiceはベクトルベースのRAG実装です。メモリの取得にはPreloadMemory（セッション開始時に自動ロード）とLoadMemory（オンデマンドでロード）の2つのビルトインツールが用意されており、プログラムからはtool_context.search_memory()でアクセスできます。 🛠 使い方 メモリサービスを設定し、エージェントにメモリツールを組み込みます。 ```python from google.adk.memory import InMemoryMemoryService from import PreloadMemory, LoadMemory # 開発用：インメモリ実装 memory_service = InMemoryMemoryService() # エージェントにメモリツールを追加 agent = Agent( name="assistant", tools=[PreloadMemory(), LoadMemory()], ... ) # Runnerにメモリサービスを設定 runner = Runner( agent=agent, memory_service=memory_service, ... ) ``` ツール内からプログラム的にメモリを検索する場合は以下のようにします。 ```python def my_tool(query: str, tool_context: ToolContext) -> str: results = tool_context.search_memory(query="過去の会話") return str(results) ``` 複数のメモリサービスを組み合わせる場合は、カスタムツールを作成して統合できます。 🏗 本番システムへの組み込み方 ・開発時はInMemoryMemoryServiceで素早くプロトタイプし、本番ではVertexAI系に切り替える ・PreloadMemoryで頻繁に必要な情報を自動ロードし、レスポンス品質を向上させる ・メモリに保存するデータの範囲を適切に設計し、不要なデータの蓄積を防ぐ ・カスタムツールで複数のメモリソースを統合し、包括的な知識ベースを構築する 💡 ユースケース 🧠 過去の会話履歴をもとに、ユーザーの好みに合わせた応答を生成 📚 プロジェクトの過去の議論や決定事項を長期記憶として保持 🔍 セマンティック検索で関連する過去のやり取りを自動的に取得 🤝 複数のエージェント間で共有知識ベースとしてメモリを活用 ⚠️ 注意点 InMemoryMemoryServiceはプロセス終了時にデータが失われるため、本番環境では使用しないでください。VertexAI系のサービスはGCPのセットアップが必要です。また、メモリに保存されるデータ量が増えると検索のレイテンシに影響するため、適切なデータ管理戦略を検討してください。 ✨ メモリ機能により、エージェントはセッションを超えた文脈を持つことができます。長期的なユーザー体験の向上に大きく貢献する機能です。 #ADK# #AIAgent#

# ADKの便利で実践的な使い方 🚀 ツールの並列実行、レスポンスサイズの最適化、レイテンシとトークンコストの削減 — ADKのTool Performanceガイドで、本番環境のスループットを最大化しましょう。 📌 タイトル：Tool Performance — ツール実行の最適化手法 🔗 URL： 🧩 概要 ADKではツールのパフォーマンスを最適化するための複数のアプローチが提供されています。読み取り専用ツールの並列実行、レスポンスサイズの削減によるトークンコスト最適化、そしてレイテンシ削減のためのベストプラクティスがあります。本番環境で高いスループットを実現するためには、これらの最適化が不可欠です。 🛠 使い方 並列実行とレスポンス最適化の例です。 まず、読み取り専用のツールを複数定義します。`get_user_profile(user_id: str)`、`get_user_orders(user_id: str)`、`get_user_preferences(user_id: str)` のように副作用のない関数を用意すると、ADKはこれらを並列に実行できます。次に、レスポンスサイズを削減するツール設計として `search_products(query: str, limit: int = 5)` を定義し、戻り値では `id`、`name`、`price` など必要なフィールドのみを返し、画像URLや説明文全文、メタデータ等は省略します。最後に `Agent` を `name="customer_service"`、`model="gemini-2.5-flash"` で作成し、`tools` リストにこれら4つの関数を渡します。 🏗 実践的な使い方 **並列実行の判断基準**: 副作用がなく、互いに依存しないツールは並列実行が安全です。LLMは複数のツールを同時に呼び出すことができ、ADKがそれらを並列に実行します。データの取得系ツール（GET相当）は並列実行の良い候補です。 並列実行に適した設計として、互いに独立した `get_weather(city: str)`、`get_news(topic: str)`、`get_stock_price(symbol: str)` のようなツールを定義します。各ツールはそれぞれ天気情報、ニュース記事、株価をdictで返すだけの読み取り専用関数です。LLMがこれら3つを同時に呼び出すと、ADKが自動的に並列で実行します。 **レスポンスサイズの最適化**: ツールの戻り値はLLMのコンテキストウィンドウを消費します。不要なフィールドの除外、データの要約、ページネーションの実装でトークンコストを大幅に削減できます。 **レイテンシ最適化のチェックリスト**: 1. キャッシュ可能な結果はキャッシュする 2. 外部API呼び出しにタイムアウトを設定する 3. 不要に大きなデータを返さない 4. 複数の小さなツールに分割して並列実行を促す 💡 ユースケース 📊 ダッシュボード用の複数データソース並列取得 🔍 検索結果のフィールド絞り込みによるトークン節約 ⚡ マイクロサービス間の並列API呼び出し 💰 大量リクエスト環境でのトークンコスト最適化 ⚠️ 注意点 - 副作用のあるツール（書き込み・削除）を並列実行すると、競合状態が発生する可能性があります。書き込み系ツールは逐次実行を推奨します。 - レスポンスを削減しすぎると、LLMが十分な情報を得られず、ユーザーへの回答品質が低下する場合があります。必要な情報は確実に含めてください。 - ツールのタイムアウトが短すぎると、正常なレスポンスも中断される可能性があります。適切なタイムアウト値を設定しましょう。 ✨ 小さな最適化の積み重ねが、本番環境での大きなパフォーマンス差を生みます。まずはレスポンスサイズの見直しから始めてみましょう！ #ADK# #AIAgent#

# ADK 2.0の便利だけど知られていない機能 🌍 エージェントの会話中に「前に聞いた情報を覚えておいてほしい」と思ったことはありませんか？状態管理の仕組みを理解すれば、それが簡単に実現できます。 ADK 2.0のState機能は、セッション内外でデータを保持・共有するためのキーバリュー型のスクラッチパッドです。プレフィックスによってスコープを使い分けることで、柔軟な状態管理が可能になります。 📌 タイトル：状態 (State) の管理 🔗 URL： 🧩 概要 Stateはキーバリュー形式のデータストアで、4つのプレフィックスによってスコープが決まります。プレフィックスなしはセッションスコープ（そのセッション内でのみ有効）、user:はユーザースコープ（同一ユーザーの複数セッションで共有）、app:はアプリケーションスコープ（全ユーザー・全セッションで共有）、temp:は一時スコープ（インボケーション終了時に破棄）です。エージェントの指示文中では{key}の形式で状態値を参照でき、動的なプロンプト構築が可能です。 🛠 使い方 状態の書き込みにはいくつかの方法があります。 ```python # 1. output_keyでエージェントの出力を自動保存 agent = Agent( name="summarizer", output_key="last_summary", ... ) # 2. EventActions.state_deltaで明示的に設定 from import EventActions actions = EventActions(state_delta={"user:preference": "dark_mode"}) # 3. ToolContext経由でツール内から設定 def my_tool(query: str, tool_context: ToolContext) -> str: tool_context.state["app:global_counter"] = 42 tool_context.state["temp:intermediate"] = "temporary_value" return "done" ``` 指示文での参照は以下のように行います。 ```python agent = Agent( instruction="ユーザーの好みは{user:preference}です。前回の要約：{last_summary}", ... ) ``` 🏗 本番システムへの組み込み方 ・スコープを適切に選択し、不要なデータの永続化を避ける（一時データにはtemp:を活用） ・user:やapp:スコープの状態は複数セッションに影響するため、慎重に設計する ・状態の読み書きは必ずCallbackContextやToolContext経由で行い、イベント追跡を確保する ・状態キーの命名規則を統一し、チーム全体での保守性を向上させる 💡 ユースケース 👤 user:プレフィックスでユーザーの好みや設定を複数セッションにわたって保持 📊 app:プレフィックスでアプリケーション全体の統計情報やカウンターを管理 🔄 output_keyで直前のエージェント出力を次のステップで自動参照 🧹 temp:プレフィックスで中間計算結果を一時保存し、メモリ効率を向上 ⚠️ 注意点 session.stateをコンテキスト外から直接変更しないでください。CallbackContextやToolContextを経由せずに変更すると、イベントトラッキングがバイパスされ、状態の変更履歴が記録されません。これにより、巻き戻し機能やデバッグに支障をきたす可能性があります。 ✨ Stateの4つのスコープを使い分けることで、エージェントの記憶と文脈を柔軟に管理できます。適切な状態管理は、質の高いエージェント体験の基盤です。 #ADK# #AIAgent#

# ADKの便利で実践的な使い方 ⚡ Python関数をそのままツールに、エージェントもツールに、そして長時間タスクもノンブロッキングで実行 — ADKのFunction Toolsは、ツール定義の柔軟性を極限まで高めます。 📌 タイトル：Function Tools — 関数・エージェント・非同期タスクをツール化 🔗 URL： 🧩 概要 ADKのFunction Toolsでは、Python/TypeScriptの関数をそのままエージェントのツールとして利用できます。さらに、AgentToolを使えばエージェント自体をツールとして別のエージェントに提供できます。Long Running Function Toolsは、動画エンコードやバッチ処理のような長時間タスクをブロッキングせずに実行するための仕組みです。 🛠 使い方 基本的な関数ツールの定義とAgentToolの活用例です。 `google.adk`から`Agent`と`AgentTool`をインポートします。シンプルな関数ツールとして`calculate_price`を定義し、`base_price`（float）、`quantity`（int）、`discount_percent`（float、デフォルト0）を受け取り、合計金額を計算してdictで返します。 エージェントのツール化には`AgentTool`を使います。まず`analysis_agent`を`name="data_analyst"`、`tools=[query_database]`で定義し、次に`main_agent`の`tools`リストに`calculate_price`関数と`AgentTool(agent=analysis_agent)`を含めます。これにより、メインエージェントは価格計算関数とデータ分析エージェントの両方をツールとして呼び出せます。 Long Running Function Toolsの使い方です。 `google.adk`から`LongRunningFunctionTool`をインポートします。非同期関数`encode_video`は`video_url`（str）と`format`（str、デフォルト"mp4"）を受け取り、`start_encoding_job`でエンコードジョブを開始してジョブIDを返します。この関数を`LongRunningFunctionTool(func=encode_video)`でラップして`video_tool`を作成し、`Agent`の`tools=[video_tool]`に渡すことで、エージェントがブロッキングなしで長時間処理を実行できます。 🏗 実践的な使い方 **AgentToolによるモジュール化**: 複雑な処理を専門エージェントとしてカプセル化し、AgentToolで公開することで、メインエージェントのinstructionをシンプルに保てます。専門エージェントは独自のツールやプロンプトを持てるため、関心の分離が実現します。 `summarizer`（3行要約）と`translator`（英語翻訳）をそれぞれ`Agent`で定義し、メインの`content_manager`エージェントの`tools`リストに`AgentTool(agent=summarizer)`と`AgentTool(agent=translator)`として登録します。これにより、メインエージェントは要約と翻訳の専門エージェントをツールとして呼び出し、コンテンツ管理の依頼を処理できます。 **Long Running Toolsの活用場面**: バッチ処理、外部APIのポーリング待ち、ファイル変換など、完了まで数秒〜数分かかる処理に最適です。エージェントはジョブIDを受け取り、他のタスクを並行して進められます。 **型アノテーションの重要性**: 関数のパラメータ型と戻り値型を明確に定義することで、LLMが正確にツールを呼び出せます。docstringもツールの説明として使われるため、簡潔で明確に書きましょう。 💡 ユースケース 🧮 計算・変換関数のツール化（価格計算、単位変換） 🤖 専門エージェントのAgentTool化による再利用 🎬 動画エンコード・画像処理の非同期実行 📊 バッチデータ処理のノンブロッキング実行 ⚠️ 注意点 - 関数のdocstringがツールの説明になるため、LLMが理解しやすい説明を書いてください。docstringがないとツールの用途が不明確になります。 - AgentToolで呼び出されたエージェントは、親エージェントのコンテキストとは別のセッションで動作します。状態の共有には注意が必要です。 - Long Running Function Toolsは完了通知の仕組みを別途実装する必要があります。ポーリングや Webhook での通知を検討してください。 ✨ Function Toolsを使えば、既存のコード資産をそのままエージェントに統合でき、AgentToolでエージェントの再利用も自在。開発効率を大幅に向上させましょう！ #ADK# #AIAgent#

# ADK 2.0の便利だけど知られていない機能 🌍 実行中のエージェントを途中で安全にキャンセルし、それまでの処理結果は保持できたら理想的だと思いませんか？ ADK 2.0のキャンセル機能は、AbortController/AbortSignalを使ってエージェントの実行をグレースフルに中断する仕組みです。キャンセルシグナルはRunner、LlmAgent、Models、Toolsへと伝播し、コミット済みのイベントは保持されます。 📌 タイトル：エージェント実行のキャンセル 🔗 URL： 🧩 概要 エージェント実行のキャンセルは、AbortControllerとAbortSignalのパターンで実現されます。AbortControllerからシグナルを発行すると、Runner→LlmAgent→Models→Toolsの順にキャンセルが伝播します。重要な点として、キャンセル時点ですでにコミットされたイベントは保持され、例外をスローせずグレースフルに完了します。AbortSignal.timeout()を使えばタイムアウトベースの自動キャンセルも可能で、AbortSignal.any()で複数のシグナルを組み合わせることもできます。 🛠 使い方 AbortControllerを作成し、そのシグナルをRunnerに渡します。 ```python from adk import App, AbortController, AbortSignal app = App(agent=my_agent) # 手動キャンセル controller = AbortController() task = "long task", signal=controller.signal) # 必要なタイミングでキャンセル controller.abort() # タイムアウトベースの自動キャンセル（2秒） signal = AbortSignal.timeout(2000) result = await "time-limited task", signal=signal) # 複数シグナルの組み合わせ combined = AbortSignal.any([ AbortSignal.timeout(5000), user_cancel_signal ]) result = await "task", signal=combined) ``` 🏗 本番システムへの組み込み方 ・ユーザー操作（キャンセルボタン）と連動したAbortControllerを実装し、UIからのキャンセルを可能にする ・AbortSignal.timeout()でAPI呼び出しの最大実行時間を制限し、リソースの無駄遣いを防ぐ ・AbortSignal.any()でユーザーキャンセルとタイムアウトの両方に対応する ・キャンセル後のコミット済みイベントを活用し、部分的な結果を返却する設計にする 💡 ユースケース ⏱ LLM呼び出しにタイムアウトを設定し、応答が遅い場合に自動キャンセル 🖱 ユーザーがUIのキャンセルボタンを押した際にエージェント実行を即座に中断 🔀 複数のエージェントを並列実行し、最初に完了したもの以外をキャンセル 💰 コスト上限に達した時点でLLM呼び出しを打ち切る ⚠️ 注意点 キャンセルはグレースフルに完了するため、abort()を呼んだ瞬間に即座に停止するわけではありません。現在実行中のオペレーション（LLM推論やツール実行）が完了するまで若干の遅延が生じる場合があります。また、コミット済みのイベントは保持されるため、キャンセル後の状態が中途半端にならないよう、ワークフロー設計時にキャンセルポイントを意識することを推奨します。ツール内部でシグナルを監視して早期リターンする実装も有効です。 ✨ キャンセル機能により、エージェント実行のライフサイクルを安全にコントロールできます。タイムアウトと組み合わせることで、本番環境での予測可能性が大きく向上します。 #ADK# #AIAgent#

# ADK 2.0の便利だけど知られていない機能 🌍 AIエージェントのワークフローを「グラフ」として設計できたら、複雑な処理フローも見通しよく管理できると思いませんか？ ADK 2.0のWorkflowクラスは、ノードとエッジでエージェントの実行パスを定義するグラフベースのワークフロー構築機能です。AIエージェント、カスタム関数、ツール、さらにはネストされたWorkflowをノードとして組み合わせ、明示的なルーティングロジックで制御できます。 📌 タイトル：Workflow クラスによるグラフベース実行 🔗 URL： 🧩 概要 Workflowクラスは、ADK 2.0におけるグラフベースのエージェントワークフローを構築するための主要な構成要素です。ノードには、AIエージェント・コード関数・ツール・ネストされたWorkflowを配置でき、エッジ（edges）パラメータで実行パスを定義します。逐次実行は `("START", node1, node2, node3)` のようにタプルで記述し、条件分岐は辞書で定義します。これにより、従来のプロンプトベースの制御では難しかった複雑な分岐構造を、明示的かつ信頼性高く実装できます。 🛠 使い方 Workflowクラスのインスタンスを作成し、nodesにノードのリスト、edgesに実行パスを指定します。 ```python from adk import Workflow, Agent agent_a = Agent(name="researcher", ...) agent_b = Agent(name="writer", ...) workflow = Workflow( name="content_pipeline", nodes=[agent_a, agent_b], edges=[("START", agent_a, agent_b)] ) ``` 逐次実行の場合はタプルでノードを列挙し、条件分岐が必要な場合は辞書ベースのルーティングを使います。ネストされたWorkflowをノードとして組み込むことで、大規模なパイプラインも階層的に管理できます。 🏗 本番システムへの組み込み方 ・各ノードの責務を明確に分離し、テスト可能な単位で設計する ・エラーハンドリングを各ノードレベルで実装し、障害の影響範囲を限定する ・ネストされたWorkflowを活用して、再利用可能なサブパイプラインを構築する ・条件分岐のルーティングロジックをドキュメント化し、チームでの保守性を確保する 💡 ユースケース 📄 論文の収集→要約→レビュー→投稿を一連のグラフとして管理 🔀 入力データの種類に応じて異なる処理パイプラインに分岐 🏢 複数部門のエージェントを組み合わせた承認フローの構築 🔁 サブワークフローを再利用した複数プロジェクト横断の自動化 ⚠️ 注意点 WorkflowクラスはLive Streamingとの互換性がなく、一部のサードパーティ統合にも対応していません。リアルタイムのストリーミング応答が必要なケースでは、別のアプローチを検討する必要があります。また、グラフの複雑さが増すとデバッグが難しくなるため、適切な粒度でノードを分割することが重要です。 ✨ 明示的なグラフ定義により、エージェントワークフローの信頼性と保守性が大きく向上します。複雑な処理フローを構築する際にはぜひ活用してみてください。 #ADK# #AIAgent#

# ADK 2.0の便利だけど知られていない機能 🌍 長時間かかるエージェントワークフローが途中で失敗したとき、最初からやり直さずに続きから再開できたら便利だと思いませんか？ ADK 2.0のResumabilityConfigは、ワークフローの実行状態をイベントログとして記録し、失敗時にinvocation_idを指定して途中から再開できる機能です。 📌 タイトル：ワークフローの再開 (ResumabilityConfig) 🔗 URL： 🧩 概要 ResumabilityConfig(is_resumable=True)をAppに設定すると、ワークフロー実行中に完了したタスクがイベントとして記録されます。失敗時にはinvocation_idを指定して再開でき、組み込みエージェントはそれぞれの状態を自動的に復元します。SequentialAgentはcurrent_sub_agentから再開し、LoopAgentはtimes_loopedの値を保持して残りの反復を実行し、ParallelAgentは未完了のサブエージェントのみを実行します。これにより、大規模なパイプラインで一部のステップが失敗しても、完了済みのステップを再実行する無駄を省けます。 🛠 使い方 AppにResumabilityConfigを設定し、再開時にinvocation_idを渡します。 ```python from adk import App, ResumabilityConfig app = App( agent=my_workflow, resumability_config=ResumabilityConfig(is_resumable=True) ) # 初回実行 result = await "process data") invocation_id = result.invocation_id # 失敗後の再開（同じinvocation_idを指定） resumed_result = await input="process data", invocation_id=invocation_id ) ``` カスタムエージェントで再開に対応する場合は、BaseAgentStateを拡張して独自の状態を保存します。 🏗 本番システムへの組み込み方 ・invocation_idをデータベースやメッセージキューに保存し、リトライ時に参照できるようにする ・ツールの冪等性を担保する（ツールは少なくとも1回実行され、再開時に再実行される可能性がある） ・カスタムエージェントではBaseAgentStateを拡張し、再開に必要な中間状態を明示的に定義する ・長時間ワークフローでは定期的にチェックポイントとなるステップを設け、再開の粒度を細かくする 💡 ユースケース 📊 数十ステップの大規模データ処理パイプラインで、途中失敗からの効率的な復旧 💰 外部API呼び出しを含むワークフローで、API課金の無駄な再実行を回避 🔄 不安定なネットワーク環境での長時間エージェント実行の信頼性向上 🏭 バッチ処理ジョブで一部アイテムの処理失敗時に残りを継続処理 ⚠️ 注意点 ツールは再開時に少なくとも1回実行されるため、副作用を持つツール（データベース書き込み、外部API呼び出しなど）は必ず冪等に設計してください。同じ入力で複数回実行しても結果が変わらないことを保証する必要があります。また、ParallelAgentの再開では完了判定がサブエージェント単位であるため、サブエージェント内部の途中状態は保持されない点に留意してください。 ✨ ResumabilityConfigにより、長時間ワークフローの運用が劇的に楽になります。特にコストのかかるステップを含むパイプラインでは、導入効果が大きいです。 #ADK# #AIAgent#

# ADK 2.0の便利だけど知られていない機能 🌍 実行時にどのエージェントを呼び出すかを動的に切り替えられたら、柔軟で堅牢なシステムが作れると思いませんか？ ADK 2.0のエージェントルーティングは、router関数を使って実行時に動的にエージェントを選択する仕組みです。設定ベースのルーティング、エラー時のフォールバック、複雑度に応じた自動振り分けなど、多彩なパターンに対応できます。 📌 タイトル：エージェントルーティング 🔗 URL： 🧩 概要 エージェントルーティングでは、router関数がagents（利用可能なエージェントのリスト）、context（現在のコンテキスト）、errorContext（直前のエージェントがイベントをyieldする前に例外を投げた場合のエラー情報）を受け取り、次に実行するエージェントを返します。あるエージェントがイベントをyieldする前に失敗した場合、routerはerrorContextを受け取るため、別のエージェントへのフォールバックが可能です。これはRoutedLlm（モデルレベルのルーティング）とは異なり、エージェント全体を切り替える点が特徴です。 🛠 使い方 router関数を定義し、エージェントに設定します。 ```python from adk import Agent def my_router(agents, context, error_context=None): # エラー時はフォールバックエージェントを選択 if error_context: return next(a for a in agents if == "fallback") # コンテキストに応じて動的にエージェントを選択 complexity = context.session.state.get("task_complexity", "low") if complexity == "high": return next(a for a in agents if == "expert") return next(a for a in agents if == "basic") parent = Agent( name="dispatcher", sub_agents=[basic_agent, expert_agent, fallback_agent], router=my_router ) ``` router関数の戻り値で次に実行されるエージェントが決まるため、任意のロジック（設定ファイル参照、外部API呼び出しなど）を組み込めます。 🏗 本番システムへの組み込み方 ・設定ベースのルーティングでは、ルーティングルールを外部設定ファイルに切り出し、再デプロイなしで変更可能にする ・errorContextを活用したフォールバックチェーンを設計し、単一障害点を排除する ・ルーティング判定のログを構造化して出力し、どのエージェントが選ばれたか追跡可能にする ・複雑度の判定ロジック自体をテスト可能な純粋関数として分離する 💡 ユースケース ⚙️ 設定ファイルでルーティングルールを管理し、運用中に振り分け先を変更 🛡 プライマリエージェント失敗時に自動でフォールバックエージェントへ切り替え 🧠 タスクの複雑度を事前判定し、軽量/高機能エージェントを自動選択 📋 プランニングモードと実行モードでエージェントを切り替える段階的処理 ⚠️ 注意点 router関数内での重い処理（外部API呼び出し等）はレイテンシに直結するため、最小限に抑えることを推奨します。また、RoutedLlm（モデルルーティング）との混同に注意してください。RoutedLlmはLLMモデル自体を切り替えるもので、エージェントルーティングはエージェント全体（プロンプト・ツール・サブエージェント含む）を切り替えるものです。errorContextはエージェントがイベントをyieldする前に失敗した場合のみ提供される点も理解しておく必要があります。 ✨ 動的ルーティングにより、単一のワークフロー定義で多様な状況に対応できる柔軟なシステムを構築できます。まずはフォールバックパターンから導入してみてください。 #ADK# #AIAgent#