使用属性图索引

属性图是一个带有属性（即元数据）的标记节点（即实体类别、文本标签等），通过关系连接成结构化路径的知识集合。

在 LlamaIndex 中，PropertyGraphIndex 提供了关于以下内容的关键编排：

• 构建图
• 查询图

用法

可以通过简单导入类并使用来找到基本用法：

from llama_index.core import PropertyGraphIndex

# 创建
index = PropertyGraphIndex.from_documents(
    documents,
)

# 使用
retriever = index.as_retriever(
    include_text=True,  # 包括与匹配路径的源块
    similarity_top_k=2,  # 向量 kg 节点检索的前 k 个
)
nodes = retriever.retrieve("Test")

query_engine = index.as_query_engine(
    include_text=True,  # 包括与匹配路径的源块
    similarity_top_k=2,  # 向量 kg 节点检索的前 k 个
)
response = query_engine.query("Test")

# 保存和加载
index.storage_context.persist(persist_dir="./storage")

from llama_index.core import StorageContext, load_index_from_storage

index = load_index_from_storage(
    StorageContext.from_defaults(persist_dir="./storage")
)

# 从现有图存储（和可选的向量存储）加载
# 从现有图/向量存储加载
index = PropertyGraphIndex.from_existing(
    property_graph_store=graph_store, vector_store=vector_store, ...
)

构建

在 LlamaIndex 中，属性图的构建是通过对每个块执行一系列 kg_extractors，并将实体和关系作为元数据附加到每个 llama-index 节点来完成的。您可以在这里使用尽可能多的提取器，并且它们都将被应用。

如果您在摄入管道中使用了转换或元数据提取器，那么这将非常熟悉（并且这些 kg_extractors 与摄入管道兼容）！

使用适当的关键字参数设置提取器：

index = PropertyGraphIndex.from_documents(
    documents,
    kg_extractors=[extractor1, extractor2, ...],
)

# 插入额外的文档/节点
index.insert(document)
index.insert_nodes(nodes)

如果未提供，默认值为 SimpleLLMPathExtractor 和 ImplicitPathExtractor。

以下是所有 kg_extractors 的详细信息。

（默认）SimpleLLMPathExtractor

使用 LLM 提取短语句，以提示和解析单跳路径的形式 (entity1, relation, entity2)

from llama_index.core.indices.property_graph import SimpleLLMPathExtractor

kg_extractor = SimpleLLMPathExtractor(
    llm=llm,
    max_paths_per_chunk=10,
    num_workers=4,
    show_progress=False,
)

如果需要，还可以自定义提示和用于解析路径的函数。

以下是一个简单（但天真）的示例：

prompt = (
    "以下提供了一些文本。给定文本，提取每行中最多 {max_paths_per_chunk} 个形式为 `subject,predicate,object` 的知识三元组。避免停用词。\n"
)


def parse_fn(response_str: str) -> List[Tuple[str, str, str]]:
    lines = response_str.split("\n")
    triples = [line.split(",") for line in lines]
    return triples


kg_extractor = SimpleLLMPathExtractor(
    llm=llm,
    extract_prompt=prompt,
    parse_fn=parse_fn,
)

（默认）ImplicitPathExtractor

使用每个 llama-index 节点对象上的 node.relationships 属性提取路径。

由于它仅解析已存在于 llama-index 节点对象上的属性，因此此提取器无需运行 LLM 或嵌入模型。

from llama_index.core.indices.property_graph import ImplicitPathExtractor

kg_extractor = ImplicitPathExtractor()

SchemaLLMPathExtractor

按照允许的实体、关系以及可以连接到哪些关系的实体的严格模式提取路径。

使用 pydantic、LLM 的结构化输出和一些巧妙的验证，我们可以动态指定模式并验证每条路径的提取。

from typing import Literal
from llama_index.core.indices.property_graph import SchemaLLMPathExtractor

# 推荐大写，下划线分隔
entities = Literal["PERSON", "PLACE", "THING"]
relations = Literal["PART_OF", "HAS", "IS_A"]
schema = {
    "PERSON": ["PART_OF", "HAS", "IS_A"],
    "PLACE": ["PART_OF", "HAS"],
    "THING": ["IS_A"],
}

kg_extractor = SchemaLLMPathExtractor(
    llm=llm,
    possible_entities=entities,
    possible_relations=relations,
    kg_validation_schema=schema,
    strict=True,  # 如果为 false，将允许超出模式范围的三元组
    num_workers=4,
    max_paths_per_chunk=10,
    show_progres=False,
)

此提取器非常可定制化，并具有自定义以下选项的选项 - 模式的各个方面（如上所示） - extract_prompt - strict=False vs. strict=True，以允许超出模式范围的三元组或不允许 - 如果你是 Pydantic 的专家，并且想要创建自定义的 Pydantic 类并进行自定义验证，你可以传入自定义的 kg_schema_cls。

