Neo4j 数据库的隐藏技能：如何用 LlamaIndex 实现智能化查询？

检索增强生成（Retrieval-augmented generation，RAG）已经成为主流技术，并且有充分的理由支持其广泛应用。它是一种强大的框架，将先进的大语言模型与目标信息检索技术相结合，从而实现更快速的相关数据访问，并生成更准确、上下文感知的响应。尽管 RAG 应用通常集中于非结构化数据，我个人非常推崇将结构化数据整合进来——这是一种重要但经常被忽视的策略。其中，我最喜欢的方式之一是利用图数据库，例如 Neo4j。

通常，从图数据库中检索数据的首选方法是 Text2Cypher，即通过自然语言查询自动转换为 Cypher 语句，以查询图数据库。这种技术依赖于语言模型（或基于规则的系统）来解释用户查询，推断其潜在意图，并将其翻译为有效的 Cypher 查询，使 RAG 应用能够从知识图谱中检索相关信息并生成准确的答案。

Text2Cypher 提供了显著的灵活性，因为它允许用户以自然语言提出问题，而无需了解底层图数据库的模式或 Cypher 语法。然而，由于语言解释的细微差别以及对精确模式特定细节的需求，其准确性仍可能存在不足，如以下 Text2Cypher 文章所示。

以下可视化展示了基准测试的最重要结果：

从高层次来看，该基准测试比较了三组模型：

• • 为 Text2Cypher 任务微调的模型

• • 开放的基础模型

• • 封闭的基础模型

基准测试使用两种指标评估这些模型在生成正确 Cypher 查询方面的表现：Google BLEU（上图）和 ExactMatch（下图）。

• • Google BLEU 指标衡量生成的查询与参考查询之间的重叠程度（以 n-grams 为单位）。较高的分数通常表示与参考查询更接近，但这并不一定保证查询在数据库上下文中可以正确运行。

• • ExactMatch 是一种基于执行的指标。它表示生成的查询与正确查询文本完全匹配的百分比，这意味着它们在执行时会产生相同的结果。ExactMatch 是一种更严格的正确性衡量标准，与查询在真实场景中的实际效用直接相关。

尽管微调模型取得了一些令人鼓舞的结果，但整体准确率表明 Text2Cypher 仍是一项不断发展的技术。一些模型在每种情况下生成完全正确查询方面仍有困难，这凸显了进一步改进的必要性。

在本文中，我们将尝试使用 LlamaIndex 工作流来实现更具智能性的 Text2Cypher 策略。与通常的单次查询生成（大多数基准测试的运行方式）不同，我们将尝试一种多步骤方法，允许重试或替代查询形式。通过引入这些额外步骤和回退选项，我们旨在提高整体准确性并减少错误 Cypher 生成的情况。

代码已在 GitHub 上提供（https://github.com/tomasonjo-labs/text2cypher_llama_agent）。我们还提供了应用的托管版本。感谢 Anej Gorkic 对应用的贡献以及调试的帮助。:)

LlamaIndex 工作流

LlamaIndex 工作流是一种实用的方法，通过事件驱动系统将不同的操作连接起来，从而组织多步骤的 AI 处理过程。它有助于将复杂任务分解为更小、更易管理的部分，这些部分可以以结构化的方式相互通信。工作流中的每一步处理特定事件并生成新事件，从而创建一系列操作链，完成诸如文档处理、问答或内容生成等任务。系统会自动处理步骤之间的协调，使构建和维护复杂的 AI 应用程序变得更加容易。

简单的 Text2Cypher 流程

简单的 Text2Cypher 架构是一种将自然语言问题转换为 Neo4j 图数据库 Cypher 查询的精简方法。它通过以下三阶段工作流运行：

1. 1. 使用存储在向量数据库中的类似示例，通过少样本学习生成输入问题的 Cypher 查询。

2. 2. 系统针对图数据库执行生成的 Cypher 查询。

