Introduction#

GitHub this note shows

  • Bedrock Python SDK batch and stream
  • Create a simple Flask app
  • JavaScript handle stream response
  • Deploy on Lambda web adapter

Beckrock SDK#

Let call bedrock with batch response

import boto3
import json
from flask import jsonify
client = boto3.client('bedrock-runtime')
def bedrock_batch(topic='chicken soup'):
"""
demo
"""
instruction = f"""
You are a world class writer. Please write a sweet bedtime story about {topic}.
"""
body = json.dumps({
'prompt': f'Human:{instruction}\n\nAssistant:',
'max_tokens_to_sample': 1028,
'temperature': 1,
'top_k': 250,
'top_p': 0.999,
'stop_sequences': ['\n\nHuman:']
})
response = client.invoke_model(
body=body,
contentType="application/json",
accept="*/*",
modelId="anthropic.claude-v2",
)
response_body = json.loads(
response.get('body').read()
)
print(response_body)

Let call becrock with stream response

def bedrock_stream(topic='chicken soup'):
"""
demo
"""
# prompt
instruction = f"""
You are a world class writer. Please write a sweet bedtime story about {topic}.
"""
# request body
body = json.dumps({
'prompt': f'Human:{instruction}\n\nAssistant:',
'max_tokens_to_sample': 1028,
'temperature': 1,
'top_k': 250,
'top_p': 0.999,
'stop_sequences': ['\n\nHuman:']
})
# response
response = client.invoke_model_with_response_stream(
body=body,
contentType="application/json",
accept="*/*",
modelId="anthropic.claude-v2",
)
# parse stream
stream = response.get('body')
if stream:
for event in stream:
print(event)

Flask App#

Create a flask app with project structure as the following

|--static
|--output.css
|--script.js
|--templates
|--bedrock.html
|--index.html
|--app.py
|--Dockerfile
|--requirements.txt

JavaScript Stream Response#

Let create a simple javascript at client side to handle stream response from Flask

const storyOutput = document.getElementById('story-output')
const callBedrock = async () => {
const topic = document.getElementById('topic').value
storyOutput.innerText = 'thinking ...'
console.log('call bedrock request', topic)
if (topic) {
try {
const response = await fetch('/bedrock', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({ topic: topic })
})
const body = await response.json()
console.log(body)
storyOutput.innerText = body.completion
} catch (error) {
console.log(error)
}
}
}
const callBedrockStream = async () => {
console.log('call bedrock request')
try {
const response = await fetch('/bedrock-stream', {
method: 'POST',
headers: {
'Content-Type': 'application/json'
},
body: JSON.stringify({ topic: 'chicken soup' })
})
console.log(response)
const reader = response.body.getReader()
const decoder = new TextDecoder()
while (true) {
const { done, value } = await reader.read()
if (done) {
break
}
const text = decoder.decode(value)
console.log(text)
storyOutput.innerText += text
}
} catch (error) {
console.log(error)
}
}
document
.getElementById('submit-button')
.addEventListener('click', callBedrockStream)

Lambda Web Adapter#

Let create a lambda web adapter for hosting the web. Please note that it supports only batch response

import { Stack, StackProps, aws_iam, aws_lambda } from 'aws-cdk-lib'
import { Effect } from 'aws-cdk-lib/aws-iam'
import { Construct } from 'constructs'
import path = require('path')
export class LambdaWebAdapter extends Stack {
constructor(scope: Construct, id: string, props: StackProps) {
super(scope, id, props)
const func = new aws_lambda.Function(this, 'FlaskWebAdapter', {
functionName: 'FlaskWebAdatper',
code: aws_lambda.EcrImageCode.fromAssetImage(
path.join(__dirname, './../../flask/app/')
),
runtime: aws_lambda.Runtime.FROM_IMAGE,
handler: aws_lambda.Handler.FROM_IMAGE,
memorySize: 1024
})
func.addToRolePolicy(
new aws_iam.PolicyStatement({
effect: Effect.ALLOW,
resources: [
'arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-v2'
],
actions: [
'bedrock:InvokeModel',
'bedrock:InvokeModelWithResponseStream'
]
})
)
}
}

Reference#