[TIL-0410] Neo4j Aura로 그래프 DB 구축하기

jiny·2026년 4월 11일
AI AgentNeo4j Aura

AI Agent 실습

목록 보기
13/21

🌟 개요: 그래프 DB와 Neo4j Aura

🔍 프로젝트 배경

이 프로젝트는 AI Agent 기반 인플루언서 매칭 플랫폼이다. 광고 컨셉, 광고 아이템, 광고주 정보를 입력하면 LangGraph React Agent가 DB를 탐색해 최적의 인플루언서를 추천해준다.
처음에는 SQLite + SQLAlchemy로 시작했다. 가짜 인플루언서 50명 정보를 테이블에 넣고, 카테고리·팔로워·예산 조건으로 WHERE 절을 쌓아서 필터링하는 방식이었다. 기본적인 검색은 동작했지만, 기능을 확장하면서 RDB의 한계가 금방 드러났다.

⚠️ RDB로 풀기 어려웠던 문제

  1. 다대다 관계가 너무 많다
    인플루언서는 여러 카테고리를 가질 수 있고, 여러 플랫폼에서 활동한다. 광고도 마찬가지이다. RDB에서는 influencer_categories, influencer_platforms, ad_categories, ad_platforms 같은 중간 테이블이 계속 늘어난다. JOIN이 4~5개씩 붙기 시작하면서 쿼리가 복잡해졌다.

  2. "관계 자체"에 데이터를 담고 싶다
    인플루언서와 광고 사이의 매칭은 단순한 연결이 아니다. 계약 금액, 매칭 상태, 평점, 계약 일시가 관계 위에 붙어야 한다. RDB에서는 이걸 별도 matchings 테이블로 분리해야 하는데, 그러면 "이 인플루언서의 과거 광고 이력을 광고주 정보까지 포함해서 가져와" 같은 쿼리가 금방 3~4 depth JOIN이 된다.

    --- RDB였다면 이런 식의 쿼리가 필요했을 것임
SELECT adv.name, ad.title, m.contract_amount, m.rating
FROM matchings m
JOIN influencers i ON m.influencer_id = i.id
JOIN advertisements ad ON m.advertisement_id = ad.id
JOIN advertisers adv ON ad.advertiser_id = adv.id
WHERE i.id = ?
ORDER BY m.matched_at DESC

  3. 그래프 탐색 자체가 핵심 기능이다
    "이 인플루언서와 비슷한 인플루언서를 찾아라" 라는 요구사항이 있었다. 공통 카테고리와 플랫폼을 많이 공유할수록 유사하다고 보는 방식이다. RDB에서 이 쿼리를 짜면 셀프 조인에 집계까지 붙어서 가독성이 거의 없어진다.
    반면 Cypher로 표현하면 의도가 그대로 보인다.

    MATCH (target:Influencer {id: $id})-[:SPECIALIZES_IN]->(c:Category)
MATCH (similar:Influencer)-[:SPECIALIZES_IN]->(c)
WHERE similar.id <> $id AND similar.is_active = true
WITH similar, count(DISTINCT c) as shared_categories
RETURN similar.name, shared_categories
ORDER BY shared_categories DESC

    "카테고리 노드를 공유하는 인플루언서를 찾는다"는 의도가 쿼리 자체에서 읽힌다. 데이터 구조와 비즈니스 로직이 일치하는 느낌이다.

✨ Neo4j Aura를 선택한 이유

그래프 DB를 써보기로 결정하고 나서 고민은 "어디에서 실행할 것인가?"였다. 로컬에 Neo4j를 도커로 띄우는 방법도 있었지만, 두 가지 이유로 Neo4j Aura Free를 선택했다.

  1. 무료 클라우드 인스턴스
    Aura Free 티어는 별도 서버 없이 1GB 제한의 Neo4j 인스턴스를 무료로 제공한다. 개인 프로젝트나 프로토타입 단계에서 인프라 관리 없이 바로 시작할 수 있다는 게 큰 장점이었다.

  2. 보안 연결 기본 제공
    Aura는 연결 URI가 neo4j+s:// 형태로 TLS 암호화가 기본 적용된다. 별도 SSL 설정 없이 드라이버만 연결하면 끝이다. 설정 파일도 단순하다.

    # config/settings.py
class Settings(BaseSettings)"
   neo4j_uri: str = ""
   neo4j_username: str = ""
   neo4j_password: str = ""
   
   model_config = {
      "env_file": ".env",
   }

    .env에 세 줄만 추가하면 연결 준비가 끝난다. Pydantic BaseSettings를 쓰면 환경변수와 .env 파일을 자동으로 읽어주기 때문에 설정 관리도 깔끔하다.

    # .env
NEO4J_URI=neo4j+s://ff02cf33.databases.neo4j.io
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=your-password

🌟 Neo4j Aura 셋업

🤔 Neo4j Aura란?

Neo4j가 공식 제공하는 완전 관리형 클라우드 그래프 DB 서비스이다. 직접 서버를 띄울 필요 없이 브라우저에서 인스턴스를 생성하면 즉시 사용 가능하다.
Free 티어는 1GB 스토리지 제한이지만 개인 프로젝트나 프로토타입에는 충분하다.

🛠️ Aura 인스턴스 생성

console.neo4j.io에 접속해서 계정을 만들면 된다. 가입 후 "New Instance" → "AuraDB Free"를 선택하면 수십 초 안에 인스턴스가 뜬다.인스턴스가 생성되는 순간 딱 한 번만 자격증명 파일(.txt)을 다운로드할 수 있다. 이 타이밍을 놓치면 비밀번호를 재설정해야 하므로 반드시 저장해둔다.
다운로드한 파일 안에는 아래 6가지 정보가 들어 있다.

🔍 URI 구조 이해하기

neo4j+s://ff92cf33.databases.neo4j.io
│     │   │         └─ Aura 도메인
│     │   └─ 인스턴스 고유 ID
│     └─ +s : TLS 암호화 강제 (Bolt over TLS)
└─ 드라이버 프로토콜 (Bolt)

neo4j+s는 Bolt 프로토콜에 TLS를 강제 적용한 스킴이다. 로컬 Neo4j라면 bolt:// 또는 neo4j://를 쓰지만, Aura는 +s 접미사로 암호화 연결만 허용한다. 덕분에 SSL 인증서 설정을 따로 건드릴 필요가 없다.

⚙️ .env 파일 설정

프로젝트 루트에 .env 파일을 생성하고 Aura에서 받은 값을 채운다..env 파일은 .gitignore에 반드시 추가해야 한다. 자격증명이 그대로 레포에 올라가는 사고를 막기 위해서이다.

⚙️ Pydantic Settings로 자격증명 관리

환경변수를 코드에서 읽을 때 os.environ.get()을 직접 쓰는 방법도 있지만, 이 프로젝트에서는 pydantic-settings를 사용했다. 이유는 두 가지이다.

  1. .env 파일과 실제 환경변수를 자동으로 통합해서 읽어준다.
  2. 타입 힌트 기반으로 자동 형변환과 유효성 검사가 된다.
# config/settings.py
from pydantic_settings import BaseSettings
from typing from Optional

class Settings(BaseSettings):
   # Neo4j
   neo4j_uri: str = ""
   neo4j_username: str = ""
   neo4j_password: str = ""
   
   # LLM
   llm_provider: str = "ollama"
   llm_model: Optional[str] = "qwen2.5:7b-instruct"
   openai_api_key: Optional[str] = None
   anthropic_api_key: Optional[str] = None
   ollama_base_url: str = "http://localhost:11434"
   
   model_config = {
      "env_file": ".env",
      "env_file_encoding": "utf-8",
      "extra": "ignore",  # .env에 정의되지 않은 키가 있어도 에러 없이 무시
   }

BaseSettings를 상속받으면 Settings() 인스턴스를 만드는 시점에 .env 파일을 자동으로 읽는다. 환경변수 이름은 필드 이름의 대문자 버전과 매핑된다. (neo4j_uri → NEO4J_URI) BaseSettings가 내부적으로 필드 이름을 .upper()해서 환경변수 이름으로 사용한다. 즉, .env 파일에 대문자로 써야 인식된다.

⚙️ 의존성 주입 컨테이너에서 연결 초기화

Settings에서 읽어온 자격증명은 Container 싱글톤에서 Neo4j 연결 객체를 만들 때 사용한다.

# infrastructure/container.py
class Container:
   @property
   def neo4j_database(self) -> Neo4jDatabase:
      if self._neo4j_database is None:
         self._neo4j_database = Neo4jDatabase(
            uri=self.settings.neo4j_uri,
            username=self.settings.neo4j_username,
            password=self.settings.neo4j_password,
         )
      return self._neo4j_database

@property로 lazy initialization(지연 초기화)을 구현했다. neo4j_database에 처음 접근하는 순간에만 실제 드라이버 연결이 생성되고, 이후에는 같은 인스턴스를 재사용한다. 앱 시작 시점에 불필요하게 DB 연결을 열지 않아도 된다.