3. 3. 通过语言模型处理数据库结果，生成直接回答原始问题的自然语言响应。

这种架构保持了简单但高效的管道，利用向量相似性搜索（例如少样本检索）和大语言模型进行 Cypher 查询生成和响应格式化。

以下是简单 Text2Cypher 工作流的可视化：

值得注意的是，大多数 Neo4j 模式生成方法在处理多标签节点时表现不佳。这不仅是由于复杂性增加，还因为标签的组合爆炸可能会使提示词超载。为缓解这一问题，我们在模式生成过程中排除了 Actor
 和 Director
 标签：

@step
async def generate_cypher(self, ctx: Context, ev: StartEvent) -> ExecuteCypherEvent:
    question = ev.input
    # Cypher query generation using an LLM
    cypher_query = await generate_cypher_step(
        self.llm, question, self.few_shot_retriever
    )
    # Streaming event information to the web UI.
    ctx.write_event_to_stream(
        SseEvent(
            label="Cypher generation",
            message=f"Generated Cypher: {cypher_query}",
        )
    )

    # Return for the next step
    return ExecuteCypherEvent(question=question, cypher=cypher_query)

管道从 generate_cypher
 步骤开始：

@step
async def execute_query(
    self, ctx: Context, ev: ExecuteCypherEvent
) -> SummarizeEvent | CorrectCypherEvent:
    # Get global var
    retries = await ctx.get("retries")
    try:
        database_output = str(graph_store.structured_query(ev.cypher))
    except Exception as e:
        database_output = str(e)
        # Retry
        if retries < self.max_retries:
            await ctx.set("retries", retries + 1)
            return CorrectCypherEvent(
                question=ev.question, cypher=ev.cypher, error=database_output
            )

    return SummarizeEvent(
        question=ev.question, cypher=ev.cypher, context=database_output
    )

generate_cypher
 步骤通过使用语言模型和从向量存储中检索类似示例，将自然语言问题转换为 Cypher 查询。该步骤还会实时将生成的 Cypher 查询流式传输回用户界面，提供查询生成过程的即时反馈。您可以查看完整代码和提示词。

带重试机制的简单 Text2Cypher 流程

这种增强版的 Text2Cypher 流程在原始架构的基础上增加了自我修正机制。当生成的 Cypher 查询执行失败时，系统不会直接报错，而是通过 CorrectCypherEvent
 步骤将错误信息反馈回语言模型以修正查询。这使得系统更具弹性，能够处理初始错误，类似于人类在收到错误反馈后调整方法的方式。

以下是带重试机制的简单 Text2Cypher 工作流的可视化：

以下是 ExecuteCypherEvent
 的示例：

@step
async def evaluate_context(
    self, ctx: Context, ev: EvaluateEvent
) -> SummarizeEvent | CorrectCypherEvent:
    # Get global var
    retries = await ctx.get("retries")
    evaluation = await evaluate_database_output_step(
        self.llm, ev.question, ev.cypher, ev.context
    )
    if retries < self.max_retries and not evaluation == "Ok":
        await ctx.set("retries", retries + 1)
        return CorrectCypherEvent(
            question=ev.question, cypher=ev.cypher, error=evaluation
        )
    return SummarizeEvent(
        question=ev.question, cypher=ev.cypher, context=ev.context
    )

execute
 函数首先尝试运行查询，如果成功，则将结果传递给后续的总结步骤。然而，如果出现问题，它不会立即放弃，而是检查是否还有剩余的重试次数。如果有，它会将查询连同错误信息一起发送回修正步骤。这种机制创建了一个更容错的系统，能够从错误中学习，就像我们在收到反馈后调整方法一样。您可以查看完整代码和提示词（https://github.com/tomasonjo-labs/text2cypher_llama_agent/blob/main/app/workflows/naive_text2cypher_retry.py）。

带重试和评估机制的简单 Text2Cypher 流程

