3장: Neo4j — 프로퍼티 그래프 데이터베이스

// 노드 생성
CREATE (p:Person {name: "김개발", role: "Engineer"})
 
// 관계 생성
MATCH (p:Person {name: "김개발"})
MATCH (t:Technology {name: "Neo4j"})
CREATE (p)-[:USES {since: 2023}]->(t)
 
// 패턴 매칭 - Cypher의 핵심
MATCH (p:Person)-[:USES]->(t:Technology)
WHERE p.name = "김개발"
RETURN p.name, t.name
 
// 가변 길이 경로
MATCH path = (a:Technology)-[:DEPENDS_ON*1..3]->(b:Technology)
RETURN path

// CREATE: 노드와 관계 생성
CREATE (d:Document {title: "GraphRAG 가이드", createdAt: datetime()})
 
// READ: 조건부 조회
MATCH (d:Document)
WHERE d.createdAt > datetime("2026-01-01")
RETURN d.title, d.createdAt
ORDER BY d.createdAt DESC
LIMIT 10
 
// UPDATE: 속성 수정
MATCH (d:Document {title: "GraphRAG 가이드"})
SET d.updatedAt = datetime(), d.status = "published"
 
// DELETE: 노드와 관련 관계 삭제
MATCH (d:Document {title: "임시 문서"})
DETACH DELETE d

// 기술별 문서 수 집계
MATCH (d:Document)-[:COVERS]->(t:Technology)
RETURN t.name, count(d) AS docCount
ORDER BY docCount DESC
 
// 저자별 전문 분야 목록
MATCH (a:Author)-[:EXPERT_IN]->(t:Technology)
RETURN a.name, collect(t.name) AS expertises
 
// CASE 표현식
MATCH (t:Technology)
RETURN t.name,
  CASE
    WHEN t.type = "Database" THEN "데이터베이스"
    WHEN t.type = "Framework" THEN "프레임워크"
    ELSE "기타"
  END AS typeKorean

// B-tree 인덱스 (기본)
CREATE INDEX idx_person_name FOR (p:Person) ON (p.name)
 
// 복합 인덱스
CREATE INDEX idx_doc_category_date
FOR (d:Document) ON (d.category, d.publishedAt)
 
// 유일성 제약 (자동으로 인덱스도 생성)
CREATE CONSTRAINT uniq_tech_name
FOR (t:Technology) REQUIRE t.name IS UNIQUE

// 전문 검색 인덱스 생성
CREATE FULLTEXT INDEX ft_document_content
FOR (d:Document) ON EACH [d.title, d.content, d.summary]
 
// 전문 검색 쿼리
CALL db.index.fulltext.queryNodes("ft_document_content", "GraphRAG knowledge graph")
YIELD node, score
RETURN node.title, score
ORDER BY score DESC
LIMIT 5

// 벡터 인덱스 생성 (1536차원, OpenAI 임베딩 기준)
CREATE VECTOR INDEX vec_document_embedding
FOR (d:Document) ON (d.embedding)
OPTIONS {
  indexConfig: {
    `vector.dimensions`: 1536,
    `vector.similarity_function`: 'cosine'
  }
}
 
// 벡터 유사도 검색
WITH [0.1, 0.2, ...] AS queryEmbedding  // 실제로는 1536차원 벡터
CALL db.index.vector.queryNodes(
  "vec_document_embedding",
  10,
  queryEmbedding
) YIELD node, score
RETURN node.title, score

// 1. 그래프 프로젝션 생성
CALL gds.graph.project(
  'techGraph',
  ['Technology', 'Document'],
  {
    COVERS: {orientation: 'UNDIRECTED'},
    DEPENDS_ON: {orientation: 'NATURAL'}
  }
)
 
// 2. PageRank 실행
CALL gds.pageRank.stream('techGraph')
YIELD nodeId, score
WITH gds.util.asNode(nodeId) AS node, score
RETURN node.name, score
ORDER BY score DESC
LIMIT 10
 
