Neo4jvector

Neo4jVectorStore

Bases: BasePydanticVectorStore

Neo4j向量存储。

示例: pip install llama-index-vector-stores-neo4jvector

from llama_index.vector_stores.neo4jvector import Neo4jVectorStore

username = "neo4j"
password = "pleaseletmein"
url = "bolt://localhost:7687"
embed_dim = 1536

neo4j_vector = Neo4jVectorStore(username, password, url, embed_dim)

Source code in llama_index/vector_stores/neo4jvector/base.py

class Neo4jVectorStore(BasePydanticVectorStore):
    """Neo4j向量存储。

示例:
`pip install llama-index-vector-stores-neo4jvector`

```python
from llama_index.vector_stores.neo4jvector import Neo4jVectorStore

username = "neo4j"
password = "pleaseletmein"
url = "bolt://localhost:7687"
embed_dim = 1536

neo4j_vector = Neo4jVectorStore(username, password, url, embed_dim)
```"""

    stores_text: bool = True
    flat_metadata = True

    distance_strategy: str
    index_name: str
    keyword_index_name: str
    hybrid_search: bool
    node_label: str
    embedding_node_property: str
    text_node_property: str
    retrieval_query: str
    embedding_dimension: int

    _driver: neo4j.GraphDatabase.driver = PrivateAttr()
    _database: str = PrivateAttr()
    _support_metadata_filter: bool = PrivateAttr()
    _is_enterprise: bool = PrivateAttr()

    def __init__(
        self,
        username: str,
        password: str,
        url: str,
        embedding_dimension: int,
        database: str = "neo4j",
        index_name: str = "vector",
        keyword_index_name: str = "keyword",
        node_label: str = "Chunk",
        embedding_node_property: str = "embedding",
        text_node_property: str = "text",
        distance_strategy: str = "cosine",
        hybrid_search: bool = False,
        retrieval_query: str = "",
        **kwargs: Any,
    ) -> None:
        super().__init__(
            distance_strategy=distance_strategy,
            index_name=index_name,
            keyword_index_name=keyword_index_name,
            hybrid_search=hybrid_search,
            node_label=node_label,
            embedding_node_property=embedding_node_property,
            text_node_property=text_node_property,
            retrieval_query=retrieval_query,
            embedding_dimension=embedding_dimension,
        )

        if distance_strategy not in ["cosine", "euclidean"]:
            raise ValueError("distance_strategy must be either 'euclidean' or 'cosine'")

        self._driver = neo4j.GraphDatabase.driver(url, auth=(username, password))
        self._database = database

        # Verify connection
        try:
            self._driver.verify_connectivity()
        except neo4j.exceptions.ServiceUnavailable:
            raise ValueError(
                "Could not connect to Neo4j database. "
                "Please ensure that the url is correct"
            )
        except neo4j.exceptions.AuthError:
            raise ValueError(
                "Could not connect to Neo4j database. "
                "Please ensure that the username and password are correct"
            )

        # Verify if the version support vector index
        self._verify_version()

        # Verify that required values are not null
        check_if_not_null(
            [
                "index_name",
                "node_label",
                "embedding_node_property",
                "text_node_property",
            ],
            [index_name, node_label, embedding_node_property, text_node_property],
        )

        index_already_exists = self.retrieve_existing_index()
        if not index_already_exists:
            self.create_new_index()
        if hybrid_search:
            fts_node_label = self.retrieve_existing_fts_index()
            # If the FTS index doesn't exist yet
            if not fts_node_label:
                self.create_new_keyword_index()
            else:  # Validate that FTS and Vector index use the same information
                if not fts_node_label == self.node_label:
                    raise ValueError(
                        "Vector and keyword index don't index the same node label"
                    )

    def _verify_version(self) -> None:
        """检查连接的Neo4j数据库版本是否支持向量索引。

查询Neo4j数据库以检索其版本，并将其与已知支持向量索引的目标版本（5.11.0）进行比较。如果连接的Neo4j版本不受支持，则引发ValueError。
"""
        db_data = self.database_query("CALL dbms.components()")
        version = db_data[0]["versions"][0]
        if "aura" in version:
            version_tuple = (*tuple(map(int, version.split("-")[0].split("."))), 0)
        else:
            version_tuple = tuple(map(int, version.split(".")))

        target_version = (5, 11, 0)

        if version_tuple < target_version:
            raise ValueError(
                "Version index is only supported in Neo4j version 5.11 or greater"
            )

        # Flag for metadata filtering
        metadata_target_version = (5, 18, 0)
        if version_tuple < metadata_target_version:
            self._support_metadata_filter = False
        else:
            self._support_metadata_filter = True
        # Flag for enterprise
        self._is_enterprise = db_data[0]["edition"] == "enterprise"

    def create_new_index(self) -> None:
        """这个方法构造一个Cypher查询并执行它来在Neo4j中创建一个新的向量索引。
"""
        index_query = (
            "CALL db.index.vector.createNodeIndex("
            "$index_name,"
            "$node_label,"
            "$embedding_node_property,"
            "toInteger($embedding_dimension),"
            "$similarity_metric )"
        )

        parameters = {
            "index_name": self.index_name,
            "node_label": self.node_label,
            "embedding_node_property": self.embedding_node_property,
            "embedding_dimension": self.embedding_dimension,
            "similarity_metric": self.distance_strategy,
        }
        self.database_query(index_query, params=parameters)

    def retrieve_existing_index(self) -> bool:
        """检查向量索引是否存在于Neo4j数据库中，并返回其嵌入维度。

该方法查询Neo4j数据库中的现有索引，并尝试检索具有指定名称的向量索引的维度。如果索引存在，则返回其维度。如果索引不存在，则返回`None`。

返回：
    int或None：如果找到现有索引，则返回其嵌入维度。
"""
        index_information = self.database_query(
            "SHOW INDEXES YIELD name, type, labelsOrTypes, properties, options "
            "WHERE type = 'VECTOR' AND (name = $index_name "
            "OR (labelsOrTypes[0] = $node_label AND "
            "properties[0] = $embedding_node_property)) "
            "RETURN name, labelsOrTypes, properties, options ",
            params={
                "index_name": self.index_name,
                "node_label": self.node_label,
                "embedding_node_property": self.embedding_node_property,
            },
        )
        # sort by index_name
        index_information = sort_by_index_name(index_information, self.index_name)
        try:
            self.index_name = index_information[0]["name"]
            self.node_label = index_information[0]["labelsOrTypes"][0]
            self.embedding_node_property = index_information[0]["properties"][0]
            self.embedding_dimension = index_information[0]["options"]["indexConfig"][
                "vector.dimensions"
            ]

            return True
        except IndexError:
            return False

    def retrieve_existing_fts_index(self) -> Optional[str]:
        """检查Neo4j数据库中是否存在全文索引。

此方法查询具有指定名称的现有fts索引的Neo4j数据库。

返回：
    （元组）：关键字索引信息
"""
        index_information = self.database_query(
            "SHOW INDEXES YIELD name, type, labelsOrTypes, properties, options "
            "WHERE type = 'FULLTEXT' AND (name = $keyword_index_name "
            "OR (labelsOrTypes = [$node_label] AND "
            "properties = $text_node_property)) "
            "RETURN name, labelsOrTypes, properties, options ",
            params={
                "keyword_index_name": self.keyword_index_name,
                "node_label": self.node_label,
                "text_node_property": self.text_node_property,
            },
        )
        # sort by index_name
        index_information = sort_by_index_name(index_information, self.index_name)
        try:
            self.keyword_index_name = index_information[0]["name"]
            self.text_node_property = index_information[0]["properties"][0]
            return index_information[0]["labelsOrTypes"][0]
        except IndexError:
            return None

    def create_new_keyword_index(self, text_node_properties: List[str] = []) -> None:
        """这个方法构造一个Cypher查询并执行它来在Neo4j中创建一个新的全文索引。
"""
        node_props = text_node_properties or [self.text_node_property]
        fts_index_query = (
            f"CREATE FULLTEXT INDEX {self.keyword_index_name} "
            f"FOR (n:`{self.node_label}`) ON EACH "
            f"[{', '.join(['n.`' + el + '`' for el in node_props])}]"
        )
        self.database_query(fts_index_query)

    def database_query(
        self, query: str, params: Optional[dict] = None
    ) -> List[Dict[str, Any]]:
        """这个方法将一个Cypher查询发送到连接的Neo4j数据库，并将结果作为字典列表返回。

Args:
    query (str): 要执行的Cypher查询。
    params (dict, 可选): 查询参数的字典。默认为{}。

Returns:
    List[Dict[str, Any]]: 包含查询结果的字典列表。
"""
        params = params or {}
        with self._driver.session(database=self._database) as session:
            try:
                data = session.run(query, params)
                return [r.data() for r in data]
            except CypherSyntaxError as e:
                raise ValueError(f"Cypher Statement is not valid\n{e}")

    def add(self, nodes: List[BaseNode], **add_kwargs: Any) -> List[str]:
        ids = [r.node_id for r in nodes]
        import_query = (
            "UNWIND $data AS row "
            "CALL { WITH row "
            f"MERGE (c:`{self.node_label}` {{id: row.id}}) "
            "WITH c, row "
            f"CALL db.create.setVectorProperty(c, "
            f"'{self.embedding_node_property}', row.embedding) "
            "YIELD node "
            f"SET c.`{self.text_node_property}` = row.text "
            "SET c += row.metadata } IN TRANSACTIONS OF 1000 ROWS"
        )

        self.database_query(
            import_query,
            params={"data": clean_params(nodes)},
        )

        return ids

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        if query.filters:
            # Verify that 5.18 or later is used
            if not self._support_metadata_filter:
                raise ValueError(
                    "Metadata filtering is only supported in "
                    "Neo4j version 5.18 or greater"
                )
            # Metadata filtering and hybrid doesn't work
            if self.hybrid_search:
                raise ValueError(
                    "Metadata filtering can't be use in combination with "
                    "a hybrid search approach"
                )
            parallel_query = (
                "CYPHER runtime = parallel parallelRuntimeSupport=all "
                if self._is_enterprise
                else ""
            )
            base_index_query = parallel_query + (
                f"MATCH (n:`{self.node_label}`) WHERE "
                f"n.`{self.embedding_node_property}` IS NOT NULL AND "
                f"size(n.`{self.embedding_node_property}`) = "
                f"toInteger({self.embedding_dimension}) AND "
            )
            base_cosine_query = (
                " WITH n as node, vector.similarity.cosine("
                f"n.`{self.embedding_node_property}`, "
                "$embedding) AS score ORDER BY score DESC LIMIT toInteger($k) "
            )
            filter_snippets, filter_params = construct_metadata_filter(query.filters)
            index_query = base_index_query + filter_snippets + base_cosine_query
        else:
            index_query = _get_search_index_query(self.hybrid_search)
            filter_params = {}

        default_retrieval = (
            f"RETURN node.`{self.text_node_property}` AS text, score, "
            "node.id AS id, "
            f"node {{.*, `{self.text_node_property}`: Null, "
            f"`{self.embedding_node_property}`: Null, id: Null }} AS metadata"
        )

        retrieval_query = self.retrieval_query or default_retrieval
        read_query = index_query + retrieval_query

        parameters = {
            "index": self.index_name,
            "k": query.similarity_top_k,
            "embedding": query.query_embedding,
            "keyword_index": self.keyword_index_name,
            "query": remove_lucene_chars(query.query_str),
            **filter_params,
        }

        results = self.database_query(read_query, params=parameters)

        nodes = []
        similarities = []
        ids = []
        for record in results:
            node = metadata_dict_to_node(record["metadata"])
            node.set_content(str(record["text"]))
            nodes.append(node)
            similarities.append(record["score"])
            ids.append(record["id"])

        return VectorStoreQueryResult(nodes=nodes, similarities=similarities, ids=ids)

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        self.database_query(
            f"MATCH (n:`{self.node_label}`) WHERE n.ref_doc_id = $id DETACH DELETE n",
            params={"id": ref_doc_id},
        )

