Neo4j와 Unstructured.io로 기술 문서 기반 GenAI 챗봇 구축하기

기술적인 사용자를 위한 GenAI 애플리케이션을 구축할 때 흔히 겪는 어려움은 차트, 다이어그램, 표와 같이 관련된 비정형 정보까지 포함해서 정확한 답변을 제공해야 한다는 점이에요.

이번 블로그에서는 간단한 GenAI 챗봇을 만들어 볼 건데요. 에너지 산업의 오픈 소스 기술 문서를 활용할 거예요. Unstructured.io를 사용해서 고해상도 청킹을 수행하고, 테이블과 이미지를 추출한 다음, Neo4j로 Knowledge Graph를 구축해서 문서 컨텍스트에서 청크 순서를 유지할 거예요. 그리고 Neo4j의 Needle Starter Kit로 챗봇을 배포할 건데요. 이 키트는 Python용 Neo4j GraphRAG 패키지를 사용해서 검사 가능한 검색 및 생성 작업을 수행하죠(neo4j-graphrag). 그럼, 다음 방법들을 단계별로 살펴볼까요?

청크에 Vector Index를 구축하고, 엔터티 용어에 전체 텍스트 Index를 구축해요.
문서 텍스트, 이미지, 표를 구문 분석하고 Knowledge Graph로 수집해요.
문서 청크에 Vector Embedding을 수행해요.
문서 청크에서 엔터티 추출을 수행해요.
이미지에 메타데이터를 추가해서 불필요한 이미지를 걸러내요.
neo4j-graphrag HybridCypherRetriever를 사용해서 관련 청크, 이미지, 테이블에 대한 최근접 순회를 통해 강화된 동시 Semantic Search 및 전체 텍스트 청크 검색을 수행하는 챗봇 애플리케이션을 배포해요.

하지만 먼저 완성된 결과가 어떤 모습인지, 왜 이런 방식으로 접근하는지부터 알아볼게요.

문서를 이해하는 챗봇

먼저 Knowledge Graph 스키마부터 살펴볼게요. :Document Node는 :Chunk Node를 가지고 있고, :Chunk Node는 문서 순서대로 순서가 지정되어 있어요. :NEXT_CHUNK Relationship을 통해서요. :Chunk Node는 Neo4j Vector Index가 지원하는 벡터에 대한 임베딩 속성을 가지고 있죠. 텍스트 청크의 이미지와 테이블은 :Image와 :Table Node로 저장되고, 컨텍스트는 :RELATED_CONTENT Relationship으로 유지돼요. 추출된 엔터티는 :Entity Node(Neo4j Lucene 전체 텍스트 Index로 지원)로 인스턴스화되고, 청크(문서 내 및 문서 간)에 매핑되죠. :HAS_ENTITY Relationship을 통해서요. 이렇게 문서를 다양한 요소로 분해했기 때문에 어휘 그래프라고 부르는 거예요.

다음 Query를 사용해서 Neo4j에서 문서가 어떻게 표현되는지 살펴볼 수 있어요.

MATCH path0 = (n:Document {name: "StatoilHydro_Volve_15_9_F_14_Report.pdf"})<-[:PART_OF_DOCUMENT]-(c0:Chunk)
WHERE NOT (c0)<-[:NEXT_CHUNK]-()
MATCH (n)<-[:PART_OF_DOCUMENT]-(c:Chunk)
WITH path0, COLLECT(c) AS chunks
UNWIND chunks AS chunk
CALL(chunk) {
    MATCH path1 = (chunk)-[:NEXT_CHUNK|RELATED_CONTENT]-()
    RETURN path1
}
RETURN path0, path1

그래프는 문서의 기본적인 구조, 즉 일련의 문서 구조를 유지해요. :Chunk 노드와 함께 나타나는 순서대로 연결된 :Image and :Table 노드들이죠. 이 특정 문서에 대해선 47개의 서술형 텍스트 덩어리, 32개의 이미지, 그리고 11개의 테이블을 구문 분석했어요.

확대해서 보면 어떻게 보이는지 더 잘 알 수 있을 거예요. :Document 노드는 왼쪽 하단에 있고, 순서대로 정렬된 :Chunk 노드와 관련된 :Image and :Table 노드들을 확인할 수 있죠.

:Document

:Chunk

:Image

, 그리고

:Table

노드를 보여주는 어휘 그래프의 클로즈업

Neo4j GraphRAG를 통한 정확성과 설명 가능성

그렇다면 이것이 어떻게 GenAI 정확도를 향상시키는 데 도움이 될까요? 이 어휘 그래프 구조를 사용하면 네 가지 검색 방법을 통해 텍스트(및 관련 콘텐츠)를 검색할 수 있어요.

