엔터프라이즈 Knowledge Graph 구축하기

기관의 전문 지식을 유지하고 재사용하는 것은 지식 관리의 성배와 같아요.

수년에 걸쳐 기업은 기업 전체에서 지식을 유지 및 재사용하고, 바퀴를 재발명하는 것을 방지하고 생산성을 향상시키기 위해 여러 세대의 지식 관리 제품을 활용해 왔어요.

기업 Knowledge Graph를 구축해야 하는 이유

기업 Knowledge Graph는 AI/ML 기반 예측 기능과 결합되어 지식 관리 기술의 미래라고 할 수 있죠.

이는 현재의 지식 관리 구현에 비해 크게 개선된 것이에요. 그래프 기술은 개별 정보를 올바른 컨텍스트와 연결해서 매우 빠르고 유연한 query 기능을 제공하거든요.

AI/ML 기반 추천 또는 예측 기능과 그래프를 결합하면 검색이 특정 정보를 가져올 뿐만 아니라 관련성이 높은 다양한 대체 제안과 복잡한 질문에 대한 답변을 실시간으로 제공할 수 있어요.

엔터프라이즈 앱으로 수집된 숨겨진 데이터의 풍부함

현대 기업은 운영을 간소화하고 기업 데이터를 캡처하기 위해 수많은 다양한 앱을 사용하고 있어요. 여기에는 Google Drive, Slack, Salesforce, Zendesk 등이 포함될 수 있죠.

우리 모두는 이것이 일상적인 생산성을 높이는 데 얼마나 도움이 되는지 알고 있지만, 데이터를 기업을 대표하는 보물 창고로 생각하지 않았을 수도 있어요.

예를 들어 Google 드라이브에서 가져와서 수집된 문서 메타데이터를 에 넣으면 현재 인기 있는 문서 주제가 무엇인지, 어떤 프로젝트가 많은 관심을 받고 있는지, 조직의 Google 드라이브 슈퍼 사용자(예: 많은 문서를 검토/편집하는 사람)가 누구인지 알려줄 수 있어요. 여기에서의 가능성은 무한하며 분석의 힘은 데이터의 양/품질에 의해서만 제한될 거예요.

추가 비즈니스 통찰력에는 문서 유사성, 재사용 가능한 콘텐츠 식별, 직원의 기술 목록, 고객 프로젝트, 사용된 기술, 필요한 기술 등이 포함돼요.

Knowledge Graph 구현

Google Drive API를 사용하여 Knowledge Graph의 구현을 처음부터 한번 살펴볼까요? Neo4j Python 드라이버를 사용할 거예요.

이는 다음을 포함하는 더 큰 엔드투엔드 지식 관리 솔루션의 일부일 뿐이에요*.

• 기업 Knowledge Graph: 기업 지식 저장소 (이 게시물에서 다룰 내용이에요)
• GraphQL API: Knowledge Graph의 꼭대기에 위치해요.
• React 앱: 기술적이지 않은 최종 사용자가 그래프를 탐색할 수 있도록 도와줘요.
• 추천 엔진: 그래프 패턴/알고리즘을 활용하고 React 앱 내에 표시할 관련 콘텐츠를 제공해요.

위의 구성 요소를 포함하는 아키텍처 다이어그램은 다음과 같아요.

*메모: Neo4j 및/또는 Python에 대한 사전 경험이 있으면 도움이 되겠지만 꼭 필요한 건 아니에요!

Google Drive API에 연결

다행히 드라이브 문서 작업 시 Google은 문서 메타데이터를 추출할 수 있는 강력한 API 세트를 제공하고 있어요.

Google Drive API를 통해 문서에 액세스하려면 API 키, Google 서비스 계정 또는 OAuth 통합이 필요해요. 우리의 목적을 위해서는 서비스 계정을 활용하는 것이 가장 합리적이에요. 서비스 계정은 이메일과 같은 특별한 Google 계정이지만 주로 사람이 아닌 개체(예: 코드 조각)와 Google Drive API 간의 통신을 촉진하는 데 사용돼요.

