Las traducciones son generadas a través de traducción automática. En caso de conflicto entre la traducción y la version original de inglés, prevalecerá la version en inglés.

Conversación con un flujo de Amazon Bedrock

La conversación de varios turnos de Flujos de Amazon Bedrock permite mantener conversaciones dinámicas de ida y vuelta entre los usuarios y los flujos, similares a un diálogo natural. Cuando un nodo de agente requiere aclaraciones o contexto adicional, puede pausar de forma inteligente la ejecución del flujo y solicitar al usuario información específica. Esto crea una experiencia más interactiva y con mayor contexto, ya que el nodo puede adaptar su comportamiento en función de las respuestas de los usuarios. Por ejemplo, si una consulta inicial de un usuario es ambigua o está incompleta, el nodo puede hacer preguntas de seguimiento para recopilar los detalles necesarios. Una vez que el usuario proporciona la información solicitada, el flujo reanuda la ejecución sin problemas con la entrada enriquecida, lo que garantiza resultados más precisos y pertinentes. Esta capacidad es particularmente útil en escenarios complejos en los que una sola interacción puede no ser suficiente para comprender y abordar completamente las necesidades del usuario.

Cómo procesar una conversación de varios turnos en un flujo

Para utilizar una conversación de varios turnos en un flujo, necesita un nodo de agente conectado a un agente de Amazon Bedrock. Al ejecutar el flujo, se produce una conversación de varios turnos cuando el agente necesita más información del usuario para poder continuar. En esta sección se describe un flujo que utiliza un agente con las siguientes instrucciones:

You are a playlist creator for a radio station. 
When asked to create a playlist, ask for the number of songs,
the genre of music, and a theme for the playlist.

Para obtener información acerca de la creación de un agente, consulte Automatización de las tareas en la aplicación mediante agentes de IA.

Paso 1: inicio del flujo

Inicie un flujo llamando a la operación InvokeFlow. Incluya el contenido inicial que desea enviar al flujo. En el siguiente ejemplo, el campo document contiene una solicitud para crear una lista de reproducción. Cada conversación tiene un identificador único (ID de ejecución) que identifica la conversación dentro del flujo. Para obtener el ID de ejecución, no envíe el campo executionID en su primera llamada a InvokeFlow. La respuesta de InvokeFlow incluye el ID de ejecución. En el código, utilice el identificador para realizar un seguimiento de varias conversaciones e identificar una conversación en las siguientes llamadas a la operación InvokeFlow.

A continuación, se muestra un ejemplo de solicitud JSON para InvokeFlow.

{
  "flowIdentifier": "XXXXXXXXXX",
  "flowAliasIdentifier": "YYYYYYYYYY",
  "inputs": [
    {
      "content": {
        "document": "Create a playlist."
      },
      "nodeName": "FlowInputNode",
      "nodeOutputName": "document"
    }
  ]
}

Paso 2: recuperación de las solicitudes del agente

Si el nodo de agente del flujo decide que necesita más información del usuario, el flujo de respuesta (responseStream) de InvokeFlow incluye un objeto del evento FlowMultiTurnInputRequestEvent. El evento tiene la información solicitada en el campo content (FlowMultiTurnInputContent). En el siguiente ejemplo, la solicitud del campo document es para obtener información sobre el número de canciones, el género musical y el tema de la lista de reproducción. Luego, en su código, necesita obtener esa información del usuario.

A continuación se muestra un objeto JSON FlowMultiTurnInputRequestEvent de ejemplo:

{
    "nodeName": "AgentsNode_1",
    "nodeType": "AgentNode",
    "content": {
        "document": "Certainly! I'd be happy to create a playlist for you. To make sure it's tailored to your preferences, could you please provide me with the following information:
        1. How many songs would you like in the playlist?
        2. What genre of music do you prefer? 
        3. Is there a specific theme or mood you'd like for the playlist? Once you provide these details, I'll be able to create a customized playlist just for you."
    }
}