Semantic Search: :Chunk 텍스트 임베딩을 사용한 벡터 인덱싱
추출된 엔티티의 전체 텍스트 Index 검색 및 해당 소스 텍스트 탐색
가장 가까운 이웃 텍스트로 순회: 방법 1과 2로 발견된 :Chunk 노드
관련 이미지 및 테이블 순회: 방법 1과 2로 발견된 :Chunk 노드

방법 1~3의 결과는 LLM 요약을 위한 풍부하고 정확하며 구조화된 컨텍스트를 제공하기 위해 패키지로 제공돼요. 방법 4(이미지 및 표)의 결과는 생성 후 챗봇에 표시될 거예요 (아래 참조).

이 모든 것이 복잡해 보이지만, 다행히도 어려운 작업은 하이브리드사이퍼리트리버 neo4j-graphrag 패키지의 메소드가 대신 해준답니다:

#this query pulls the adjacent Chunks, Entities
RETRIEVAL_QUERY = (
"""
WITH node, score
OPTIONAL MATCH (node)-[:NEXT_CHUNK]-(c) // get chunk neighbors
OPTIONAL MATCH (node)<-[HAS_ENTITY]-(e) // get entity context chunks
ORDER BY score DESC LIMIT 100
RETURN apoc.convert.toSet(COLLECT(elementId(node))+COLLECT(elementId(e))+COLLECT(elementId(c))) AS listIds,
COLLECT (e.id) as contextNodes, node.text as nodeText, score ORDER BY score DESC
"""
)

def __init__(self, driver, embedder, vector_index_name, fulltext_index_name):
        self._retriever = HybridCypherRetriever(
            driver,
            vector_index_name=vector_index_name,
            fulltext_index_name=fulltext_index_name,
            retrieval_query=self.RETRIEVAL_QUERY,
            result_formatter=self.formatter,
            embedder=embedder,
            neo4j_database="neo4j",
        )

검색자에게 사용할 벡터 및 전체 텍스트 인덱스의 이름을 알려주고, 사용자 프롬프트에 대한 임베더를 전달하고, 증대를 지정합니다. Cypher 쿼리를 사용해서 가장 가까운 이웃을 탐색하는데, :Chunk 노드를 검색하고 사용할 데이터베이스와 함께 검색된 노드를 찾죠.

예제 프롬프트

실제로 한번 살펴볼까요? 몇 가지 예시 질문이 저장소에 있는데, 다음 중 하나를 시도해 볼 수 있어요.

What is meant by a sinusoid in the EcoScope analyses?

정말 강력하고 구체적인 답변을 얻었죠? (전직 지구과학자로서 이 문서를 자세히 읽어본 결과, 이건 좋고 정확한 결과라고 말할 수 있어요!) 우리가 정말로 확신하는지 한번 살펴볼까요? 석유 엔지니어인 척하면서 그래프 아이콘을 클릭하면 응답 준비에 사용된 내용을 검사할 수 있어요.

전적으로 어휘 Knowledge Graph를 기반으로 한 사용자 프롬프트에 대한 자세한 응답

이제 HybridCypherRetriever의 내부 작동을 볼 수 있어요. 우리는 탐색할 어휘 하위 그래프를 가지고 있어요. :Chunk and :Entity 의미 검색과 전체 텍스트 검색을 통해 각각 발견된 노드 문서 시퀀스 내에서 로컬 순회를 통해 발견된 가장 가까운 이웃 청크, 그리고 관련된 모든 :Table and :Image 노드까지요.

GraphRAG retriever에 의해 발견되고 LLM에 의해 요약된 Node 검사

시각화에서 이러한 Node 중 하나를 클릭하면 해당 콘텐츠를 볼 수 있어요.

neo4j-graphrag HybridCypherRetriever에 의해 발견된 예제 텍스트 Chunk

그리고 내가 :Image Node를 클릭하면 EcoScope 분석이 어떻게 작동하는지 보여주는 다이어그램을 볼 수 있는데, 이는 원래 질문에 대한 가장 좋고 직관적인 대답일 수 있습니다.

그래서 무엇? GenAI 퀵 테이크

이 블로그에서는 Neo4j Knowledge Graph 기술이 기술 청중을 위해 상황에 맞는 정확하고 검사 가능한 GenAI 응답(이미지 및 표 포함)을 제공할 수 있는 방법을 보여줘요. LLM이나 벡터 데이터베이스만 사용해서는 이러한 수준의 특이성과 신뢰도를 달성하기는 어렵죠.

