Behandeln Sie asynchrone und lang laufende Agenten mit Amazon AgentCore Bedrock Runtime

Amazon Bedrock AgentCore Runtime kann asynchrone Verarbeitung und Agenten mit langer Laufzeit verarbeiten. Asynchrone Aufgaben ermöglichen es Ihrem Agenten, die Verarbeitung fortzusetzen, nachdem er dem Client geantwortet hat, und lang andauernde Operationen abzuwickeln, ohne Antworten zu blockieren. Mit der asynchronen Verarbeitung kann Ihr Agent:

Die wichtigsten Konzepte

Asynchrones Verarbeitungsmodell

Das Amazon Bedrock AgentCore SDK unterstützt sowohl synchrone als auch asynchrone Verarbeitung über eine einheitliche API. Dadurch entsteht ein flexibles Implementierungsmuster sowohl für Kunden als auch für Agentenentwickler. Agent-Clients können mit derselben API arbeiten, ohne auf der Clientseite zwischen synchroner und asynchroner API unterscheiden zu müssen. Durch die Möglichkeit, dieselbe Sitzung bei mehreren Aufrufen aufzurufen, können Agentenentwickler den Kontext wiederverwenden und schrittweise auf diesem Kontext aufbauen, ohne eine komplexe Aufgabenverwaltungslogik implementieren zu müssen.

Verwaltung des Lebenszyklus von Runtime-Sitzungen

Der Agentencode übermittelt seinen Verarbeitungsstatus mithilfe des Integritätsstatus des Endpunkts „/ping“. Der /ping Endpunkt muss eine HTTP 200-Antwort mit der folgenden JSON-Payload zurückgeben:

{"status": "HealthyBusy"}

Die Antwort enthält ein erforderliches Feld und ein optionales Feld:

Eine Sitzung, die sich 15 Minuten lang im Ruhezustand ("Healthy") befindet, wird automatisch beendet. Eine zurückkehrende Sitzung "HealthyBusy" bleibt über das Timeout im Leerlauf hinaus aktiv.

Implementierung asynchroner Aufgaben

Um zu beginnen, installieren Sie das bedrock-agentcore Paket:

pip install bedrock-agentcore

AgentCore Das SDK bietet die folgenden Integrationsoptionen für die asynchrone Verarbeitung.

Vollständiges Beispiel

Installieren Sie zunächst das erforderliche Paket:

pip install strands-agents

Erstellen Sie dann eine Python-Datei mit dem folgenden Code:

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

Dieses Beispiel zeigt:

Häufige Probleme und Lösungen

Long-running Der Agent wird nach 15 Minuten beendet

Dies kann passieren, wenn die Anwendung über einen einzigen Thread verfügt und der Ping-Thread blockiert ist.