使用属性图索引

属性图是由带标签的节点（即实体类别、文本标签等）及其属性（即元数据）组成的知识集合，通过关系相互连接形成结构化路径。

在LlamaIndex中，PropertyGraphIndex提供了以下核心功能： - 构建图结构 - 查询图数据

基本用法

通过简单导入类即可使用基础功能：

from llama_index.core import PropertyGraphIndex

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

# 使用
retriever = index.as_retriever(
    include_text=True,  # 包含匹配路径的源文本块
    similarity_top_k=2,  # 向量知识图谱节点检索的top k值
)
nodes = retriever.retrieve("Test")

query_engine = index.as_query_engine(
    include_text=True,  # 包含匹配路径的源文本块
    similarity_top_k=2,  # 向量知识图谱节点检索的top 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提取短语句，解析单跳路径格式(实体1, 关系, 实体2)

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

# 推荐使用大写、下划线分隔
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_triplets_per_chunk=10,
)

该提取器高度可定制，可调整： - 模式各方面（如上所示） - extract_prompt - strict=False与strict=True，决定是否允许模式外的三元组 - 如熟悉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

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("^")

    # 大写以标准化输入
    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 = (
    "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,
    # 自定义LLM，默认为Settings.llm
    llm=llm,
    # 自定义文本转Cypher模板
    # 需要`schema`和`question`模板参数
    text_to_cypher_template=DEFAULT_TEXT_TO_CYPHER_TEMPLATE,
    # 自定义Cypher结果如何插入文本节点
    # 需要`query`和`response`模板参数
    response_template=DEFAULT_RESPONSE_TEMPLATE,
    # 可选的可清理/验证生成Cypher的回调
    cypher_validator=None,
    # 结果中允许的字段
    allowed_output_field=DEFAULT_ALLOWED_FIELDS,
)

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

CypherTemplateRetriever

这是TextToCypherRetriever的约束版本。不同于让LLM自由生成任何Cypher语句，我们可以提供Cypher模板并让LLM填空。

以下示例说明其工作原理：

# 注意：当前需要v1版本
from pydantic 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类表示查询参数
# 类字段直接用作运行Cypher查询的参数
class TemplateParams(BaseModel):
    """Cypher查询的模板参数"""

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


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

# 创建
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 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",
)

# 创建索引
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)

# 可选操作：我们也可以插入文本块
source_chunk = TextNode(id_="source", text="My llama has an index.")

# 为每个实体创建关联关系
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):
    # 初始化可选
    # 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
• 使用 Nebula
• Neo4j 与本地模型的高级用法
• 使用属性图存储
• 创建自定义图检索器
• 知识图谱提取器对比