구조화되지 않은.io는 추가 설명 컨텍스트를 위한 관련 이미지 및 테이블과 함께 문서 컨텍스트에서 설명 텍스트를 지능적으로 추출할 수 있어요. 네오4j는 비정형 데이터의 설명 정보와 정형 데이터의 비즈니스 사실 및 계층 구조를 결합한 Knowledge Graph를 쉽게 구축할 수 있게 해주죠.

Neo4j Knowledge Graph를 GenAI 애플리케이션의 지속성 계층으로 활용하면 최종 사용자에게 성능이 뛰어나고 신뢰할 수 있는 경험을 제공할 수 있어요. 에이전트 프레임워크 and MCP는 더욱 정교한 지식 발견 방법을 위해 필요에 따라 통합됩니다.

여기에 제시된 개념은 저장소 코드를 약간만 수정하면 모든 기술 문서 모음으로 확장할 수 있어요. 다음 섹션에서는 이러한 접근 방식을 활용해서 여러분의 문서에서 Neo4j Knowledge Graph와 챗봇을 구축하는 방법을 단계별로 설명해 드릴게요.

Unstructured.io를 사용한 문서 구문 분석

가장 먼저 기술 문서를 구문 분석해야겠죠? 이를 위해 우리는 훌륭한 도구인 Unstructured.io를 사용할 거예요. Neo4j의 새로운 파트너랍니다.

Unstructured.io는 문서에서 다양한 콘텐츠 요소를 추출하는 정교한 도구를 제공하고, 문서 레이아웃을 검사해서 추가 컨텍스트를 추론할 수 있게 해줘요. Unstructured.io는 최근 병렬화된 Unstructured.io 문서 처리 파이프라인의 일부로 텍스트 청크의 어휘 그래프를 구축할 수 있는 Neo4j 커넥터를 출시했어요. Unstructured.io 웹 콘솔이나 API를 통해 서비스를 이용할 수 있답니다.

예제 구문 분석 파이프라인은 Unstructured.io Neo4j 그래프 대상을 사용하는 콘솔

Partitioner는 Unstructured.io 플랫폼의 핵심이에요. 병렬로 작동할 수 있어서 문서를 빠르게 분할한 다음 제목, 머리글, 바닥글, 페이지 번호, 설명 텍스트, 목록, 수식 등과 같은 요소를 식별하고 태그를 지정할 수 있어요. 나중에 살펴보겠지만 이미지와 표도 추출할 수 있답니다.

        partitioner_config=PartitionerConfig(
            partition_by_api=True,
            api_key=(UNSTRUCTURED_API_KEY),
            partition_endpoint=UNSTRUCTURED_API_URL,
            additional_partition_args={
                "split_pdf_page": True,
                "split_pdf_allow_failed": True,
                "split_pdf_concurrency_level": 15
            }

Chunker는 Large Language Model (LLM)에 표시하기 위해 Partitioner에서 생성된 요소를 더 큰 텍스트 블록으로 모으는 구성 가능한 상황별 인텔리전스 계층이에요. by_title 옵션을 사용하면 문서 들여쓰기와 공백을 살펴서 내러티브의 자연스러운 경계를 결정하죠.

        chunker_config=ChunkerConfig(
            chunking_strategy="by_title"
            max_characters=1500
            ),

Embedder는 Semantic Search를 지원하기 위해 청크에 대한 Vector Embedding을 생성하고, Entity Extractor는 청크에서 엔터티를 구문 분석하고 LLM에 프롬프트를 전달할 수 있게 해줘요.

        embedder_config=EmbedderConfig(
            embedding_provider="openai",
            embedding_api_key=OPENAI_API_KEY
        ),

Neo4j 대상 커넥터는 몇 가지 제약 조건과 기본 벡터 Index를 구축한 다음 파이프라인의 출력을 Neo4j에 로드해서 처리된 문서를 나타내는 Node에 시퀀스된 청크 Node를 연결하는 어휘 그래프를 구축해요. 엔터티 Node는 추가 탐색 가능한 컨텍스트를 위해 청크 Node를 연결한답니다.

        destination_connection_config=Neo4jConnectionConfig(
            access_config=Neo4jAccessConfig(password=NEO4J_PASSWORD),
            username=NEO4J_USERNAME,
            uri=NEO4J_URI,
            database=NEO4J_DATABASE,
        ),
        stager_config=Neo4jUploadStagerConfig(),
        uploader_config=Neo4jUploaderConfig(batch_size=100)

우리 애플리케이션에서는 Unstructured.io 구문 분석 API를 사용해서 모든 구문 분석 및 청크 작업을 처리하고 있어요. GitHub 저장소에 모든 노트북, 애플리케이션 코드, 예제 파일 링크가 있으니 참고해주세요. 무료 Neo4j Aura Graph Database를 프로비저닝하고 Unstructured.io API 키를 얻는 방법은 저장소 지침을 따라하면 돼요.

다음 섹션에서는 그래프를 단계별로 만들어볼게요.

어휘 Knowledge Graph 파이프라인

어휘 Knowledge Graph를 구축하는 데 사용된 전체 파이프라인은 다음과 같아요.

1단계. 문서 청크에 대한 Vector Index 및 엔터티 용어에 대한 전체 텍스트 Index 작성

먼저 neo4j-graphrag 패키지를 사용해서 Neo4j에서 기본적인 Semantic Search 및 전체 텍스트 검색에 필요한 몇 가지 구성을 하고, 필요한 Index를 설정할 거예요.

종속성 설치

!pip install unstructured-client
!pip install neo4j
!pip install neo4j-graphrag

변수 구성

unstructured_api_key="UNSTRUCTURED_API_KEY"
neo4j_uri = "NEO4J_URI"
neo4j_database = "neo4j" 
neo4j_user = "neo4j" 
neo4j_password = "NEO4J_PASSWORD"
openai_api_key ="OPENAI_API_KEY"

Neo4j에서 벡터 및 전체 텍스트 Index 설정

from neo4j import GraphDatabase
from neo4j_graphrag.indexes import create_vector_index
from neo4j_graphrag.indexes import create_fulltext_index

# Neo4j driver setup
driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))

