Ada lebih banyak contoh AWS SDK yang tersedia di repo Contoh [SDK AWS Doc](https://github.com/awsdocs/aws-doc-sdk-examples). GitHub
Terjemahan disediakan oleh mesin penerjemah. Jika konten terjemahan yang diberikan bertentangan dengan versi bahasa Inggris aslinya, utamakan versi bahasa Inggris.
# Contoh kode untuk Amazon Bedrock Agents Runtime menggunakan AWS SDKs
Contoh kode berikut menunjukkan cara menggunakan Amazon Bedrock Agents Runtime dengan kit pengembangan AWS perangkat lunak (SDK).
*Dasar-dasar* adalah contoh kode yang menunjukkan kepada Anda bagaimana melakukan operasi penting dalam suatu layanan.
*Tindakan* merupakan kutipan kode dari program yang lebih besar dan harus dijalankan dalam konteks. Sementara tindakan menunjukkan cara memanggil fungsi layanan individual, Anda dapat melihat tindakan dalam konteks dalam skenario terkait.
*Skenario* adalah contoh kode yang menunjukkan kepada Anda bagaimana menyelesaikan tugas tertentu dengan memanggil beberapa fungsi dalam layanan atau dikombinasikan dengan yang lain Layanan AWS.
**Sumber daya lainnya**
+ **[Panduan Pengguna Amazon Bedrock Agents Runtime](https://docs.aws.amazon.com/bedrock/latest/userguide/agents.html)** — Informasi lebih lanjut tentang Runtime Agen Amazon Bedrock.
+ **[Referensi API Runtime Agen Amazon Bedrock](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_Operations_Agents_for_Amazon_Bedrock_Runtime.html)** — Detail tentang semua tindakan Runtime Agen Batuan Dasar Amazon yang tersedia.
+ **[AWS Pusat Pengembang](https://aws.amazon.com/developer/code-examples/?awsf.sdk-code-examples-product=product%23bedrock-agents)** - Contoh kode yang dapat Anda filter berdasarkan kategori atau pencarian teks lengkap.
+ **[AWS Contoh SDK](https://github.com/awsdocs/aws-doc-sdk-examples)** — GitHub repo dengan kode lengkap dalam bahasa pilihan. Termasuk instruksi untuk mengatur dan menjalankan kode.
**Contents**
+ [Hal-hal mendasar](bedrock-agent-runtime_code_examples_basics.md)
+ [Pelajari dasar-dasarnya](bedrock-agent-runtime_example_bedrock-agent-runtime_Scenario_ConverseWithFlow_section.md)
+ [Tindakan](bedrock-agent-runtime_code_examples_actions.md)
+ [`InvokeAgent`](bedrock-agent-runtime_example_bedrock-agent-runtime_InvokeAgent_section.md)
+ [`InvokeFlow`](bedrock-agent-runtime_example_bedrock-agent-runtime_InvokeFlow_section.md)
+ [Skenario](bedrock-agent-runtime_code_examples_scenarios.md)
+ [Buat dan panggil alur](bedrock-agent-runtime_example_bedrock-agent_GettingStartedWithBedrockFlows_section.md)
+ [Mengatur aplikasi AI generatif dengan Step Functions](bedrock-agent-runtime_example_cross_ServerlessPromptChaining_section.md)
# Contoh dasar untuk Amazon Bedrock Agents Runtime menggunakan AWS SDKs
Contoh kode berikut menunjukkan cara menggunakan dasar-dasar Amazon Bedrock Agents Runtime dengan. AWS SDKs
**Contents**
+ [Pelajari dasar-dasarnya](bedrock-agent-runtime_example_bedrock-agent-runtime_Scenario_ConverseWithFlow_section.md)
+ [Tindakan](bedrock-agent-runtime_code_examples_actions.md)
+ [`InvokeAgent`](bedrock-agent-runtime_example_bedrock-agent-runtime_InvokeAgent_section.md)
+ [`InvokeFlow`](bedrock-agent-runtime_example_bedrock-agent-runtime_InvokeFlow_section.md)
# Berkomunikasi dengan aliran Amazon Bedrock
Contoh kode berikut menunjukkan cara menggunakan InvokeFlow untuk berkomunikasi dengan aliran Amazon Bedrock yang menyertakan node agen.
Untuk informasi selengkapnya, [lihat Berbicara dengan alur Amazon Bedrock](https://docs.aws.amazon.com/bedrock/latest/userguide/flows-multi-turn-invocation.html).
------
#### [ 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-runtime#code-examples).
```
"""
Shows how to run an Amazon Bedrock flow with InvokeFlow and handle muli-turn interaction
for a single conversation.
For more information, see https://docs.aws.amazon.com/bedrock/latest/userguide/flows-multi-turn-invocation.html.
"""
import logging
import boto3
import botocore
import botocore.exceptions
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def invoke_flow(client, flow_id, flow_alias_id, input_data, execution_id):
"""
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.
execution_id: Execution ID for continuing a flow. Use the value None on first run.
Returns:
Dict containing flow_complete status, input_required info, and execution_id
"""
response = None
request_params = None
if execution_id is None:
# Don't pass execution ID for first run.
request_params = {
"flowIdentifier": flow_id,
"flowAliasIdentifier": flow_alias_id,
"inputs": [input_data],
"enableTrace": True
}
else:
request_params = {
"flowIdentifier": flow_id,
"flowAliasIdentifier": flow_alias_id,
"executionId": execution_id,
"inputs": [input_data],
"enableTrace": True
}
response = client.invoke_flow(**request_params)
if "executionId" not in request_params:
execution_id = response['executionId']
input_required = None
flow_status = ""
# Process the streaming response
for event in response['responseStream']:
# Check if flow is complete.
if 'flowCompletionEvent' in event:
flow_status = event['flowCompletionEvent']['completionReason']
# Check if more input us needed from user.
elif 'flowMultiTurnInputRequestEvent' in event:
input_required = event
# Print the model output.
elif 'flowOutputEvent' in event:
print(event['flowOutputEvent']['content']['document'])
# Log trace events.
elif 'flowTraceEvent' in event:
logger.info("Flow trace: %s", event['flowTraceEvent'])
return {
"flow_status": flow_status,
"input_required": input_required,
"execution_id": execution_id
}
def converse_with_flow(bedrock_agent_client, flow_id, flow_alias_id):
"""
Run a conversation with the supplied 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.
"""
flow_execution_id = None
finished = False
# Get the intial prompt from the user.
user_input = input("Enter input: ")
# Use prompt to create input data.
flow_input_data = {
"content": {
"document": user_input
},
"nodeName": "FlowInputNode",
"nodeOutputName": "document"
}
try:
while not finished:
# Invoke the flow until successfully finished.
result = invoke_flow(
bedrock_agent_client, flow_id, flow_alias_id, flow_input_data, flow_execution_id)
status = result['flow_status']
flow_execution_id = result['execution_id']
more_input = result['input_required']
if status == "INPUT_REQUIRED":
# The flow needs more information from the user.
logger.info("The flow %s requires more input", flow_id)
user_input = input(
more_input['flowMultiTurnInputRequestEvent']['content']['document'] + ": ")
flow_input_data = {
"content": {
"document": user_input
},
"nodeName": more_input['flowMultiTurnInputRequestEvent']['nodeName'],
"nodeInputName": "agentInputText"
}
elif status == "SUCCESS":
# The flow completed successfully.
finished = True
logger.info("The flow %s successfully completed.", flow_id)
except botocore.exceptions.ClientError as e:
print(f"Client error: {str(e)}")
logger.error("Client error: %s", {str(e)})
except Exception as e:
print(f"An error occurred: {str(e)}")
logger.error("An error occurred: %s", {str(e)})
logger.error("Error type: %s", {type(e)})
def main():
"""
Main entry point for the script.
"""
# Replace these with your actual flow ID and flow alias ID.
FLOW_ID = 'YOUR_FLOW_ID'
FLOW_ALIAS_ID = 'YOUR_FLOW_ALIAS_ID'
logger.info("Starting conversation with FLOW: %s ID: %s",
FLOW_ID, FLOW_ALIAS_ID)
# Get the Bedrock agent runtime client.
session = boto3.Session(profile_name='default')
bedrock_agent_client = session.client('bedrock-agent-runtime')
# Start the conversation.
converse_with_flow(bedrock_agent_client, FLOW_ID, FLOW_ALIAS_ID)
logger.info("Conversation with FLOW: %s ID: %s finished",
FLOW_ID, FLOW_ALIAS_ID)
if __name__ == "__main__":
main()
```
+ Untuk detail API, lihat [InvokeFlow](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-runtime-2023-12-12/InvokeFlow)di *AWS SDK for Python (Boto3) Referensi* API.
------
# Tindakan untuk Amazon Bedrock Agents Runtime menggunakan AWS SDKs
Contoh kode berikut menunjukkan cara melakukan tindakan Runtime Amazon Bedrock Agents individual dengan. AWS SDKs Setiap contoh menyertakan tautan ke GitHub, di mana Anda dapat menemukan instruksi untuk mengatur dan menjalankan kode.
Kutipan ini memanggil Amazon Bedrock Agents Runtime API dan merupakan kutipan kode dari program yang lebih besar yang harus dijalankan dalam konteks. Anda dapat melihat tindakan dalam konteks di[Skenario untuk Amazon Bedrock Agents Runtime menggunakan AWS SDKs](bedrock-agent-runtime_code_examples_scenarios.md).
Contoh berikut hanya mencakup tindakan yang paling umum digunakan. Untuk daftar lengkapnya, lihat Referensi [API Amazon Bedrock Agents Runtime](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_Operations_Agents_for_Amazon_Bedrock_Runtime.html).
**Topics**
+ [`InvokeAgent`](bedrock-agent-runtime_example_bedrock-agent-runtime_InvokeAgent_section.md)
+ [`InvokeFlow`](bedrock-agent-runtime_example_bedrock-agent-runtime_InvokeFlow_section.md)
# Gunakan `InvokeAgent` dengan AWS SDK
Contoh kode berikut menunjukkan cara menggunakan`InvokeAgent`.
------
#### [ JavaScript ]
**SDK untuk JavaScript (v3)**
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/javascriptv3/example_code/bedrock-agent-runtime#code-examples).
```
import {
BedrockAgentRuntimeClient,
InvokeAgentCommand,
} from "@aws-sdk/client-bedrock-agent-runtime";
/**
* @typedef {Object} ResponseBody
* @property {string} completion
*/
/**
* Invokes a Bedrock agent to run an inference using the input
* provided in the request body.
*
* @param {string} prompt - The prompt that you want the Agent to complete.
* @param {string} sessionId - An arbitrary identifier for the session.
*/
export const invokeBedrockAgent = async (prompt, sessionId) => {
const client = new BedrockAgentRuntimeClient({ region: "us-east-1" });
// const client = new BedrockAgentRuntimeClient({
// region: "us-east-1",
// credentials: {
// accessKeyId: "accessKeyId", // permission to invoke agent
// secretAccessKey: "accessKeySecret",
// },
// });
const agentId = "AJBHXXILZN";
const agentAliasId = "AVKP1ITZAA";
const command = new InvokeAgentCommand({
agentId,
agentAliasId,
sessionId,
inputText: prompt,
});
try {
let completion = "";
const response = await client.send(command);
if (response.completion === undefined) {
throw new Error("Completion is undefined");
}
for await (const chunkEvent of response.completion) {
const chunk = chunkEvent.chunk;
console.log(chunk);
const decodedResponse = new TextDecoder("utf-8").decode(chunk.bytes);
completion += decodedResponse;
}
return { sessionId: sessionId, completion };
} catch (err) {
console.error(err);
}
};
// Call function if run directly
import { fileURLToPath } from "node:url";
if (process.argv[1] === fileURLToPath(import.meta.url)) {
const result = await invokeBedrockAgent("I need help.", "123");
console.log(result);
}
```
+ Untuk detail API, lihat [InvokeAgent](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/bedrock-agent-runtime/command/InvokeAgentCommand)di *Referensi AWS SDK untuk JavaScript API*.
------
#### [ 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-runtime#code-examples).
Meminta agen.
```
def invoke_agent(self, agent_id, agent_alias_id, session_id, prompt):
"""
Sends a prompt for the agent to process and respond to.
:param agent_id: The unique identifier of the agent to use.
:param agent_alias_id: The alias of the agent to use.
:param session_id: The unique identifier of the session. Use the same value across requests
to continue the same conversation.
:param prompt: The prompt that you want Claude to complete.
:return: Inference response from the model.
"""
try:
# Note: The execution time depends on the foundation model, complexity of the agent,
# and the length of the prompt. In some cases, it can take up to a minute or more to
# generate a response.
response = self.agents_runtime_client.invoke_agent(
agentId=agent_id,
agentAliasId=agent_alias_id,
sessionId=session_id,
inputText=prompt,
)
completion = ""
for event in response.get("completion"):
chunk = event["chunk"]
completion = completion + chunk["bytes"].decode()
except ClientError as e:
logger.error(f"Couldn't invoke agent. {e}")
raise
return completion
```
+ Untuk detail API, lihat [InvokeAgent](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-runtime-2023-12-12/InvokeAgent)di *AWS SDK for Python (Boto3) Referensi* API.
------
#### [ Rust ]
**SDK for Rust**
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/rustv1/examples/bedrock-agent-runtime#code-examples).
```
use aws_config::{BehaviorVersion, SdkConfig};
use aws_sdk_bedrockagentruntime::{
self as bedrockagentruntime,
types::{error::ResponseStreamError, ResponseStream},
};
#[allow(unused_imports)]
use mockall::automock;
const BEDROCK_AGENT_ID: &str = "AJBHXXILZN";
const BEDROCK_AGENT_ALIAS_ID: &str = "AVKP1ITZAA";
const BEDROCK_AGENT_REGION: &str = "us-east-1";
#[cfg(not(test))]
pub use EventReceiverImpl as EventReceiver;
#[cfg(test)]
pub use MockEventReceiverImpl as EventReceiver;
pub struct EventReceiverImpl {
inner: aws_sdk_bedrockagentruntime::primitives::event_stream::EventReceiver<
ResponseStream,
ResponseStreamError,
>,
}
#[cfg_attr(test, automock)]
impl EventReceiverImpl {
#[allow(dead_code)]
pub fn new(
inner: aws_sdk_bedrockagentruntime::primitives::event_stream::EventReceiver<
ResponseStream,
ResponseStreamError,
>,
) -> Self {
Self { inner }
}
pub async fn recv(
&mut self,
) -> Result<
Option,
aws_sdk_bedrockagentruntime::error::SdkError<
ResponseStreamError,
aws_smithy_types::event_stream::RawMessage,
>,
> {
self.inner.recv().await
}
}
#[tokio::main]
async fn main() -> Result<(), Box> {
let result = invoke_bedrock_agent("I need help.".to_string(), "123".to_string()).await?;
println!("{}", result);
Ok(())
}
async fn invoke_bedrock_agent(
prompt: String,
session_id: String,
) -> Result {
let sdk_config: SdkConfig = aws_config::defaults(BehaviorVersion::latest())
.region(BEDROCK_AGENT_REGION)
.load()
.await;
let bedrock_client = bedrockagentruntime::Client::new(&sdk_config);
let command_builder = bedrock_client
.invoke_agent()
.agent_id(BEDROCK_AGENT_ID)
.agent_alias_id(BEDROCK_AGENT_ALIAS_ID)
.session_id(session_id)
.input_text(prompt);
let response = command_builder.send().await?;
let response_stream = response.completion;
let event_receiver = EventReceiver::new(response_stream);
process_agent_response_stream(event_receiver).await
}
async fn process_agent_response_stream(
mut event_receiver: EventReceiver,
) -> Result {
let mut full_agent_text_response = String::new();
while let Some(event_result) = event_receiver.recv().await? {
match event_result {
ResponseStream::Chunk(chunk) => {
if let Some(bytes) = chunk.bytes {
match String::from_utf8(bytes.into_inner()) {
Ok(text_chunk) => {
full_agent_text_response.push_str(&text_chunk);
}
Err(e) => {
eprintln!("UTF-8 decoding error for chunk: {}", e);
}
}
}
}
_ => {
panic!("received an unhandled event type from Bedrock stream",);
}
}
}
Ok(full_agent_text_response)
}
#[cfg(test)]
mod test {
use super::*;
#[tokio::test]
async fn test_process_agent_response_stream() {
let mut mock = MockEventReceiverImpl::default();
mock.expect_recv().times(1).returning(|| {
Ok(Some(
aws_sdk_bedrockagentruntime::types::ResponseStream::Chunk(
aws_sdk_bedrockagentruntime::types::PayloadPart::builder()
.set_bytes(Some(aws_smithy_types::Blob::new(vec![
116, 101, 115, 116, 32, 99, 111, 109, 112, 108, 101, 116, 105, 111, 110,
])))
.build(),
),
))
});
// end the stream
mock.expect_recv().times(1).returning(|| Ok(None));
let response = process_agent_response_stream(mock).await.unwrap();
assert_eq!("test completion", response);
}
#[tokio::test]
#[should_panic(expected = "received an unhandled event type from Bedrock stream")]
async fn test_process_agent_response_stream_error() {
let mut mock = MockEventReceiverImpl::default();
mock.expect_recv().times(1).returning(|| {
Ok(Some(
aws_sdk_bedrockagentruntime::types::ResponseStream::Trace(
aws_sdk_bedrockagentruntime::types::TracePart::builder().build(),
),
))
});
let _ = process_agent_response_stream(mock).await.unwrap();
}
}
```
+ Untuk detail API, lihat [InvokeAgent](https://docs.rs/aws-sdk-bedrockagentruntime/latest/aws_sdk_bedrockagentruntime/client/struct.Client.html#method.invoke_agent)*referensi AWS SDK for Rust API*.
------
#### [ SAP ABAP ]
**SDK for SAP ABAP**
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/sap-abap/services/bdz#code-examples).
```
DATA(lo_result) = lo_bdz->invokeagent(
iv_agentid = iv_agentid
iv_agentaliasid = iv_agentaliasid
iv_enabletrace = abap_true
iv_sessionid = CONV #( cl_system_uuid=>create_uuid_c26_static( ) )
iv_inputtext = |Let's play "rock, paper, scissors". I choose rock.| ).
DATA(lo_stream) = lo_result->get_completion( ).
TRY.
" loop while there are still events in the stream
WHILE lo_stream->/aws1/if_rt_stream_reader~data_available( ) = abap_true.
DATA(lo_evt) = lo_stream->read( ).
" each /AWS1/CL_BDZRESPONSESTREAM_EV event contains exactly one member
" all others are INITIAL. For each event, process the non-initial
" member if desired
IF lo_evt->get_chunk( ) IS NOT INITIAL.
" Process a Chunk event
DATA(lv_xstr) = lo_evt->get_chunk( )->get_bytes( ).
DATA(lv_answer) = /aws1/cl_rt_util=>xstring_to_string( lv_xstr ).
" the answer says something like "I chose paper, so you lost"
ELSEIF lo_evt->get_files( ) IS NOT INITIAL.
" process a Files event if desired
ELSEIF lo_evt->get_returncontrol( ) IS NOT INITIAL.
" process a ReturnControl event if desired
ELSEIF lo_evt->get_trace( ) IS NOT INITIAL.
" process a Trace event if desired
ENDIF.
ENDWHILE.
" the stream of events can possibly contain an exception
" which will be raised to break the loop
" catch /AWS1/CX_BDZACCESSDENIEDEX.
" catch /AWS1/CX_BDZINTERNALSERVEREX.
" catch /AWS1/CX_BDZMODELNOTREADYEX.
" catch /AWS1/CX_BDZVALIDATIONEX.
" catch /AWS1/CX_BDZTHROTTLINGEX.
" catch /AWS1/CX_BDZDEPENDENCYFAILEDEX.
" catch /AWS1/CX_BDZBADGATEWAYEX.
" catch /AWS1/CX_BDZRESOURCENOTFOUNDEX.
" catch /AWS1/CX_BDZSERVICEQUOTAEXCDEX.
" catch /AWS1/CX_BDZCONFLICTEXCEPTION.
ENDTRY.
```
+ Untuk detail API, lihat [InvokeAgent](https://docs.aws.amazon.com/sdk-for-sap-abap/v1/api/latest/index.html)di *AWS SDK untuk referensi SAP ABAP* API.
------
# Gunakan `InvokeFlow` dengan AWS SDK
Contoh kode berikut menunjukkan cara menggunakan`InvokeFlow`.
Contoh tindakan adalah kutipan kode dari program yang lebih besar dan harus dijalankan dalam konteks. Anda dapat melihat tindakan ini dalam konteks dalam contoh kode berikut:
+ [Pelajari dasar-dasarnya](bedrock-agent-runtime_example_bedrock-agent-runtime_Scenario_ConverseWithFlow_section.md)
+ [Buat dan panggil alur](bedrock-agent-runtime_example_bedrock-agent_GettingStartedWithBedrockFlows_section.md)
------
#### [ JavaScript ]
**SDK untuk JavaScript (v3)**
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/javascriptv3/example_code/bedrock-agent-runtime#code-examples).
```
import { fileURLToPath } from "node:url";
import {
BedrockAgentRuntimeClient,
InvokeFlowCommand,
} from "@aws-sdk/client-bedrock-agent-runtime";
/**
* Invokes an alias of a flow to run the inputs that you specify and return
* the output of each node as a stream.
*
* @param {{
* flowIdentifier: string,
* flowAliasIdentifier: string,
* prompt?: string,
* region?: string
* }} options
* @returns {Promise} An object containing information about the output from flow invocation.
*/
export const invokeBedrockFlow = async ({
flowIdentifier,
flowAliasIdentifier,
prompt = "Hi, how are you?",
region = "us-east-1",
}) => {
const client = new BedrockAgentRuntimeClient({ region });
const command = new InvokeFlowCommand({
flowIdentifier,
flowAliasIdentifier,
inputs: [
{
content: {
document: prompt,
},
nodeName: "FlowInputNode",
nodeOutputName: "document",
},
],
});
let flowResponse = {};
const response = await client.send(command);
for await (const chunkEvent of response.responseStream) {
const { flowOutputEvent, flowCompletionEvent } = chunkEvent;
if (flowOutputEvent) {
flowResponse = { ...flowResponse, ...flowOutputEvent };
console.log("Flow output event:", flowOutputEvent);
} else if (flowCompletionEvent) {
flowResponse = { ...flowResponse, ...flowCompletionEvent };
console.log("Flow completion event:", flowCompletionEvent);
}
}
return flowResponse;
};
// Call function if run directly
import { parseArgs } from "node:util";
import {
isMain,
validateArgs,
} from "@aws-doc-sdk-examples/lib/utils/util-node.js";
const loadArgs = () => {
const options = {
flowIdentifier: {
type: "string",
required: true,
},
flowAliasIdentifier: {
type: "string",
required: true,
},
prompt: {
type: "string",
},
region: {
type: "string",
},
};
const results = parseArgs({ options });
const { errors } = validateArgs({ options }, results);
return { errors, results };
};
if (isMain(import.meta.url)) {
const { errors, results } = loadArgs();
if (!errors) {
invokeBedrockFlow(results.values);
} else {
console.error(errors.join("\n"));
}
}
```
+ Untuk detail API, lihat [InvokeFlow](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/client/bedrock-agent-runtime/command/InvokeFlowCommand)di *Referensi AWS SDK untuk JavaScript API*.
------
#### [ 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-runtime#code-examples).
Memanggil aliran.
```
def invoke_flow(self, flow_id, flow_alias_id, input_data, execution_id):
"""
Invoke an Amazon Bedrock flow and handle the response stream.
Args:
param flow_id: The ID of the flow to invoke.
param flow_alias_id: The alias ID of the flow.
param input_data: Input data for the flow.
param execution_id: Execution ID for continuing a flow. Use the value None on first run.
Return: Response from the flow.
"""
try:
request_params = None
if execution_id is None:
# Don't pass execution ID for first run.
request_params = {
"flowIdentifier": flow_id,
"flowAliasIdentifier": flow_alias_id,
"inputs": input_data,
"enableTrace": True
}
else:
request_params = {
"flowIdentifier": flow_id,
"flowAliasIdentifier": flow_alias_id,
"executionId": execution_id,
"inputs": input_data,
"enableTrace": True
}
response = self.agents_runtime_client.invoke_flow(**request_params)
if "executionId" not in request_params:
execution_id = response['executionId']
result = ""
# Get the streaming response
for event in response['responseStream']:
result = result + str(event) + '\n'
print(result)
except ClientError as e:
logger.error("Couldn't invoke flow %s.", {e})
raise
return result
```
+ Untuk detail API, lihat [InvokeFlow](https://docs.aws.amazon.com/goto/boto3/bedrock-agent-runtime-2023-12-12/InvokeFlow)di *AWS SDK for Python (Boto3) Referensi* API.
------
# Skenario untuk Amazon Bedrock Agents Runtime menggunakan AWS SDKs
Contoh kode berikut menunjukkan cara menerapkan skenario umum di Amazon Bedrock Agents Runtime dengan. AWS SDKs Skenario ini menunjukkan kepada Anda cara menyelesaikan tugas tertentu dengan memanggil beberapa fungsi dalam Amazon Bedrock Agents Runtime 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-runtime_example_bedrock-agent_GettingStartedWithBedrockFlows_section.md)
+ [Mengatur aplikasi AI generatif dengan Step Functions](bedrock-agent-runtime_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)
------
# 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
+ Runtime Amazon Bedrock
+ Agen Batuan Dasar Amazon
+ Runtime Agen Batuan Dasar Amazon
+ Step Functions
------