nl2cypher不存在了？使用agent框架和ollama快速搭建检索图数据库的demo

前言

最近Agent和LLM Tool Calling相关的主题很火，并且网上涌现了一系列agent框架，恰巧家里的电脑上部署了一个ollama，又意外发现agentscope可以很方便的在llm中嵌入一系列函数。

既然llm有思考能力，如果把图数据库相关的信息和调用方式通过agentscope喂给llm，那么模型是不是就可以基于图数据库中的知识进行问答了？

随手写了个demo，效果如下：

思考过程如下：

前提准备

• 安装最新版本的ollama（对tools提供支持）
• python环境，安装agentscope
• 配置GES图数据库
• 下载ges4jupyter并了解如何使用（地址：https://bbs.huaweicloud.com/blogs/361607）

ges4jupyter和图数据库配置，可以参考：https://bbs.huaweicloud.com/blogs/361607以及图引擎官方文档。

大概思路

由于agentscope支持工具调用，所以思考的重点应该是llm做不了什么：

• LLM不知道图结构，图中有哪些label
• LLM可能不知道怎么写cypher语句
• LLM无法发送语句给图数据库

那么tools函数编写，就围绕这三个问题展开
首先根据ges4jupyter的参考文档，初始化工具，而后解决三个问题：

async def main() -> None:
    eip = ''
    project_id = ''
    graph_name = ''
    iam_url = ''
    user_name = ''
    password = ''
    domain_name = ''
    project_name = ''
    port = 80
    eip, project_id, graph_name, iam_url, user_name, password, domain_name, project_name, port = read_csv_config(
        'cn_north_4_graph.csv')
    config = GESConfig(eip, project_id, graph_name,
                       iam_url=iam_url,
                       user_name=user_name,
                       password=password,
                       domain_name=domain_name,
                       project_name=project_name,
                       port=port)
    ges_util = GES4Jupyter(config, False)

编写函数获取图结构

修改ges4jupyter代码，添加获取schema的代码：

    def schema(self):
        url = ('{}://{}:{}/ges/v1.0/{}/graphs/{}/schema').format(self.scheme, self.eip, self.port,
                                                                                                self.project_id,
                                                                                                self.graph_name)
        r = requests.get(url=url, headers=self.headers, verify=False)
        output = r.json()
        return output

编写工具函数：

    def get_graph_schema() -> ToolResponse:
        """
        获取图的schema，schema中关联的label和属性，用于拼接cypher语句
        Args:
        Returns:
             图的schema，json格式
        """
        return ToolResponse(content=[TextBlock(text=str(ges_util.schema()), type='text')])

编写函数执行cypher语句：

    def execute_cypher_query(stmt) -> ToolResponse:
        """
        执行cypher查询
        Args:
            stmt: cypher语句
        Returns:
             json格式的cypher响应体
        """
        #[TextBlock(text=str(ges_util.cypher_query(stmt)), type='text')]
        return ToolResponse(content=[TextBlock(text=str(ges_util.cypher_query(stmt)), type='text')])

编写函数提供cypher示例

这里提供了几个示例，可能没被LLM用到，如果想让LLM侧重于解决什么问题，就多写类似的示例

    def get_example_query_get_node_by_id() -> ToolResponse:
        """
        样例cypher语句,根据点id来获取点详情
        Args:
        Returns:
             样例cypher语句,根据点id来获取点详情
        """
        #
        return ToolResponse(content=[TextBlock(text="match (n) where id(n)='{{id_place_holder}}' return n", type='text')])

    def get_example_query_expand_all() -> ToolResponse:
        """
        样例cypher语句,获取点的一跳邻居（邻居点和邻居边）
        Args:
        Returns:
            样例cypher语句,获取点的一跳邻居（邻居点和邻居边）
        """
        return ToolResponse(content=[TextBlock(text="match (n)-[r]-(m) where id(n)='{{id_place_holder}}' return r,m", type='text')])

    def get_example_query_expand_all_with_filter() -> ToolResponse:
        """
        样例cypher语句,获取点的一跳邻居（邻居点和邻居边）,并附带过滤条件
        Args:
        Returns:
            样例cypher语句
        """
        return ToolResponse(content=[TextBlock(text="match (n)-[r:{{label_place_holder}}]-(m) where id(n)='{{id_place_holder}}' and n.property_name='property_value' return r,m", type='text')])

最后对工具集进行打包，输入agent中：

toolkit = Toolkit()
    toolkit.register_tool_function(get_graph_schema)
    toolkit.register_tool_function(execute_cypher_query)
    toolkit.register_tool_function(get_example_query_expand_all)
    toolkit.register_tool_function(get_example_query_expand_all_with_filter)
    # 创建智能体
    agent = ReActAgent(
        name="Friday",
        sys_prompt="你是一个智能助手，手头有一系列工具集，可以根据用户需求拼接cypher语句，或者根据用户的需求生成cypher语句并执行，可以调用工具集中的无参数工具获取样例语句以及图中的label属性信息。注意语句中如果出现label和属性，则必须是通过工具函数拿到的schema中的label和属性。最后，对于执行后了的cypher语句，需要检查一下结果，并用相对友好的方式(如表格、文字）返回给用户。如果用户输入的cypher语句未给定id，建议尾部通过limit语句限制返回数目，可以配置为20。",
        model=OllamaChatModel(model_name='qwen3:8b', host='http://127.0.0.1:11434', enable_thinking=True),
        formatter=OllamaChatFormatter(),
        toolkit=toolkit,
    )

