Neo4j GraphRag 실행 예정 코드 정리

JERRY·2025년 12월 2일

Project

목록 보기
7/14

1. APOC/GDS 확인

RETURN apoc.version();
CALL gds.version();

2. 데이터베이스 생성

:use system
SHOW DATABASES;
CREATE DATABASE auditdb IF NOT EXISTS;
SHOW DATABASE auditdb;

3. Neo4j 세팅 코드 (auditdb에서 실행)

1) DB 전환

:use auditdb
RETURN "ok" AS status;

2) 필수 제약조건 4개 생성 (중복 적재 방지/auditdb에서 1회 실행)

:use auditdb

// Document
CREATE CONSTRAINT document_doc_code_unique IF NOT EXISTS
FOR (d:Document) REQUIRE d.doc_code IS UNIQUE;

// Case
CREATE CONSTRAINT case_doc_sub_unique IF NOT EXISTS
FOR (cs:Case) REQUIRE (cs.doc_code, cs.sub_order) IS UNIQUE;

// CaseHeader
CREATE CONSTRAINT caseheader_subcode_unique IF NOT EXISTS
FOR (h:CaseHeader) REQUIRE h.sub_code IS UNIQUE;

// Chunk
CREATE CONSTRAINT chunk_subcode_chunkid_unique IF NOT EXISTS
FOR (c:Chunk) REQUIRE (c.sub_code, c.chunk_id) IS UNIQUE;

# 확인
SHOW CONSTRAINTS;

3) v1용 Fulltext 인덱스(먼저 만들어두면 폴백이 됨(권장, 1회))

:use auditdb
CREATE FULLTEXT INDEX chunk_text_ft IF NOT EXISTS
FOR (c:Chunk) ON EACH [c.text, c.sub_title, c.problem, c.action, c.criteria];

# 확인 
SHOW INDEXES;

4) v2용 Vector 인덱스(임베딩 dim 확정 후, 1회)

:use auditdb
CREATE VECTOR INDEX caseheader_embedding_idx IF NOT EXISTS
FOR (h:CaseHeader) ON (h.embedding)
OPTIONS {indexConfig: {'vector.dimensions': 1536, 'vector.similarity_function': 'cosine'}};

4. 적재(ingestion) 규격: n8n이 Neo4j에 넣는 MERGE 템플릿

n8n은 JSON을 만들고 아래 MERGE 쿼리로 넣으면 됩니다(파라미터 방식 권장)

1) Document upsert

:use auditdb
WITH $doc AS doc
MERGE (d:Document {doc_code: doc.doc_code})
SET d.title            = doc.title,
    d.purpose          = doc.purpose,
    d.site             = doc.site,
    d.category         = doc.category,
    d.audit_start_date = CASE WHEN doc.audit_start_date IS NULL OR doc.audit_start_date = "" THEN NULL ELSE date(doc.audit_start_date) END,
    d.audit_end_date   = CASE WHEN doc.audit_end_date   IS NULL OR doc.audit_end_date   = "" THEN NULL ELSE date(doc.audit_end_date) END,
    d.date             = CASE WHEN doc.date             IS NULL OR doc.date             = "" THEN NULL ELSE date(doc.date) END,
    d.download_url     = doc.download_url,
    d.save_place       = doc.save_place,
    d.file_name        = doc.file_name,
    d.file_hash        = doc.file_hash,
    d.created_at       = coalesce(d.created_at, datetime()),
    d.updated_at       = datetime()

WITH d,
     // category(list) -> "A|B|C" 문자열로 join (APOC 없이)
     reduce(s = "", x IN coalesce(d.category, []) |
       CASE WHEN s = "" THEN toString(x) ELSE s + "|" + toString(x) END
     ) AS category_joined,
     $cases AS cases
UNWIND cases AS cs

2) ChunkHeader upsert + Document 연결

