利用知识图谱提升RAG应用的准确性

本文转载自：lucas大叔 : 利用知识图谱提升rag应用的准确性
https://zhuanlan.zhihu.com/p/692595027
英文原文：enhancing the accuracy of rag applications with knowledge graphs
https://neo4j.com/developer-blog/enhance-rag-knowledge-graph/

一、关于 graphrag

图检索增强生成（graphrag）利用图数据库的结构化特性，以节点和关系的方式进行组织数据，增加了检索到信息的深度和关联上下文，是传统向量检索方法的有效补充。

图擅长以结构化的方式表示和存储异构和互连的信息，可以轻松地捕捉不同数据类型之间的复杂关系和属性。
相比之下，向量数据库往往难以处理此类结构化信息，因为它们的优势在于通过高维向量处理非结构化数据。
在rag应用中，可以将结构化图数据与非结构化文本的向量搜索相结合，以实现优势互补。

虽然知识图谱的概念已经比较普及，但构建知识图还是一项有挑战性的工作。
它涉及到数据的收集和结构化，需要对领域和图建模有深入的了解。
为了简化图谱构建过程，我们尝试利用llm来构建。

llm对语言和上下文有着深刻的理解，可以实现知识图创建过程重要部分的自动化。
通过分析文本数据，llm可以识别实体，理解实体之间的关系，并建议如何在图结构中最好地表示它们。
作为实验的结果，我们在langchain中添加了图构建模块的第一个版本，并将在这篇博客中进行演示。

相关代码：https://github.com/tomasonjo/blogs/blob/master/llm/enhancing_rag_with_graph.ipynb

二、neo4j环境配置

首先创建一个neo4j实例。
最简单的方法是在neo4j aura上启动一个免费实例，该实例提供neo4j数据库的云实例。
或者，你也可以下载 neo4j desktop 应用并创建neo4j数据库的本地实例。

os.environ["openai_api_key"] = "sk-"
os.environ["neo4j_uri"] = "bolt://localhost:7687"
os.environ["neo4j_username"] = "neo4j"
os.environ["neo4j_password"] = "password"

graph = neo4jgraph()

三、数据提取

本演示使用 elizabeth i 的维基百科页面，我们用langchain loader无缝地从维基百科抓取和分割文档。

# read the wikipedia article
raw_documents = wikipedialoader(query="elizabeth i").load()

# define chunking strategy
text_splitter = tokentextsplitter(chunk_size=512, chunk_overlap=24)
documents = text_splitter.split_documents(raw_documents[:3])

现在用分割后的文档构建图谱。
为此，我们实现了llmgraphtransformer模块，它大大简化了在图数据库中构建和存储知识图谱。

llmgraphtransformer类利用llm将文档转化为图谱文档，允许指定输出图谱中节点和关系类型的约束，不支持抽取节点或者关系的属性。
它的参数如下：

• llm (baselanguagemodel):支持结构化输出的语言模型实例
• allowed_nodes (list[str], optional): 指定图谱中包含哪些节点类型，默认是空list，允许所有节点类型
• allowed_relationships (list[str], optional): 指定图谱中包含哪些关系类型，默认是空list，允许所有关系类型
• prompt (optional[chatprompttemplate], optional): 传给llm的带有其他指令的prompt
• strict_mode (bool, optional): 确定转化是否应该使用筛选以严格遵守allowed_nodes 和 allowed_relationships，默认为true

本例allowed_nodes和allowed_relationships都采取默认设置，即图谱中允许所有的节点和关系类型。

llm=chatopenai(temperature=0, model_name="gpt-4-0125-preview")
llm_transformer = llmgraphtransformer(llm=llm)

# extract graph data
graph_documents = llm_transformer.convert_to_graph_documents(documents)
# store to neo4j
graph.add_graph_documents(
  graph_documents, 
  baseentitylabel=true, 
  include_source=true
)

