Amazon Bedrock AgentCore ランタイムを使用して非同期エージェントと長時間稼働エージェントを処理する

Amazon Bedrock AgentCore ランタイムは、非同期処理と長時間実行エージェントを処理できます。非同期タスクを使用すると、エージェントはクライアントに応答した後も処理を継続し、応答をブロックせずに長時間実行されるオペレーションを処理できます。非同期処理を使用すると、エージェントは次のことができます。

主要なコンセプト

非同期処理モデル

Amazon Bedrock AgentCore SDK は、統合 API による同期処理と非同期処理の両方をサポートします。これにより、クライアント開発者とエージェント開発者の両方に柔軟な実装パターンが作成されます。エージェントクライアントは、クライアント側で同期と非同期を区別することなく、同じ API を使用できます。呼び出し間で同じセッションを呼び出す機能により、エージェントデベロッパーは複雑なタスク管理ロジックを実装することなく、コンテキストを再利用し、このコンテキストを段階的に構築できます。

ランタイムセッションライフサイクル管理

エージェントコードは、「/ping」エンドポイントのヘルスステータスを使用して処理ステータスを通信します。/ping エンドポイントは、次の JSON ペイロードを含む HTTP 200 レスポンスを返す必要があります。

{"status": "HealthyBusy"}

レスポンスには、必須フィールドとオプションフィールドが含まれます。

アイドル状態 ("Healthy") のセッションは 15 分間自動的に終了します。返されるセッション"HealthyBusy"は、アイドルタイムアウトを超えて存続します。

非同期タスクの実装

開始するには、 bedrock-agentcoreパッケージをインストールします。

pip install bedrock-agentcore

AgentCore SDK には、統合非同期処理のための以下のオプションが用意されています。

完全な例

まず、必要なパッケージをインストールします。

pip install strands-agents

次に、次のコードを使用して Python ファイルを作成します。

import threading
import time
from strands import Agent, tool
from bedrock_agentcore.runtime import BedrockAgentCoreApp

# Initialize app with debug mode for task management
app = BedrockAgentCoreApp()

@tool
def start_background_task(duration: int = 5) -> str:
    """Start a simple background task that runs for specified duration."""
    # Start tracking the async task
    task_id = app.add_async_task("background_processing", {"duration": duration})

    # Run task in background thread
    def background_work():
        time.sleep(duration)  # Simulate work
        app.complete_async_task(task_id)  # Mark as complete

    threading.Thread(target=background_work, daemon=True).start()
    return f"Started background task (ID: {task_id}) for {duration} seconds. Agent status is now BUSY."

# Create agent with the tool
agent = Agent(tools=[start_background_task])

@app.entrypoint
def main(payload):
    """Main entrypoint - handles user messages."""
    user_message = payload.get("prompt", "Try: start_background_task(3)")
    return {"message": agent(user_message).message}

if __name__ == "__main__":
    print("🚀 Simple Async Strands Example")
    print("Test: curl -X POST http://localhost:8080/invocations -H 'Content-Type: application/json' -d '{\"prompt\": \"start a 3 second task\"}'")
    app.run()

この例では、以下を示します。

一般的な の問題と解決策

長時間稼働エージェントが 15 分後に終了する

これは、アプリケーションがシングルスレッドで、ping スレッドがブロックされている場合に発生する可能性があります。