🧐 연결 확인

설정이 끝났으면 연결이 제대로 되는지 확인해볼 수 있다.

from infrastructure.container import Container

container = Container.get_instance()
db = container.neo4j_database

if db.verify_connectivity():
   print("Neo4j Aura 연결 성공")
else:
   print("연결 실패 - URI, 계정 정보를 확인하세요")

verity_connectivity()는 내부적으로 드라이버의 driver.verify_connectivity()를 호출해 실제 핸드셰이크를 시도한다. 비밀번호가 틀리거나 URI가 잘못됐으면 여기서 바로 잡힌다.

🌟 그래프 스키마 설계

❗ 설계 전에 먼저 물어야 할 것

스키마를 짜기 전에 핵심 질문을 하나 던졌다.

"이 시스템에서 가장 많이 탐색하게 될 것은 무엇인가?"

답은 관계였다. "이 인플루언서는 어떤 카테고리를 다루는가", "이 광고에 어울리는 인플루언서는 누구인가", "이 광고주는 어떤 유형의 인플루언서를 선호하는가". 전부 노드와 노드 사이의 연결을 따라가는 탐색이다.
RDB는 데이터를 테이블로 분리하고 JOIN으로 연결한다. 그래프 DB는 반대로, 연결 자체를 1급 객체로 저장한다. 탐색이 핵심인 도메인에서는 그래프 모델이 데이터 구조와 비즈니스 의도를 훨씬 직접적으로 표현할 수 있다.

📝 노드 설계

노드는 독립적으로 식별 가능한 개체를 기준으로 나눴다. 이 프로젝트에는 6종류의 노드가 있다.

  1. Influencer
    인플루언서 한 명의 프로필 정보를 담는 핵심 노드이다. 이름, 이메일, 참여율, 기본 단가, 활동 여부 등이 속성으로 붙는다. 카테고리와 플랫폼 정보는 속성이 아니라 별도 노드와의 관계로 표현한다. 이유는 아래(4, 5)에서 설명한다.

    (Influencer {
   id, name, email, description, engagement_rate,
   base_price_amount, base_price_currency,
   is_active, created_at, updated_at
})

  2. Advertiser
    광고를 집행하는 브랜드나 기업 노드이다. 업종, 규모, 총 예산 같은 정보를 가진다. 광고 캠페인은 노드로 분리했는데, 한 광고주가 여러 캠페인을 집행하는 1:N 구조이기 때문이다.

    (Advertiser {
   id, name, industry, company_size,
   contact_email, total_budget, is_active
})

  3. Advertisement
    광고 캠페인 단위 노드이다. 단순히 광고 정보만 담는 게 아니라 타겟 조건도 함께 가진다. 팔로워 최소/최대, 예산, 기간이 여기 들어간다. 인플루언서를 필터링하는 기준이 광고 노드에 내장된 셈이다.

    (Advertisement {
   id, title, concept, item, description,
   budget, start_date, end_date,
   target_follower_min, target_follower_max,
   status, created_at
})

  4. Category
    "뷰티", "패션", "테크" 같은 콘텐츠 카테고리이다. 처음에는 이걸 Influencer 노드의 배열 속성(categories: ["뷰티", "패션"])으로 저장하는 방법을 고려했다. 하지만 그렇게 하면 "뷰티 카테고리를 공유하는 인플루언서끼리 연결" 같은 그래프 탐색이 불가능해진다.
    카테고리를 독립 노드로 분리했기 때문에 Category 노드를 교차점(hub) 삼아 인플루언서끼리 연결하는 경로 탐색이 가능해진다.

    (Category { name })  // "뷰티", "패션", "테크", ...

  5. Platform
    Instagram, YouTube, TikTok, Blog 같은 소셜 플랫폼 노드이다. Category와 마찬가지로 배열 속성 대신 독립 노드로 분리했다.
    플랫폼을 노드로 두면 "같은 플랫폼에서 활동하는 인플루언서"를 Platform 노드를 경유해서 탐색할 수 있다. 거기에 팔로워 수는 Influencer 속성이 아니라 ACTIVE_ON 관계의 속성으로 붙는다. 플랫폼마다 팔로워 수가 다르기 때문이다.

    (Platform { name })  // "instagram", "youtube", "tiktok", "blog"

  6. AdPost
    인플루언서가 실제로 올린 광고 게시물을 담는 노드이다. 게시물 URL, 캡션, 해시태그 같은 원본 게시물 정보와 함께, AI가 분석한 광고 여부 판단 결과도 속성으로 함께 저장된다.
    어떤 인플루언서의 게시물인지는 속성(influencer_id)으로도 가지고 있지만, 그래프에서는 POSTED_AD 관계로 Influencer 노드와 직접 연결된다.

    (AdPost {
   id, influencer_id, platform, post_url,
   caption, hashtags, posted_at,
   brand, ad_category, confidence, ai_reason, created_at
})

    confidence와 ai_reason이 특징적인 속성이다. 인스타그램 게시물을 AI로 분석해서 "이 게시물이 광고일 확률"(confidence)과 "그 판단 근거"(ai_reason)를 함께 저장한다. 단순히 데이터를 쌓는 게 아니라 AI 분석 결과까지 그래프에 녹여낸 구조이다.

📝 관계 설계

관계는 단순한 연결선이 아니다. 방향이 있고, 속성을 가질 수 있으며, 의미를 이름으로 표현한다.

  1. SPECIALIZES_IN: 인플루언서의 전문 카테고리

    (Influencer)-[:SPECIALIZES_IN]->(Category)

    한 인플루언서가 여러 카테고리를 가질 수 있다. 예를 들어 A 인플루언서는 "뷰티"와 "라이프스타일" 두 카테고리에 SPECIALIZES_IN 관계로 연결된다. 카테고리 노드가 허브 역할을 하기 때문에, 같은 Category 노드를 가리키는 인플루언서끼리는 자연스럽게 연결된다.

  2. ACTIVE_ON: 인플루언서의 플랫폼 활동 (팔로워 수 포함)

    (Influencer)-[:ACTIVE_ON {follower_count: 250000}]->(Platform)

    관계에 속성이 붙는 첫 번째 사례이다. 팔로워 수는 인플루언서에 종속되는 동시에 플랫폼에도 종속된다. follower_count를 Influencer 노드 속성으로 넣으면 플랫폼별로 구분이 안 되고, Platform 노드에 넣으면 어떤 인플루언서의 팔로워인지 모른다. 관계 속성으로 두는 게 유일하게 올바른 위치이다.

  3. CREATED_AD: 광고주와 캠페인의 소유 관계

    (Advertiser)-[:CREATED_AD]->(Advertisement)

    광고주가 캠페인을 만든다는 소유 관계를 표현한다. 광고주 노드에서 출발해 캠페인 노드로 이동하는 방향이다. 한 광고주가 여러 캠페인을 가질 수 있고, 반대로 캠페인에서 광고주를 역으로 조회할 수도 있다.

  4. BELONGS_TO / TARGETS_PLATFORM: 광고의 타겟 조건

    (Advertisement)-[:BELONGS_TO]->(Category)
(Advertisement)-[:TARGETS_PLATFORM]->(Platform)

    광고가 원하는 카테고리와 플랫폼이다. Influencer의 SPECIALIZES_IN, ACTIVE_ON과 같은 Category/Platform 노드를 공유하기 때문에, 광고와 인플루언서의 매칭은 결국 공통 노드를 찾는 탐색으로 치환된다.

  5. MATCHED_WITH: 매칭 계약 (속성이 가장 풍부한 관계)

    (Influencer)-[:MATCHED_WITH {
   id, contract_amout, status, matched_at, rating, notes
}]->(Advertisement)

    이 프로젝트에서 관계가 단순한 연결 그 이상임을 가장 잘 보여주는 예시이다. 계약 금액, 매칭 상태(대기중/수락됨/진행중/완료/취소), 광고주 평점, 계약 일시가 전부 이 관계 위에 붙는다. RDB였다면 matchings 테이블이 됐을 데이터가 그래프 DB에서는 관계의 속성으로 자연스럽게 표현된다.

  6. PROMOTED: 과거 광고 이력

    (Influencer)-[:PROMOTED]->(Advertisement)

    인스타그램 게시물에서 수집한 실제 광고 이력 데이터이다. MATCHED_WITH가 우리 플랫폼에서 생성한 공식 매칭이라면, PROMOTED는 크롤링으로 수집한 과거 실적 데이터이다. 두 관계를 분리해서 데이터 출처를 명확히 구분했다.

  7. POSTED_AD: 광고 게시물

    (Influencer)-[:POSTED_AD]->(AdPost)

    인플루언서가 실제로 올린 광고 게시물 노드로의 연결이다. AdPost 노드에는 게시물 URL, 캡션, 해시태그, AI가 분석한 브랜드·카테고리 정보가 들어간다.

📊 전체 스키마 다이어그램

🤔 이 구조가 유리한 이유