VECTOR_INDEX_NAME = "chunk_embedding"

# Creating the vector index
create_vector_index(
    driver,
    VECTOR_INDEX_NAME,
    label="Chunk",
    embedding_property="embedding",
    dimensions=1536,
    similarity_fn="cosine",
    fail_if_exists=False,
)

FULLTEXT_INDEX_NAME = "entity_text"

# Creating the full text index
create_fulltext_index(
    driver,
    FULLTEXT_INDEX_NAME,
    label="Entity",
    node_properties= ["text", "variants"],
    fail_if_exists=False,
)

query = '''
SHOW INDEXES
'''

with driver.session() as session:
            result = session.run(query)
            for record in result:
                print(record)

2단계. 문서 텍스트, 이미지, 표를 구문 분석해서 Knowledge Graph로 수집하기

다음으로는 문서 세트를 구문 분석하고 Knowledge Graph를 만들어볼게요. :Document, :Chunk, :Image, 그리고 :Table에 대한 Node를 생성할 거예요. apoc.nodes.link() 프로시저를 사용해서 모든 :Chunk Node가 나타나는 순서대로 :Document에 연결되도록 할 거예요. 또한 :Image와 :Table Node가 :Chunk 콘텐츠의 일부인 경우 추가하고 있다는 점도 알아두세요. 이미지는 바이너리와 OCR 텍스트로 저장하고, 테이블은 이미지, HTML, 텍스트로 저장할 거예요.

Unstructured.io를 사용해서 PDF 추출 및 청크

import os
import base64
import zlib
import json
import logging
import nltk

from neo4j import GraphDatabase
from unstructured_client import UnstructuredClient
from unstructured_client.models import operations, shared
from unstructured.staging.base import elements_from_dicts, elements_to_json

# Disable logging
logging.disable(logging.CRITICAL)

# Configuration
directory_path = "PATH_TO_DOCUMENTS/"

client = UnstructuredClient(
    api_key_auth=unstructured_api_key,
    server_url="https://api.unstructuredapp.io"
)

driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))

CHUNK_QUERY = '''
WITH apoc.convert.fromJsonList($json) AS maps
UNWIND maps AS map
WITH apoc.map.clean(map,[],["  ",""]) AS m
MERGE (d:Document {name: m.metadata.filename})
WITH m, d
CALL(m, d) {
  CREATE (n:Chunk {id: m.element_id})
  SET
    n.type = "NarrativeText",
    n.text = m.text,
    n.filename = m.metadata.filename,
    n.filetype = m.metadata.filetype,
    n.languages = m.metadata.languages,
    n.page_number = m.metadata.page_number,
    n.tokens = m.tokens
  CREATE (n)-[:PART_OF_DOCUMENT]->(d)
  RETURN n
}
WITH m, d, n
CALL(m, d, n) {
  WITH m, d, n
  WHERE m.metadata.type IN ["Image", "Table"]
  CREATE (i:$(m.metadata.type) {id: m.element_id})
  SET i.type = m.metadata.type,
      i.figure_caption = m.metadata.figure_caption,
      i.text = m.metadata.text,
      i.filename = m.metadata.filename,
      i.filetype = m.metadata.filetype,
      i.languages = m.metadata.languages,
      i.page_number = m.metadata.page_number,
      i.image_base64 = m.metadata.image_base64,
      i.image_mime_type = m.metadata.image_mime_type,
      i.text_as_html = m.metadata.text_as_html
  MERGE (n)-[:RELATED_CONTENT]->(i)
  MERGE (i)-[:PART_OF_DOCUMENT]->(d)
}
WITH DISTINCT d, n
WITH d, COLLECT(n) AS nodes
CALL apoc.nodes.link(nodes, "NEXT_CHUNK")
'''

