本文為英文版的機器翻譯版本,如內容有任何歧義或不一致之處,概以英文版為準。
下列程式碼範例假設您已完成下列先決條件:
-
設定角色以擁有 Amazon Bedrock 動作的許可。如果您尚未完成,請參閱 Amazon Bedrock 入門。
-
設定您的登入資料以使用 AWS API。如果您尚未完成,請參閱 API 入門。
-
建立服務角色,代表您執行流程相關動作。如果您尚未完成,請參閱 在 Amazon Bedrock 中為 Amazon Bedrock 流程建立服務角色。
若要嘗試 Amazon Bedrock Flows 的一些程式碼範例,請選擇您偏好方法的索引標籤,然後遵循下列步驟:
-
使用 CreateFlow 請求搭配具有下列節點的 Amazon Bedrock 的 Agents 建置時間端點來建立流程:
-
輸入節點。
-
提示節點,其中包含提示定義的內嵌,可使用兩個變數 (
genre
和 ) 建立音樂播放清單number
。 -
傳回模型完成的輸出節點。
執行下列程式碼片段以載入 AWS SDK for Python (Boto3)、建立 Amazon Bedrock Agents 用戶端,以及使用節點建立流程 (將
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.titan-text-express-v1", "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 建置時間端點的代理程式提出 ListFlows 請求,列出您帳戶中的流程,包括您剛建立的流程:
client.list_flows()
-
執行下列程式碼片段,透過 Amazon Bedrock 建置時間端點的代理程式提出 GetFlow 請求,以取得您剛建立之流程的相關資訊:
client.get_flow(flowIdentifier=flow_id)
-
準備您的流程,以便套用工作草稿的最新變更,並準備好進行版本。執行下列程式碼片段,向 Amazon Bedrock 建置時間端點的代理程式提出 PrepareFlow 請求:
client.prepare_flow(flowIdentifier=flow_id)
-
版本流程的工作草稿,以建立流程的靜態快照,然後使用下列動作擷取其相關資訊:
-
透過執行下列程式碼片段來建立版本,以向 Amazon Bedrock 建置時間端點的代理程式提出 CreateFlowVersion 請求:
response = client.create_flow_version(flowIdentifier=flow_id) flow_version = response.get("version")
-
執行下列程式碼片段,使用 Amazon Bedrock 建置時間端點的代理程式提出 ListFlowVersions 請求,以列出流程的所有版本:
client.list_flow_versions(flowIdentifier=flow_id)
-
執行下列程式碼片段,透過 Amazon Bedrock 建置時間端點的代理程式提出 GetFlowVersion 請求,以取得版本的相關資訊:
client.get_flow_version(flowIdentifier=flow_id, flowVersion=flow_version)
-
-
建立別名以指向您建立的流程版本,然後使用下列動作擷取其相關資訊:
-
建立別名,並指向您剛建立的版本,方法是執行下列程式碼片段,向 Amazon Bedrock 建置時間端點的代理程式提出 CreateFlowAlias 請求:
response = client.create_flow_alias( flowIdentifier="FLOW123456", name="latest", description="Alias pointing to the latest version of the flow.", routingConfiguration=[ { "flowVersion": flow_version } ] ) flow_alias_id = response.get("id")
-
執行下列程式碼片段,向 Amazon Bedrock 建置時間端點的代理程式提出 ListFlowAliass 請求,以列出流程的所有別名:
client.list_flow_aliases(flowIdentifier=flow_id)
-
執行下列程式碼片段,向 Amazon Bedrock 建置時間端點的代理程式提出 GetFlowAlias 請求,以取得您剛建立的別名的相關資訊:
client.get_flow_alias(flowIdentifier=flow_id, aliasIdentifier=flow_alias_id)
-
-
執行下列程式碼片段來建立 Amazon Bedrock Agents Runtime 用戶端並叫用流程。請求會填入流程中提示中的變數,並傳回模型的回應,以使用 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'])
回應應會傳回包含三首歌曲的流行音樂播放清單。
-
刪除您使用下列動作建立的別名、版本和流程:
-
透過執行下列程式碼片段來刪除別名,以向 Amazon Bedrock 建置時間端點的代理程式提出 DeleteFlowAlias 請求:
client.delete_flow_alias(flowIdentifier=flow_id, aliasIdentifier=flow_alias_id)
-
透過執行下列程式碼片段來刪除版本,以向 Amazon Bedrock 建置時間端點的代理程式提出 DeleteFlowVersion 請求:
client.delete_flow_version(flowIdentifier=flow_id, flowVersion=flow_version)
-
透過執行下列程式碼片段來刪除流程,以向 Amazon Bedrock 建置時間端點的代理程式提出 DeleteFlow 請求:
client.delete_flow(flowIdentifier=flow_id)
-