기본적으로 서비스 계정과 문서나 폴더를 공유하고 서비스 계정 자격 증명을 사용하여 Google Drive API에 연결한 다음 API 연결을 통해 문서에 액세스하면 돼요. 정말 간단하죠?

다음은 우리가 구축할 기본 아키텍처와 서비스 계정이 적합한 위치를 보여주는 다이어그램이에요.

보시는 것처럼 "Python Script"와 "Google Drive API"는 Google 서비스 계정을 통해 연결돼요.

서비스 계정을 만들려면 에 방문해서 “API 및 서비스” → “자격 증명”으로 이동한 다음, 페이지 상단의 “자격 증명 만들기”를 클릭하세요. 서비스 계정을 생성하면 "credentials.json" 파일을 다운로드할 수 있어요.

이 파일은 프로젝트 디렉토리에 있어야 하고, 새로 생성된 서비스 계정을 통해 Drive API에 연결할 수 있게 해준답니다.

이제 코딩을 시작할 시간이에요!

문서 추출

먼저 pip3를 사용해서 Python용 Neo4j 드라이버와 몇몇 Google 라이브러리를 설치해야 해요.

pip3 install neo4j
pip3 install google-api-python-client google-auth-httplib2 google-auth-oauthlib

설치가 완료되면 Python 파일을 만들고 "neo4j-google-drive.py" 같은 이름을 붙여주세요. 이 파일 안에서 방금 설치한 라이브러리를 가져오고, 아래처럼 Drive API 연결을 설정하면 돼요.

from apiclient import discovery
from google.oauth2 import service_account
from neo4j import GraphDatabase, basic_auth

SCOPES = "https://www.googleapis.com/auth/drive,https://www.googleapis.com/auth/drive.file".split(",")
credentials = service_account.Credentials.from_service_account_file(
   "./credentials.json", scopes=SCOPES
)
service = discovery.build("drive", "v3", credentials=credentials)

이제 모든 설정이 끝났으니, Google Drive API 메소드 호출을 시작해 볼까요?

추출 기능을 설정해 볼게요. 이 함수는 API 연결('서비스'라고도 함)과 Google Drive 공유 드라이브의 ID를 인수로 사용해요. 공유 드라이브에 관심을 갖는 데는 몇 가지 이유가 있어요.

우선, 공유 드라이브에는 많은 문서가 포함되는 경우가 많고, 이런 문서는 보통 조직 내의 많은 사람들이 관심을 갖죠. 둘째, 개인적인 욕구를 침해하려는 건 아니에요. 그래서 이미 많은 사람이 액세스할 수 있는 대규모 공유 드라이브(보통 마케팅 팀, 엔지니어링 팀 등의 팀 소유)에 집중해서 공개 콘텐츠만 확보하고 있어요.

참고: 일반 Google Drive 폴더에서 문서를 추출하려면 아래 코드를 수정해야 한다는 점을 잊지 마세요!

파일 추출 기능의 스텁은 다음과 같아요.

# retrieve all files in specified shared drive
def get_all_files_from_shared_drive(service, driveId):
   # init function variables
   all_files = []
   page_token = None

   while True:
       if not page_token:
           break

   print("extracted ", len(all_files), " files")
   return all_files

여기 흐름은 이렇습니다.

1. 'all_files'라는 빈 목록을 초기화해요. 이 목록은 나중에 드라이브 파일을 나타내는 Python 사전 객체로 채워질 거예요.
2. "page_token" 변수를 초기화해요. 파일을 요청할 때마다 다른 결과 페이지가 있으면 페이지 토큰을 받게 돼요. 그런 다음 해당 페이지 토큰을 다음 API 호출에 제공해서 다음 파일 페이지를 원한다는 걸 알려주는 거죠. 더 이상 결과가 없으면(즉, 페이지 토큰이 없으면) 루프를 중단하고 파일 목록을 반환해요.

이제 실제 Drive API 호출로 함수 골격을 업데이트해 볼게요.

