Stream ingest information from Kafka to Amazon Bedrock Data Bases utilizing customized connectors

Retrieval Augmented Era (RAG) enhances AI responses by combining the generative AI mannequin’s capabilities with data from exterior information sources, reasonably than relying solely on the mannequin’s built-in data. On this put up, we showcase the customized information connector functionality in Amazon Bedrock Data Bases that makes it simple to construct RAG workflows with customized enter information. By way of this functionality, Amazon Bedrock Data Bases helps the ingestion of streaming information, which suggests builders can add, replace, or delete information of their data base by direct API calls.

Consider the examples of clickstream information, bank card swipes, Web of Issues (IoT) sensor information, log evaluation and commodity costs—the place each present information and historic traits are vital to make a discovered choice. Beforehand, to feed such important information inputs, you needed to first stage it in a supported information supply after which both provoke or schedule a knowledge sync job. Primarily based on the standard and amount of the information, the time to finish this course of assorted. With customized information connectors, you’ll be able to rapidly ingest particular paperwork from customized information sources with out requiring a full sync and ingest streaming information with out the necessity for middleman storage. By avoiding time-consuming full syncs and storage steps, you acquire quicker entry to information, lowered latency, and improved utility efficiency.

Nonetheless, with streaming ingestion utilizing customized connectors, Amazon Bedrock Data Bases processes such streaming information with out utilizing an middleman information supply, making it obtainable nearly instantly. This function chunks and converts enter information into embeddings utilizing your chosen Amazon Bedrock mannequin and shops all the things within the backend vector database. This automation applies to each newly created and current databases, streamlining your workflow so you’ll be able to concentrate on constructing AI functions with out worrying about orchestrating information chunking, embeddings technology, or vector retailer provisioning and indexing. Moreover, this function supplies the flexibility to ingest particular paperwork from customized information sources, all whereas decreasing latency and assuaging operational prices for middleman storage.

Amazon Bedrock

Amazon Bedrock is a completely managed service that provides a selection of high-performing basis fashions (FMs) from main AI firms corresponding to Anthropic, Cohere, Meta, Stability AI, and Amazon by a single API, together with a broad set of capabilities it’s essential to construct generative AI functions with safety, privateness, and accountable AI. Utilizing Amazon Bedrock, you’ll be able to experiment with and consider high FMs on your use case, privately customise them along with your information utilizing strategies corresponding to fine-tuning and RAG, and construct brokers that execute duties utilizing your enterprise techniques and information sources.

Amazon Bedrock Data Bases

Amazon Bedrock Data Bases permits organizations to construct absolutely managed RAG pipelines by augmenting contextual data from personal information sources to ship extra related, correct, and customised responses. With Amazon Bedrock Data Bases, you’ll be able to construct functions which are enriched by the context that’s acquired from querying a data base. It permits a quicker time to product launch by abstracting from the heavy lifting of constructing pipelines and offering you an out-of-the-box RAG answer, thus decreasing the construct time on your utility.

Amazon Bedrock Data Bases customized connector

Amazon Bedrock Data Bases helps customized connectors and the ingestion of streaming information, which suggests you’ll be able to add, replace, or delete information in your data base by direct API calls.

Answer overview: Construct a generative AI inventory value analyzer with RAG

For this put up, we implement a RAG structure with Amazon Bedrock Data Bases utilizing a customized connector and matters constructed with Amazon Managed Streaming for Apache Kafka (Amazon MSK) for a consumer who could also be to know inventory value traits. Amazon MSK is a streaming information service that manages Apache Kafka infrastructure and operations, making it simple to run Apache Kafka functions on Amazon Internet Providers (AWS). The answer permits real-time evaluation of buyer suggestions by vector embeddings and huge language fashions (LLMs).

The next structure diagram has two parts:

Preprocessing streaming information workflow famous in letters on the highest of the diagram:

1. Mimicking streaming enter, add a .csv file with inventory value information into MSK subject
2. Mechanically set off the patron AWS Lambda operate
3. Ingest consumed information right into a data base
4. Data base internally utilizing embeddings mannequin transforms into vector index
5. Data base internally storing vector index into the vector database

Runtime execution throughout consumer queries famous in numerals on the backside of the diagram:

1. Customers question on inventory costs
2. Basis mannequin makes use of the data base to seek for a solution
3. The data base returns with related paperwork
4. Person answered with related reply

Implementation design

The implementation follows these high-level steps:

1. Information supply setup – Configure an MSK subject that streams enter inventory costs
2. Amazon Bedrock Data Bases setup – Create a data base in Amazon Bedrock utilizing the short create a brand new vector retailer choice, which mechanically provisions and units up the vector retailer
3. Information consumption and ingestion – As and when information lands within the MSK subject, set off a Lambda operate that extracts inventory indices, costs, and timestamp data and feeds into the customized connector for Amazon Bedrock Data Bases
4. Take a look at the data base – Consider buyer suggestions evaluation utilizing the data base