Como el flujo no puede continuar hasta que se reciban más entradas, también emite un evento FlowCompletionEvent. Un flujo siempre emite FlowMultiTurnInputRequestEvent antes que FlowCompletionEvent. Si el valor de completionReason en el evento FlowCompletionEvent es INPUT_REQUIRED, el flujo necesita más información para poder continuar.

A continuación se muestra un objeto JSON FlowCompletionEvent de ejemplo:

{
    "completionReason": "INPUT_REQUIRED"
}

Paso 3: envío de la respuesta del usuario al flujo

Envíe la respuesta del usuario al flujo volviendo a llamar a la operación InvokeFlow. Asegúrese de incluir el executionId de la conversación.

A continuación, se muestra código JSON de ejemplo para la solicitud a InvokeFlow. El campo document contiene la respuesta del usuario.

{
  "flowIdentifier": "AUS7BMHXBE",
  "flowAliasIdentifier": "4KUDB8VBEF",
  "executionId": "b6450554-f8cc-4934-bf46-f66ed89b60a0",
  "inputs": [
    {
      "content": {
        "document": "1. 5 songs 2. Welsh rock music 3. Castles"
      },
      "nodeName": "AgentsNode_1",
      "nodeInputName": "agentInputText"
    }
  ]
}

Si el flujo necesita más información, crea más eventos FlowMultiTurnInputRequestEvent.

Paso 4: finalización del flujo

Cuando no se necesita más información, el flujo emite un evento FlowOutputEvent que contiene la respuesta final.

A continuación se muestra un objeto JSON FlowOutputEvent de ejemplo:

{
    "nodeName": "FlowOutputNode",
    "content": {
        "document": "Great news! I've created a 5-song Welsh rock playlist centered around the theme of castles. 
        Here's the playlist I've put together for you: Playlist Name: Welsh Rock Castle Anthems 
        Description: A 5-song Welsh rock playlist featuring songs about castles 
        Songs: 
        1. Castell y Bere - Super Furry Animals 
        2. The Castle - Manic Street Preachers 
        3. Caerdydd (Cardiff Castle) - Stereophonics 
        4. Castell Coch - Catatonia 
        5. Chepstow Castle - Feeder 
        This playlist combines Welsh rock bands with songs that reference castles or specific Welsh castles. 
        Enjoy your castle-themed Welsh rock music experience!"
     }
}

El flujo también emite un evento FlowCompletionEvent. El valor de completionReason es SUCCESS.

A continuación se muestra un objeto JSON FlowCompletionEvent de ejemplo:

{
    "completionReason": "SUCCESS"
}

El siguiente diagrama de secuencia muestra los pasos de un flujo de varios turnos.

Creación y ejecución de un flujo de ejemplo

En este ejemplo, se crea un flujo que utiliza un agente para crear listas de reproducción para una emisora de radio. El agente hace preguntas aclaratorias para determinar el número de canciones, el género musical y el tema de la lista de reproducción.

El siguiente ejemplo de código Python muestra cómo usar el flujo.

Para ejecutar el código, especifique lo siguiente:

Para obtener información acerca de cómo obtener estos ID, consulte Consulta de información sobre los flujos en Amazon Bedrock. El código solicita que se envíe una solicitud inicial al flujo y pide más información según sea necesario para el flujo. El código no administra otras solicitudes del agente, como las solicitudes de llamada a las funciones de AWS Lambda. Para obtener más información, consulte Funcionamiento de los agentes de Amazon Bedrock. Mientras se ejecuta, el código genera objetos FlowTraceEvent que puede usar para rastrear la ruta desde la entrada hasta la respuesta que devuelve el flujo. Para obtener más información, consulte Seguimiento de cada paso del flujo viendo su rastro en Amazon Bedrock.

"""
Runs an Amazon Bedrock flow and handles muli-turn interaction for a single conversation.

"""
import logging
import boto3
import botocore



