Gestisci agenti asincroni e di lunga durata con Amazon Bedrock Runtime AgentCore

Amazon Bedrock AgentCore Runtime è in grado di gestire l'elaborazione asincrona e gli agenti a lunga esecuzione. Le attività asincrone consentono all'agente di continuare l'elaborazione dopo aver risposto al client e di gestire operazioni di lunga durata senza bloccare le risposte. Con l'elaborazione asincrona, l'agente può:

Concetti chiave

Modello di elaborazione asincrono

L' AgentCore SDK Amazon Bedrock supporta l'elaborazione sincrona e asincrona tramite un'API unificata. Questo crea un modello di implementazione flessibile sia per i clienti che per gli agenti sviluppatori. I client agente possono lavorare con la stessa API senza distinguere tra sincroni e asincroni sul lato client. Grazie alla possibilità di richiamare la stessa sessione tra le chiamate, gli sviluppatori di agenti possono riutilizzare il contesto e svilupparlo in modo incrementale senza implementare una logica di gestione delle attività complessa.

Gestione del ciclo di vita della sessione di runtime

Il codice dell'agente comunica lo stato di elaborazione utilizzando lo stato di integrità dell'endpoint «/ping». L'/pingendpoint deve restituire una risposta HTTP 200 con il seguente payload JSON:

{"status": "HealthyBusy"}

La risposta contiene un campo obbligatorio e un campo facoltativo:

Una sessione in stato di inattività ("Healthy") per 15 minuti viene terminata automaticamente. Una sessione che ritorna "HealthyBusy" rimane attiva oltre il timeout di inattività.

Implementazione di attività asincrone

Per iniziare, installa il pacchetto: bedrock-agentcore

pip install bedrock-agentcore

AgentCore L'SDK offre le seguenti opzioni per l'integrazione dell'elaborazione asincrona.

Esempio completo

Innanzitutto, installa il pacchetto richiesto:

pip install strands-agents

Quindi, crea un file Python con il seguente codice:

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()

Questo esempio dimostra:

Problemi e soluzioni comuni

Long-running l'agente viene licenziato dopo 15 minuti

Ciò può accadere quando l'applicazione è a thread singolo e il thread di ping è bloccato.