文本嵌入在RAG应用中的局限性

大家非常喜欢文本嵌入模型，这是有充分理由的：这些模型在处理非结构化文本方面特别出色，让找到语义相似的内容变得更容易。一点也不意外，它们是大多数RAG应用程序的核心，尤其是在现在需要从文档和其他文本资源中提取和查找相关信息的时代。然而，对于某些特定的问题，文本嵌入方法在处理时可能会出现错误，提供不准确的信息。

如前所述，文本嵌入非常适合处理无结构文本。另一方面，它们在处理结构化信息和执行比如筛选、排序或聚合等操作上不是很擅长。比如说一个简单的问题如下。

2024年评分最高的是哪部电影？

为了回答这个问题，我们首先按照发行年份进行过滤，然后按评分排序。我们将看看仅用文本嵌入的简单方法能带来什么样的结果，然后展示如何处理这类问题。这篇博客文章展示了在处理诸如结构化数据的操作（例如过滤、排序或聚合）时，你需要用到不同于文本嵌入的工具。

代码可在这里找到：GitHub。

在这篇博客文章中，我们将利用Neo4j Sandbox中的推荐项目。推荐项目则使用了MovieLens数据集，该数据集包括电影、演员、评分等信息以及更多相关信息。

推荐数据库图形模式

以下代码将创建一个LangChain包装器并与Neo4j连接。

    os.environ["NEO4J_URI"] = "bolt://44.204.178.84:7687"  
    os.environ["NEO4J_USERNAME"] = "neo4j"  
    os.environ["NEO4J_PASSWORD"] = "minimums-triangle-saving"  
    # 设置Neo4j的环境变量，包括URI、用户名和密码

    graph = Neo4jGraph(refresh_schema=False)
    # 初始化Neo4j图数据库对象，不刷新模式

另外，您还需要一个OpenAI API密钥，并将其传递到以下代码中：

# 这里的API密钥部分保持不变

（请不要直接粘贴此部分代码）

os.environ["OPENAI_API_KEY"] = getpass.getpass("请输入您的OpenAI API密钥:")

数据库包含10,000部电影，但文本嵌入还未生成。为了不为所有电影计算嵌入，我们将为评分最高的1,000部电影添加一个名为 Target 的二级标签。

// 查询电影并按IMDb评分从高到低排序，选择前1000部电影，并将它们标记为Target类型
graph.query("""  
MATCH (m:Movie)  
WHERE m.imdbRating IS NOT NULL  
WITH m  
ORDER BY m.imdbRating DESC  
LIMIT 1000  
SET m:Target  
""")

决定要嵌入的内容很重要。既然我们要演示按年份过滤和评分排序，这些细节就不应该被排除在嵌入文本之外。这就是为什么我选择记录每部电影的发行年份、评分、标题和简介。

这里是一个用于《华尔街之狼》电影的嵌入文本示例：
The Wolf of Wall Street这部电影的中文名字是《华尔街之狼》：

    剧情梗概：根据乔登·贝尔福特的真实故事改编，从他成为富有的股票经纪人、过着奢华的生活，到因涉及犯罪、腐败和联邦政府的介入而落败。
    标题：华尔街之狼
    年份：2013
    IMDb评分：8.2

你可以说这不是一种很好的嵌入结构化数据的方法。我也不会反对，因为我也不知道最好的方法是什么。也许我们应该把键值对转换成文本或者其他形式。如果你有什么好主意，告诉我一下。

zh: LangChain中的Neo4j向量对象提供了一个方便的方法from_existing_graph，你可以选择要编码的哪些文本属性。

embedding = OpenAIEmbeddings(model="text-embedding-3-small")

neo4j_vector = Neo4jVector.from_existing_graph(
    embedding=embedding,
    index_name="movies",
    node_label="Target",
    text_node_properties=["plot", "title", "year", "imdbRating"],
    embedding_node_property="embedding",
)

在这个示例中，我们使用了OpenAI的text-embedding-3-small模型来生成嵌入向量。我们使用from_existing_graph方法初始化Neo4jVector对象。node_label参数用于筛选带有_Target_标签的节点。text_node_properties参数定义了需要嵌入的节点属性，包括plot、title、_year_和imdbRating。最后，embedding_node_property参数定义了用于存储嵌入向量的属性，命名为embedding。

我们先试着根据描述或情节找一部电影吧。

pretty_print(
    neo4j_vector.similarity_search(
        "一个小孩遇到他英雄的电影是关于什么的？"
    )
)