:use auditdb
MERGE (case:Case {doc_code: d.doc_code, sub_order: cs.sub_order})
SET case.sub_code      = coalesce(cs.sub_code, d.doc_code + "-" + toString(cs.sub_order)),
    case.sub_title     = cs.sub_title,
    case.keyword_list  = cs.keyword_list,
    case.problem       = cs.problem,
    case.action        = cs.action,
    case.action_type   = cs.action_type,
    case.fiscal_amount = cs.fiscal_amount,
    case.opinion       = cs.opinion,
    case.criteria      = cs.criteria,
    case.related_laws  = cs.related_laws,
    case.updated_at    = datetime(),
    case.created_at    = coalesce(case.created_at, datetime())
MERGE (d)-[:HAS_CASE]->(case)

WITH d, case, cs, category_joined

3) CaseHeader upsert + Case 연결

:use auditdb
MERGE (h:CaseHeader {sub_code: case.sub_code})
SET h.doc_code         = d.doc_code,
    h.sub_order        = cs.sub_order,
    h.title            = d.title,
    // (교체) 첫 원소만 쓰지 않고, 전체를 "A|B|C"로 저장
    h.category         = CASE WHEN category_joined = "" THEN NULL ELSE category_joined END,
    h.keyword_list     = cs.keyword_list,
    h.action_type      = cs.action_type,
    h.sub_title        = cs.sub_title,
    h.header_text      = cs.header.header_text,
    h.embedding        = cs.header.embedding,
    h.embedding_model  = cs.header.embedding_model,
    h.embedding_dim    = cs.header.embedding_dim,
    h.embedded_at      = CASE WHEN cs.header.embedded_at IS NULL OR cs.header.embedded_at = "" THEN NULL ELSE datetime(cs.header.embedded_at) END,
    h.updated_at       = datetime(),
    h.created_at       = coalesce(h.created_at, datetime())
MERGE (case)-[:HAS_HEADER]->(h)

WITH case, cs

4) Chunk upsert + Case 연결 (사례당 N개)

:use auditdb
UNWIND coalesce(cs.chunks, []) AS ck
MERGE (c:Chunk {sub_code: case.sub_code, chunk_id: toString(ck.chunk_id)})
SET c.doc_code       = case.doc_code,
    c.sub_order      = case.sub_order,
    c.sub_title      = cs.sub_title,
    c.seq            = ck.seq,
    c.text           = ck.text,
    c.token_count    = ck.token_count,
    c.keyword_list   = cs.keyword_list,
    c.problem        = cs.problem,
    c.action         = cs.action,
    c.action_type    = cs.action_type,
    c.fiscal_amount  = cs.fiscal_amount,
    c.opinion        = cs.opinion,
    c.criteria       = cs.criteria,
    c.related_laws   = cs.related_laws,
    c.updated_at     = datetime(),
    c.created_at     = coalesce(c.created_at, datetime())
MERGE (case)-[:HAS_CHUNK]->(c);

배치 적재 MERGE 템플릿 (교체본, category join 반영)

:use auditdb

// 1) Document upsert
WITH $doc AS doc
MERGE (d:Document {doc_code: doc.doc_code})
SET d.title            = doc.title,
    d.purpose          = doc.purpose,
    d.site             = doc.site,
    d.category         = doc.category,
    d.audit_start_date = CASE WHEN doc.audit_start_date IS NULL OR doc.audit_start_date = "" THEN NULL ELSE date(doc.audit_start_date) END,
    d.audit_end_date   = CASE WHEN doc.audit_end_date   IS NULL OR doc.audit_end_date   = "" THEN NULL ELSE date(doc.audit_end_date) END,
    d.date             = CASE WHEN doc.date             IS NULL OR doc.date             = "" THEN NULL ELSE date(doc.date) END,
    d.download_url     = doc.download_url,
    d.save_place       = doc.save_place,
    d.file_name        = doc.file_name,
    d.file_hash        = doc.file_hash,
    d.created_at       = coalesce(d.created_at, datetime()),
    d.updated_at       = datetime()