def run_query(tx, query, json_data):
    return tx.run(query, {"json": json_data}).consume()

def extract_orig_elements(encoded):
    decoded = base64.b64decode(encoded)
    decompressed = zlib.decompress(decoded)
    return json.loads(decompressed.decode("utf-8"))

def process_file(filepath, filename):
    print(f"\nProcessing file: {filename}")
    
    with open(filepath, "rb") as f:
        files = shared.Files(
            content=f.read(),
            file_name=filename
        )

    request = operations.PartitionRequest(
        partition_parameters=shared.PartitionParameters(
            files=files,
            strategy="hi_res",
            hi_res_model_name="yolox",
            element_exclude=["Header", "Footer", "ListItem", "Formula", "UncategorizedText"],
            extract_image_block_types=["Image", "Table"],
            chunking_strategy="by_title",
            max_characters=1500,
            split_pdf_page=True,
            split_pdf_allow_failed=True,
            split_pdf_concurrency_level=15
        )
    )

    response = client.general.partition(request=request)
    element_dicts = [e for e in response.elements]

    for i, element in enumerate(element_dicts):
        if element.get("text"):
            element["tokens"] = len(nltk.word_tokenize(element["text"]))

        metadata = element.get("metadata", {})
        if metadata.get("orig_elements"):
            orig_elements = extract_orig_elements(metadata["orig_elements"])

            for obj in orig_elements:
                if obj.get("type") == "FigureCaption" and obj.get("text", "").lower().startswith("figure"):
                    metadata["figure_caption"] = obj["text"]

                if obj.get("type") == "Image":
                    metadata.update({
                        "element_id": obj["element_id"],
                        "type": obj["type"],
                        "image_base64": obj["metadata"]["image_base64"],
                        "image_mime_type": obj["metadata"]["image_mime_type"],
                        "text": obj["text"]
                    })

                if obj.get("type") == "Table":
                    metadata.update({
                        "element_id": obj["element_id"],
                        "type": obj["type"],
                        "text_as_html": obj["metadata"]["text_as_html"],
                        "image_base64": obj["metadata"]["image_base64"],
                        "image_mime_type": obj["metadata"]["image_mime_type"],
                        "text": obj["text"]
                    })

        element_dicts[i]["metadata"].pop("orig_elements", None)

    json_data = json.dumps(element_dicts, indent=4)

    with driver.session() as session:
        summary = session.execute_write(run_query, CHUNK_QUERY, json_data)
        print(f"nodes created => {summary.counters.nodes_created}, rels created => {summary.counters.relationships_created}")
    session.close() 
    print(f"Finished processing: {filename}")

def main():
    for filename in os.listdir(directory_path):
        if filename.startswith(".") or not os.path.isfile(os.path.join(directory_path, filename)):
            continue

        try:
            process_file(os.path.join(directory_path, filename), filename)
        except Exception as e:
            print(f"Error processing {filename}: {e}")
    driver.close()
    print("Done!")

if __name__ == "__main__":
    main()

3단계. 문서 청크에 Vector Embedding 수행

다음으로 OpenAI를 사용해서 청크 텍스트에 Vector Embedding을 수행하고, 이를 속성으로 작성할 거예요. `embedding` 속성에 Vector Index를 이미 선언해 뒀기 때문에 유사성은 뒤에서 자동으로 계산될 거랍니다.

Neo4j GraphRAG를 사용해서 벡터 삽입

# vector embedding on Chunk text for semantic search
from neo4j import GraphDatabase
from neo4j_graphrag.embeddings import OpenAIEmbeddings
import logging
import sys

# Disable logging output
logging.disable(sys.maxsize)

# --- Configuration ---
EMBEDDING_MODEL = "text-embedding-ada-002"
MAX_CHUNK_LENGTH = 12000

# Initialize OpenAI embedder
embedder = OpenAIEmbeddings(model=EMBEDDING_MODEL, api_key=openai_api_key)

# Initialize Neo4j driver
driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))