create_new_index

create_new_index() -> None

这个方法构造一个Cypher查询并执行它来在Neo4j中创建一个新的向量索引。

Source code in llama_index/vector_stores/neo4jvector/base.py

    def create_new_index(self) -> None:
        """这个方法构造一个Cypher查询并执行它来在Neo4j中创建一个新的向量索引。
"""
        index_query = (
            "CALL db.index.vector.createNodeIndex("
            "$index_name,"
            "$node_label,"
            "$embedding_node_property,"
            "toInteger($embedding_dimension),"
            "$similarity_metric )"
        )

        parameters = {
            "index_name": self.index_name,
            "node_label": self.node_label,
            "embedding_node_property": self.embedding_node_property,
            "embedding_dimension": self.embedding_dimension,
            "similarity_metric": self.distance_strategy,
        }
        self.database_query(index_query, params=parameters)

retrieve_existing_index

retrieve_existing_index() -> bool

检查向量索引是否存在于Neo4j数据库中，并返回其嵌入维度。

该方法查询Neo4j数据库中的现有索引，并尝试检索具有指定名称的向量索引的维度。如果索引存在，则返回其维度。如果索引不存在，则返回None。

返回： int或None：如果找到现有索引，则返回其嵌入维度。

Source code in llama_index/vector_stores/neo4jvector/base.py

    def retrieve_existing_index(self) -> bool:
        """检查向量索引是否存在于Neo4j数据库中，并返回其嵌入维度。

该方法查询Neo4j数据库中的现有索引，并尝试检索具有指定名称的向量索引的维度。如果索引存在，则返回其维度。如果索引不存在，则返回`None`。

返回：
    int或None：如果找到现有索引，则返回其嵌入维度。
"""
        index_information = self.database_query(
            "SHOW INDEXES YIELD name, type, labelsOrTypes, properties, options "
            "WHERE type = 'VECTOR' AND (name = $index_name "
            "OR (labelsOrTypes[0] = $node_label AND "
            "properties[0] = $embedding_node_property)) "
            "RETURN name, labelsOrTypes, properties, options ",
            params={
                "index_name": self.index_name,
                "node_label": self.node_label,
                "embedding_node_property": self.embedding_node_property,
            },
        )
        # sort by index_name
        index_information = sort_by_index_name(index_information, self.index_name)
        try:
            self.index_name = index_information[0]["name"]
            self.node_label = index_information[0]["labelsOrTypes"][0]
            self.embedding_node_property = index_information[0]["properties"][0]
            self.embedding_dimension = index_information[0]["options"]["indexConfig"][
                "vector.dimensions"
            ]

            return True
        except IndexError:
            return False