import botocore.exceptions

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def invoke_flow(client, flow_id, flow_alias_id, input_data, execution_id):
    """
    Invoke an Amazon Bedrock flow and handle the response stream.

    Args:
        client: Boto3 client for Amazon Bedrock agent runtime
        flow_id: The ID of the flow to invoke
        flow_alias_id: The alias ID of the flow
        input_data: Input data for the flow
        execution_id: Execution ID for continuing a flow. Use the value None on first run.

    Returns:
        Dict containing flow_complete status, input_required info, and execution_id
    """

    response = None
    request_params = None

    if execution_id is None:
        # Don't pass execution ID for first run.
        request_params = {
            "flowIdentifier": flow_id,
            "flowAliasIdentifier": flow_alias_id,
            "inputs": [input_data],
            "enableTrace": True
        }
    else:
        request_params = {
            "flowIdentifier": flow_id,
            "flowAliasIdentifier": flow_alias_id,
            "executionId": execution_id,
            "inputs": [input_data],
            "enableTrace": True
        }

    response = client.invoke_flow(**request_params)
    if "executionId" not in request_params:
        execution_id = response['executionId']

    input_required = None
    flow_status = ""

    # Process the streaming response
    for event in response['responseStream']:
        # Check if flow is complete.
        if 'flowCompletionEvent' in event:
            flow_status = event['flowCompletionEvent']['completionReason']

        # Check if more input us needed from user.
        elif 'flowMultiTurnInputRequestEvent' in event:
            input_required = event

        # Print the model output.
        elif 'flowOutputEvent' in event:
            print(event['flowOutputEvent']['content']['document'])

        elif 'flowTraceEvent' in event:
            logger.info("Flow trace:  %s", event['flowTraceEvent'])

    return {
        "flow_status": flow_status,
        "input_required": input_required,
        "execution_id": execution_id
    }


if __name__ == "__main__":

    session = boto3.Session(profile_name='default', region_name='YOUR_FLOW_REGION')
    bedrock_agent_client = session.client('bedrock-agent-runtime')
    
    # Replace these with your actual flow ID and alias ID
    FLOW_ID = 'YOUR_FLOW_ID'
    FLOW_ALIAS_ID = 'YOUR_FLOW_ALIAS_ID'


    flow_execution_id = None
    finished = False

    # Get the intial prompt from the user.
    user_input = input("Enter input: ")

    flow_input_data = {
        "content": {
            "document": user_input
        },
        "nodeName": "FlowInputNode",
        "nodeOutputName": "document"
    }

    logger.info("Starting flow %s", FLOW_ID)

    try:
        while not finished:
            # Invoke the flow until successfully finished.

            result = invoke_flow(
                bedrock_agent_client, FLOW_ID, FLOW_ALIAS_ID, flow_input_data, flow_execution_id)
            status = result['flow_status']
            flow_execution_id = result['execution_id']
            more_input = result['input_required']
            if status == "INPUT_REQUIRED":
                # The flow needs more information from the user.
                logger.info("The flow %s requires more input", FLOW_ID)
                user_input = input(
                    more_input['flowMultiTurnInputRequestEvent']['content']['document'] + ": ")
                flow_input_data = {
                    "content": {
                        "document": user_input
                    },
                    "nodeName": more_input['flowMultiTurnInputRequestEvent']['nodeName'],
                    "nodeInputName": "agentInputText"

                }
            elif status == "SUCCESS":
                # The flow completed successfully.
                finished = True
                logger.info("The flow %s successfully completed.", FLOW_ID)

    except botocore.exceptions.ClientError as e:
        print(f"Client error: {str(e)}")
        logger.error("Client error: %s", {str(e)})

    except Exception as e:
        print(f"An error occurred: {str(e)}")
        logger.error("An error occurred: %s", {str(e)})
        logger.error("Error type: %s", {type(e)})