你可以定义知识图谱生成链要使用的llm。目前，只支持openai和mistral的function-calling模型。在本例中，我们使用最新的gpt-4。
值得注意的是，生成图谱的质量在很大程度上取决于所使用的模型。
llm graph transformer 返回图谱文档，通过add_graph_documents方法导入到neo4j。
baseentitylabel参数为每个节点分配一个额外的__entity__标签，从而提高索引和查询性能。
include_source参数将节点链接到其原始文档，便于数据跟踪和上下文理解。

在neo4j浏览器中可以检查生成的图谱。

可以看到，每种类型的节点除了自身的节点类型之外，多了一个__entity__ 标签。

同时，节点通过mentions关系与源文档连接。

四、rag混合检索

在生成图谱之后，我们将向量索引和关键字索引的混合检索与图谱检索结合起来用于rag。

上图展示了从用户提出问题开始的检索过程，问题首先输入到rag检索器，该检索器采用关键词和向量搜索非结构化文本数据，并与从知识图谱中收集的信息结合。
由于neo4j同时具有关键词和向量索引，因此可以使用单个数据库实现全部三种检索方式。
从这些数据源收集的数据输入给llm生成最终答案。

1、非结构化数据检索器

可以用 neo4jvector.from_existing_graph 方法为文档添加关键字和向量检索。
此方法为混合搜索方法配置关键字和向量搜索索引，目标节点类型为document。
另外，如果文本embedding值缺失，它还会计算创建向量索引。

vector_index = neo4jvector.from_existing_graph(
    openaiembeddings(),
    search_type="hybrid",
    node_label="document",
    text_node_properties=["text"],
    embedding_node_property="embedding"
)

可以看到，document节点原来没有embedding属性，创建非结构化数据检索器后，基于document节点的text属性新创建了embedding。

然后使用similarity_search方法就可以调用向量索引。

2、图谱检索器

另一方面，配置图谱检索更为复杂，但提供了更多的自由度。
本示例将使用全文索引 识别相关节点并返回其一阶邻居。

图谱检索器首先识别输入中的相关实体。
为简单起见，我们让llm识别人、组织和地点等通用实体，用lcel和新添加的with_structured_output 方法来提取。

# extract entities from text
class entities(basemodel):
    """identifying information about entities."""

    names: list[str] = field(
        ...,
        description="all the person, organization, or business entities that "
        "appear in the text",
    )

prompt = chatprompttemplate.from_messages(
    [
        (
            "system",
            "you are extracting organization and person entities from the text.",
        ),
        (
            "human",
            "use the given format to extract information from the following "
            "input: {question}",
        ),
    ]
)

entity_chain = prompt | llm.with_structured_output(entities)

让我们测试一下：

entity_chain.invoke({"question": "where was amelia earhart born?"}).names
# ['amelia earhart']

现在实现了从问题中检测出实体，接下来用全文索引将它们映射到知识图谱。
首先，我们需要定义全文索引和一个函数，该函数将生成允许有些拼写错误的全文查询。

graph.query(
    "create fulltext index entity if not exists for (e:__entity__) on each [e.id]")

def generate_full_text_query(input: str) -> str:
    """
    generate a full-text search query for a given input string.

    this function constructs a query string suitable for a full-text search.
    it processes the input string by splitting it into words and appending a
    similarity threshold (~2 changed characters) to each word, then combines 
    them using the and operator. useful for mapping entities from user questions
    to database values, and allows for some misspelings.
    """
    full_text_query = ""
    words = [el for el in remove_lucene_chars(input).split() if el]
    for word in words[:-1]:
        full_text_query += f" {word}~2 and"
    full_text_query += f" {words[-1]}~2"
    return full_text_query.strip()

把上面的功能组装在一起实现图谱检索的结构化检索器。