스키마를 이렇게 잡고 나면, 복잡한 비즈니스 쿼리가 직관적인 경로 탐색으로 바뀐다.

  • 예시 1: 광고에 맞는 인플루언서 찾기
    광고와 인플루언서가 같은 Category, Platform 노드를 공유하는지 확인하면 끝이다. 별도 JOIN 테이블 없이 그래프 경로 자체가 필터 조건이 된다.
    MATCH (ad:Advertisement {id: $id})-[:BELONGS_TO]->(c:Category)
MATCH (ad)-[:TARGETS_PLATFORM]->(p:Platform)
MATCH (i:Influencer)-[:SPECIALIZES_IN]->(c)
MATCH (i)-[a:ACTIVE_ON]->(p)
WHERE i.is_active = true
   AND a.follower_count >= ad.target_follower_min
   AND i.base_price_amount <= ad.budget
RETURN i.name, i.engagement_rate, a.follower_count
ORDER BY i.engagement_rate DESC
  • 예시 2: 유사 인플루언서 추천
    공통 카테고리/플랫폼 노드를 경유한 인플루언서를 찾는 것만으로 유사도가 계산된다.
    MATCH (target:Influencer {id: $id})-[:SPECIALIZES_IN]->(c:Category)
MATCH (similar:Influencer)-[:SPECIALIZES_IN]->(c)
WHERE similar.id <> $id AND similar.is_active = true
WITH similar, count(DISTINCT c) as shared_categories
RETURN similar.name, shared_categories
ORDER BY shared_categories DESC
  • 예시 3: 광고주 선호 패턴 분석
    광고주 → 캠페인 → 매칭된 인플루언서 → 카테고리 경로를 한 번에 탐색한다. 평점 4.0 이상인 성공적인 매칭 패턴을 분석해 광고주가 선호하는 인플루언서 유형을 도출한다.
    MATCH (adv:Advertiser {id: $id})-[:CREATED_AD]->(ad:Advertisement)
MATCH (i:Influencer)-[m:MATCHED_WITH]->(ad)
WHERE m.status = '완료' AND m.rating >= 4.0
MATCH (i)-[:SPECIALIZES_IN]->(c:Category)
RETURN collect(DISTINCT c.name) as preferred_categories,
       avg(i.engagement_rate) as avg_engagement,
       avg(m.contract_amount) as avg_contract

세 쿼리 모두 RDB로 짰다면 4~5개 테이블 JOIN에 서브쿼리가 붙는 구조였을 것이다. 그래프 스키마에서는 탐색 경로가 곧 쿼리가 된다.

🌟 Python Neo4j Driver 연결 관리

⚙️ Neo4j Python 드라이버 설치

pip install neo4j>=5.15.0

드라이버 하나로 Aura 연결부터 쿼리 실행까지 전부 처리한다. 이 프로젝트에서는 연결 관리 로직을 Neo4jDatabase 클래스 하나에 모아뒀다.

⚙️ Neo4jDatabase 클래스 전체 구조

# adapters/outbound/persistence/neo4j/database.py
from neo4j import GraphDatabase, Driver

class Neo4jDatabase:
   def __init__(
      self,
      uri: str,
      username: str,
      password: str,
      database: str = "neo4j"
   ):
      self.uri = uri
      self.database = database
      self._driver: Optional[Driver] = None  # 아직 연결 안 함
      self._username = username
      self._password = password

생성자에서 드라이버를 바로 만들지 않는다. 자격증명과 URI만 보관해두고, 실제 연결은 처음 필요한 시점에 열린다.

⚙️ Lazy Initialization

_driver를 @property로 감싸서 처음 접근하는 순간에만 GraphDatabase.driver()를 호출한다.

@property
def driver(self) -> Driver:
   if self._driver is None:
      self._driver = GraphDatabase.driver(
         self.uri,
         auth=(self._username, self._password)
      )
   return self._driver

이 패턴의 이점은 두 가지이다.

  1. 앱 시작 비용이 낮다
    Container 싱글톤이 초기화될 때 Neo4jDatabase 객체는 만들어지지만, 실제 TCP 연결은 DB를 처음 쿼리하는 순간까지 미뤄진다.

    ⭐ TCP 연결이란?

    TCP(Transmission Control Protocol)는 인터넷에서 두 컴퓨터가 데이터를 주고받을 때 사용하는 통신 규약이다.
    Neo4j Aura는 클라우드에 있는 원격 서버이다. Python 코드에서 쿼리를 실행하려면 내 컴퓨터 → Neo4j Aura 서버 사이에 물리적인 네트워크 연결이 먼저 열려야 하는데, 이 연결 자체가 TCP 연결이다.

    내 Python 코드
     │
     │  (TCP 연결 수립 — 핸드셰이크, 인증)
     ▼
