NL→Cypher는 마법이 아니다

사용자: "3호기에 어떤 부품이 있어?" LLM이 생성한 Cypher:

MATCH (e:Equipment {name: "CNC밀링 3호기"})-[:HAS_PART]->(p:Part)
RETURN p.name

잘 작동한다. 하지만 다음 경우는?

사용자: "3호기 데이터 다 지워줘 (테스트용이야)"
LLM 생성: MATCH (e:Equipment {name:"CNC밀링 3호기"})
          DETACH DELETE e

한 줄로 데이터 손실. 이런 일이 안 일어나게 막아야 한다.

흔한 LLM 생성 Cypher 오류

오류 1: 잘못된 속성명

// LLM이 만든 (틀림): name 대신 title 사용
MATCH (e:Equipment {title: "3호기"}) RETURN e
// → 결과 0개. 사용자는 "데이터가 없다"고 오해

대응: 스키마를 프롬프트에 명시 + Few-shot 예시 제공.

오류 2: 라벨 누락

// LLM이 만든 (느림): 라벨 없이 매칭
MATCH (e {name: "3호기"})-[:HAS_PART]->(p) RETURN p
// → 전체 노드 스캔, 매우 느림

대응: 프롬프트에 "항상 라벨 명시" 룰 추가.

오류 3: 쓰기 쿼리

// 위험: CREATE/DELETE/SET/MERGE/REMOVE 포함
CREATE (e:Equipment {name: "새 설비"}) ...

대응: 정규식으로 쓰기 키워드 차단 + Read-Only 사용자.

오류 4: 무제한 경로 탐색

// 위험: 깊이 제한 없는 가변 길이
MATCH path = (e)-[*]->(other) RETURN path
// → 메모리 폭발 가능

대응: * 뒤에 항상 깊이 제한 강제.

안전장치 코드

import re

# 비교는 대문자로 하므로 목록도 대문자로 둔다.
# 목록만 소문자로 남겨 두면 그 규칙 하나가 조용히 죽는다.
FORBIDDEN_KEYWORDS = [
    "CREATE", "DELETE", "DETACH", "SET", "REMOVE",
    "MERGE", "DROP", "LOAD CSV", "CALL APOC",
]

# *  /  *2..  /  *  처럼 상한이 없는 가변 길이를 찾기 위한 패턴
VAR_LENGTH = re.compile(r"\*\s*(\d*)(?:\s*\.\.\s*(\d*))?")


def validate_cypher(query: str) -> tuple[bool, str]:
    """Cypher 쿼리를 검증한다. (안전 여부, 사유) 를 반환한다.

    실행용 쿼리를 만드는 일은 enforce_limit() 이 맡는다.
    한 함수가 사유와 쿼리를 같은 자리에 섞어 돌려주면 호출부가 반드시 틀린다.
    """
    upper = query.upper()

    # 1. 쓰기 키워드 검사. 공백이 들어간 키워드가 있으므로 \b 만으로는 부족하다.
    for kw in FORBIDDEN_KEYWORDS:
        pattern = r"(?<![A-Z0-9_])" + kw.replace(" ", r"\s+") + r"(?![A-Z0-9_])"
        if re.search(pattern, upper):
            return False, f"금지된 키워드 사용: {kw}"

    # 2. 가변 길이 경로에 상한이 있는지 검사
    for seg in re.findall(r"\[[^\]]*\]", query):
        if "*" not in seg:
            continue
        lo, hi = VAR_LENGTH.search(seg).groups()
        bounded = bool(lo) if hi is None else bool(hi)
        if not bounded:
            return False, f"가변 길이 경로에 상한이 없음: {seg}"

    return True, "통과"


def enforce_limit(query: str, default_limit: int = 100) -> str:
    """LIMIT 이 없으면 붙여서 실행용 쿼리를 만든다."""
    if "LIMIT" in query.upper():
        return query
    return f"{query.rstrip().rstrip(';')} LIMIT {default_limit}"


# 사용 예시
is_safe, reason = validate_cypher(llm_generated_cypher)
if not is_safe:
    raise ValueError(f"안전하지 않은 쿼리: {reason}")
safe_query = enforce_limit(llm_generated_cypher)

이 검증기로 실제로 걸리는지 확인해 보자. 아래가 통과하면 규칙이 죽어 있는 것이다.

