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.
Ejecución de ejemplos de código de Flujos de Amazon Bedrock
En los siguientes ejemplos de código se supone que cumple los siguientes requisitos previos:
Para crear un flujo, envíe una solicitud CreateFlow con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock. Para ver el código de ejemplo, consulte Ejecución de ejemplos de código de Flujos de Amazon Bedrock
Los siguientes campos son obligatorios:
Los siguientes campos son opcionales:
Si bien el campo definition es opcional, es necesario para que el flujo funcione. Puede optar por crear primero un flujo sin la definición y actualizar el flujo más adelante.
Para cada nodo de la lista de nodes, especifique el tipo de nodo en el campo type y proporcione la configuración correspondiente del nodo en el campo config. Para obtener más información sobre la estructura de API de los distintos tipos de nodo, consulte Tipos de nodos para el flujo.
Para probar algunos ejemplos de código para Flujos de Amazon Bedrock, seleccione la pestaña correspondiente al método que prefiera y siga los pasos:
- Python
-
-
Cree un flujo con una solicitud CreateFlow con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock con los siguientes nodos:
-
un nodo de entrada,
-
un nodo de petición con una petición definida en línea que crea una lista de reproducción de música con dos variables (genre y number),
-
un nodo de salida que devuelva la finalización del modelo.
Ejecute el siguiente fragmento de código para cargar AWS SDK para Python (Boto3), crear un cliente de Agentes para Amazon Bedrock y crear un flujo con los nodos (sustituya el campo executionRoleArn por el ARN del rol de servicio que ha creado para el flujo):
# Import Python SDK and create client
import boto3
client = boto3.client(service_name='bedrock-agent')
# Replace with the service role that you created. For more information, see https://docs.aws.amazon.com/bedrock/latest/userguide/flows-permissions.html
FLOWS_SERVICE_ROLE = "arn:aws:iam::123456789012:role/MyFlowsRole"
# Define each node
# The input node validates that the content of the InvokeFlow request is a JSON object.
input_node = {
"type": "Input",
"name": "FlowInput",
"outputs": [
{
"name": "document",
"type": "Object"
}
]
}
# This prompt node defines an inline prompt that creates a music playlist using two variables.
# 1. {{genre}} - The genre of music to create a playlist for
# 2. {{number}} - The number of songs to include in the playlist
# It validates that the input is a JSON object that minimally contains the fields "genre" and "number", which it will map to the prompt variables.
# The output must be named "modelCompletion" and be of the type "String".
prompt_node = {
"type": "Prompt",
"name": "MakePlaylist",
"configuration": {
"prompt": {
"sourceConfiguration": {
"inline": {
"modelId": "amazon.nova-lite-v1:0",
"templateType": "TEXT",
"inferenceConfiguration": {
"text": {
"temperature": 0.8
}
},
"templateConfiguration": {
"text": {
"text": "Make me a {{genre}} playlist consisting of the following number of songs: {{number}}."
}
}
}
}
}
},
"inputs": [
{
"name": "genre",
"type": "String",
"expression": "$.data.genre"
},
{
"name": "number",
"type": "Number",
"expression": "$.data.number"
}
],
"outputs": [
{
"name": "modelCompletion",
"type": "String"
}
]
}
# The output node validates that the output from the last node is a string and returns it as is. The name must be "document".
output_node = {
"type": "Output",
"name": "FlowOutput",
"inputs": [
{
"name": "document",
"type": "String",
"expression": "$.data"
}
]
}
# Create connections between the nodes
connections = []
# First, create connections between the output of the flow input node and each input of the prompt node
for input in prompt_node["inputs"]:
connections.append(
{
"name": "_".join([input_node["name"], prompt_node["name"], input["name"]]),
"source": input_node["name"],
"target": prompt_node["name"],
"type": "Data",
"configuration": {
"data": {
"sourceOutput": input_node["outputs"][0]["name"],
"targetInput": input["name"]
}
}
}
)
# Then, create a connection between the output of the prompt node and the input of the flow output node
connections.append(
{
"name": "_".join([prompt_node["name"], output_node["name"]]),
"source": prompt_node["name"],
"target": output_node["name"],
"type": "Data",
"configuration": {
"data": {
"sourceOutput": prompt_node["outputs"][0]["name"],
"targetInput": output_node["inputs"][0]["name"]
}
}
}
)
# Create the flow from the nodes and connections
response = client.create_flow(
name="FlowCreatePlaylist",
description="A flow that creates a playlist given a genre and number of songs to include in the playlist.",
executionRoleArn=FLOWS_SERVICE_ROLE,
definition={
"nodes": [input_node, prompt_node, output_node],
"connections": connections
}
)
flow_id = response.get("id")
-
Muestre los flujos de la cuenta, incluido el que acaba de crear, ejecutando el siguiente fragmento de código para realizar una solicitud ListFlows con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.list_flows()
-
Obtenga información sobre el flujo que acaba de crear ejecutando el siguiente fragmento de código para realizar una solicitud GetFlow con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.get_flow(flowIdentifier=flow_id)
-
Prepare el flujo para que se apliquen los cambios más recientes del borrador de trabajo y esté listo para la versión. Ejecute el siguiente fragmento de código para realizar una solicitud PrepareFlow con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.prepare_flow(flowIdentifier=flow_id)
-
Cree la versión del borrador de trabajo del flujo para crear una instantánea estática del flujo y, a continuación, obtenga la información sobre el trabajo mediante las siguientes acciones:
-
Cree una versión ejecutando el siguiente fragmento de código para realizar una solicitud CreateFlowVersion con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
response = client.create_flow_version(flowIdentifier=flow_id)
flow_version = response.get("version")
-
Muestre todas las versiones del flujo ejecutando el siguiente fragmento de código para realizar una solicitud ListFlowVersions con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.list_flow_versions(flowIdentifier=flow_id)
-
Obtenga información sobre la versión ejecutando el siguiente fragmento de código para realizar una solicitud GetFlowVersion con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.get_flow_version(flowIdentifier=flow_id, flowVersion=flow_version)
-
Cree un alias que apunte a la versión del flujo creado y, a continuación, recupere la información del flujo realizando las siguientes acciones:
-
Cree una alias y apúntelo hacia la versión que acaba de crear ejecutando el siguiente fragmento de código para realizar una solicitud CreateFlowAlias con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
response = client.create_flow_alias(
flowIdentifier=flow_id,
name="latest",
description="Alias pointing to the latest version of the flow.",
routingConfiguration=[
{
"flowVersion": flow_version
}
]
)
flow_alias_id = response.get("id")
-
Muestre todos los alias del flujo ejecutando el siguiente fragmento de código para realizar una solicitud ListFlowAliass con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.list_flow_aliases(flowIdentifier=flow_id)
-
Obtenga información sobre el alias que acaba de crear ejecutando el siguiente fragmento de código para realizar una solicitud GetFlowAlias con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.get_flow_alias(flowIdentifier=flow_id, aliasIdentifier=flow_alias_id)
-
Ejecute el siguiente fragmento de código para crear un cliente de tiempo de ejecución para agentes de Amazon Bedrock e invoque un flujo. La solicitud rellena las variables de la petición del flujo y devuelve la respuesta del modelo para realizar una solicitud InvokeFlow con un punto de conexión en tiempo de ejecución de Agentes para Amazon Bedrock:
client_runtime = boto3.client('bedrock-agent-runtime')
response = client_runtime.invoke_flow(
flowIdentifier=flow_id,
flowAliasIdentifier=flow_alias_id,
inputs=[
{
"content": {
"document": {
"genre": "pop",
"number": 3
}
},
"nodeName": "FlowInput",
"nodeOutputName": "document"
}
]
)
result = {}
for event in response.get("responseStream"):
result.update(event)
if result['flowCompletionEvent']['completionReason'] == 'SUCCESS':
print("Flow invocation was successful! The output of the flow is as follows:\n")
print(result['flowOutputEvent']['content']['document'])
else:
print("The flow invocation completed because of the following reason:", result['flowCompletionEvent']['completionReason'])
La respuesta debería devolver una lista de reproducción de música pop con tres canciones.
-
Elimine el alias, la versión y el flujo que ha creado con las siguientes acciones:
-
Elimine el alias ejecutando el siguiente fragmento de código para realizar una solicitud DeleteFlowAlias con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.delete_flow_alias(flowIdentifier=flow_id, aliasIdentifier=flow_alias_id)
-
Elimine la versión ejecutando el siguiente fragmento de código para realizar una solicitud DeleteFlowVersion con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.delete_flow_version(flowIdentifier=flow_id, flowVersion=flow_version)
-
Elimine el flujo ejecutando el siguiente fragmento de código para realizar una solicitud DeleteFlow con un punto de conexión en tiempo de compilación de Agentes para Amazon Bedrock:
client.delete_flow(flowIdentifier=flow_id)
