大模型实战：如何使用图数据库提高向量搜索精确度？

文本嵌入和向量搜索技术可以帮助我们根据文档的含义及其相似性来检索文档。但当需要根据日期或类别等特定标准来筛选信息时，这些技术就显得力不从心。

为了解决这个问题，我们可以引入元数据过滤或过滤向量搜索，这允许我们根据用户的特定需求来缩小搜索范围。喜欢本文记得收藏、关注、点赞。

在这里插入图片描述

技术交流

技术要学会分享、交流，不建议闭门造车。一个人可以走的很快、一堆人可以走的更远。

成立了大模型面试和技术交流群，相关源码、资料、技术交流&答疑，均可加我们的交流群获取，群友已超过2000人，添加时最好的备注方式为：来源+兴趣方向，方便找到志同道合的朋友。

方式①、微信搜索公众号：机器学习社区，后台回复：加群
方式②、添加微信号：mlc2040，备注：来自CSDN + 技术交流

例如，用户可能想要了解 2021 年实施的新政策。通过使用元数据过滤器，系统可以先筛选出 2021 年的文档，然后在这些文档中执行向量相似性搜索，以找到与用户兴趣最相关的文档。这种先进行元数据过滤再执行向量搜索的两步策略，能够显著提高搜索的相关性和准确性。

近期，Neo4j 引入了基于节点属性的 LangChain 元数据过滤支持。由于图形数据库能够存储复杂的结构化和非结构化数据，我们可以利用这些数据来执行更精细的元数据过滤。

在这里插入图片描述

以一个包含文章和组织信息的数据集为例，文章节点包含了文本和嵌入值，而与文章相关联的组织节点则包含了日期、情感、作者等更多信息。通过这些信息，我们可以构建复杂的查询，以回答如

Rod Johnson 所在的公司是否实施了新的在家工作政策？
Neo4j 投资的公司是否有负面新闻？
与为现代汽车供应的公司相关的供应链问题是否有任何值得注意的新闻？

等问题。

在本篇博客中，Tomaz Bratanic 将向我们展示如何结合 LangChain 和 OpenAI 函数调用代理来实现基于图的元数据过滤。

概览

我们将使用 Neo4j 托管的公共演示服务器上的 companies 图数据集。您可以通过以下凭据访问该数据集：

URI: https://demo.neo4jlabs.com:7473/browser/
用户名: companies
密码: companies
数据库: companies

在这里插入图片描述

数据集的完整模式包括以 Organization 节点为中心的丰富信息，涵盖供应商、竞争对手、位置、董事会成员等。此外，还有提及特定组织的文章及其相应的文本块。

我们将实现一个 OpenAI 代理，它可以根据用户输入动态生成 Cypher 语句，并从图形数据库检索相关文本块。这个工具将提供四个可选输入参数：

主题：用户感兴趣的特定信息或主题。
组织：用户希望查询信息的组织。
国家：用户感兴趣的组织的国家。
情感：文章的情感倾向。

我们将根据这些输入参数动态构建相应的 Cypher 语句，从图形数据库检索相关信息，并利用大型语言模型（LLM）生成最终答案。

要跟随代码实践，您将需要一个 OpenAI API 密钥。

功能实现

我们从设置 Neo4j 的连接凭证和相关连接开始。

import os

os.environ["OPENAI_API_KEY"] = "sk-"
os.environ["NEO4J_URI"] = "neo4j+s://demo.neo4jlabs.com"
os.environ["NEO4J_USERNAME"] = "companies"
os.environ["NEO4J_PASSWORD"] = "companies"
os.environ["NEO4J_DATABASE"] = "companies"

embeddings = OpenAIEmbeddings()
graph = Neo4jGraph()
vector_index = Neo4jVector.from_existing_index(
 embeddings,
 index_name="news"
)