retrieve_existing_fts_index

retrieve_existing_fts_index() -> Optional[str]

检查Neo4j数据库中是否存在全文索引。

此方法查询具有指定名称的现有fts索引的Neo4j数据库。

返回： （元组）：关键字索引信息

Source code in llama_index/vector_stores/neo4jvector/base.py

    def retrieve_existing_fts_index(self) -> Optional[str]:
        """检查Neo4j数据库中是否存在全文索引。

此方法查询具有指定名称的现有fts索引的Neo4j数据库。

返回：
    （元组）：关键字索引信息
"""
        index_information = self.database_query(
            "SHOW INDEXES YIELD name, type, labelsOrTypes, properties, options "
            "WHERE type = 'FULLTEXT' AND (name = $keyword_index_name "
            "OR (labelsOrTypes = [$node_label] AND "
            "properties = $text_node_property)) "
            "RETURN name, labelsOrTypes, properties, options ",
            params={
                "keyword_index_name": self.keyword_index_name,
                "node_label": self.node_label,
                "text_node_property": self.text_node_property,
            },
        )
        # sort by index_name
        index_information = sort_by_index_name(index_information, self.index_name)
        try:
            self.keyword_index_name = index_information[0]["name"]
            self.text_node_property = index_information[0]["properties"][0]
            return index_information[0]["labelsOrTypes"][0]
        except IndexError:
            return None

