codekingpro/portable-devtools
114k
1from __future__ import annotations2 3import enum4import logging5import os6from hashlib import md57from typing import (8 Any,9 Callable,10 Dict,11 Iterable,12 List,13 Optional,14 Tuple,15 Type,16)17 18import numpy as np19from langchain_core._api.deprecation import deprecated20from langchain_core.documents import Document21from langchain_core.embeddings import Embeddings22from langchain_core.utils import get_from_dict_or_env23from langchain_core.vectorstores import VectorStore24 25from langchain_community.graphs import Neo4jGraph26from langchain_community.vectorstores.utils import (27 DistanceStrategy,28 maximal_marginal_relevance,29)30 31DEFAULT_DISTANCE_STRATEGY = DistanceStrategy.COSINE32DISTANCE_MAPPING = {33 DistanceStrategy.EUCLIDEAN_DISTANCE: "euclidean",34 DistanceStrategy.COSINE: "cosine",35}36 37COMPARISONS_TO_NATIVE = {38 "$eq": "=",39 "$ne": "<>",40 "$lt": "<",41 "$lte": "<=",42 "$gt": ">",43 "$gte": ">=",44}45 46SPECIAL_CASED_OPERATORS = {47 "$in",48 "$nin",49 "$between",50}51 52TEXT_OPERATORS = {53 "$like",54 "$ilike",55}56 57LOGICAL_OPERATORS = {"$and", "$or"}58 59SUPPORTED_OPERATORS = (60 set(COMPARISONS_TO_NATIVE)61 .union(TEXT_OPERATORS)62 .union(LOGICAL_OPERATORS)63 .union(SPECIAL_CASED_OPERATORS)64)65 66 67@deprecated(68 since="0.3.8",69 removal="1.0",70 alternative_import="langchain_neo4j.vectorstores.neo4j_vector.SearchType",71)72class SearchType(str, enum.Enum):73 """Enumerator of the Distance strategies."""74 75 VECTOR = "vector"76 HYBRID = "hybrid"77 78 79DEFAULT_SEARCH_TYPE = SearchType.VECTOR80 81 82@deprecated(83 since="0.3.8",84 removal="1.0",85 alternative_import="langchain_neo4j.vectorstores.neo4j_vector.IndexType",86)87class IndexType(str, enum.Enum):88 """Enumerator of the index types."""89 90 NODE = "NODE"91 RELATIONSHIP = "RELATIONSHIP"92 93 94DEFAULT_INDEX_TYPE = IndexType.NODE95 96 97@deprecated(98 since="0.3.8",99 removal="1.0",100 alternative_import="langchain_neo4j.vectorstores.neo4j_vector._get_search_index_query",101)102def _get_search_index_query(103 search_type: SearchType, index_type: IndexType = DEFAULT_INDEX_TYPE104) -> str:105 if index_type == IndexType.NODE:106 type_to_query_map = {107 SearchType.VECTOR: (108 "CALL db.index.vector.queryNodes($index, $k, $embedding) "109 "YIELD node, score "110 ),111 SearchType.HYBRID: (112 "CALL { "113 "CALL db.index.vector.queryNodes($index, $k, $embedding) "114 "YIELD node, score "115 "WITH collect({node:node, score:score}) AS nodes, max(score) AS max "116 "UNWIND nodes AS n "117 # We use 0 as min118 "RETURN n.node AS node, (n.score / max) AS score UNION "119 "CALL db.index.fulltext.queryNodes($keyword_index, $query, "120 "{limit: $k}) YIELD node, score "121 "WITH collect({node:node, score:score}) AS nodes, max(score) AS max "122 "UNWIND nodes AS n "123 # We use 0 as min124 "RETURN n.node AS node, (n.score / max) AS score "125 "} "126 # dedup127 "WITH node, max(score) AS score ORDER BY score DESC LIMIT $k "128 ),129 }130 return type_to_query_map[search_type]131 else:132 return (133 "CALL db.index.vector.queryRelationships($index, $k, $embedding) "134 "YIELD relationship, score "135 )136 137 138@deprecated(139 since="0.3.8",140 removal="1.0",141 alternative_import="langchain_neo4j.vectorstores.neo4j_vector.check_if_not_null",142)143def check_if_not_null(props: List[str], values: List[Any]) -> None:144 """Check if the values are not None or empty string"""145 for prop, value in zip(props, values):146 if not value:147 raise ValueError(f"Parameter `{prop}` must not be None or empty string")148 149 150@deprecated(151 since="0.3.8",152 removal="1.0",153 alternative_import="langchain_neo4j.vectorstores.neo4j_vector.sort_by_index_name",154)155def sort_by_index_name(156 lst: List[Dict[str, Any]], index_name: str157) -> List[Dict[str, Any]]:158 """Sort first element to match the index_name if exists"""159 return sorted(lst, key=lambda x: x.get("name") != index_name)160 161 162@deprecated(163 since="0.3.8",164 removal="1.0",165 alternative_import="langchain_neo4j.vectorstores.neo4j_vector.remove_lucene_chars",166)167def remove_lucene_chars(text: str) -> str:168 """Remove Lucene special characters"""169 special_chars = [170 "+",171 "-",172 "&",173 "|",174 "!",175 "(",176 ")",177 "{",178 "}",179 "[",180 "]",181 "^",182 '"',183 "~",184 "*",185 "?",186 ":",187 "\\",188 ]189 for char in special_chars:190 if char in text:191 text = text.replace(char, " ")192 return text.strip()193 194 195@deprecated(196 since="0.3.8",197 removal="1.0",198 alternative_import="langchain_neo4j.vectorstores.neo4j_vector.dict_to_yaml_str",199)200def dict_to_yaml_str(input_dict: Dict, indent: int = 0) -> str:201 """202 Convert a dictionary to a YAML-like string without using external libraries.203 204 Parameters:205 - input_dict (dict): The dictionary to convert.206 - indent (int): The current indentation level.207 208 Returns:209 - str: The YAML-like string representation of the input dictionary.210 """211 yaml_str = ""212 for key, value in input_dict.items():213 padding = " " * indent214 if isinstance(value, dict):215 yaml_str += f"{padding}{key}:\n{dict_to_yaml_str(value, indent + 1)}"216 elif isinstance(value, list):217 yaml_str += f"{padding}{key}:\n"218 for item in value:219 yaml_str += f"{padding}- {item}\n"220 else:221 yaml_str += f"{padding}{key}: {value}\n"222 return yaml_str223 224 225@deprecated(226 since="0.3.8",227 removal="1.0",228 alternative_import="langchain_neo4j.vectorstores.neo4j_vector.combine_queries",229)230def combine_queries(231 input_queries: List[Tuple[str, Dict[str, Any]]], operator: str232) -> Tuple[str, Dict[str, Any]]:233 """Combine multiple queries with an operator."""234 235 # Initialize variables to hold the combined query and parameters236 combined_query: str = ""237 combined_params: Dict = {}238 param_counter: Dict = {}239 240 for query, params in input_queries:241 # Process each query fragment and its parameters242 new_query = query243 for param, value in params.items():244 # Update the parameter name to ensure uniqueness245 if param in param_counter:246 param_counter[param] += 1247 else:248 param_counter[param] = 1249 new_param_name = f"{param}_{param_counter[param]}"250 251 # Replace the parameter in the query fragment252 new_query = new_query.replace(f"${param}", f"${new_param_name}")253 # Add the parameter to the combined parameters dictionary254 combined_params[new_param_name] = value255 256 # Combine the query fragments with an AND operator257 if combined_query:258 combined_query += f" {operator} "259 combined_query += f"({new_query})"260 261 return combined_query, combined_params262 263 264@deprecated(265 since="0.3.8",266 removal="1.0",267 alternative_import="langchain_neo4j.vectorstores.neo4j_vector.collect_params",268)269def collect_params(270 input_data: List[Tuple[str, Dict[str, str]]],271) -> Tuple[List[str], Dict[str, Any]]:272 """Transform the input data into the desired format.273 274 Args:275 - input_data (list of tuples): Input data to transform.276 Each tuple contains a string and a dictionary.277 278 Returns:279 - tuple: A tuple containing a list of strings and a dictionary.280 """281 # Initialize variables to hold the output parts282 query_parts = []283 params = {}284 285 # Loop through each item in the input data286 for query_part, param in input_data:287 # Append the query part to the list288 query_parts.append(query_part)289 # Update the params dictionary with the param dictionary290 params.update(param)291 292 # Return the transformed data293 return (query_parts, params)294 295 296@deprecated(297 since="0.3.8",298 removal="1.0",299 alternative_import="langchain_neo4j.vectorstores.neo4j_vector._handle_field_filter",300)301def _handle_field_filter(302 field: str, value: Any, param_number: int = 1303) -> Tuple[str, Dict]:304 """Create a filter for a specific field.305 306 Args:307 field: name of field308 value: value to filter309 If provided as is then this will be an equality filter310 If provided as a dictionary then this will be a filter, the key311 will be the operator and the value will be the value to filter by312 param_number: sequence number of parameters used to map between param313 dict and Cypher snippet314 315 Returns a tuple of316 - Cypher filter snippet317 - Dictionary with parameters used in filter snippet318 """319 if not isinstance(field, str):320 raise ValueError(321 f"field should be a string but got: {type(field)} with value: {field}"322 )323 324 if field.startswith("$"):325 raise ValueError(326 f"Invalid filter condition. Expected a field but got an operator: {field}"327 )328 329 # Allow [a-zA-Z0-9_], disallow $ for now until we support escape characters330 if not field.isidentifier():331 raise ValueError(f"Invalid field name: {field}. Expected a valid identifier.")332 333 if isinstance(value, dict):334 # This is a filter specification335 if len(value) != 1:336 raise ValueError(337 "Invalid filter condition. Expected a value which "338 "is a dictionary with a single key that corresponds to an operator "339 f"but got a dictionary with {len(value)} keys. The first few "340 f"keys are: {list(value.keys())[:3]}"341 )342 operator, filter_value = list(value.items())[0]343 # Verify that that operator is an operator344 if operator not in SUPPORTED_OPERATORS:345 raise ValueError(346 f"Invalid operator: {operator}. Expected one of {SUPPORTED_OPERATORS}"347 )348 else: # Then we assume an equality operator349 operator = "$eq"350 filter_value = value351 352 if operator in COMPARISONS_TO_NATIVE:353 # Then we implement an equality filter354 # native is trusted input355 native = COMPARISONS_TO_NATIVE[operator]356 query_snippet = f"n.`{field}` {native} $param_{param_number}"357 query_param = {f"param_{param_number}": filter_value}358 return (query_snippet, query_param)359 elif operator == "$between":360 low, high = filter_value361 query_snippet = (362 f"$param_{param_number}_low <= n.`{field}` <= $param_{param_number}_high"363 )364 query_param = {365 f"param_{param_number}_low": low,366 f"param_{param_number}_high": high,367 }368 return (query_snippet, query_param)369 370 elif operator in {"$in", "$nin", "$like", "$ilike"}:371 # We'll do force coercion to text372 if operator in {"$in", "$nin"}:373 for val in filter_value:374 if not isinstance(val, (str, int, float)):375 raise NotImplementedError(376 f"Unsupported type: {type(val)} for value: {val}"377 )378 if operator in {"$in"}:379 query_snippet = f"n.`{field}` IN $param_{param_number}"380 query_param = {f"param_{param_number}": filter_value}381 return (query_snippet, query_param)382 elif operator in {"$nin"}:383 query_snippet = f"n.`{field}` NOT IN $param_{param_number}"384 query_param = {f"param_{param_number}": filter_value}385 return (query_snippet, query_param)386 elif operator in {"$like"}:387 query_snippet = f"n.`{field}` CONTAINS $param_{param_number}"388 query_param = {f"param_{param_number}": filter_value.rstrip("%")}389 return (query_snippet, query_param)390 elif operator in {"$ilike"}:391 query_snippet = f"toLower(n.`{field}`) CONTAINS $param_{param_number}"392 query_param = {f"param_{param_number}": filter_value.rstrip("%")}393 return (query_snippet, query_param)394 else:395 raise NotImplementedError()396 else:397 raise NotImplementedError()398 399 400@deprecated(401 since="0.3.8",402 removal="1.0",403 alternative_import="langchain_neo4j.vectorstores.neo4j_vector.construct_metadata_filter",404)405def construct_metadata_filter(filter: Dict[str, Any]) -> Tuple[str, Dict]:406 """Construct a metadata filter.407 408 Args:409 filter: A dictionary representing the filter condition.410 411 Returns:412 Tuple[str, Dict]413 """414 415 if isinstance(filter, dict):416 if len(filter) == 1:417 # The only operators allowed at the top level are $AND and $OR418 # First check if an operator or a field419 key, value = list(filter.items())[0]420 if key.startswith("$"):421 # Then it's an operator422 if key.lower() not in ["$and", "$or"]:423 raise ValueError(424 f"Invalid filter condition. Expected $and or $or but got: {key}"425 )426 else:427 # Then it's a field428 return _handle_field_filter(key, filter[key])429 430 # Here we handle the $and and $or operators431 if not isinstance(value, list):432 raise ValueError(433 f"Expected a list, but got {type(value)} for value: {value}"434 )435 if key.lower() == "$and":436 and_ = combine_queries(437 [construct_metadata_filter(el) for el in value], "AND"438 )439 if len(and_) >= 1:440 return and_441 else:442 raise ValueError(443 "Invalid filter condition. Expected a dictionary "444 "but got an empty dictionary"445 )446 elif key.lower() == "$or":447 or_ = combine_queries(448 [construct_metadata_filter(el) for el in value], "OR"449 )450 if len(or_) >= 1:451 return or_452 else:453 raise ValueError(454 "Invalid filter condition. Expected a dictionary "455 "but got an empty dictionary"456 )457 else:458 raise ValueError(459 f"Invalid filter condition. Expected $and or $or but got: {key}"460 )461 elif len(filter) > 1:462 # Then all keys have to be fields (they cannot be operators)463 for key in filter.keys():464 if key.startswith("$"):465 raise ValueError(466 f"Invalid filter condition. Expected a field but got: {key}"467 )468 # These should all be fields and combined using an $and operator469 and_multiple = collect_params(470 [471 _handle_field_filter(k, v, index)472 for index, (k, v) in enumerate(filter.items())473 ]474 )475 if len(and_multiple) >= 1:476 return " AND ".join(and_multiple[0]), and_multiple[1]477 else:478 raise ValueError(479 "Invalid filter condition. Expected a dictionary "480 "but got an empty dictionary"481 )482 else:483 raise ValueError("Got an empty dictionary for filters.")484 485 486@deprecated(487 since="0.3.8",488 removal="1.0",489 alternative_import="langchain_neo4j.Neo4jVector",490)491class Neo4jVector(VectorStore):492 """`Neo4j` vector index.493 494 To use, you should have the ``neo4j`` python package installed.495 496 Args:497 url: Neo4j connection url498 username: Neo4j username.499 password: Neo4j password500 database: Optionally provide Neo4j database501 Defaults to "neo4j"502 embedding: Any embedding function implementing503 `langchain.embeddings.base.Embeddings` interface.504 distance_strategy: The distance strategy to use. (default: COSINE)505 search_type: The type of search to be performed, either506 'vector' or 'hybrid'507 node_label: The label used for nodes in the Neo4j database.508 (default: "Chunk")509 embedding_node_property: The property name in Neo4j to store embeddings.510 (default: "embedding")511 text_node_property: The property name in Neo4j to store the text.512 (default: "text")513 retrieval_query: The Cypher query to be used for customizing retrieval.514 If empty, a default query will be used.515 index_type: The type of index to be used, either516 'NODE' or 'RELATIONSHIP'517 pre_delete_collection: If True, will delete existing data if it exists.518 (default: False). Useful for testing.519 520 Example:521 .. code-block:: python522 523 from langchain_community.vectorstores.neo4j_vector import Neo4jVector524 from langchain_community.embeddings.openai import OpenAIEmbeddings525 526 url="bolt://localhost:7687"527 username="neo4j"528 password="pleaseletmein"529 embeddings = OpenAIEmbeddings()530 vectorestore = Neo4jVector.from_documents(531 embedding=embeddings,532 documents=docs,533 url=url534 username=username,535 password=password,536 )537 538 539 """540 541 def __init__(542 self,543 embedding: Embeddings,544 *,545 search_type: SearchType = SearchType.VECTOR,546 username: Optional[str] = None,547 password: Optional[str] = None,548 url: Optional[str] = None,549 keyword_index_name: Optional[str] = "keyword",550 database: Optional[str] = None,551 index_name: str = "vector",552 node_label: str = "Chunk",553 embedding_node_property: str = "embedding",554 text_node_property: str = "text",555 distance_strategy: DistanceStrategy = DEFAULT_DISTANCE_STRATEGY,556 logger: Optional[logging.Logger] = None,557 pre_delete_collection: bool = False,558 retrieval_query: str = "",559 relevance_score_fn: Optional[Callable[[float], float]] = None,560 index_type: IndexType = DEFAULT_INDEX_TYPE,561 graph: Optional[Neo4jGraph] = None,562 ) -> None:563 try:564 import neo4j565 except ImportError:566 raise ImportError(567 "Could not import neo4j python package. "568 "Please install it with `pip install neo4j`."569 )570 571 # Allow only cosine and euclidean distance strategies572 if distance_strategy not in [573 DistanceStrategy.EUCLIDEAN_DISTANCE,574 DistanceStrategy.COSINE,575 ]:576 raise ValueError(577 "distance_strategy must be either 'EUCLIDEAN_DISTANCE' or 'COSINE'"578 )579 580 # Graph object takes precedent over env or input params581 if graph:582 self._driver = graph._driver583 self._database = graph._database584 else:585 # Handle if the credentials are environment variables586 # Support URL for backwards compatibility587 if not url:588 url = os.environ.get("NEO4J_URL")589 590 url = get_from_dict_or_env({"url": url}, "url", "NEO4J_URI")591 username = get_from_dict_or_env(592 {"username": username}, "username", "NEO4J_USERNAME"593 )594 password = get_from_dict_or_env(595 {"password": password}, "password", "NEO4J_PASSWORD"596 )597 database = get_from_dict_or_env(598 {"database": database}, "database", "NEO4J_DATABASE", "neo4j"599 )600 601 self._driver = neo4j.GraphDatabase.driver(url, auth=(username, password))602 self._database = database603 # Verify connection604 try:605 self._driver.verify_connectivity()606 except neo4j.exceptions.ServiceUnavailable:607 raise ValueError(608 "Could not connect to Neo4j database. "609 "Please ensure that the url is correct"610 )611 except neo4j.exceptions.AuthError:612 raise ValueError(613 "Could not connect to Neo4j database. "614 "Please ensure that the username and password are correct"615 )616 617 self.schema = ""618 # Verify if the version support vector index619 self._is_enterprise = False620 self.verify_version()621 622 # Verify that required values are not null623 check_if_not_null(624 [625 "index_name",626 "node_label",627 "embedding_node_property",628 "text_node_property",629 ],630 [index_name, node_label, embedding_node_property, text_node_property],631 )632 633 self.embedding = embedding634 self._distance_strategy = distance_strategy635 self.index_name = index_name636 self.keyword_index_name = keyword_index_name637 self.node_label = node_label638 self.embedding_node_property = embedding_node_property639 self.text_node_property = text_node_property640 self.logger = logger or logging.getLogger(__name__)641 self.override_relevance_score_fn = relevance_score_fn642 self.retrieval_query = retrieval_query643 self.search_type = search_type644 self._index_type = index_type645 # Calculate embedding dimension646 self.embedding_dimension = len(embedding.embed_query("foo"))647 648 # Delete existing data if flagged649 if pre_delete_collection:650 from neo4j.exceptions import DatabaseError651 652 self.query(653 f"MATCH (n:`{self.node_label}`) "654 "CALL (n) { DETACH DELETE n } "655 "IN TRANSACTIONS OF 10000 ROWS;"656 )657 # Delete index658 try:659 self.query(f"DROP INDEX {self.index_name}")660 except DatabaseError: # Index didn't exist yet661 pass662 663 def query(664 self,665 query: str,666 *,667 params: Optional[dict] = None,668 ) -> List[Dict[str, Any]]:669 """Query Neo4j database with retries and exponential backoff.670 671 Args:672 query (str): The Cypher query to execute.673 params (dict, optional): Dictionary of query parameters. Defaults to {}.674 675 Returns:676 List[Dict[str, Any]]: List of dictionaries containing the query results.677 """678 from neo4j import Query679 from neo4j.exceptions import Neo4jError680 681 params = params or {}682 try:683 data, _, _ = self._driver.execute_query(684 query, database_=self._database, parameters_=params685 )686 return [r.data() for r in data]687 except Neo4jError as e:688 if not (689 (690 ( # isCallInTransactionError691 e.code == "Neo.DatabaseError.Statement.ExecutionFailed"692 or e.code693 == "Neo.DatabaseError.Transaction.TransactionStartFailed"694 )695 and "in an implicit transaction" in e.message696 )697 or ( # isPeriodicCommitError698 e.code == "Neo.ClientError.Statement.SemanticError"699 and (700 "in an open transaction is not possible" in e.message701 or "tried to execute in an explicit transaction" in e.message702 )703 )704 ):705 raise706 # Fallback to allow implicit transactions707 with self._driver.session(database=self._database) as session:708 data = session.run(Query(text=query), params)709 return [r.data() for r in data]710 711 def verify_version(self) -> None:712 """713 Check if the connected Neo4j database version supports vector indexing.714 715 Queries the Neo4j database to retrieve its version and compares it716 against a target version (5.11.0) that is known to support vector717 indexing. Raises a ValueError if the connected Neo4j version is718 not supported.719 """720 db_data = self.query("CALL dbms.components()")721 version = db_data[0]["versions"][0]722 if "aura" in version:723 version_tuple = tuple(map(int, version.split("-")[0].split("."))) + (0,)724 else:725 version_tuple = tuple(map(int, version.split(".")))726 727 target_version = (5, 11, 0)728 729 if version_tuple < target_version:730 raise ValueError(731 "Version index is only supported in Neo4j version 5.11 or greater"732 )733 734 # Flag for metadata filtering735 metadata_target_version = (5, 18, 0)736 if version_tuple < metadata_target_version:737 self.support_metadata_filter = False738 else:739 self.support_metadata_filter = True740 # Flag for enterprise741 self._is_enterprise = True if db_data[0]["edition"] == "enterprise" else False742 743 def retrieve_existing_index(self) -> Tuple[Optional[int], Optional[str]]:744 """745 Check if the vector index exists in the Neo4j database746 and returns its embedding dimension.747 748 This method queries the Neo4j database for existing indexes749 and attempts to retrieve the dimension of the vector index750 with the specified name. If the index exists, its dimension is returned.751 If the index doesn't exist, `None` is returned.752 753 Returns:754 int or None: The embedding dimension of the existing index if found.755 """756 757 index_information = self.query(758 "SHOW INDEXES YIELD name, type, entityType, labelsOrTypes, "759 "properties, options WHERE type = 'VECTOR' AND (name = $index_name "760 "OR (labelsOrTypes[0] = $node_label AND "761 "properties[0] = $embedding_node_property)) "762 "RETURN name, entityType, labelsOrTypes, properties, options ",763 params={764 "index_name": self.index_name,765 "node_label": self.node_label,766 "embedding_node_property": self.embedding_node_property,767 },768 )769 # sort by index_name770 index_information = sort_by_index_name(index_information, self.index_name)771 try:772 self.index_name = index_information[0]["name"]773 self.node_label = index_information[0]["labelsOrTypes"][0]774 self.embedding_node_property = index_information[0]["properties"][0]775 self._index_type = index_information[0]["entityType"]776 embedding_dimension = None777 index_config = index_information[0]["options"]["indexConfig"]778 if "vector.dimensions" in index_config:779 embedding_dimension = index_config["vector.dimensions"]780 781 return embedding_dimension, index_information[0]["entityType"]782 except IndexError:783 return None, None784 785 def retrieve_existing_fts_index(786 self, text_node_properties: List[str] = []787 ) -> Optional[str]:788 """789 Check if the fulltext index exists in the Neo4j database790 791 This method queries the Neo4j database for existing fts indexes792 with the specified name.793 794 Returns:795 (Tuple): keyword index information796 """797 798 index_information = self.query(799 "SHOW INDEXES YIELD name, type, labelsOrTypes, properties, options "800 "WHERE type = 'FULLTEXT' AND (name = $keyword_index_name "801 "OR (labelsOrTypes = [$node_label] AND "802 "properties = $text_node_property)) "803 "RETURN name, labelsOrTypes, properties, options ",804 params={805 "keyword_index_name": self.keyword_index_name,806 "node_label": self.node_label,807 "text_node_property": text_node_properties or [self.text_node_property],808 },809 )810 # sort by index_name811 index_information = sort_by_index_name(index_information, self.index_name)812 try:813 self.keyword_index_name = index_information[0]["name"]814 self.text_node_property = index_information[0]["properties"][0]815 node_label = index_information[0]["labelsOrTypes"][0]816 return node_label817 except IndexError:818 return None819 820 def create_new_index(self) -> None:821 """822 This method constructs a Cypher query and executes it823 to create a new vector index in Neo4j.824 """825 index_query = (826 f"CREATE VECTOR INDEX {self.index_name} IF NOT EXISTS "827 f"FOR (m:`{self.node_label}`) ON m.`{self.embedding_node_property}` "828 "OPTIONS { indexConfig: { "829 "`vector.dimensions`: toInteger($embedding_dimension), "830 "`vector.similarity_function`: $similarity_metric }}"831 )832 833 parameters = {834 "embedding_dimension": self.embedding_dimension,835 "similarity_metric": DISTANCE_MAPPING[self._distance_strategy],836 }837 self.query(index_query, params=parameters)838 839 def create_new_keyword_index(self, text_node_properties: List[str] = []) -> None:840 """841 This method constructs a Cypher query and executes it842 to create a new full text index in Neo4j.843 """844 node_props = text_node_properties or [self.text_node_property]845 fts_index_query = (846 f"CREATE FULLTEXT INDEX {self.keyword_index_name} "847 f"FOR (n:`{self.node_label}`) ON EACH "848 f"[{', '.join(['n.`' + el + '`' for el in node_props])}]"849 )850 self.query(fts_index_query)851 852 @property853 def embeddings(self) -> Embeddings:854 return self.embedding855 856 @classmethod857 def __from(858 cls,859 texts: List[str],860 embeddings: List[List[float]],861 embedding: Embeddings,862 metadatas: Optional[List[dict]] = None,863 ids: Optional[List[str]] = None,864 create_id_index: bool = True,865 search_type: SearchType = SearchType.VECTOR,866 **kwargs: Any,867 ) -> Neo4jVector:868 if ids is None:869 ids = [md5(text.encode("utf-8")).hexdigest() for text in texts]870 871 if not metadatas:872 metadatas = [{} for _ in texts]873 874 store = cls(875 embedding=embedding,876 search_type=search_type,877 **kwargs,878 )879 # Check if the vector index already exists880 embedding_dimension, index_type = store.retrieve_existing_index()881 882 # Raise error if relationship index type883 if index_type == "RELATIONSHIP":884 raise ValueError(885 "Data ingestion is not supported with relationship vector index."886 )887 888 # If the vector index doesn't exist yet889 if not index_type:890 store.create_new_index()891 # If the index already exists, check if embedding dimensions match892 elif (893 embedding_dimension and not store.embedding_dimension == embedding_dimension894 ):895 raise ValueError(896 f"Index with name {store.index_name} already exists."897 "The provided embedding function and vector index "898 "dimensions do not match.\n"899 f"Embedding function dimension: {store.embedding_dimension}\n"900 f"Vector index dimension: {embedding_dimension}"901 )902 903 if search_type == SearchType.HYBRID:904 fts_node_label = store.retrieve_existing_fts_index()905 # If the FTS index doesn't exist yet906 if not fts_node_label:907 store.create_new_keyword_index()908 else: # Validate that FTS and Vector index use the same information909 if not fts_node_label == store.node_label:910 raise ValueError(911 "Vector and keyword index don't index the same node label"912 )913 914 # Create unique constraint for faster import915 if create_id_index:916 store.query(917 "CREATE CONSTRAINT IF NOT EXISTS "918 f"FOR (n:`{store.node_label}`) REQUIRE n.id IS UNIQUE;"919 )920 921 store.add_embeddings(922 texts=texts, embeddings=embeddings, metadatas=metadatas, ids=ids, **kwargs923 )924 925 return store926 927 def add_embeddings(928 self,929 texts: Iterable[str],930 embeddings: List[List[float]],931 metadatas: Optional[List[dict]] = None,932 ids: Optional[List[str]] = None,933 **kwargs: Any,934 ) -> List[str]:935 """Add embeddings to the vectorstore.936 937 Args:938 texts: Iterable of strings to add to the vectorstore.939 embeddings: List of list of embedding vectors.940 metadatas: List of metadatas associated with the texts.941 kwargs: vectorstore specific parameters942 """943 if ids is None:944 ids = [md5(text.encode("utf-8")).hexdigest() for text in texts]945 946 if not metadatas:947 metadatas = [{} for _ in texts]948 949 import_query = (950 "UNWIND $data AS row "951 "CALL (row) { WITH row "952 f"MERGE (c:`{self.node_label}` {{id: row.id}}) "953 "WITH c, row "954 f"CALL db.create.setNodeVectorProperty(c, "955 f"'{self.embedding_node_property}', row.embedding) "956 f"SET c.`{self.text_node_property}` = row.text "957 "SET c += row.metadata "958 "} IN TRANSACTIONS OF 1000 ROWS "959 )960 961 parameters = {962 "data": [963 {"text": text, "metadata": metadata, "embedding": embedding, "id": id}964 for text, metadata, embedding, id in zip(965 texts, metadatas, embeddings, ids966 )967 ]968 }969 970 self.query(import_query, params=parameters)971 972 return ids973 974 def add_texts(975 self,976 texts: Iterable[str],977 metadatas: Optional[List[dict]] = None,978 ids: Optional[List[str]] = None,979 **kwargs: Any,980 ) -> List[str]:981 """Run more texts through the embeddings and add to the vectorstore.982 983 Args:984 texts: Iterable of strings to add to the vectorstore.985 metadatas: Optional list of metadatas associated with the texts.986 kwargs: vectorstore specific parameters987 988 Returns:989 List of ids from adding the texts into the vectorstore.990 """991 embeddings = self.embedding.embed_documents(list(texts))992 return self.add_embeddings(993 texts=texts, embeddings=embeddings, metadatas=metadatas, ids=ids, **kwargs994 )995 996 def similarity_search(997 self,998 query: str,999 k: int = 4,1000 params: Dict[str, Any] = {},1001 filter: Optional[Dict[str, Any]] = None,1002 **kwargs: Any,1003 ) -> List[Document]:1004 """Run similarity search with Neo4jVector.1005 1006 Args:1007 query (str): Query text to search for.1008 k (int): Number of results to return. Defaults to 4.1009 params (Dict[str, Any]): The search params for the index type.1010 Defaults to empty dict.1011 filter (Optional[Dict[str, Any]]): Dictionary of argument(s) to1012 filter on metadata.1013 Defaults to None.1014 1015 Returns:1016 List of Documents most similar to the query.1017 """1018 embedding = self.embedding.embed_query(text=query)1019 return self.similarity_search_by_vector(1020 embedding=embedding,1021 k=k,1022 query=query,1023 params=params,1024 filter=filter,1025 **kwargs,1026 )1027 1028 def similarity_search_with_score(1029 self,1030 query: str,1031 k: int = 4,1032 params: Dict[str, Any] = {},1033 filter: Optional[Dict[str, Any]] = None,1034 **kwargs: Any,1035 ) -> List[Tuple[Document, float]]:1036 """Return docs most similar to query.1037 1038 Args:1039 query: Text to look up documents similar to.1040 k: Number of Documents to return. Defaults to 4.1041 params (Dict[str, Any]): The search params for the index type.1042 Defaults to empty dict.1043 filter (Optional[Dict[str, Any]]): Dictionary of argument(s) to1044 filter on metadata.1045 Defaults to None.1046 1047 Returns:1048 List of Documents most similar to the query and score for each1049 """1050 embedding = self.embedding.embed_query(query)1051 docs = self.similarity_search_with_score_by_vector(1052 embedding=embedding,1053 k=k,1054 query=query,1055 params=params,1056 filter=filter,1057 **kwargs,1058 )1059 return docs1060 1061 def similarity_search_with_score_by_vector(1062 self,1063 embedding: List[float],1064 k: int = 4,1065 filter: Optional[Dict[str, Any]] = None,1066 params: Dict[str, Any] = {},1067 **kwargs: Any,1068 ) -> List[Tuple[Document, float]]:1069 """1070 Perform a similarity search in the Neo4j database using a1071 given vector and return the top k similar documents with their scores.1072 1073 This method uses a Cypher query to find the top k documents that1074 are most similar to a given embedding. The similarity is measured1075 using a vector index in the Neo4j database. The results are returned1076 as a list of tuples, each containing a Document object and1077 its similarity score.1078 1079 Args:1080 embedding (List[float]): The embedding vector to compare against.1081 k (int, optional): The number of top similar documents to retrieve.1082 filter (Optional[Dict[str, Any]]): Dictionary of argument(s) to1083 filter on metadata.1084 Defaults to None.1085 params (Dict[str, Any]): The search params for the index type.1086 Defaults to empty dict.1087 1088 Returns:1089 List[Tuple[Document, float]]: A list of tuples, each containing1090 a Document object and its similarity score.1091 """1092 if filter:1093 # Verify that 5.18 or later is used1094 if not self.support_metadata_filter:1095 raise ValueError(1096 "Metadata filtering is only supported in "1097 "Neo4j version 5.18 or greater"1098 )1099 # Metadata filtering and hybrid doesn't work1100 if self.search_type == SearchType.HYBRID:1101 raise ValueError(1102 "Metadata filtering can't be use in combination with "1103 "a hybrid search approach"1104 )1105 parallel_query = (1106 "CYPHER runtime = parallel parallelRuntimeSupport=all "1107 if self._is_enterprise1108 else ""1109 )1110 base_index_query = parallel_query + (1111 f"MATCH (n:`{self.node_label}`) WHERE "1112 f"n.`{self.embedding_node_property}` IS NOT NULL AND "1113 f"size(n.`{self.embedding_node_property}`) = "1114 f"toInteger({self.embedding_dimension}) AND "1115 )1116 base_cosine_query = (1117 " WITH n as node, vector.similarity.cosine("1118 f"n.`{self.embedding_node_property}`, "1119 "$embedding) AS score ORDER BY score DESC LIMIT toInteger($k) "1120 )1121 filter_snippets, filter_params = construct_metadata_filter(filter)1122 index_query = base_index_query + filter_snippets + base_cosine_query1123 1124 else:1125 index_query = _get_search_index_query(self.search_type, self._index_type)1126 filter_params = {}1127 1128 if self._index_type == IndexType.RELATIONSHIP:1129 if kwargs.get("return_embeddings"):1130 default_retrieval = (1131 f"RETURN relationship.`{self.text_node_property}` AS text, score, "1132 f"relationship {{.*, `{self.text_node_property}`: Null, "1133 f"`{self.embedding_node_property}`: Null, id: Null, "1134 f"_embedding_: relationship.`{self.embedding_node_property}`}} "1135 "AS metadata"1136 )1137 else:1138 default_retrieval = (1139 f"RETURN relationship.`{self.text_node_property}` AS text, score, "1140 f"relationship {{.*, `{self.text_node_property}`: Null, "1141 f"`{self.embedding_node_property}`: Null, id: Null }} AS metadata"1142 )1143 1144 else:1145 if kwargs.get("return_embeddings"):1146 default_retrieval = (1147 f"RETURN node.`{self.text_node_property}` AS text, score, "1148 f"node {{.*, `{self.text_node_property}`: Null, "1149 f"`{self.embedding_node_property}`: Null, id: Null, "1150 f"_embedding_: node.`{self.embedding_node_property}`}} AS metadata"1151 )1152 else:1153 default_retrieval = (1154 f"RETURN node.`{self.text_node_property}` AS text, score, "1155 f"node {{.*, `{self.text_node_property}`: Null, "1156 f"`{self.embedding_node_property}`: Null, id: Null }} AS metadata"1157 )1158 1159 retrieval_query = (1160 self.retrieval_query if self.retrieval_query else default_retrieval1161 )1162 1163 read_query = index_query + retrieval_query1164 parameters = {1165 "index": self.index_name,1166 "k": k,1167 "embedding": embedding,1168 "keyword_index": self.keyword_index_name,1169 "query": remove_lucene_chars(kwargs["query"]),1170 **params,1171 **filter_params,1172 }1173 1174 results = self.query(read_query, params=parameters)1175 1176 if any(result["text"] is None for result in results):1177 if not self.retrieval_query:1178 raise ValueError(1179 f"Make sure that none of the `{self.text_node_property}` "1180 f"properties on nodes with label `{self.node_label}` "1181 "are missing or empty"1182 )1183 else:1184 raise ValueError(1185 "Inspect the `retrieval_query` and ensure it doesn't "1186 "return None for the `text` column"1187 )1188 if kwargs.get("return_embeddings") and any(1189 result["metadata"]["_embedding_"] is None for result in results1190 ):1191 if not self.retrieval_query:1192 raise ValueError(1193 f"Make sure that none of the `{self.embedding_node_property}` "1194 f"properties on nodes with label `{self.node_label}` "1195 "are missing or empty"1196 )1197 else:1198 raise ValueError(1199 "Inspect the `retrieval_query` and ensure it doesn't "1200 "return None for the `_embedding_` metadata column"