我们使用 OpenAI 的文本嵌入技术，您需要一个 API 密钥来使用它。接下来，我们定义了与 Neo4j 的连接，这使我们能够执行任意的 Cypher 语句。最后，我们创建了一个 Neo4jVector 连接，它可以通过查询现有的向量索引来检索信息。目前，我们不能将向量索引与预过滤方法结合使用，只能与后过滤方法结合使用。但本文将专注于预过滤方法与全面向量相似性搜索的结合使用。

本文的核心是一个名为 get_organization_news 的函数，它能够根据用户的需求动态生成 Cypher 查询语句并检索相关信息。为了清晰起见，我将代码分成了多个部分。

首先，我们定义了一组输入参数，这些参数都是可选的文字输入。特别地，topic 参数用来在文档中搜索特定的信息。在实际应用中，我们会将 topic 参数的值用于向量相似性搜索。另外三个参数则用于展示预过滤的方法。如果所有预过滤参数都没有提供，我们可以直接利用现有的向量索引来检索相关文档。如果提供了预过滤参数，我们会开始构建一个基础的 Cypher 查询语句，这个语句将用于后续的预过滤元数据方法。我们使用 CYPHER runtime = parallel parallelRuntimeSupport=all 指令来告诉 Neo4j 数据库，在可能的情况下使用并行运行时。然后，我们准备一个匹配语句来选择 Chunk 节点和它们关联的 Article 节点。

def get_organization_news(
 topic: Optional[str] = None,
 organization: Optional[str] = None,
 country: Optional[str] = None,
 sentiment: Optional[str] = None,
) -> str:
 # 如果没有预过滤条件，我们可以直接使用向量索引进行搜索
 if topic and not organization and not country and not sentiment:
     return vector_index.similarity_search(topic)
 # 使用并行运行时（如果可用）
 base_query = (
     "CYPHER runtime = parallel parallelRuntimeSupport=all "
     "MATCH (c:Chunk)<-[:HAS_CHUNK]-(a:Article) WHERE "
 )
 where_queries = []
 params = {"k": 5}  # 设置要检索的文本块数量

接下来，我们动态地向 Cypher 语句添加元数据过滤器。我们从 Organization 过滤器开始。

if organization:
    # 将组织名称映射到数据库中的候选项
    candidates = get_candidates(organization)
    if len(candidates) > 1:  # 如果候选选项太多，则需要用户进一步明确
        return f"请明确指出用户指的是以下哪个组织：{candidates}"
    # 添加一个过滤条件，筛选出提及特定组织的 articles
    where_queries.append(f"EXISTS {{(a)-[:MENTIONS]->(:Organization {{name: $organization}})}}")
    # 将组织名称作为参数传入
    params["organization"] = candidates[0]

如果系统识别出用户感兴趣的特定组织，我们会使用 get_candidates 函数将该组织的名称映射到数据库中的候选项。如果找到多个匹配项，我们会要求用户进一步明确。如果没有找到多个匹配项，我们会添加一个过滤条件，筛选出提及特定组织的 articles。为了安全起见，我们使用参数化查询而不是直接拼接查询字符串。

随后，我们处理用户可能基于提及的组织的国家进行预过滤的情况。

if country:
    # 由于国家名称标准化，不需要额外的映射
    where_queries.append(f"EXISTS {{(a)-[:MENTIONS]->(:Organization)-[:IN_CITY]->()-[:IN_COUNTRY]->(:Country {{name: $country}})}}")
    params["country"] = country

由于国家名称通常是标准化的，我们不需要将国家名称映射到数据库中的值，因为大型语言模型（LLM）已经熟悉大多数国家的名称。

随后，我们处理情感元数据的过滤。

if sentiment:
    if sentiment == "positive":
        where_queries.append("a.sentiment > $sentiment")
        params["sentiment"] = 0.5
    else:
        where_queries.append("a.sentiment < $sentiment")
        params["sentiment"] = -0.5

