2,896 讀數

如何使用知识图谱进行检索增强生成（无需图形数据库）

经过 DataStax12m2024/04/23

太長; 讀書

这篇文章探讨了知识图谱在 RAG 中的应用，使用 DataStax Astra DB 进行存储。示例代码在此笔记本中，使用一些原型代码通过 Astra DB 存储和检索知识图谱

featured image - 如何使用知识图谱进行检索增强生成（无需图形数据库）

检索增强生成（RAG）是指检索信息并使用其为生成式 AI 提供上下文信息的各种技术。最常见的形式是针对文本块进行操作，包括：

从原始文档（HTML、PDF、Markdown 等）中提取文本。
根据文档结构和语义将文本分块为特定大小。
将块存储在以块嵌入为键的向量数据库中。
检索与问题相关的块，以用作生成答案时的上下文。

然而，基于向量相似度的 RAG 也存在一些缺点。由于它关注与问题类似的信息，因此很难回答涉及多个主题和/或需要多次跳跃的问题。此外，它还限制了检索到的块数。

每个块来自不同的来源，因此在多个地方存在大量相似信息的情况下，它需要在检索信息的多个副本（可能错过其他信息）或仅选择一份副本以获得更多不同的块之间进行选择，然后就会错过其他来源的细微差别。

知识图谱可以用作基于向量的块检索的替代或补充。在知识图谱中，节点对应于特定实体，边表示实体之间的关系。当用于 RAG 时，会提取与问题相关的实体，然后检索包含这些实体及其相关信息的知识子图。

与基于相似性的方法相比，这种方法有几个优点：

可以从单一来源提取许多事实，并将其与知识图谱中的各种实体相关联。这样就可以从给定来源中检索相关事实，而不是检索整个块（包括不相关的信息）。
如果多个来源说的是同一件事，它们会产生相同的节点或边。它们可以被视为相同的节点或边，只检索一次，而不是将它们视为不同的事实（并检索多个副本）。这样就可以检索更广泛的事实和/或只关注出现在多个来源中的事实。
该图谱可能经过多个步骤 — 不仅检索与问题中的实体直接相关的信息，而且还会提取 2 或 3 步之外的内容。在传统的 RAG 方法中，这需要多轮查询。

除了使用知识图谱进行 RAG 的好处之外，LLM 还使创建知识图谱变得更加容易。LLM 和提示可用于从文档中提取信息，而不需要主题专家精心制作知识图谱。

这篇文章探讨了知识图谱在 RAG 中的应用，使用数据Stax Astra 数据库用于存储。示例代码位于此处笔记本使用一些原型代码，通过 Astra DB 存储和检索知识图谱这个存储库我们将利用 LangChain 的“LLMGraphTransformer”从文档中提取知识图谱，将其写入 Astra，并讨论调整用于知识提取的提示的技术。

然后，我们将创建 LangChain Runnable，用于从问题中提取实体并检索相关子图。我们将看到，使用知识图谱实现 RAG 所需的操作不需要图形数据库或图形查询语言，因此可以使用您可能已经在使用的典型数据存储来应用该方法。

知识图谱

如前所述，知识图谱将不同的实体表示为节点。例如，一个节点可能表示“居里夫人”这个人，或“法国人”这个语言。在 LangChain 中，每个节点都有一个名称和一个类型。在唯一标识节点时，我们会同时考虑两者，以区分“法国人”这个语言和“法国人”这个国籍。

实体之间的关系对应于图中的边。每条边包括源（例如，居里夫人这个人）、目标（诺贝尔奖奖项）和类型，指示源与目标的关系（例如，“获奖”）。

下面是使用 LangChain 从关于居里夫人的一段话中提取的示例知识图谱：

根据您的目标，您可以选择向节点和边添加属性。例如，您可以使用属性来识别诺贝尔奖的获奖时间和类别。在检索过程中遍历图时，这些属性可用于过滤边和节点。

提取：创建知识图谱

知识图谱中的实体和关系可以直接创建，也可以从现有的已知良好数据源导入。当您希望仔细整理知识时，这种方法很有用，但它很难快速整合新信息或处理大量信息。

