使用属性图索引

属性图是一种知识集合，由带有属性（即元数据）的标签节点（例如，实体类别、文本标签等）构成，这些节点通过关系链接形成结构化路径。

在 LlamaIndex 中，PropertyGraphIndex 主要负责以下编排：

• 构建图
• 查询图

用法

基本用法只需导入类并使用即可。

from llama_index.core import PropertyGraphIndex

# create
index = PropertyGraphIndex.from_documents(
    documents,
)

# use
retriever = index.as_retriever(
    include_text=True,  # include source chunk with matching paths
    similarity_top_k=2,  # top k for vector kg node retrieval
)
nodes = retriever.retrieve("Test")

query_engine = index.as_query_engine(
    include_text=True,  # include source chunk with matching paths
    similarity_top_k=2,  # top k for vector kg node retrieval
)
response = query_engine.query("Test")

# save and load
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")
)

# loading from existing graph store (and optional vector store)
# load from existing graph/vector store
index = PropertyGraphIndex.from_existing(
    property_graph_store=graph_store, vector_store=vector_store, ...
)

构建

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

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

提取器通过相应的 kwarg 设置

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

# insert additional documents / nodes
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 = (
    "Some text is provided below. Given the text, extract up to "
    "{max_paths_per_chunk} "
    "knowledge triples in the form of `subject,predicate,object` on each line. Avoid stopwords.\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 属性提取路径。

这个提取器不需要 LLM 或嵌入模型来运行，因为它只是解析 llama-index 节点对象上已经存在的属性。

from llama_index.core.indices.property_graph import ImplicitPathExtractor

kg_extractor = ImplicitPathExtractor()

DynamicLLMPathExtractor

将根据允许的实体类型和关系类型的可选列表提取路径（包括实体类型！）。如果未提供任何列表，LLM 将根据其判断分配类型。如果提供了列表，将有助于引导 LLM，但不会强制精确使用这些类型。

from llama_index.core.indices.property_graph import DynamicLLMPathExtractor

kg_extractor = DynamicLLMPathExtractor(
    llm=llm,
    max_triplets_per_chunk=20,
    num_workers=4,
    allowed_entity_types=["POLITICIAN", "POLITICAL_PARTY"],
    allowed_relation_types=["PRESIDENT_OF", "MEMBER_OF"],
)

SchemaLLMPathExtractor

提取遵循严格模式的路径，该模式规定了允许的实体、关系以及哪些实体可以连接到哪些关系。

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

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

# recommended uppercase, underscore separated
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,  # if false, will allow triplets outside of the schema
    num_workers=4,
    max_triplets_per_chunk=10,
)

这个提取器高度可定制，并提供了多种选项来定制 - 模式的各个方面（如上所示） - extract_prompt - strict=False vs. strict=True，以决定是否允许模式外的三元组 - 如果您是 pydantic 专家并希望创建带有自定义验证的 pydantic 类，可以传入您自己的自定义 kg_schema_cls。

检索与查询

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

# create a retriever
retriever = index.as_retriever(sub_retrievers=[retriever1, retriever2, ...])

# create a query engine
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

prompt = (
    "Given some initial query, generate synonyms or related keywords up to {max_keywords} in total, "
    "considering possible cases of capitalization, pluralization, common expressions, etc.\n"
    "Provide all synonyms/keywords separated by '^' symbols: 'keyword1^keyword2^...'\n"
    "Note, result should be in one-line, separated by '^' symbols."
    "----\n"
    "QUERY: {query_str}\n"
    "----\n"
    "KEYWORDS: "
)


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

    # capitalize to normalize with ingestion
    return [x.strip().capitalize() for x in matches if x.strip()]