def embed_chunks():
    with driver.session() as session:
        result = session.run("""
            MATCH (n:Chunk)
            WHERE n.text IS NOT NULL AND n.embedding IS NULL
            RETURN n.id AS id, n.text AS data
        """)

        for record in result:
            chunk_id = record["id"]
            data = record["data"]

            if len(data) > MAX_CHUNK_LENGTH:
                print(f"Skipping chunk {chunk_id} (length: {len(data)})")
                continue

            print(f"\rEmbedding chunk: {chunk_id}", end="", flush=True)

            try:
                embedding = embedder.embed_query(data)

                session.run("""
                    MATCH (n:Chunk {id: $chunk_id})
                    SET n.embedding = $embedding
                """, chunk_id=chunk_id, embedding=embedding)

            except Exception as e:
                print(f"\nFailed to embed chunk {chunk_id}: {e}")


def main():
    embed_chunks()
    driver.close()
    print("\nDone embedding chunks!")


if __name__ == "__main__":
    main()

4단계. 문서 청크에서 엔터티 추출 수행

OpenAI를 사용해서 청크 텍스트에서 엔터티 추출을 수행하고, 엔터티를 서로 연결할 거예요. 이는 계산 집약적인 작업이기 때문에, 다음을 사용하는 걸 선호해요. :ProcessMe 쿼리 속도를 높이기 위한 Label이죠.

엔터티 추출 및 결과 형식을 안내하기 위해 데이터 도메인별 Prompt를 사용하고 있다는 점도 알아두세요!

Extract all the entities from the following text. 

Identify only entities, abbreviations and technical terms commonly used in the petroleum exploration, petroleum geology, reservoir analysis, and oil & gas production.

Return entities in this format: ["entity1", "entity2"]

엔터티를 올바르게 추출하려면 자신의 데이터 도메인에 맞게 이를 사용자 정의해야 해요.

처리를 위해 청크 Node에 Label 지정

# entity extraction on Chunk text for full text search

from neo4j import GraphDatabase

# Neo4j driver setup
driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))

# to perform selective entity extraction using "ProcessMe" label execute this query

PROCESS_ME ='''
MATCH (n:Chunk {type:"NarrativeText"}) 
WHERE NOT (n)-[:HAS_ENTITY]->() AND n.entities IS NULL
SET n:ProcessMe
'''

with driver.session() as session:
    res = session.run(PROCESS_ME)
session.close() 
print("done!")

Label이 지정된 청크 Node에서 엔터티 추출

# prompted entity extraction with some very light entity resolution
import logging
import sys
from neo4j import GraphDatabase
from openai import OpenAI

# --- Config ---
MAX_CHUNKS = 1000
OPENAI_MODEL = "gpt-4o"  # or "gpt-4"

# --- Logging ---
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logging.disable(sys.maxsize)  # Disable logging if needed

# --- Clients ---
client = OpenAI(api_key=openai_api_key)
driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))

# --- Cypher Queries ---
FETCH_CHUNKS_QUERY = """
MATCH (n:Chunk:ProcessMe)
RETURN n.id AS id, replace(n.text,"\n","") AS text
LIMIT $limit
"""

ENTITY_INSERT_QUERY = """
WITH $entities AS entities
MATCH (n:Chunk:ProcessMe {id: $id})
WITH n, entities
CALL apoc.do.when(
    entities[0] = "[]" OR entities[0] STARTS WITH "The text provided does not contain",
    "WITH n SET n.entities = 'failed' REMOVE n:ProcessMe RETURN 0 AS rels",
    "WITH n, apoc.convert.fromJsonList(entities[0]) AS names
     UNWIND names AS name
     MERGE (e:Entity {text: toLower(name)})
     ON CREATE SET e.variants = [name]
     ON MATCH SET e.variants = apoc.convert.toSet(e.variants + [name])
     MERGE (n)-[:HAS_ENTITY]->(e)
     WITH DISTINCT n, COUNT(e) AS rels
     REMOVE n:ProcessMe
     RETURN rels",
    {n: n, entities: entities}
) YIELD value
RETURN value
"""

FAIL_MARK_QUERY = """
MATCH (n:Chunk:ProcessMe {id: $id})
SET n.entities = "failed"
REMOVE n:ProcessMe
"""