WITH d,
     // category(list) -> "A|B|C" 문자열로 join (APOC 없이)
     reduce(s = "", x IN coalesce(d.category, []) |
       CASE WHEN s = "" THEN toString(x) ELSE s + "|" + toString(x) END
     ) AS category_joined,
     $cases AS cases
UNWIND cases AS cs

// 2) Case upsert + Document 연결
MERGE (case:Case {doc_code: d.doc_code, sub_order: cs.sub_order})
SET case.sub_code      = coalesce(cs.sub_code, d.doc_code + "-" + toString(cs.sub_order)),
    case.sub_title     = cs.sub_title,
    case.keyword_list  = cs.keyword_list,
    case.problem       = cs.problem,
    case.action        = cs.action,
    case.action_type   = cs.action_type,
    case.fiscal_amount = cs.fiscal_amount,
    case.opinion       = cs.opinion,
    case.criteria      = cs.criteria,
    case.related_laws  = cs.related_laws,
    case.updated_at    = datetime(),
    case.created_at    = coalesce(case.created_at, datetime())
MERGE (d)-[:HAS_CASE]->(case)

WITH d, case, cs, category_joined

// 3) CaseHeader upsert + Case 연결
MERGE (h:CaseHeader {sub_code: case.sub_code})
SET h.doc_code         = d.doc_code,
    h.sub_order        = cs.sub_order,
    h.title            = d.title,
    // (교체) 첫 원소만 쓰지 않고, 전체를 "A|B|C"로 저장
    h.category         = CASE WHEN category_joined = "" THEN NULL ELSE category_joined END,
    h.keyword_list     = cs.keyword_list,
    h.action_type      = cs.action_type,
    h.sub_title        = cs.sub_title,
    h.header_text      = cs.header.header_text,
    h.embedding        = cs.header.embedding,
    h.embedding_model  = cs.header.embedding_model,
    h.embedding_dim    = cs.header.embedding_dim,
    h.embedded_at      = CASE WHEN cs.header.embedded_at IS NULL OR cs.header.embedded_at = "" THEN NULL ELSE datetime(cs.header.embedded_at) END,
    h.updated_at       = datetime(),
    h.created_at       = coalesce(h.created_at, datetime())
MERGE (case)-[:HAS_HEADER]->(h)

WITH case, cs

// 4) Chunk upsert + Case 연결 (사례당 N개)
UNWIND coalesce(cs.chunks, []) AS ck
MERGE (c:Chunk {sub_code: case.sub_code, chunk_id: toString(ck.chunk_id)})
SET c.doc_code       = case.doc_code,
    c.sub_order      = case.sub_order,
    c.sub_title      = cs.sub_title,
    c.seq            = ck.seq,
    c.text           = ck.text,
    c.token_count    = ck.token_count,
    c.keyword_list   = cs.keyword_list,
    c.problem        = cs.problem,
    c.action         = cs.action,
    c.action_type    = cs.action_type,
    c.fiscal_amount  = cs.fiscal_amount,
    c.opinion        = cs.opinion,
    c.criteria       = cs.criteria,
    c.related_laws   = cs.related_laws,
    c.updated_at     = datetime(),
    c.created_at     = coalesce(c.created_at, datetime())
MERGE (case)-[:HAS_CHUNK]->(c);

5. Retrieval 쿼리 3종: v1 / v2 / hybrid

1) v1(Fulltext) 검색: 질문 문자열로 Chunk 후보 뽑기

:use auditdb
CALL db.index.fulltext.queryNodes('chunk_text_ft', $q) YIELD node, score
WITH node AS c, score
OPTIONAL MATCH (case:Case)-[:HAS_CHUNK]->(c)
OPTIONAL MATCH (d:Document {doc_code: c.doc_code})
RETURN
  c.doc_code       AS doc_code,
  case.sub_code    AS sub_code,
  c.sub_title      AS sub_title,
  c.chunk_id       AS chunk_id,
  c.seq            AS seq,
  c.text           AS text,
  score            AS score,
  d.title          AS doc_title,
  d.site           AS site,
  d.download_url   AS download_url
ORDER BY score DESC, c.seq ASC
LIMIT $k;

