[Feature][Integration][Java] add ElasticsearchVectorStore in Java

api/src/main/java/org/apache/flink/agents/api/annotation/VectorStore.java

@@ -0,0 +1,36 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.agents.api.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Annotation to mark a field or method as a vector store resource that should be managed by the
+ * agent plan.
+ *
+ * <p>Fields annotated with @VectorStore will be scanned during agent plan creation and
+ * corresponding resource providers will be created to manage the vector store instances. Methods
+ * annotated with @VectorStore must be static and will be wrapped as vector store resources.
+ */
+@Target({ElementType.FIELD, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface VectorStore {}

api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalRequestEvent.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.agents.api.event;
+
+import org.apache.flink.agents.api.Event;
+
+/** Event representing a request for context retrieval. */
+public class ContextRetrievalRequestEvent extends Event {
+
+    private static final int DEFAULT_MAX_RESULTS = 3;
+
+    private final String query;
+    private final String vectorStore;
+    private final int maxResults;
+
+    public ContextRetrievalRequestEvent(String query, String vectorStore) {
+        this.query = query;
+        this.vectorStore = vectorStore;
+        this.maxResults = DEFAULT_MAX_RESULTS;
+    }
+
+    public ContextRetrievalRequestEvent(String query, String vectorStore, int maxResults) {
+        this.query = query;
+        this.vectorStore = vectorStore;
+        this.maxResults = maxResults;
+    }
+
+    public String getQuery() {
+        return query;
+    }
+
+    public String getVectorStore() {
+        return vectorStore;
+    }
+
+    public int getMaxResults() {
+        return maxResults;
+    }
+
+    @Override
+    public String toString() {
+        return "ContextRetrievalRequestEvent{"
+                + "query='"
+                + query
+                + '\''
+                + ", vectorStore='"
+                + vectorStore
+                + '\''
+                + ", maxResults="
+                + maxResults
+                + '}';
+    }
+}

api/src/main/java/org/apache/flink/agents/api/event/ContextRetrievalResponseEvent.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.agents.api.event;
+
+import org.apache.flink.agents.api.Event;
+import org.apache.flink.agents.api.vectorstores.Document;
+
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Event representing retrieved context results.
+ *
+ * @param <ContentT>
+ */
+public class ContextRetrievalResponseEvent<ContentT> extends Event {
+
+    private final UUID requestId;
+    private final String query;
+    private final List<Document<ContentT>> documents;
+
+    public ContextRetrievalResponseEvent(
+            UUID requestId, String query, List<Document<ContentT>> documents) {
+        this.requestId = requestId;
+        this.query = query;
+        this.documents = documents;
+    }
+
+    public UUID getRequestId() {
+        return requestId;
+    }
+
+    public String getQuery() {
+        return query;
+    }
+
+    public List<Document<ContentT>> getDocuments() {
+        return documents;
+    }
+
+    @Override
+    public String toString() {
+        return "ContextRetrievalResponseEvent{"
+                + "requestId="
+                + requestId
+                + ", query='"
+                + query
+                + '\''
+                + ", documents="
+                + documents
+                + '}';
+    }
+}

api/src/main/java/org/apache/flink/agents/api/vectorstores/BaseVectorStore.java

@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.agents.api.vectorstores;
+
+import org.apache.flink.agents.api.embedding.model.BaseEmbeddingModelSetup;
+import org.apache.flink.agents.api.resource.Resource;
+import org.apache.flink.agents.api.resource.ResourceDescriptor;
+import org.apache.flink.agents.api.resource.ResourceType;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.BiFunction;
+
+/**
+ * Base abstract class for vector store. Provides vector store functionality that integrates
+ * embedding models for text-based semantic search. Handles both connection management and embedding
+ * generation internally.
+ *
+ * @param <ContentT> The type of content stored in the vector store documents
+ */
+public abstract class BaseVectorStore<ContentT> extends Resource {
+
+    /** Name of the embedding model resource to use. */
+    protected final String embeddingModel;
+
+    public BaseVectorStore(
+            ResourceDescriptor descriptor, BiFunction<String, ResourceType, Resource> getResource) {
+        super(descriptor, getResource);
+        this.embeddingModel = descriptor.getArgument("embedding_model");
+    }
+
+    @Override
+    public ResourceType getResourceType() {
+        return ResourceType.VECTOR_STORE;
+    }
+
+    /**
+     * Returns vector store setup settings passed to connection. These parameters are merged with
+     * query-specific parameters when performing vector search operations.
+     *
+     * @return A map containing the store configuration parameters
+     */
+    public abstract Map<String, Object> getStoreKwargs();
+
+    /**
+     * Performs vector search using structured query object. Converts text query to embeddings and
+     * returns structured query result.
+     *
+     * @param query VectorStoreQuery object containing query parameters
+     * @return VectorStoreQueryResult containing the retrieved documents
+     */
+    public VectorStoreQueryResult<ContentT> query(VectorStoreQuery query) {
+        final BaseEmbeddingModelSetup embeddingModel =
+                (BaseEmbeddingModelSetup)
+                        this.getResource.apply(this.embeddingModel, ResourceType.EMBEDDING_MODEL);
+
+        // TODO
+        // for now, we don't need to use additional parameters.
+        final float[] queryEmbedding = embeddingModel.embed(query.getQueryText(), Map.of());
+
+        final Map<String, Object> storeKwargs = this.getStoreKwargs();
+        storeKwargs.putAll(query.getExtraArgs());
+
+        final List<Document<ContentT>> documents =
+                this.queryEmbedding(queryEmbedding, query.getLimit(), storeKwargs);
+
+        return new VectorStoreQueryResult<>(documents);
+    }
+
+    /**
+     * Performs vector search using a pre-computed embedding.
+     *
+     * @param embedding The embedding vector to search with
+     * @param limit Maximum number of results to return
+     * @param args Additional arguments for the vector search
+     * @return List of documents matching the query embedding
+     */
+    public abstract List<Document<ContentT>> queryEmbedding(
+            float[] embedding, int limit, Map<String, Object> args);
+}

api/src/main/java/org/apache/flink/agents/api/vectorstores/Document.java

@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.flink.agents.api.vectorstores;
+
+import java.util.Map;
+
+/**
+ * A document retrieved from vector store search.
+ *
+ * <p>Represents a single piece of content with associated metadata. This class is generic to
+ * support different content types while maintaining consistent metadata and identification
+ * structure.
+ *
+ * @param <ContentT> the type of the document content
+ */
+public class Document<ContentT> {
+
+    /** Unique identifier of the document (if available). */
+    private final String id;
+
+    /** The actual content of the document. */
+    private final ContentT content;
+
+    /** Document metadata such as source, author, timestamp, etc. */
+    private final Map<String, Object> metadata;
+
+    public Document(ContentT content, Map<String, Object> metadata, String id) {
+        this.content = content;
+        this.metadata = metadata;
+        this.id = id;
+    }
+
+    public ContentT getContent() {
+        return content;
+    }
+
+    public Map<String, Object> getMetadata() {
+        return metadata;
+    }
+
+    public String getId() {
+        return id;
+    }
+}