Gestione agentes asíncronos y de larga ejecución con Amazon Bedrock Runtime AgentCore

Amazon Bedrock AgentCore Runtime puede gestionar el procesamiento asíncrono y los agentes de larga ejecución. Las tareas asincrónicas permiten a su agente continuar procesando después de responder al cliente y gestionar las operaciones de larga duración sin bloquear las respuestas. Con el procesamiento asíncrono, su agente puede:

Conceptos clave

Modelo de procesamiento asíncrono

El AgentCore SDK de Amazon Bedrock admite el procesamiento sincrónico y asíncrono a través de una API unificada. Esto crea un patrón de implementación flexible tanto para los clientes como para los agentes desarrolladores. Los clientes agentes pueden trabajar con la misma API sin diferenciar entre síncronos y asíncronos en el lado del cliente. Gracias a la posibilidad de invocar la misma sesión en todas las invocaciones, los desarrolladores de agentes pueden reutilizar el contexto y aprovechar este contexto de forma incremental sin necesidad de implementar una lógica compleja de administración de tareas.

Administración del ciclo de vida de las sesiones en

El código del agente comunica su estado de procesamiento mediante el estado del punto final «/ping». El /ping punto final debe devolver una respuesta HTTP 200 con la siguiente carga útil JSON:

{"status": "HealthyBusy"}

La respuesta contiene un campo obligatorio y un campo opcional:

Una sesión en estado inactivo ("Healthy") durante 15 minutos finaliza automáticamente. Una sesión que regresa "HealthyBusy" permanece activa después del tiempo de espera de inactividad.

Implementación de tareas asíncronas

Para empezar, instale el paquete: bedrock-agentcore

pip install bedrock-agentcore

AgentCore El SDK proporciona las siguientes opciones para el procesamiento asíncrono de integración.

Ejemplo completo

Primero, instale el paquete requerido:

pip install strands-agents

A continuación, cree un archivo Python con el siguiente código:

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

Este ejemplo demuestra lo siguiente:

Problemas y soluciones comunes de

Long-running el agente es despedido después de 15 minutos

Esto puede ocurrir cuando la aplicación tiene un solo subproceso y el subproceso de ping está bloqueado.