2) v2(Vector) 검색: 질문 임베딩으로 ChunkHeader Top-k → 연결 Chunk 가져오기

:use auditdb
CALL db.index.vector.queryNodes('caseheader_embedding_idx', $kHeader, $q_embedding)
YIELD node, score
WITH node AS h, score
MATCH (case:Case {doc_code:h.doc_code, sub_order:h.sub_order})-[:HAS_CHUNK]->(c:Chunk)
OPTIONAL MATCH (d:Document {doc_code:h.doc_code})
RETURN
  h.doc_code        AS doc_code,
  h.sub_code        AS sub_code,
  h.sub_title       AS sub_title,
  c.chunk_id        AS chunk_id,
  c.seq             AS seq,
  c.text            AS text,
  score             AS score,
  d.title           AS doc_title,
  d.site            AS site,
  d.download_url    AS download_url
ORDER BY score DESC, c.seq ASC
LIMIT $kChunk;

3) Hybrid(권장 실행 로직)

v2를 먼저 시도 → 결과가 비었거나 너무 적으면(v2 실패/임베딩 누락) v1으로 폴백


6. LangFlow: Neo4j Hybrid Retriever 커스텀 컴포넌트 (drop-in)

목표: 기존 플로우의 Qdrant “검색 노드” 자리에 넣고, DataFrame을 출력해서 ReRanker로 그대로 넘기기

1) 설치 필요(랭플로우 컨테이너/환경)

LangFlow 실행 환경에 neo4j 드라이버가 있어야 합니다.

  • neo4j Python package

2) 컴포넌트 코드

아래는 “질문(text) + (선택) 질문 임베딩(list)”을 받아서
임베딩이 있으면 v2 먼저, 없거나 결과 부족하면 v1으로 폴백, DataFrame으로 반환

from __future__ import annotations

from typing import List, Optional

import pandas as pd
from neo4j import GraphDatabase

from langflow.custom import Component
from langflow.io import (
    Output,
    MessageTextInput,
    StrInput,
    SecretStrInput,
    IntInput,
    BoolInput,
    DataInput,
)
from langflow.schema.dataframe import DataFrame
  