# retrieve all files in specified shared drive
def get_all_files_from_shared_drive(service, driveId):
   # init function variables
   all_files = []
   page_token = None

   while True:
       # init query params
       params = {}
       params[
           "fields"
       ] = "nextPageToken, files(id, name, mimeType, fileExtension, webViewLink, webContentLink, iconLink, createdTime, modifiedTime, driveId, parents, trashed)"
       params["spaces"] = "drive"
       params["corpora"] = "drive"
       params["driveId"] = driveId
       params["supportsAllDrives"] = True
       params["includeItemsFromAllDrives"] = True
       params["q"] = "trashed = false"
       if page_token:
           params["pageToken"] = page_token

       # call drive api
       page = service.files().list(**params).execute()
       files = page.get("files")
       page_token = page.get("nextPageToken")
       all_files.extend(files)
       if not page_token:
           break

   print("extracted ", len(all_files), " files")
   return all_files

그리고 스크립트를 실행할 때 호출될 기본 함수를 설정해 볼게요.

if __name__ == "__main__":
   all_files = get_all_files_from_shared_drive(service, "YOUR SHARED DRIVE ID")

터미널에서 python3 neo4j-google-drive.py를 입력해서 스크립트를 실행할 수 있어요. 이제 쿼리의 params 객체에 있는 "fields" 값으로 지정된 필드를 포함하는 객체 목록이 있을 거예요.

결과는 다음과 같을 거예요:

[{id: "12345", name: "document1", mimeType: "application/json", ...}, {id: "23456", name: "document2", mimeType: "text/javascript", ...}, ...]

다음으로 그래프를 채워볼까요!

문서 수집

드라이브 파일 목록이 있다면 Neo4j로 가져올 수 있는 몇 가지 방법이 있어요.

한 가지 방법은 CSV 파일에 기록한 다음, CSV 파일을 파싱해서 데이터를 로드하는 거예요. 약간 돌아가는 것처럼 보일 수도 있지만, 필요한 경우 나중에 CSV 파일을 내보낼 수 있는 유연성을 제공하기도 해요.

여기서는 단순화를 위해 전체 CSV 작성/파싱 프로세스를 다루지 않고, Neo4j에 수집할 준비가 된 Google 드라이브 문서 목록(위와 같은)이 있다고 가정할게요.

물론 로컬 컴퓨터나 서버 어딘가에서 실행되는 빈 Neo4j 인스턴스도 필요하겠죠? 설정하는 데 도움이 필요하면 를 참고해주세요.

시작하기 전에 Cypher 쿼리를 사용해서 문서를 로드하기 위해 Neo4j 드라이버와의 상호 작용을 추상화하는 일반 유틸리티 함수를 정의해 볼게요. 이 함수는 데이터베이스 인증, 드라이버 열고 닫기, 제공된 기능 실행을 처리해 줄 거예요.

def ingest_data(func, data):
   driver = GraphDatabase.driver(
       "YOUR_NEO4J_URI",
       auth=basic_auth("YOUR_NEO4J_USERNAME", "YOUR_NEO4J_PASSWORD"),
   )
   with driver.session() as session:
       session.write_transaction(func, data)
   driver.close()

다음으로 수집 함수를 작성해 볼게요. 이 함수는 제공된 문서 목록을 반복하고 데이터베이스에 nodes를 생성하는 Cypher 블록을 포함하고 있어요.

def write_files(tx, files):
   return tx.run(
       """
      UNWIND {files} AS file
      WITH file.id AS id, file.name AS name, file.fileExtension AS fileExtension,
          file.mimeType AS mimeType, datetime(file.createdTime) AS dateCreated,
          datetime(file.modifiedTime) AS dateLastModified, file.webContentLink AS webContentLink,
          file.webViewLink AS webViewLink, file.iconLink AS iconLink, file.trashed AS trashed,
          file.parents AS parents, file.driveId AS driveId
      MERGE (f:DriveFile {id: id})
      SET f.name = name, f.fileExtension = fileExtension, f.type = mimeType,
          f.dateCreated = dateCreated, f.dateLastModified = dateLastModified, f.parents = parents,
          f.iconLink = iconLink, f.driveId = driveId, f.trashed = trashed, f.source = 'drive', f.webViewLink = webViewLink
      """,
       files=files,
   )

