codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import contextlib4import enum5import logging6import uuid7from typing import (8 Any,9 Callable,10 Dict,11 Generator,12 Iterable,13 List,14 Optional,15 Tuple,16 Type,17 Union,18)19 20import numpy as np21import sqlalchemy22from sqlalchemy import delete, func23from sqlalchemy.dialects.postgresql import JSON, UUID24from sqlalchemy.exc import ProgrammingError25from sqlalchemy.orm import Session26from sqlalchemy.sql import quoted_name27 28from langchain_community.vectorstores.utils import maximal_marginal_relevance29 30try:31 from sqlalchemy.orm import declarative_base32except ImportError:33 from sqlalchemy.ext.declarative import declarative_base34 35from langchain_core.documents import Document36from langchain_core.embeddings import Embeddings37from langchain_core.utils import get_from_dict_or_env38from langchain_core.vectorstores import VectorStore39 40ADA_TOKEN_COUNT = 153641_LANGCHAIN_DEFAULT_COLLECTION_NAME = "langchain"42 43 44def _results_to_docs(docs_and_scores: Any) -> List[Document]:45 """Return docs from docs and scores."""46 return [doc for doc, _ in docs_and_scores]47 48 49class BaseEmbeddingStore:50 """Base class for the Lantern embedding store."""51 52 53def get_embedding_store(54 distance_strategy: DistanceStrategy, collection_name: str55) -> Any:56 """Get the embedding store class."""57 58 embedding_type = None59 60 if distance_strategy == DistanceStrategy.HAMMING:61 embedding_type = sqlalchemy.INTEGER62 else:63 embedding_type = sqlalchemy.REAL # type: ignore[assignment]64 65 DynamicBase = declarative_base(class_registry=dict()) # type: Any66 67 class EmbeddingStore(DynamicBase, BaseEmbeddingStore):68 __tablename__ = collection_name69 uuid = sqlalchemy.Column(70 UUID(as_uuid=True), primary_key=True, default=uuid.uuid471 )72 __table_args__ = {"extend_existing": True}73 document = sqlalchemy.Column(sqlalchemy.String, nullable=True)74 cmetadata = sqlalchemy.Column(JSON, nullable=True)75 # custom_id : any user defined id76 custom_id = sqlalchemy.Column(sqlalchemy.String, nullable=True)77 embedding = sqlalchemy.Column(sqlalchemy.ARRAY(embedding_type)) # type: ignore[arg-type,var-annotated]78 79 return EmbeddingStore80 81 82class QueryResult:83 """Result from a query."""84 85 EmbeddingStore: BaseEmbeddingStore86 distance: float87 88 89class DistanceStrategy(str, enum.Enum):90 """Enumerator of the Distance strategies."""91 92 EUCLIDEAN = "l2sq"93 COSINE = "cosine"94 HAMMING = "hamming"95 96 97DEFAULT_DISTANCE_STRATEGY = DistanceStrategy.COSINE98 99 100class Lantern(VectorStore):101 """`Postgres` with the `lantern` extension as a vector store.102 103 lantern uses sequential scan by default. but you can create a HNSW index104 using the create_hnsw_index method.105 - `connection_string` is a postgres connection string.106 - `embedding_function` any embedding function implementing107 `langchain.embeddings.base.Embeddings` interface.108 - `collection_name` is the name of the collection to use. (default: langchain)109 - NOTE: This is the name of the table in which embedding data will be stored110 The table will be created when initializing the store (if not exists)111 So, make sure the user has the right permissions to create tables.112 - `distance_strategy` is the distance strategy to use. (default: EUCLIDEAN)113 - `EUCLIDEAN` is the euclidean distance.114 - `COSINE` is the cosine distance.115 - `HAMMING` is the hamming distance.116 - `pre_delete_collection` if True, will delete the collection if it exists.117 (default: False)118 - Useful for testing.119 """120 121 def __init__(122 self,123 connection_string: str,124 embedding_function: Embeddings,125 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,126 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,127 collection_metadata: Optional[dict] = None,128 pre_delete_collection: bool = False,129 logger: Optional[logging.Logger] = None,130 relevance_score_fn: Optional[Callable[[float], float]] = None,131 ) -> None:132 self.connection_string = connection_string133 self.embedding_function = embedding_function134 self.collection_name = collection_name135 self.collection_metadata = collection_metadata136 self._distance_strategy = distance_strategy137 self.pre_delete_collection = pre_delete_collection138 self.logger = logger or logging.getLogger(__name__)139 self.override_relevance_score_fn = relevance_score_fn140 self.EmbeddingStore = get_embedding_store(141 self.distance_strategy, collection_name142 )143 self.__post_init__()144 145 def __post_init__(146 self,147 ) -> None:148 self._conn = self.connect()149 self.create_hnsw_extension()150 self.create_collection()151 152 @property153 def distance_strategy(self) -> DistanceStrategy:154 if isinstance(self._distance_strategy, DistanceStrategy):155 return self._distance_strategy156 157 if self._distance_strategy == DistanceStrategy.EUCLIDEAN.value:158 return DistanceStrategy.EUCLIDEAN159 elif self._distance_strategy == DistanceStrategy.COSINE.value:160 return DistanceStrategy.COSINE161 elif self._distance_strategy == DistanceStrategy.HAMMING.value:162 return DistanceStrategy.HAMMING163 else:164 raise ValueError(165 f"Got unexpected value for distance: {self._distance_strategy}. "166 f"Should be one of {', '.join([ds.value for ds in DistanceStrategy])}."167 )168 169 @property170 def embeddings(self) -> Embeddings:171 return self.embedding_function172 173 @classmethod174 def connection_string_from_db_params(175 cls,176 driver: str,177 host: str,178 port: int,179 database: str,180 user: str,181 password: str,182 ) -> str:183 """Return connection string from database parameters."""184 return f"postgresql+{driver}://{user}:{password}@{host}:{port}/{database}"185 186 def connect(self) -> sqlalchemy.engine.Connection:187 engine = sqlalchemy.create_engine(self.connection_string)188 conn = engine.connect()189 return conn190 191 @property192 def distance_function(self) -> Any:193 if self.distance_strategy == DistanceStrategy.EUCLIDEAN:194 return "l2sq_dist"195 elif self.distance_strategy == DistanceStrategy.COSINE:196 return "cos_dist"197 elif self.distance_strategy == DistanceStrategy.HAMMING:198 return "hamming_dist"199 200 def create_hnsw_extension(self) -> None:201 try:202 with Session(self._conn) as session:203 statement = sqlalchemy.text("CREATE EXTENSION IF NOT EXISTS lantern")204 session.execute(statement)205 session.commit()206 except Exception as e:207 self.logger.exception(e)208 209 def create_tables_if_not_exists(self) -> None:210 try:211 self.create_collection()212 except ProgrammingError:213 pass214 215 def drop_table(self) -> None:216 try:217 self.EmbeddingStore.__table__.drop(self._conn.engine)218 except ProgrammingError:219 pass220 221 def drop_tables(self) -> None:222 self.drop_table()223 224 def _hamming_relevance_score_fn(self, distance: float) -> float:225 return distance226 227 def _select_relevance_score_fn(self) -> Callable[[float], float]:228 """229 The 'correct' relevance function230 may differ depending on a few things, including:231 - the distance / similarity metric used by the VectorStore232 - the scale of your embeddings (OpenAI's are unit normed. Many others are not!)233 - embedding dimensionality234 - etc.235 """236 if self.override_relevance_score_fn is not None:237 return self.override_relevance_score_fn238 239 # Default strategy is to rely on distance strategy provided240 # in vectorstore constructor241 if self.distance_strategy == DistanceStrategy.COSINE:242 return self._cosine_relevance_score_fn243 elif self.distance_strategy == DistanceStrategy.EUCLIDEAN:244 return self._euclidean_relevance_score_fn245 elif self.distance_strategy == DistanceStrategy.HAMMING:246 return self._hamming_relevance_score_fn247 else:248 raise ValueError(249 "No supported normalization function"250 f" for distance_strategy of {self._distance_strategy}."251 "Consider providing relevance_score_fn to Lantern constructor."252 )253 254 def _get_op_class(self) -> str:255 if self.distance_strategy == DistanceStrategy.COSINE:256 return "dist_cos_ops"257 elif self.distance_strategy == DistanceStrategy.EUCLIDEAN:258 return "dist_l2sq_ops"259 elif self.distance_strategy == DistanceStrategy.HAMMING:260 return "dist_hamming_ops"261 else:262 raise ValueError(263 "No supported operator class"264 f" for distance_strategy of {self._distance_strategy}."265 )266 267 def _get_operator(self) -> str:268 if self.distance_strategy == DistanceStrategy.COSINE:269 return "<=>"270 elif self.distance_strategy == DistanceStrategy.EUCLIDEAN:271 return "<->"272 elif self.distance_strategy == DistanceStrategy.HAMMING:273 return "<+>"274 else:275 raise ValueError(276 "No supported operator"277 f" for distance_strategy of {self._distance_strategy}."278 )279 280 def _typed_arg_for_distance(281 self, embedding: List[Union[float, int]]282 ) -> List[Union[float, int]]:283 if self.distance_strategy == DistanceStrategy.HAMMING:284 return list(map(lambda x: int(x), embedding))285 return embedding286 287 @property288 def _index_name(self) -> str:289 return f"langchain_{self.collection_name}_idx"290 291 def create_hnsw_index(292 self,293 dims: int = ADA_TOKEN_COUNT,294 m: int = 16,295 ef_construction: int = 64,296 ef_search: int = 64,297 **_kwargs: Any,298 ) -> None:299 """Create HNSW index on collection.300 301 Optional Keyword Args for HNSW Index:302 engine: "nmslib", "faiss", "lucene"; default: "nmslib"303 304 ef: Size of the dynamic list used during k-NN searches. Higher values305 lead to more accurate but slower searches; default: 64306 307 ef_construction: Size of the dynamic list used during k-NN graph creation.308 Higher values lead to more accurate graph but slower indexing speed;309 default: 64310 311 m: Number of bidirectional links created for each new element. Large impact312 on memory consumption. Between 2 and 100; default: 16313 314 dims: Dimensions of the vectors in collection. default: 1536315 """316 create_index_query = sqlalchemy.text(317 "CREATE INDEX IF NOT EXISTS {} "318 "ON {} USING hnsw (embedding {}) "319 "WITH ("320 "dim = :dim, "321 "m = :m, "322 "ef_construction = :ef_construction, "323 "ef = :ef"324 ");".format(325 quoted_name(self._index_name, True),326 quoted_name(self.collection_name, True),327 self._get_op_class(),328 )329 )330 331 with Session(self._conn) as session:332 # Create the HNSW index333 session.execute(334 create_index_query,335 {336 "dim": dims,337 "m": m,338 "ef_construction": ef_construction,339 "ef": ef_search,340 },341 )342 session.commit()343 self.logger.info("HNSW extension and index created successfully.")344 345 def drop_index(self) -> None:346 with Session(self._conn) as session:347 # Drop the HNSW index348 session.execute(349 sqlalchemy.text(350 "DROP INDEX IF EXISTS {}".format(351 quoted_name(self._index_name, True)352 )353 )354 )355 session.commit()356 357 def create_collection(self) -> None:358 if self.pre_delete_collection:359 self.delete_collection()360 self.drop_table()361 362 with self._conn.begin():363 try:364 self.EmbeddingStore.__table__.create(self._conn.engine)365 except ProgrammingError as e:366 # Duplicate table367 if e.code == "f405":368 pass369 else:370 raise e371 372 def delete_collection(self) -> None:373 self.logger.debug("Trying to delete collection")374 self.drop_table()375 376 @contextlib.contextmanager377 def _make_session(self) -> Generator[Session, None, None]:378 """Create a context manager for the session, bind to _conn string."""379 yield Session(self._conn)380 381 def delete(382 self,383 ids: Optional[List[str]] = None,384 **kwargs: Any,385 ) -> None:386 """Delete vectors by ids or uuids.387 388 Args:389 ids: List of ids to delete.390 """391 with Session(self._conn) as session:392 if ids is not None:393 self.logger.debug(394 "Trying to delete vectors by ids (represented by the model "395 "using the custom ids field)"396 )397 stmt = delete(self.EmbeddingStore).where(398 self.EmbeddingStore.custom_id.in_(ids)399 )400 session.execute(stmt)401 session.commit()402 403 @classmethod404 def _initialize_from_embeddings(405 cls,406 texts: List[str],407 embeddings: List[List[float]],408 embedding: Embeddings,409 metadatas: Optional[List[dict]] = None,410 ids: Optional[List[str]] = None,411 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,412 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,413 pre_delete_collection: bool = False,414 **kwargs: Any,415 ) -> Lantern:416 """417 Order of elements for lists `ids`, `embeddings`, `texts`, `metadatas`418 should match, so each row will be associated with correct values.419 420 Postgres connection string is required421 "Either pass it as `connection_string` parameter422 or set the LANTERN_CONNECTION_STRING environment variable.423 424 - `texts` texts to insert into collection.425 - `embeddings` an Embeddings to insert into collection426 - `embedding` is :class:`Embeddings` that will be used for427 embedding the text sent. If none is sent, then the428 multilingual Tensorflow Universal Sentence Encoder will be used.429 - `metadatas` row metadata to insert into collection.430 - `ids` row ids to insert into collection.431 - `collection_name` is the name of the collection to use. (default: langchain)432 - NOTE: This is the name of the table in which embedding data will be stored433 The table will be created when initializing the store (if not exists)434 So, make sure the user has the right permissions to create tables.435 - `distance_strategy` is the distance strategy to use. (default: EUCLIDEAN)436 - `EUCLIDEAN` is the euclidean distance.437 - `COSINE` is the cosine distance.438 - `HAMMING` is the hamming distance.439 - `pre_delete_collection` if True, will delete the collection if it exists.440 (default: False)441 - Useful for testing.442 """443 if ids is None:444 ids = [str(uuid.uuid4()) for _ in texts]445 446 if not metadatas:447 metadatas = [{} for _ in texts]448 449 connection_string = cls.__get_connection_string(kwargs)450 451 store = cls(452 connection_string=connection_string,453 collection_name=collection_name,454 embedding_function=embedding,455 pre_delete_collection=pre_delete_collection,456 distance_strategy=distance_strategy,457 )458 459 store.add_embeddings(460 texts=texts, embeddings=embeddings, metadatas=metadatas, ids=ids, **kwargs461 )462 463 store.create_hnsw_index(**kwargs)464 465 return store466 467 def add_embeddings(468 self,469 texts: List[str],470 embeddings: List[List[float]],471 metadatas: List[dict],472 ids: List[str],473 **kwargs: Any,474 ) -> None:475 with Session(self._conn) as session:476 for text, metadata, embedding, id in zip(texts, metadatas, embeddings, ids):477 embedding_store = self.EmbeddingStore(478 embedding=embedding,479 document=text,480 cmetadata=metadata,481 custom_id=id,482 )483 session.add(embedding_store)484 session.commit()485 486 def add_texts(487 self,488 texts: Iterable[str],489 metadatas: Optional[List[dict]] = None,490 ids: Optional[List[str]] = None,491 **kwargs: Any,492 ) -> List[str]:493 if ids is None:494 ids = [str(uuid.uuid4()) for _ in texts]495 496 embeddings = self.embedding_function.embed_documents(list(texts))497 498 if not metadatas:499 metadatas = [{} for _ in texts]500 501 with Session(self._conn) as session:502 for text, metadata, embedding, id in zip(texts, metadatas, embeddings, ids):503 embedding_store = self.EmbeddingStore(504 embedding=embedding,505 document=text,506 cmetadata=metadata,507 custom_id=id,508 )509 session.add(embedding_store)510 session.commit()511 512 return ids513 514 def _results_to_docs_and_scores(self, results: Any) -> List[Tuple[Document, float]]:515 """Return docs and scores from results."""516 docs = [517 (518 Document(519 page_content=result.EmbeddingStore.document,520 metadata=result.EmbeddingStore.cmetadata,521 ),522 result.distance if self.embedding_function is not None else None,523 )524 for result in results525 ]526 return docs527 528 def similarity_search(529 self,530 query: str,531 k: int = 4,532 filter: Optional[dict] = None,533 **kwargs: Any,534 ) -> List[Document]:535 embedding = self.embedding_function.embed_query(text=query)536 return self.similarity_search_by_vector(537 embedding=embedding,538 k=k,539 filter=filter,540 )541 542 def similarity_search_with_score(543 self,544 query: str,545 k: int = 4,546 filter: Optional[dict] = None,547 ) -> List[Tuple[Document, float]]:548 embedding = self.embedding_function.embed_query(query)549 docs = self.similarity_search_with_score_by_vector(550 embedding=embedding, k=k, filter=filter551 )552 return docs553 554 def similarity_search_with_score_by_vector(555 self,556 embedding: List[float],557 k: int = 4,558 filter: Optional[dict] = None,559 ) -> List[Tuple[Document, float]]:560 results = self.__query_collection(embedding=embedding, k=k, filter=filter)561 562 return self._results_to_docs_and_scores(results)563 564 def __query_collection(565 self,566 embedding: List[float],567 k: int = 4,568 filter: Optional[dict] = None,569 ) -> List[Any]:570 with Session(self._conn) as session:571 set_enable_seqscan_stmt = sqlalchemy.text("SET enable_seqscan = off")572 set_init_k = sqlalchemy.text("SET hnsw.init_k = :k")573 session.execute(set_enable_seqscan_stmt)574 session.execute(set_init_k, {"k": k})575 576 filter_by = None577 if filter is not None:578 filter_clauses = []579 for key, value in filter.items():580 IN = "in"581 if isinstance(value, dict) and IN in map(str.lower, value):582 value_case_insensitive = {583 k.lower(): v for k, v in value.items()584 }585 filter_by_metadata = self.EmbeddingStore.cmetadata[586 key587 ].astext.in_(value_case_insensitive[IN])588 filter_clauses.append(filter_by_metadata)589 else:590 filter_by_metadata = self.EmbeddingStore.cmetadata[591 key592 ].astext == str(value)593 filter_clauses.append(filter_by_metadata)594 595 filter_by = sqlalchemy.and_(*filter_clauses)596 597 embedding = self._typed_arg_for_distance(embedding)598 query = session.query(599 self.EmbeddingStore,600 getattr(func, self.distance_function)(601 self.EmbeddingStore.embedding, embedding602 ).label("distance"),603 ) # Specify the columns you need here, e.g., EmbeddingStore.embedding604 605 if filter_by is not None:606 query = query.filter(filter_by)607 608 results: List[QueryResult] = (609 query.order_by(610 self.EmbeddingStore.embedding.op(self._get_operator())(embedding)611 ) # Using PostgreSQL specific operator with the correct column name612 .limit(k)613 .all()614 )615 616 return results617 618 def similarity_search_by_vector(619 self,620 embedding: List[float],621 k: int = 4,622 filter: Optional[dict] = None,623 **kwargs: Any,624 ) -> List[Document]:625 docs_and_scores = self.similarity_search_with_score_by_vector(626 embedding=embedding, k=k, filter=filter627 )628 return _results_to_docs(docs_and_scores)629 630 @classmethod631 def from_texts(632 cls: Type[Lantern],633 texts: List[str],634 embedding: Embeddings,635 metadatas: Optional[List[dict]] = None,636 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,637 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,638 ids: Optional[List[str]] = None,639 pre_delete_collection: bool = False,640 **kwargs: Any,641 ) -> Lantern:642 """643 Initialize Lantern vectorstore from list of texts.644 The embeddings will be generated using `embedding` class provided.645 646 Order of elements for lists `ids`, `texts`, `metadatas` should match,647 so each row will be associated with correct values.648 649 Postgres connection string is required650 "Either pass it as `connection_string` parameter651 or set the LANTERN_CONNECTION_STRING environment variable.652 653 - `connection_string` is fully populated connection string for postgres database654 - `texts` texts to insert into collection.655 - `embedding` is :class:`Embeddings` that will be used for656 embedding the text sent. If none is sent, then the657 multilingual Tensorflow Universal Sentence Encoder will be used.658 - `metadatas` row metadata to insert into collection.659 - `collection_name` is the name of the collection to use. (default: langchain)660 - NOTE: This is the name of the table in which embedding data will be stored661 The table will be created when initializing the store (if not exists)662 So, make sure the user has the right permissions to create tables.663 - `distance_strategy` is the distance strategy to use. (default: EUCLIDEAN)664 - `EUCLIDEAN` is the euclidean distance.665 - `COSINE` is the cosine distance.666 - `HAMMING` is the hamming distance.667 - `ids` row ids to insert into collection.668 - `pre_delete_collection` if True, will delete the collection if it exists.669 (default: False)670 - Useful for testing.671 """672 embeddings = embedding.embed_documents(list(texts))673 674 return cls._initialize_from_embeddings(675 texts,676 embeddings,677 embedding,678 metadatas=metadatas,679 ids=ids,680 collection_name=collection_name,681 pre_delete_collection=pre_delete_collection,682 distance_strategy=distance_strategy,683 **kwargs,684 )685 686 @classmethod687 def from_embeddings(688 cls,689 text_embeddings: List[Tuple[str, List[float]]],690 embedding: Embeddings,691 metadatas: Optional[List[dict]] = None,692 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,693 ids: Optional[List[str]] = None,694 pre_delete_collection: bool = False,695 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,696 **kwargs: Any,697 ) -> Lantern:698 """Construct Lantern wrapper from raw documents and pre-699 generated embeddings.700 701 Postgres connection string is required702 "Either pass it as `connection_string` parameter703 or set the LANTERN_CONNECTION_STRING environment variable.704 705 Order of elements for lists `ids`, `text_embeddings`, `metadatas` should match,706 so each row will be associated with correct values.707 708 - `connection_string` is fully populated connection string for postgres database709 - `text_embeddings` is array with tuples (text, embedding)710 to insert into collection.711 - `embedding` is :class:`Embeddings` that will be used for712 embedding the text sent. If none is sent, then the713 multilingual Tensorflow Universal Sentence Encoder will be used.714 - `metadatas` row metadata to insert into collection.715 - `collection_name` is the name of the collection to use. (default: langchain)716 - NOTE: This is the name of the table in which embedding data will be stored717 The table will be created when initializing the store (if not exists)718 So, make sure the user has the right permissions to create tables.719 - `ids` row ids to insert into collection.720 - `pre_delete_collection` if True, will delete the collection if it exists.721 (default: False)722 - Useful for testing.723 - `distance_strategy` is the distance strategy to use. (default: EUCLIDEAN)724 - `EUCLIDEAN` is the euclidean distance.725 - `COSINE` is the cosine distance.726 - `HAMMING` is the hamming distance.727 """728 texts = [t[0] for t in text_embeddings]729 embeddings = [t[1] for t in text_embeddings]730 731 return cls._initialize_from_embeddings(732 texts,733 embeddings,734 embedding,735 metadatas=metadatas,736 ids=ids,737 collection_name=collection_name,738 pre_delete_collection=pre_delete_collection,739 distance_strategy=distance_strategy,740 **kwargs,741 )742 743 @classmethod744 def from_existing_index(745 cls: Type[Lantern],746 embedding: Embeddings,747 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,748 pre_delete_collection: bool = False,749 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,750 **kwargs: Any,751 ) -> Lantern:752 """753 Get instance of an existing Lantern store.This method will754 return the instance of the store without inserting any new755 embeddings756 757 Postgres connection string is required758 "Either pass it as `connection_string` parameter759 or set the LANTERN_CONNECTION_STRING environment variable.760 761 - `connection_string` is a postgres connection string.762 - `embedding` is :class:`Embeddings` that will be used for763 embedding the text sent. If none is sent, then the764 multilingual Tensorflow Universal Sentence Encoder will be used.765 - `collection_name` is the name of the collection to use. (default: langchain)766 - NOTE: This is the name of the table in which embedding data will be stored767 The table will be created when initializing the store (if not exists)768 So, make sure the user has the right permissions to create tables.769 - `ids` row ids to insert into collection.770 - `pre_delete_collection` if True, will delete the collection if it exists.771 (default: False)772 - Useful for testing.773 - `distance_strategy` is the distance strategy to use. (default: EUCLIDEAN)774 - `EUCLIDEAN` is the euclidean distance.775 - `COSINE` is the cosine distance.776 - `HAMMING` is the hamming distance.777 """778 connection_string = cls.__get_connection_string(kwargs)779 780 store = cls(781 connection_string=connection_string,782 collection_name=collection_name,783 embedding_function=embedding,784 pre_delete_collection=pre_delete_collection,785 distance_strategy=distance_strategy,786 )787 788 return store789 790 @classmethod791 def __get_connection_string(cls, kwargs: Dict[str, Any]) -> str:792 connection_string: str = get_from_dict_or_env(793 data=kwargs,794 key="connection_string",795 env_key="LANTERN_CONNECTION_STRING",796 )797 798 if not connection_string:799 raise ValueError(800 "Postgres connection string is required"801 "Either pass it as `connection_string` parameter"802 "or set the LANTERN_CONNECTION_STRING variable."803 )804 805 return connection_string806 807 @classmethod808 def from_documents(809 cls: Type[Lantern],810 documents: List[Document],811 embedding: Embeddings,812 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,813 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,814 ids: Optional[List[str]] = None,815 pre_delete_collection: bool = False,816 **kwargs: Any,817 ) -> Lantern:818 """819 Initialize a vector store with a set of documents.820 821 Postgres connection string is required822 "Either pass it as `connection_string` parameter823 or set the LANTERN_CONNECTION_STRING environment variable.824 825 - `connection_string` is a postgres connection string.826 - `documents` is list of :class:`Document` to initialize the vector store with827 - `embedding` is :class:`Embeddings` that will be used for828 embedding the text sent. If none is sent, then the829 multilingual Tensorflow Universal Sentence Encoder will be used.830 - `collection_name` is the name of the collection to use. (default: langchain)831 - NOTE: This is the name of the table in which embedding data will be stored832 The table will be created when initializing the store (if not exists)833 So, make sure the user has the right permissions to create tables.834 - `distance_strategy` is the distance strategy to use. (default: EUCLIDEAN)835 - `EUCLIDEAN` is the euclidean distance.836 - `COSINE` is the cosine distance.837 - `HAMMING` is the hamming distance.838 - `ids` row ids to insert into collection.839 - `pre_delete_collection` if True, will delete the collection if it exists.840 (default: False)841 - Useful for testing.842 """843 texts = [d.page_content for d in documents]844 metadatas = [d.metadata for d in documents]845 connection_string = cls.__get_connection_string(kwargs)846 847 kwargs["connection_string"] = connection_string848 849 return cls.from_texts(850 texts=texts,851 pre_delete_collection=pre_delete_collection,852 embedding=embedding,853 metadatas=metadatas,854 ids=ids,855 collection_name=collection_name,856 distance_strategy=distance_strategy,857 **kwargs,858 )859 860 def max_marginal_relevance_search_with_score_by_vector(861 self,862 embedding: List[float],863 k: int = 4,864 fetch_k: int = 20,865 lambda_mult: float = 0.5,866 filter: Optional[Dict[str, str]] = None,867 **kwargs: Any,868 ) -> List[Tuple[Document, float]]:869 """Return docs selected using the maximal marginal relevance with score870 to embedding vector.871 872 Maximal marginal relevance optimizes for similarity to query AND diversity873 among selected documents.874 875 Args:876 embedding: Embedding to look up documents similar to.877 k (int): Number of Documents to return. Defaults to 4.878 fetch_k (int): Number of Documents to fetch to pass to MMR algorithm.879 Defaults to 20.880 lambda_mult (float): Number between 0 and 1 that determines the degree881 of diversity among the results with 0 corresponding882 to maximum diversity and 1 to minimum diversity.883 Defaults to 0.5.884 filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.885 886 Returns:887 List[Tuple[Document, float]]: List of Documents selected by maximal marginal888 relevance to the query and score for each.889 """890 results = self.__query_collection(embedding=embedding, k=fetch_k, filter=filter)891 embedding_list = [result.EmbeddingStore.embedding for result in results]892 893 mmr_selected = maximal_marginal_relevance(894 np.array(embedding, dtype=np.float32),895 embedding_list,896 k=k,897 lambda_mult=lambda_mult,898 )899 900 candidates = self._results_to_docs_and_scores(results)901 902 return [r for i, r in enumerate(candidates) if i in mmr_selected]903 904 def max_marginal_relevance_search(905 self,906 query: str,907 k: int = 4,908 fetch_k: int = 20,909 lambda_mult: float = 0.5,910 filter: Optional[Dict[str, str]] = None,911 **kwargs: Any,912 ) -> List[Document]:913 """Return docs selected using the maximal marginal relevance.914 915 Maximal marginal relevance optimizes for similarity to query AND diversity916 among selected documents.917 918 Args:919 query (str): Text to look up documents similar to.920 k (int): Number of Documents to return. Defaults to 4.921 fetch_k (int): Number of Documents to fetch to pass to MMR algorithm.922 Defaults to 20.923 lambda_mult (float): Number between 0 and 1 that determines the degree924 of diversity among the results with 0 corresponding925 to maximum diversity and 1 to minimum diversity.926 Defaults to 0.5.927 filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.928 929 Returns:930 List[Document]: List of Documents selected by maximal marginal relevance.931 """932 embedding = self.embedding_function.embed_query(query)933 return self.max_marginal_relevance_search_by_vector(934 embedding,935 k=k,936 fetch_k=fetch_k,937 lambda_mult=lambda_mult,938 filter=filter,939 **kwargs,940 )941 942 def max_marginal_relevance_search_with_score(943 self,944 query: str,945 k: int = 4,946 fetch_k: int = 20,947 lambda_mult: float = 0.5,948 filter: Optional[dict] = None,949 **kwargs: Any,950 ) -> List[Tuple[Document, float]]:951 """Return docs selected using the maximal marginal relevance with score.952 953 Maximal marginal relevance optimizes for similarity to query AND diversity954 among selected documents.955 956 Args:957 query (str): Text to look up documents similar to.958 k (int): Number of Documents to return. Defaults to 4.959 fetch_k (int): Number of Documents to fetch to pass to MMR algorithm.960 Defaults to 20.961 lambda_mult (float): Number between 0 and 1 that determines the degree962 of diversity among the results with 0 corresponding963 to maximum diversity and 1 to minimum diversity.964 Defaults to 0.5.965 filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.966 967 Returns:968 List[Tuple[Document, float]]: List of Documents selected by maximal marginal969 relevance to the query and score for each.970 """971 embedding = self.embedding_function.embed_query(query)972 docs = self.max_marginal_relevance_search_with_score_by_vector(973 embedding=embedding,974 k=k,975 fetch_k=fetch_k,976 lambda_mult=lambda_mult,977 filter=filter,978 **kwargs,979 )980 return docs981 982 def max_marginal_relevance_search_by_vector(983 self,984 embedding: List[float],985 k: int = 4,986 fetch_k: int = 20,987 lambda_mult: float = 0.5,988 filter: Optional[Dict[str, str]] = None,989 **kwargs: Any,990 ) -> List[Document]:991 """Return docs selected using the maximal marginal relevance992 to embedding vector.993 994 Maximal marginal relevance optimizes for similarity to query AND diversity995 among selected documents.996 997 Args:998 embedding (str): Text to look up documents similar to.999 k (int): Number of Documents to return. Defaults to 4.1000 fetch_k (int): Number of Documents to fetch to pass to MMR algorithm.1001 Defaults to 20.1002 lambda_mult (float): Number between 0 and 1 that determines the degree1003 of diversity among the results with 0 corresponding1004 to maximum diversity and 1 to minimum diversity.1005 Defaults to 0.5.1006 filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.1007 1008 Returns:1009 List[Document]: List of Documents selected by maximal marginal relevance.1010 """1011 docs_and_scores = self.max_marginal_relevance_search_with_score_by_vector(1012 embedding,1013 k=k,1014 fetch_k=fetch_k,1015 lambda_mult=lambda_mult,1016 filter=filter,1017 **kwargs,1018 )1019 1020 return _results_to_docs(docs_and_scores)1021