Neo4j Aura 서버 (neo4j+s://ff92cf33.databases.neo4j.io)

    TCP 연결을 여는 것 자체에 시간 비용이 있다. 서버와 "연결해도 돼?" → "응, 연결해" → "알겠어" 하는 핸드셰이크 과정이 필요하기 때문이다.
    그래서 Lazy init의 의미는

    # 앱 시작 시점 - Neo4jDatabse 객체만 만들고 TCP 연결은 아직 안 열림
db = Neo4jDatabase(uri, username, password)
# 첫 쿼리 시점 - 이 순간 비로소 TCP 연결이 열림
db.execute_query("MATCH (n) RETURN n")

    앱이 시작할 때마다 무조건 연결을 열면, DB를 실제로 쓰지 않는 상황에서도 연결 비용이 발생한다. Lazy init은 실제로 필요한 순간까지 그 비용을 미루는 패턴이다.

  2. 연결 실패를 늦출 수 있다
    Aura 인스턴스가 잠깐 다운됐을 때 앱 자체가 시작 불가가 되는 상황을 막는다. 쿼리 시점에 실패하면 해당 요청만 에러가 나고 앱은 계속 살아있다.

Container에서도 같은 패턴을 한 번 더 적용한다.

# infrastructure/container.py
@property
def neo4j_database(self) -> Neo4jDatabase:
   if self._neo4j_database is None:
      self._neo4j_database = Neo4jDatabase(
         uri=self.settings.neo4j_uri,
         username=self.settings.neo4j_username,
         password=self.settings.neo4j_password,
      )
   return self._neo4j_database

Container → Neo4jDatabase → Driver 세 단계 모두 lazy하게 초기화된다.

⚙️ 세션 관리 - Context Manager

드라이버가 만들어지면, 실제 쿼리는 세션 단위로 실행된다. 세션은 열고 나면 반드시 닫아야 한다. 이걸 수동으로 관리하면 실수가 생기기 쉽기 때문에 contextmanager로 감쌌다.

from contextlib import contextmanager

@contextmanager
def get_session(self):
   session = self.driver.session(database=self.database)
   try:
      yield session
   finally:
      session.close()  # 예외가 발생해도 반드시 닫힘

try/finally 구조 덕분에 쿼리 도중 예외가 터져도 세션이 누수되지 않는다. 사용하는 쪽에서는 with 블록만 쓰면 된다.

with self.get_session() as session:
   result = session.run("MATCH (n) RETURN count(n)")

⚙️ execute_read / execute_write 분리

세션을 직접 열어서 쿼리를 날리는 방법 외에, Neo4j 드라이버는 트랜잭션 함수 방식도 지원한다. 이 프로젝트에서는 세 가지 실행 메서드를 계층으로 나눴다.

  1. 읽기 전용: 빠른 단순 조회

    def execute_query(self, query: str, parameters: dict = None) -> list:
   with self.get_session() as session:
      result = session.run(query, parameters or {})
      return [record.data() for record in result]

  2. DB 레벨 읽기 트랜잭션: 쓰기 시도 시 DB가 직접 거부

    def execute_read(self, query: str, parameters: dict = None) -> list:
   with self.get_session() as session:
      result = session.execute_read(
         lambda tx: list(tx.run(query, parameters or {}))
      )
      return [record.data() for record in result]

  3. 쓰기 트랜잭션: 실패 시 자동 롤백

    def execute_write(self, query: str, parameters: dict = None) -> list:
   with self.get_session() as session:
      result = session.execute_write(
         lambda tx: list(tx.run(query, parameters or {}))
      )
      return [record.data() for record in result]

세 메서드의 차이를 정리하면 다음과 같다.

메서드트랜잭션 보장쓰기 차단주 용도
execute_query없음없음간단한 조회
execute_readDB 레벨 보장DB가 직접 거부안전한 읽기
execute_write있음. 실패 시 롤백-노드/관계 생성, 수정

실제 사용 패턴을 보면 의도가 명확하게 구분된다.

# 조회 - execute_query
def find_similar_influencers(self, influencer_id):
   return self.db.execute_query("""
      MATCH (i:Influencer {id: $id})-[:SPECIALIZES_IN]->(c:Category)
      ...
   """, {"id": str(influencer_id)})
   
# 생성/수정 - execute_write
def create_influencer_node(self, influencer):
   return self.db.execute_write("""
      MERGE (i:Influencer {id: $id})
      SET i.name = $name, ...
   """, {...})

execute_write에서 MERGE를 쓰면 동일 id가 이미 있을 때 새로 만들지 않고 업데이트한다. INSERT OR UPDATE를 별도 분기 없이 한 쿼리로 처리할 수 있다.

⚙️ 연결 재사용과 드라이버 수정

Neo4j Python 드라이버는 내부적으로 커넥션 풀을 관리한다. GraphDatabase.driver()는 앱 전체에서 한 번만 호출하고 계속 재사용하도록 설계됐다. 매 쿼리마다 드라이버를 새로 만들면 TCP 연결 비용이 반복 발생한다.
이 프로젝트에서는 Container 싱글톤이 Neo4jDatabase를 하나만 유지하고, 모든 Repository가 같은 인스턴스를 공유한다.

# 어디서 호출해도 같은 Neo4jDatabsae 인스턴스를 반환
container = Container.get_instance()
influencer_repo = container.influencer_repository()  # Neo4jDatabase 공유
neo4j_repo = container.neo4j_repository()            # 동일 인스턴스 사용

앱이 종료될 때는 close()를 명시적으로 호출해서 드라이버 리소스를 해제한다.

def close(self):
   if self._driver:
      self._driver.close()
      self._driver = None

Container에도 reset_instance()를 두어 테스트 환경에서 강제로 드라이버를 닫고 재초기화할 수 있게 됐다.

@classmethod
def reset_instance(cls) -> None:
   if cls._instance and cls._instance._neo4j_database:
      cls._instance._neo4j_database.close()
   cls._instance = None

⚙️ Context Manager로 직접 사용하기

Neo4jDatabase는 __enter__ / __exit__도 구현하고 있어서 with문으로 쓸 수도 있다.

def __enter__(self):
   return self
   
def __exit__(self, exc_type, exc_val, exc_tb):
   self.close()

with문을 사용하면 블록이 끝나는 순간 __exit__가 자동으로 호출되면서 close()가 실행된다. 예외가 발생해도 마찬가지이다. 그래서 이 두 코드는 완전히 동일하다.

# with문 사용: close()를 직접 호출하지 않아도 됨
with Neo4jDatabase(uri, username, password) as db:
   db.execute_write("MERGE (i:Influencer {id: $id}) SET ...", {...})
# with 블록을 벗어나는 순간 __exit__ → close() 자동 호출

# with문 미사용: close()를 직접 챙겨야 함
db = Neo4jDatabase(uri, username, password)
db.execute_write("MERGE (i:Influencer {id: $id}) SET ...", {...})
db.close()  # 빠뜨리면 연결이 안 닫힘

이 프로젝트에서 평소에는 Container 싱글톤이 Neo4jDatabase의 수명을 관리하기 때문에 with문을 쓸 일이 없다. 반면 seed_bulk_data.py같은 한 번 실행하고 끝내는 스크립트에서는 with문으로 감싸두면 close() 호출을 잊어버릴 걱정 없이 깔끔하게 마무리할 수 있다.

⚙️ 연결 확인

코드를 처음 붙이고 나서 연결이 제대로 됐는지 확인할 때는 verify_connectivity()를 쓴다.

def verify_connectivity(self) -> bool:
   try:
      self.driver.verify_connectivity()
      return True
   except Exception as e:
      print(f"Neo4j 연결 실패: {e}")
      return False

현재 프로젝트에서는 verify_connectivity()는 정의만 되어 있고 호출하지 않는다. 연결 문제를 디버깅하거나 헬스체크가 필요할 때 직접 호출해서 쓸 수 있는 유틸리티 메서드이다.

# 필요할 때 직접 호출하는 방식
db = container.neo4j_database
if db.verify_connectivity():
   print("연결 성공")

실제로는 첫 execute_query()가 호출되는 순간 연결이 열리고, 연결이 안 되면 그 시점에 에러가 발생하는 구조라 별도 연결 확인 단계가 없어도 동작에는 문제가 없다.

🌟 Repository 패턴으로 Cypher 쿼리 추상화

🤔 Repositoy 패턴이란?

Repository 패턴은 데이터 저장소 접근 로직을 비즈니스 로직으로부터 분리하는 설계 패턴이다.
가장 쉽게 이해하는 방법은 "창고 관리인" 비유이다. 비즈니스 로직(서비스)은 창고 안이 어떻게 생겼는지 모른다. 그냥 관리인(Repositoy)에게 "인플루언서 찾아줘"라고 요청할 뿐이다. 관리인이 내부적으로 Neo4j를 뒤지든, SQLite를 뒤지든 서비스 입장에서는 상관없다.이 구조의 핵심은 서비스 코드가 구현체를 직접 참조하지 않는다는 점이다. 내일 Neo4j를 PostgreSQL로 바꾸기로 했다면, PostgreSQLInfluencerRepository를 새로 만들고 Container에서 갈아끼우면 된다. 서비스 코드는 한 줄도 건드리지 않아도 된다.

# 서비스 코드는 포트(인터페이스)만 바라봄
class AgentService:
   def __init__(self, influencer_repo: InfluencerRepositoryPort):
      self.repo = influencer_repo  # Neo4j인지 SQLite인지 모름
      
# 어떤 구현체를 쓸지는 Container에서만 결정
def influencer_repository(self) -> InfluencerRepositoryPort:
   return Neo4jInfluencerRepository(self.neo4j_database)  # 여기만 바꾸면 됨

🤔 왜 Repository 패턴인가?

Neo4jDatabase는 Cypher 쿼리를 실행하는 저수준 연결 관리자이다. 비즈니스 로직에서 이 클래스를 직접 쓰면 Cypher 쿼리가 서비스 코드 곳곳에 흩어진다. DB를 바꾸거나 쿼리를 수정할 때 고쳐야 할 곳이 너무 많아진다.
Repositoy 패턴은 이 문제를 인터페이스와 구현체를 분리해서 해결한다. 비즈니스 로직은 인터페이스만 바라보고, 실제 Cypher 쿼리는 구현체 안에 완전히 숨겨진다.

✨ InfluencerRepositoryPort: 인터페이스 정의

# application/ports/outbound/influencer_repository_port.py
from abc import ABC, abstractmethod

class InfluencerRepositoryPort(ABC):
   @abstractmethod
   def save(self, influencer: Influencer) -> Influencer: ...
   
   @abstractmethod
   def find_by_id(self, influencer_id: UUID) -> Optional[Influencer]: ...
   
   @abstractmethod
   def find_by_categories(self, categories: List[str]) -> List[Influencer]: ...
   
   @abstractmethod
   def find_by_platform(self, platform: str) -> List[Influencer]: ...
   
   @abstractmethod
   def find_by_follower_range(self, min_followers: int, max_followers: Optional[int]) -> List[Influencer]: ...
   
   @abstractmethod
   def find_eligible_for_campaign(self, categories, platforms, min_followers, max_followers, max_price) -> List[Influencer]: ...
   
   @abstractmethod
   def delete(self, influencer_id: UUID) -> bool: ...
   
   @abstractmethod
   def search(self, query: str) -> List[Influencer]: ...

ABC와 @abstractmethod로 선언된 인터페이스이다. 구현체가 메서드를 빠뜨리면 인스턴스화 시점에 에러가 난다. 비즈니스 로직은 이 포트만 바라본다. Neo4j를 쓰든, SQLite를 쓰든, 인터페이스가 동일하면 서비스 코드는 한 줄도 바뀌지 않는다.

✨ Neo4jInfluencerRepository: 구현체

# adapters/outbound/persistence/neo4j/neo4j_influencer_repository.py
class Neo4jInfluencerRepository(InfluencerRepositoryPort):
   def __init__(self, database: Neo4jDatabase):
      self.db = database

InfluencerRepositoryPort를 상속받아 모든 추상 메서드를 Cypher로 구현한다. Neo4jDatabase를 생성자로 주입받기 때문에 테스트에서 mock으로 교체하기도 쉽다.

🤔 의존성 주입 (Dependency Injection)

의존성 주입 덕분에 테스트할 때 실제 Neo4j 대신 가짜 객체를 끼워 넣을 수 있다.

# 실제 코드
class Neo4jInfluencerRepository(InfluencerRepositoryPort):
   def __init__(self, database: Neo4jDatabase):
      self.db = database  # 외부에서 주입받음

Neo4jDatabase를 내부에서 직접 만들지 않고, 생성자 파라미터로 받는다. 그래서 테스트할 때 이렇게 할 수 있다.

# 테스트 코드
class MockDatabase:
   def execute_query(self, query, params):
      return [{"i": {"id": "123", "name": "김소영"}, ...}]  # 가짜 데이터 반환     
   def execute_write(self, query, params):
      return []
# 실제 Neo4j Aura 대신 Mock을 주입
mock_db = MockDatabase()
repo = Neo4jInfluencerRepository(mock_db)  # 가짜 DB로 교체
result = repo.find_by_id("123")  # 실제 Neo4j 없이 테스트 가능

반대로 Neo4jDatabase를 내부에서 직접 작성했다면

# 이렇게 짰다면 테스트가 어려움
class Neo4jInfluencerRepository:
   def __init__(self):
      self.db = Neo4jDatabase(uri="neo4j+s://...", ...)  # 내부에서 직접 생성

이 경우 테스트를 돌릴 때마다 실제 Neo4j Aura에 네트워크 연결이 필요하고, 테스트 데이터가 실제 DB에 쌓이는 부작용이 생긴다.
생성자로 주입받는 구조 덕분에 테스트 환경에서는 가짜 DB를, 실제 환경에서는 진짜 DB를 갈아끼울 수 있다.

✨ 대표 Cypher 쿼리

  1. MERGE: INSERT OR UPDATE를 한 번에
    save() 메서드에서 인플루언서 노드를 저장할 때 CREATE 대신 MERGE를 쓴다.
    MERGE (i:Influencer {id: $id})
SET i.name = $name,
    i.email = $email,
    i.engagement_rate = $engagement_rate,
    i.created_at = datetime($created_at)
    MERGE는 {id: $id} 조건으로 노드를 먼저 찾고, 없으면 새로 만들고, 있으면 SET으로 덮어쓴다. 덕분에 중복 체크 없이 저장 로직을 단순하게 유지할 수 있다.
  1. 관계 생성: MATCH + MERGE 조합
    노드를 저장한 뒤에 카테고리, 플랫폼과의 관계를 연결한다.
    -- 카테고리 연결
MATCH (i:Influencer {id: $id})
MERGE (c:Category {name: $category_name})
MERGE (i)-[:SPECIALIZES_IN]->(c)

-- 플랫폼 연결 (관계에 팔로워 수 포함)
MATCH (i:Influencer {id: $id})
MERGE (p:Platform {name: $platform_name})
MERGE (i)-[r:ACTIVE_ON]->(p)
SET r.follower_count = $follower_count
    패턴이 일정하다. MATCH로 이미 저장된 Influencer 노드를 찾고, MERGE로 Category/Platform 노드를 찾거나 없으면 새로 만들고, 다시 MERGE로 관계를 연결한다.
    관계도 MERGE이기 때문에 같은 인플루언서를 두 번 저장해도 관계가 중복 생성되지 않는다.
  1. OPTIONAL MATCH + collect: 노드와 관계를 한 번에 조회
    find_by_id()를 보면 Influencer 노드와 연결된 카테고리, 플랫폼을 한 쿼리에서 함께 가져온다.
    MATCH (i:Influencer {id: $id, is_active: true})
OPTIONAL MATCH (i)-[:SPECIALIZES_IN]->(c:Category)
OPTIONAL MATCH (i)-[r:ACTIVE_ON]->(p:Platform)
RETURN i,
       collect(DISTINCT c.name) as categories,
       collect(DISTINCT {platform: p.name, followers: r.follower_count}) as platforms
    OPTIONAL MATCH는 SQL의 LEFT_JOIN과 같다. 카테고리나 플랫폼이 없는 인플루언서도 결과에서 누락되지 않는다.
    collect(DISTINCT ...)는 매칭된 여러 노드를 리스트로 묶어준다. 이 한 쿼리로 아래 결과가 나온다.
    {
  "i": {"id": "...", "name": "김소영", ...},
  "categories": ["뷰티"],
  "platforms": [
    {"platform": "instagram", "followers": 250000},
    {"platform": "youtube", "followers": 180000}
  ]
}
  1. WITH + 집계 필터: 팔로워 범위 조회
    find_by_follower_range()에서는 플랫폼별 팔로워를 합산한 값으로 필터링한다. WHERE는 집계 결과에 바로 걸 수 없기 때문에 WITH로 중간 결과를 한 번 내보낸 뒤 조건을 건다.
    MATCH (i:Influencer {is_active: true})
OPTIONAL MATCH (i)-[r:ACTIVE_ON]->(p:Platform)
WITH i,
     collect(DISTINCT {platform: p.name, followers: r.follower_count}) as platforms,
     sum(r.follower_count) as total_followers  -- 플랫폼별 팔로워 합산
WHERE total_followers >= $min_followers
  AND ($max_followers IS NULL OR total_followers <= $max_followers)
RETURN i, platforms
    WITH는 Cypher에서 파이프 역할을 한다. 앞 단계의 결과를 변수로 묶어 다음 단계로 넘기고, 그 시점에 WHERE이나 추가 MATCH를 붙일 수 있다.
  1. soft Delete: is_active 플래그
    delete()는 노드를 실제로 지우지 않는다.
    MATCH (i:Influencer {id: $id})
SET i.is_active = false
RETURN i
    is_active = false로만 바꾼다. 모든 쿼리에는 {is_active: true} 조건이 붙어있기 때문에 삭제된 인플루언서는 자연스럽게 결과에서 빠진다. 관계와 이력 데이터를 보존하면서도 논리적으로 삭제된 것처럼 동작한다.

✨ Neo4j 노드 → Python 엔티티 변환

쿼리 결과는 딕셔너리 형태로 반환된다. 이걸 도메인 엔티티로 바꾸는 작업을 _to_entity()가 담당한다.

def _to_entity(self, node_data: dict, categories: List[str], platforms: List[dict]) -> Influencer:

   # Neo4j DateTime → Python datetime 변환
   created_at = node_data.get("created_at")
   if hasattr(created_at, "to_native"):
      created_at = created_at.to_native()
   
   # 문자열 플랫폼명 → PlatformType Enum 변환
   platform_dict = {}
   for p in platforms:
      try:
         platform_type = PlatformType.from_string(p["platform"])
         platform_dict[platform_type] = int(p["followers"])
      except Exception:
         pass  # 알 수 없는 플랫폼은 스킵
         
   return Influencer(
      id=UUID(node_data["id"]),
      categories=Category.from_list(categories),
      platforms=platform_dict,
      ...
   )

두 가지 변환이 핵심이다.
첫째, Neo4j의 datetime 타입은 Python의 datetime과 다르기 때문에 .to_native()로 변환한다.
둘째, Neo4j에 문자열로 저장된 플랫폼명("instagram")을 PlatformType Enum으로 바꿔서 타입 안정성을 확보한다. 알 수 없는 플랫폼명이 들어와도 except로 스킵하기 때문에 데이터 불일치로 앱이 죽는 상황을 방지한다.

🌟 그래프 탐색 쿼리: RDB와의 차이점

그래프 DB는 노드와 노드 사이를 따라가는 탐색이 특징이다.

🔍 유사 인플루언서 찾기

📋 비즈니스 요구사항

특정 인플루언서와 카테고리, 플랫폼을 많이 공유할수록 유사하다고 본다.

  • RDB: influencer_categories 중간 테이블을 셀프 조인

    SELECT i2.id, i2.name, COUNT(DISTINCT ic2.category_id) as shared_categories
FROM influencers i1
JOIN influencer_categories ic1 ON i1.id = ic1.influencer_id
JOIN influencer_categories ic2 ON ic1.category_id = ic2.category_id
JOIN influencer_platforms ip1 ON i1.id = ip1.influencer_id
JOIN influencer_platforms ip2 ON ip1.platform_id = ip2.platform_id
JOIN influencers i2 ON ic2.influencer_id = i2.id
WHERE i1.id = ? AND i2.id <> i1.id AND i2.is_active = true
GROUP BY i2.id, i2.name
ORDER BY shared_categories DESC
LIMIT 5

  • 그래프 DB: 공유 Category 노드 수를 카운트

    # adapters/outbound/persistence/neo4j/neo4j_repository.py
def find_similar_influencers(self, influencer_id: UUID, limit: int = 5):
   query = """
   MATCH (target:Influencer {id: $influencer_id})-[:SPECIALIZES_IN]->(c:Category)
   MATCH (target)-[ta:ACTIVE_ON]->(p:Platform)
   MATCH (similar:Influencer)-[:SPECIALIZES_IN]->(c)
   MATCH (similar)-[sa:ACTIVE_ON]->(p)
   WHERE similar.id <> $influencer_id
     AND similar.is_active = true
   WITH similar,
        count(DISTINCT c) as shared_categories,
        count(DISTINCT p) as shard_platforms,
        avg(sa.follower_count) as avg_followers
   RETURN similar.id as id,
          similar.name as name,
          similar.engagement_rate as engagement_rate,
          shared_categories,
          shared_platforms,
          avg_followers
   ORDER BY shared_categories DESC, shared_platforms DESC
   LIMIT $limit
   """
   return self.db.execute_query(query, {
      "influencer_id": str(influencer_id),
      "limit": limit
   })

    쿼리를 단계별로 읽으면 탐색 경로가 그대로 보인다.

    1. targer 인플루언서에서 출발
    2. SPECIALIZES_IN 관계를 따라 Category 노드로 이동
    3. 같은 Category를 가진 다른 Influencer(similar)를 찾음
    4. 그 similar가 target과 같은 Platform도 사용하는지 확인
    5. 공유 카테고리/플랫폼 수를 집계해서 정렬

    중간 테이블을 JOIN하는 대신 관계 자체를 경로로 따라가는 방식이다.
    count(DISTINCT c)는 두 인플루언서가 공유하는 Category 노드의 수를 세는 것이고, 이 숫자가 유사도 점수가 된다.

🔍 광고 캠페인에 맞는 인플루언서 필터링

📋 비즈니스 요구사항

광고의 타겟 카테고리, 플랫폼, 팔로워 조건, 예산을 모두 만족하는 인플루언서를 찾는다.

  • RDB: 카테고리/플랫폼 중간 테이블을 4번 JOIN하고 팔로워·예산 조건은 서브쿼리로 처리

    SELECT i.id, i.name, i.engagement_rate, i.base_price_amount,
       COUNT(DISTINCT ic.category_id) as category_matches,
       SUM(ip.follower_count) as total_followers
FROM influencers i
JOIN influencer_categories ic ON i.id = ic.influencer_id
JOIN ad_categories as ON ic.category_id = ac.category_id
JOIN influencer_platforms ip ON i.id = ip.influencer_id
JOIN ad_platforms ap ON ip.platform_id = ap.platform_id
WHERE ac.advertisement_id = ?
  AND ap.advertisement_id = ?
  AND i.is_active = true
  AND ip.follower_count >= (SELECT target_follower_min FROM advertisements WHERE id = ?)
  AND i.base_price_amount <= (SELECT budget FROM advertisements WHERE id = ?)
GROUP BY i.id
ORDER BY category_matches DESC, i.engagement_rate DESC

  • 그래프 DB: 광고 노드에서 출발해 공유 Category·Platform 노드를 경유하는 인플루언서를 검색

    def find_matching_influencers_for_ad(self, advertisement_id: UUID, limit: int = 10):
   query = """
   MATCH (ad:Advertisement {id: $advertisement_id})-[:BELONGS_TO]->(c:Category)
   MATCH (ad)-[:TARGETS_PLATFORM]->(p:Platform)
   MATCH (i:Influencer)-[:SPECIALIZES_IN]->(c)
   MATCH (i)-[a:ACTIVE_ON]->(p)
   WHERE i.is_active = true
     AND a.follower_count >= ad.target_follower_min
     AND (ad.target_follower_max IS NULL OR a.follower_count <= ad.target_follower_max)
     AND i.base_price_amount <= ad.budget
   WITH i, ad,
        count(DISTINCT c) as category_matches,
        sum(a.follower_count) as total_followers
   RETURN i.id as id,
          i.engagement_rate as engagement_rate,
          i.base_price_amount as price,
          category_matches,
          total_followers
   ORDER BY category_matches DESC, i.engagement_rate DESC
   LIMIT $limit
   """
   return self.db.execute_query(query, {
      "advertisement_id": str(advertisement_id),
      "limit": limit
   })

    이 쿼리의 핵심은 광고와 인플루언서가 같은 Category, Platform 노드를 공유하는지 확인하는 부분이다.

    [Advertisement] -BELONGS_TO-> [Category] <-SPECIALIZES_IN- [Influencer]
[Advertisement] -TARGETS_PLATFORM-> [Platform] <-ACTIVE_ON- [Influencer]

    광고와 인플루언서가 동일한 Category 노드, Platform 노드를 가리키고 있으면 자연스럽게 조건을 만족한다. '그래프 스키마 설계' 단계에서 Category와 Platform을 독립 노드로 분리한 이유가 여기서 드러난다.
    두 노드가 교차점 역할을 하기 때문에 별도 매핑 테이블 없이 경로 탐색만으로 매칭이 이루어진다.

🔍 광고주 선호 패턴 분석

📋 비즈니스 요구사항

특정 광고주가 과거에 높은 평점을 준 매칭들을 분석해서 선호하는 인플루언서 유형을 도출한다.

  • RDB: 광고주→광고→매칭→인플루언서→카테고리/플랫폼까지 6개 테이블을 JOIN하고 WHERE로 평점 필터

    SELECT GROUP_CONCAT(DISTINCT c.name) as preferred_categories,
       GROUP_CONCAT(DISTINCT p.name) as preferred_platforms,
       AVG(ip.follower_count) as avg_followers,
       AVG(i.engagement_rate) as avg_engagement,
       AVG(m.contract_amount) as avg_contract
FROM advertisers adv
JOIN advertisements ad ON adv.id = ad.advertiser_id
JOIN matchings m ON ad.id = m.advertisement_id
JOIN influencers i ON m.influencer_id = i.id
JOIN influencer_categories ic ON i.id = ic.influencer_id
JOIN categories c ON ic.category_id = c.id
JOIN influencer_platforms ip ON i.id = ip.influencer_id
JOIN platforms p ON ip.platform_id = p.id
WHERE adv.id = ? AND m.status '완료' AND m.rating >= 4.0

  • 그래프 DB: Advertiser에서 출발해 4단계 경로를 따라가면서 MATCHED_WITH 관계 속성으로 바로 평점 필터

    def get_advertiser_preference_pattern(self, advertiser_id: UUID):
   query = """
   MATCH (adv:Advertiser {id: $advertiser_id})-[:CREATED_AD]->(ad:Advertisement)
   MATCH (i:Influencer)-[m:MATCHED_WITH]->(ad)
   WHERE m.status = '완료' AND m.rating >= 4.0
   MATCH (i)-[:SPECIALIZES_IN]->(c:Category)
   MATCH (i)-[a:ACTIVE_ON]->(p:Platform)
   WITH adv,
        collect(DISTINCT c.name) as preferred_categories,
        collect(DISTINCT p.name) as preferred_platforms,
        avg(a.follower_count) as avg_followers,
        avg(i.engagement_rate) as avg_engagement,
        avg(m.contract_amount) as avg_contract
   RETURN preferred_categories,
          preferred_platforms,
          avg_followers,
          avg_engagement,
          avg_contract
   """
   results = self.db.execute_query(query, {
      "advertiser_id": str(advertiser_id)
   })
   return results[0] if results else {}

    이 쿼리는 4단계 경로 탐색을 한 번에 처리한다.

    Advertiser
  └─ CREATED_AD ──> Advertisement
                       └─ MATCHED_WITH (rating >= 4.0) ──> Influencer
                                                               ├─ SPECIALIZES_IN ──> Category
                                                               └─ ACTIVE_ON ──> Platform

    특히 주목할 부분은 MATCHED_WITH 관계의 속성을 필터 조건으로 쓰는 부분(m.status = '완료' AND m.rating >= 4.0)이다. 관계가 단순한 연결이 아니라 계약 상태와 평점을 직접 들고 있기 때문에 가능한 필터링이다.
    RDB에서라면 matching 테이블을 JOIN한 뒤 WHERE 절을 추가하는 방식이지만, 그래프에서는 경로를 탐색하면서 관계 속성으로 바로 거른다.
    결과적으로는 이런 데이터가 나온다.

    {
  "preferred_categories": ["뷰티", "라이프스타일"],
  "preferred_platforms": ["instagram", "youtube"],
  "avg_followers": 185000.0,
  "avg_engagement": 0.052,
  "avg_contract": 750000.0
}

    이 데이터를 AI Agent가 받아서 "이 광고주는 인스타그램 중심의 뷰티 인플루언서를 선호하고, 평균 계약금은 75만원대입니다"라는 형태로 해석해 다음 매칭 추천에 활용한다.

📌 RDB vs. 그래프 DB

RDB그래프 DB
관계 표현중간 테이블 + JOIN관계 자체가 경로
유사도 계산셀프 조인 + 집계공유 노드 수 카운트
다단계 탐색JOIN 중첩경로를 순서대로 기술
관계 속성 필터WHERE + JOIN관계 변수로 직접 접근

세 쿼리 모두 공통점이 있다. 쿼리가 데이터 탐색 경로를 그대로 서술한다. "광고주에서 출발해서, 그 광고주가 만든 캠페인을 찾고, 그 캠페인에 매칭된 인플루언서를 찾고, 그 인플루언서의 카테고리와 플랫폼을 수집한다." 이 문장이 Cypher 쿼리와 1:1로 대응된다.

🌟 LangGraph AI Agent에 Neo4j 도구 연결

🪄 구조 개요

이 프로젝트의 AI Agent는 사용자 입력을 받아 스스로 어떤 도구를 쓸지 판단하고, 도구 실행 결과를 바탕으로 다음 행동을 결정하는 LangGraph React Agent이다.
Neo4j는 단순한 데이터 저장소가 아니라 Agent가 직접 호출하는 도구(Tool)의 백엔드로 연결된다.

🛠️ @tool 데코레이터: LangChain 도구 등록

Neo4j 쿼리 함수를 Agent가 쓸 수 있는 도구로 만드는 방법은 간단하다. @tool 데코레이터를 붙이면 된다.

from langchain_core.tools import tool

@tool
def find_similar_influencers(influencer_id: str, limit: int = 5) -> str:
   """특정 인플루언서와 유사한 인플루언서를 찾습니다.
   
   같은 카테고리, 비슷한 플랫폼에서 활동하는 인플루언서를 추천합니다.
   Knowledge Graph 기반으로 관계를 분석하여 유사도를 계산합니다.
   
   Args:
      influencer_id: 기준 인플루언서의 UUID
      limit: 반환할 최대 인플루언서 수 (기본: 5)
   """
   from infrastructure.container import Container
   container = Container.get_instance()
   repo = container.neo4j_repository()
   results = repo.find_similar_influencers(UUID(influencer_id), limit)
   return json.dumps(results, ensure_ascii=False, indent=2)

@tool 데코레이터가 하는 일이 두 가지이다.
첫째, 함수를 LangChain이 인식할 수 있는 Tool 객체로 변환한다.
둘째, docstring을 그대로 읽어서 LLM에게 도구 설명으로 전달한다. LLM은 이 설명을 보고 "유사 인플루언서가 필요한 상황이면 이 도구를 써야겠다"라고 판단한다. docstring이 곧 도구의 사용 설명서인 셈이다.

🛠️ 7개 NEO4J_TOOLS

# adapters/outbound/agent/tools/neo4j_tools.py
NEO4J_TOOLS = [
   find_similar_influencers,    # 유사 인플루언서 추천
   get_influencer_ad_history,   # 인플루언서 과거 광고 이력 조회
   find_best_match_for_ad,      # 광고 캠페인에 맞는 인플루언서 추천
   analyze_advertiser_pattern,  # 광고주 선호 패턴 분석
   get_all_advertisers,         # 전체 광고주 목록 조회
   get_active_advertisements,   # 진행 중인 캠페인 목록 조회
   query_knowledge_graph,       # 자유 Cypher 쿼리 실행
]

앞 6개는 자주 쓰이는 탐색 패턴을 미리 함수로 감싼 것이다. 마지막 query_knowledge_graph는 성격이 다르다.

🛠️ query_knowledge_graph: Agent가 직접 Cypher를 작성

가장 강력하면서도 위험할 수 있는 도구이다. LLM이 상황에 따라 Cypher 쿼리를 직접 작성해서 실행한다.

@tool
def query_knowledge_graph(cypher_query: str) -> str:
   """Knowledge Graph에 직접 Cypher 쿼리를 실행합니다.
   복잡한 관계 탐색이나 분석이 필요할 때 사용합니다.
   주의: 읽기 전용 쿼리만 사용해야 합니다 (MATCH, RETURN만 사용).
   """
   # 쓰기 키워드 차단
   forbidden = ["CREATE", "DELETE", "SET", "REMOVE", "MERGE", "DROP"]
   query_upper = cypher_query.upper()
   for word in forbidden:
      if word in query_upper:
         return f"오류: 쓰기 쿼리는 허용되지 않습니다. ({word} 감지됨)"
         
   container = Container.get_instance()
   db = container.neo4j_database
   results = db.execute_read(cypher_query)  # 읽기 전용 트랜잭션
   return json.dumps(results, ensure_ascii=False, indent=2, default=str)

데이터 변조를 막기 위해 이중 안전장치를 걸었다.
첫째, Python 레벨에서 CREATE, DELETE, SET, MERGE 등의 키워드가 포함된 쿼리는 실행 전에 차단한다.
둘째, 통과하더라도 execute_read()로 실행하기 때문에 Neo4j 서버 레벨에서도 쓰기를 거부한다. 두 단계를 모두 통과해야 쿼리가 실행된다.
미리 만들어둔 6개 도구로 해결이 안 되는 복잡한 탐색이 필요할 때 Agent가 이 도구를 꺼낸다.

🛠️ Agent에 도구 등록

AgentFactory에서 LLM과 도구를 조합해 Agent를 만든다.

# adapters/outbound/agent/agent_factory.py
tools = [
   # Neo4j Repository 기반 기본 검색 도구
   create_search_by_category_tool(influencer_repo),
   create_search_by_platform_tool(influencer_repo),
   create_search_by_followers_tool(influencer_repo),
   create_get_influencer_details_tool(influencer_repo),
   create_calculate_match_score_tool(influencer_repo),
   
   # Neo4j Knowledge Graph 도구
   *NEO4J_TOOLS,
   
   # LlamaIndex 시맨틱 검색 도구
   *LLAMAINDEX_TOOLS,
   
   # DuckDuckGo 웹 검색 도구
   *REPUTATION_TOOLS,
]

agent = create_agent(
   model=llm,
   tools=tools,
   system_prompt=AGENT_SYSTEM_PROMPT,
)

*NEO4J_TOOLS로 리스트를 펼쳐서 다른 도구들과 함께 등록한다.
Agent는 이 전체 도구 목록을 받아 상황에 따라 어떤 도구를 쓸지 스스로 판단한다. Neo4j 그래프 탐색이 필요하면 NEO4J_TOOLS를, 자연어 기반 의미 검색이 필요하면 LLAMAINDEX_TOOLS를, 인플루언서의 최근 화제성을 확인해야 하면 REPUTATION_TOOLS를 선택한다.

📌 전체 흐름 정리

Neo4j가 단순한 데이터 저장소가 아니라 Agent의 추론 과정에 직접 참여하는 지식 그래프로 동작하는 구조이다.

🌟 LlamaIndex + Neo4j 지식 그래프 인덱싱

🤔 Neo4j 도구와 무엇이 다른가?

NEO4J_TOOLS는 Cypher 쿼리 기반이다. "카테고리가 뷰티인 인플루언서를 찾아라" 같은 조건이 명확할 때 강력하다. 반면 "친환경 느낌의 MZ 타겟 광고에 어울리는 인플루언서"처럼 조건이 모호하고 의미 기반인 질문은 Cypher로 표현하기 어렵다.
LLAMAINDEX_TOOLS는 이 간극을 채운다. Neo4j 그래프 데이터를 자연어 문서로 변환해서 인덱싱하고, 자연어 질문으로 그래프를 탐색할 수 있게 한다.

  • NEO4J_TOOLS → 조건이 명확할 때 (카테고리, 팔로워 범위,
    예산)
  • LLAMAINDEX_TOOLS → 의미 기반 탐색이 필요할 때 (컨셉, 분위기, 과거 성공 사례)

🛠️ KnowledgeGraphQueryIndex: 핵심 클래스

# adapters/outbound/persistence/llamaindex/kg_index.py
class KnowledgeGraphQueryIndex:
   """Neo4j Knowledge Graph를 LlamaIndex로 래핑하여 자연어 질의를 지원"""
   
   def __init__(self, neo4j_uri: str, neo4j_username: str, neo4j_password: str):
      self._graph_store = Neo4jGraphStore(
         username=neo4j_username,
         password=neo4j_password,
         url=neo4j_uri,
      )
      self._index: Optional[KnowledgeGraphIndex] = None

Neo4jGraphStore는 LlamaIndex가 Neo4j를 백엔드로 사용할 수 있게 해주는 연결 객체이다. 이 위에 KnowledgeGraphIndex를 올려서 자연어 질의 엔진을 만든다.

🛠️ 그래프 데이터 → Document 반환

LlamaIndex가 Neo4j 데이터를 이해하려면 그래프 노드와 관계를 자연어 문장(Document)으로 변환해야 한다.
_load_graph_documents()가 이 역할을 한다.
두 종류의 데이터를 각각 문장으로 만든다.

  1. [쿼리 1] 과거 광고 이력 데이터 (PROMOTED 관계)
    results = session.run("""
   MATCH (adv:Advertiser)-[:CREATED_AD]->(ad:Advertisement)
   OPTIONAL MATCH (i:Influencer)-[:PROMOTED]->(ad)
   OPTIONAL MATCH (i)-[:SPECIALIZES_IN]->(c:Category)
   OPTIONAL MATCH (i)-[:ACTIVE_ON]->(p:Platform)
   WITH adv, ad,
        collect(DISTINCT i.name) as promoters,
        collect(DISTINCT c.name) as categories,
        collect(DISTINCT p.name) as platforms,
        max(i.follower_count) as follower_count,
        max(i.engagement_rate) as engagement_rate
   RETURN adv.name as advertiser, adv.industry as industry,
          ad.product as product, ad.product_category as product_category,
          ad.concept_keywords as concept_keywords,
          ad.likes as likes, ad.comments as comments, ad.views as views,
          promoters, categories, platforms, follower_count, engagement_rate
""")
    이 결과를 아래처럼 자연어 문장으로 조립한다.
    doc = Document(
   text=f"광고주 '{data['advertiser']}'(업종: {data['industry']})이 "
        f"'{data['product']}' 제품 광고를 생성함. "
        f"카테고리: {data['product_category']}, "
        f"컨셉: {data['concept_keywords']}. "
        f"성과: 좋아요 {data['likes']}, 댓글 {data['comments']}, 조회수 {data['views']}. "
        f"프로모션 참여 인플루언서: {promoter_names}(카테고리: ..., 팔로워: ...).",
    metadata={"node_type": "advertisement", "advertiser": data["advertiser"]},
)
  1. [쿼리 2] 플랫폼 매칭 데이터 (MATCHED_WITH 관계)
    results = session.run("""
   MATCH (i:Influencer)-[m:MATCHED_WITH]->(ad:Advertisement)
   MATCH (adv:Advertiser)-[:CREATED_AD]->(ad)
   OPTIONAL MATCH (ad)-[:BELONGS_TO]->(c:Category)
   OPTIONAL MATCH (ad)-[:TARGETS_PLATFORM]->(p:Platform)
   WITH i, adv, ad, m,
        collect(DISTINCT c.name) as categories,
        collect(DISTINCT p.name) as platforms
   RETURN i.name as influencer, adv.name as advertiser,
          ad.title as title, ad.concept as concept,
          ad.item as item, ad.budget as budget,
          m.status as status,
          m.contract_amount as contract_amount,
          m.rating as rating,
          categories, platforms
""")
    이 결과를 아래처럼 자연어 문장으로 조립한다.
    doc = Document(
   text=f"광고주 '{data['advertiser']}'이 '{data['title']}' 캠페인을 등록함. "
        f"컨셉: {data['concept']}, 아이템: {data['item']}, 예산: {data['budget']}원. "
        f"타겟 카테고리: {categories_text}, 타겟 플랫폼: {platforms_text}. "
        f"인플루언서: '{data['influencer']}'과 매칭됨 "
        f"(상태: {data['status']}, 계약금: {data['contract_amount']}원). 평점: {data['rating']}점. "
    metadata={"node_type": "matching", "advertiser": ..., "influencer": ...},
)

그래프의 노드·관계가 사람이 읽을 수 있는 문장으로 바뀐다. LlamaIndex는 이 문장들을 임베딩 벡터로 변환해서 인덱싱한다.

🛠️ 인덱스 빌드: build_index vs. load_existing

인덱스를 만드는 방법이 두 가지이다.

  1. Document를 새로 만들고 LLM이 트리플 추출 (시간 소요)

    def build_index(self) -> KnowledgeGraphIndex:
   documents = self._load_graph_documents()
   storage_context = StorageContext.from_defaults(graph_store=self._graph_store)
   self._index = KnowledgeGraphIndex.from_documents(
      documents,
      storage_context=storage_context,
      include_embeddings=True,
      max_triplets_per_chunk=10,  # 청크당 최대 트리플 수
   )
   return self._index

    build_index()는 Document를 LLM이 분석해서 주어-관계-목적어 형태의 트리플을 추출하고 Neo4j에 저장한다. 처음 한 번만 실행하면 되고 시간이 걸린다.

  2. 이미 Neo4j에 저장된 트리플을 그대로 로드 (빠름)

    def load_existing(self) -> KnowledgeGraphIndex:
   storage_context = StorageContext.from_defaults(graph_store=self._graph_store)
   self._index = KnowledgeGraphIndex(
      nodes=[],
      storage_context=storage_context,
      include_embeddings=True,
   )
   return self._index

    load_existing()은 이미 Neo4j에 저장된 트리플을 그대로 불러온다. 앱이 재시작될 때마다 build_index()를 다시 돌릴 필요 없이 기존 데이터를 빠르게 로드한다.

실제로는 get_index()가 두 방법을 자동으로 선택한다.

def get_index(self) -> KnowledgeGraphIndex:
   if self._index is None:
      self.build_index()  # 인덱스가 없으면 빌드
   return self._index

🛠️ 자연어 질의

인덱스가 준비되면 자연어로 그래프를 탐색할 수 있다.

def query(self, query_text: str) -> str:
   index = self.get_index()
   query_engine = index.as_query_engine(
      include_text=True,
      response_mode="tree_summarize",  # 여러 결과를 트리 구조로 요약
   )
   response = query_engine.query(query_text)
   return str(response)

tree_summarize는 여러 노드에서 찾은 결과를 계층적으로 요약해서 하나의 자연어 답변으로 만들어주는 모드이다. Agent가 이 도구를 쓰면 이런 질문이 가능해진다.

"삼성전자가 과거에 협업했던 인플루언서는 누구야?"
"뷰티 카테고리에서 좋아요가 가장 많았던 캠페인은?"
"패션 광고에서 조회수가 높았던 인플루언서의 특징은?"

🛠️ Agent 도구로 연결

KnowledgeGraphQueryIndex는 LLAMAINDEX_TOOLS 중 query_kg_natural_language를 통해 Agent에 연결된다.

@tool
def query_kg_natural_language(query: str) -> str:
   """Knowledge Graph를 자연어로 탐색합니다. Cypher 쿼리를 작성할 필요 없이
   자연어 질문을 입력하세요.
   
   예: "삼성전자가 과거에 협업했던 인플루언서는?",
       "뷰티 카테고리 인플루언서의 과거 광고 캠페인 성과는?"
   """
   container = Container.get_instance()
   kg_index = container.kg_query_index()
   result = kg_index.query(query)  # 자연어 → 그래프 탐색 → 자연어 응답
   return result

🛠️ Neo4j가 두 가지 역할을 동시에

결국 이 프로젝트에서 Neo4j는 두 가지 역할을 동시에 수행한다.Cypher로 해결하기 어려운 의미 기반 탐색은 LlamaIndex가, 조건이 명확한 필터링은 Cypher가 처리한다.
Agent는 상황에 따라 두 도구 중 적합한 것을 선택해서 사용한다.

🌟 마무리

⚠️ Neo4j Aura Free 티어 제한

Aura Free는 빠르게 시작하기에 좋지만 몇 가지 제한을 알고 들어가야 한다.

스토리지 1GB 제한이 가장 크다. 인플루언서 50명, 광고주·캠페인 데이터, 매칭 이력 정도는 여유롭게 들어가지만, 인스타그램 게시물 데이터를 대량으로 수집해서 AdPost 노드로 쌓기 시작하면 금방 한계에 다다른다.

또, 일정 기간 미사용 시 인스턴스가 일시 정지된다. 팀 프로젝트에서 며칠 손을 놓고 있다가 돌아왔을 때 인스턴스가 꺼져있는 경험을 했다. Aura 콘솔에 들어가서 재시작하면 되지만, 자동화된 서비스라면 주의해야 할 부분이다.

즉, 무료 티어는 프로토타입이나 학습 용도라면 충분하지만, 실제 서비스라면 AuraDB Professional로 업그레이드 해야 한다.

😃 그래프 DB가 어울리는 상황

이 프로젝트를 통해 그래프 DB가 특히 강점을 발휘하는 상황을 직접 체감했다.

첫째, 다대다 관계가 여러 개 겹치는 경우이다. 인플루언서-카테고리, 인플루언서-플랫폼, 광고-카테고리, 광고-플랫폼이 전부 다대다이다. RDB였다면 중간 테이블이 4개 생기고 JOIN이 쌓였을 구조가, 그래프에서는 관계 자체로 자연스럽게 표현됐다.

둘째, 관계 자체에 데이터가 붙는 경우이다. MATCHED_WITH 관계에 계약금, 상태, 평점이 붙는 것처럼, 연결 자체가 의미 있는 데이터를 들고 있을 때 그래프 모델이 가장 깔끔하다. RDB에서는 이걸 별도 테이블로 분리해야 한다.

셋째, 탐색 깊이가 가변적인 경우이다. "광고주 → 캠페인 → 매칭된 인플루언서 → 카테고리"처럼 몇 단계를 거쳐가는 탐색이 핵심인 도메인에서 Cypher는 경로를 그대로 서술하는 방식으로 쿼리를 할 수 있어서 직관적이다.

☹️ 그래프 DB가 어울리지 않는 상황

그래프 DB가 불필요한 부분도 있었다.

첫째, 단순한 CRUD와 집계가 전부인 경우이다. 예를 들어 단순히 "인플루언서 목록을 페이지네이션해서 보여줘" 같은 쿼리는 RDB가 더 직관적이다. 그래프의 강점인 관계 탐색이 필요 없는 상황에서는 Cypher가 오히려 번거롭게 느껴질 수 있다.

둘째, 집계·통계 쿼리를 사용하는 경우이다. GROUP BY, HAVING, 윈도우 함수 같은 SQL의 강력한 집계 기능에 익숙하다면, Cypher의 WITH 기반 집계는 처음에 낯설다.

profile
jiny
이전 포스트

[TIL-0328] LangGraph + LlamaIndex: 이중 LLM 아키텍처 설계

다음 포스트

[TIL-0411] 임베딩 모델에 접두어를 붙이면 검색 성능이 올라갈까?

0개의 댓글