检索和查询

标记属性图可以通过多种方式进行查询，以检索节点和路径。在 LlamaIndex 中，我们可以同时组合多种节点检索方法！

# 创建一个检索器
retriever = index.as_retriever(sub_retrievers=[retriever1, retriever2, ...])

# 创建一个查询引擎
query_engine = index.as_query_engine(
    sub_retrievers=[retriever1, retriever2, ...]
)

如果没有提供子检索器，默认的是 LLMSynonymRetriever 和 VectorContextRetriever（如果启用了嵌入）。

目前所有的检索器包括： - LLMSynonymRetriever - 基于 LLM 生成的关键词/同义词进行检索 - VectorContextRetriever - 基于嵌入的图节点进行检索 - TextToCypherRetriever - 请求 LLM 根据属性图的模式生成 Cypher 查询 - CypherTemplateRetriever - 使用由 LLM 推断的参数的 Cypher 模板 - CustomPGRetriever - 易于子类化并实现自定义检索逻辑

通常，你会定义一个或多个这些子检索器，并将它们传递给 PGRetriever：

from llama_index.core.indices.property_graph import (
    PGRetriever,
    VectorContextRetriever,
    LLMSynonymRetriever,
)

sub_retrievers = [
    VectorContextRetriever(index.property_graph_store, ...),
    LLMSynonymRetriever(index.property_graph_store, ...),
]

retriever = PGRetriever(sub_retrievers=sub_retrievers)

nodes = retriever.retrieve("<query>")

继续阅读以下内容，了解所有检索器的更多细节。

(默认) LLMSynonymRetriever

LLMSynonymRetriever 接受查询，并尝试生成关键词和同义词以检索节点（因此也检索与这些节点连接的路径）。

显式声明检索器允许你自定义多个选项。以下是默认值：

from llama_index.core.indices.property_graph import LLMSynonymRetriever

DEFAULT_SYNONYM_EXPAND_TEMPLATE = (
    "给定一些初始查询，生成同义词或相关关键词，总共最多 {max_keywords} 个，"
    "考虑到大写、复数形式、常见表达等。\n"
    "提供所有同义词/关键词，用 '^' 符号分隔：'keyword1^keyword2^...'\n"
    "注意，结果应该在一行上，用 '^' 符号分隔。"
    "----\n"
    "查询: {query_str}\n"
    "----\n"
    "关键词: "
)


def parse_fn(self, output: str) -> list[str]:
    matches = output.strip().split("^")

    # 大写化以与摄入进行规范化
    return [x.strip().capitalize() for x in matches if x.strip()]


synonym_retriever = LLMSynonymRetriever(
    index.property_graph_store,
    llm=llm,
    # 包括检索路径的源块文本
    include_text=False,
    synonym_prompt=prompt,
    output_parsing_fn=parse_fn,
    max_keywords=10,
    # 节点检索后要遵循的关系深度
    path_depth=1,
)

retriever = index.as_retriever(sub_retrievers=[synonym_retriever])

(默认，如果支持) VectorContextRetriever

VectorContextRetriever 根据它们的向量相似性检索节点，然后获取与这些节点连接的路径。

如果你的图存储支持向量，那么你只需要管理该图存储进行存储。否则，你需要额外提供一个向量存储，除了图存储（默认使用内存中的 SimpleVectorStore）。

from llama_index.core.indices.property_graph import VectorContextRetriever

vector_retriever = VectorContextRetriever(
    index.property_graph_store,
    # 仅在图存储不支持向量查询时需要
    # vector_store=index.vector_store,
    embed_model=embed_model,
    # 包括检索路径的源块文本
    include_text=False,
    # 要获取的节点数
    similarity_top_k=2,
    # 节点检索后要遵循的关系深度
    path_depth=1,
    # 可以为 VectorStoreQuery 类提供任何其他关键字参数
    ...,
)

retriever = index.as_retriever(sub_retrievers=[vector_retriever])

TextToCypherRetriever

TextToCypherRetriever 使用图存储模式、你的查询和一个用于文本到 Cypher 的提示模板来生成和执行 Cypher 查询。

注意： 由于 SimplePropertyGraphStore 实际上不是图数据库，它不支持 Cypher 查询。

你可以使用 index.property_graph_store.get_schema_str() 来检查模式。

from llama_index.core.indices.property_graph import TextToCypherRetriever

DEFAULT_RESPONSE_TEMPLATE = (
    "生成的 Cypher 查询语句：\n{query}\n\n" "Cypher 响应：\n{response}"
)
DEFAULT_ALLOWED_FIELDS = ["text", "label", "type"]

DEFAULT_TEXT_TO_CYPHER_TEMPLATE = (
    index.property_graph_store.text_to_cypher_template,
)


