As traduções são geradas por tradução automática. Em caso de conflito entre o conteúdo da tradução e da versão original em inglês, a versão em inglês prevalecerá.

Conversar com um fluxo do Amazon Bedrock

A conversa de vários turnos do recurso Fluxos do Amazon Bedrock permite conversas dinâmicas entre usuários e fluxos, de modo semelhante a um diálogo em linguagem natural. Quando um nó de agente exige esclarecimento ou contexto adicional, ele pode pausar de forma inteligente a execução do fluxo e solicitar ao usuário informações específicas. Isso cria uma experiência mais interativa e sensível ao contexto, pois o nó pode adaptar seu próprio comportamento com base nas respostas do usuário. Por exemplo, se uma consulta inicial do usuário for ambígua ou incompleta, o nó poderá fazer perguntas complementares para coletar os detalhes necessários. Depois que o usuário fornece as informações solicitadas, o fluxo retoma perfeitamente a execução com a entrada ampliada, garantindo resultados mais precisos e relevantes. Esse recurso é particularmente valioso para cenários complexos em que uma única interação pode não ser suficiente para entender e atender totalmente às necessidades do usuário.

Como processar uma conversa de vários turnos em um fluxo

Para usar uma conversa de vários turnos em um fluxo, você precisa de um nó de agente conectado a um agente do Amazon Bedrock. Quando você executa o fluxo, uma conversa de vários turnos ocorre quando o agente precisa de mais informações do usuário para continuar. Esta seção descreve um fluxo que usa um agente com as seguintes instruções:

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 ter informações sobre como criar um agente, consulte Automatizar tarefas em sua aplicação usando agentes de IA.

Etapa 1: iniciar o fluxo

Inicie um fluxo chamando a operação InvokeFlow. Inclua o conteúdo inicial que deseja enviar ao fluxo. No exemplo a seguir, o campo document contém uma solicitação para criar uma playlist. Cada conversa tem um identificador exclusivo (ID de execução) que identifica a conversa dentro do fluxo. Para obter o ID de execução, não envie o campo executionID em sua primeira chamada a InvokeFlow. A resposta de InvokeFlow inclui o ID de execução. Em seu código, use o identificador para rastrear várias conversas e identificar uma conversa em outras chamadas à operação InvokeFlow.

Veja a seguir um exemplo JSON para uma solicitação a InvokeFlow.

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

Etapa 2: recuperar as solicitações do agente

Se o nó do agente no fluxo decidir que precisa de mais informações do usuário, o fluxo de resposta (responseStream) de InvokeFlow incluirá um objeto de evento FlowMultiTurnInputRequestEvent. O evento tem as informações solicitadas no campo content (FlowMultiTurnInputContent). No exemplo a seguir, a solicitação no campo document é para obter informações sobre o número de músicas, o gênero musical e o tema da playlist. Em seu código, você precisa obter essas informações do usuário.

Veja a seguir um exemplo de um objeto JSON FlowMultiTurnInputRequestEvent.

{
    "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 o fluxo não pode continuar enquanto mais entradas não forem recebidas, o fluxo também emite um evento FlowCompletionEvent. Um fluxo sempre emite FlowMultiTurnInputRequestEvent antes de FlowCompletionEvent. Se o valor de completionReason no evento FlowCompletionEvent for INPUT_REQUIRED, o fluxo precisará de mais informações para continuar.

Veja a seguir um exemplo de um objeto JSON FlowCompletionEvent.

{
    "completionReason": "INPUT_REQUIRED"
}

Etapa 3: enviar a resposta do usuário ao fluxo

Envie a resposta do usuário de volta ao fluxo chamando novamente a operação InvokeFlow. Você dever incluir o executionId da conversa.

Veja a seguir um exemplo de JSON para a solicitação a InvokeFlow. O campo document contém toda a resposta do usuário.

{
  "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"
    }
  ]
}

Se o fluxo precisar de mais informações, ele criará outros eventos FlowMultiTurnInputRequestEvent.

Etapa 4: encerrar o fluxo

Quando não são necessárias mais informações, o fluxo emite um evento FlowOutputEvent, que contém a resposta final.

Veja a seguir um exemplo de um objeto JSON FlowOutputEvent.

{
    "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!"
     }
}

O fluxo também emite um evento FlowCompletionEvent. O valor de completionReason é SUCCESS.

Veja a seguir um exemplo de um objeto JSON FlowCompletionEvent.

{
    "completionReason": "SUCCESS"
}

O diagrama de sequência a seguir mostra as etapas de um fluxo de vários turnos.

Criar e executar um exemplo de fluxo

Neste exemplo, você criará um fluxo que usa um agente para criar playlists para uma estação de rádio. O agente faz perguntas de esclarecimento para determinar o número de músicas, o gênero musical e o tema da playlist.

O exemplo de código Python a seguir mostra como usar o fluxo.

Para executar o código, especifique o seguinte:

Para ter informações sobre como obter os IDs, consulte Visualizar informações sobre fluxos no Amazon Bedrock. O código solicita que uma solicitação inicial seja enviada ao fluxo e pede mais informações de acordo com a necessidade do fluxo. O código não gerencia outras solicitações do agente, como solicitações para chamar funções do AWS Lambda. Para obter mais informações, consulte Como o Amazon Bedrock Agents funciona. Durante a execução, o código gera objetos FlowTraceEvent que você pode usar para rastrear o caminho da entrada até a resposta que o fluxo exibe. Para obter mais informações, consulte Rastrear cada etapa no fluxo visualizando os respectivos rastreamentos no 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)})