提示词中要写明，要求llm首先获取图的schema，相关语句语句拼接时，label和properties都要来自于获取的schema中。
以下是完整示例代码。

示例代码

import asyncio
from agentscope.agent import ReActAgent, UserAgent
from agentscope.formatter import OllamaChatFormatter
from agentscope.message import TextBlock
from agentscope.model import OllamaChatModel
from agentscope.tool import Toolkit
from agentscope.tool import ToolResponse
from ges4jupyter import GESConfig, GES4Jupyter, read_csv_config

async def main() -> None:
    eip = ''
    project_id = ''
    graph_name = ''
    iam_url = ''
    user_name = ''
    password = ''
    domain_name = ''
    project_name = ''
    port = 80
    eip, project_id, graph_name, iam_url, user_name, password, domain_name, project_name, port = read_csv_config(
        'cn_north_4_graph.csv')
    config = GESConfig(eip, project_id, graph_name,
                       iam_url=iam_url,
                       user_name=user_name,
                       password=password,
                       domain_name=domain_name,
                       project_name=project_name,
                       port=port)
    ges_util = GES4Jupyter(config, False)

    async def get_graph_schema() -> ToolResponse:
        """
        获取图的schema，schema中关联的label和属性，用于拼接cypher语句
        Args:
        Returns:
             图的schema，json格式
        """
        return ToolResponse(content=[TextBlock(text=str(ges_util.schema()), type='text')])

    def execute_cypher_query(stmt) -> ToolResponse:
        """
        执行cypher查询
        Args:
            stmt: cypher语句
        Returns:
             json格式的cypher响应体
        """
        return ToolResponse(content=[TextBlock(text=str(ges_util.cypher_query(stmt)), type='text')])

    def get_example_query_get_node_by_id() -> ToolResponse:
        """
        样例cypher语句,根据点id来获取点详情
        Args:
        Returns:
             样例cypher语句,根据点id来获取点详情
        """
        #
        return ToolResponse(content=[TextBlock(text="match (n) where id(n)='{{id_place_holder}}' return n", type='text')])

    def get_example_query_expand_all() -> ToolResponse:
        """
        样例cypher语句,获取点的一跳邻居（邻居点和邻居边）
        Args:
        Returns:
            样例cypher语句,获取点的一跳邻居（邻居点和邻居边）
        """
        return ToolResponse(content=[TextBlock(text="match (n)-[r]-(m) where id(n)='{{id_place_holder}}' return r,m", type='text')])

    def get_example_query_expand_all_with_filter() -> ToolResponse:
        """
        样例cypher语句,获取点的一跳邻居（邻居点和邻居边）,并附带过滤条件
        Args:
        Returns:
            样例cypher语句
        """
        return ToolResponse(content=[TextBlock(text="match (n)-[r:{{label_place_holder}}]-(m) where id(n)='{{id_place_holder}}' and n.property_name='property_value' return r,m", type='text')])

    toolkit = Toolkit()
    toolkit.register_tool_function(get_graph_schema)
    toolkit.register_tool_function(execute_cypher_query)
    toolkit.register_tool_function(get_example_query_expand_all)
    toolkit.register_tool_function(get_example_query_expand_all_with_filter)
    # 创建智能体
    agent = ReActAgent(
        name="Friday",
        sys_prompt="你是一个智能助手，手头有一系列工具集，可以根据用户需求拼接cypher语句，或者根据用户的需求生成cypher语句并执行，可以调用工具集中的无参数工具获取样例语句以及图中的label属性信息。注意语句中如果出现label和属性，则必须是通过工具函数拿到的schema中的label和属性。最后，对于执行后了的cypher语句，需要检查一下结果，并用相对友好的方式(如表格、文字）返回给用户。如果用户输入的cypher语句未给定id，建议尾部通过limit语句限制返回数目，可以配置为20。",
        model=OllamaChatModel(model_name='qwen3:8b', host='http://127.0.0.1:11434', enable_thinking=True),
        formatter=OllamaChatFormatter(),
        toolkit=toolkit,
    )

    # 创建用户输入的代理
    user = UserAgent(name="user")

    import json
    print(json.dumps(toolkit.get_json_schemas(), indent=4, ensure_ascii=False))

    # 通过消息的显式传递构建对话逻辑
    msg = None
    while True:
        msg = await agent(msg)
        msg = await user(msg)
        print(msg)
        if msg.get_text_content() == "exit":
            break


asyncio.run(main())

其他

这里只是一个demo，真实检索场景往往还涉及向量库的使用（比如一些模糊或者近似的语义），以及对结果的剪枝和重排序。如果只是为了做nl2cypher类的工作，则可以不使用向量库，毕竟nl2cypher只需要知道图的元数据、以及cypher怎么写，就可以做很多事了。