在带重试机制的简单 Text2Cypher 流程基础上，这种增强版增加了一个评估阶段，用于检查查询结果是否足以回答用户的问题。如果结果被认为不足，系统会将查询返回修正步骤，并附上改进建议。如果结果令人满意，流程则继续进入最终总结步骤。这一额外的验证层进一步增强了管道的弹性，确保用户最终获得最准确和完整的答案。

以下是带重试和评估机制的简单 Text2Cypher 工作流的可视化：

附加的评估步骤实现如下：

@step
async def evaluate_context(
    self, ctx: Context, ev: EvaluateEvent
) -> SummarizeEvent | CorrectCypherEvent:
    # Get global var
    retries = await ctx.get("retries")
    evaluation = await evaluate_database_output_step(
        self.llm, ev.question, ev.cypher, ev.context
    )
    if retries < self.max_retries and not evaluation == "Ok":
        await ctx.set("retries", retries + 1)
        return CorrectCypherEvent(
            question=ev.question, cypher=ev.cypher, error=evaluation
        )
    return SummarizeEvent(
        question=ev.question, cypher=ev.cypher, context=ev.context
    )

evaluate_check
 函数是一个简单的检查，用于确定查询结果是否充分回答了用户的问题。如果评估表明结果不足且还有剩余的重试次数，它会返回 CorrectCypherEvent
，以便进一步改进查询。否则，它会继续执行 SummarizeEvent
，表明结果适合进行最终总结。

后来我意识到，捕捉成功自我修正无效 Cypher 语句的实例是一个绝佳主意。这些实例可以作为动态少样本提示词，用于未来的 Cypher 生成。这种方法不仅使智能体能够自我修复，还能随着时间的推移不断自我学习和改进。

@step
async def summarize_answer(self, ctx: Context, ev: SummarizeEvent) -> StopEvent:
    retries = await ctx.get("retries")
    # If retry was successful:
    if retries > 0 and check_ok(ev.evaluation):
        # print(f"Learned new example: {ev.question}, {ev.cypher}")
        # Store success retries to be used as fewshots!
        store_fewshot_example(ev.question, ev.cypher, self.llm.model)

迭代规划流程

最后的流程是最复杂的，巧合的是，这是我最初雄心勃勃设计的流程。我保留了代码，以便您可以从我的探索中学习。

迭代规划流程通过引入迭代规划系统实现了更复杂的方法。它并不是直接生成 Cypher 查询，而是首先创建一个子查询计划，在执行前验证每个子查询 Cypher 语句，并包含一个信息检查机制。如果初始结果不足，它可以修改计划。系统最多进行三次信息收集迭代，每次根据先前结果优化方法。这种方式创建了一个更全面的问题回答系统，可以通过分解复杂查询为可管理的步骤并在每个阶段验证信息来处理复杂查询。

以下是迭代规划工作流的可视化：

让我们来看看查询规划提示词。我在一开始时非常有雄心壮志，期望语言模型生成如下响应：

class SubqueriesOutput(BaseModel):
    """Defines the output format for transforming a question into parallel-optimized retrieval steps."""

    plan: List[List[str]] = Field(
        description=(
            """A list of query groups where:
        - Each group (inner list) contains queries that can be executed in parallel
        - Groups are ordered by dependency (earlier groups must be executed before later ones)
        - Each query must be a specific information retrieval request
        - Split into multiple steps only if intermediate results return ≤25 values
        - No reasoning or comparison tasks, only data fetching queries"""
        )
    )

输出代表了一个将复杂问题转化为顺序和并行查询步骤的结构化计划。每一步包括一组可以并行执行的查询，后续步骤依赖于前面的结果。查询严格用于信息检索，避免推理任务，并在需要时拆分为更小的步骤以管理结果大小。例如，以下计划首先并行列出两位演员的电影，然后在第二步中从第一步的结果中找出票房最高的电影。

plan = [
# 2 steps in parallel
    [
        "List all movies made by Tom Hanks in the 2000s.",
        "List all movies made by Tom Cruise in the 2000s.",
    ],
# Second step
    ["Find the highest profiting movie among winner of step 1"],
]

