Vector Embeddings

Striim enables you to integrate vector embeddings into your streaming applications using the Euclid AI Agent. This agent can be added directly from Flow Designer to generate embeddings from selected data fields and write them to downstream systems. Vector embeddings are commonly stored in vector databases such as Postgres or other supported targets, and Striim provides target adapters for integrating with these systems.

The Euclid AI Agent encapsulates the setup required to connect to an AI model (such as OpenAI or Vertex AI) and generate high-dimensional vector representations from your application data. Once configured, you can use built-in SQL functions to transform and pass selected data to the agent for embedding generation.

You can apply SQL-based transformations to prepare your input data for embedding. To use vector embeddings, first configure the Euclid agent in Flow Designer and specify its parameters. The agent supports both OpenAI and Vertex AI models.

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:

Batch processing can improve performance for use cases where you want to generate vector embeddings from streaming data. Batching aggregates the events and data to be embedded, generates the embeddings using the generateEmbeddingsPerBatch function, and flattens the resulting nested data structure into a single list of embeddings.

When using the Euclid AI agent component in Flow Designer, batching is supported if the selected fields (columns) are of type List<String>. The system will automatically generate a continuous query (CQ) that uses the generateEmbeddingsPerBatch function to perform batch processing.

In the Euclid AI agent configuration panel, you can select one or more fields to embed. When batching is enabled (via selecting a list-type field), the Euclid component configures an internal window, batch CQ, and embedding CQ to optimize the embedding process.

// 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 per batch and aggregate 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;

// Flatten 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;