Skip to content

Neo4j

Neo4jQueryToolSpec #

Bases: BaseToolSpec

这个类负责根据提供的模式定义查询Neo4j图数据库。

Source code in llama_index/tools/neo4j/base.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class Neo4jQueryToolSpec(BaseToolSpec):
    """这个类负责根据提供的模式定义查询Neo4j图数据库。"""

    spec_functions = ["run_request"]

    def __init__(
        self, url, user, password, database, llm: LLM, validate_cypher: bool = False
    ):
        """初始化Neo4jSchemaWiseQuery对象。

Args:
    url (str): Neo4j数据库的连接字符串。
    user (str): Neo4j数据库的用户名。
    password (str): Neo4j数据库的密码。
    llm (obj): 用于生成Cypher查询的语言模型。
    validate_cypher (bool): 验证生成的Cypher语句中的关系方向。默认值: False
"""
        if find_spec("neo4j") is None:
            raise ImportError(
                "`neo4j` package not found, please run `pip install neo4j`"
            )

        self.graph_store = Neo4jGraphStore(
            url=url, username=user, password=password, database=database
        )
        self.llm = llm
        self.cypher_query_corrector = None
        if validate_cypher:
            corrector_schema = [
                Schema(el["start"], el["type"], el["end"])
                for el in self.graph_store.structured_schema.get("relationships")
            ]
            self.cypher_query_corrector = CypherQueryCorrector(corrector_schema)

    def get_system_message(self):
        """生成一个详细描述任务和模式的系统消息。

返回：
    str：系统消息。
"""
        return f"""
        Task: Generate Cypher queries to query a Neo4j graph database based on the provided schema definition.
        Instructions:
        Use only the provided relationship types and properties.
        Do not use any other relationship types or properties that are not provided.
        If you cannot generate a Cypher statement based on the provided schema, explain the reason to the user.
        Schema:
        {self.graph_store.schema}

        Note: Do not include any explanations or apologies in your responses.
        """

    def query_graph_db(self, neo4j_query, params=None):
        """查询Neo4j数据库。

Args:
    neo4j_query (str): 要执行的Cypher查询。
    params (dict, optional): Cypher查询的参数。默认为None。

Returns:
    list: 查询结果。
"""
        if params is None:
            params = {}
        with self.graph_store.client.session() as session:
            result = session.run(neo4j_query, params)
            output = [r.values() for r in result]
            output.insert(0, list(result.keys()))
            return output

    def construct_cypher_query(self, question, history=None):
        """根据给定的问题和历史记录构建一个Cypher查询。

Args:
    question (str): 要为其构建Cypher查询的问题。
    history (list, optional): 用于提供上下文的先前交互的列表。默认为None。

Returns:
    str: 构建的Cypher查询。
"""
        messages = [
            ChatMessage(role=MessageRole.SYSTEM, content=self.get_system_message()),
            ChatMessage(role=MessageRole.USER, content=question),
        ]
        # Used for Cypher healing flows
        if history:
            messages.extend(history)

        completions = self.llm.chat(messages)
        return completions.message.content

    def run_request(self, question, history=None, retry=True):
        """根据给定的问题执行Cypher查询。

Args:
    question (str): 要执行Cypher查询的问题。
    history (list, optional): 用于上下文的先前交互列表。默认为None。
    retry (bool, optional): 是否在出现语法错误时重试。默认为True。

Returns:
    list/str: 查询结果或错误消息。
"""
        from neo4j.exceptions import CypherSyntaxError

        # Construct Cypher statement
        cypher = self.construct_cypher_query(question, history)
        # Validate Cypher statement
        if self.cypher_query_corrector:
            cypher = self.cypher_query_corrector(cypher)
        print(cypher)
        try:
            return self.query_graph_db(cypher)
        # Self-healing flow
        except CypherSyntaxError as e:
            # If out of retries
            if not retry:
                return "Invalid Cypher syntax"
            # Self-healing Cypher flow by
            # providing specific error to GPT-4
            print("Retrying")
            return self.run_request(
                question,
                [
                    ChatMessage(role=MessageRole.ASSISTANT, content=cypher),
                    ChatMessage(
                        role=MessageRole.SYSTEM,
                        content=f"This query returns an error: {e!s}\n"
                        "Give me a improved query that works without any explanations or apologies",
                    ),
                ],
                retry=False,
            )

