NL→Cypher 변환의 함정과 안전장치

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", "SET", "REMOVE",
    "MERGE", "DROP", "CALL apoc.",
]

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

    # 1. 쓰기 키워드 검사
    for kw in FORBIDDEN_KEYWORDS:
        if re.search(rf"\b{kw}\b", upper):
            return False, f"금지된 키워드 사용: {kw}"

    # 2. 무제한 가변 길이 검사
    if re.search(r"\[\*\]", query) or re.search(r"\[\*\.\.[^0-9]", query):
        return False, "가변 길이 경로에 깊이 제한이 없음"

    # 3. LIMIT 강제
    if "LIMIT" not in upper:
        query += " LIMIT 100"

    return True, query

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

Neo4j Read-Only 사용자 생성

// 권한 분리: 읽기 전용 사용자
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 시도해도 실패

쿼리 화이트리스트(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 생성.