# --- Entity Extraction Prompt ---
# modify as needed for your data domain
def extract_entities(text: str) -> list:
    prompt = f"""
Extract all the entities from the following text. 

Identify only entities, abbreviations and technical terms commonly used in the petroleum exploration, petroleum geology, reservoir analysis, and oil & gas production.

Return entities in this format: ["entity1", "entity2"]

Do not include any extra text or explanation.

Text: {text}
"""

    messages = [
        {"role": "system", "content": "You help extract entities from petroleum-related text."},
        {"role": "user", "content": prompt}
    ]

    response = client.chat.completions.create(
        model=OPENAI_MODEL,
        messages=messages,
        max_tokens=500,
        temperature=0
    )

    return [response.choices[0].message.content.strip()]


# --- Main Function ---
def main():
    processed_count = 0

    with driver.session() as session:
        chunks = session.run(FETCH_CHUNKS_QUERY, limit=MAX_CHUNKS)

        for chunk in chunks:
            chunk_id = chunk["id"]
            text = chunk["text"]
            processed_count += 1

            try:
                entities = extract_entities(text)
                result = session.run(ENTITY_INSERT_QUERY, id=chunk_id, entities=entities)

                for record in result:
                    rels = record["value"]["rels"]
                    print(f"\rRelationships created: {rels} | Processed: {processed_count}     ", end="", flush=True)

                result.consume()

            except Exception as e:
                logging.warning(f"\nFailed to process chunk {chunk_id}: {e}")
                session.run(FAIL_MARK_QUERY, id=chunk_id)
    driver.close()
    print("\nDone extracting entities!")


if __name__ == "__main__":
    main()

이제 거의 다 왔어요! 캡처된 이미지 중 일부를 살펴보면 로고 및 기타 추출 아티팩트와 같이 흥미롭지 않은 그래픽이 많이 보일 거예요. 마지막 단계로 이미지 메타데이터를 캡처해서 검색 시 흥미롭지 않은 이미지를 필터링할 수 있어요.

이미지 크기 및 치수 계산

#filtering properties for the frontend to hide logos and other uninteresting images
import base64
from io import BytesIO
from PIL import Image
from neo4j import GraphDatabase
import json

driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))

# Cypher query to fetch nodes with base64 images
QUERY_IMAGES= """
MATCH (n:Image|Table)
WHERE n.image_base64 IS NOT NULL AND n.bytes IS NULL
RETURN n.id AS id, n.image_base64 AS image_base64
"""

def get_image_properties(image_base64: str):
    try:
        image_data = base64.b64decode(image_base64)
        with Image.open(BytesIO(image_data)) as image:
            width, height = image.size
            aspect_ratio = max(width / height, height / width) if width and height else None
            return {
                "bytes": len(image_base64),
                "width": width,
                "height": height,
                "aspect_ratio": aspect_ratio
            }
    except Exception as e:
        print(f"Error processing image: {e}")
        return None

def update_image_properties(driver):
    with driver.session() as session:
        result = session.run(QUERY_IMAGES)
        
        for record in result:
            node_id = record["id"]
            image_base64 = record["image_base64"]
            props = get_image_properties(image_base64)

            if props:
                session.run(
                    """
                    WITH apoc.convert.fromJsonMap($json) AS map
                    MATCH (n:Image|Table {id: $id})
                    SET n += map
                    """,
                    {"id": node_id, "json": json.dumps(props)}
                )


if __name__ == "__main__":
    update_image_properties(driver)
    driver.close()
    print("Image metadata updated.")

GraphRAG 챗봇

이제 그래프가 구축되었으니 챗봇을 연결해 볼까요? 다음 지침은 GitHub 저장소에서 확인할 수 있어요.

이 애플리케이션은 Neo4j 니들 스타터 키트로 구축되었는데요, Python 백엔드와 React 프런트 엔드를 포함하고 몇 가지 기본적인 그래프 시각화 기능도 제공해요. 시각화는 Neo4j 시각화 라이브러리를 사용했고요. 저희가 집중할 부분은 백엔드 구성 요소 중 retriever.py 파일이에요.

여러분은 아마 Semantic Search와 전체 텍스트 검색 인덱스가 어떻게 함께 작동하는지 궁금할 텐데요. 간단히 말하면, 하이브리드 리트리버가 그 역할을 담당하고, 이 리트리버는 neo4j-graphrag의 일부로 제공돼요. HybridCypherRetriever는 사용자의 프롬프트를 받아서 동시에 Vector Embedding 검색(텍스트 임베딩)과 전체 텍스트 검색(엔티티 용어)에 대한 입력으로 사용한 다음, 각 점수의 가중 평균을 사용하여 순위를 매긴 검색된 Node의 결합 목록을 반환하는 방식이에요. 더 자세한 내용은 GraphRAG Python 패키지를 사용하는 GraphRAG 애플리케이션에 대한 하이브리드 검색에서 확인해 보세요.

