Graphdb cypher

GraphDBCypherReader

Bases: BaseReader

图形数据库Cypher读取器。

将所有Cypher查询结果组合成LlamaIndex使用的Document类型。

Source code in llama_index/readers/graphdb_cypher/base.py

class GraphDBCypherReader(BaseReader):
    """图形数据库Cypher读取器。

将所有Cypher查询结果组合成LlamaIndex使用的Document类型。

Args:
    uri（str）：图形数据库URI
    username（str）：用户名
    password（str）：密码"""

    def __init__(self, uri: str, username: str, password: str, database: str) -> None:
        """使用参数进行初始化。"""
        try:
            from neo4j import GraphDatabase, basic_auth

        except ImportError:
            raise ImportError(
                "`neo4j` package not found, please run `pip install neo4j`"
            )
        if uri:
            if uri is None:
                raise ValueError("`uri` must be provided.")
            self.client = GraphDatabase.driver(
                uri=uri, auth=basic_auth(username, password)
            )
            self.database = database

    def load_data(
        self, query: str, parameters: Optional[Dict] = None
    ) -> List[Document]:
        """运行带有可选参数的Cypher，并将结果转换为文档。

Args:
    query (str): 图形Cypher查询字符串。
    parameters (Optional[Dict]): 可选的查询参数。

Returns:
    List[Document]: 文档列表。
"""
        if parameters is None:
            parameters = {}

        records, summary, keys = self.client.execute_query(
            query, parameters, database_=self.database
        )

        return [Document(text=yaml.dump(entry.data())) for entry in records]

load_data

load_data(
    query: str, parameters: Optional[Dict] = None
) -> List[Document]

运行带有可选参数的Cypher，并将结果转换为文档。

Parameters:

Name	Type	Description	Default
query	str	图形Cypher查询字符串。	required
parameters	Optional[Dict]	可选的查询参数。	None

Returns:

Type	Description
List[Document]	List[Document]: 文档列表。

Source code in llama_index/readers/graphdb_cypher/base.py

    def load_data(
        self, query: str, parameters: Optional[Dict] = None
    ) -> List[Document]:
        """运行带有可选参数的Cypher，并将结果转换为文档。

Args:
    query (str): 图形Cypher查询字符串。
    parameters (Optional[Dict]): 可选的查询参数。

Returns:
    List[Document]: 文档列表。
"""
        if parameters is None:
            parameters = {}

        records, summary, keys = self.client.execute_query(
            query, parameters, database_=self.database
        )

        return [Document(text=yaml.dump(entry.data())) for entry in records]