这一想法无疑很酷。它是一种聪明的方法，可以将复杂问题分解为更小的、可操作的步骤，甚至使用并行化来优化检索。这听起来像是能够真正加速流程的策略。但在实践中，期望语言模型可靠地执行这一策略有些过于雄心勃勃。并行化虽然在理论上高效，但引入了很多复杂性。步骤间的依赖关系、中间结果以及保持并行步骤之间的逻辑一致性可能会让即使是高级模型也容易出错。顺序执行虽然不那么炫酷，但目前更可靠，并显著减少了模型的认知负担。

此外，语言模型在处理诸如列表嵌套等结构化工具输出时往往表现不佳，尤其是在推理步骤之间的依赖关系时。在这里，我很想看看仅通过提示词（而不依赖工具输出）能在这些任务中提升模型表现到何种程度。

查看迭代规划流程的代码：https://github.com/tomasonjo-labs/text2cypher_llama_agent/blob/main/app/workflows/iterative_planner.py

基准测试

为在 LlamaIndex 工作流架构中评估 Text2Cypher 智能体创建基准测试数据集，感觉像是向前迈出的令人兴奋的一步。

我们寻找了一种替代传统单次 Cypher 执行指标（如 ExactMatch）的方法，因为这些指标往往无法全面反映像迭代规划这样的工作流的潜力。在这些工作流中，通过多步骤流程来优化查询并检索相关信息，使得单步骤执行指标显得不足。

因此，我们选择了 Ragas 的**答案相关性（answer relevancy）**指标——它更符合我们想要衡量的内容。在这里，我们使用语言模型生成答案，然后将其作为裁判，与真实答案进行比较。我们准备了约 50 个样本的自定义数据集，设计时避免生成过多或过于详细的数据库输出（即过大的输出可能使语言模型裁判难以有效评估相关性）。保持结果简洁可以确保对单步骤和多步骤工作流的公平比较。

以下是结果：

Claude 3.5 Sonnet、Deepseek-V3 和 GPT-4o 成为答案相关性方面的前三名模型，每个模型得分均超过 0.80。NaiveText2CypherRetryCheckFlow 通常产生最高的相关性，而 IterativePlanningFlow 的排名始终较低（最低降至 0.163）。

尽管 OpenAI o1 模型相当准确，但由于多次超时（设定为 90 秒），可能未能跻身榜首。Deepseek-V3 尤其令人期待，其得分较高且延迟相对较低。总体来看，这些结果强调了实际部署场景中不仅需要关注准确性，还需要关注稳定性和速度的重要性。

另附一张表格，便于查看不同流程之间的提升效果：

Sonnet 3.5 的分数从 NaiveText2CypherFlow 的 0.596 稳步上升到 NaiveText2CypherRetryFlow 的 0.616，然后大幅跃升至 NaiveText2CypherRetryCheckFlow 的 0.843。GPT-4o 的整体模式类似，从 NaiveText2CypherFlow 的 0.622 略微下降到 NaiveText2CypherRetryFlow 的 0.603，但随后显著上升至 NaiveText2CypherRetryCheckFlow 的 0.837。这些改进表明，添加重试机制和最终验证步骤显著提高了答案相关性。

查看基准测试代码：https://github.com/tomasonjo-labs/text2cypher_llama_agent/blob/main/benchmark/benchmark_gridsearch.ipynb。

请注意，基准测试结果可能会有至少 5% 的波动，这意味着您可能会在不同运行中观察到略有不同的结果和表现最佳的模型。

收获与生产部署

这是一个为期两个月的项目，我在此过程中学到了很多。一个亮点是，在测试基准中达到了 84% 的相关性，这是一项重要成就。然而，这是否意味着您能在生产中达到 84% 的准确率？可能未必。

