Daddy Makers: 그래프 데이터베이스 Neo4J 기반 데이터 질의 서비스 개발하기

이 글은 그래프 데이터베이스 Neo4J 기반 데이터 질의 서비스 개발하는 방법을 간략히 설명한다. Neo4j는 ACID(Atomicity,Consistency,Isolation,Durability)의 모든 속성을 지원하는 완전 트랜잭션을 지원하는 그래프 데이터베이스이다.

Neo4j는 2007년에 노이만 라디칼 시스템즈(Neo Technology)라는 회사에 의해 개발되었다. 그래프 데이터베이스는 관계형 데이터베이스와 달리 데이터를 노드와 엣지(관계)로 구성된 그래프 형태로 저장한다. 이러한 방식은 실제 세계의 복잡한 관계를 모델링하기에 적합하며, 네트워크, 사회 네트워크, 지도 및 추천 시스템 등 다양한 분야에서 유용하게 사용된다. neo4j는 가장 인기 있는 그래프 데이터베이스 중 하나로, 고성능, 확장성 및 질의 언어 지원을 특징으로 한다.

Neo4j는 SNS분석, 지도 위치 기반 서비스, IoT 등 그래프형 데이터 관리에 주로 사용된다.

Neo4j구조는 여러 프로젝트와 프로젝트에 포함된 데이터베이스로 구성된다.

기본 기능

시작하기

다음 링크에서 프로그램 다운로드 후 설치한다.

Neo4j Free Graph Database Download

설치 후, new project 후 add - local DBMS 선택한다. DB 생성 후 Open 하면 다음같이 브라우저 창을 볼 수 있다. 생성 시 입력 암호를 잘 기억한다.

다음과 같이 neo4j$ 에 명령을 입력해 레코드를 생성해 본다.

CREATE (john:Person {name: 'John'})

CREATE (joe:Person {name: 'Joe'})

CREATE (steve:Person {name: 'Steve'})

CREATE (sara:Person {name: 'Sara'})

CREATE (maria:Person {name: 'Maria'})

CREATE (john)-[:FRIEND]->(joe)-[:FRIEND]->(steve)

CREATE (john)-[:FRIEND]->(sara)-[:FRIEND]->(maria)

다음 명령을 입력, 실행한다.

MATCH (n) RETURN n LIMIT 5

결과는 다음과 같이, 질의한 그래프를 표시해준다. 이전 명령이 아래에 표시된 것을 알 수 있다.

http://127.0.0.1:7474/browser 브라우저로 접속해 본다. 초기 ID/PWD는 neo4j, neo4j이다. 여기서, 데스크탑 프로그램과 동일하게 데이터 관리 가능하다.

그래프 DB와 질의어

그래프 DB는 다음 구성요소를 가진다.

Node: 그래프 데이터 레코드
Relationship: Node 간의 관계
Property: Node의 속성
Label: Node를 묶는 단위

Cypher는 그래프 쿼리 언어이다. 문법은 다음과 같다.

CREATE: 노드, 관계 생성
MATCH: 기존 노트, 관계 검색. RETURN이나 WITH로 매칭된 대상 반환.

WHERE: 조건을 지정

MERGE: CREATE와 MATCH를 합친 함수
SET: 노드 LABEL과 PROPERTY를 업데이트 함
DELETE: 노드, 관계를 삭제

Relation 표현은 다음과 같다.

(node)-[relationship]->(node)

예를 들어, A-[:Knows(since:2020)]->B 경우는 A와 B가 knows 관계가 있으며, since 속성이 2020값을 가진다는 것을 의미한다.

다음은 cypher언어의 기본 형식을 보여준다.

Nodes: ()

Relationships: -[:DIRTECTED]->

Pattern: ()-[]-()

: ()-[]->()

: ()<-[]-()

다음은 이를 이용한 질의 예이다.

MATCH (p:Person)-[:ACTED_IN]->(m:Movie),

(d:Person)-[:DIRECTED]->(m:Movie)

WHERE p.name = 'Tom Hanks' AND p.born = 1956

AND d.name = 'Robert Zemeckis' AND d.born = 1951

RETURN m.title, m.released

파이썬으로 그래프 데이터 관리

이 예에서는 파이썬을 이용해 그래프 데이터를 관리한다. 다음을 설치한다.

pip install neo4j

파이썬에서 드라이버를 설정한다.

from neo4j import GraphDatabase

uri = "bolt://localhost:7687"

username = "neo4j"

password = "neo4jneo4j"

driver = GraphDatabase.driver(uri, auth=(username, password))

session = driver.session()

쿼리를 실행한다.

q = 'MATCH (n) RETURN n LIMIT 5'

nodes = session.run(q)

for node in nodes:

print(node)

query = '''

MERGE (n:Person {name: 'Joe'})

RETURN n

'''

results, summary, keys = driver.execute_query(query, database_='neo4j')

그래프 DB RAG 처리 예시

다음은 그래프 DB를 RAG처리하는 간단한 예시를 보여준다. 이 코드는 text를 langchain의 실험적 기능을 이용해 graph로 변환하고, neo4j에 저장한 후, 이를 질의하는 순서로 실행된다. 이를 실행하기 위해선 미리 PC에 neo4j 설치 후 서버로 우선 실행해야 한다.

text = """

Marie Curie, born in 1867, was a Polish and naturalised-French physicist and chemist who conducted pioneering research on radioactivity.

She was the first woman to win a Nobel Prize, the first person to win a Nobel Prize twice, and the only person to win a Nobel Prize in two scientific fields.

Her husband, Pierre Curie, was a co-winner of her first Nobel Prize, making them the first-ever married couple to win the Nobel Prize and launching the Curie family legacy of five Nobel Prizes.

She was, in 1906, the first woman to become a professor at the University of Paris.

"""

