Weitere AWS SDK-Beispiele sind im GitHub Repo [AWS Doc SDK Examples](https://github.com/awsdocs/aws-doc-sdk-examples) verfügbar.
Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.
# Szenarien für die Verwendung von Amazon Bedrock Agents Runtime AWS SDKs
Die folgenden Codebeispiele zeigen Ihnen, wie Sie allgemeine Szenarien in Amazon Bedrock Agents Runtime mit AWS SDKs implementieren. Diese Szenarien zeigen Ihnen, wie Sie bestimmte Aufgaben durch den Aufruf mehrerer Funktionen innerhalb von Runtime der Agenten für Amazon Bedrock oder in Kombination mit anderen AWS-Services ausführen können. Jedes Szenario enthält einen Link zum vollständigen Quell-Code, wo Sie Anleitungen zum Einrichten und Ausführen des Codes finden.
Szenarien zielen auf eine mittlere Erfahrungsebene ab, um Ihnen zu helfen, Service-Aktionen im Kontext zu verstehen.
**Topics**
+ [Erstellen und Aufrufen eines Flows](bedrock-agent-runtime_example_bedrock-agent_GettingStartedWithBedrockFlows_section.md)
+ [Orchestrieren von Anwendungen mit generativer KI mit Step Functions](bedrock-agent-runtime_example_cross_ServerlessPromptChaining_section.md)
# Ein end-to-end Beispiel, das zeigt, wie ein Amazon Bedrock-Flow mithilfe eines SDK erstellt und aufgerufen wird AWS
Wie das aussehen kann, sehen Sie am nachfolgenden Beispielcode:
+ Erstellen einer Ausführungsrolle für den Flow.
+ Erstellen Sie den Flow.
+ Stellen Sie den vollständig konfigurierten Flow bereit.
+ Rufen Sie den Flow mit vom Benutzer bereitgestellten Prompts auf.
+ Löschen aller erstellten Ressourcen.
------
#### [ Python ]
**SDK für Python (Boto3)**
Es gibt noch mehr dazu GitHub. Hier finden Sie das vollständige Beispiel und erfahren, wie Sie das [AWS -Code-Beispiel-](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/bedrock-agent#code-examples) einrichten und ausführen.
Erstellt eine Musik-Playlist basierend auf dem vom Benutzenden angegebenen Genre und der Anzahl der Songs.
```
from datetime import datetime
import logging
import boto3
from botocore.exceptions import ClientError
from roles import create_flow_role, delete_flow_role, update_role_policy
from flow import create_flow, prepare_flow, delete_flow
from run_flow import run_playlist_flow
from flow_version import create_flow_version, delete_flow_version
from flow_alias import create_flow_alias, delete_flow_alias
logging.basicConfig(
level=logging.INFO
)
logger = logging.getLogger(__name__)
def create_input_node(name):
"""
Creates an input node configuration for an Amazon Bedrock flow.
The input node serves as the entry point for the flow and defines
the initial document structure that will be passed to subsequent nodes.
Args:
name (str): The name of the input node.
Returns:
dict: The input node configuration.
"""
return {
"type": "Input",
"name": name,
"outputs": [
{
"name": "document",
"type": "Object"
}
]
}
def create_prompt_node(name, model_id):
"""
Creates a prompt node configuration for a Bedrock flow that generates music playlists.
The prompt node defines an inline prompt template that creates a music playlist based on
a specified genre and number of songs. The prompt uses two variables that are mapped from
the input JSON object:
- {{genre}}: The genre of music to create a playlist for
- {{number}}: The number of songs to include in the playlist
Args:
name (str): The name of the prompt node.
model_id (str): The identifier of the foundation model to use for the prompt.
Returns:
dict: The prompt node.
"""
return {
"type": "Prompt",
"name": name,
"configuration": {
"prompt": {
"sourceConfiguration": {
"inline": {
"modelId": model_id,
"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"
}
]
}
def create_output_node(name):
"""
Creates an output node configuration for a Bedrock flow.
The output node validates that the output from the last node is a string
and returns it unmodified. The input name must be "document".
Args:
name (str): The name of the output node.
Returns:
dict: The output node configuration containing the output node:
"""
return {
"type": "Output",
"name": name,
"inputs": [
{
"name": "document",
"type": "String",
"expression": "$.data"
}
]
}
def create_playlist_flow(client, flow_name, flow_description, role_arn, prompt_model_id):
"""
Creates the playlist generator flow.
Args:
client: bedrock agent boto3 client.
role_arn (str): Name for the new IAM role.
prompt_model_id (str): The id of the model to use in the prompt node.
Returns:
dict: The response from the create_flow operation.
"""
input_node = create_input_node("FlowInput")
prompt_node = create_prompt_node("MakePlaylist", prompt_model_id)
output_node = create_output_node("FlowOutput")
# 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 prompt_node_input in prompt_node["inputs"]:
connections.append(
{
"name": "_".join([input_node["name"], prompt_node["name"],
prompt_node_input["name"]]),
"source": input_node["name"],
"target": prompt_node["name"],
"type": "Data",
"configuration": {
"data": {
"sourceOutput": input_node["outputs"][0]["name"],
"targetInput": prompt_node_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"]
}
}
}
)
flow_def = {
"nodes": [input_node, prompt_node, output_node],
"connections": connections
}
# Create the flow.
response = create_flow(
client, flow_name, flow_description, role_arn, flow_def)
return response
def get_model_arn(client, model_id):
"""
Gets the Amazon Resource Name (ARN) for a model.
Args:
client (str): Amazon Bedrock boto3 client.
model_id (str): The id of the model.
Returns:
str: The ARN of the model.
"""
try:
# Call GetFoundationModelDetails operation
response = client.get_foundation_model(modelIdentifier=model_id)
# Extract model ARN from the response
model_arn = response['modelDetails']['modelArn']
return model_arn
except ClientError as e:
logger.exception("Client error getting model ARN: %s", {str(e)})
raise
except Exception as e:
logger.exception("Unexpected error getting model ARN: %s", {str(e)})
raise
def prepare_flow_version_and_alias(bedrock_agent_client,
flow_id):
"""
Prepares the flow and then creates a flow version and flow alias.
Args:
bedrock_agent_client: Amazon Bedrock Agent boto3 client.
flowd_id (str): The ID of the flow that you want to prepare.
Returns: The flow_version and flow_alias.
"""
status = prepare_flow(bedrock_agent_client, flow_id)
flow_version = None
flow_alias = None
if status == 'Prepared':
# Create the flow version and alias.
flow_version = create_flow_version(bedrock_agent_client,
flow_id,
f"flow version for flow {flow_id}.")
flow_alias = create_flow_alias(bedrock_agent_client,
flow_id,
flow_version,
"latest",
f"Alias for flow {flow_id}, version {flow_version}")
return flow_version, flow_alias
def delete_role_resources(bedrock_agent_client,
iam_client,
role_name,
flow_id,
flow_version,
flow_alias):
"""
Deletes the flow, flow alias, flow version, and IAM roles.
Args:
bedrock_agent_client: Amazon Bedrock Agent boto3 client.
iam_client: Amazon IAM boto3 client.
role_name (str): The name of the IAM role.
flow_id (str): The id of the flow.
flow_version (str): The version of the flow.
flow_alias (str): The alias of the flow.
"""
if flow_id is not None:
if flow_alias is not None:
delete_flow_alias(bedrock_agent_client, flow_id, flow_alias)
if flow_version is not None:
delete_flow_version(bedrock_agent_client,
flow_id, flow_version)
delete_flow(bedrock_agent_client, flow_id)
if role_name is not None:
delete_flow_role(iam_client, role_name)
def main():
"""
Creates, runs, and optionally deletes a Bedrock flow for generating music playlists.
Note:
Requires valid AWS credentials in the default profile
"""
delete_choice = "y"
try:
# Get various boto3 clients.
session = boto3.Session(profile_name='default')
bedrock_agent_runtime_client = session.client('bedrock-agent-runtime')
bedrock_agent_client = session.client('bedrock-agent')
bedrock_client = session.client('bedrock')
iam_client = session.client('iam')
role_name = None
flow_id = None
flow_version = None
flow_alias = None
#Change the model as needed.
prompt_model_id = "amazon.nova-pro-v1:0"
# Base the flow name on the current date and time
current_time = datetime.now()
timestamp = current_time.strftime("%Y-%m-%d-%H-%M-%S")
flow_name = f"FlowPlayList_{timestamp}"
flow_description = "A flow to generate a music playlist."
# Create a role for the flow.
role_name = f"BedrockFlowRole-{flow_name}"
role = create_flow_role(iam_client, role_name)
role_arn = role['Arn']
# Create the flow.
response = create_playlist_flow(
bedrock_agent_client, flow_name, flow_description, role_arn, prompt_model_id)
flow_id = response.get('id')
if flow_id:
# Update accessible resources in the role.
model_arn = get_model_arn(bedrock_client, prompt_model_id)
update_role_policy(iam_client, role_name, [
response.get('arn'), model_arn])
# Prepare the flow and flow version.
flow_version, flow_alias = prepare_flow_version_and_alias(
bedrock_agent_client, flow_id)
# Run the flow.
if flow_version and flow_alias:
run_playlist_flow(bedrock_agent_runtime_client,
flow_id, flow_alias)
delete_choice = input("Delete flow? y or n : ").lower()
else:
print("Couldn't run. Deleting flow and role.")
delete_flow(bedrock_agent_client, flow_id)
delete_flow_role(iam_client, role_name)
else:
print("Couldn't create flow.")
except Exception as e:
print(f"Fatal error: {str(e)}")
finally:
if delete_choice == 'y':
delete_role_resources(bedrock_agent_client,
iam_client,
role_name,
flow_id,
flow_version,
flow_alias)
else:
print("Flow not deleted. ")
print(f"\tFlow ID: {flow_id}")
print(f"\tFlow version: {flow_version}")
print(f"\tFlow alias: {flow_alias}")
print(f"\tRole ARN: {role_arn}")
print("Done!")
if __name__ == "__main__":
main()
def invoke_flow(client, flow_id, flow_alias_id, input_data):
"""
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.
Returns:
Dict containing flow status and flow output.
"""
response = None
request_params = None
request_params = {
"flowIdentifier": flow_id,
"flowAliasIdentifier": flow_alias_id,
"inputs": [input_data],
"enableTrace": True
}
response = client.invoke_flow(**request_params)
flow_status = ""
output= ""
# Process the streaming response
for event in response['responseStream']:
# Check if flow is complete.
if 'flowCompletionEvent' in event:
flow_status = event['flowCompletionEvent']['completionReason']
# Save the model output.
elif 'flowOutputEvent' in event:
output = event['flowOutputEvent']['content']['document']
logger.info("Output : %s", output)
# Log trace events.
elif 'flowTraceEvent' in event:
logger.info("Flow trace: %s", event['flowTraceEvent'])
return {
"flow_status": flow_status,
"output": output
}
def run_playlist_flow(bedrock_agent_client, flow_id, flow_alias_id):
"""
Runs the playlist generator flow.
Args:
bedrock_agent_client: Boto3 client for Amazon Bedrock agent runtime.
flow_id: The ID of the flow to run.
flow_alias_id: The alias ID of the flow.
"""
print ("Welcome to the playlist generator flow.")
# Get the initial prompt from the user.
genre = input("Enter genre: ")
number_of_songs = int(input("Enter number of songs: "))
# Use prompt to create input data for the input node.
flow_input_data = {
"content": {
"document": {
"genre" : genre,
"number" : number_of_songs
}
},
"nodeName": "FlowInput",
"nodeOutputName": "document"
}
try:
result = invoke_flow(
bedrock_agent_client, flow_id, flow_alias_id, flow_input_data)
status = result['flow_status']
if status == "SUCCESS":
# The flow completed successfully.
logger.info("The flow %s successfully completed.", flow_id)
print(result['output'])
else:
logger.warning("Flow status: %s",status)
except ClientError as e:
print(f"Client error: {str(e)}")
logger.error("Client error: %s", {str(e)})
raise
except Exception as e:
logger.error("An error occurred: %s", {str(e)})
logger.error("Error type: %s", {type(e)})
raise
def create_flow_role(client, role_name):
"""
Creates an IAM role for Amazon Bedrock with permissions to run a flow.
Args:
role_name (str): Name for the new IAM role.
Returns:
str: The role Amazon Resource Name.
"""
# Trust relationship policy - allows Amazon Bedrock service to assume this role.
trust_policy = {
"Version":"2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {
"Service": "bedrock.amazonaws.com"
},
"Action": "sts:AssumeRole"
}]
}
# Basic inline policy for for running a flow.
resources = "*"
bedrock_policy = {
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"bedrock:InvokeModel",
"bedrock:Retrieve",
"bedrock:RetrieveAndGenerate"
],
# Using * as placeholder - Later you update with specific ARNs.
"Resource": resources
}
]
}
try:
# Create the IAM role with trust policy
logging.info("Creating role: %s",role_name)
role = client.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=json.dumps(trust_policy),
Description="Role for Amazon Bedrock operations"
)
# Attach inline policy to the role
print("Attaching inline policy")
client.put_role_policy(
RoleName=role_name,
PolicyName=f"{role_name}-policy",
PolicyDocument=json.dumps(bedrock_policy)
)
logging.info("Create Role ARN: %s", role['Role']['Arn'])
return role['Role']
except ClientError as e:
logging.warning("Error creating role: %s", str(e))
raise
except Exception as e:
logging.warning("Unexpected error: %s", str(e))
raise
def update_role_policy(client, role_name, resource_arns):
"""
Updates an IAM role's inline policy with specific resource ARNs.
Args:
role_name (str): Name of the existing role.
resource_arns (list): List of resource ARNs to allow access to.
"""
updated_policy = {
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"bedrock:GetFlow",
"bedrock:InvokeModel",
"bedrock:Retrieve",
"bedrock:RetrieveAndGenerate"
],
"Resource": resource_arns
}
]
}
try:
client.put_role_policy(
RoleName=role_name,
PolicyName=f"{role_name}-policy",
PolicyDocument=json.dumps(updated_policy)
)
logging.info("Updated policy for role: %s",role_name)
except ClientError as e:
logging.warning("Error updating role policy: %s", str(e))
raise
def delete_flow_role(client, role_name):
"""
Deletes an IAM role.
Args:
role_name (str): Name of the role to delete.
"""
try:
# Detach and delete inline policies
policies = client.list_role_policies(RoleName=role_name)['PolicyNames']
for policy_name in policies:
client.delete_role_policy(RoleName=role_name, PolicyName=policy_name)
# Delete the role
client.delete_role(RoleName=role_name)
logging.info("Deleted role: %s", role_name)
except ClientError as e:
logging.info("Error Deleting role: %s", str(e))
raise
```
+ Weitere API-Informationen finden Sie in den folgenden Themen der *API-Referenz zum AWS SDK für Python (Boto3)*.
+ [CreateFlow](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreateFlow)
+ [CreateFlowAlias](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreateFlowAlias)
+ [CreateFlowVersion](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreateFlowVersion)
+ [DeleteFlow](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/DeleteFlow)
+ [DeleteFlowAlias](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/DeleteFlowAlias)
+ [DeleteFlowVersion](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/DeleteFlowVersion)
+ [GetFlow](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/GetFlow)
+ [GetFlowAlias](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/GetFlowAlias)
+ [GetFlowVersion](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/GetFlowVersion)
+ [InvokeFlow](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-runtime-2023-12-12/InvokeFlow)
+ [PrepareFlow](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/PrepareFlow)
------
# Entwickeln und Orchestrieren von Anwendungen mit generativer KI mit Amazon Bedrock und Step Functions
Im folgenden Codebeispiel wird gezeigt, wie Anwendungen mit generativer KI mit Amazon Bedrock und Step Functions erstellt und orchestriert werden.
------
#### [ Python ]
**SDK für Python (Boto3)**
Das Szenario „Amazon Bedrock Serverless Prompt Chaining” zeigt, wie [AWS Step Functions](https://docs.aws.amazon.com/step-functions/latest/dg/welcome.html), [Amazon Bedrock](https://docs.aws.amazon.com/bedrock/latest/userguide/what-is-bedrock.html) und [https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html](https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html) zum Erstellen und Orchestrieren komplexer und hoch skalierbarer Serverless-Anwendungen mit generativer KI verwendet werden können. Es enthält die folgenden praktischen Beispiele:
+ Verfassen einer Analyse eines bestimmten Romans für einen Literaturblog. Dieses Beispiel veranschaulicht eine einfache, sequentielle Kette von Prompts.
+ Generieren einer Kurzgeschichte zu einem bestimmten Thema. Dieses Beispiel veranschaulicht, wie die KI eine Liste von zuvor generierten Elementen iterativ verarbeiten kann.
+ Erstellen eines Reiseplans für einen Wochenendurlaub an einem bestimmten Zielort. Dieses Beispiel veranschaulicht, wie mehrere unterschiedliche Prompts parallelisiert werden können.
+ Präsentieren von Filmideen für einen menschlichen Benutzer, der als Filmproduzent fungiert. Dieses Beispiel zeigt, wie derselbe Prompt mit unterschiedlichen Inferenzparametern parallelisiert wird, wie man zu einem vorherigen Schritt in der Kette zurückkehrt und wie menschliche Eingaben in den Workflow einbezogen werden können.
+ Planen einer Mahlzeit auf Grundlage der Zutaten, die der Benutzer zur Hand hat. Dieses Beispiel zeigt, wie Prompt-Chains zwei unterschiedliche KI-Konversationen beinhalten können, bei denen zwei KI-Personas miteinander debattieren, um das Endergebnis zu verbessern.
+ Finden Sie das aktuelle Repository mit den meisten Trends GitHub und fassen Sie es zusammen. Dieses Beispiel veranschaulicht die Verkettung mehrerer KI-Agenten, die mit externen Agenten interagieren. APIs
Den vollständigen Quellcode und Anweisungen zur Einrichtung und Ausführung finden Sie im vollständigen Projekt unter [GitHub](https://github.com/aws-samples/amazon-bedrock-serverless-prompt-chaining).
**In diesem Beispiel verwendete Dienste**
+ Amazon Bedrock
+ Amazon Bedrock Runtime
+ Agenten für Amazon Bedrock
+ Runtime der Agenten für Amazon Bedrock
+ Step Functions
------