使用具有以下节点的 Amazon Bedrock 代理构建时端点发送 CreateFlow 请求，创建流：
运行以下代码片段来加载适用于 Python (Boto3) 的 AWS SDK，创建 Amazon Bedrock 代理客户端，然后使用节点创建流（将 executionRoleArn 字段替换为您为流创建的服务角色的 ARN）：
# 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")
运行以下代码片段，创建 Amazon Bedrock 代理运行时客户端并调用工作流。请求会在流的提示中填写变量，并返回模型的响应，以便使用 Amazon Bedrock 代理运行时端点发送 InvokeFlow 请求：
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'])
响应应返回包含三首歌曲的流行音乐播放列表。
