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

在 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 与摄取流水线是兼容的）！

抽取器通过相应的关键字参数设置：

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`

按照一份严格 schema 抽取路径：规定允许的实体、关系，以及哪些实体可以经由哪些关系相连。

借助 pydantic、LLM 的结构化输出，以及若干巧妙的校验，我们可以动态指定 schema 并逐路径校验抽取结果。

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,
)

这个抽取器极其可定制，以下方面都有相应选项

schema 的各个环节（如上文所示）
extract_prompt
strict=False 与 strict=True，用于决定是否允许 schema 之外的三元组
raise_on_error=False 与 raise_on_error=True，用于控制 LLM API 报错是被吞掉并记录（默认），还是让流水线直接中断。
传入你自己的 kg_schema_cls——如果你是 pydantic 高手，想自建一个带自定义校验的 pydantic 类。

检索与查询

带标签的属性图可以用多种方式查询以取回节点与路径。而在 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 根据属性图的 schema 生成 cypher
CypherTemplateRetriever —— 使用一个 cypher 模板，参数由 LLM 推断
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 使用图存储的 schema、你的查询，以及一个 text-to-cypher 的提示词模板，来生成并执行 cypher 查询。

注意：SimplePropertyGraphStore 并不是真正的图数据库，因此不支持 cypher 查询。

你可以用 index.property_graph_store.get_schema_str() 查看 schema。

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
)

存储

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

	In-Memory	Native Embedding Support	Async	Server or disk based?
SimplePropertyGraphStore	✅	❌	❌	Disk
Neo4jPropertyGraphStore	❌	✅	❌	Server
NebulaPropertyGraphStore	❌	❌	❌	Server
TiDBPropertyGraphStore	❌	✅	❌	Server
FalkorDBPropertyGraphStore	❌	✅	❌	Server

保存到/加载自磁盘

默认的图存储 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：

索引

数据加载

查询

存储

成本分析

应用

聊天机器人

问答

结构化数据

智能体

Chat Engines

Query Engine

连接器

文档与节点

摄取流水线

节点解析器

大语言模型

提示词

回调

节点后处理器

响应合成器

检索器

路由

结构化输出

现有数据

Presets and Modes

能力

起步

能力

Examples

Azure Deployment

Databases and Queues

LLM Integrations

用量

构建

（默认）SimpleLLMPathExtractor

（默认）ImplicitPathExtractor

DynamicLLMPathExtractor

SchemaLLMPathExtractor