# 라이브러리를 임포트한다.

from dotenv import load_dotenv

from langchain.chains import GraphCypherQAChain

from langchain_community.graphs import Neo4jGraph

from langchain_core.documents import Document

from langchain_experimental.graph_transformers import LLMGraphTransformer

from langchain_openai import ChatOpenAI

load_dotenv()

# llm을 생성한다.

llm = ChatOpenAI(temperature=0, model_name="gpt-4o", openai_api_key=os.getenv("OPENAI_API_KEY")) # LLM 모델 설정. gpt-4o는 실험적 모델로, 성능이 다를 수 있음

llm_transformer = LLMGraphTransformer(llm=llm) # 실험적 모듈. 언어 모델(LLM)을 사용하여 텍스트 데이터를 그래프 데이터로 변환하는 데 사용

# neo4j에 접속하고, 텍스트를 통해 문서를 만든 후, 문서를 그래프 형식으로 변환한다.

def build_graph():

graph = Neo4jGraph(url='bolt://localhost:7687', username='neo4j', password='neo4jneo4j') # Neo4j 그래프 데이터베이스에 연결

documents = [Document(page_content=text)]

graph_documents = llm_transformer.convert_to_graph_documents(documents)

graph.add_graph_documents(graph_documents)

return graph

# 그래프 DB에 질의한다.

def query_graph(graph, query):

chain = GraphCypherQAChain.from_llm(graph=graph, llm=llm, verbose=True, validate_cypher=True)

response = chain.invoke({"query": query})

return response

graph = build_graph()

response = query_graph(graph, "In what university Marie Curie was professor and when she did it?")

print(response)

만약, neo4j가 설치되지 않았다면, 다음 도커 명령을 통해 실행 후 시작한다.

docker run --publish=7474:7474 --publish=7687:7687 --volume=neo_data:/data --env=NEO4J_AUTH=neo4j/12345678oo --env NEO4J_PLUGINS='["apoc"]' -d neo4j

공학용 IFC 그래프로 변환

다음은 IFC파일을 그래프 데이터베이스로 변환하는 파이썬 코드이다.

import ifcopenshell, sys, time

from py2neo import Graph, Node

def typeDict(key):

f = ifcopenshell.file()

value = f.create_entity(key).wrapped_data.get_attribute_names()

return value

ifc_path = "input.ifc" # 입력파일

nodes = []

edges = []

# 노드 설정

f = ifcopenshell.open(ifc_path)

for el in f:

if el.is_a() == "IfcOwnerHistory":

continue

tid = el.id()

cls = el.is_a()

pairs = []

keys = []

try:

keys = [x for x in el.get_info() if x not in ["type", "id", "OwnerHistory"]]

except RuntimeError:

pass

for key in keys:

val = el.get_info()[key]

if any(hasattr(val, "is_a") and val.is_a(thisTyp)

for thisTyp in ["IfcBoolean", "IfcLabel", "IfcText", "IfcReal"]):

val = val.wrappedValue

if val and type(val) is tuple and type(val[0]) in (str, bool, float, int):

val = ",".join(str(x) for x in val)

if type(val) not in (str, bool, float, int):

continue

pairs.append((key, val))

nodes.append((tid, cls, pairs))

for i in range(len(el)):

try:

el[i]

except RuntimeError as e:

if str(e) != "Entity not found":

print("ID", tid, e, file=sys.stderr)

continue

if isinstance(el[i], ifcopenshell.entity_instance):

if el[i].is_a() == "IfcOwnerHistory":

continue

if el[i].id() != 0:

edges.append((tid, el[i].id(), typeDict(cls)[i]))

continue

try:

iter(el[i])

except TypeError:

continue

destinations = [x.id() for x in el[i] if isinstance(x, ifcopenshell.entity_instance)]

for connectedTo in destinations:

edges.append((tid, connectedTo, typeDict(cls)[i]))

if len(nodes) == 0:

print("no nodes in file", file=sys.stderr)

sys.exit(1)

# 그래프 데이터베이스 생성

graph = Graph(auth=('neo4j', 'neo4jneo4j')) # http://localhost:7474

graph.delete_all()

for node in nodes:

nId, cls, pairs = node

one_node = Node("IfcNode", ClassName=cls, nid=nId)

for k, v in pairs:

one_node[k] = v

graph.create(one_node)

# graph.run("CREATE INDEX ON :IfcNode(nid)")

print("Node creat prosess done. Take for ", time.time() - start)

print(time.strftime("%Y/%m/%d %H:%M", time.strptime(time.ctime())))

query_rel = """

MATCH (a:IfcNode)

WHERE a.nid = {:d}

MATCH (b:IfcNode)

WHERE b.nid = {:d}

CREATE (a)-[:{:s}]->(b)

"""

for (nId1, nId2, relType) in edges:

graph.run(query_rel.format(nId1, nId2, relType))

생성 결과는 다음과 같다.

MATCH (n1)-[r]->(n2)

WHERE n1.ClassName = "IfcBuildingStorey"

RETURN r, n1, n2;

결론

Neo4j 그래프 데이터베이스는 데이터를 노드와 엣지(관계)로 구성된 그래프 형태로 저장한다. 이러한 방식은 실제 세계의 복잡한 관계를 모델링하기에 적합하며, 네트워크, 사회 네트워크, 지도 및 추천 시스템 등 다양한 분야에서 유용하게 사용된다. 이 글을 통해 그래프 데이터를 질의하고 정보를 생성하는 서비스 개발 방법을 확인해 보았다.

레퍼런스

Daddy Makers

2024년 5월 22일 수요일

그래프 데이터베이스 Neo4J 기반 데이터 질의 서비스 개발하기

댓글 없음:

댓글 쓰기