get_system_message #

get_system_message()

生成一个详细描述任务和模式的系统消息。

返回： str：系统消息。

Source code in llama_index/tools/neo4j/base.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
    def get_system_message(self):
        """生成一个详细描述任务和模式的系统消息。

返回：
    str：系统消息。
"""
        return f"""
        Task: Generate Cypher queries to query a Neo4j graph database based on the provided schema definition.
        Instructions:
        Use only the provided relationship types and properties.
        Do not use any other relationship types or properties that are not provided.
        If you cannot generate a Cypher statement based on the provided schema, explain the reason to the user.
        Schema:
        {self.graph_store.schema}

        Note: Do not include any explanations or apologies in your responses.
        """

query_graph_db #

query_graph_db(neo4j_query, params=None)

查询Neo4j数据库。

Parameters:

Name Type Description Default
neo4j_query str

要执行的Cypher查询。

 required
params dict

Cypher查询的参数。默认为None。

 None

Returns:

Name Type Description
list

查询结果。
Source code in llama_index/tools/neo4j/base.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
    def query_graph_db(self, neo4j_query, params=None):
        """查询Neo4j数据库。

Args:
    neo4j_query (str): 要执行的Cypher查询。
    params (dict, optional): Cypher查询的参数。默认为None。

Returns:
    list: 查询结果。
"""
        if params is None:
            params = {}
        with self.graph_store.client.session() as session:
            result = session.run(neo4j_query, params)
            output = [r.values() for r in result]
            output.insert(0, list(result.keys()))
            return output

construct_cypher_query #

construct_cypher_query(question, history=None)

根据给定的问题和历史记录构建一个Cypher查询。

Parameters:

Name Type Description Default
question str

要为其构建Cypher查询的问题。

 required
history list

用于提供上下文的先前交互的列表。默认为None。

 None

Returns:

Name Type Description
str

构建的Cypher查询。
Source code in llama_index/tools/neo4j/base.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
    def construct_cypher_query(self, question, history=None):
        """根据给定的问题和历史记录构建一个Cypher查询。

Args:
    question (str): 要为其构建Cypher查询的问题。
    history (list, optional): 用于提供上下文的先前交互的列表。默认为None。

Returns:
    str: 构建的Cypher查询。
"""
        messages = [
            ChatMessage(role=MessageRole.SYSTEM, content=self.get_system_message()),
            ChatMessage(role=MessageRole.USER, content=question),
        ]
        # Used for Cypher healing flows
        if history:
            messages.extend(history)

        completions = self.llm.chat(messages)
        return completions.message.content

run_request #

run_request(question, history=None, retry=True)

根据给定的问题执行Cypher查询。

Parameters:

Name Type Description Default
question str

要执行Cypher查询的问题。

 required
history list

用于上下文的先前交互列表。默认为None。

 None
retry bool

是否在出现语法错误时重试。默认为True。

 True

Returns:

Type Description

list/str: 查询结果或错误消息。
Source code in llama_index/tools/neo4j/base.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
    def run_request(self, question, history=None, retry=True):
        """根据给定的问题执行Cypher查询。

Args:
    question (str): 要执行Cypher查询的问题。
    history (list, optional): 用于上下文的先前交互列表。默认为None。
    retry (bool, optional): 是否在出现语法错误时重试。默认为True。

Returns:
    list/str: 查询结果或错误消息。
"""
        from neo4j.exceptions import CypherSyntaxError

        # Construct Cypher statement
        cypher = self.construct_cypher_query(question, history)
        # Validate Cypher statement
        if self.cypher_query_corrector:
            cypher = self.cypher_query_corrector(cypher)
        print(cypher)
        try:
            return self.query_graph_db(cypher)
        # Self-healing flow
        except CypherSyntaxError as e:
            # If out of retries
            if not retry:
                return "Invalid Cypher syntax"
            # Self-healing Cypher flow by
            # providing specific error to GPT-4
            print("Retrying")
            return self.run_request(
                question,
                [
                    ChatMessage(role=MessageRole.ASSISTANT, content=cypher),
                    ChatMessage(
                        role=MessageRole.SYSTEM,
                        content=f"This query returns an error: {e!s}\n"
                        "Give me a improved query that works without any explanations or apologies",
                    ),
                ],
                retry=False,
            )