synonym_retriever = LLMSynonymRetriever(
    index.property_graph_store,
    llm=llm,
    # include source chunk text with retrieved paths
    include_text=False,
    synonym_prompt=prompt,
    output_parsing_fn=parse_fn,
    max_keywords=10,
    # the depth of relations to follow after node retrieval
    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,
    # only needed when the graph store doesn't support vector queries
    # vector_store=index.vector_store,
    embed_model=embed_model,
    # include source chunk text with retrieved paths
    include_text=False,
    # the number of nodes to fetch
    similarity_top_k=2,
    # the depth of relations to follow after node retrieval
    path_depth=1,
    # can provide any other kwargs for the VectorStoreQuery class
    ...,
)

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 = (
    "Generated Cypher query:\n{query}\n\n" "Cypher Response:\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,
    # customize the LLM, defaults to Settings.llm
    llm=llm,
    # customize the text-to-cypher template.
    # Requires `schema` and `question` template args
    text_to_cypher_template=DEFAULT_TEXT_TO_CYPHER_TEMPLATE,
    # customize how the cypher result is inserted into
    # a text node. Requires `query` and `response` template args
    response_template=DEFAULT_RESPONSE_TEMPLATE,
    # an optional callable that can clean/verify generated cypher
    cypher_validator=None,
    # allowed fields in the resulting
    allowed_output_field=DEFAULT_ALLOWED_FIELDS,
)

注意： 执行任意 Cypher 查询存在风险。请确保采取必要措施（只读角色、沙盒环境等）以确保在生产环境中的安全使用。

CypherTemplateRetriever

这是 TextToCypherRetriever 的一个更受限制的版本。我们不再让 LLM 自由生成任何 Cypher 语句，而是提供一个 Cypher 模板，让 LLM 填充空白。

为了说明其工作原理，这里有一个小示例。

# NOTE: current v1 is needed
from pydantic import BaseModel, Field
from llama_index.core.indices.property_graph import CypherTemplateRetriever

# write a query with template params
cypher_query = """
MATCH (c:Chunk)-[:MENTIONS]->(o)
WHERE o.name IN $names
RETURN c.text, o.name, o.label;
"""


# create a pydantic class to represent the params for our query
# the class fields are directly used as params for running the cypher query
class TemplateParams(BaseModel):
    """Template params for a cypher query."""

    names: list[str] = Field(
        description="A list of entity names or keywords to use for lookup in a knowledge graph."
    )


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

存储

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

	内存中	原生嵌入支持	异步	基于服务器还是磁盘？
SimplePropertyGraphStore	✅	❌	❌	磁盘
Neo4jPropertyGraphStore	❌	✅	❌	服务器
NebulaPropertyGraphStore	❌	❌	❌	服务器
TiDBPropertyGraphStore	❌	✅	❌	服务器
FalkorDBPropertyGraphStore	❌	✅	❌	服务器

保存到/从磁盘加载

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

这里有一个使用默认图存储保存/加载索引的示例。

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

# create
index = PropertyGraphIndex.from_documents(documents)

# save
index.storage_context.persist("./storage")

# load
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 Neo4jPropertyGraphStore
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 = Neo4jPropertyGraphStore(
    username="neo4j",
    password="<password>",
    url="bolt://localhost:7687",
)

# creates an index
index = PropertyGraphIndex.from_documents(
    documents,
    property_graph_store=graph_store,
    # optional, neo4j also supports vectors directly
    vector_store=vector_store,
    embed_kg_nodes=True,
)

# load from existing graph/vector store
index = PropertyGraphIndex.from_existing(
    property_graph_store=graph_store,
    # optional, neo4j also supports vectors directly
    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 查询（假定图存储支持）

此外，所有这些方法都存在以 a 开头的异步版本（即 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):
    # the init is optional
    # def __init__(self, ...):
    #     ...

    def __call__(
        self, llama_nodes: list[BaseNode], **kwargs
    ) -> list[BaseNode]:
        for llama_node in llama_nodes:
            # be sure to not overwrite existing entities/relations

            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={},
                )
            )

            # add back to the metadata

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

        return llama_nodes

    # optional async method
    # 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:
        """Uses any kwargs passed in from class constructor."""
        self.my_option_1 = my_option_1
        # optionally do something with self.graph_store

    def custom_retrieve(self, query_str: str) -> CUSTOM_RETRIEVE_TYPE:
        # some some operation with self.graph_store
        return "result"

    # optional async method
    # 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 的示例 Notebook。

• 基本用法
• 使用 Neo4j
• 使用 Nebula
• 使用 Neo4j 和本地模型的高级用法
• 使用属性图存储
• 创建自定义图检索器
• 比较 KG 提取器