cypher_retriever = TextToCypherRetriever(
    index.property_graph_store,
    # 自定义 LLM， 默认为 Settings.llm
    llm=llm,
    # 自定义文本到 Cypher 的模板。
    # 需要 `schema` 和 `question` 模板参数
    text_to_cypher_template=DEFAULT_TEXT_TO_CYPHER_TEMPLATE,
    # 自定义 Cypher 结果如何插入到文本节点中。
    # 需要 `query` 和 `response` 模板参数
    response_template=DEFAULT_RESPONSE_TEMPLAT,
    # 一个可选的可调用对象，用于清理/验证生成的 Cypher 查询语句
    cypher_validator=None,
    # 结果中允许的字段
    allowed_output_field=DEFAULT_ALLOWED_FIELDS,
)

注意：执行任意密码有其风险。确保您采取必要的措施（只读角色，沙盒环境等）以确保在生产环境中安全使用。

CypherTemplateRetriever

这是TextToCypherRetriever的更受限制的版本。与其让LLM自由生成任何密码语句，我们可以提供一个密码模板，让LLM填写空白处。

为了说明这是如何工作的，这里有一个小例子：

# 注意：需要当前的v1版本
from pydantic.v1 import BaseModel, Field
from llama_index.core.indices.property_graph import CypherTemplateRetriever

# 使用模板参数编写查询
cypher_query = """
MATCH (c:Chunk)-[:MENTIONS]->(o)
WHERE o.name IN $names
RETURN c.text, o.name, o.label;
"""


# 创建一个pydantic类来表示我们查询的参数
# 类字段直接用作运行密码查询的参数
class TemplateParams(BaseModel):
    """密码查询的模板参数。"""

    names: list[str] = Field(
        description="用于在知识图中查找的实体名称或关键词列表。"
    )


template_retriever = CypherTemplateRetriever(
    index.property_graph_store, TemplateParams, cypher_query
)

存储

目前，用于属性图的支持图存储包括：

	内存中	本地嵌入支持	异步	服务器或基于磁盘？
SimplePropertyGraphStore	✅	❌	❌	磁盘
Neo4jPropertyGraphStore	❌	✅	❌	服务器

存储到/从磁盘

默认的属性图存储SimplePropertyGraphStore将所有内容存储在内存中，并持久化和从磁盘加载。

以下是使用默认图存储保存/加载索引的示例：

from llama_index.core import StorageContext, load_index_from_storage
from llama_index.core.indices import PropertyGraphIndex

# 创建
index = PropertyGraphIndex.from_documents(documents)

# 保存
index.storage_context.persist("./storage")

# 加载
storage_context = StorageContext.from_defaults(persist_dir="./storage")
index = load_index_from_storage(storage_context)

集成的保存和加载

集成通常会自动保存。一些图存储将支持向量，而其他一些可能不支持。您也可以将图存储与外部向量数据库结合使用。

此示例显示了如何使用Neo4j和Qdrant保存/加载属性图索引。

注意：如果未传入qdrant，则neo4j将自行存储和使用嵌入。此示例说明了超出此范围的灵活性。

pip install llama-index-graph-stores-neo4j llama-index-vector-stores-qdrant

from llama_index.core import StorageContext, load_index_from_storage
from llama_index.core.indices import PropertyGraphIndex
from llama_index.graph_stores.neo4j import Neo4jPGStore
from llama_index.vector_stores.qdrant import QdrantVectorStore
from qdrant_client import QdrantClient, AsyncQdrantClient

vector_store = QdrantVectorStore(
    "graph_collection",
    client=QdrantClient(...),
    aclient=AsyncQdrantClient(...),
)

graph_store = Neo4jPGStore(
    username="neo4j",
    password="<password>",
    url="bolt://localhost:7687",
)

# 创建一个索引
index = PropertyGraphIndex.from_documents(
    documents,
    property_graph_store=graph_store,
    # 可选，neo4j也直接支持向量
    vector_store=vector_store,
    embed_kg_nodes=True,
)

# 从现有的图/向量存储加载
index = PropertyGraphIndex.from_existing(
    property_graph_store=graph_store,
    # 可选，neo4j也直接支持向量
    vector_store=vector_store,
    embed_kg_nodes=True,
)

直接使用属性图存储

属性图的基本存储类是PropertyGraphStore。这些属性图存储是使用不同类型的LabeledNode对象构建的，并使用Relation对象连接。

我们可以自己创建这些，并且也可以自己插入！

from llama_index.core.graph_stores import (
    SimplePropertyGraphStore,
    EntityNode,
    Relation,
)
from llama_index.core.schema import TextNode

graph_store = SimplePropertyGraphStore()

entities = [
    EntityNode(name="llama", label="ANIMAL", properties={"key": "val"}),
    EntityNode(name="index", label="THING", properties={"key": "val"}),
]