我们要求 LLM 仅接受正面或负面两种情感输入值，并将这些值映射到适当的过滤器上。

对于 topic 参数，我们采取了略有不同的处理方式，因为它不用于预过滤，而是用于向量相似性搜索。

if topic:  # 执行向量比较
    vector_snippet = (
        "WITH c, a, vector.similarity.cosine(c.embedding,$embedding) AS score "
        "ORDER BY score DESC LIMIT toInteger($k)"
    )
    params["embedding"] = embeddings.embed_query(topic)
else:  # 只返回最新的数据
    vector_snippet = "WITH c, a ORDER BY a.date DESC LIMIT toInteger($k)"

如果系统识别出用户对新闻中的特定主题感兴趣，我们使用主题输入的文本嵌入来找到最相关的文档。如果没有识别出特定主题，我们简单地返回最新的几篇文章，并避免向量相似性搜索。

最后，我们将 Cypher 语句组合起来，并用它来从数据库中检索信息。

return_snippet = "RETURN '#title ' + a.title + '\n#date ' + toString(a.date) + '\n#text ' + c.text AS output"

complete_query = (
    base_query + " AND ".join(where_queries) + vector_snippet + return_snippet
)

# 从数据库检索信息
data = graph.query(complete_query, params)
print(f"Cypher: {complete_query}\n")
# 在打印前安全地移除嵌入
params.pop('embedding', None)
print(f"参数: {params}")
return "###文章: ".join([el["output"] for el in data])

我们通过组合所有查询片段来构建最终的 complete_query。然后，我们使用动态生成的 Cypher 语句从数据库检索信息并返回结果。让我们通过一个示例输入来看看生成的 Cypher 语句。

get_organization_news(
  organization='neo4j',
  sentiment='positive',
  topic='远程工作'
)

# Cypher: CYPHER runtime = parallel parallelRuntimeSupport=all
# MATCH (c:Chunk)<-[:HAS_CHUNK]-(a:Article) WHERE 
# EXISTS {(a)-[:MENTIONS]->(:Organization {name: $organization})} AND 
# a.sentiment > $sentiment 
# WITH c, a, vector.similarity.cosine(c.embedding,$embedding) AS score 
# ORDER BY score DESC LIMIT toInteger($k) 
# RETURN '#title ' + a.title + '\n#date ' + toString(a.date) + '\n#text ' + c.text AS output

# 参数: {'k': 5, 'organization': 'Neo4j', 'sentiment': 0.5}

动态查询生成按预期工作，能够从数据库中检索到相关的信息。

构建新闻信息代理工具

接下来，我们将创建一个代理工具，用于处理新闻信息查询。首先，我们需要为输入参数编写一些说明。

fewshot_examples = """{输入：Google员工的健康福利在新闻中有哪些？查询：健康福利}
{输入：关于Google的最新正面新闻是什么？查询：无}
{输入：有关VertexAI和Google的新闻有哪些？查询：VertexAI}
{输入：关于Google的新产品有哪些新闻？查询：新产品}
"""

class NewsInput(BaseModel):
    topic: Optional[str] = Field(
        description="除了组织、国家和情感倾向之外，如果您对其他特定信息或话题感兴趣，请告诉我们。以下是一些示例："
        + fewshot_examples
    )
    organization: Optional[str] = Field(
        description="您希望了解信息的组织名称"
    )
    country: Optional[str] = Field(
        description="您感兴趣的组织的所在国家。请使用正式的国家名称，例如‘美利坚合众国’或‘法国’。"
    )
    sentiment: Optional[str] = Field(
        description="您想要查询的文章情感倾向", enum=["正面", "负面"]
    )

在定义预过滤参数时，我遇到了一些困难，特别是如何让 topic 参数按预期工作。为了解决这个问题，我提供了一些示例，帮助语言模型更好地理解用户的需求。同时，我们还向模型提供了关于国家名称格式的指导，并对情感倾向选项进行了枚举。

