Striim Platform 5.0 documentation

Striim allows you to configure an embeddings model that you can use to generate vector embeddings. After generating the vector embeddings, we expose the capability for the user to store their embeddings in an appropriate data storage. Vector embeddings are commonly stored in vector databases, such as Pinecone, ChromaDB, and FAISS. We will provide such options to the user in the form of target adapters for your embeddings.

To use vector embeddings within your application, you can create an Embedding Generator object which you can invoke within a built-in function inside a CQ. This object contains the connection and setup details for a single embeddings model you can use to generate vector embeddings, which you can write to supported Striim targets.

By exposing a built-in function to generate embeddings from user data, you have the ability to apply any SQL related transformations to the data columns you would like to create embeddings from. There are two ways in which you can create and configure the embedding generator object:

When creating the embedding generator object, you provide the name of the object, the name of the AI model (see options below), your API key, and any other necessary account details to use the embeddings API.

CREATE OR REPLACE EMBEDDINGGENERATOR OpenAIEmbedder2 USING OpenAI (
modelName: 'text-embedding-ada-002',
apiKey: '**',
'organizationID': 'orgID' // optional
);

CREATE CQ EmbGenCQ
INSERT INTO EmbeddingCDCStream
SELECT putUserData(e, 'embedding', java.util.Arrays.toString(generateEmbeddings("admin.OpenAIEmbedder26",
  TO_STRING(GETDATA(e, "description")))))
FROM sourceCDCStream e;

CREATE OR REPLACE TARGET DeliverEmbeddingChangesToPostgresDB USING DatabaseWriter (
  Tables: 'AIDEMO.PRODUCTS,aidemo.products2
    ColumnMap(product_id=product_id,product_name=product_name,description=description,list_price=
    list_price,embedding=@USERDATA(embedding))',
  Username: 'example',
  Password: 'example',
  BatchPolicy: 'EventCount:100,Interval:2',
  CommitPolicy: 'EventCount:100,Interval:2',
  ConnectionURL: 'jdbc:postgresql://url:port/postgres?stringtype=unspecified'
 ) INPUT FROM EmbeddingCDCStream;

The following are the supported targets for writing vector embeddings and the recommended data types.

The following are the model types available and the connection parameters needed:

The use of batch processing can improve performance for use cases where you want to generate vector embeddings. Batching involves aggregating the events and data to be embedded, generating the embeddings and aggregating into a list using the generateEmbeddingsPerBatch function, and flattening the nested data structure into a single, one-dimensional list for processing by the model.

// Window to batch
CREATE JUMPING WINDOW <window_name> OVER <Stream>
KEEP <window_policy>; 

// Aggregate the events and data to be embedded.
CREATE OR REPLACE CQ <CQ_name>
INSERT INTO <stream>
SELECT list(w) as events, list(TO_STRING(<col_name>)) as data FROM <window_name> w; 

// Generate embeddings and aggregated it into a list.
CREATE CQ <CQ_Name>
INSERT INTO <out_stream>
SELECT makeTupleList(a.events, generateEmbeddingsPerBatch(<namespace>.<object_name>, List<data>)) as objs FROM <int_stream> a; 

// unRoll the list.CREATE CQ <CQ_name>
INSERT INTO <out_stream>
SELECT putUserData(cast(origevent.get(0) as com.webaction.proc.events.WAEvent), 'embedding', java.util.Arrays.toString(cast(origevent.get(1) as java.lang.Float[]))) FROM embOut e, ITERATOR (e.objs, java.util.List) as origevent;