class Neo4jHybridRetrieverB(Component):
    display_name = "Neo4j Hybrid Retriever (B: Case+Chunk)"
    description = "Vector-first CaseHeader retrieval + Fulltext fallback Chunk retrieval. Returns DataFrame for reranker."
    icon = "database"
    name = "Neo4jHybridRetrieverB"

    inputs = [
        StrInput(name="uri", display_name="Neo4j URI", required=True, value="bolt://127.0.0.1:7687"),
        StrInput(name="user", display_name="Neo4j User", required=True, value="neo4j"),
        SecretStrInput(name="password", display_name="Neo4j Password", required=True),
        StrInput(name="database", display_name="Database", required=True, value="auditdb"),

        MessageTextInput(name="query", display_name="User Query", required=True),
        DataInput(name="query_embedding", display_name="Query Embedding (optional)", required=False),

        StrInput(name="vector_index", display_name="Vector Index Name", value="caseheader_embedding_idx"),
        StrInput(name="fulltext_index", display_name="Fulltext Index Name", value="chunk_text_ft"),

        IntInput(name="k_header", display_name="Top-K CaseHeaders (vector)", value=8),
        IntInput(name="k_chunk", display_name="Max rows returned (vector route)", value=24),
        IntInput(name="k_fulltext", display_name="Top-K rows (fulltext)", value=24),

        IntInput(name="min_results", display_name="Min results before fallback", value=3),
        BoolInput(name="enable_fulltext_fallback", display_name="Enable fulltext fallback", value=True),
    ]

    outputs = [
        Output(display_name="Docs (DataFrame)", name="docs", method="run"),
    ]

    def _run(self, driver, database: str, cypher: str, params: dict) -> List[dict]:
        with driver.session(database=database) as session:
            res = session.run(cypher, params)
            return [r.data() for r in res]

    def run(self) -> DataFrame:
        q = (self.query or "").strip()
        if not q:
            return DataFrame(pd.DataFrame([]))

        # normalize embedding input (LangFlow DataInput can be dict or list)
        q_embedding: Optional[list] = None
        if self.query_embedding is not None:
            if isinstance(self.query_embedding, dict) and "data" in self.query_embedding:
                q_embedding = self.query_embedding["data"]
            else:
                q_embedding = self.query_embedding

        if isinstance(q_embedding, list) and len(q_embedding) > 0:
            try:
                q_embedding = [float(x) for x in q_embedding]
            except Exception:
                q_embedding = None
        else:
            q_embedding = None

        cypher_vector = """
        CALL db.index.vector.queryNodes($vector_index, $k_header, $q_embedding)
        YIELD node, score
        WITH node AS h, score
        MATCH (case:Case {doc_code:h.doc_code, sub_order:h.sub_order})-[:HAS_CHUNK]->(c:Chunk)
        OPTIONAL MATCH (d:Document {doc_code:h.doc_code})
        RETURN
          h.doc_code AS doc_code,
          h.sub_code AS sub_code,
          h.sub_title AS sub_title,
          c.chunk_id AS chunk_id,
          c.seq AS seq,
          c.text AS text,
          score AS score,
          d.title AS doc_title,
          d.site AS site,
          d.download_url AS download_url
        ORDER BY score DESC, c.seq ASC
        LIMIT $k_chunk;
        """

        cypher_fulltext = """
        CALL db.index.fulltext.queryNodes($fulltext_index, $q) YIELD node, score
        WITH node AS c, score
        OPTIONAL MATCH (case:Case {doc_code:c.doc_code, sub_order:c.sub_order})
        OPTIONAL MATCH (d:Document {doc_code:c.doc_code})
        RETURN
          c.doc_code AS doc_code,
          case.sub_code AS sub_code,
          c.sub_title AS sub_title,
          c.chunk_id AS chunk_id,
          c.seq AS seq,
          c.text AS text,
          score AS score,
          d.title AS doc_title,
          d.site AS site,
          d.download_url AS download_url
        ORDER BY score DESC
        LIMIT $k_fulltext;
        """

        driver = GraphDatabase.driver(self.uri, auth=(self.user, self.password))
        try:
            rows: List[dict] = []

            # Vector-first route
            if q_embedding is not None:
                rows = self._run(
                    driver,
                    self.database,
                    cypher_vector,
                    {
                        "vector_index": self.vector_index,
                        "k_header": int(self.k_header),
                        "k_chunk": int(self.k_chunk),
                        "q_embedding": q_embedding,
                    },
                )

            # Fallback
            if (not rows or len(rows) < int(self.min_results)) and bool(self.enable_fulltext_fallback):
                rows = self._run(
                    driver,
                    self.database,
                    cypher_fulltext,
                    {
                        "fulltext_index": self.fulltext_index,
                        "q": q,
                        "k_fulltext": int(self.k_fulltext),
                    },
                )

            df = pd.DataFrame(rows)
            if not df.empty:
                # downstream compatibility
                df["text"] = df["text"].fillna("")
                for col in ["doc_code", "sub_code", "sub_title", "chunk_id", "doc_title", "site", "download_url"]:
                    if col in df.columns:
                        df[col] = df[col].fillna("")
                if "score" in df.columns:
                    df["score"] = df["score"].fillna(0.0)

            return DataFrame(df)

        finally:
            driver.close()

7. LangFlow 플로우 연결

MultiQuery + ReRanker를 넣어 flow에서 검색 노드만 교체

  • Chat Input
  • MultiQuery
  • (각 쿼리에 대해) Embedding 생성(필요 시) → Neo4jHybridRetriever
  • 결과를 합치고(중복 제거: sub_code 기준) ReRanker
  • Prompt Template
  • LLM
  • Chat Output
profile
JERRY
이전 포스트

Langflow - audit_rag_multi_reranker

다음 포스트

이번주 진행 사항 정리

0개의 댓글