Add opensearch support (LangStream#574)

examples/applications/query-opensearch/.gitignore

@@ -0,0 +1 @@
+java/lib/*

examples/applications/query-opensearch/README.md

@@ -0,0 +1,66 @@
+# Querying a Pinecone Index
+
+This sample application shows how to perform queries against a Pinecone index.
+
+## Prerequisites
+
+Run OpenSearch locally.
+```
+docker network create n1
+docker run --rm -it --network n1 --name opensearch -p 9200:9200 -p 9600:9600 -e "discovery.type=single-node" -e "plugins.security.disabled=true" opensearchproject/opensearch:latest
+```
+
+
+Or if you want to use Amazon AWS OpenSearch Serverless:
+```
+export OPENSEARCH_USERNAME=<aws-access-key-id>
+export OPENSEARCH_PASSWORD=<aws-secret-access-key>
+export OPENSEARCH_HOST=xxxx.<region>.aoss.amazonaws.com
+export OPENSEARCH_REGION=<region>
+```
+
+Note that the you need to create a AWS OpenSearch collection.
+This examples uses both document IDs and vector.
+
+This is not supported by the current Vector Search type collection so you either remove the document IDs or you use the new Vector Search type collection.
+
+## Configure access to the Vector Database
+
+Configure OpenAI access key to generate embeddings:
+
+```
+export OPEN_AI_ACCESS_KEY= ...
+```
+
+
+## Deploy the LangStream application
+
+```
+langstream docker run test -app examples/applications/query-opensearch -i examples/instances/kafka-kubernetes.yaml -s examples/secrets/secrets.yaml --docker-args --network --docker-args n1
+```
+
+or with AWS OpenSearch:
+
+```
+langstream docker run test -app examples/applications/query-opensearch -i examples/instances/kafka-kubernetes.yaml -s examples/secrets/secrets.yaml
+```
+
+## Fill the index
+
+Let's start a produce that sends messages to the vectors-topic:
+
+```
+langstream gateway produce test fill-index -v "My cat eats carrots" 
+langstream gateway produce test fill-index -v "My dog is called Jim"
+```
+
+## Search 
+
+Search via k-NN (k-Nearest Neighbors):
+
+```
+langstream gateway chat test -g chat
+$ > Food
+My cat eats carrots
+```
+

examples/applications/query-opensearch/configuration.yaml

@@ -0,0 +1,35 @@
+#
+#
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+configuration:
+  resources:
+    - type: "open-ai-configuration"
+      name: "OpenAI Azure configuration"
+      configuration:
+        url: "${secrets.open-ai.url}"
+        access-key: "${secrets.open-ai.access-key}"
+        provider: "${secrets.open-ai.provider}"
+    - type: "vector-database"
+      name: "opensearch-datasource"
+      configuration:
+        service: "opensearch"
+        username: "${secrets.opensearch.username}"
+        password: "${secrets.opensearch.password}"
+        host: "${secrets.opensearch.host}"
+        port: "${secrets.opensearch.port}"
+        https: "${secrets.opensearch.https}"
+        region: "${secrets.opensearch.region}"

examples/applications/query-opensearch/gateways.yaml

@@ -0,0 +1,29 @@
+#
+#
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+gateways:
+  - id: fill-index
+    type: produce
+    topic: vectors-topic
+
+  - id: chat
+    type: chat
+    chat-options:
+      headers:
+      - value-from-parameters: sessionId
+      questions-topic: input-topic
+      answers-topic: output-topic

examples/applications/query-opensearch/query.yaml

@@ -0,0 +1,64 @@
+#
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+name: "Index Products on Vector Database"
+topics:
+  - name: "input-topic"
+    creation-mode: create-if-not-exists
+  - name: "output-topic"
+    creation-mode: create-if-not-exists
+errors:
+    on-failure: skip
+pipeline:
+  - name: "convert-to-json"
+    type: "document-to-json"
+    input: "input-topic"
+    configuration:
+      text-field: "question"
+  - name: "compute-embeddings"
+    type: "compute-ai-embeddings"
+    configuration:
+      model: "${secrets.open-ai.embeddings-model}" # This needs to match the name of the model deployment, not the base model
+      embeddings-field: "value.embeddings"
+      text: "{{ value.question }}"
+  - name: "Execute Query"
+    type: "query-vector-db"
+    configuration:
+      datasource: "opensearch-datasource"
+      query: |
+        {
+          "index": "langstream-index",
+          "query": {
+            "knn": {
+              "embeddings": {
+                "vector": ?,
+                "k": 1
+              }
+            }
+          }
+        }
+      fields:
+        - "value.embeddings"
+      output-field: "value.query_result"
+      only-first: true
+  - name: "Format response"
+    type: compute
+    output: "output-topic"
+    configuration:
+      fields:
+        - name: "value"
+          type: STRING
+          expression: "value.query_result.document.content"

examples/applications/query-opensearch/write.yaml

@@ -0,0 +1,74 @@
+#
+# Copyright DataStax, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+name: "Index Products on Vector Database"
+topics:
+  - name: "vectors-topic"
+    creation-mode: create-if-not-exists
+assets:
+  - name: "os-index"
+    asset-type: "opensearch-index"
+    creation-mode: create-if-not-exists
+    config:
+      index-name: "langstream-index"
+      datasource: "opensearch-datasource"
+      settings: |
+        {
+             "index": {
+                   "knn": true,
+                   "knn.algo_param.ef_search": 100
+             }
+         }
+      mappings: |
+        {
+             "properties": {
+                   "content": {
+                         "type": "text"
+                   },
+                   "embeddings": {
+                         "type": "knn_vector",
+                         "dimension": 1536
+                   }
+             }
+         }
+errors:
+    on-failure: skip
+pipeline:
+  - name: "convert-to-json"
+    type: "document-to-json"
+    input: "vectors-topic"
+    configuration:
+      text-field: "document"
+  - name: "compute-embeddings"
+    type: "compute-ai-embeddings"
+    configuration:
+      model: "${secrets.open-ai.embeddings-model}" # This needs to match the name of the model deployment, not the base model
+      embeddings-field: "value.embeddings"
+      text: "{{ value.document }}"
+      batch-size: 10
+      flush-interval: 500
+  - name: "Write to vector db"
+    type: "vector-db-sink"
+    configuration:
+      datasource: "opensearch-datasource"
+      index-name: langstream-index
+      bulk-parameters:
+        refresh: "wait_for"
+      fields:
+        - name: "embeddings"
+          expression: "value.embeddings"
+        - name: "content"
+          expression: "value.document"

examples/secrets/secrets.yaml

@@ -104,4 +104,13 @@ secrets:
       username: "${SOLR_USERNAME:-}"
       password: "${SOLR_PASSWORD:-}"
       host: "${SOLR_HOST:-localhost}"
-      port: "${SOLR_PORT:-8983}"
+      port: "${SOLR_PORT:-8983}"
+  - name: opensearch
+    id: opensearch
+    data:
+      username: "${OPENSEARCH_USERNAME:-admin}"
+      password: "${OPENSEARCH_PASSWORD:-admin}"
+      host: "${OPENSEARCH_HOST:-localhost}"
+      port: "${OPENSEARCH_PORT:-9200}"
+      https: "${OPENSEARCH_HTTPS:-false}"
+      region: "${OPENSEARCH_REGION}"

langstream-agents/langstream-vector-agents/pom.xml

@@ -30,6 +30,17 @@
     <maven.compiler.source>17</maven.compiler.source>
     <maven.compiler.target>17</maven.compiler.target>
   </properties>
+  <dependencyManagement>
+    <dependencies>
+      <dependency>
+        <groupId>software.amazon.awssdk</groupId>
+        <artifactId>bom</artifactId>
+        <version>2.20.162</version>
+        <type>pom</type>
+        <scope>import</scope>
+      </dependency>
+    </dependencies>
+  </dependencyManagement>
   <dependencies>
     <dependency>
       <groupId>${project.groupId}</groupId>
@@ -76,6 +87,15 @@
       <artifactId>solr-solrj</artifactId>
       <version>9.3.0</version>
     </dependency>
+    <dependency>
+      <groupId>org.opensearch.client</groupId>
+      <artifactId>opensearch-java</artifactId>
+      <version>2.6.0</version>
+    </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>opensearch</artifactId>
+    </dependency>
     <dependency>
       <groupId>org.projectlombok</groupId>
       <artifactId>lombok</artifactId>
@@ -110,6 +130,12 @@
       <artifactId>cassandra</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+        <groupId>org.opensearch</groupId>
+        <artifactId>opensearch-testcontainers</artifactId>
+        <version>2.0.0</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
   <build>
     <plugins>