Answer walkthrough

To construct a generative AI inventory evaluation device with Amazon Bedrock Data Bases customized connector, use directions within the following sections.

Configure the structure

To do this structure, deploy the AWS CloudFormation template from this GitHub repository in your AWS account. This template deploys the next parts:

1. Useful digital personal clouds (VPCs), subnets, safety teams and AWS Identification and Entry Administration (IAM) roles
2. An MSK cluster internet hosting Apache Kafka enter subject
3. A Lambda operate to eat Apache Kafka subject information
4. An Amazon SageMaker Studio pocket book for granular setup and enablement

Create an Apache Kafka subject

Within the precreated MSK cluster, the required brokers are deployed prepared to be used. The subsequent step is to make use of a SageMaker Studio terminal occasion to connect with the MSK cluster and create the take a look at stream subject. On this step, you comply with the detailed directions which are talked about at Create a subject within the Amazon MSK cluster. The next are the overall steps concerned:

1. Obtain and set up the latest Apache Kafka client
2. Connect with the MSK cluster dealer occasion
3. Create the take a look at stream subject on the dealer occasion

Create a data base in Amazon Bedrock

To create a data base in Amazon Bedrock, comply with these steps:

1. On the Amazon Bedrock console, within the left navigation web page below Builder instruments, select Data Bases.

2. To provoke data base creation, on the Create dropdown menu, select Data Base with vector retailer, as proven within the following screenshot.

3. Within the Present Data Base particulars pane, enter BedrockStreamIngestKnowledgeBase because the Data Base title.
4. Beneath IAM permissions, select the default choice, Create and use a brand new service function, and (non-obligatory) present a Service function title, as proven within the following screenshot.

5. On the Select information supply pane, choose Customized as the information supply the place your dataset is saved
6. Select Subsequent, as proven within the following screenshot

7. On the Configure information supply pane, enter BedrockStreamIngestKBCustomDS because the Information supply title.
8. Beneath Parsing technique, choose Amazon Bedrock default parser and for Chunking technique, select Default chunking. Select Subsequent, as proven within the following screenshot.

9. On the Choose embeddings mannequin and configure vector retailer pane, for Embeddings mannequin, select Titan Textual content Embeddings v2. For Embeddings kind, select Floating-point vector embeddings. For Vector dimensions, choose 1024, as proven within the following screenshot. Be sure to have requested and acquired entry to the chosen FM in Amazon Bedrock. To study extra, check with Add or take away entry to Amazon Bedrock basis fashions.

10. On the Vector database pane, choose Fast create a brand new vector retailer and select the brand new Amazon OpenSearch Serverless choice because the vector retailer.

11. On the following display screen, overview your alternatives. To finalize the setup, select Create.
12. Inside a couple of minutes, the console will show your newly created data base.

Configure AWS Lambda Apache Kafka shopper

Now, utilizing API calls, you configure the patron Lambda operate so it will get triggered as quickly because the enter Apache Kafka subject receives information.

1. Configure the manually created Amazon Bedrock Data Base ID and its customized Information Supply ID as setting variables throughout the Lambda operate. Once you use the pattern pocket book, the referred operate names and IDs will probably be crammed in mechanically.

response = lambda_client.update_function_configuration(
        FunctionName=<Shopper Lambda Operate Identify>,
        Surroundings={
            'Variables': {
                'KBID': <Data Base ID>,
                'DSID': <Information Supply ID>
            }
        }
    )

2. When it’s accomplished, you tie the Lambda shopper operate to pay attention for occasions within the supply Apache Kafka subject:

response = lambda_client.create_event_source_mapping(
        EventSourceArn=<MSK Cluster’s ARN>,
        FunctionName=<Shopper Lambda Operate Identify>,
        StartingPosition='LATEST',
        Enabled=True,
        Matters=['streamtopic']
    )

Evaluate AWS Lambda Apache Kafka shopper

The Apache Kafka shopper Lambda operate reads information from the Apache Kafka subject, decodes it, extracts inventory value data, and ingests it into the Amazon Bedrock data base utilizing the customized connector.

1. Extract the data base ID and the information supply ID:

kb_id = os.environ['KBID']
ds_id = os.environ['DSID']

2. Outline a Python operate to decode enter occasions:

def decode_payload(event_data):
    agg_data_bytes = base64.b64decode(event_data)
    decoded_data = agg_data_bytes.decode(encoding="utf-8") 
    event_payload = json.masses(decoded_data)
    return event_payload   

3. Decode and parse required information on the enter occasion acquired from the Apache Kafka subject. Utilizing them, create a payload to be ingested into the data base:

data = occasion['records']['streamtopic-0']
for rec in data:
        # Every report has separate eventID, and so on.
        event_payload = decode_payload(rec['value'])
        ticker = event_payload['ticker']
        value = event_payload['price']
        timestamp = event_payload['timestamp']
        myuuid = uuid.uuid4()
        payload_ts = datetime.utcfromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')
        payload_string = "At " + payload_ts + " the value of " + ticker + " is " + str(value) + "."