幸运的是，LLM 可以轻松地从内容中提取信息，因此我们可以使用它们来提取知识图谱。

下面，我使用LLMGraphTransformer来自 LangChain，从有关居里夫人的一些信息中提取图表。这使用提示来指示 LLM 从文档中提取节点和边。它可以与 LangChain 可以加载的任何文档一起使用，从而轻松添加到现有的 LangChain 项目中。

LangChain 支持其他选项，例如区分机器人你也可以看看一些可用的知识提取模型，比如反叛。

 from langchain_experimental.graph_transformers import LLMGraphTransformer from langchain_openai import ChatOpenAI from langchain_core.documents import Document # Prompt used by LLMGraphTransformer is tuned for Gpt4. llm = ChatOpenAI(temperature=0, model_name="gpt-4") llm_transformer = LLMGraphTransformer(llm=llm) text = """ Marie Curie, was a Polish and naturalised-French physicist and chemist who conducted pioneering research on radioactivity. She was the first woman to win a Nobel Prize, the first person to win a Nobel Prize twice, and the only person to win a Nobel Prize in two scientific fields. Her husband, Pierre Curie, was a co-winner of her first Nobel Prize, making them the first-ever married couple to win the Nobel Prize and launching the Curie family legacy of five Nobel Prizes. She was, in 1906, the first woman to become a professor at the University of Paris. """ documents = [Document(page_content=text)] graph_documents = llm_transformer.convert_to_graph_documents(documents) print(f"Nodes:{graph_documents[0].nodes}") print(f"Relationships:{graph_documents[0].relationships}")

这展示了如何使用 LangChain 的LLMGraphTransformer提取知识图谱。您可以使用存储库中的render_graph_document来呈现 LangChain GraphDocument以供目视检查。

在以后的文章中，我们将讨论如何检查整个知识图谱以及从每个文档中提取的子图，以及如何应用提示工程和知识工程来改进自动提取。

检索：使用子知识图谱进行回答

使用知识图谱回答问题需要几个步骤。我们首先确定从哪里开始遍历知识图谱。在这个例子中，我将提示 LLM 从问题中提取实体。然后，遍历知识图谱以检索这些起点给定距离内的所有关系。默认遍历深度为 3。检索到的关系和原始问题用于为 LLM 回答问题创建提示和上下文。

从问题中提取实体

与知识图谱的提取一样，可以使用特殊模型或具有特定提示的 LLM 来提取问题中的实体。为简单起见，我们将使用具有以下提示的 LLM，其中包含问题和有关要提取的格式的信息。我们使用具有名称和类型的 Pydantic 模型来获取正确的结构。

 QUERY_ENTITY_EXTRACT_PROMPT = ( "A question is provided below. Given the question, extract up to 5 " "entity names and types from the text. Focus on extracting the key entities " "that we can use to best lookup answers to the question. Avoid stopwords.\n" "---------------------\n" "{question}\n" "---------------------\n" "{format_instructions}\n" ) def extract_entities(llm): prompt = ChatPromptTemplate.from_messages([keyword_extraction_prompt]) class SimpleNode(BaseModel): """Represents a node in a graph with associated properties.""" id: str = Field(description="Name or human-readable unique identifier.") type: str = optional_enum_field(node_types, description="The type or label of the node.") class SimpleNodeList(BaseModel): """Represents a list of simple nodes.""" nodes: List[SimpleNode] output_parser = JsonOutputParser(pydantic_object=SimpleNodeList) return ( RunnablePassthrough.assign( format_instructions=lambda _: output_parser.get_format_instructions(), ) | ChatPromptTemplate.from_messages([QUERY_ENTITY_EXTRACT_PROMPT]) | llm | output_parser | RunnableLambda( lambda node_list: [(n["id"], n["type"]) for n in node_list["nodes"]]) )

运行上面的例子我们可以看到提取的实体：

 # Example showing extracted entities (nodes) extract_entities(llm).invoke({ "question": "Who is Marie Curie?"}) # Output: [Marie Curie(Person)]

当然，LangChain Runnable 可以在链中使用，从问题中提取实体。

