{"version":3,"file":"analyticdb.cjs","names":["VectorStore","embeddings: EmbeddingsInterface","args: AnalyticDBArgs","documents: Document[]","vectors: number[][]","chunksTableData: DataType[]","Readable","query: number[]","k: number","filter?: this[\"FilterType\"]","result: [Document, number][]","Document","texts: string[]","metadatas: object[] | object","dbConfig: AnalyticDBArgs","docs: Document[]"],"sources":["../../src/vectorstores/analyticdb.ts"],"sourcesContent":["import * as uuid from \"uuid\";\nimport pg, { Pool, PoolConfig } from \"pg\";\nimport { from as copyFrom } from \"pg-copy-streams\";\nimport { pipeline } from \"node:stream/promises\";\nimport { Readable } from \"node:stream\";\n\nimport { VectorStore } from \"@langchain/core/vectorstores\";\nimport type { EmbeddingsInterface } from \"@langchain/core/embeddings\";\nimport { Document } from \"@langchain/core/documents\";\n\nconst _LANGCHAIN_DEFAULT_COLLECTION_NAME = \"langchain_document\";\n\n/**\n * Interface defining the arguments required to create an instance of\n * `AnalyticDBVectorStore`.\n */\nexport interface AnalyticDBArgs {\n  connectionOptions: PoolConfig;\n  embeddingDimension?: number;\n  collectionName?: string;\n  preDeleteCollection?: boolean;\n}\n\n/**\n * Interface defining the structure of data to be stored in the\n * AnalyticDB.\n */\ninterface DataType {\n  id: string;\n  embedding: number[];\n  document: string;\n  // eslint-disable-next-line @typescript-eslint/no-explicit-any\n  metadata: Record<string, any>;\n}\n\n/**\n * Class that provides methods for creating and managing a collection of\n * documents in an AnalyticDB, adding documents or vectors to the\n * collection, performing similarity search on vectors, and creating an\n * instance of `AnalyticDBVectorStore` from texts or documents.\n */\nexport class AnalyticDBVectorStore extends VectorStore {\n  // eslint-disable-next-line @typescript-eslint/no-explicit-any\n  declare FilterType: Record<string, any>;\n\n  private pool: Pool;\n\n  private embeddingDimension?: number;\n\n  private collectionName: string;\n\n  private preDeleteCollection: boolean;\n\n  private isCreateCollection = false;\n\n  _vectorstoreType(): string {\n    return \"analyticdb\";\n  }\n\n  constructor(embeddings: EmbeddingsInterface, args: AnalyticDBArgs) {\n    super(embeddings, args);\n\n    this.pool = new pg.Pool({\n      host: args.connectionOptions.host,\n      port: args.connectionOptions.port,\n      database: args.connectionOptions.database,\n      user: args.connectionOptions.user,\n      password: args.connectionOptions.password,\n    });\n    this.embeddingDimension = args.embeddingDimension;\n    this.collectionName =\n      args.collectionName || _LANGCHAIN_DEFAULT_COLLECTION_NAME;\n    this.preDeleteCollection = args.preDeleteCollection || false;\n  }\n\n  /**\n   * Closes all the clients in the pool and terminates the pool.\n   * @returns Promise that resolves when all clients are closed and the pool is terminated.\n   */\n  async end(): Promise<void> {\n    return this.pool.end();\n  }\n\n  /**\n   * Creates a new table in the database if it does not already exist. The\n   * table is created with columns for id, embedding, document, and\n   * metadata. An index is also created on the embedding column if it does\n   * not already exist.\n   * @returns Promise that resolves when the table and index are created.\n   */\n  async createTableIfNotExists(): Promise<void> {\n    if (!this.embeddingDimension) {\n      this.embeddingDimension = (\n        await this.embeddings.embedQuery(\"test\")\n      ).length;\n    }\n    const client = await this.pool.connect();\n    try {\n      await client.query(\"BEGIN\");\n      // Create the table if it doesn't exist\n      await client.query(`\n        CREATE TABLE IF NOT EXISTS ${this.collectionName} (\n          id TEXT PRIMARY KEY DEFAULT NULL,\n          embedding REAL[],\n          document TEXT,\n          metadata JSON\n        );\n      `);\n\n      // Check if the index exists\n      const indexName = `${this.collectionName}_embedding_idx`;\n      const indexQuery = `\n        SELECT 1\n        FROM pg_indexes\n        WHERE indexname = '${indexName}';\n      `;\n      const result = await client.query(indexQuery);\n\n      // Create the index if it doesn't exist\n      if (result.rowCount === 0) {\n        const indexStatement = `\n          CREATE INDEX ${indexName}\n          ON ${this.collectionName} USING ann(embedding)\n          WITH (\n            \"dim\" = ${this.embeddingDimension},\n            \"hnsw_m\" = 100\n          );\n        `;\n        await client.query(indexStatement);\n      }\n      await client.query(\"COMMIT\");\n    } catch (err) {\n      await client.query(\"ROLLBACK\");\n      throw err;\n    } finally {\n      client.release();\n    }\n  }\n\n  /**\n   * Deletes the collection from the database if it exists.\n   * @returns Promise that resolves when the collection is deleted.\n   */\n  async deleteCollection(): Promise<void> {\n    const dropStatement = `DROP TABLE IF EXISTS ${this.collectionName};`;\n    await this.pool.query(dropStatement);\n  }\n\n  /**\n   * Creates a new collection in the database. If `preDeleteCollection` is\n   * true, any existing collection with the same name is deleted before the\n   * new collection is created.\n   * @returns Promise that resolves when the collection is created.\n   */\n  async createCollection(): Promise<void> {\n    if (this.preDeleteCollection) {\n      await this.deleteCollection();\n    }\n    await this.createTableIfNotExists();\n    this.isCreateCollection = true;\n  }\n\n  /**\n   * Adds an array of documents to the collection. The documents are first\n   * converted to vectors using the `embedDocuments` method of the\n   * `embeddings` instance.\n   * @param documents Array of Document instances to be added to the collection.\n   * @returns Promise that resolves when the documents are added.\n   */\n  async addDocuments(documents: Document[]): Promise<void> {\n    // When the pageContent is empty in certain scenarios (such as when using unstructuredIo), an error occurs during embedding.\n    const filteredDocs = documents.filter((doc) => doc.pageContent);\n    if (filteredDocs.length !== documents.length) {\n      console.warn(\n        `[AnalyticDB]: Filtered out ${\n          documents.length - filteredDocs.length\n        } empty documents.`\n      );\n    }\n    const texts = filteredDocs.map(({ pageContent }) => pageContent);\n    return this.addVectors(\n      await this.embeddings.embedDocuments(texts),\n      filteredDocs\n    );\n  }\n\n  /**\n   * Adds an array of vectors and corresponding documents to the collection.\n   * The vectors and documents are batch inserted into the database.\n   * @param vectors Array of vectors to be added to the collection.\n   * @param documents Array of Document instances corresponding to the vectors.\n   * @returns Promise that resolves when the vectors and documents are added.\n   */\n  async addVectors(vectors: number[][], documents: Document[]): Promise<void> {\n    if (vectors.length === 0) {\n      return;\n    }\n    if (vectors.length !== documents.length) {\n      throw new Error(`Vectors and documents must have the same length`);\n    }\n    if (!this.embeddingDimension) {\n      this.embeddingDimension = (\n        await this.embeddings.embedQuery(\"test\")\n      ).length;\n    }\n    if (vectors[0].length !== this.embeddingDimension) {\n      throw new Error(\n        `Vectors must have the same length as the number of dimensions (${this.embeddingDimension})`\n      );\n    }\n\n    if (!this.isCreateCollection) {\n      await this.createCollection();\n    }\n\n    const client = await this.pool.connect();\n    try {\n      const chunkSize = 500;\n      const chunksTableData: DataType[] = [];\n\n      for (let i = 0; i < documents.length; i += 1) {\n        chunksTableData.push({\n          id: uuid.v4(),\n          embedding: vectors[i],\n          document: documents[i].pageContent,\n          metadata: documents[i].metadata,\n        });\n\n        // Execute the batch insert when the batch size is reached\n        if (chunksTableData.length === chunkSize) {\n          const rs = new Readable();\n          let currentIndex = 0;\n          rs._read = function () {\n            if (currentIndex === chunkSize) {\n              rs.push(null);\n            } else {\n              const data = chunksTableData[currentIndex];\n              rs.push(\n                `${data.id}\\t{${data.embedding.join(\",\")}}\\t${\n                  data.document\n                }\\t${JSON.stringify(data.metadata)}\\n`\n              );\n              currentIndex += 1;\n            }\n          };\n          const ws = client.query(\n            copyFrom(\n              `COPY ${this.collectionName}(id, embedding, document, metadata) FROM STDIN`\n            )\n          );\n\n          // @ts-expect-error - Overload issues & ReadableStream issue\n          await pipeline(rs, ws);\n          // Clear the chunksTableData list for the next batch\n          chunksTableData.length = 0;\n        }\n      }\n\n      // Insert any remaining records that didn't make up a full batch\n      if (chunksTableData.length > 0) {\n        const rs = new Readable();\n        let currentIndex = 0;\n        rs._read = function () {\n          if (currentIndex === chunksTableData.length) {\n            rs.push(null);\n          } else {\n            const data = chunksTableData[currentIndex];\n            rs.push(\n              `${data.id}\\t{${data.embedding.join(\",\")}}\\t${\n                data.document\n              }\\t${JSON.stringify(data.metadata)}\\n`\n            );\n            currentIndex += 1;\n          }\n        };\n        const ws = client.query(\n          copyFrom(\n            `COPY ${this.collectionName}(id, embedding, document, metadata) FROM STDIN`\n          )\n        );\n\n        // @ts-expect-error - Overload issues & ReadableStream issue\n        await pipeline(rs, ws);\n      }\n    } finally {\n      client.release();\n    }\n  }\n\n  /**\n   * Performs a similarity search on the vectors in the collection. The\n   * search is performed using the given query vector and returns the top k\n   * most similar vectors along with their corresponding documents and\n   * similarity scores.\n   * @param query Query vector for the similarity search.\n   * @param k Number of top similar vectors to return.\n   * @param filter Optional. Filter to apply on the metadata of the documents.\n   * @returns Promise that resolves to an array of tuples, each containing a Document instance and its similarity score.\n   */\n  async similaritySearchVectorWithScore(\n    query: number[],\n    k: number,\n    filter?: this[\"FilterType\"]\n  ): Promise<[Document, number][]> {\n    if (!this.isCreateCollection) {\n      await this.createCollection();\n    }\n\n    let filterCondition = \"\";\n    const filterEntries = filter ? Object.entries(filter) : [];\n    if (filterEntries.length > 0) {\n      const conditions = filterEntries.map(\n        (_, index) => `metadata->>$${2 * index + 3} = $${2 * index + 4}`\n      );\n      filterCondition = `WHERE ${conditions.join(\" AND \")}`;\n    }\n\n    const sqlQuery = `\n      SELECT *, l2_distance(embedding, $1::real[]) AS distance\n      FROM ${this.collectionName}\n      ${filterCondition}\n      ORDER BY embedding <-> $1\n      LIMIT $2;\n    `;\n\n    // Execute the query and fetch the results\n    const { rows } = await this.pool.query(sqlQuery, [\n      query,\n      k,\n      ...filterEntries.flatMap(([key, value]) => [key, value]),\n    ]);\n\n    const result: [Document, number][] = rows.map((row) => [\n      new Document({ pageContent: row.document, metadata: row.metadata }),\n      row.distance,\n    ]);\n\n    return result;\n  }\n\n  /**\n   * Creates an instance of `AnalyticDBVectorStore` from an array of texts\n   * and corresponding metadata. The texts are first converted to Document\n   * instances before being added to the collection.\n   * @param texts Array of texts to be added to the collection.\n   * @param metadatas Array or object of metadata corresponding to the texts.\n   * @param embeddings Embeddings instance used to convert the texts to vectors.\n   * @param dbConfig Configuration for the AnalyticDB.\n   * @returns Promise that resolves to an instance of `AnalyticDBVectorStore`.\n   */\n  static async fromTexts(\n    texts: string[],\n    metadatas: object[] | object,\n    embeddings: EmbeddingsInterface,\n    dbConfig: AnalyticDBArgs\n  ): Promise<AnalyticDBVectorStore> {\n    const docs = [];\n    for (let i = 0; i < texts.length; i += 1) {\n      const metadata = Array.isArray(metadatas) ? metadatas[i] : metadatas;\n      const newDoc = new Document({\n        pageContent: texts[i],\n        metadata,\n      });\n      docs.push(newDoc);\n    }\n    return AnalyticDBVectorStore.fromDocuments(docs, embeddings, dbConfig);\n  }\n\n  /**\n   * Creates an instance of `AnalyticDBVectorStore` from an array of\n   * Document instances. The documents are added to the collection.\n   * @param docs Array of Document instances to be added to the collection.\n   * @param embeddings Embeddings instance used to convert the documents to vectors.\n   * @param dbConfig Configuration for the AnalyticDB.\n   * @returns Promise that resolves to an instance of `AnalyticDBVectorStore`.\n   */\n  static async fromDocuments(\n    docs: Document[],\n    embeddings: EmbeddingsInterface,\n    dbConfig: AnalyticDBArgs\n  ): Promise<AnalyticDBVectorStore> {\n    const instance = new this(embeddings, dbConfig);\n    await instance.addDocuments(docs);\n    return instance;\n  }\n\n  /**\n   * Creates an instance of `AnalyticDBVectorStore` from an existing index\n   * in the database. A new collection is created in the database.\n   * @param embeddings Embeddings instance used to convert the documents to vectors.\n   * @param dbConfig Configuration for the AnalyticDB.\n   * @returns Promise that resolves to an instance of `AnalyticDBVectorStore`.\n   */\n  static async fromExistingIndex(\n    embeddings: EmbeddingsInterface,\n    dbConfig: AnalyticDBArgs\n  ): Promise<AnalyticDBVectorStore> {\n    const instance = new this(embeddings, dbConfig);\n    await instance.createCollection();\n    return instance;\n  }\n}\n"],"mappings":";;;;;;;;;;;;AAUA,MAAM,qCAAqC;;;;;;;AA+B3C,IAAa,wBAAb,MAAa,8BAA8BA,0CAAY;CAIrD,AAAQ;CAER,AAAQ;CAER,AAAQ;CAER,AAAQ;CAER,AAAQ,qBAAqB;CAE7B,mBAA2B;AACzB,SAAO;CACR;CAED,YAAYC,YAAiCC,MAAsB;EACjE,MAAM,YAAY,KAAK;EAEvB,KAAK,OAAO,IAAI,WAAG,KAAK;GACtB,MAAM,KAAK,kBAAkB;GAC7B,MAAM,KAAK,kBAAkB;GAC7B,UAAU,KAAK,kBAAkB;GACjC,MAAM,KAAK,kBAAkB;GAC7B,UAAU,KAAK,kBAAkB;EAClC;EACD,KAAK,qBAAqB,KAAK;EAC/B,KAAK,iBACH,KAAK,kBAAkB;EACzB,KAAK,sBAAsB,KAAK,uBAAuB;CACxD;;;;;CAMD,MAAM,MAAqB;AACzB,SAAO,KAAK,KAAK,KAAK;CACvB;;;;;;;;CASD,MAAM,yBAAwC;AAC5C,MAAI,CAAC,KAAK,oBACR,KAAK,sBACH,MAAM,KAAK,WAAW,WAAW,OAAO,EACxC;EAEJ,MAAM,SAAS,MAAM,KAAK,KAAK,SAAS;AACxC,MAAI;GACF,MAAM,OAAO,MAAM,QAAQ;GAE3B,MAAM,OAAO,MAAM,CAAC;mCACS,EAAE,KAAK,eAAe;;;;;;MAMnD,CAAC,CAAC;GAGF,MAAM,YAAY,GAAG,KAAK,eAAe,cAAc,CAAC;GACxD,MAAM,aAAa,CAAC;;;2BAGC,EAAE,UAAU;MACjC,CAAC;GACD,MAAM,SAAS,MAAM,OAAO,MAAM,WAAW;AAG7C,OAAI,OAAO,aAAa,GAAG;IACzB,MAAM,iBAAiB,CAAC;uBACT,EAAE,UAAU;aACtB,EAAE,KAAK,eAAe;;oBAEf,EAAE,KAAK,mBAAmB;;;QAGtC,CAAC;IACD,MAAM,OAAO,MAAM,eAAe;GACnC;GACD,MAAM,OAAO,MAAM,SAAS;EAC7B,SAAQ,KAAK;GACZ,MAAM,OAAO,MAAM,WAAW;AAC9B,SAAM;EACP,UAAS;GACR,OAAO,SAAS;EACjB;CACF;;;;;CAMD,MAAM,mBAAkC;EACtC,MAAM,gBAAgB,CAAC,qBAAqB,EAAE,KAAK,eAAe,CAAC,CAAC;EACpE,MAAM,KAAK,KAAK,MAAM,cAAc;CACrC;;;;;;;CAQD,MAAM,mBAAkC;AACtC,MAAI,KAAK,qBACP,MAAM,KAAK,kBAAkB;EAE/B,MAAM,KAAK,wBAAwB;EACnC,KAAK,qBAAqB;CAC3B;;;;;;;;CASD,MAAM,aAAaC,WAAsC;EAEvD,MAAM,eAAe,UAAU,OAAO,CAAC,QAAQ,IAAI,YAAY;AAC/D,MAAI,aAAa,WAAW,UAAU,QACpC,QAAQ,KACN,CAAC,2BAA2B,EAC1B,UAAU,SAAS,aAAa,OACjC,iBAAiB,CAAC,CACpB;EAEH,MAAM,QAAQ,aAAa,IAAI,CAAC,EAAE,aAAa,KAAK,YAAY;AAChE,SAAO,KAAK,WACV,MAAM,KAAK,WAAW,eAAe,MAAM,EAC3C,aACD;CACF;;;;;;;;CASD,MAAM,WAAWC,SAAqBD,WAAsC;AAC1E,MAAI,QAAQ,WAAW,EACrB;AAEF,MAAI,QAAQ,WAAW,UAAU,OAC/B,OAAM,IAAI,MAAM,CAAC,+CAA+C,CAAC;AAEnE,MAAI,CAAC,KAAK,oBACR,KAAK,sBACH,MAAM,KAAK,WAAW,WAAW,OAAO,EACxC;AAEJ,MAAI,QAAQ,GAAG,WAAW,KAAK,mBAC7B,OAAM,IAAI,MACR,CAAC,+DAA+D,EAAE,KAAK,mBAAmB,CAAC,CAAC;AAIhG,MAAI,CAAC,KAAK,oBACR,MAAM,KAAK,kBAAkB;EAG/B,MAAM,SAAS,MAAM,KAAK,KAAK,SAAS;AACxC,MAAI;GACF,MAAM,YAAY;GAClB,MAAME,kBAA8B,CAAE;AAEtC,QAAK,IAAI,IAAI,GAAG,IAAI,UAAU,QAAQ,KAAK,GAAG;IAC5C,gBAAgB,KAAK;KACnB,IAAI,KAAK,IAAI;KACb,WAAW,QAAQ;KACnB,UAAU,UAAU,GAAG;KACvB,UAAU,UAAU,GAAG;IACxB,EAAC;AAGF,QAAI,gBAAgB,WAAW,WAAW;KACxC,MAAM,KAAK,IAAIC;KACf,IAAI,eAAe;KACnB,GAAG,QAAQ,WAAY;AACrB,UAAI,iBAAiB,WACnB,GAAG,KAAK,KAAK;WACR;OACL,MAAM,OAAO,gBAAgB;OAC7B,GAAG,KACD,GAAG,KAAK,GAAG,GAAG,EAAE,KAAK,UAAU,KAAK,IAAI,CAAC,GAAG,EAC1C,KAAK,SACN,EAAE,EAAE,KAAK,UAAU,KAAK,SAAS,CAAC,EAAE,CAAC,CACvC;OACD,gBAAgB;MACjB;KACF;KACD,MAAM,KAAK,OAAO,gCAEd,CAAC,KAAK,EAAE,KAAK,eAAe,8CAA8C,CAAC,CAC5E,CACF;KAGD,yCAAe,IAAI,GAAG;KAEtB,gBAAgB,SAAS;IAC1B;GACF;AAGD,OAAI,gBAAgB,SAAS,GAAG;IAC9B,MAAM,KAAK,IAAIA;IACf,IAAI,eAAe;IACnB,GAAG,QAAQ,WAAY;AACrB,SAAI,iBAAiB,gBAAgB,QACnC,GAAG,KAAK,KAAK;UACR;MACL,MAAM,OAAO,gBAAgB;MAC7B,GAAG,KACD,GAAG,KAAK,GAAG,GAAG,EAAE,KAAK,UAAU,KAAK,IAAI,CAAC,GAAG,EAC1C,KAAK,SACN,EAAE,EAAE,KAAK,UAAU,KAAK,SAAS,CAAC,EAAE,CAAC,CACvC;MACD,gBAAgB;KACjB;IACF;IACD,MAAM,KAAK,OAAO,gCAEd,CAAC,KAAK,EAAE,KAAK,eAAe,8CAA8C,CAAC,CAC5E,CACF;IAGD,yCAAe,IAAI,GAAG;GACvB;EACF,UAAS;GACR,OAAO,SAAS;EACjB;CACF;;;;;;;;;;;CAYD,MAAM,gCACJC,OACAC,GACAC,QAC+B;AAC/B,MAAI,CAAC,KAAK,oBACR,MAAM,KAAK,kBAAkB;EAG/B,IAAI,kBAAkB;EACtB,MAAM,gBAAgB,SAAS,OAAO,QAAQ,OAAO,GAAG,CAAE;AAC1D,MAAI,cAAc,SAAS,GAAG;GAC5B,MAAM,aAAa,cAAc,IAC/B,CAAC,GAAG,UAAU,CAAC,YAAY,EAAE,IAAI,QAAQ,EAAE,IAAI,EAAE,IAAI,QAAQ,GAAG,CACjE;GACD,kBAAkB,CAAC,MAAM,EAAE,WAAW,KAAK,QAAQ,EAAE;EACtD;EAED,MAAM,WAAW,CAAC;;WAEX,EAAE,KAAK,eAAe;MAC3B,EAAE,gBAAgB;;;IAGpB,CAAC;EAGD,MAAM,EAAE,MAAM,GAAG,MAAM,KAAK,KAAK,MAAM,UAAU;GAC/C;GACA;GACA,GAAG,cAAc,QAAQ,CAAC,CAAC,KAAK,MAAM,KAAK,CAAC,KAAK,KAAM,EAAC;EACzD,EAAC;EAEF,MAAMC,SAA+B,KAAK,IAAI,CAAC,QAAQ,CACrD,IAAIC,oCAAS;GAAE,aAAa,IAAI;GAAU,UAAU,IAAI;EAAU,IAClE,IAAI,QACL,EAAC;AAEF,SAAO;CACR;;;;;;;;;;;CAYD,aAAa,UACXC,OACAC,WACAZ,YACAa,UACgC;EAChC,MAAM,OAAO,CAAE;AACf,OAAK,IAAI,IAAI,GAAG,IAAI,MAAM,QAAQ,KAAK,GAAG;GACxC,MAAM,WAAW,MAAM,QAAQ,UAAU,GAAG,UAAU,KAAK;GAC3D,MAAM,SAAS,IAAIH,oCAAS;IAC1B,aAAa,MAAM;IACnB;GACD;GACD,KAAK,KAAK,OAAO;EAClB;AACD,SAAO,sBAAsB,cAAc,MAAM,YAAY,SAAS;CACvE;;;;;;;;;CAUD,aAAa,cACXI,MACAd,YACAa,UACgC;EAChC,MAAM,WAAW,IAAI,KAAK,YAAY;EACtC,MAAM,SAAS,aAAa,KAAK;AACjC,SAAO;CACR;;;;;;;;CASD,aAAa,kBACXb,YACAa,UACgC;EAChC,MAAM,WAAW,IAAI,KAAK,YAAY;EACtC,MAAM,SAAS,kBAAkB;AACjC,SAAO;CACR;AACF"}