现在，我们可以定义一个自定义工具，为其指定一个名称和一段包含使用说明的描述。

class NewsTool(BaseTool):
    name = "新闻信息工具"
    description = (
        "当你需要在新闻中查找相关信息时，这个工具会非常有用。"
    )
    args_schema:Type[BaseModel] = NewsInput

    def _run(
        self,
        topic: Optional[str] = None,
        organization: Optional[str] = None,
        country: Optional[str] = None,
        sentiment: Optional[str] = None,
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> str:
        ""“使用这个工具来获取新闻信息。”""
        return get_organization_news(topic, organization, country, sentiment)

最后，我们需要定义一个代理执行器。这里，我使用了之前实现的 OpenAI 代理的 LCEL 实现。

llm = ChatOpenAI(temperature=0, model="gpt-4-turbo", streaming=True)
tools = [NewsTool()]

llm_with_tools = llm.bind(functions=[format_tool_to_openai_function(t) for t in tools])

prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            “你是一个乐于助人的助手，可以找到关于电影的信息并进行推荐。如果工具需要进一步的问题，请确保向用户询问以获得澄清。确保在后续问题中包含任何需要澄清的可用选项。只做用户明确请求的事情。”
        ),
        MessagesPlaceholder(variable_name="chat_history"),
        ("user", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)

agent = (
    {
        "input": lambda x: x["input"],
        "chat_history": lambda x: _format_chat_history(x["chat_history"])
        if x.get("chat_history")
        else [],
        "agent_scratchpad": lambda x: format_to_openai_function_messages(
            x["intermediate_steps"]
        ),
    }
    | prompt
    | llm_with_tools
    | OpenAIFunctionsAgentOutputParser()
)

agent_executor = AgentExecutor(agent=agent, tools=tools)

这个代理工具可以用于检索新闻信息。我们还添加了 聊天记录 消息占位符，这样代理就可以进行对话，并允许提出后续问题和回复。

实施测试

让我们尝试几个查询，看看生成的 Cypher 语句和参数是什么样的。

agent_executor.invoke(
  {"输入": "关于 neo4j 的一些正面新闻是什么？"}
)

# Cypher: CYPHER runtime = parallel parallelRuntimeSupport=all 
# MATCH (c:Chunk)<-[:HAS_CHUNK]-(a:Article) WHERE 
# EXISTS {(a)-[:MENTIONS]->(:Organization {name: $organization})} AND 
# a.sentiment > $sentiment WITH c, a 
# ORDER BY a.date DESC LIMIT toInteger($k) 
# RETURN '#标题 ' + a.title + '日期 ' + toString(a.date) + '文本 ' + c.text AS output
# 参数: {'k': 5, 'organization': 'Neo4j', 'sentiment': 0.5}

生成的 Cypher 语句是有效的。由于没有指定具体的主题，它返回了提到 Neo4j 的最后五篇正面文章的文本块。让我们尝试一个更复杂的例子：

agent_executor.invoke(
   {"输入": "关于法国公司的员工幸福感，有哪些最新的负面新闻？"}
)

# Cypher: CYPHER runtime = parallel parallelRuntimeSupport=all 
# MATCH (c:Chunk)<-[:HAS_CHUNK]-(a:Article) WHERE 
# EXISTS {(a)-[:MENTIONS]->(:Organization)-[:IN_CITY]->()-[:IN_COUNTRY]->(:Country {name: $country})} AND 
# a.sentiment < $sentiment 
# WITH c, a, vector.similarity.cosine(c.embedding,$embedding) AS score 
# ORDER BY score DESC LIMIT toInteger($k) 
# RETURN '#标题 ' + a.title + '日期 ' + toString(a.date) + '文本 ' + c.text AS output
# 参数: {'k': 5, 'country': 'France', 'sentiment': -0.5, 'topic': '员工幸福感'}