create_new_keyword_index

create_new_keyword_index(
    text_node_properties: List[str] = [],
) -> None

这个方法构造一个Cypher查询并执行它来在Neo4j中创建一个新的全文索引。

Source code in llama_index/vector_stores/neo4jvector/base.py

    def create_new_keyword_index(self, text_node_properties: List[str] = []) -> None:
        """这个方法构造一个Cypher查询并执行它来在Neo4j中创建一个新的全文索引。
"""
        node_props = text_node_properties or [self.text_node_property]
        fts_index_query = (
            f"CREATE FULLTEXT INDEX {self.keyword_index_name} "
            f"FOR (n:`{self.node_label}`) ON EACH "
            f"[{', '.join(['n.`' + el + '`' for el in node_props])}]"
        )
        self.database_query(fts_index_query)

database_query

database_query(
    query: str, params: Optional[dict] = None
) -> List[Dict[str, Any]]

这个方法将一个Cypher查询发送到连接的Neo4j数据库，并将结果作为字典列表返回。

Parameters:

Name	Type	Description	Default
query	str	要执行的Cypher查询。	required
params	(dict, 可选)	查询参数的字典。默认为{}。	None

Returns:

Type	Description
List[Dict[str, Any]]	List[Dict[str, Any]]: 包含查询结果的字典列表。

Source code in llama_index/vector_stores/neo4jvector/base.py

    def database_query(
        self, query: str, params: Optional[dict] = None
    ) -> List[Dict[str, Any]]:
        """这个方法将一个Cypher查询发送到连接的Neo4j数据库，并将结果作为字典列表返回。

Args:
    query (str): 要执行的Cypher查询。
    params (dict, 可选): 查询参数的字典。默认为{}。

Returns:
    List[Dict[str, Any]]: 包含查询结果的字典列表。
"""
        params = params or {}
        with self._driver.session(database=self._database) as session:
            try:
                data = session.run(query, params)
                return [r.data() for r in data]
            except CypherSyntaxError as e:
                raise ValueError(f"Cypher Statement is not valid\n{e}")