结果如下：

    plot: 一个年轻男孩与来自外太空的巨大机器人成为了朋友，而一个神经紧张的政府特工想要摧毁这个机器人。
    title: 铁巨人
    year: 1999
    imdbRating: 8.0

    plot: 一个男孩的朋友去世后，一位作家回忆起自己童年时寻找失踪男孩尸体的旅程。
    title: Stand by Me
    year: 1986
    imdbRating: 8.1

    plot: 一个天真无邪的小男孩独自上路寻找离家出走的母亲。不久，他发现了一个不寻常的保护者，一个脾气暴躁的老头，两人在途中经历了一系列意想不到的冒险。
    title: 菊次郎的夏天 (Kikujiro)
    year: 1999
    imdbRating: 7.9

    plot: 在一个生病卧床的日子里，一个男孩的祖父给他读了一个叫做《公主 Bride》的故事。
    title: 公主新娘
    year: 1987
    imdbRating: 8.1

结果看来总体上相当可靠。每次都有一个小男孩出现，不过我不清楚他每次是否都能遇到他的英雄。再说，数据集中只有1,000部电影，因此选择余地不大。

现在我们来试试一个需要简单筛选的查询：

    pretty_print(  
        neo4j_vector.similarity_search(  
            "2016年的哪些电影？"  
        )  
    )

以下是结果：

    plot: 六个短篇故事，探索处于困境中的人们行为的极端。
    title: 狂野故事
    year: 2014  
    imdbRating: 8.1  

    plot: 一名年轻男子在海上灾难中幸存，被卷入了一段冒险与发现的史诗旅程。在荒岛上，他与另一位幸存者——凶猛的孟加拉虎——建立了意想不到的联系。
    title: 少年派的奇幻漂流
    year: 2012  
    imdbRating: 8.0  

    plot: 讲述他从一个中西部小镇的年轻人，成长为华尔街金融巨鳄的过程，到因贪婪、腐败和政府调查而堕落的结局。
    title: 华尔街之狼
    year: 2013  
    imdbRating: 8.2  

    plot: 年轻的莱利从美国中西部被搬到旧金山，她在新城市、新家和新学校的经历让她感受到了各种情感：欢乐、恐惧、愤怒、厌恶和悲伤。
    title: 头脑特工队
    year: 2015  
    imdbRating: 8.3

这很有趣，但2016年的电影竟然没有一部被选中。也许我们可以尝试不同的文本准备方法来取得更好的结果。然而，文本向量化在这里并不适用，因为我们处理的是一个简单的结构化数据操作，需要根据元数据属性来筛选文档，比如这里的电影。元数据筛选是一种常用的方法，常被用来提高RAG系统的准确性。

接下来的查询语句，首先需要做一些排序工作。

    pretty_print(  
        neo4j_vector.similarity_search("哪部电影的IMDb评分是最高的？")  
    )

结果是：

    plot: 一家默片制作公司及其演员们艰难地过渡到有声电影。
    title: 雨中曲
    year: 1952
    imdbRating: 8.3

    plot: 纪录了伍德斯托克音乐节前最伟大的摇滚音乐节之一。
    title: 蒙特雷流行音乐节
    year: 1968
    imdbRating: 8.1

    plot: 这部电影记录了阿波罗任务，也许是任何两小时内电影中最全面的。阿尔·莱内特观看了所有拍摄的任务镜头——超过600万英尺……
    title: 为全人类
    year: 1989
    imdbRating: 8.2

    plot: 一名不择手段的电影制片人利用一位女演员、一位导演和一位编剧来取得成功。
    title: 坏种子与美丽
    year: 1952
    imdbRating: 7.9

如果你对IMDb评分有所了解，你就会知道有很多电影的评分高于8.3。评分最高的是一个系列剧——《兄弟连》（Band of Brothers），获得了令人印象深刻的9.6分。同样地，文本嵌入在结果排序时表现不佳。

我们再来看一个问题，这个问题需要汇总相关数据。

pretty_print(neo4j_vector.similarity_search("有几部电影？"))

这里的结果是：

    plot: 十部电视剧情片，每部都基于十诫中的一个。  
    title: 十诫  
    year: 1989  
    imdbRating: 9.2  

    plot: 一部纪录片，要求前印度尼西亚死亡队领导人重现他们的屠杀事件，可以任意选择电影类型，包括经典好莱坞犯罪剧情和奢华歌舞场面。  
    title: 杀戮行动  
    year: 2012  
    imdbRating: 8.2  

    plot: 一个温和的霍比特人和他的八个同伴踏上了摧毁至尊戒和黑暗领主索伦的旅程。  
    title: 魔戒：远行  
    year: 2001  
    imdbRating: 8.8  

    plot: 当弗罗多和山姆在狡猾的咕噜的帮助下逐渐接近魔多时，分裂的队伍对索伦的新盟友萨鲁曼及其伊森加德大军展开了抵抗。  
    title: 魔戒：双塔奇兵  
    year: 2002  
    imdbRating: 8.7