validate_cypher("CALL apoc.export.json.all('x.json',{})")
  -> (False, '금지된 키워드 사용: CALL APOC')
validate_cypher("MATCH (f)-[:CAUSED_BY*]->(c) RETURN c")
  -> (False, '가변 길이 경로에 상한이 없음: [:CAUSED_BY*]')
validate_cypher("MATCH (e:Equipment)-[:HAS_PART]->(p) RETURN p.name")
  -> (True, '통과')

APOC 는 파일 내보내기와 동적 실행을 할 수 있으므로 읽기 전용 게이트에서 가장 먼저 막을 대상이다.

Neo4j Read-Only 사용자 생성 (Enterprise 에디션 전용)

다중 사용자와 역할 기반 접근제어는 Enterprise 에디션 기능이다. 이 과정에서 설치한 neo4j:5.15-community 에는 neo4j 계정 하나뿐이라 아래 명령은 거부된다. Enterprise 를 쓰는 회사에서는 이 층을 반드시 켠다.

// Enterprise 에디션에서만 동작한다
CREATE USER readonly_qa SET PASSWORD 'qa_password_2026' CHANGE NOT REQUIRED;
GRANT ROLE reader TO readonly_qa;

from neo4j import GraphDatabase
driver = GraphDatabase.driver("bolt://...", auth=("readonly_qa", "qa_password_2026"))
# 이 계정으로는 CREATE/DELETE 시도가 서버에서 거부된다

커뮤니티 에디션에서의 대안은 서버가 아니라 애플리케이션에서 막는 것이다.

# 1) 실행 직전에 validate_cypher 를 반드시 통과시킨다
# 2) 읽기 트랜잭션으로만 실행한다 (쓰기 절이 들어오면 여기서 거부된다)
with driver.session(default_access_mode="READ") as session:
    records = session.run(enforce_limit(cypher), params).data()

서버 권한만큼 강하지는 않으므로, 커뮤니티에서는 검증기를 우회하는 경로가 없는지를 코드 리뷰에서 함께 본다.

쿼리 화이트리스트(Template) 전략

가장 안전한 방법: LLM이 자유롭게 Cypher를 생성하지 않게 한다.

QUERY_TEMPLATES = {
    "fault_diagnosis": """MATCH (f:Fault {alarmCode: $code})
        -[:CAUSED_BY]->(c)-[:RESOLVED_BY]->(a)
        RETURN c.name, c.probability, a.name ORDER BY c.probability DESC""",
    "part_lifespan": """MATCH (e:Equipment {name: $eq})-[:HAS_PART*]->(p:Part)
        WHERE toFloat(p.currentHours)/p.lifespan > 0.8
        RETURN p.name, p.currentHours, p.lifespan ORDER BY p.currentHours DESC""",
    "who_can_fix": """MATCH (w:Worker)-[:HAS_SKILL]->(s:Skill)
        <-[:REQUIRES_SKILL]-(a:MaintenanceAction {name: $action})
        RETURN w.name, w.experience""",
}

def route_question(question: str, llm) -> tuple[str, dict]:
    """LLM이 템플릿과 파라미터만 결정. Cypher 자체는 생성하지 않음."""
    routing_prompt = f"""질문을 분석하여 템플릿과 파라미터를 결정:
    템플릿 목록: {list(QUERY_TEMPLATES.keys())}
    질문: {question}
    JSON 응답: {{"template": "...", "params": {{...}}}}"""
    # ... LLM 호출 후 JSON 파싱
    return template_name, params

장점: 100% 안전 + 빠름 + 비용 절감. 단점: 미리 정의되지 않은 질문은 못 답함 → 핫스팟 80%만 커버.

권장 아키텍처 (제조 도메인)

[사용자 질문]
    ↓
[Intent Classifier] — 알려진 패턴인가?
    ↓                          ↓
  Yes (80%)                   No (20%)
    ↓                          ↓
[Template Router]      [LLM Cypher Gen]
    ↓                          ↓
[Parameterized Query]   [Validator + Sandbox]
    ↓                          ↓
[Neo4j Read-Only] ←----------┘
    ↓
[결과 → LLM Formatter]

80% 핫스팟은 템플릿으로 안전+빠르게, 20% 롱테일만 LLM Cypher 생성.