Tomaz Bratanic • 2025-01-15

使用 LlamaIndex 工作流构建知识图谱代理

知识图谱

检索增强生成 (RAG) 将持续存在，这有充分的理由。它是一个强大的框架，将先进的语言模型与有针对性的信息检索技术相结合，从而更快地访问相关数据并生成更准确、更具上下文意识的响应。虽然 RAG 应用程序通常专注于非结构化数据，但我非常喜欢将结构化数据整合到其中，这是一个至关重要但经常被忽视的方法。我最喜欢的方法之一是利用像 Neo4j 这样的图数据库。

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

使用 LLM 生成 Cypher。图片来源：https://neo4j.ac.cn/developer-blog/fine-tuned-text2cypher-2024-model/

Text2cypher 提供了非凡的灵活性，因为它允许用户使用自然语言提问，而无需了解底层的图模式或 Cypher 语法。然而，由于语言解释的细微差别以及对精确的模式特定细节的需求，其准确性仍有不足，如这篇 text2cypher 文章所示。

基准测试的最重要结果如下所示。

Text2cypher 基准测试结果。图片来自 https://medium.com/neo4j/benchmarking-using-the-neo4j-text2cypher-2024-dataset-d77be96ab65a

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

针对 text2cypher 任务进行微调的模型
开放基础模型
闭源基础模型

该基准测试使用两个指标评估它们生成正确 Cypher 查询的性能：GoogleBLEU（上图）和 ExactMatch（下图）。

GoogleBLEU 指标衡量生成查询与参考查询之间（在 n-grams 方面）的重叠程度。分数越高通常表示与参考更接近，但并不一定保证查询在数据库环境中能正确运行。

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

尽管微调取得了一些有前景的结果，但总体准确率水平表明 text2cypher 仍然是一项不断发展的技术。有些模型在所有情况下仍然难以生成完全正确的查询，这凸显了在该领域进一步改进的必要性。

在本文中，我们将通过实验 LlamaIndex 工作流来实现更具代理性的 text2cypher 策略。我们不会依赖于通常用于运行大多数基准测试的单次查询生成，而是尝试一种多步骤方法，允许重试或备用查询表达。通过加入这些额外的步骤和回退选项，我们的目标是提高整体准确性并减少生成有缺陷的 Cypher 查询的情况。

代码可在 GitHub 上获取。我们还有一个托管版本的应用程序可在此处访问。感谢 Anej Gorkic 对应用程序的贡献和调试帮助 :)

包含所有代理的托管 Web 应用程序可在以下地址访问：https://text2cypher-llama-agent.up.railway.app/

LlamaIndex 工作流

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

朴素的 text2cypher 流程

朴素的 text2cypher 架构是一种将自然语言问题转换为 Neo4j 图数据库 Cypher 查询的简化方法。它通过一个三阶段工作流程运行：首先，它利用存储在向量数据库中的相似示例进行少样本学习，从输入问题生成 Cypher 查询。然后，系统针对图数据库执行生成的 Cypher 查询。最后，它通过语言模型处理数据库结果，生成直接回答原始问题的自然语言响应。这种架构保持了一个简单而有效的流程，利用向量相似性搜索进行示例少样本检索，并利用 LLM 进行 Cypher 查询生成和响应格式化。

这是朴素 text2cypher 工作流的可视化。

值得注意的是，大多数 Neo4j 模式生成方法在处理多标签节点时都存在困难。这个问题不仅是由于增加了复杂性，还因为标签的组合爆炸，这会使提示词不堪重负。为了缓解这个问题，我们在模式生成过程中排除了 Actor 和 Director 标签。

schema = graph_store.get_schema_str(exclude_types=["Actor", "Director"])

该流程从 generate_cypher 步骤开始。

@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 步骤接收一个自然语言问题，并通过利用语言模型和从向量存储中检索相似示例将其转换为 Cypher 查询。该步骤还将生成的 Cypher 查询实时流式传输回用户界面，提供有关查询生成过程的即时反馈。您可以在此处查看完整的代码和提示词。

带有重试功能的朴素 text2cypher 流程

这个增强版的带有重试功能的 text2cypher 在原始架构的基础上增加了自我修正机制。当生成的 Cypher 查询执行失败时，系统不会直接失败，而是尝试通过在 CorrectCypherEvent 步骤中将错误信息反馈给语言模型来修复查询。这使得系统更具弹性，能够处理最初的错误，就像人类在收到错误反馈后可能会修改其方法一样。

这是带有重试功能的朴素 text2cypher 工作流的可视化。

让我们看看 ExecuteCypherEvent。

@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
    )