# fulltext index query
def structured_retriever(question: str) -> str:
    """
    collects the neighborhood of entities mentioned
    in the question
    """
    result = ""
    entities = entity_chain.invoke({"question": question})
    for entity in entities.names:
        response = graph.query(
            """call db.index.fulltext.querynodes('entity', $query, {limit:2})
            yield node,score
            call {
              match (node)-[r:!mentions]->(neighbor)
              return node.id + ' - ' + type(r) + ' -> ' + neighbor.id as output
              union
              match (node)<-[r:!mentions]-(neighbor)
              return neighbor.id + ' - ' + type(r) + ' -> ' +  node.id as output
            }
            return output limit 50
            """,
            {"query": generate_full_text_query(entity)},
        )
        result += "\n".join([el['output'] for el in response])
    return result

structured_retriever 函数从检测用户问题中的实体开始，迭代检测到的实体，使用cypher模板来检索相关节点的一阶邻居。

print(structured_retriever("who is elizabeth i?"))
# elizabeth i - born_on -> 7 september 1533
# elizabeth i - died_on -> 24 march 1603
# elizabeth i - title_held_from -> queen of england and ireland
# elizabeth i - title_held_until -> 17 november 1558
# elizabeth i - member_of -> house of tudor
# elizabeth i - child_of -> henry viii
# and more...

3、最终的检索器

如开头所述，我们将结合非结构化和图谱检索器来创建传递给llm的最终上下文。

def retriever(question: str):
    print(f"search query: {question}")
    structured_data = structured_retriever(question)
    unstructured_data = [el.page_content for el in vector_index.similarity_search(question)]
    final_data = f"""structured data:
{structured_data}
unstructured data:
{"#document ". join(unstructured_data)}
    """
    return final_data

正如处理python一样，可以简单地使用f-string拼接输出。

五、定义rag chain

我们已经成功地实现了rag的检索组件。
首先引入查询重写功能，允许根据对话历史对当前问题进行改写。

# condense a chat history and follow-up question into a standalone question
_template = """given the following conversation and a follow up question, rephrase the follow up question to be a standalone question,
in its original language.
chat history:
{chat_history}
follow up input: {question}
standalone question:"""  # noqa: e501
condense_question_prompt = prompttemplate.from_template(_template)

def _format_chat_history(chat_history: list[tuple[str, str]]) -> list:
    buffer = []
    for human, ai in chat_history:
        buffer.append(humanmessage(content=human))
        buffer.append(aimessage(content=ai))
    return buffer

_search_query = runnablebranch(
    # if input includes chat_history, we condense it with the follow-up question
    (
        runnablelambda(lambda x: bool(x.get("chat_history"))).with_config(
            run_name="haschathistorycheck"
        ),  # condense follow-up question and chat into a standalone_question
        runnablepassthrough.assign(
            chat_history=lambda x: _format_chat_history(x["chat_history"])
        )
        | condense_question_prompt
        | chatopenai(temperature=0)
        | stroutputparser(),
    ),
    # else, we have no chat history, so just pass through the question
    runnablelambda(lambda x : x["question"]),
)

接下来，引入prompt利用集成混合检索器提供的上下文生成响应，完成rag链的实现。

template = """answer the question based only on the following context:
{context}

question: {question}
"""
prompt = chatprompttemplate.from_template(template)

chain = (
    runnableparallel(
        {
            "context": _search_query | retriever,
            "question": runnablepassthrough(),
        }
    )
    | prompt
    | llm
    | stroutputparser()
)

最后，继续测试混合rag实现。

chain.invoke({"question": "which house did elizabeth i belong to?"})
# search query: which house did elizabeth i belong to?
# 'elizabeth i belonged to the house of tudor.'

前面实现了查询重写功能，使rag链能够适配允许后续问题的对话设置。
考虑到我们使用向量和关键字搜索，必须重写后续问题以优化搜索过程。

chain.invoke(
    {
        "question": "when was she born?",
        "chat_history": [("which house did elizabeth i belong to?", "house of tudor")],
    }
)
# search query: when was elizabeth i born?
# 'elizabeth i was born on 7 september 1533.'

可以看到 when was she born? 首先被改写为“when was elizabeth i born?”，然后使用重写后的查询来检索相关上下文并回答问题。

2024-05-12（日）