執行 Amazon Bedrock Flows 程式碼範例

下列程式碼範例假設您已完成下列先決條件：

1. 設定角色以擁有 Amazon Bedrock 動作的許可。如果您尚未，請參閱 Amazon Bedrock 入門。

2. 設定您的登入資料以使用 AWS API。如果尚未，請參閱 API 入門。

3. 建立服務角色，以代表您執行流程相關動作。如果您尚未，請參閱 在 Amazon Bedrock 中為 Amazon Bedrock 流程建立服務角色。

若要建立流程，請使用 Amazon Bedrock 建置時間端點的代理程式傳送 CreateFlow 請求。如需範例程式碼，請參閱 執行 Amazon Bedrock Flows 程式碼範例

下列是必要欄位：

欄位	基本描述
name	流程的名稱。
executionRoleArn	具有建立和管理流程許可的服務角色 ARN。

下列欄位為選用：

欄位	使用案例
定義	包含構成流程的 connections nodes和 。
description	描述流程。
標籤	將標籤與流程建立關聯。如需詳細資訊，請參閱標記 Amazon Bedrock 資源。
customerEncryptionKeyArn	使用 KMS 金鑰加密資源。如需詳細資訊，請參閱Amazon Bedrock Flows 資源的加密。
clientToken	為確保 API 請求僅完成一次。如需詳細資訊，請參閱確保冪等性。

雖然 definition 欄位是選用的，但流程需要正常運作。您可以選擇先建立沒有定義的流程，稍後再更新流程。

對於nodes清單中的每個節點，您可以在 type 欄位中指定節點的類型，並在 config 欄位中提供節點的對應組態。如需不同節點類型之 API 結構的詳細資訊，請參閱 流程中的節點類型。

若要嘗試 Amazon Bedrock Flows 的一些程式碼範例，請選擇您偏好方法的索引標籤，然後遵循下列步驟：

Python

1. 使用 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")

2. 執行下列程式碼片段，向 Amazon Bedrock 建置時間端點的代理程式提出 ListFlows 請求，以列出您帳戶中的流程，包括您剛建立的流程：

   client.list_flows()

3. 執行下列程式碼片段，向 Amazon Bedrock 建置時間端點的代理程式提出 GetFlow 請求，以取得您剛建立之流程的相關資訊：

   client.get_flow(flowIdentifier=flow_id)

4. 準備您的流程，以便套用工作草稿的最新變更，並準備好進行版本。執行下列程式碼片段，向 Amazon Bedrock 建置時間端點的代理程式提出 PrepareFlow 請求：

   client.prepare_flow(flowIdentifier=flow_id)

5. 版本流程的工作草稿，以建立流程的靜態快照，然後使用下列動作擷取其相關資訊：

   1. 透過執行下列程式碼片段來建立版本，以使用 Amazon Bedrock 建置時間端點的代理程式提出 CreateFlowVersion 請求：

      response = client.create_flow_version(flowIdentifier=flow_id)
                                      
      flow_version = response.get("version")

   2. 執行下列程式碼片段，以使用 Amazon Bedrock 建置時間端點的代理程式提出 ListFlowVersions 請求，列出流程的所有版本：

      client.list_flow_versions(flowIdentifier=flow_id)

   3. 執行下列程式碼片段，使用 Amazon Bedrock 的 Agents 建置時間端點提出 GetFlowVersion 請求，以取得版本的相關資訊：

      client.get_flow_version(flowIdentifier=flow_id, flowVersion=flow_version)

6. 建立別名以指向您建立的流程版本，然後使用下列動作擷取其相關資訊：

   1. 建立別名，並指向您剛建立的版本，方法是執行下列程式碼片段，向 Amazon Bedrock 建置時間端點的代理程式提出 CreateFlowAlias 請求：

      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")

   2. 執行下列程式碼片段，向 Amazon Bedrock 建置時間端點的代理程式提出 ListFlowAliass 請求，以列出流程的所有別名：

      client.list_flow_aliases(flowIdentifier=flow_id)

   3. 執行下列程式碼片段，向 Amazon Bedrock 建置時間端點的代理程式提出 GetFlowAlias 請求，以取得您剛建立的別名的相關資訊：

      client.get_flow_alias(flowIdentifier=flow_id, aliasIdentifier=flow_alias_id)

7. 執行下列程式碼片段來建立 Amazon Bedrock Agents 執行期用戶端並叫用流程。請求會填入流程中提示中的變數，並傳回模型的回應，以使用 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'])

   回應應會傳回播放三首歌曲的流行音樂。

8. 刪除您使用下列動作建立的別名、版本和流程：

   1. 執行下列程式碼片段來刪除別名，以向 Amazon Bedrock 建置時間端點的代理程式提出 DeleteFlowAlias 請求：

      client.delete_flow_alias(flowIdentifier=flow_id, aliasIdentifier=flow_alias_id)

   2. 透過執行下列程式碼片段來刪除版本，以使用 Amazon Bedrock 建置時間端點的代理程式提出 DeleteFlowVersion 請求：

      client.delete_flow_version(flowIdentifier=flow_id, flowVersion=flow_version)

   3. 透過執行下列程式碼片段來刪除流程，以使用 Amazon Bedrock 的 Agents 建置時間端點提出 DeleteFlow 請求：

      client.delete_flow(flowIdentifier=flow_id)