다음으로 드라이브 문서 nodes를 서로 연결해서 부모-자식 계층 구조를 만드는 함수를 정의해 볼게요. 즉, 다음과 같은 계층 구조가 있는 경우:

folder1 > folder2 > document

Google 드라이브 내에서 다음을 얻을 수 있어요.

(folder1)<-[:HAS_PARENT]-(folder2)<-[:HAS_PARENT]-(document)

Neo4j 내부 코드를 살펴볼까요?

기능은 다음과 같아요.

def write_parent_rels(tx, files):
   return tx.run(
       """
      UNWIND {files} AS file
      WITH file.id AS id, file.parents AS parentIds
      UNWIND parentIds AS parentId
      WITH id, parentId
      MATCH (f:DriveFile {id: id})
      MATCH (p:DriveFile {id: parentId})
      MERGE (f)-[:HAS_PARENT]->(p)
      """,
       files=files,
   )

마지막으로 `main` 함수에 다음 코드를 추가해서 모든 것을 하나로 묶어봐요.

if __name__ == "__main__":
   all_files = get_all_files_from_shared_drive(service, "YOUR SHARED DRIVE ID")
   ingest_data(write_files, all_files)
   ingest_data(write_parent_rels, all_files)

짜잔! 이제 이전 섹션에서 추출한 공유 드라이브의 그래프가 나타날 거예요.

데이터 모델은 이제 다음과 같은 모습일 텐데요.

데이터 모델 확장

위의 데이터 모델도 꽤 흥미롭고 Google 드라이브 공유 드라이브의 구조를 꽤 정확하게 묘사하지만, 이걸로는 충분하지 않죠. 그래프에서 더 많은 유용성을 얻기 위해 n-gram이라는 또 다른 엔터티를 추가해 볼 거예요.

N-그램은 일반적으로 Natural Language Processing(NLP) 시나리오에 사용되는 단어 그룹이에요. 부정관사, 구두점, 숫자, 관련성이 낮은 용어는 제외되며 수동 태그 지정의 대안이 될 수 있죠.

Knowledge Graph에 사용되는 N-그램은 문서 유사성을 신속하게 평가하고 핵심 용어를 식별하는 좋은 방법이에요. 여기서는 문서 제목을 기준으로 유니그램(단어)과 바이그램(두 단어)을 계산할 거예요. 물론 이걸 수행하는 더 정교한 방법도 있지만, 이건 n-그램을 그래프로 가져오고 문서 `Node`를 연결하는 빠르고 간단한 방법이랍니다.

먼저 NLTK를 설치해 볼게요. 터미널에서 다음 명령을 실행하세요.

pip3 install nltk
python3 -m nltk.downloader stopwords

NLTK는 "Natural Language ToolKit"을 의미하며 n-gram을 만드는 데 필요한 기능이 포함되어 있어요.

Python 스크립트 상단에 다음을 import 해주세요.

import re
import nltk
import string
from nltk.corpus import stopwords
from nltk.tokenize import sent_tokenize, word_tokenize

다음으로, Google Drive에서 가져온 원시 텍스트를 정리하는 함수를 정의해 볼게요. 이 함수는 문자열을 인수(이 경우 문서 제목)로 사용하여 소문자로 바꾸고, 구두점을 제거하고, 불용어(예: "a", "an" 및 "to")를 제거하고, 파일 확장자를 제거해요.

def normalize_and_tokenize_text(str):
   lower_case = str.lower()
   tokens = re.findall(r"[^W_]+|[.,!?;]", lower_case)
   tokens = list(filter(lambda token: token not in string.punctuation, tokens))
   stop_words = list(stopwords.words("english"))
   file_extensions = ["docx", "txt", "xlsx", "pdf", "zip", "tar", "mp4", "pptx", "gz"]
   low_relevancy_terms = ["any terms you think don't hold much meaning in your corpora"]
   stop_words = stop_words + file_extensions + low_relevancy_terms
   tokens = [t for t in tokens if not t in stop_words]
   tokens = [t for t in tokens if len(t) > 1]
   return tokens

다음으로 실제로 n-gram을 계산하는 함수를 정의해 볼게요.

