Source code for langchain.vectorstores.hologres
"""VectorStore wrapper around a Hologres database."""
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.embeddings.base import Embeddings
from langchain.utils import get_from_dict_or_env
from langchain.vectorstores.base import VectorStore
ADA_TOKEN_COUNT = 1536
_LANGCHAIN_DEFAULT_TABLE_NAME = "langchain_pg_embedding"
class HologresWrapper:
    def __init__(self, connection_string: str, ndims: int, table_name: str) -> None:
        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
    def create_vector_extension(self) -> None:
        self.cursor.execute("create extension if not exists proxima")
        self.conn.commit()
    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}),
metadata json,
document text);"""
        )
        self.cursor.execute(
            f"call set_table_property('{self.table_name}'"
            + """, 'proxima_vectors', 
'{"embedding":{"algorithm":"Graph",
"distance_method":"SquaredEuclidean",
rtdocs_stable/api.python.langchain.com/en/stable/_modules/langchain/vectorstores/hologres.html