Stream ingest knowledge 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 info from exterior knowledge sources, quite than relying solely on the mannequin’s built-in information. On this put up, we showcase the customized knowledge connector functionality in Amazon Bedrock Data Bases that makes it easy to construct RAG workflows with customized enter knowledge. By way of this functionality, Amazon Bedrock Data Bases helps the ingestion of streaming knowledge, which implies builders can add, replace, or delete knowledge of their information base by direct API calls.

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

Nonetheless, with streaming ingestion utilizing customized connectors, Amazon Bedrock Data Bases processes such streaming knowledge with out utilizing an middleman knowledge supply, making it obtainable virtually instantly. This characteristic chunks and converts enter knowledge into embeddings utilizing your chosen Amazon Bedrock mannequin and shops every part within the backend vector database. This automation applies to each newly created and present databases, streamlining your workflow so you’ll be able to concentrate on constructing AI purposes with out worrying about orchestrating knowledge chunking, embeddings era, or vector retailer provisioning and indexing. Moreover, this characteristic offers the power to ingest particular paperwork from customized knowledge sources, all whereas lowering latency and assuaging operational prices for middleman storage.

Amazon Bedrock

Amazon Bedrock is a totally managed service that gives a selection of high-performing basis fashions (FMs) from main AI corporations similar to Anthropic, Cohere, Meta, Stability AI, and Amazon by a single API, together with a broad set of capabilities it is advisable to construct generative AI purposes with safety, privateness, and accountable AI. Utilizing Amazon Bedrock, you’ll be able to experiment with and consider high FMs in your use case, privately customise them together with your knowledge utilizing methods similar to fine-tuning and RAG, and construct brokers that execute duties utilizing your enterprise programs and knowledge sources.

Amazon Bedrock Data Bases

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

Amazon Bedrock Data Bases customized connector

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

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

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

The next structure diagram has two parts:

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

1. Mimicking streaming enter, add a .csv file with inventory worth knowledge into MSK matter
2. Routinely set off the buyer AWS Lambda perform
3. Ingest consumed knowledge right into a information 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 information base to seek for a solution
3. The information base returns with related paperwork
4. Person answered with related reply

Implementation design

The implementation follows these high-level steps:

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

Answer walkthrough

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

Configure the structure

To do that 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 Id and Entry Administration (IAM) roles
2. An MSK cluster internet hosting Apache Kafka enter matter
3. A Lambda perform to devour Apache Kafka matter knowledge
4. An Amazon SageMaker Studio pocket book for granular setup and enablement

Create an Apache Kafka matter

Within the precreated MSK cluster, the required brokers are deployed prepared to be used. The following 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 matter. 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 final steps concerned:

1. Obtain and set up the newest Apache Kafka shopper
2. Hook up with the MSK cluster dealer occasion
3. Create the take a look at stream matter on the dealer occasion

Create a information base in Amazon Bedrock

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

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

2. To provoke information 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 identify.
4. Underneath IAM permissions, select the default choice, Create and use a brand new service function, and (non-compulsory) present a Service function identify, as proven within the following screenshot.

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

7. On the Configure knowledge supply pane, enter BedrockStreamIngestKBCustomDS because the Knowledge supply identify.
8. Underneath 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. Ensure you have requested and acquired entry to the chosen FM in Amazon Bedrock. To be taught 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 information base.

Configure AWS Lambda Apache Kafka shopper

Now, utilizing API calls, you configure the buyer Lambda perform so it will get triggered as quickly because the enter Apache Kafka matter receives knowledge.

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

response = lambda_client.update_function_configuration(
        FunctionName=,
        Atmosphere={
            'Variables': {
                'KBID': ,
                'DSID': 
            }
        }
    )

2. When it’s accomplished, you tie the Lambda shopper perform to hear for occasions within the supply Apache Kafka matter:

response = lambda_client.create_event_source_mapping(
        EventSourceArn=,
        FunctionName=,
        StartingPosition='LATEST',
        Enabled=True,
        Matters=['streamtopic']
    )

Overview AWS Lambda Apache Kafka shopper

The Apache Kafka shopper Lambda perform reads knowledge from the Apache Kafka matter, decodes it, extracts inventory worth info, and ingests it into the Amazon Bedrock information base utilizing the customized connector.

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

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

2. Outline a Python perform 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 knowledge on the enter occasion acquired from the Apache Kafka matter. Utilizing them, create a payload to be ingested into the information base:

data = occasion['records']['streamtopic-0']
for rec in data:
        # Every document has separate eventID, and many others.
        event_payload = decode_payload(rec['value'])
        ticker = event_payload['ticker']
        worth = 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 worth of " + ticker + " is " + str(worth) + "."

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 completed, you set off the workflow by ingesting take a look at knowledge into your Apache Kafka matter hosted with the MSK cluster. For finest outcomes, repeat this part by altering the .csv enter file to indicate inventory worth enhance or lower.

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

2. Outline a Python perform to place knowledge to the subject. Use pykafka shopper to ingest knowledge:

def put_to_topic(kafka_host, topic_name, ticker, quantity, timestamp):    
    shopper = KafkaClient(hosts = kafka_host)
    matter = shopper.subjects[topic_name]
    payload = {
        'ticker': ticker,
        'worth': quantity,
        'timestamp': timestamp
    }
    ret_status = True
    knowledge = json.dumps(payload)
    encoded_message = knowledge.encode("utf-8")
    print(f'Sending ticker knowledge: {ticker}...')
    with matter.get_sync_producer() as producer:
        consequence=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 info ingestion and subsequent processing is profitable, navigate to the Amazon Bedrock Data Bases knowledge supply web page to verify the uploaded info.

Querying the information base

Inside the Amazon Bedrock Data Bases console, you may have entry to question the ingested knowledge 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 based mostly on the ingested knowledge. Word how Amazon Bedrock Data Bases exhibits the way it derived the reply, even pointing to the granular knowledge aspect from its supply.

Cleanup

To ensure you’re not paying for sources, delete and clear up the sources created.

1. Delete the Amazon Bedrock information 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 atmosphere.
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 site visitors.
6. Delete the mechanically created Amazon Bedrock Data Bases execution IAM function.
7. Cease the kernel cases with Amazon SageMaker Studio.
8. Delete the CloudFormation stack.

Conclusion

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

Ship suggestions to AWS re:Submit for Amazon Bedrock or by your ordinary AWS contacts, and interact with the generative AI builder group at group.aws.

In regards to the Creator

Prabhakar Chandrasekaran is a Senior Technical Account Supervisor with AWS Enterprise Help. 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.