Source code for langchain.vectorstores.hologres
from __future__ import annotations
import json
import logging
import uuid
from typing import Any, Dict, Iterable, List, Optional, Tuple, Type
from langchain.docstore.document import Document
from langchain.schema.embeddings import Embeddings
from langchain.schema.vectorstore import VectorStore
from langchain.utils import get_from_dict_or_env
ADA_TOKEN_COUNT = 1536
_LANGCHAIN_DEFAULT_TABLE_NAME = "langchain_pg_embedding"
[docs]class HologresWrapper:
    """`Hologres API` wrapper."""
[docs]    def __init__(self, connection_string: str, ndims: int, table_name: str) -> None:
        """Initialize the wrapper.
        Args:
            connection_string: Hologres connection string.
            ndims: Number of dimensions of the embedding output.
            table_name: Name of the table to store embeddings and data.
        """
        import psycopg2
        self.table_name = table_name
        self.conn = psycopg2.connect(connection_string)
        self.cursor = self.conn.cursor()
        self.conn.autocommit = False
        self.ndims = ndims
[docs]    def create_vector_extension(self) -> None:
        self.cursor.execute("create extension if not exists proxima")
        self.conn.commit()
[docs]    def create_table(self, drop_if_exist: bool = True) -> None:
        if drop_if_exist:
            self.cursor.execute(f"drop table if exists {self.table_name}")
        self.conn.commit()
        self.cursor.execute(
            f"""create table if not exists {self.table_name} (
id text,
embedding float4[] check(array_ndims(embedding) = 1 and \
array_length(embedding, 1) = {self.ndims}),