Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Skenario untuk Agen Bedrock Amazon menggunakan AWS SDKs
Contoh kode berikut menunjukkan cara menerapkan skenario umum di Amazon Bedrock Agents dengan AWS SDKs. Skenario ini menunjukkan kepada Anda cara menyelesaikan tugas tertentu dengan memanggil beberapa fungsi dalam Amazon Bedrock Agents atau digabungkan dengan yang lain. Layanan AWS Setiap skenario menyertakan tautan ke kode sumber lengkap, di mana Anda dapat menemukan instruksi tentang cara mengatur dan menjalankan kode.
Skenario menargetkan tingkat pengalaman menengah untuk membantu Anda memahami tindakan layanan dalam konteks.
**Topics**
+ [Buat dan panggil alur](bedrock-agent_example_bedrock-agent_GettingStartedWithBedrockFlows_section.md)
+ [Membuat dan memanggil prompt terkelola](bedrock-agent_example_bedrock-agent_GettingStartedWithBedrockPrompts_section.md)
+ [Buat dan panggil agen](bedrock-agent_example_bedrock-agent_GettingStartedWithBedrockAgents_section.md)
+ [Mengatur aplikasi AI generatif dengan Step Functions](bedrock-agent_example_cross_ServerlessPromptChaining_section.md)
# end-to-endContoh yang menunjukkan cara membuat dan menjalankan alur Amazon Bedrock menggunakan SDK AWS
Contoh kode berikut ini menunjukkan cara untuk melakukan:
+ Buat peran eksekusi untuk alur.
+ Buat alur.
+ Terapkan aliran yang sepenuhnya dikonfigurasi.
+ Memanggil alur dengan petunjuk yang disediakan pengguna.
+ Hapus semua sumber daya yang dibuat.
------
#### [ Python ]
**SDK untuk Python (Boto3)**
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/bedrock-agent#code-examples).
Menghasilkan daftar putar musik berdasarkan genre yang ditentukan pengguna dan jumlah lagu.
```
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
```
+ Untuk detail API, lihat topik berikut di *Referensi API AWS SDK untuk 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)
------
Untuk daftar lengkap panduan pengembang AWS SDK dan contoh kode, lihat[Menggunakan Amazon Bedrock dengan SDK AWS](sdk-general-information-section.md). Topik ini juga mencakup informasi tentang memulai dan detail tentang versi SDK sebelumnya.
# end-to-endContoh yang menunjukkan cara membuat dan memanggil prompt terkelola Amazon Bedrock menggunakan SDK AWS
Contoh kode berikut ini menunjukkan cara untuk melakukan:
+ Buat prompt terkelola.
+ Buat versi prompt.
+ Panggil prompt menggunakan versi.
+ Bersihkan sumber daya (opsional).
------
#### [ Python ]
**SDK untuk Python (Boto3)**
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/bedrock-agent#code-examples).
Buat dan panggil prompt terkelola.
```
import argparse
import boto3
import logging
import time
# Now import the modules
from prompt import create_prompt, create_prompt_version, delete_prompt
from run_prompt import invoke_prompt
logging.basicConfig(
level=logging.INFO,
format='%(levelname)s: %(message)s'
)
logger = logging.getLogger(__name__)
def run_scenario(bedrock_client, bedrock_runtime_client, model_id, cleanup=True):
"""
Runs the Amazon Bedrock managed prompt scenario.
Args:
bedrock_client: The Amazon Bedrock Agent client.
bedrock_runtime_client: The Amazon Bedrock Runtime client.
model_id (str): The model ID to use for the prompt.
cleanup (bool): Whether to clean up resources at the end of the scenario.
Returns:
dict: A dictionary containing the created resources.
"""
prompt_id = None
try:
# Step 1: Create a prompt
print("\n=== Step 1: Creating a prompt ===")
prompt_name = f"PlaylistGenerator-{int(time.time())}"
prompt_description = "Playlist generator"
prompt_template = """
Make me a {{genre}} playlist consisting of the following number of songs: {{number}}."""
create_response = create_prompt(
bedrock_client,
prompt_name,
prompt_description,
prompt_template,
model_id
)
prompt_id = create_response['id']
print(f"Created prompt: {prompt_name} with ID: {prompt_id}")
# Create a version of the prompt
print("\n=== Creating a version of the prompt ===")
version_response = create_prompt_version(
bedrock_client,
prompt_id,
description="Initial version of the product description generator"
)
prompt_version_arn = version_response['arn']
prompt_version = version_response['version']
print(f"Created prompt version: {prompt_version}")
print(f"Prompt version ARN: {prompt_version_arn}")
# Step 2: Invoke the prompt directly
print("\n=== Step 2: Invoking the prompt ===")
input_variables = {
"genre": "pop",
"number": "2",
}
# Use the ARN from the create_prompt_version response
result = invoke_prompt(
bedrock_runtime_client,
prompt_version_arn,
input_variables
)
# Display the playlist
print(f"\n{result}")
# Step 3: Clean up resources (optional)
if cleanup:
print("\n=== Step 3: Cleaning up resources ===")
# Delete the prompt
print(f"Deleting prompt {prompt_id}...")
delete_prompt(bedrock_client, prompt_id)
print("Cleanup complete")
else:
print("\n=== Resources were not cleaned up ===")
print(f"Prompt ID: {prompt_id}")
except Exception as e:
logger.exception("Error in scenario: %s", str(e))
# Attempt to clean up if an error occurred and cleanup was requested
if cleanup and prompt_id:
try:
print("\nCleaning up resources after error...")
# Delete the prompt
try:
delete_prompt(bedrock_client, prompt_id)
print("Cleanup after error complete")
except Exception as cleanup_error:
logger.error("Error during cleanup: %s", str(cleanup_error))
except Exception as final_error:
logger.error("Final error during cleanup: %s", str(final_error))
# Re-raise the original exception
raise
def main():
"""
Entry point for the Amazon Bedrock managed prompt scenario.
"""
parser = argparse.ArgumentParser(
description="Run the Amazon Bedrock managed prompt scenario."
)
parser.add_argument(
'--region',
default='us-east-1',
help="The AWS Region to use."
)
parser.add_argument(
'--model-id',
default='anthropic.claude-v2',
help="The model ID to use for the prompt."
)
parser.add_argument(
'--cleanup',
action='store_true',
default=True,
help="Clean up resources at the end of the scenario."
)
parser.add_argument(
'--no-cleanup',
action='store_false',
dest='cleanup',
help="Don't clean up resources at the end of the scenario."
)
args = parser.parse_args()
bedrock_client = boto3.client('bedrock-agent', region_name=args.region)
bedrock_runtime_client = boto3.client('bedrock-runtime', region_name=args.region)
print("=== Amazon Bedrock Managed Prompt Scenario ===")
print(f"Region: {args.region}")
print(f"Model ID: {args.model_id}")
print(f"Cleanup resources: {args.cleanup}")
try:
run_scenario(
bedrock_client,
bedrock_runtime_client,
args.model_id,
args.cleanup
)
except Exception as e:
logger.exception("Error running scenario: %s", str(e))
if __name__ == "__main__":
main()
```
+ Untuk detail API, lihat topik berikut di *Referensi API AWS SDK untuk Python (Boto3)*.
+ [Bercakap](https://docs.aws.amazon.com/goto/boto3/bedrock-runtime-2023-09-30/Converse)
+ [CreatePrompt](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreatePrompt)
+ [CreatePromptVersion](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreatePromptVersion)
+ [DeletePrompt](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/DeletePrompt)
------
Untuk daftar lengkap panduan pengembang AWS SDK dan contoh kode, lihat[Menggunakan Amazon Bedrock dengan SDK AWS](sdk-general-information-section.md). Topik ini juga mencakup informasi tentang memulai dan detail tentang versi SDK sebelumnya.
# end-to-endContoh yang menunjukkan cara membuat dan memanggil Amazon Bedrock Agents menggunakan SDK AWS
Contoh kode berikut ini menunjukkan cara untuk melakukan:
+ Buat peran eksekusi untuk agen.
+ Buat agen dan gunakan versi DRAFT.
+ Buat fungsi Lambda yang mengimplementasikan kemampuan agen.
+ Buat grup tindakan yang menghubungkan agen ke fungsi Lambda.
+ Menyebarkan agen yang sepenuhnya dikonfigurasi.
+ Panggil agen dengan petunjuk yang disediakan pengguna.
+ Hapus semua sumber daya yang dibuat.
------
#### [ Python ]
**SDK untuk Python (Boto3)**
Ada lebih banyak tentang GitHub. Temukan contoh lengkapnya dan pelajari cara mengatur dan menjalankannya di [Repositori Contoh Kode AWS](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/bedrock-agent#code-examples).
Buat dan panggil agen.
```
REGION = "us-east-1"
ROLE_POLICY_NAME = "agent_permissions"
class BedrockAgentScenarioWrapper:
"""Runs a scenario that shows how to get started using Amazon Bedrock Agents."""
def __init__(
self, bedrock_agent_client, runtime_client, lambda_client, iam_resource, postfix
):
self.iam_resource = iam_resource
self.lambda_client = lambda_client
self.bedrock_agent_runtime_client = runtime_client
self.postfix = postfix
self.bedrock_wrapper = BedrockAgentWrapper(bedrock_agent_client)
self.agent = None
self.agent_alias = None
self.agent_role = None
self.prepared_agent_details = None
self.lambda_role = None
self.lambda_function = None
def run_scenario(self):
print("=" * 88)
print("Welcome to the Amazon Bedrock Agents demo.")
print("=" * 88)
# Query input from user
print("Let's start with creating an agent:")
print("-" * 40)
name, foundation_model = self._request_name_and_model_from_user()
print("-" * 40)
# Create an execution role for the agent
self.agent_role = self._create_agent_role(foundation_model)
# Create the agent
self.agent = self._create_agent(name, foundation_model)
# Prepare a DRAFT version of the agent
self.prepared_agent_details = self._prepare_agent()
# Create the agent's Lambda function
self.lambda_function = self._create_lambda_function()
# Configure permissions for the agent to invoke the Lambda function
self._allow_agent_to_invoke_function()
self._let_function_accept_invocations_from_agent()
# Create an action group to connect the agent with the Lambda function
self._create_agent_action_group()
# If the agent has been modified or any components have been added, prepare the agent again
components = [self._get_agent()]
components += self._get_agent_action_groups()
components += self._get_agent_knowledge_bases()
latest_update = max(component["updatedAt"] for component in components)
if latest_update > self.prepared_agent_details["preparedAt"]:
self.prepared_agent_details = self._prepare_agent()
# Create an agent alias
self.agent_alias = self._create_agent_alias()
# Test the agent
self._chat_with_agent(self.agent_alias)
print("=" * 88)
print("Thanks for running the demo!\n")
if q.ask("Do you want to delete the created resources? [y/N] ", q.is_yesno):
self._delete_resources()
print("=" * 88)
print(
"All demo resources have been deleted. Thanks again for running the demo!"
)
else:
self._list_resources()
print("=" * 88)
print("Thanks again for running the demo!")
def _request_name_and_model_from_user(self):
existing_agent_names = [
agent["agentName"] for agent in self.bedrock_wrapper.list_agents()
]
while True:
name = q.ask("Enter an agent name: ", self.is_valid_agent_name)
if name.lower() not in [n.lower() for n in existing_agent_names]:
break
print(
f"Agent {name} conflicts with an existing agent. Please use a different name."
)
models = ["anthropic.claude-instant-v1", "anthropic.claude-v2"]
model_id = models[
q.choose("Which foundation model would you like to use? ", models)
]
return name, model_id
def _create_agent_role(self, model_id):
role_name = f"AmazonBedrockExecutionRoleForAgents_{self.postfix}"
model_arn = f"arn:aws:bedrock:{REGION}::foundation-model/{model_id}*"
print("Creating an an execution role for the agent...")
try:
role = self.iam_resource.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=json.dumps(
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "bedrock.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
),
)
role.Policy(ROLE_POLICY_NAME).put(
PolicyDocument=json.dumps(
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "bedrock:InvokeModel",
"Resource": model_arn,
}
],
}
)
)
except ClientError as e:
logger.error(f"Couldn't create role {role_name}. Here's why: {e}")
raise
return role
def _create_agent(self, name, model_id):
print("Creating the agent...")
instruction = """
You are a friendly chat bot. You have access to a function called that returns
information about the current date and time. When responding with date or time,
please make sure to add the timezone UTC.
"""
agent = self.bedrock_wrapper.create_agent(
agent_name=name,
foundation_model=model_id,
instruction=instruction,
role_arn=self.agent_role.arn,
)
self._wait_for_agent_status(agent["agentId"], "NOT_PREPARED")
return agent
def _prepare_agent(self):
print("Preparing the agent...")
agent_id = self.agent["agentId"]
prepared_agent_details = self.bedrock_wrapper.prepare_agent(agent_id)
self._wait_for_agent_status(agent_id, "PREPARED")
return prepared_agent_details
def _create_lambda_function(self):
print("Creating the Lambda function...")
function_name = f"AmazonBedrockExampleFunction_{self.postfix}"
self.lambda_role = self._create_lambda_role()
try:
deployment_package = self._create_deployment_package(function_name)
lambda_function = self.lambda_client.create_function(
FunctionName=function_name,
Description="Lambda function for Amazon Bedrock example",
Runtime="python3.11",
Role=self.lambda_role.arn,
Handler=f"{function_name}.lambda_handler",
Code={"ZipFile": deployment_package},
Publish=True,
)
waiter = self.lambda_client.get_waiter("function_active_v2")
waiter.wait(FunctionName=function_name)
except ClientError as e:
logger.error(
f"Couldn't create Lambda function {function_name}. Here's why: {e}"
)
raise
return lambda_function
def _create_lambda_role(self):
print("Creating an execution role for the Lambda function...")
role_name = f"AmazonBedrockExecutionRoleForLambda_{self.postfix}"
try:
role = self.iam_resource.create_role(
RoleName=role_name,
AssumeRolePolicyDocument=json.dumps(
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {"Service": "lambda.amazonaws.com"},
"Action": "sts:AssumeRole",
}
],
}
),
)
role.attach_policy(
PolicyArn="arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
)
print(f"Created role {role_name}")
except ClientError as e:
logger.error(f"Couldn't create role {role_name}. Here's why: {e}")
raise
print("Waiting for the execution role to be fully propagated...")
wait(10)
return role
def _allow_agent_to_invoke_function(self):
policy = self.iam_resource.RolePolicy(
self.agent_role.role_name, ROLE_POLICY_NAME
)
doc = policy.policy_document
doc["Statement"].append(
{
"Effect": "Allow",
"Action": "lambda:InvokeFunction",
"Resource": self.lambda_function["FunctionArn"],
}
)
self.agent_role.Policy(ROLE_POLICY_NAME).put(PolicyDocument=json.dumps(doc))
def _let_function_accept_invocations_from_agent(self):
try:
self.lambda_client.add_permission(
FunctionName=self.lambda_function["FunctionName"],
SourceArn=self.agent["agentArn"],
StatementId="BedrockAccess",
Action="lambda:InvokeFunction",
Principal="bedrock.amazonaws.com",
)
except ClientError as e:
logger.error(
f"Couldn't grant Bedrock permission to invoke the Lambda function. Here's why: {e}"
)
raise
def _create_agent_action_group(self):
print("Creating an action group for the agent...")
try:
with open("./scenario_resources/api_schema.yaml") as file:
self.bedrock_wrapper.create_agent_action_group(
name="current_date_and_time",
description="Gets the current date and time.",
agent_id=self.agent["agentId"],
agent_version=self.prepared_agent_details["agentVersion"],
function_arn=self.lambda_function["FunctionArn"],
api_schema=json.dumps(yaml.safe_load(file)),
)
except ClientError as e:
logger.error(f"Couldn't create agent action group. Here's why: {e}")
raise
def _get_agent(self):
return self.bedrock_wrapper.get_agent(self.agent["agentId"])
def _get_agent_action_groups(self):
return self.bedrock_wrapper.list_agent_action_groups(
self.agent["agentId"], self.prepared_agent_details["agentVersion"]
)
def _get_agent_knowledge_bases(self):
return self.bedrock_wrapper.list_agent_knowledge_bases(
self.agent["agentId"], self.prepared_agent_details["agentVersion"]
)
def _create_agent_alias(self):
print("Creating an agent alias...")
agent_alias_name = "test_agent_alias"
agent_alias = self.bedrock_wrapper.create_agent_alias(
agent_alias_name, self.agent["agentId"]
)
self._wait_for_agent_status(self.agent["agentId"], "PREPARED")
return agent_alias
def _wait_for_agent_status(self, agent_id, status):
while self.bedrock_wrapper.get_agent(agent_id)["agentStatus"] != status:
wait(2)
def _chat_with_agent(self, agent_alias):
print("-" * 88)
print("The agent is ready to chat.")
print("Try asking for the date or time. Type 'exit' to quit.")
# Create a unique session ID for the conversation
session_id = uuid.uuid4().hex
while True:
prompt = q.ask("Prompt: ", q.non_empty)
if prompt == "exit":
break
response = asyncio.run(self._invoke_agent(agent_alias, prompt, session_id))
print(f"Agent: {response}")
async def _invoke_agent(self, agent_alias, prompt, session_id):
response = self.bedrock_agent_runtime_client.invoke_agent(
agentId=self.agent["agentId"],
agentAliasId=agent_alias["agentAliasId"],
sessionId=session_id,
inputText=prompt,
)
completion = ""
for event in response.get("completion"):
chunk = event["chunk"]
completion += chunk["bytes"].decode()
return completion
def _delete_resources(self):
if self.agent:
agent_id = self.agent["agentId"]
if self.agent_alias:
agent_alias_id = self.agent_alias["agentAliasId"]
print("Deleting agent alias...")
self.bedrock_wrapper.delete_agent_alias(agent_id, agent_alias_id)
print("Deleting agent...")
agent_status = self.bedrock_wrapper.delete_agent(agent_id)["agentStatus"]
while agent_status == "DELETING":
wait(5)
try:
agent_status = self.bedrock_wrapper.get_agent(
agent_id, log_error=False
)["agentStatus"]
except ClientError as err:
if err.response["Error"]["Code"] == "ResourceNotFoundException":
agent_status = "DELETED"
if self.lambda_function:
name = self.lambda_function["FunctionName"]
print(f"Deleting function '{name}'...")
self.lambda_client.delete_function(FunctionName=name)
if self.agent_role:
print(f"Deleting role '{self.agent_role.role_name}'...")
self.agent_role.Policy(ROLE_POLICY_NAME).delete()
self.agent_role.delete()
if self.lambda_role:
print(f"Deleting role '{self.lambda_role.role_name}'...")
for policy in self.lambda_role.attached_policies.all():
policy.detach_role(RoleName=self.lambda_role.role_name)
self.lambda_role.delete()
def _list_resources(self):
print("-" * 40)
print(f"Here is the list of created resources in '{REGION}'.")
print("Make sure you delete them once you're done to avoid unnecessary costs.")
if self.agent:
print(f"Bedrock Agent: {self.agent['agentName']}")
if self.lambda_function:
print(f"Lambda function: {self.lambda_function['FunctionName']}")
if self.agent_role:
print(f"IAM role: {self.agent_role.role_name}")
if self.lambda_role:
print(f"IAM role: {self.lambda_role.role_name}")
@staticmethod
def is_valid_agent_name(answer):
valid_regex = r"^[a-zA-Z0-9_-]{1,100}$"
return (
answer
if answer and len(answer) <= 100 and re.match(valid_regex, answer)
else None,
"I need a name for the agent, please. Valid characters are a-z, A-Z, 0-9, _ (underscore) and - (hyphen).",
)
@staticmethod
def _create_deployment_package(function_name):
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, "w") as zipped:
zipped.write(
"./scenario_resources/lambda_function.py", f"{function_name}.py"
)
buffer.seek(0)
return buffer.read()
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
postfix = "".join(
random.choice(string.ascii_lowercase + "0123456789") for _ in range(8)
)
scenario = BedrockAgentScenarioWrapper(
bedrock_agent_client=boto3.client(
service_name="bedrock-agent", region_name=REGION
),
runtime_client=boto3.client(
service_name="bedrock-agent-runtime", region_name=REGION
),
lambda_client=boto3.client(service_name="lambda", region_name=REGION),
iam_resource=boto3.resource("iam"),
postfix=postfix,
)
try:
scenario.run_scenario()
except Exception as e:
logging.exception(f"Something went wrong with the demo. Here's what: {e}")
```
+ Untuk detail API, lihat topik berikut di *Referensi API AWS SDK untuk Python (Boto3)*.
+ [CreateAgent](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreateAgent)
+ [CreateAgentActionGroup](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreateAgentActionGroup)
+ [CreateAgentAlias](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/CreateAgentAlias)
+ [DeleteAgent](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/DeleteAgent)
+ [DeleteAgentAlias](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/DeleteAgentAlias)
+ [GetAgent](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/GetAgent)
+ [ListAgentActionGroups](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/ListAgentActionGroups)
+ [ListAgentKnowledgeBases](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/ListAgentKnowledgeBases)
+ [ListAgents](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/ListAgents)
+ [PrepareAgent](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-2023-12-12/PrepareAgent)
------
Untuk daftar lengkap panduan pengembang AWS SDK dan contoh kode, lihat[Menggunakan Amazon Bedrock dengan SDK AWS](sdk-general-information-section.md). Topik ini juga mencakup informasi tentang memulai dan detail tentang versi SDK sebelumnya.
# Membangun dan mengatur aplikasi AI generatif dengan Amazon Bedrock dan Step Functions
Contoh kode berikut menunjukkan cara membangun dan mengatur aplikasi AI generatif dengan Amazon Bedrock dan Step Functions.
------
#### [ Python ]
**SDK untuk Python (Boto3)**
Skenario Amazon Bedrock Serverless Prompt Chaining menunjukkan bagaimana [AWS Step Functions](https://docs.aws.amazon.com/step-functions/latest/dg/welcome.html), [Amazon](https://docs.aws.amazon.com/bedrock/latest/userguide/what-is-bedrock.html) Bedrock, [https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html](https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html) dan dapat digunakan untuk membangun dan mengatur aplikasi AI generatif yang kompleks, tanpa server, dan sangat skalabel. Ini berisi contoh kerja berikut:
+ Tulis analisis novel yang diberikan untuk blog sastra. Contoh ini menggambarkan rantai petunjuk yang sederhana dan berurutan.
+ Hasilkan cerita pendek tentang topik tertentu. Contoh ini menggambarkan bagaimana AI dapat secara iteratif memproses daftar item yang dihasilkan sebelumnya.
+ Buat rencana perjalanan untuk liburan akhir pekan ke tujuan tertentu. Contoh ini menggambarkan cara memparalelkan beberapa prompt yang berbeda.
+ Pitch ide film untuk pengguna manusia yang bertindak sebagai produser film. Contoh ini menggambarkan cara memparalelkan prompt yang sama dengan parameter inferensi yang berbeda, cara mundur ke langkah sebelumnya dalam rantai, dan cara memasukkan input manusia sebagai bagian dari alur kerja.
+ Rencanakan makanan berdasarkan bahan-bahan yang dimiliki pengguna. Contoh ini menggambarkan bagaimana rantai cepat dapat menggabungkan dua percakapan AI yang berbeda, dengan dua persona AI terlibat dalam debat satu sama lain untuk meningkatkan hasil akhir.
+ Temukan dan rangkum repositori tren GitHub tertinggi hari ini. Contoh ini menggambarkan rantai beberapa agen AI yang berinteraksi dengan eksternal. APIs
Untuk kode sumber lengkap dan instruksi untuk menyiapkan dan menjalankan, lihat proyek lengkap di [GitHub](https://github.com/aws-samples/amazon-bedrock-serverless-prompt-chaining).
**Layanan yang digunakan dalam contoh ini**
+ Amazon Bedrock
+ Waktu Jalan Amazon Bedrock
+ Agen Bedrock Amazon
+ Runtime Agen Batuan Dasar Amazon
+ Step Functions
------
Untuk daftar lengkap panduan pengembang AWS SDK dan contoh kode, lihat[Menggunakan Amazon Bedrock dengan SDK AWS](sdk-general-information-section.md). Topik ini juga mencakup informasi tentang memulai dan detail tentang versi SDK sebelumnya.