将来，我们将讨论改进实体提取的方法，例如考虑节点属性或使用向量嵌入和相似性搜索来识别相关的起点。为了让这篇第一篇文章保持简单，我们将坚持上述提示，然后继续遍历知识图谱以检索knowledge-subgraph并将其作为提示中的上下文。

检索子知识图谱

上一个链为我们提供了相关节点。我们可以使用这些实体和图形存储来检索相关知识三元组。与 RAG 一样，我们将它们作为上下文的一部分放入提示中并生成答案。

 def _combine_relations(relations): return "\n".join(map(repr, relations)) ANSWER_PROMPT = ( "The original question is given below." "This question has been used to retrieve information from a knowledge graph." "The matching triples are shown below." "Use the information in the triples to answer the original question.\n\n" "Original Question: {question}\n\n" "Knowledge Graph Triples:\n{context}\n\n" "Response:" ) chain = ( { "question": RunnablePassthrough() } # extract_entities is provided by the Cassandra knowledge graph library # and extracts entitise as shown above. | RunnablePassthrough.assign(entities = extract_entities(llm)) | RunnablePassthrough.assign( # graph_store.as_runnable() is provided by the CassandraGraphStore # and takes one or more entities and retrieves the relevant sub-graph(s). triples = itemgetter("entities") | graph_store.as_runnable()) | RunnablePassthrough.assign( context = itemgetter("triples") | RunnableLambda(_combine_relations)) | ChatPromptTemplate.from_messages([ANSWER_PROMPT]) | llm )

可以执行上述链来回答问题。例如：

 chain.invoke("Who is Marie Curie?") # Output AIMessage( content="Marie Curie is a Polish and French chemist, physicist, and professor who " "researched radioactivity. She was married to Pierre Curie and has worked at " "the University of Paris. She is also a recipient of the Nobel Prize.", response_metadata={ 'token_usage': {'completion_tokens': 47, 'prompt_tokens': 213, 'total_tokens': 260}, 'model_name': 'gpt-4', ... } )

遍历，不要查询

虽然使用图形数据库来存储知识图谱似乎很直观，但实际上并非如此。检索几个节点周围的子知识图谱是一个简单的图形遍历，而图形数据库则设计用于更复杂的查询，搜索具有特定属性序列的路径。此外，遍历通常只有 2 或 3 个深度，因为距离较远的节点很快就会与问题无关。这可以表示为几轮简单查询（每个步骤一个）或 SQL 连接。

无需单独的图形数据库，知识图谱的使用就更加简单。此外，使用 Astra DB 或 Apache Cassandra 可以简化对图形和存储在同一位置的其他数据的事务写入，并且可能具有更好的扩展性。只有当您计划使用 Gremlin 或 Cypher 或类似工具生成和执行图形查询时，这种开销才是值得的。

但这对于检索子知识图谱来说只是过度了，并且会引发许多其他问题，例如查询在性能方面脱轨。

这种遍历在 Python 中很容易实现。使用 CQL 和 Cassandra 驱动程序实现此操作（同步和异步）的完整代码可以在存储库异步遍历的核心如下图所示：

 def fetch_relation(tg: asyncio.TaskGroup, depth: int, source: Node) -> AsyncPagedQuery: paged_query = AsyncPagedQuery( depth, session.execute_async(query, (source.name, source.type)) ) return tg.create_task(paged_query.next()) results = set() async with asyncio.TaskGroup() as tg: if isinstance(start, Node): start = [start] discovered = {t: 0 for t in start} pending = {fetch_relation(tg, 1, source) for source in start} while pending: done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED) for future in done: depth, relations, more = future.result() for relation in relations: results.add(relation) # Schedule the future for more results from the same query. if more is not None: pending.add(tg.create_task(more.next())) # Schedule futures for the next step. if depth < steps: # We've found a path of length `depth` to each of the targets. # We need to update `discovered` to include the shortest path. # And build `to_visit` to be all of the targets for which this is # the new shortest path. to_visit = set() for r in relations: previous = discovered.get(r.target, steps + 1) if depth < previous: discovered[r.target] = depth to_visit.add(r.target) for source in to_visit: pending.add(fetch_relation(tg, depth + 1, source)) return results