Property Graph 인덱스 사용하기

Property Graph 인덱스 사용하기

프로퍼티 그래프는 레이블이 지정된 노드(즉, 엔티티 카테고리, 텍스트 레이블 등)를 프로퍼티(즉, 메타데이터)와 함께 관계로 연결해 구조화된 경로를 만든 지식 컬렉션이에요. LlamaIndex에서 PropertyGraphIndex는 그래프 구축과 그래프 쿼리에 관한 핵심 오케스트레이션을 제공해요.

출처: 문서

본문

사용법

간단한 사용법은 클래스를 import 해서 사용하는 거예요:

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

구축 (Construction)

LlamaIndex의 프로퍼티 그래프 구축은 각 청크에 대해 일련의 kg_extractors를 수행하고 엔티티와 관계를 각 llama-index 노드의 메타데이터로 첨부하는 방식으로 동작해요. 여기서 원하는 만큼 많이 사용할 수 있고 모두 적용돼요.

ingestion pipeline과 함께 transformations이나 metadata extractors를 사용해 본 적이 있다면 익숙할 거예요(그리고 이 kg_extractors는 ingestion pipeline과 호환돼요)!

추출기는 적절한 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 속성을 사용해 경로를 추출해요.

이 추출기는 llama-index 노드 객체에 이미 존재하는 프로퍼티를 파싱할 뿐이므로 실행에 LLM이나 임베딩 모델이 필요 없어요.

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
• LLM API 오류를 삼키고 로깅할지(기본), 아니면 파이프라인을 중단시킬지 제어하는 raise_on_error=False vs. raise_on_error=True
• pydantic 전문가라면 커스텀 검증이 포함된 자체 pydantic 클래스를 만들어 전달할 수 있는 커스텀 kg_schema_cls

검색과 쿼리 (Retrieval and Querying)

레이블이 지정된 프로퍼티 그래프는 노드와 경로를 검색하기 위해 여러 방식으로 쿼리될 수 있어요. 그리고 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이 추론한 params와 함께 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는 그래프 스토어 스키마, 쿼리, 그리고 text-to-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
)

저장 (Storage)

현재 프로퍼티 그래프를 위해 지원되는 그래프 스토어는 다음과 같아요:

	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)

통합과 함께 저장하고 로딩

통합은 보통 자동으로 저장해요. 일부 그래프 스토어는 벡터를 지원하고, 일부는 그렇지 않을 수 있어요. 그래프 스토어를 외부 벡터 DB와 결합할 수도 있어요.

이 예시는 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,
)

Property Graph 스토어 직접 사용하기

프로퍼티 그래프의 기본 저장 클래스는 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 쿼리 실행(그래프 스토어가 지원하는 경우)

또한 이 모든 것의 async 지원을 위한 a 버전(aget, adelete 등)도 존재해요.

고급 커스터마이징

LlamaIndex의 모든 컴포넌트와 마찬가지로, 모듈을 서브클래싱하고 정확히 필요한 대로 커스터마이즈하거나 새 아이디어를 실험하고 새 모듈을 연구할 수 있어요!

추출기 서브클래싱

LlamaIndex의 그래프 추출기는 TransformComponent 클래스를 서브클래싱해요. ingestion pipeline을 다뤄봤다면 같은 클래스라서 익숙할 거예요.

추출기의 요구사항은 그래프 데이터를 노드의 메타데이터에 삽입하는 것인데, 이후 인덱스가 이를 처리해요.

커스텀 추출기를 만드는 서브클래싱 예시:

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를 보여주는 예시 노트북을 찾을 수 있어요.

• Basic Usage
• Neo4j 사용하기
• Nebula 사용하기
• Neo4j와 로컬 모델을 쓴 고급 사용법
• Property Graph Store 사용하기
• 커스텀 그래프 검색기 만들기
• KG 추출기 비교

더 알아보기 (Learn more)

• Property Graph Index 가이드