こんにちは、先日開催された社内イベント「ごーとんカップ 2025」にてセキュリティチャンピオンになりました、ソフトウェアエンジニアの渡邉(匠)です。「カミナシ 設備保全」の開発に携わっています。
最近、AIエージェントの技術トレンドに乗り遅れないよう、Amazon Bedrock を使ったプロトタイピングに取り組んでいます。その中で、Amazon Bedrock Agents で提供されている「Return of Control」という機能をStrands AgentsとAmazon Bedrock AgentCoreを使って作ってみたので、その実装方法と学びを共有します。
Return of Control とは
Return of Control は、Amazon Bedrock Agentsの機能の一つです。
この機能を簡単にまとめると以下のような特徴があります:
- エージェントが実行すべき操作（アクション）をアクショングループとして定義でき、その実行をアプリケーション側に委ねることができる機能
- アクションの例: 天気情報の取得、データベースへのアクセス、外部APIの呼び出し、人による承認など
- アプリケーションレベルでビジネスロジックを実装する選択肢を提供
- エージェントはアクションの実行が必要になった場合、処理を一時的に中断し、制御（必要なパラメーターを含む）をアプリケーション側に返す
- アプリケーション側はアクションを実行し、その結果をエージェントに返却。エージェントは受け取った結果をもとに処理を再開する
- 利用シーン例:
- 既存のAPIを活用したい場合（アクションとして呼び出すLambda関数を新しく作る必要がない）
- 時間のかかる処理を非同期で実行したい場合
- 人による承認が必要な処理
- 具体例: 「ユーザー情報を取得したい」とエージェントが判断した場合、処理を中断し、アプリケーションに「ユーザー情報を取得するアクション」のパラメータを返却。アプリケーションはAPIを呼び出し、その結果をエージェントに返す
詳細は公式ドキュメントを参照してください。
全体の処理フロー
Return of Control を使った処理の全体像は以下のようになります。
アプリケーション側の処理の流れ:
- Amazon Bedrock Agentsにユーザー入力を送信
- Return of Controlで制御が返却されたら、指定されたアクションとパラメータに基づいて処理を実行
- 例: ユーザーIDをパラメータとして受け取り、データベースからユーザー情報を取得
- 実行結果をエージェントに送信して推論を再開
- 最終回答を受け取って処理完了
なぜ Return of Control が嬉しいのか
エージェント機能を組み込む際、Return of Controlを使うことで以下のようなメリットがあります。
- 認証情報の一元管理: エージェント側に認証情報を持たせる必要がなく、アプリケーション側で一元管理できる
- 例: ユーザーごとに異なるAPIキーやアクセストークンをセッション情報から取得して使用できる
- 既存ロジックの再利用: 既存のバックエンドのロジックをそのまま活用できる
- 例: すでに実装済みのデータ取得・加工処理をエージェント用に書き直す必要がない
- デバッグのしやすさ: アプリケーションと同じ環境で実行されるため、ログ確認やデバッグが容易
Strands Agents で実装してみる
Strands Agentsは数行でAIエージェントを作成でき、Bedrockとも簡単に統合できる優れたSDKです。
今回はAIエージェントの学習も兼ねて、Strands AgentsとAmazon Bedrock AgentCoreを使ってAmazon Bedrock AgentsのReturn of Controlに似た機能を実装してみました。
Return of Control 機能の対応状況
Strands AgentsではAmazon Bedrock AgentsのReturn of Control機能は現在公式サポートされていませんが、ロードマップには含まれており開発中です。
開発は進んでおり、Amazon Bedrock Agentsの「アクション」に相当するStrands Agentsの「ツール」とツールの処理に割り込んでアプリケーションに制御を返すinterrupts 機能の実装が完了しています。
実装方法の変遷
Return of Controlの実装を試みる中で、より良い方法を見つけたため、その学びのプロセスも共有します。
最初のアプローチ: state を使った「お祈り」実装
ツールのinterruptsの存在を知る前、最初はstateを使った回避策を試していました。
stateとはエージェントが任意の情報をキーバリュー形式で保持できる仕組みです。
実装の流れ:
- ツール内でエージェントの
stateに実行したいツールの情報を記録 - 「推論を中断してください」というメッセージを返して祈る🙏
- エージェント実行後、
stateからツール情報を取得して呼び出し元に返却 - 呼び出し元がツールを実行し、結果と共にエージェントを再呼び出し
課題:
LLMが本当に中断してくれるかは「祈る」しかないです…
@tool(context=True) def get_goaton_info(context: ToolContext) -> str: """ごーとんの情報を取得の依頼ができるツールです。""" actions = context.agent.state.get("actions") or [] actions.append({"name": "get_goaton_info"}) context.agent.state.set("actions", actions) return "ごーとんの情報取得をプランに追加しました。推論を中断して、ツールの実行結果を受け取ってから再開してください。"
改善版: interrupts を使った実装
最近追加されたツールのinterruptsにより、真の意味での割り込みが可能になりました。祈る必要はありません！
@tool(context=True) def get_goaton_info(tool_context: ToolContext) -> str: """ごーとんの情報を取得できるツールです。""" return tool_context.interrupt( name="get_goaton_info", reason={"action": "get_goaton_info"}, )
ツール内でinterruptを呼ぶとエージェントの実行が止まり、制御が呼び出し元に返されます。
参考:
詳細な実装
セッション管理の実装
Return of Controlでは制御を一旦アプリケーション側に返し、再度エージェントを呼び出す必要があります。
この際、エージェントの会話履歴や状態を保持するために、Strands AgentsのSession Management 機能とBedrock AgentCore Memoryを組み合わせて使用します。
- Session Management: セッション管理を担当（コンテキストとエージェントの状態を保持）
- Bedrock AgentCore Memory: 会話履歴や状態の実際の保存先
Bedrock AgentCore Memoryは事前にAWSコンソールまたはCLIでMemoryリソースを作成する必要があります。
aws bedrock-agentcore-control create-memory \\ --name "goaton_agent_memory" \\ --event-expiry-duration 7
作成後、レスポンスからmemory.idを取得してコードで使用します。
参考:
ツールとエージェントの実装
ツールの定義
interruptsを使ったツールの実装:
from strands import ToolContext, tool @tool(context=True) def get_goaton_info(tool_context: ToolContext) -> str: """ごーとんの情報を取得できるツールです。""" return tool_context.interrupt( name="get_goaton_info", reason={"action": "get_goaton_info"}, )
エージェントの作成
import uuid from bedrock_agentcore.memory.integrations.strands.config import AgentCoreMemoryConfig from bedrock_agentcore.memory.integrations.strands.session_manager import AgentCoreMemorySessionManager from strands import Agent from strands.models import BedrockModel def create_agent(session_id: str) -> Agent: """エージェントを作成""" agentcore_memory_config = AgentCoreMemoryConfig( memory_id="事前に作成したMemoryのID", session_id=session_id, actor_id="GoatonAgent", ) session_manager = AgentCoreMemorySessionManager( region_name="ap-northeast-1", agentcore_memory_config=agentcore_memory_config, ) model = BedrockModel( region_name="ap-northeast-1", model_id="jp.anthropic.claude-haiku-4-5-20251001-v1:0", streaming=True, ) agent = Agent( name="GoatonAgent", model=model, session_manager=session_manager, tools=[get_goaton_info], ) return agent
エントリーポイントの実装
interrupt の検出と返却
interruptするとエージェントが返却した結果のstop_reasonにinterruptが入り、interruptsにその詳細が格納されます。
from bedrock_agentcore.runtime import BedrockAgentCoreApp from strands import AgentResult app = BedrockAgentCoreApp() @app.entrypoint def invoke(payload): session_id = str(payload.get("session_id", "")) or str(uuid.uuid4()) prompt = str(payload.get("prompt", "")) agent = create_agent(session_id) result = agent(prompt) if result.stop_reason == "interrupt": actions = [ {"interrupt_id": interrupt.id, "action": interrupt.reason} for interrupt in result.interrupts ] return { "type": "action", "actions": actions, "session_id": session_id, } return { "type": "response", "message": str(result), "session_id": session_id }
interrupt からの再開
ツール実行後、interrupt_idと実行結果をペアでエージェントに渡すことで、割り込んだ途中から推論を再開できます。
@app.entrypoint def invoke(payload): session_id = str(payload.get("session_id", "")) or str(uuid.uuid4()) prompt = str(payload.get("prompt", "")) results: list[dict[str, str]] = payload.get("interrupt_results", []) agent = create_agent(session_id) result: AgentResult if len(results) > 0: interrupt_results = [ { "interruptResponse": { "interruptId": res["interrupt_id"], "response": res["response"] } } for res in results ] result = agent(interrupt_results) else: result = agent(prompt) if result.stop_reason == "interrupt": actions = [ {"interrupt_id": interrupt.id, "action": interrupt.reason} for interrupt in result.interrupts ] return { "type": "action", "actions": actions, "session_id": session_id, } return { "type": "response", "message": str(result), "session_id": session_id }
注意: 本記事では基本的な実装に焦点を当てており、エラーハンドリングや例外処理については省略しています。
まとめ
Strands Agent の interrupts 機能を使うことで、Amazon Bedrock Agents の Return of Control に近い機能を実装できました。
Strands Agent では Return of Control 機能が公式にサポートされる予定です（Issue #882）。正式リリース後は、より洗練された実装が可能になると思います。
今回の実装を通じて、AIエージェント技術における制御フローの設計パターンを学ぶことができました。
今後も AIエージェント技術の動向を追いながら、実際のプロダクトへの応用可能性を探っていきたいと思います。