执行函数首先尝试运行查询，如果成功，则将结果转发进行总结。然而，如果出现问题，它不会立即放弃——相反，它会检查是否还有剩余的重试尝试，如果有，则将查询连同错误信息一起发送回去进行更正。这创建了一个更宽容、能够从错误中学习的系统，就像我们在收到反馈后可能会修改方法一样。您可以在此处查看完整的代码和提示词。

带有重试和评估功能的朴素 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 语句，并包含一个信息检查机制，如果初始结果不足，该机制可以修改计划。系统可以进行最多三次信息收集迭代，每次都根据之前的结果改进其方法。这创建了一个更全面的问答系统，可以通过将复杂查询分解为可管理的步骤并在每个阶段验证信息来处理它们。

这是迭代规划器工作流的可视化。

让我们检查一下查询规划器提示。当我开始时，我相当雄心勃勃。我期望 LLM 产生以下响应。

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"],
]

这个想法无疑很酷。这是一种将复杂问题分解为更小、可执行的步骤，甚至利用并行性优化检索的巧妙方法。这听起来像是能够真正加速事情的策略。但是，在实践中，期望 LLM 可靠地执行此操作有点雄心勃勃。并行性虽然在理论上高效，但引入了许多复杂性。依赖关系、中间结果以及保持并行步骤之间的逻辑一致性很容易让即使是先进的模型也出错。顺序执行虽然不那么引人注目，但目前更可靠，并显著降低了模型的认知开销。

此外，LLM 通常难以遵循像列表嵌套列表这样的结构化工具输出，尤其是在推理步骤之间的依赖关系时。在这里，我很好奇仅凭提示（没有工具输出）能在多大程度上提高模型在这些任务上的性能

迭代规划流程的代码可在此处获取。

基准测试

为评估 LlamaIndex 工作流架构中的 text2ypher 代理创建基准测试数据集感觉像是向前迈出了令人兴奋的一步。

我们寻求传统一次性 Cypher 执行指标（例如开头提到的 ExactMatch）的替代方案，因为这些指标通常无法充分捕捉迭代规划等工作流的全部潜力。在这些工作流中，采用多个步骤来改进查询和检索相关信息，使得单步执行指标不足以衡量性能。

这就是为什么我们选择使用 RAGAS 的 answer_relevancy 指标——它与我们想要衡量的目标更一致。在这里，我们使用一个 LLM 来生成答案，然后使用另一个 LLM 作为评判者将其与真实情况进行比较。我们准备了一个包含大约 50 个样本的自定义数据集，经过精心设计，以避免生成压倒性的输出，即过于庞大或详细的数据库结果。这样的输出会使得 LLM 评判者难以有效评估相关性，因此保持结果简洁可以确保公平、有重点地比较单步和多步工作流。

结果如下。

Sonnet 3.5、Deepsek-v3 和 GPT-4o 在答案相关性方面脱颖而出，位居前三名，各自得分均高于 0.80。总体而言，NaiveText2CypherRetryCheckFlow 倾向于产生最高的相关性，而 IterativePlanningFlow 的排名始终较低（最低降至 0.163）。

虽然 o1 模型相当准确，但由于多次超时（设置为 90 秒），它可能没有排在首位。Deepsek-v3 因其高分数和相对较低的延迟而显得尤为有前景。总体而言，这些结果强调了在实际部署场景中，不仅原始准确性重要，稳定性和速度也同样重要。

这是另一个表格，可以轻松查看流程之间的提升情况。

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

基准测试代码可在 此处找到。

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

经验总结和生产部署

这是一个为期两个月的项目，我在其中学到了很多东西。项目的一个亮点是在测试基准测试中达到了 84% 的相关性，这是一个重要的成就。然而，这是否意味着您在生产环境中也能达到 84% 的准确率？可能不会。

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

在这些设置中，您需要实施某种防护措施，以阻止不相关的问题通过 text-to-Cypher 管道。

我们在此处提供了一个防护措施的示例实现。除了简单地重定向不相关的问题外，初始的防护措施步骤还可以用于通过引导用户了解他们可以提出的问题类型、展示可用工具以及演示如何有效使用这些工具来帮助教育用户。

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

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

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

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

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

由于 LLM 经过训练旨在取悦用户，LLM 仍然会提供一些符合模式但实际上无效的内容。我目前还不清楚如何最好地处理这类示例。

我想提到的最后一点是查询规划部分。请记住，之前我使用以下计划查询来回答问题

在 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 语句高效处理。

这个例子突显了另一个问题：最终答案是模糊的，因为只向 LLM 提供了有限的信息。具体来说，只提到了赢家汤姆·克鲁斯和电影《世界大战》。在这种情况下，推理已经在数据库内部完成，因此 LLM 不需要处理该逻辑。然而，LLM 倾向于默认以这种方式运行，这强调了向 LLM 提供完整上下文以确保准确和清晰响应的重要性。

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