这样的结果肯定帮不上忙，因为我们得到的是四部随机的电影。几乎不可能从这四部随机的电影中得出结论，即我们在这个例子中总共标注了1,000部电影。

所以，解决方法是什么呢？很简单：涉及filtering、sorting、aggregation等结构化处理的问题，就需要用到处理结构化数据的工具。

目前看来，大多数人考虑的是文本转查询的方法，其中大型语言模型根据问题和模式生成数据库查询。对于Neo4j，这是文本转Cypher；而对于SQL数据库，则是文本转SQL。然而，在实践中发现这种方法不够可靠，也不够健壮，不适合在实际环境中使用。

Cypher 生成语句评估。摘自我写的一篇关于 Cypher 评估的文章。

你可以使用链式思考、少量样本等技术，或者进行微调，但在现阶段要达到高准确度几乎是不可能的。text2query方法在处理简单问题和结构清晰的数据库模式时表现良好，但这并不是生产环境中的实际情况。为了解决这个问题，我们将生成数据库查询的复杂度从大型语言模型中移出，将其视为一个代码生成问题，根据函数输入来生成数据库查询。这种方法的优势在于显著提高了鲁棒性，尽管代价是灵活性的降低。与其试图回答所有问题但不准确，不如缩小RAG应用程序的范围，准确地回答相关问题。

由于我们要根据函数输入生成数据库查询指令——在这种情况下是Cypher语句——我们可以利用大语言模型（LLMs）的工具能力来完成这一任务。在这一过程中，LLM根据用户输入填充相关参数，而函数则负责获取所需的信息。在这个演示中，我们将首先实现两个工具函数：一个用于统计电影的数量，另一个用于列出电影，然后使用LangGraph创建一个LLM代理来进行演示。

我们从实现一个根据预定义过滤器来计数电影的工具开始。首先，我们需要定义这些过滤器是什么，并描述何时何地以及如何使用这些过滤器给LLM模型。

    class MovieCountInput(BaseModel):  
        min_year: Optional[int] = Field(  
            description="电影的最小发行年份"  
        )  
        max_year: Optional[int] = Field(  
            description="电影的最大发行年份"  
        )  
        min_rating: Optional[float] = Field(description="IMDb评分的最小值")  
        grouping_key: Optional[str] = Field(  
            description="用于分组聚合的分组键", enum=["year",]  
        )

LangChain 提供了几种定义函数输入的方法，但我更喜欢使用 Pydantic 方法。在这个例子中，我们提供了三个可选的过滤器来筛选电影搜索结果：min_year、max_year 和 min_rating。这些过滤器基于结构化数据，是可选的，用户可以选择包含其中任意一个、所有或完全不使用这些过滤器。此外，我们还增加了一个 grouping_key 参数，用于指定是否按特定属性来分组计算结果。在这种情况下，唯一支持的分组方式是按年份分组，如在 enum 部分定义的那样。

现在我们来定义这个具体的函数：

    @tool("movie-count", args_schema=MovieCountInput)  
    def movie_count(  
        min_year: Optional[int],  
        max_year: Optional[int],  
        min_rating: Optional[float],  
        grouping_key: Optional[str],  
    ) -> List[Dict]:  
        """基于特定过滤条件计算电影数量"""  

        filters = [  
            ("t.year >= $min_year", min_year),  
            ("t.year <= $max_year", max_year),  
            ("t.imdbRating >= $min_rating", min_rating),  
        ]  

        # 根据函数输入动态创建参数  
        params = {  
            extract_param_name(condition): value  
            for condition, value in filters  
            if value is not None  
        }  
        where_clause = " AND ".join(  
            [condition for condition, value in filters if value is not None]  
        )  

        cypher_statement = "MATCH (t:Target) "  
        if where_clause:  
            cypher_statement += f"WHERE {where_clause} "  

        return_clause = (  
            f"t.`{grouping_key}`, count(t) AS movie_count"  
            if grouping_key is not None  
            else "count(t) AS movie_count"  
        )  

        cypher_statement += f"RETURN {return_clause}"  

        print(cypher_statement)  # 调试输出：  
        return graph.query(cypher_statement, params=params)

