codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import contextlib4import enum5import json6import logging7import uuid8from typing import (9 Any,10 Callable,11 Dict,12 Generator,13 Iterable,14 List,15 Mapping,16 Optional,17 Tuple,18 Type,19 Union,20)21 22import numpy as np23import sqlalchemy24from langchain_core._api import deprecated, warn_deprecated25from sqlalchemy import delete, func26from sqlalchemy.dialects.postgresql import JSON, JSONB, UUID27from sqlalchemy.orm import Session, relationship28 29try:30 from sqlalchemy.orm import declarative_base31except ImportError:32 from sqlalchemy.ext.declarative import declarative_base33 34try:35 from sqlalchemy import SQLColumnExpression36except ImportError:37 # for sqlalchemy < 238 SQLColumnExpression = Any # type: ignore[assignment,misc]39 40from langchain_core.documents import Document41from langchain_core.embeddings import Embeddings42from langchain_core.runnables.config import run_in_executor43from langchain_core.utils import get_from_dict_or_env44from langchain_core.vectorstores import VectorStore45 46from langchain_community.vectorstores.utils import maximal_marginal_relevance47 48 49class DistanceStrategy(str, enum.Enum):50 """Enumerator of the Distance strategies."""51 52 EUCLIDEAN = "l2"53 COSINE = "cosine"54 MAX_INNER_PRODUCT = "inner"55 56 57DEFAULT_DISTANCE_STRATEGY = DistanceStrategy.COSINE58 59Base = declarative_base() # type: Any60 61 62_LANGCHAIN_DEFAULT_COLLECTION_NAME = "langchain"63 64 65class BaseModel(Base):66 """Base model for the SQL stores."""67 68 __abstract__ = True69 uuid = sqlalchemy.Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)70 71 72_classes: Any = None73 74COMPARISONS_TO_NATIVE = {75 "$eq": "==",76 "$ne": "!=",77 "$lt": "<",78 "$lte": "<=",79 "$gt": ">",80 "$gte": ">=",81}82 83SPECIAL_CASED_OPERATORS = {84 "$in",85 "$nin",86 "$between",87}88 89TEXT_OPERATORS = {90 "$like",91 "$ilike",92}93 94LOGICAL_OPERATORS = {"$and", "$or"}95 96SUPPORTED_OPERATORS = (97 set(COMPARISONS_TO_NATIVE)98 .union(TEXT_OPERATORS)99 .union(LOGICAL_OPERATORS)100 .union(SPECIAL_CASED_OPERATORS)101)102 103 104def _get_embedding_collection_store(105 vector_dimension: Optional[int] = None, *, use_jsonb: bool = True106) -> Any:107 global _classes108 if _classes is not None:109 return _classes110 111 from pgvector.sqlalchemy import Vector112 113 class CollectionStore(BaseModel):114 """Collection store."""115 116 __tablename__ = "langchain_pg_collection"117 118 name = sqlalchemy.Column(sqlalchemy.String)119 cmetadata = sqlalchemy.Column(JSON)120 121 embeddings = relationship(122 "EmbeddingStore",123 back_populates="collection",124 passive_deletes=True,125 )126 127 @classmethod128 def get_by_name(129 cls, session: Session, name: str130 ) -> Optional["CollectionStore"]:131 return session.query(cls).filter(cls.name == name).first()132 133 @classmethod134 def get_or_create(135 cls,136 session: Session,137 name: str,138 cmetadata: Optional[dict] = None,139 ) -> Tuple["CollectionStore", bool]:140 """141 Get or create a collection.142 Returns [Collection, bool] where the bool is True if the collection was created.143 """ # noqa: E501144 created = False145 collection = cls.get_by_name(session, name)146 if collection:147 return collection, created148 149 collection = cls(name=name, cmetadata=cmetadata)150 session.add(collection)151 session.commit()152 created = True153 return collection, created154 155 if use_jsonb:156 # TODO(PRIOR TO LANDING): Create a gin index on the cmetadata field157 class EmbeddingStore(BaseModel):158 """Embedding store."""159 160 __tablename__ = "langchain_pg_embedding"161 162 collection_id = sqlalchemy.Column(163 UUID(as_uuid=True),164 sqlalchemy.ForeignKey(165 f"{CollectionStore.__tablename__}.uuid",166 ondelete="CASCADE",167 ),168 )169 collection = relationship(CollectionStore, back_populates="embeddings")170 171 embedding: Vector = sqlalchemy.Column(Vector(vector_dimension))172 document = sqlalchemy.Column(sqlalchemy.String, nullable=True)173 cmetadata = sqlalchemy.Column(JSONB, nullable=True)174 175 # custom_id : any user defined id176 custom_id = sqlalchemy.Column(sqlalchemy.String, nullable=True)177 178 __table_args__ = (179 sqlalchemy.Index(180 "ix_cmetadata_gin",181 "cmetadata",182 postgresql_using="gin",183 postgresql_ops={"cmetadata": "jsonb_path_ops"},184 ),185 )186 187 else:188 # For backwards comaptibilty with older versions of pgvector189 # This should be removed in the future (remove during migration)190 class EmbeddingStore(BaseModel): # type: ignore[no-redef]191 """Embedding store."""192 193 __tablename__ = "langchain_pg_embedding"194 195 collection_id = sqlalchemy.Column(196 UUID(as_uuid=True),197 sqlalchemy.ForeignKey(198 f"{CollectionStore.__tablename__}.uuid",199 ondelete="CASCADE",200 ),201 )202 collection = relationship(CollectionStore, back_populates="embeddings")203 204 embedding: Vector = sqlalchemy.Column(Vector(vector_dimension))205 document = sqlalchemy.Column(sqlalchemy.String, nullable=True)206 cmetadata = sqlalchemy.Column(JSON, nullable=True)207 208 # custom_id : any user defined id209 custom_id = sqlalchemy.Column(sqlalchemy.String, nullable=True)210 211 _classes = (EmbeddingStore, CollectionStore)212 213 return _classes214 215 216def _results_to_docs(docs_and_scores: Any) -> List[Document]:217 """Return docs from docs and scores."""218 return [doc for doc, _ in docs_and_scores]219 220 221@deprecated(222 since="0.0.31",223 message=(224 "This class is pending deprecation and may be removed in a future version. "225 "You can swap to using the `PGVector` "226 "implementation in `langchain_postgres`. "227 "Please read the guidelines in the doc-string of this class "228 "to follow prior to migrating as there are some differences "229 "between the implementations. "230 "See <https://github.com/langchain-ai/langchain-postgres> for details about "231 "the new implementation."232 ),233 alternative="from langchain_postgres import PGVector;",234 pending=True,235)236class PGVector(VectorStore):237 """`Postgres`/`PGVector` vector store.238 239 **DEPRECATED**: This class is pending deprecation and will likely receive240 no updates. An improved version of this class is available in241 `langchain_postgres` as `PGVector`. Please use that class instead.242 243 When migrating please keep in mind that:244 * The new implementation works with psycopg3, not with psycopg2245 (This implementation does not work with psycopg3).246 * Filtering syntax has changed to use $ prefixed operators for JSONB247 metadata fields. (New implementation only uses JSONB field for metadata)248 * The new implementation made some schema changes to address issues249 with the existing implementation. So you will need to re-create250 your tables and re-index your data or else carry out a manual251 migration.252 253 To use, you should have the ``pgvector`` python package installed.254 255 Args:256 connection_string: Postgres connection string.257 embedding_function: Any embedding function implementing258 `langchain.embeddings.base.Embeddings` interface.259 embedding_length: The length of the embedding vector. (default: None)260 NOTE: This is not mandatory. Defining it will prevent vectors of261 any other size to be added to the embeddings table but, without it,262 the embeddings can't be indexed.263 collection_name: The name of the collection to use. (default: langchain)264 NOTE: This is not the name of the table, but the name of the collection.265 The tables will be created when initializing the store (if not exists)266 So, make sure the user has the right permissions to create tables.267 distance_strategy: The distance strategy to use. (default: COSINE)268 pre_delete_collection: If True, will delete the collection if it exists.269 (default: False). Useful for testing.270 engine_args: SQLAlchemy's create engine arguments.271 use_jsonb: Use JSONB instead of JSON for metadata. (default: True)272 Strongly discouraged from using JSON as it's not as efficient273 for querying.274 It's provided here for backwards compatibility with older versions,275 and will be removed in the future.276 create_extension: If True, will create the vector extension if it doesn't exist.277 disabling creation is useful when using ReadOnly Databases.278 279 Example:280 281 .. code-block:: python282 283 from langchain_community.vectorstores import PGVector284 from langchain_community.embeddings.openai import OpenAIEmbeddings285 CONNECTION_STRING = "postgresql+psycopg2://hwc@localhost:5432/test3"286 COLLECTION_NAME = "state_of_the_union_test"287 embeddings = OpenAIEmbeddings()288 vectorestore = PGVector.from_documents(289 embedding=embeddings,290 documents=docs,291 collection_name=COLLECTION_NAME,292 connection_string=CONNECTION_STRING,293 use_jsonb=True,294 295 """ # noqa: E501296 297 def __init__(298 self,299 connection_string: str,300 embedding_function: Embeddings,301 embedding_length: Optional[int] = None,302 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,303 collection_metadata: Optional[dict] = None,304 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,305 pre_delete_collection: bool = False,306 logger: Optional[logging.Logger] = None,307 relevance_score_fn: Optional[Callable[[float], float]] = None,308 *,309 connection: Optional[sqlalchemy.engine.Connection] = None,310 engine_args: Optional[dict[str, Any]] = None,311 use_jsonb: bool = False,312 create_extension: bool = True,313 ) -> None:314 """Initialize the PGVector store."""315 self.connection_string = connection_string316 self.embedding_function = embedding_function317 self._embedding_length = embedding_length318 self.collection_name = collection_name319 self.collection_metadata = collection_metadata320 self._distance_strategy = distance_strategy321 self.pre_delete_collection = pre_delete_collection322 self.logger = logger or logging.getLogger(__name__)323 self.override_relevance_score_fn = relevance_score_fn324 self.engine_args = engine_args or {}325 self._bind = connection if connection else self._create_engine()326 self.use_jsonb = use_jsonb327 self.create_extension = create_extension328 329 if not use_jsonb:330 # Replace with a deprecation warning.331 warn_deprecated(332 "0.0.29",333 pending=True,334 message=(335 "Please use JSONB instead of JSON for metadata. "336 "This change will allow for more efficient querying that "337 "involves filtering based on metadata. "338 "Please note that filtering operators have been changed "339 "when using JSONB metadata to be prefixed with a $ sign "340 "to avoid name collisions with columns. "341 "If you're using an existing database, you will need to create a "342 "db migration for your metadata column to be JSONB and update your "343 "queries to use the new operators. "344 ),345 alternative=(346 "Instantiate with use_jsonb=True to use JSONB instead "347 "of JSON for metadata."348 ),349 )350 self.__post_init__()351 352 def __post_init__(353 self,354 ) -> None:355 """Initialize the store."""356 if self.create_extension:357 self.create_vector_extension()358 359 EmbeddingStore, CollectionStore = _get_embedding_collection_store(360 self._embedding_length, use_jsonb=self.use_jsonb361 )362 self.CollectionStore = CollectionStore363 self.EmbeddingStore = EmbeddingStore364 self.create_tables_if_not_exists()365 self.create_collection()366 367 def __del__(self) -> None:368 if isinstance(self._bind, sqlalchemy.engine.Connection):369 self._bind.close()370 371 @property372 def embeddings(self) -> Embeddings:373 return self.embedding_function374 375 def _create_engine(self) -> sqlalchemy.engine.Engine:376 return sqlalchemy.create_engine(url=self.connection_string, **self.engine_args)377 378 def create_vector_extension(self) -> None:379 try:380 with Session(self._bind) as session:381 # The advisor lock fixes issue arising from concurrent382 # creation of the vector extension.383 # https://github.com/langchain-ai/langchain/issues/12933384 # For more information see:385 # https://www.postgresql.org/docs/16/explicit-locking.html#ADVISORY-LOCKS386 statement = sqlalchemy.text(387 "BEGIN;"388 "SELECT pg_advisory_xact_lock(1573678846307946496);"389 "CREATE EXTENSION IF NOT EXISTS vector;"390 "COMMIT;"391 )392 session.execute(statement)393 session.commit()394 except Exception as e:395 raise Exception(f"Failed to create vector extension: {e}") from e396 397 def create_tables_if_not_exists(self) -> None:398 with Session(self._bind) as session, session.begin():399 Base.metadata.create_all(session.get_bind())400 401 def drop_tables(self) -> None:402 with Session(self._bind) as session, session.begin():403 Base.metadata.drop_all(session.get_bind())404 405 def create_collection(self) -> None:406 if self.pre_delete_collection:407 self.delete_collection()408 with Session(self._bind) as session:409 self.CollectionStore.get_or_create(410 session, self.collection_name, cmetadata=self.collection_metadata411 )412 413 def delete_collection(self) -> None:414 self.logger.debug("Trying to delete collection")415 with Session(self._bind) as session:416 collection = self.get_collection(session)417 if not collection:418 self.logger.warning("Collection not found")419 return420 session.delete(collection)421 session.commit()422 423 @contextlib.contextmanager424 def _make_session(self) -> Generator[Session, None, None]:425 """Create a context manager for the session, bind to _conn string."""426 yield Session(self._bind)427 428 def delete(429 self,430 ids: Optional[List[str]] = None,431 collection_only: bool = False,432 **kwargs: Any,433 ) -> None:434 """Delete vectors by ids or uuids.435 436 Args:437 ids: List of ids to delete.438 collection_only: Only delete ids in the collection.439 """440 with Session(self._bind) as session:441 if ids is not None:442 self.logger.debug(443 "Trying to delete vectors by ids (represented by the model "444 "using the custom ids field)"445 )446 447 stmt = delete(self.EmbeddingStore)448 449 if collection_only:450 collection = self.get_collection(session)451 if not collection:452 self.logger.warning("Collection not found")453 return454 455 stmt = stmt.where(456 self.EmbeddingStore.collection_id == collection.uuid457 )458 459 stmt = stmt.where(self.EmbeddingStore.custom_id.in_(ids))460 session.execute(stmt)461 session.commit()462 463 def get_collection(self, session: Session) -> Any:464 return self.CollectionStore.get_by_name(session, self.collection_name)465 466 @classmethod467 def _from(468 cls,469 texts: List[str],470 embeddings: List[List[float]],471 embedding: Embeddings,472 metadatas: Optional[List[dict]] = None,473 ids: Optional[List[str]] = None,474 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,475 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,476 connection_string: Optional[str] = None,477 pre_delete_collection: bool = False,478 *,479 use_jsonb: bool = False,480 **kwargs: Any,481 ) -> PGVector:482 if ids is None:483 ids = [str(uuid.uuid4()) for _ in texts]484 485 if not metadatas:486 metadatas = [{} for _ in texts]487 if connection_string is None:488 connection_string = cls.get_connection_string(kwargs)489 490 store = cls(491 connection_string=connection_string,492 collection_name=collection_name,493 embedding_function=embedding,494 distance_strategy=distance_strategy,495 pre_delete_collection=pre_delete_collection,496 use_jsonb=use_jsonb,497 **kwargs,498 )499 500 store.add_embeddings(501 texts=texts, embeddings=embeddings, metadatas=metadatas, ids=ids, **kwargs502 )503 504 return store505 506 def add_embeddings(507 self,508 texts: Iterable[str],509 embeddings: List[List[float]],510 metadatas: Optional[List[dict]] = None,511 ids: Optional[List[str]] = None,512 **kwargs: Any,513 ) -> List[str]:514 """Add embeddings to the vectorstore.515 516 Args:517 texts: Iterable of strings to add to the vectorstore.518 embeddings: List of list of embedding vectors.519 metadatas: List of metadatas associated with the texts.520 kwargs: vectorstore specific parameters521 """522 if ids is None:523 ids = [str(uuid.uuid4()) for _ in texts]524 525 if not metadatas:526 metadatas = [{} for _ in texts]527 528 with Session(self._bind) as session:529 collection = self.get_collection(session)530 if not collection:531 raise ValueError("Collection not found")532 documents = []533 for text, metadata, embedding, id in zip(texts, metadatas, embeddings, ids):534 embedding_store = self.EmbeddingStore(535 embedding=embedding,536 document=text,537 cmetadata=metadata,538 custom_id=id,539 collection_id=collection.uuid,540 )541 documents.append(embedding_store)542 session.bulk_save_objects(documents)543 session.commit()544 545 return ids546 547 def add_texts(548 self,549 texts: Iterable[str],550 metadatas: Optional[List[dict]] = None,551 ids: Optional[List[str]] = None,552 **kwargs: Any,553 ) -> List[str]:554 """Run more texts through the embeddings and add to the vectorstore.555 556 Args:557 texts: Iterable of strings to add to the vectorstore.558 metadatas: Optional list of metadatas associated with the texts.559 kwargs: vectorstore specific parameters560 561 Returns:562 List of ids from adding the texts into the vectorstore.563 """564 embeddings = self.embedding_function.embed_documents(list(texts))565 return self.add_embeddings(566 texts=texts, embeddings=embeddings, metadatas=metadatas, ids=ids, **kwargs567 )568 569 def similarity_search(570 self,571 query: str,572 k: int = 4,573 filter: Optional[dict] = None,574 **kwargs: Any,575 ) -> List[Document]:576 """Run similarity search with PGVector with distance.577 578 Args:579 query (str): Query text to search for.580 k (int): Number of results to return. Defaults to 4.581 filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.582 583 Returns:584 List of Documents most similar to the query.585 """586 embedding = self.embedding_function.embed_query(text=query)587 return self.similarity_search_by_vector(588 embedding=embedding,589 k=k,590 filter=filter,591 )592 593 def similarity_search_with_score(594 self,595 query: str,596 k: int = 4,597 filter: Optional[dict] = None,598 ) -> List[Tuple[Document, float]]:599 """Return docs most similar to query.600 601 Args:602 query: Text to look up documents similar to.603 k: Number of Documents to return. Defaults to 4.604 filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.605 606 Returns:607 List of Documents most similar to the query and score for each.608 """609 embedding = self.embedding_function.embed_query(query)610 docs = self.similarity_search_with_score_by_vector(611 embedding=embedding, k=k, filter=filter612 )613 return docs614 615 @property616 def distance_strategy(self) -> Any:617 if self._distance_strategy == DistanceStrategy.EUCLIDEAN:618 return self.EmbeddingStore.embedding.l2_distance619 elif self._distance_strategy == DistanceStrategy.COSINE:620 return self.EmbeddingStore.embedding.cosine_distance621 elif self._distance_strategy == DistanceStrategy.MAX_INNER_PRODUCT:622 return self.EmbeddingStore.embedding.max_inner_product623 else:624 raise ValueError(625 f"Got unexpected value for distance: {self._distance_strategy}. "626 f"Should be one of {', '.join([ds.value for ds in DistanceStrategy])}."627 )628 629 def similarity_search_with_score_by_vector(630 self,631 embedding: List[float],632 k: int = 4,633 filter: Optional[dict] = None,634 ) -> List[Tuple[Document, float]]:635 results = self._query_collection(embedding=embedding, k=k, filter=filter)636 637 return self._results_to_docs_and_scores(results)638 639 def _results_to_docs_and_scores(self, results: Any) -> List[Tuple[Document, float]]:640 """Return docs and scores from results."""641 docs = [642 (643 Document(644 page_content=result.EmbeddingStore.document,645 metadata=result.EmbeddingStore.cmetadata,646 ),647 result.distance if self.embedding_function is not None else None,648 )649 for result in results650 ]651 return docs652 653 def _handle_field_filter(654 self,655 field: str,656 value: Any,657 ) -> SQLColumnExpression:658 """Create a filter for a specific field.659 660 Args:661 field: name of field662 value: value to filter663 If provided as is then this will be an equality filter664 If provided as a dictionary then this will be a filter, the key665 will be the operator and the value will be the value to filter by666 667 Returns:668 sqlalchemy expression669 """670 if not isinstance(field, str):671 raise ValueError(672 f"field should be a string but got: {type(field)} with value: {field}"673 )674 675 if field.startswith("$"):676 raise ValueError(677 f"Invalid filter condition. Expected a field but got an operator: "678 f"{field}"679 )680 681 # Allow [a-zA-Z0-9_], disallow $ for now until we support escape characters682 if not field.isidentifier():683 raise ValueError(684 f"Invalid field name: {field}. Expected a valid identifier."685 )686 687 if isinstance(value, dict):688 # This is a filter specification689 if len(value) != 1:690 raise ValueError(691 "Invalid filter condition. Expected a value which "692 "is a dictionary with a single key that corresponds to an operator "693 f"but got a dictionary with {len(value)} keys. The first few "694 f"keys are: {list(value.keys())[:3]}"695 )696 operator, filter_value = list(value.items())[0]697 # Verify that that operator is an operator698 if operator not in SUPPORTED_OPERATORS:699 raise ValueError(700 f"Invalid operator: {operator}. "701 f"Expected one of {SUPPORTED_OPERATORS}"702 )703 else: # Then we assume an equality operator704 operator = "$eq"705 filter_value = value706 707 if operator in COMPARISONS_TO_NATIVE:708 # Then we implement an equality filter709 # native is trusted input710 native = COMPARISONS_TO_NATIVE[operator]711 return func.jsonb_path_match(712 self.EmbeddingStore.cmetadata,713 f"$.{field} {native} $value",714 json.dumps({"value": filter_value}),715 )716 elif operator == "$between":717 # Use AND with two comparisons718 low, high = filter_value719 720 lower_bound = func.jsonb_path_match(721 self.EmbeddingStore.cmetadata,722 f"$.{field} >= $value",723 json.dumps({"value": low}),724 )725 upper_bound = func.jsonb_path_match(726 self.EmbeddingStore.cmetadata,727 f"$.{field} <= $value",728 json.dumps({"value": high}),729 )730 return sqlalchemy.and_(lower_bound, upper_bound)731 elif operator in {"$in", "$nin", "$like", "$ilike"}:732 # We'll do force coercion to text733 if operator in {"$in", "$nin"}:734 for val in filter_value:735 if not isinstance(val, (str, int, float)):736 raise NotImplementedError(737 f"Unsupported type: {type(val)} for value: {val}"738 )739 740 queried_field = self.EmbeddingStore.cmetadata[field].astext741 742 if operator in {"$in"}:743 return queried_field.in_([str(val) for val in filter_value])744 elif operator in {"$nin"}:745 return queried_field.not_in([str(val) for val in filter_value])746 elif operator in {"$like"}:747 return queried_field.like(filter_value)748 elif operator in {"$ilike"}:749 return queried_field.ilike(filter_value)750 else:751 raise NotImplementedError()752 else:753 raise NotImplementedError()754 755 def _create_filter_clause_deprecated(756 self, key: str, value: dict[str, Any]757 ) -> SQLColumnExpression:758 """Deprecated functionality.759 760 This is for backwards compatibility with the JSON based schema for metadata.761 It uses incorrect operator syntax (operators are not prefixed with $).762 763 This implementation is not efficient, and has bugs associated with764 the way that it handles numeric filter clauses.765 """766 IN, NIN, BETWEEN, GT, LT, NE = "in", "nin", "between", "gt", "lt", "ne"767 EQ, LIKE, CONTAINS, OR, AND = "eq", "like", "contains", "or", "and"768 769 value_case_insensitive = {k.lower(): v for k, v in value.items()}770 if IN in map(str.lower, value):771 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext.in_(772 value_case_insensitive[IN]773 )774 elif NIN in map(str.lower, value):775 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext.not_in(776 value_case_insensitive[NIN]777 )778 elif BETWEEN in map(str.lower, value):779 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext.between(780 str(value_case_insensitive[BETWEEN][0]),781 str(value_case_insensitive[BETWEEN][1]),782 )783 elif GT in map(str.lower, value):784 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext > str(785 value_case_insensitive[GT]786 )787 elif LT in map(str.lower, value):788 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext < str(789 value_case_insensitive[LT]790 )791 elif NE in map(str.lower, value):792 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext != str(793 value_case_insensitive[NE]794 )795 elif EQ in map(str.lower, value):796 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext == str(797 value_case_insensitive[EQ]798 )799 elif LIKE in map(str.lower, value):800 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext.like(801 value_case_insensitive[LIKE]802 )803 elif CONTAINS in map(str.lower, value):804 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext.contains(805 value_case_insensitive[CONTAINS]806 )807 elif OR in map(str.lower, value):808 or_clauses = [809 self._create_filter_clause_deprecated(key, sub_value)810 for sub_value in value_case_insensitive[OR]811 ]812 filter_by_metadata = sqlalchemy.or_(*or_clauses)813 elif AND in map(str.lower, value):814 and_clauses = [815 self._create_filter_clause_deprecated(key, sub_value)816 for sub_value in value_case_insensitive[AND]817 ]818 filter_by_metadata = sqlalchemy.and_(*and_clauses)819 820 else:821 filter_by_metadata = None822 823 return filter_by_metadata824 825 def _create_filter_clause_json_deprecated(826 self, filter: Mapping[str, Union[str, dict[str, Any]]]827 ) -> List[SQLColumnExpression]:828 """Convert filters from IR to SQL clauses.829 830 **DEPRECATED** This functionality will be deprecated in the future.831 832 It implements translation of filters for a schema that uses JSON833 for metadata rather than the JSONB field which is more efficient834 for querying.835 """836 filter_clauses = []837 for key, value in filter.items():838 if isinstance(value, dict):839 filter_by_metadata = self._create_filter_clause_deprecated(key, value)840 841 if filter_by_metadata is not None:842 filter_clauses.append(filter_by_metadata)843 else:844 filter_by_metadata = self.EmbeddingStore.cmetadata[key].astext == str(845 value846 )847 filter_clauses.append(filter_by_metadata)848 return filter_clauses849 850 def _create_filter_clause(self, filters: Any) -> Any:851 """Convert LangChain IR filter representation to matching SQLAlchemy clauses.852 853 At the top level, we still don't know if we're working with a field854 or an operator for the keys. After we've determined that we can855 call the appropriate logic to handle filter creation.856 857 Args:858 filters: Dictionary of filters to apply to the query.859 860 Returns:861 SQLAlchemy clause to apply to the query.862 """863 if isinstance(filters, dict):864 if len(filters) == 1:865 # The only operators allowed at the top level are $AND and $OR866 # First check if an operator or a field867 key, value = list(filters.items())[0]868 if key.startswith("$"):869 # Then it's an operator870 if key.lower() not in ["$and", "$or"]:871 raise ValueError(872 f"Invalid filter condition. Expected $and or $or "873 f"but got: {key}"874 )875 else:876 # Then it's a field877 return self._handle_field_filter(key, filters[key])878 879 # Here we handle the $and and $or operators880 if not isinstance(value, list):881 raise ValueError(882 f"Expected a list, but got {type(value)} for value: {value}"883 )884 if key.lower() == "$and":885 and_ = [self._create_filter_clause(el) for el in value]886 if len(and_) > 1:887 return sqlalchemy.and_(*and_)888 elif len(and_) == 1:889 return and_[0]890 else:891 raise ValueError(892 "Invalid filter condition. Expected a dictionary "893 "but got an empty dictionary"894 )895 elif key.lower() == "$or":896 or_ = [self._create_filter_clause(el) for el in value]897 if len(or_) > 1:898 return sqlalchemy.or_(*or_)899 elif len(or_) == 1:900 return or_[0]901 else:902 raise ValueError(903 "Invalid filter condition. Expected a dictionary "904 "but got an empty dictionary"905 )906 else:907 raise ValueError(908 f"Invalid filter condition. Expected $and or $or but got: {key}"909 )910 elif len(filters) > 1:911 # Then all keys have to be fields (they cannot be operators)912 for key in filters.keys():913 if key.startswith("$"):914 raise ValueError(915 f"Invalid filter condition. Expected a field but got: {key}"916 )917 # These should all be fields and combined using an $and operator918 and_ = [self._handle_field_filter(k, v) for k, v in filters.items()]919 if len(and_) > 1:920 return sqlalchemy.and_(*and_)921 elif len(and_) == 1:922 return and_[0]923 else:924 raise ValueError(925 "Invalid filter condition. Expected a dictionary "926 "but got an empty dictionary"927 )928 else:929 raise ValueError("Got an empty dictionary for filters.")930 else:931 raise ValueError(932 f"Invalid type: Expected a dictionary but got type: {type(filters)}"933 )934 935 def _query_collection(936 self,937 embedding: List[float],938 k: int = 4,939 filter: Optional[Dict[str, str]] = None,940 ) -> List[Any]:941 """Query the collection."""942 with Session(self._bind) as session:943 collection = self.get_collection(session)944 if not collection:945 raise ValueError("Collection not found")946 947 filter_by = [self.EmbeddingStore.collection_id == collection.uuid]948 if filter:949 if self.use_jsonb:950 filter_clauses = self._create_filter_clause(filter)951 if filter_clauses is not None:952 filter_by.append(filter_clauses)953 else:954 # Old way of doing things955 filter_clauses = self._create_filter_clause_json_deprecated(filter)956 filter_by.extend(filter_clauses)957 958 _type = self.EmbeddingStore959 960 results: List[Any] = (961 session.query(962 self.EmbeddingStore,963 self.distance_strategy(embedding).label("distance"),964 )965 .filter(*filter_by)966 .order_by(sqlalchemy.asc("distance"))967 .join(968 self.CollectionStore,969 self.EmbeddingStore.collection_id == self.CollectionStore.uuid,970 )971 .limit(k)972 .all()973 )974 975 return results976 977 def similarity_search_by_vector(978 self,979 embedding: List[float],980 k: int = 4,981 filter: Optional[dict] = None,982 **kwargs: Any,983 ) -> List[Document]:984 """Return docs most similar to embedding vector.985 986 Args:987 embedding: Embedding to look up documents similar to.988 k: Number of Documents to return. Defaults to 4.989 filter (Optional[Dict[str, str]]): Filter by metadata. Defaults to None.990 991 Returns:992 List of Documents most similar to the query vector.993 """994 docs_and_scores = self.similarity_search_with_score_by_vector(995 embedding=embedding, k=k, filter=filter996 )997 return _results_to_docs(docs_and_scores)998 999 @classmethod1000 def from_texts(1001 cls: Type[PGVector],1002 texts: List[str],1003 embedding: Embeddings,1004 metadatas: Optional[List[dict]] = None,1005 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,1006 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,1007 ids: Optional[List[str]] = None,1008 pre_delete_collection: bool = False,1009 *,1010 use_jsonb: bool = False,1011 **kwargs: Any,1012 ) -> PGVector:1013 """1014 Return VectorStore initialized from texts and embeddings.1015 Postgres connection string is required1016 "Either pass it as a parameter1017 or set the PGVECTOR_CONNECTION_STRING environment variable.1018 """1019 embeddings = embedding.embed_documents(list(texts))1020 1021 return cls._from(1022 texts,1023 embeddings,1024 embedding,1025 metadatas=metadatas,1026 ids=ids,1027 collection_name=collection_name,1028 distance_strategy=distance_strategy,1029 pre_delete_collection=pre_delete_collection,1030 use_jsonb=use_jsonb,1031 **kwargs,1032 )1033 1034 @classmethod1035 def from_embeddings(1036 cls,1037 text_embeddings: List[Tuple[str, List[float]]],1038 embedding: Embeddings,1039 metadatas: Optional[List[dict]] = None,1040 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,1041 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,1042 ids: Optional[List[str]] = None,1043 pre_delete_collection: bool = False,1044 **kwargs: Any,1045 ) -> PGVector:1046 """Construct PGVector wrapper from raw documents and pre-1047 generated embeddings.1048 1049 Return VectorStore initialized from documents and embeddings.1050 Postgres connection string is required1051 "Either pass it as a parameter1052 or set the PGVECTOR_CONNECTION_STRING environment variable.1053 1054 Example:1055 .. code-block:: python1056 1057 from langchain_community.vectorstores import PGVector1058 from langchain_community.embeddings import OpenAIEmbeddings1059 embeddings = OpenAIEmbeddings()1060 text_embeddings = embeddings.embed_documents(texts)1061 text_embedding_pairs = list(zip(texts, text_embeddings))1062 faiss = PGVector.from_embeddings(text_embedding_pairs, embeddings)1063 """1064 texts = [t[0] for t in text_embeddings]1065 embeddings = [t[1] for t in text_embeddings]1066 1067 return cls._from(1068 texts,1069 embeddings,1070 embedding,1071 metadatas=metadatas,1072 ids=ids,1073 collection_name=collection_name,1074 distance_strategy=distance_strategy,1075 pre_delete_collection=pre_delete_collection,1076 **kwargs,1077 )1078 1079 @classmethod1080 def from_existing_index(1081 cls: Type[PGVector],1082 embedding: Embeddings,1083 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,1084 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,1085 pre_delete_collection: bool = False,1086 **kwargs: Any,1087 ) -> PGVector:1088 """1089 Get instance of an existing PGVector store.This method will1090 return the instance of the store without inserting any new1091 embeddings1092 """1093 1094 connection_string = cls.get_connection_string(kwargs)1095 1096 store = cls(1097 connection_string=connection_string,1098 collection_name=collection_name,1099 embedding_function=embedding,1100 distance_strategy=distance_strategy,1101 pre_delete_collection=pre_delete_collection,1102 )1103 1104 return store1105 1106 @classmethod1107 def get_connection_string(cls, kwargs: Dict[str, Any]) -> str:1108 connection_string: str = get_from_dict_or_env(1109 data=kwargs,1110 key="connection_string",1111 env_key="PGVECTOR_CONNECTION_STRING",1112 )1113 1114 if not connection_string:1115 raise ValueError(1116 "Postgres connection string is required"1117 "Either pass it as a parameter"1118 "or set the PGVECTOR_CONNECTION_STRING environment variable."1119 )1120 1121 return connection_string1122 1123 @classmethod1124 def from_documents(1125 cls: Type[PGVector],1126 documents: List[Document],1127 embedding: Embeddings,1128 collection_name: str = _LANGCHAIN_DEFAULT_COLLECTION_NAME,1129 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,1130 ids: Optional[List[str]] = None,1131 pre_delete_collection: bool = False,1132 *,1133 use_jsonb: bool = False,1134 **kwargs: Any,1135 ) -> PGVector:1136 """1137 Return VectorStore initialized from documents and embeddings.1138 Postgres connection string is required1139 "Either pass it as a parameter1140 or set the PGVECTOR_CONNECTION_STRING environment variable.1141 """1142 1143 texts = [d.page_content for d in documents]1144 metadatas = [d.metadata for d in documents]1145 connection_string = cls.get_connection_string(kwargs)1146 1147 kwargs["connection_string"] = connection_string1148 1149 return cls.from_texts(1150 texts=texts,1151 pre_delete_collection=pre_delete_collection,1152 embedding=embedding,1153 distance_strategy=distance_strategy,1154 metadatas=metadatas,1155 ids=ids,1156 collection_name=collection_name,1157 use_jsonb=use_jsonb,1158 **kwargs,1159 )1160 1161 @classmethod1162 def connection_string_from_db_params(1163 cls,1164 driver: str,1165 host: str,1166 port: int,1167 database: str,1168 user: str,1169 password: str,1170 ) -> str:1171 """Return connection string from database parameters."""1172 return f"postgresql+{driver}://{user}:{password}@{host}:{port}/{database}"1173 1174 def _select_relevance_score_fn(self) -> Callable[[float], float]:1175 """1176 The 'correct' relevance function1177 may differ depending on a few things, including:1178 - the distance / similarity metric used by the VectorStore1179 - the scale of your embeddings (OpenAI's are unit normed. Many others are not!)1180 - embedding dimensionality1181 - etc.1182 """1183 if self.override_relevance_score_fn is not None:1184 return self.override_relevance_score_fn1185 1186 # Default strategy is to rely on distance strategy provided1187 # in vectorstore constructor1188 if self._distance_strategy == DistanceStrategy.COSINE:1189 return self._cosine_relevance_score_fn1190 elif self._distance_strategy == DistanceStrategy.EUCLIDEAN:1191 return self._euclidean_relevance_score_fn1192 elif self._distance_strategy == DistanceStrategy.MAX_INNER_PRODUCT:1193 return self._max_inner_product_relevance_score_fn1194 else:1195 raise ValueError(1196 "No supported normalization function"1197 f" for distance_strategy of {self._distance_strategy}."1198 "Consider providing relevance_score_fn to PGVector constructor."1199 )1200