Diagrams to CDK/Terraform using Claude 3 on Amazon Bedrock

import base64
import json
import os

import boto3
import click
from botocore.exceptions import ClientError

def call_claude_multi_model(bedrock_runtime, model_id, input_text, image, max_tokens):
"""
Streams the response from a multimodal prompt.
Args:
bedrock_runtime: The Amazon Bedrock boto3 client.
model_id (str): The model ID to use.
input_text (str) : The prompt text
image (str) : The path to an image that you want in the prompt.
max_tokens (int) : The maximum number of tokens to generate.
Returns:
None.
"""

with open(image, "rb") as image_file:
encoded_string = base64.b64encode(image_file.read())

body = json.dumps(
{
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": max_tokens,
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": input_text},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": encoded_string.decode("utf-8"),
},
},
],
}
],
}
)

response = bedrock_runtime.invoke_model_with_response_stream(
body=body, modelId=model_id
)

for event in response.get("body"):
chunk = json.loads(event["chunk"]["bytes"])

if chunk["type"] == "content_block_delta":
if chunk["delta"]["type"] == "text_delta":
print(chunk["delta"]["text"], end="")

@click.command()
@click.option("--image_path", prompt="path to image", help="Image you want to parse")
def main(image_path):
"""
Entrypoint for Anthropic Claude Sonnet multimodal prompt example.
"""

model_id = "anthropic.claude-3-sonnet-20240229-v1:0"
input_text = """
You are a AWS solution architect,
The image provided is an architecture diagram. Use two heading to explain below.
1. Explain the technical data flow in detail.
2. Provide cdk typescript code to implement using aws-cdk-lib

Do not:
1. use preambles.
2. make assumptions.
"""
max_tokens = 4000

try:

bedrock_runtime = boto3.client("bedrock-runtime")

call_claude_multi_model(
bedrock_runtime,
model_id,
input_text,
os.path.abspath(image_path),
max_tokens,
)

except ClientError as err:
message = err.response["Error"]["Message"]
logger.error("A client error occurred: %s", message)
print("A client error occured: " + format(message))

if __name__ == "__main__":
main()

$ python src/utils/claude_vision.py --image_path ~/Desktop/s3_extractor.png

Technical Data Flow Explanation

The architecture diagram depicts a serverless data pipeline using AWS services. The data flow can be described as follows:

1. An Amazon S3 bucket is the source of data, which likely contains files or objects that need to be processed.
2. When a new object is uploaded or modified in the S3 bucket, it triggers an AWS Lambda function named "LambdaExtractor."
3. The LambdaExtractor function is responsible for extracting relevant data or events from the input files or objects in the S3 bucket.
4. After processing, the extracted data or events are stored in an Amazon DynamoDB table named "StoreEventsDB."

CDK Typescript Code Implementation

To implement this architecture using the AWS Cloud Development Kit (CDK) and TypeScript, you can use the following code:

```typescript
import * as cdk from 'aws-cdk-lib';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
import * as lambdaEventSources from 'aws-cdk-lib/aws-lambda-event-sources';

const app = new cdk.App();
const stack = new cdk.Stack(app, 'DataPipelineStack');

// Create the S3 bucket
const sourceBucket = new s3.Bucket(stack, 'SourceBucket');

// Create the DynamoDB table
const eventsTable = new dynamodb.Table(stack, 'StoreEventsDB', {
partitionKey: { name: 'id', type: dynamodb.AttributeType.STRING },
billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
});

// Create the Lambda function
const lambdaExtractor = new lambda.Function(stack, 'LambdaExtractor', {
runtime: lambda.Runtime.NODEJS_14_X,
code: lambda.Code.fromAsset('lambda'),
handler: 'hello.handler',
environment: {
EVENTS_TABLE_NAME: eventsTable.tableName,
},
});

// Grant permissions for the Lambda function to access the DynamoDB table
eventsTable.grantWriteData(lambdaExtractor);

// Create the S3 event source and associate it with the Lambda function
lambdaExtractor.addEventSource(
new lambdaEventSources.S3EventSource(sourceBucket, {
events: [s3.EventType.OBJECT_CREATED, s3.EventType.OBJECT_REMOVED],
recursive: true,
})
);
```

In this code:

1. An S3 bucket named "SourceBucket" is created to serve as the source of data.
2. A DynamoDB table named "StoreEventsDB" is created with a partition key "id" of type string.
3. A Lambda function named "LambdaExtractor" is created with the provided source code and environment variables.
4. The Lambda function is granted write permissions to the DynamoDB table.
5. An S3 event source is created and associated with the Lambda function, triggering the function when objects are created or removed from the "SourceBucket."

Note: You need to replace 'lambda/hello.js' to your Lambda function code, and provide the necessary logic in the Lambda function to extract data from the S3 objects and store it in the DynamoDB table.

> cdk init --language=typescript --generate-only
Applying project template app for typescript
# Welcome to your CDK TypeScript project

This is a blank project for CDK development with TypeScript.

The `cdk.json` file tells the CDK Toolkit how to execute your app.

## Useful commands

* `npm run build` compile typescript to js
* `npm run watch` watch for changes and compile
* `npm run test` perform the jest unit tests
* `npx cdk deploy` deploy this stack to your default AWS account/region
* `npx cdk diff` compare deployed stack with current state
* `npx cdk synth` emits the synthesized CloudFormation template

✅ All done!
****************************************************
*** Newer version of CDK is available [2.131.0] ***
*** Upgrade recommended (npm install -g aws-cdk) ***
****************************************************

cat lib/genai-iac-stack.ts
import * as cdk from 'aws-cdk-lib';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as s3 from 'aws-cdk-lib/aws-s3';
import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
import * as lambdaEventSources from 'aws-cdk-lib/aws-lambda-event-sources';
import { Construct } from 'constructs';

export class GenAiIacStack extends cdk.Stack {
constructor(scope: Construct, id: string, props?: cdk.StackProps) {
super(scope, id, props);

// Create the S3 bucket
const sourceBucket = new s3.Bucket(this, 'SourceBucket');

// Create the DynamoDB table
const eventsTable = new dynamodb.Table(this, 'StoreEventsDB', {
partitionKey: { name: 'id', type: dynamodb.AttributeType.STRING },
billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
});

// Create the Lambda function
const lambdaExtractor = new lambda.Function(this, 'LambdaExtractor', {
runtime: lambda.Runtime.NODEJS_20_X,
code: lambda.Code.fromAsset('lambda'),
handler: 'hello.handler',
environment: {
EVENTS_TABLE_NAME: eventsTable.tableName,
},
});

// Grant permissions for the Lambda function to access the DynamoDB table
eventsTable.grantWriteData(lambdaExtractor);

// Create the S3 event source and associate it with the Lambda function
lambdaExtractor.addEventSource(
new lambdaEventSources.S3EventSource(sourceBucket, {
events: [s3.EventType.OBJECT_CREATED, s3.EventType.OBJECT_REMOVED],
})
);
}
}

exports.handler = async function(event) {
console.log("request:", JSON.stringify(event, undefined, 2));
return {
statusCode: 200,
headers: { "Content-Type": "text/plain" },
body: `Hello, CDK! You've hit ${event.path}\n`
};
};