4. Ingest the payload into Amazon Bedrock Data Bases utilizing the customized connector:

response = bedrock_agent_client.ingest_knowledge_base_documents(
                knowledgeBaseId = kb_id,
                dataSourceId = ds_id,
                paperwork= [
                    {
                        'content': {
                            'custom' : {
                                'customDocumentIdentifier': {
                                    'id' : str(myuuid)
                                },
                                'inlineContent' : {
                                    'textContent' : {
                                        'data' : payload_string
                                    },
                                    'type' : 'TEXT'
                                },
                                'sourceType' : 'IN_LINE'
                            },
                            'dataSourceType' : 'CUSTOM'
                        }
                    }
                ]
            )

Testing

Now that the required setup is finished, you set off the workflow by ingesting take a look at information into your Apache Kafka subject hosted with the MSK cluster. For finest outcomes, repeat this part by altering the .csv enter file to point out inventory value improve or lower.

1. Put together the take a look at information. In my case, I had the next information enter as a .csv file with a header.

2. Outline a Python operate to place information to the subject. Use pykafka shopper to ingest information:

def put_to_topic(kafka_host, topic_name, ticker, quantity, timestamp):    
    shopper = KafkaClient(hosts = kafka_host)
    subject = shopper.matters[topic_name]
    payload = {
        'ticker': ticker,
        'value': quantity,
        'timestamp': timestamp
    }
    ret_status = True
    information = json.dumps(payload)
    encoded_message = information.encode("utf-8")
    print(f'Sending ticker information: {ticker}...')
    with subject.get_sync_producer() as producer:
        end result=producer.produce(encoded_message)        
    return ret_status

3. Learn the .csv file and push the data to the subject:

df = pd.read_csv('TestData.csv')
start_test_time = time.time() 
print(datetime.utcfromtimestamp(start_test_time).strftime('%Y-%m-%d %H:%M:%S'))
df = df.reset_index()
for index, row in df.iterrows():
    put_to_topic(BootstrapBrokerString, KafkaTopic, row['ticker'], row['price'], time.time())
end_test_time = time.time()
print(datetime.utcfromtimestamp(end_test_time).strftime('%Y-%m-%d %H:%M:%S'))

Verification

If the information ingestion and subsequent processing is profitable, navigate to the Amazon Bedrock Data Bases information supply web page to examine the uploaded data.

Querying the data base

Inside the Amazon Bedrock Data Bases console, you could have entry to question the ingested information instantly, as proven within the following screenshot.

To do this, choose an Amazon Bedrock FM that you’ve entry to. In my case, I selected Amazon Nova Lite 1.0, as proven within the following screenshot.

When it’s accomplished, the query, “How is ZVZZT trending?”, yields the outcomes primarily based on the ingested information. Word how Amazon Bedrock Data Bases reveals the way it derived the reply, even pointing to the granular information factor from its supply.

Cleanup

To be sure to’re not paying for assets, delete and clear up the assets created.

1. Delete the Amazon Bedrock data base.
2. Delete the mechanically created Amazon OpenSearch Serverless cluster.
3. Delete the mechanically created Amazon Elastic File System (Amazon EFS) shares backing the SageMaker Studio setting.
4. Delete the mechanically created safety teams related to the Amazon EFS share. You would possibly have to take away the inbound and outbound guidelines earlier than they are often deleted.
5. Delete the mechanically created elastic community interfaces connected to the Amazon MSK safety group for Lambda visitors.
6. Delete the mechanically created Amazon Bedrock Data Bases execution IAM function.
7. Cease the kernel situations with Amazon SageMaker Studio.
8. Delete the CloudFormation stack.

Conclusion

On this put up, we confirmed you the way Amazon Bedrock Data Bases helps customized connectors and the ingestion of streaming information, by which builders can add, replace, or delete information of their data base by direct API calls. Amazon Bedrock Data Bases provides absolutely managed, end-to-end RAG workflows to create extremely correct, low-latency, safe, and customized generative AI functions by incorporating contextual data out of your firm’s information sources. With this functionality, you’ll be able to rapidly ingest particular paperwork from customized information sources with out requiring a full sync, and ingest streaming information with out the necessity for middleman storage.

Ship suggestions to AWS re:Post for Amazon Bedrock or by your ordinary AWS contacts, and have interaction with the generative AI builder neighborhood at community.aws.

Concerning the Creator

Prabhakar Chandrasekaran is a Senior Technical Account Supervisor with AWS Enterprise Assist. Prabhakar enjoys serving to clients construct cutting-edge AI/ML options on the cloud. He additionally works with enterprise clients offering proactive steering and operational help, serving to them enhance the worth of their options when utilizing AWS. Prabhakar holds eight AWS and 7 different skilled certifications. With over 22 years {of professional} expertise, Prabhakar was a knowledge engineer and a program chief within the monetary providers area previous to becoming a member of AWS.