relations = [
    Relation(
        label="HAS",
        source_id=entities[0].id,
        target_id=entities[1].id,
        properties={},
    )
]

graph_store.upsert_nodes(entities)
graph_store.upsert_relations(relations)

# optionally, we can also insert text chunks
source_chunk = TextNode(id_="source", text="My llama has an index.")

# create relation for each of our entities
source_relations = [
    Relation(
        label="HAS_SOURCE",
        source_id=entities[0].id,
        target_id="source",
    ),
    Relation(
        label="HAS_SOURCE",
        source_id=entities[1].id,
        target_id="source",
    ),
]
graph_store.upsert_llama_nodes([source_chunk])
graph_store.upsert_relations(source_relations)

图存储的其他有用方法包括： - graph_store.get(ids=[]) - 根据 id 获取节点 - graph_store.get(properties={"key": "val"}) - 根据匹配的属性获取节点 - graph_store.get_rel_map([entity_node], depth=2) - 获取到特定深度的三元组 - graph_store.get_llama_nodes(['id1']) - 获取原始文本节点 - graph_store.delete(ids=['id1']) - 根据 id 删除 - graph_store.delete(properties={"key": "val"}) - 根据属性删除 - graph_store.structured_query("<cypher query>") - 运行 Cypher 查询（假设图存储支持）

此外，所有这些方法都有对应的异步版本（例如 aget, adelete 等）。

高级定制

与 LlamaIndex 中的所有组件一样，您可以对模块进行子类化并自定义功能，使其完全按照您的需求工作，或者尝试新的想法并研究新的模块！

子类化提取器

LlamaIndex 中的图提取器是 TransformComponent 类的子类。如果您之前使用过摄取管道，这将是熟悉的，因为它是相同的类。

提取器的要求是将图数据插入节点的元数据中，然后稍后由索引进行处理。

以下是一个子类化的小例子，用于创建自定义提取器：

from llama_index.core.graph_store.types import (
    EntityNode,
    Relation,
    KG_NODES_KEY,
    KG_RELATIONS_KEY,
)
from llama_index.core.schema import BaseNode, TransformComponent


class MyGraphExtractor(TransformComponent):
    # init 是可选的
    # def __init__(self, ...):
    #     ...

    def __call__(
        self, llama_nodes: list[BaseNode], **kwargs
    ) -> list[BaseNode]:
        for llama_node in llama_nodes:
            # 确保不覆盖现有的实体/关系

            existing_nodes = llama_node.metadata.pop(KG_NODES_KEY, [])
            existing_relations = llama_node.metadata.pop(KG_RELATIONS_KEY, [])

            existing_nodes.append(
                EntityNode(
                    name="llama", label="ANIMAL", properties={"key": "val"}
                )
            )
            existing_nodes.append(
                EntityNode(
                    name="index", label="THING", properties={"key": "val"}
                )
            )

            existing_relations.append(
                Relation(
                    label="HAS",
                    source_id="llama",
                    target_id="index",
                    properties={},
                )
            )

            # 添加回元数据

            llama_node.metadata[KG_NODES_KEY] = existing_nodes
            llama_node.metadata[KG_RELATIONS_KEY] = existing_relations

        return llama_nodes

    # 可选的异步方法
    # async def acall(self, llama_nodes: list[BaseNode], **kwargs) -> list[BaseNode]:
    #    ...

子类化检索器

检索器比提取器更复杂，有自己的特殊类来帮助简化子类化。

检索的返回类型非常灵活。它可以是 - 一个字符串 - 一个 TextNode - 一个 NodeWithScore - 上述任何一种的列表

以下是一个小例子，用于创建自定义检索器：

from llama_index.core.indices.property_graph import (
    CustomPGRetriever,
    CUSTOM_RETRIEVE_TYPE,
)


class MyCustomRetriever(CustomPGRetriever):
    def init(self, my_option_1: bool = False, **kwargs) -> None:
        """使用从类构造函数传入的任何 kwargs。"""
        self.my_option_1 = my_option_1
        # 可选择地对 self.graph_store 进行一些操作

    def custom_retrieve(self, query_str: str) -> CUSTOM_RETRIEVE_TYPE:
        # 使用 self.graph_store 进行一些操作
        return "result"

    # 可选的异步方法
    # async def acustom_retrieve(self, query_str: str) -> str:
    #     ...


custom_retriever = MyCustomRetriever(graph_store, my_option_1=True)

retriever = index.as_retriever(sub_retrievers=[custom_retriever])

对于更复杂的定制和用例，建议查看源代码并直接对 BasePGRetriever 进行子类化。

示例

下面，您可以找到一些展示 PropertyGraphIndex 的示例笔记本

• 基本用法
• 使用 Neo4j
• 使用 Neo4j 和本地模型的高级用法
• 使用图存储
• 创建自定义图检索器