生产环境带来了自己的挑战——真实世界的数据通常比基准数据集更嘈杂、更多样化且更不结构化。我们尚未讨论的一点是，您会在真实应用和用户中看到，生产环境需要生产就绪的步骤。这不仅意味着专注于在受控基准测试中实现高准确率，还意味着确保系统在真实条件下可靠、适应性强并提供一致的结果。

在这些场景中，您需要实现某种类型的防护措施，以阻止无关问题通过 Text2Cypher 管道。

我们有一个防护措施实现示例（https://github.com/tomasonjo-labs/text2cypher_llama_agent/blob/main/app/workflows/steps/iterative_planner/guardrails.py）。除了简单地重新路由无关问题，初始防护步骤还可以通过引导用户了解他们可以提出的问题类型、展示可用工具并演示如何有效使用它们来帮助教育用户。

在以下示例中，我们强调了添加一个将用户输入值映射到数据库的过程的重要性。这一步对于确保用户提供的信息与数据库模式一致至关重要，从而实现准确的查询执行并最大限度地减少因数据不匹配或模糊导致的错误。

这是一个用户请求“科幻电影”的示例。问题在于，数据库中该类型存储为“Sci-Fi”，导致查询没有返回结果。

经常被忽视的是空值的存在。在真实世界数据中，空值很常见，必须加以考虑，尤其是在执行排序或类似操作时。未能妥善处理它们可能会导致意外结果或错误。

在此示例中，我们得到了一部评分为 Null
 的随机电影。为解决此问题，查询需要添加一个附加子句 WHERE m.imdbRating IS NOT NULL
。

还有一些情况，缺失信息不仅仅是数据问题，而是模式限制。例如，如果我们请求奥斯卡获奖电影，但模式中不包含任何关于奖项的信息，查询就无法返回所需结果。

由于大语言模型被训练为取悦用户，它仍然会生成一个符合模式但无效的结果。我还不确定如何最好地处理此类示例。

最后，我想提到查询规划部分。我使用以下查询计划来回答问题：

谁在 2000 年代制作了更多电影，汤姆·汉克斯还是汤姆·克鲁斯？对于获胜者，找到其票房最高的电影。

计划如下：

plan = [
# 2 steps in parallel
    [
        "List all movies made by Tom Hanks in the 2000s.",
        "List all movies made by Tom Cruise in the 2000s.",
    ],
# Second step
    ["Find the highest profiting movie among winner of step 1"],
]

看起来很令人印象深刻，但现实是，Cypher 非常灵活，GPT-4o 可以在单个查询中处理这个问题。

我认为并行化在这种情况下是多余的。如果您正在处理真正需要的复杂问题类型，可以包括查询规划器，但请记住，许多多跳问题可以通过单个 Cypher 语句高效处理。

这个示例突出了一个不同的问题：最终答案是模棱两可的，因为语言模型只获得了有限的信息，具体来说是汤姆·克鲁斯的《世界大战》。在这种情况下，推理已经在数据库中完成，因此语言模型不需要处理该逻辑。然而，语言模型往往默认以这种方式操作，因此提供完整上下文以确保准确且明确的响应非常重要。

最后，您还需要考虑如何处理返回大量结果的问题。

在我们的实现中，我们对结果强制设置了 100 条记录的硬限制。虽然这有助于管理数据量，但在某些情况下仍可能过多，甚至可能在推理过程中误导语言模型。

此外，并非本文中介绍的所有智能体都具备对话功能。您可能需要在开始时添加一个问题重写步骤以使其具有对话性，或者将其作为防护步骤的一部分。如果您有一个无法完全传递到提示中的大型图模式，则需要设计一个动态获取相关图模式的系统。

在生产环境中需要注意很多事项！

总结

智能体非常有用，但最好从简单入手，避免一开始就陷入过于复杂的实现中。专注于建立一个可靠的基准，以有效评估和比较不同架构。在工具输出方面，考虑尽量减少使用或坚持最简单的工具，因为许多智能体难以有效处理工具输出，通常需要手动解析。

学习资源