def get_unigrams_from_dict_list_column(dict_list, col, primary_key):
   unigram_list = []
   for d in dict_list:
       tokens = normalize_and_tokenize_text(d[col])
       unigram_list.append({"id": d[primary_key], "type": "unigram", "ngrams": tokens})
   return unigram_list


def get_bigrams_from_dict_list_column(dict_list, col, primary_key):
   bigram_list = []
   for d in dict_list:
       tokens = normalize_and_tokenize_text(d[col])
       bigrams = list(nltk.bigrams(tokens))
       bigrams = list(" ".join(b) for b in bigrams)
       bigram_list.append({"id": d[primary_key], "type": "bigram", "ngrams": bigrams})
   return bigram_list

이전의 Drive 문서 목록, 처리하려는 열의 키, 그리고 문서 기록의 기본 키/ID에 대한 키를 입력으로 사용해요.

이 모든 걸 종합하면, 다음 코드를 실행해서 n-gram 목록을 계산할 수 있어요.

unigrams = get_unigrams_from_dict_list_column(list_of_drive_documents, "name", "id")
bigrams = get_bigrams_from_dict_list_column(list_of_drive_documents, "name", "id")

이제 unigram과 bigram 목록이 준비됐으니, 이걸 Neo4j에 넣어볼까요?

def write_ngrams(tx, ngrams):
   return tx.run(
       """
      UNWIND {ngrams} AS ng
      WITH ng.id AS id, ng.type AS type, ng.ngrams AS ngrams
      UNWIND ngrams AS ngram
      WITH ngram, id, type
      MATCH (f:DriveFile {id: id})
      MERGE (n:NGram {ngram: ngram})
      ON CREATE SET n.type = type
      MERGE (f)-[:HAS_NGRAM]->(n)
      """,
       ngrams=ngrams,
   )

마지막으로, 다음과 유사하게 기본 함수를 업데이트해서 그래프에 작성해 볼게요.

if __name__ == "__main__":
   all_files = get_all_files_from_shared_drive(service, "0ALB-g6xFmljEUk9PVA")
   ingest_data(write_files, all_files)
   ingest_data(write_parent_rels, all_files)
   unigrams = get_unigrams_from_dict_list_column(all_files, "name", "id")
   bigrams = get_bigrams_from_dict_list_column(all_files, "name", "id")
   ingest_data(write_ngrams, unigrams)
   ingest_data(write_ngrams, bigrams)

이제 데이터 모델은 다음과 같은 모습일 거예요.

n-gram을 추가하면 이제 문서의 연결성을 분석하고, 특정 키워드/구문의 인기나 중요도를 파악할 수 있어요. 또, 공유하는 n-gram 수와 문서 계층 구조에서의 위치 (:HAS_PARENT 관계)를 기반으로 문서 간의 유사성을 계산할 수도 있죠.

사용자 개정 내역을 포함해서 데이터 모델을 더 확장할 수도 있어요. Google Drive API는 특정 문서를 수정한 사용자 목록을 반환하는 엔드포인트를 제공하거든요.

이걸 통해 여러분의 조직에 대한 어떤 종류의 통찰력을 얻을 수 있을지 한번 생각해 보셨으면 좋겠어요!

저희는 실제로 Neo4j 내부적으로 "Knowledge Graph 구현" 섹션에 설명된 것과 유사한 시스템을 활용하고 있고, 이걸 고객에게도 제공하고 있답니다.

완성된 애플리케이션의 스크린샷은 다음과 같아요.

궁금한 점이 있다면 언제든지 solutions@neo4j.com으로 문의해주세요! 여러분에게 딱 맞는 솔루션 데모를 보여드릴게요.

앞으로 올라올 블로그 포스팅에서는 추가적인 데이터 소스나 전체 솔루션의 다른 부분들을 다룰 예정이니 기대해주세요!

Neo4j 기술을 한 단계 더 발전시키고 싶으신가요? 저희 온라인 교육 강좌 중 하나를 수강하거나 인증을 받아보세요. GraphAcademy에서 레벨 업!

• Cypher
• 지식 관리 기술
• Machine Learning
• Python

---

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

👉 에이치시스템즈 홈페이지