top_k에 대한 가중 평균 점수 = neo4j-graphrag HybridCypherRetriever를 통한 3개의 동시 벡터 및 전체 텍스트 검색

저희 애플리케이션에서는 추가 순회를 위해 Cypher 쿼리를 추가할 수 있는 HybridCypherRetriever를 사용하고 있어요. 여기에서 추가 컨텍스트를 위해 Knowledge Graph를 마이닝할 수 있죠. 이 경우 검색에서 발견된 청크의 앞과 뒤(문서 순서에서)에 있는 청크를 발견하는데요. 이러한 이웃 청크는 더 완전하고 정확한 GenAI 결과를 위해 유용한 설명 컨텍스트를 제공한답니다.

#this query pulls the adjacent Chunks, Entities
    RETRIEVAL_QUERY = (
            """
            WITH node, score
            OPTIONAL MATCH (node)-[:NEXT_CHUNK]-(c) // get chunk neighbors
            OPTIONAL MATCH (node)<-[HAS_ENTITY]-(e) // get entity context chunks
            ORDER BY score DESC LIMIT 100
            RETURN apoc.convert.toSet(COLLECT(elementId(node))+COLLECT(elementId(e))+COLLECT(elementId(c))) AS listIds,
            COLLECT (e.id) as contextNodes, node.text as nodeText, score ORDER BY score DESC
            """
        )

    def __init__(self, driver, embedder, vector_index_name, fulltext_index_name):
        self._retriever = HybridCypherRetriever(
            driver,
            vector_index_name=vector_index_name,
            fulltext_index_name=fulltext_index_name,
            retrieval_query=self.RETRIEVAL_QUERY,
            result_formatter=self.formatter,
            embedder=embedder,
            neo4j_database="neo4j",
        )

프론트엔드에는 구성 요소가 있는데요. 검색정보.tsx(결과 창의 작은 그래프 아이콘으로 실행됨)는 :Chunk 노드를 요약을 위해 LLM으로 전송해요. 또한 :Image 와 :Table 노드는 요약된 결과와 관련된 노드들이죠. 이들 각각은 사용자가 개별적으로 검사할 수 있답니다. 이러한 검사 가능성은 시스템에 대한 신뢰도를 높여주는데요. 최종 사용자는 최종 LLM 요약 결과에 기여한 검색된 텍스트 및 관련 콘텐츠를 볼 수 있어요.

RetrievalInformation.tsx  

function run() {
    const formattedSources = sources.map((source) => `"${source}"`).join(',');

    const query1 = `
    MATCH (a:Chunk)-[r:PART_OF_DOCUMENT]->(b:Document)
    WHERE elementId(a) in [${formattedSources}]
    RETURN DISTINCT a,r,b
    UNION
    MATCH (a:Chunk)-[r:HAS_ENTITY|NEXT_CHUNK]-(b:Chunk|Entity)
    WHERE elementId(a) in [${formattedSources}] AND elementId(b) in [${formattedSources}]
    RETURN DISTINCT a,r,b
    UNION
    MATCH (a:Chunk)-[r:RELATED_CONTENT]-(b:Image|Table)
    WHERE elementId(a) in [${formattedSources}] AND b.aspect_ratio < 10 AND b.bytes > 1024 * 9
    RETURN DISTINCT a,r,b
    LIMIT 500
    `;

저는 결과 하위 그래프를 클릭하는 것을 좋아하지만, 이러한 항목들을 텍스트 결과와 함께 인라인으로 쉽게 표시할 수도 있어요. Unstructured.io가 이미지에 대한 OCR 텍스트 결과와 테이블에 대한 HTML도 제공한다는 사실을 눈치채셨을 텐데요. 또 다른 잠재적인 개선 사항은 이러한 텍스트도 검색 가능하게 만드는 것이랍니다.

위 프롬프트 결과에 기여하는 발견된 subgraph nodes의 검사 예시예요. (이 경우 텍스트 청크 node의 가장 가까운 이웃인 이미지 node)

요약

이 GenAI 챗봇 애플리케이션에 추가할 수 있는 잠재적인 개선 사항과 통합이 정말 많아요. 앞서 말씀드린 것처럼 코드를 약간만 수정하면 모든 문서 모음에 Knowledge Graph 파이프라인을 적용할 수 있고요. Neo4j GraphRAG 프로젝트를 시작하는 방법에 대해 궁금한 점이 있다면 언제든지 편하게 문의해주세요!

에이치시스템즈의 LogTree는 Neo4j 기반 GraphRAG 플랫폼으로, 데이터를 자동으로 지식그래프화하고 자연어 질의로 즉시 답을 제공합니다.