movie_count函数生成一个Cypher查询来根据可选的过滤条件和分组键统计电影数量。它首先定义一个过滤条件列表，并将提供的参数作为对应的值。这些过滤条件项用于动态构建WHERE子句，这个WHERE子句负责在Cypher语句中应用指定的过滤条件，只包括那些值不是None的条件。

RETURN子句随后会根据提供的grouping_key进行构建，要么根据grouping_key分组，要么直接计算总数。最后，函数执行查询并将结果返回。

该函数可以根据需要添加更多参数和更复杂的逻辑，但重要的是保持其清晰，以确保LLM可以正确调用它。

还是得从定义函数参数开始。

    class MovieListInput(BaseModel):  
        sort_by: str = Field(  
            description="如何排序电影，可以是最新或评分中的一个",  
            enum=["latest", "rating"],  
        )  
        k: Optional[int] = Field(description="需要返回的电影数量")  
        description: Optional[str] = Field(description="电影的描述")  
        min_year: Optional[int] = Field(  
            description="放映年份的最小值"  
        )  
        max_year: Optional[int] = Field(  
            description="放映年份的最大值"  
        )  
        min_rating: Optional[float] = Field(description="最低 IMDb 评分")

我们保留了电影计数函数中的三个相同过滤器，但增加了 description 参数。此参数使我们能够使用向量相似度搜索来搜索和列出电影。即使我们使用的是结构化工具和过滤器，这并不妨碍我们利用文本嵌入和向量搜索方法。我们通常不希望返回所有电影，因此我们包含了一个可选的 k 输入，默认值已设定。此外，在列表时，我们希望只返回最相关的电影。因此，我们可以根据评分或发行年份来排序电影。

咱们来实现这个函数吧：

    @tool("movie-list", args_schema=MovieListInput)  
    def movie_list(  
        sort_by: str = "rating",  
        k : int = 4,  
        description: Optional[str] = None,  
        min_year: Optional[int] = None,  
        max_year: Optional[int] = None,  
        min_rating: Optional[float] = None,  
    ) -> List[Dict]:  
        """根据特定筛选条件列出电影"""  

        # 处理仅使用向量查询的情况，当没有预筛选时  
        if description and not min_year and not max_year and not min_rating:  
            return neo4j_vector.similarity_search(description, k=k)  

        filters = [  
            ("t.year >= $min_year", min_year),  
            ("t.year <= $max_year", max_year),  
            ("t.imdbRating >= $min_rating", min_rating),  
        ]  

        # 根据函数参数动态创建参数  
        params = {  
            key.split("$")[1]: value for key, value in filters if value is not None  
        }  
        where_clause = " AND ".join(  
            [condition for condition, value in filters if value is not None]  
        )  

        cypher_statement = "MATCH (t:Target) "  
        if where_clause:  
            cypher_statement += f"WHERE {where_clause} "  

        # 添加带有排序的返回子句  
        cypher_statement += " RETURN t.title AS title, t.year AS year, t.imdbRating AS rating ORDER BY "  

        # 根据描述或其他标准处理排序逻辑  
        if description:  
            cypher_statement += (  
                "vector.similarity.cosine(t.embedding, $embedding) DESC "  
            )  
            params["embedding"] = embedding.embed_query(description)  
        elif sort_by == "rating":  
            cypher_statement += "t.imdbRating降序 "  
        else:  # 按最新年份进行排序  
            cypher_statement += "t.year降序 "  

        cypher_statement += " 限制到整数($limit)"  
        params["limit"] = k or 4  

        print(cypher_statement)  # 调试打印  
        data = graph.query(cypher_statement, params=params)  
        返回 data

此功能根据多个可选过滤条件查找电影列表。如果只提供了描述而没有其他筛选条件，它将进行向量索引相似度搜索来找出相关电影。当添加了额外筛选条件时，该功能构建一个Cypher查询语句，以根据指定标准（如发行年份和IMDb评分）来匹配电影，并可选择性地结合描述相似度。然后根据相似度分数、IMDb评分或年份对结果进行排序，并限制返回_k_部电影的结果。