// 3. 커뮤니티 감지 (Louvain)
CALL gds.louvain.stream('techGraph')
YIELD nodeId, communityId
WITH gds.util.asNode(nodeId) AS node, communityId
RETURN communityId, collect(node.name) AS members
ORDER BY size(collect(node.name)) DESC
 
// 4. 노드 임베딩 (FastRP)
CALL gds.fastRP.stream('techGraph', {embeddingDimension: 128})
YIELD nodeId, embedding
WITH gds.util.asNode(nodeId) AS node, embedding
SET node.graphEmbedding = embedding
 
// 5. 프로젝션 정리
CALL gds.graph.drop('techGraph')

from neo4j import GraphDatabase
 
# 드라이버 생성
driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=("neo4j", "password")
)
 
# 세션을 통한 쿼리 실행
def get_technologies(driver, category: str) -> list[dict]:
    """특정 카테고리의 기술 목록을 조회합니다."""
    query = """
    MATCH (t:Technology)-[:BELONGS_TO]->(c:Category {name: $category})
    RETURN t.name AS name, t.version AS version
    ORDER BY t.name
    """
    records, _, _ = driver.execute_query(
        query,
        category=category,
        database_="neo4j"
    )
    return [record.data() for record in records]
 
# 사용
techs = get_technologies(driver, "Database")
for tech in techs:
    print(f"{tech['name']} v{tech['version']}")
 
driver.close()

from neo4j import GraphDatabase, ManagedTransaction
 
driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
 
def create_document(tx: ManagedTransaction, title: str, techs: list[str]) -> str:
    """문서를 생성하고 기술과 연결합니다."""
    result = tx.run("""
        CREATE (d:Document {
            title: $title,
            createdAt: datetime()
        })
        WITH d
        UNWIND $techs AS techName
        MATCH (t:Technology {name: techName})
        CREATE (d)-[:COVERS]->(t)
        RETURN elementId(d) AS docId
    """, title=title, techs=techs)
    record = result.single()
    return record["docId"]
 
# 쓰기 트랜잭션
with driver.session(database="neo4j") as session:
    doc_id = session.execute_write(
        create_document,
        title="Neo4j 벡터 인덱스 활용법",
        techs=["Neo4j", "Python"]
    )
    print(f"생성된 문서 ID: {doc_id}")
 
driver.close()

from neo4j import GraphDatabase
from openai import OpenAI
 
openai_client = OpenAI()
neo4j_driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password"))
 
def hybrid_search(question: str, top_k: int = 5) -> list[dict]:
    """벡터 + 그래프 하이브리드 검색을 수행합니다."""
    # 1. 질문을 벡터로 변환
    response = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=question
    )
    query_embedding = response.data[0].embedding
 
    # 2. 벡터 유사도로 후보 문서를 찾고, 그래프를 순회하여 관련 기술 수집
    query = """
    CALL db.index.vector.queryNodes("vec_document_embedding", $topK, $embedding)
    YIELD node AS doc, score
    OPTIONAL MATCH (doc)-[:COVERS]->(tech:Technology)
    OPTIONAL MATCH (tech)-[:DEPENDS_ON]->(dep:Technology)
    RETURN doc.title AS title,
           score,
           collect(DISTINCT tech.name) AS technologies,
           collect(DISTINCT dep.name) AS dependencies
    ORDER BY score DESC
    """
    records, _, _ = neo4j_driver.execute_query(
        query,
        topK=top_k,
        embedding=query_embedding,
        database_="neo4j"
    )
    return [record.data() for record in records]
 
# 사용
results = hybrid_search("GraphRAG에서 커뮤니티 감지는 어떻게 작동하나요?")
for r in results:
    print(f"[{r['score']:.3f}] {r['title']}")
    print(f"  기술: {', '.join(r['technologies'])}")

from neo4j import GraphDatabase
 
# AuraDB는 neo4j+s:// 프로토콜 사용 (TLS 포함)
driver = GraphDatabase.driver(
    "neo4j+s://xxxxxxxx.databases.neo4j.io",
    auth=("neo4j", "your-auradb-password")
)
 
# 연결 확인
driver.verify_connectivity()
print("AuraDB 연결 성공")