feat(ts-sdk): add Elasticsearch vector store provider (#5866)

Co-authored-by: kartik-mem0 <kartik.labhshetwar@mem0.ai>
This commit is contained in:
404 Ameyy
2026-07-07 17:10:25 +05:30
committed by GitHub
co-authored by kartik-mem0
parent d122479687
commit 94e46526bc
8 changed files with 1007 additions and 26 deletions
+1
View File
@@ -32,6 +32,7 @@ export * from "./vector_stores/langchain";
export * from "./vector_stores/vectorize";
export * from "./vector_stores/azure_ai_search";
export * from "./vector_stores/pgvector";
export * from "./vector_stores/elasticsearch";
export * from "./vector_stores/upstash_vector";
export * from "./vector_stores/azure_mysql";
export * from "./vector_stores/cassandra";
+3
View File
@@ -43,6 +43,7 @@ import { LangchainEmbedder } from "../embeddings/langchain";
import { LangchainVectorStore } from "../vector_stores/langchain";
import { AzureAISearch } from "../vector_stores/azure_ai_search";
import { PGVector } from "../vector_stores/pgvector";
import { ElasticsearchDB } from "../vector_stores/elasticsearch";
import { OpenSearchDB } from "../vector_stores/opensearch";
import { UpstashVector } from "../vector_stores/upstash_vector";
import { AzureMySQLDB } from "../vector_stores/azure_mysql";
@@ -141,6 +142,8 @@ export class VectorStoreFactory {
return new VertexAIVectorSearch(config as any);
case "pgvector":
return new PGVector(config as any);
case "elasticsearch":
return new ElasticsearchDB(config as any);
case "opensearch":
return new OpenSearchDB(config as any);
case "upstash_vector":
@@ -0,0 +1,333 @@
import { Client } from "@elastic/elasticsearch";
import { VectorStore } from "./base";
import { SearchFilters, VectorStoreConfig, VectorStoreResult } from "../types";
interface ElasticsearchConfig extends VectorStoreConfig {
client?: Client;
host?: string;
port?: number;
cloudId?: string;
apiKey?: string;
username?: string;
password?: string;
collectionName: string;
embeddingModelDims: number;
dimension?: number;
useSsl?: boolean;
caCerts?: string;
verifyCerts?: boolean;
autoCreateIndex?: boolean;
headers?: Record<string, string>;
}
const SAFE_FILTER_KEY = /^[a-zA-Z_][a-zA-Z0-9_]*$/;
// Mirrors the Python provider's _validate_filter. Filter keys are interpolated
// into `metadata.${key}` field paths, so reject anything that is not a plain
// identifier and reject non-scalar values before they reach the query.
function validateFilter(key: string, value: unknown): void {
if (typeof key !== "string" || !SAFE_FILTER_KEY.test(key)) {
throw new Error(`Invalid filter key: ${JSON.stringify(key)}`);
}
if (
typeof value !== "string" &&
typeof value !== "number" &&
typeof value !== "boolean"
) {
throw new Error(
`Filter value for ${JSON.stringify(key)} must be string, number, or boolean`,
);
}
}
export class ElasticsearchDB implements VectorStore {
private client: Client;
private readonly collectionName: string;
private readonly dimension: number;
private readonly autoCreateIndex: boolean;
private _initPromise?: Promise<void>;
constructor(config: ElasticsearchConfig) {
this.collectionName = config.collectionName;
this.dimension = config.dimension || config.embeddingModelDims || 1536;
this.autoCreateIndex = config.autoCreateIndex !== false;
if (config.client) {
this.client = config.client;
} else {
const params: Record<string, any> = {};
if (config.cloudId) {
params.cloud = { id: config.cloudId };
if (config.apiKey) {
params.auth = { apiKey: config.apiKey };
}
} else {
const host = config.host || "localhost";
const port = config.port || 9200;
params.node = `${config.useSsl !== false ? "https" : "http"}://${host}:${port}`;
if (config.apiKey) {
params.auth = { apiKey: config.apiKey };
} else if (config.username && config.password) {
params.auth = {
username: config.username,
password: config.password,
};
}
}
if (config.verifyCerts !== undefined) {
params.tls = { rejectUnauthorized: config.verifyCerts };
}
if (config.caCerts) {
params.tls = { ...params.tls, ca: config.caCerts };
}
if (config.headers) {
params.headers = config.headers;
}
this.client = new Client(params);
}
this.initialize().catch(console.error);
}
async initialize(): Promise<void> {
if (!this._initPromise) {
this._initPromise = this._doInitialize();
}
return this._initPromise;
}
private async _doInitialize(): Promise<void> {
try {
if (this.autoCreateIndex) {
await this.ensureIndex(this.collectionName, this.dimension);
}
await this.ensureIndex("memory_migrations", 1);
} catch (error) {
console.error("Error initializing Elasticsearch:", error);
throw error;
}
}
private async ensureIndex(name: string, size: number): Promise<void> {
const exists = await this.client.indices.exists({ index: name });
if (!exists) {
await this.client.indices.create({
index: name,
settings: {
number_of_replicas: 1,
number_of_shards: 5,
refresh_interval: "1s",
},
mappings: {
properties: {
vector: {
type: "dense_vector",
dims: size,
index: true,
similarity: "cosine",
},
metadata: {
type: "object",
properties: {
user_id: { type: "keyword" },
agent_id: { type: "keyword" },
run_id: { type: "keyword" },
},
},
},
},
});
}
}
async insert(
vectors: number[][],
ids: string[],
payloads: Record<string, any>[],
): Promise<void> {
const operations: any[] = [];
for (let i = 0; i < vectors.length; i++) {
operations.push(
{ index: { _index: this.collectionName, _id: ids[i] } },
{ vector: vectors[i], metadata: payloads[i] || {} },
);
}
await this.client.bulk({ operations });
}
async search(
query: number[],
topK: number = 5,
filters?: SearchFilters,
): Promise<VectorStoreResult[]> {
const searchBody: Record<string, any> = {
knn: {
field: "vector",
query_vector: query,
k: topK,
num_candidates: topK * 2,
},
};
if (filters && Object.keys(filters).length > 0) {
const filterConditions = Object.entries(filters).map(([key, value]) => {
validateFilter(key, value);
return { term: { [`metadata.${key}`]: value } };
});
searchBody.knn.filter = { bool: { must: filterConditions } };
}
const response = await this.client.search({
index: this.collectionName,
...searchBody,
});
return response.hits.hits.map((hit: any) => ({
id: hit._id,
score: hit._score,
payload: hit._source?.metadata || {},
}));
}
async get(vectorId: string): Promise<VectorStoreResult | null> {
try {
const response = await this.client.get({
index: this.collectionName,
id: vectorId,
});
return {
id: response._id,
payload: (response._source as any)?.metadata || {},
};
} catch (error: any) {
if (error?.meta?.statusCode === 404) return null;
throw error;
}
}
async update(
vectorId: string,
vector: number[],
payload: Record<string, any>,
): Promise<void> {
const doc: Record<string, any> = {};
if (vector) doc.vector = vector;
if (payload) doc.metadata = payload;
await this.client.update({
index: this.collectionName,
id: vectorId,
doc,
});
}
async delete(vectorId: string): Promise<void> {
await this.client.delete({
index: this.collectionName,
id: vectorId,
});
}
async deleteCol(): Promise<void> {
await this.client.indices.delete({ index: this.collectionName });
}
async list(
filters?: SearchFilters,
topK: number = 100,
): Promise<[VectorStoreResult[], number]> {
const query: Record<string, any> = { query: { match_all: {} } };
if (filters && Object.keys(filters).length > 0) {
const filterConditions = Object.entries(filters).map(([key, value]) => {
validateFilter(key, value);
return { term: { [`metadata.${key}`]: value } };
});
query.query = { bool: { must: filterConditions } };
}
query.size = topK;
const response = await this.client.search({
index: this.collectionName,
...query,
});
const results = response.hits.hits.map((hit: any) => ({
id: hit._id,
payload: hit._source?.metadata || {},
}));
return [results, results.length];
}
private generateUUID(): string {
return "xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx".replace(/[xy]/g, (c) => {
const r = (Math.random() * 16) | 0;
const v = c === "x" ? r : (r & 0x3) | 0x8;
return v.toString(16);
});
}
async getUserId(): Promise<string> {
try {
const response = await this.client.search({
index: "memory_migrations",
query: { match_all: {} },
size: 1,
});
if (response.hits.hits.length > 0) {
return (response.hits.hits[0]._source as any)?.metadata
?.user_id as string;
}
const randomUserId =
Math.random().toString(36).substring(2, 15) +
Math.random().toString(36).substring(2, 15);
// memory_migrations is only ever read via match_all, never by vector,
// so store no vector. A zero vector would be rejected by the cosine
// dense_vector mapping ("does not support vectors with zero magnitude").
await this.client.index({
index: "memory_migrations",
id: this.generateUUID(),
document: { metadata: { user_id: randomUserId } },
});
return randomUserId;
} catch (error) {
console.error("Error getting user ID:", error);
throw error;
}
}
async setUserId(userId: string): Promise<void> {
try {
const response = await this.client.search({
index: "memory_migrations",
query: { match_all: {} },
size: 1,
});
const docId =
response.hits.hits.length > 0
? response.hits.hits[0]._id
: this.generateUUID();
await this.client.index({
index: "memory_migrations",
id: docId,
document: { metadata: { user_id: userId } },
});
} catch (error) {
console.error("Error setting user ID:", error);
throw error;
}
}
}
@@ -0,0 +1,524 @@
/// <reference types="jest" />
type MockFn = jest.Mock;
const mockIndicesExists = jest.fn();
const mockIndicesCreate = jest.fn();
const mockBulk = jest.fn();
const mockSearch = jest.fn();
const mockGet = jest.fn();
const mockUpdate = jest.fn();
const mockDelete = jest.fn();
const mockIndicesDelete = jest.fn();
const mockIndex_ = jest.fn();
const mockClient = jest.fn().mockImplementation(() => ({
indices: {
exists: mockIndicesExists,
create: mockIndicesCreate,
delete: mockIndicesDelete,
},
bulk: mockBulk,
search: mockSearch,
get: mockGet,
update: mockUpdate,
delete: mockDelete,
index: mockIndex_,
}));
jest.mock("@elastic/elasticsearch", () => ({
Client: mockClient,
}));
const { ElasticsearchDB } = require("../src/vector_stores/elasticsearch");
beforeEach(() => {
jest.clearAllMocks();
mockIndicesExists.mockResolvedValue(false);
mockIndicesCreate.mockResolvedValue({});
mockBulk.mockResolvedValue({});
mockSearch.mockResolvedValue({ hits: { hits: [] } });
mockGet.mockResolvedValue({
_id: "test-id",
_source: { metadata: { key: "val" } },
});
mockUpdate.mockResolvedValue({});
mockDelete.mockResolvedValue({});
mockIndicesDelete.mockResolvedValue({});
mockIndex_.mockResolvedValue({ _id: "new-id" });
});
describe("ElasticsearchDB", () => {
describe("constructor", () => {
it("creates client with self-hosted host:port", () => {
new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 768,
host: "localhost",
port: 9200,
username: "user",
password: "pass",
});
expect(mockClient).toHaveBeenCalledWith(
expect.objectContaining({
node: "https://localhost:9200",
auth: { username: "user", password: "pass" },
}),
);
});
it("creates client with cloud config", () => {
new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 1536,
cloudId:
"my-cloud:dXMtZWFzdDQuZ2NwLmVsYXN0aWMtY2xvdWQuY29tOjQ0MyQxMjM0NTY3ODkw",
apiKey: "base64-key",
});
expect(mockClient).toHaveBeenCalledWith(
expect.objectContaining({
cloud: {
id: "my-cloud:dXMtZWFzdDQuZ2NwLmVsYXN0aWMtY2xvdWQuY29tOjQ0MyQxMjM0NTY3ODkw",
},
auth: { apiKey: "base64-key" },
}),
);
});
it("accepts pre-configured client", () => {
const preExisting = new mockClient();
new ElasticsearchDB({
client: preExisting,
collectionName: "mem0",
embeddingModelDims: 1536,
});
expect(mockClient).toHaveBeenCalledTimes(1);
});
it("defaults to port 9200 and https", () => {
new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 384,
host: "es.example.com",
});
expect(mockClient).toHaveBeenCalledWith(
expect.objectContaining({
node: "https://es.example.com:9200",
}),
);
});
});
describe("initialize", () => {
it("creates index with correct dense_vector mapping", async () => {
const store = new ElasticsearchDB({
collectionName: "my_memories",
embeddingModelDims: 768,
host: "localhost",
});
await store.initialize();
expect(mockIndicesExists).toHaveBeenCalledWith({ index: "my_memories" });
expect(mockIndicesCreate).toHaveBeenCalledWith(
expect.objectContaining({
index: "my_memories",
mappings: {
properties: {
vector: {
type: "dense_vector",
dims: 768,
index: true,
similarity: "cosine",
},
metadata: {
type: "object",
properties: {
user_id: { type: "keyword" },
agent_id: { type: "keyword" },
run_id: { type: "keyword" },
},
},
},
},
}),
);
});
it("creates memory_migrations index with dims 1", async () => {
const store = new ElasticsearchDB({
collectionName: "test",
embeddingModelDims: 1536,
host: "localhost",
});
await store.initialize();
expect(mockIndicesCreate).toHaveBeenCalledWith(
expect.objectContaining({
index: "memory_migrations",
mappings: {
properties: {
vector: {
type: "dense_vector",
dims: 1,
index: true,
similarity: "cosine",
},
metadata: {
type: "object",
properties: {
user_id: { type: "keyword" },
agent_id: { type: "keyword" },
run_id: { type: "keyword" },
},
},
},
},
}),
);
});
it("is idempotent (initialize does not re-run)", async () => {
const store = new ElasticsearchDB({
collectionName: "test",
embeddingModelDims: 1536,
host: "localhost",
});
await store.initialize();
mockIndicesExists.mockClear();
await store.initialize();
expect(mockIndicesExists).not.toHaveBeenCalled();
});
it("skips index creation when autoCreateIndex is false", async () => {
const store = new ElasticsearchDB({
collectionName: "test",
embeddingModelDims: 1536,
host: "localhost",
autoCreateIndex: false,
});
await store.initialize();
expect(mockIndicesExists).not.toHaveBeenCalledWith({ index: "test" });
});
});
describe("insert", () => {
it("bulk indexes vectors with ids and payloads", async () => {
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
await store.insert(
[
[0.1, 0.2, 0.3, 0.4],
[0.5, 0.6, 0.7, 0.8],
],
["id-1", "id-2"],
[{ user_id: "u1" }, { user_id: "u2" }],
);
expect(mockBulk).toHaveBeenCalledWith({
operations: [
{ index: { _index: "mem0", _id: "id-1" } },
{ vector: [0.1, 0.2, 0.3, 0.4], metadata: { user_id: "u1" } },
{ index: { _index: "mem0", _id: "id-2" } },
{ vector: [0.5, 0.6, 0.7, 0.8], metadata: { user_id: "u2" } },
],
});
});
});
describe("search", () => {
it("builds correct knn query", async () => {
mockSearch.mockResolvedValue({
hits: {
hits: [
{
_id: "id-1",
_score: 0.95,
_source: { metadata: { user_id: "u1" } },
},
],
},
});
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
const results = await store.search([0.1, 0.2, 0.3, 0.4], 3);
expect(mockSearch).toHaveBeenCalledWith({
index: "mem0",
knn: {
field: "vector",
query_vector: [0.1, 0.2, 0.3, 0.4],
k: 3,
num_candidates: 6,
},
});
expect(results).toEqual([
{ id: "id-1", score: 0.95, payload: { user_id: "u1" } },
]);
});
it("applies metadata filters", async () => {
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
await store.search([0.1, 0.2, 0.3, 0.4], 5, {
user_id: "u1",
agent_id: "a1",
});
expect(mockSearch).toHaveBeenCalledWith({
index: "mem0",
knn: {
field: "vector",
query_vector: [0.1, 0.2, 0.3, 0.4],
k: 5,
num_candidates: 10,
filter: {
bool: {
must: [
{ term: { "metadata.user_id": "u1" } },
{ term: { "metadata.agent_id": "a1" } },
],
},
},
},
});
});
it("rejects filter keys and values that are not safe scalars", async () => {
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
await expect(
store.search([0.1, 0.2, 0.3, 0.4], 5, { "bad.key": "x" }),
).rejects.toThrow(/Invalid filter key/);
await expect(
store.search([0.1, 0.2, 0.3, 0.4], 5, { user_id: { $ne: null } }),
).rejects.toThrow(/must be string, number, or boolean/);
expect(mockSearch).not.toHaveBeenCalled();
});
});
describe("get", () => {
it("returns document by id", async () => {
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
const result = await store.get("id-1");
expect(mockGet).toHaveBeenCalledWith({ index: "mem0", id: "id-1" });
expect(result).toEqual({ id: "test-id", payload: { key: "val" } });
});
it("returns null on 404", async () => {
mockGet.mockRejectedValue({ meta: { statusCode: 404 } });
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
const result = await store.get("missing");
expect(result).toBeNull();
});
});
describe("update", () => {
it("updates vector and payload", async () => {
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
await store.update("id-1", [1, 2, 3, 4], { key: "val" });
expect(mockUpdate).toHaveBeenCalledWith({
index: "mem0",
id: "id-1",
doc: { vector: [1, 2, 3, 4], metadata: { key: "val" } },
});
});
});
describe("delete", () => {
it("deletes document by id", async () => {
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
await store.delete("id-1");
expect(mockDelete).toHaveBeenCalledWith({ index: "mem0", id: "id-1" });
});
});
describe("deleteCol", () => {
it("deletes the index", async () => {
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
await store.deleteCol();
expect(mockIndicesDelete).toHaveBeenCalledWith({ index: "mem0" });
});
});
describe("list", () => {
it("returns all documents with count", async () => {
mockSearch.mockResolvedValue({
hits: {
hits: [
{ _id: "a", _source: { metadata: { user_id: "u1" } } },
{ _id: "b", _source: { metadata: { user_id: "u2" } } },
],
},
});
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
const [results, count] = await store.list();
expect(mockSearch).toHaveBeenCalledWith({
index: "mem0",
query: { match_all: {} },
size: 100,
});
expect(results).toHaveLength(2);
expect(count).toBe(2);
});
it("applies filters to list query", async () => {
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
await store.list({ user_id: "u1" }, 10);
expect(mockSearch).toHaveBeenCalledWith({
index: "mem0",
query: { bool: { must: [{ term: { "metadata.user_id": "u1" } }] } },
size: 10,
});
});
});
describe("getUserId / setUserId", () => {
it("generates new userId when memory_migrations is empty", async () => {
mockSearch.mockResolvedValue({ hits: { hits: [] } });
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
const userId = await store.getUserId();
expect(mockSearch).toHaveBeenCalledWith({
index: "memory_migrations",
query: { match_all: {} },
size: 1,
});
expect(mockIndex_).toHaveBeenCalledWith(
expect.objectContaining({
index: "memory_migrations",
document: { metadata: { user_id: expect.any(String) } },
}),
);
expect(userId).toBeTruthy();
});
it("returns existing userId", async () => {
mockSearch.mockResolvedValue({
hits: {
hits: [
{
_id: "doc-1",
_source: { metadata: { user_id: "existing-user" } },
},
],
},
});
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
const userId = await store.getUserId();
expect(userId).toBe("existing-user");
});
it("setUserId upserts into memory_migrations", async () => {
const store = new ElasticsearchDB({
collectionName: "mem0",
embeddingModelDims: 4,
host: "localhost",
});
await store.setUserId("custom-user");
expect(mockIndex_).toHaveBeenCalledWith(
expect.objectContaining({
index: "memory_migrations",
document: { metadata: { user_id: "custom-user" } },
}),
);
});
});
describe("factory registration", () => {
it("is registered in VectorStoreFactory", () => {
const { VectorStoreFactory } = require("../src/utils/factory");
const store = VectorStoreFactory.create("elasticsearch", {
collectionName: "test",
embeddingModelDims: 128,
host: "localhost",
client: new mockClient(),
});
expect(store).toBeInstanceOf(ElasticsearchDB);
});
});
});