我们将实现一个直接的[ReAct](https://langchain-ai.github.io/langgraph/how-tos/create-react-agent/]智能代理，使用“LangGraph”。

LangGraph代理实现

代理包括一个LLM和一个工具步骤。当我们与代理互动时，我们首先会调用LLM来判断是否需要使用工具。然后我们会进入一个循环：

1. 如果代理指示执行某个动作（即，调用工具），我们会运行相应的工具并将结果反馈给代理。
2. 如果没有代理提出运行工具的需求，我们将结束流程（回复给用户）。

代码实现尽可能简单。首先，我们将工具与LLM绑定，然后定义助手步骤。

    llm = ChatOpenAI(model='gpt-4-turbo')  

    tools = [movie_count, movie_list]  
    llm_with_tools = llm.bind_tools(tools)  

    # 系统信息  
    sys_msg = SystemMessage(content="你是一个乐于助人的电影信息助手，任务是查找并提供有关电影的相关信息的解释。")  

    # 定义了一个名为assistant的函数，接收一个MessagesState类型的参数  
    def assistant(state: MessagesState):  
       return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

接下来我们来定义LangGraph的流程：

    # 图
    builder = StateGraph(MessagesState)

    # 定义节点：这些节点执行任务
    builder.add_node("assistant", assistant)
    builder.add_node("tools", ToolNode(tools))

    # 定义边：这些边决定了控制流如何移动
    builder.add_edge(START, "assistant")
    builder.add_conditional_edges(
        "assistant",
        # 如果助理返回的最新消息（结果）是一个工具请求 -> tools_condition 将转向工具节点
        # 如果助理返回的最新消息（结果）不是一个工具请求 -> tools_condition 将转向结束
        tools_condition,
    )
    builder.add_edge("tools", "assistant")
    react_graph = builder.compile()

我们定义 LangGraph 中的两个节点，并用条件连接将它们连接起来。如果调用了一个工具，流程就会转向工具；否则，将结果返回给用户。

我们现在来测试一下我们的代理。

    messages = [  
        HumanMessage(  
            content="有哪些电影是关于女孩遇到她的英雄的?"  
        )  
    ]  
    messages = react_graph.invoke({"messages": messages})  
    # pretty_print()用于格式化输出消息内容
    for m in messages["messages"]:  
        m.pretty_print()

结果如下:

结果显示如下

首先，代理选择使用movie-list工具，并设置适当的description参数。虽然不清楚为何选择5这个数字，但它似乎对这个数字情有独钟。该工具返回了五部与情节最相关的电影，LLM最后只是简单地为用户总结了这些电影。

如果我们问ChatGPT为什么它喜欢k值为5，它会给出以下的回答。

接下来，我们来问一个稍微复杂点，需要筛选元数据的问题。

    messages = [  
        HumanMessage(  
            content="90年代有哪些电影是关于一个女孩遇见她的英雄的?"  
        )  
    ]  
    messages = react_graph.invoke({"messages": messages})  
    # 以下循环将遍历每个消息：
    for m in messages["messages"]:  
        # 美化输出
        m.pretty_print()

结果显示了：

得到了结果

这一次，额外的条件被用来筛选仅限于1990年代的影片。这将是一个使用预筛选方法进行元数据过滤的典型例子。生成的Cypher查询首先通过过滤上映年份来缩小电影范围至特定年份。在接下来的部分，Cypher查询利用文本嵌入和向量相似度搜索来寻找关于一个小女孩遇见她英雄的电影故事。

我们试着按不同标准数数电影。

    messages = [  
        HumanMessage(  
            content="在90年代评分高于9.1的电影有多少部?"  
        )  
    ]  
    messages = react_graph.invoke({"messages": messages}),  
    for m in messages["messages"]:  
        m.pretty_print()

以下是结果：

结果如下

使用一个专门的计数工具，复杂度从大模型转移到工具，使得大模型只需关注于填充函数参数。这种任务分离让系统更加高效和稳健，从而降低了大模型输入的复杂度。

由于软件代理可以一个接一个或者同时调用多个工具，我们用更复杂的功能来试一试。

messages = [  
    HumanMessage(  
        content="评分最高的电影之后每年发行了多少部电影？"  
    )  
]  
messages = react_graph.invoke({"messages": messages})  
for m in messages["messages"]:  
    m.pretty_print()

结果是:

结果是

如前所述，该代理可以调用多个工具来回答问题。在这个例子中，它首先列出评分最高的电影，来确定最高评分的电影是哪一年发布的。一旦获得了这些信息，它就调用电影计数工具功能，来统计在指定年份之后发布的电影数量，使用问题中提到的分组键。

虽然文本嵌入在搜索非结构化数据方面表现出色，但在进行结构化操作，如过滤、排序和聚合时却显得力不从心。这些任务需要专门设计用于结构化数据的工具，这些工具提供了进行这些操作所需的精确性和灵活性。关键是，扩展系统中的工具集可以让你应对更广泛的用户查询，从而让你的应用程序更加健壮和灵活。结合结构化数据方法和非结构化文本搜索技术可以提供更准确和相关的结果，从而增强RAG应用程序中的用户体验。

一如既往，代码可以在这个Github链接上找到。