3. 실습 ② — 분류 노드 (2차 LLM)

import config
from prompts import CLASSIFY_PROMPT
from nodes.generator import _llm
from langchain_core.output_parsers import StrOutputParser

_classify_chain = CLASSIFY_PROMPT | _llm | StrOutputParser()
VALID = {"greeting", "calc", "scope", "document"}

def classifier_node(state) -> dict:
    q = state.get("question", "")

    # ── 1차: 규칙 (비용 0) ──
    intent = _rule_classify(q)
    if intent:
        return {"intent": intent,
                "log": [f"분류(규칙): {intent}"]}

    # ── 2차: LLM ──
    if not config.USE_LLM_CLASSIFY:
        return {"intent": "document",
                "log": ["분류: 기본값 document"]}

    try:
        raw = _classify_chain.invoke({"question": q}).strip().lower()
        intent = raw if raw in VALID else "document"
    except Exception as e:
        intent = "document"       # 실패하면 안전한 기본값
        return {"intent": intent,
                "log": [f"분류 오류({type(e).__name__}) -> document"]}

    return {"intent": intent, "log": [f"분류(LLM): {intent}"]}

# ==============================================================
# 29차시. PPT 29차시_ConditionalEdge실습. 코드 폴더는 src/29
# PPT: 29차시 질문 분류
# 역할: Classifier Node. 규칙을 먼저 쓰고, 규칙이 못 가르면 LLM으로 intent를 정합니다.
# 읽는 키: question
# 쓰는 키: intent, log
# 그래프:
#   classifier_node
#     |-- 인사 단어, 20자 이하 --> greeting
#     |-- 숫자와 연산자만 ------> calc
#     |-- 날씨·주가 등 ---------> scope
#     |-- 규칙 실패 + LLM 끔 --> document
#     +-- 규칙 실패 + LLM 켬 --> greeting|calc|scope|document
# --------------------------------------------------------------
# 파이썬만 아는 분을 위한 말
# State는 노드들이 함께 쓰는 딕셔너리입니다. 함수는 state를 읽고, 바꿀 키만 return 합니다.
# return에 적지 않은 키는 이전 값이 남습니다.
# log와 tried_queries는 리스트라서, 새로 return한 항목이 뒤에 이어 붙습니다.
# Retriever: 검색 담당. 질문과 비슷한 문서 조각만 고릅니다. 답 문장은 만들지 않습니다.
# Generator: 고른 문서를 근거로 답 문장을 만듭니다. 문서가 없으면 안내 문장만 돌려줍니다.
# Verifier: 답이 근거와 맞는지 검사하고 grade를 적습니다.
#   grade=pass     답을 확정하고 그래프를 끝냅니다.
#   grade=retry    같은 문서로 답 문장만 다시 씁니다. retries가 1 늘어납니다.
#   grade=research 문서가 부족합니다. 검색어를 바꾼 뒤 다시 찾습니다. rewrites가 늘어납니다.
#   grade=giveup   더 진행하지 않고 fallback으로 안내 문장만 돌려줍니다.
# fallback: 실패했을 때 답을 지어내지 않고 안내 문장을 반환하는 노드입니다.
# intent: 질문 종류. greeting 인사, calc 계산, scope 문서 밖, document 문서 검색.
# rewrite: 사용자가 친 question은 유지하고, 검색에 쓰는 query만 바꿉니다.
# safe_node: 노드를 try로 감쌉니다. 오류가 나도 그래프는 멈추지 않고 node_error에 적습니다.
# intent는 다음 칸을 고르는 표식입니다. 이 노드는 답을 쓰지 않습니다.
# 인사 목록에 '안녕'이 있고, 20자 이하만 인사로 봅니다. 긴 문장 속 단어는 인사로 치지 않습니다.
# CALC_PATTERN은 숫자와 +-*/(). 만 허용합니다. 연산자가 실제로 있어야 calc입니다.
# OUT_OF_SCOPE_WORDS에 '날씨'가 있어 날씨 질문은 scope입니다.
# LLM 결과가 네 단어 중 하나가 아니면 document로 바꿉니다. 호출 오류도 document입니다.
# config.USE_LLM_CLASSIFY가 True이면 규칙이 None일 때 LLM을 부릅니다.
# ==============================================================

# ============================================================
# Classifier Node
# ============================================================
# 사용자의 질문을 인사, 계산, 범위 밖 질문, 문서 질문으로 분류합니다.
# 먼저 규칙으로 판단하고, 판단하기 어려운 질문만 LLM에게 전달합니다.
# ============================================================

import os
import sys
import re

# ============================================================
# 다른 차시의 파일을 import하기 위한 경로 설정
# ============================================================

CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
SRC_DIR = os.path.abspath(os.path.join(CURRENT_DIR, ".."))

# 공통 src 폴더를 추가합니다.
sys.path.insert(0, SRC_DIR)

# 25차시 Generator 모듈 경로를 추가합니다.
sys.path.insert(0, os.path.join(SRC_DIR, "25"))

# 13차시 프롬프트 모듈 경로를 추가합니다.
sys.path.insert(0, os.path.join(SRC_DIR, "13"))

# 24차시 Retriever 모듈 경로를 추가합니다.
sys.path.insert(0, os.path.join(SRC_DIR, "24"))

# 다른 모듈을 가져옵니다.
import config

from generator import _llm
from prompts import CLASSIFY_PROMPT
from langchain_core.output_parsers import StrOutputParser

# ============================================================
# 분류 기준
# ============================================================

# 인사말로 판단할 단어 목록입니다.
GREETING_WORDS = [
    "안녕",
    "반가",
    "하이",
    "헬로",
    "고마",
    "감사",
    "수고",
    "잘 부탁",
]

# 문서와 관련이 없다고 판단할 단어 목록입니다.
OUT_OF_SCOPE_WORDS = [
    "날씨",
    "주가",
    "환율",
    "로또",
    "뉴스",
    "맛집",
    "영화",
    "축구",
]

# 숫자와 계산 기호로만 이루어진 질문인지 확인합니다.
CALC_PATTERN = re.compile(
    r"^[\d\s+\-*/().,]+[=?]?$"
)

# ============================================================
# 1차 분류: 규칙 기반 분류
# ============================================================

def rule_classify(question):
    """
    규칙만 사용하여 질문을 빠르게 분류합니다.

    반환값:
        greeting : 인사말
        calc     : 계산식
        scope    : 문서 범위 밖 질문
        None     : 규칙으로 판단하지 못함
    """

    # 질문 앞뒤의 공백을 제거합니다.
    question = question.strip()

    # 빈 질문은 문서와 관련 없는 질문으로 처리합니다.
    if question == "":
        return "scope"

    # --------------------------------------------------------
    # 1. 인사말인지 확인합니다.
    # --------------------------------------------------------

    # 긴 문장 전체를 인사말로 잘못 판단하지 않도록 제한합니다.
    if len(question) <= 20:

        # 인사말 단어를 하나씩 확인합니다.
        for word in GREETING_WORDS:

            # 질문에 인사말 단어가 포함되어 있으면 인사로 분류합니다.
            if word in question:
                return "greeting"

    # --------------------------------------------------------
    # 2. 계산식인지 확인합니다.
    # --------------------------------------------------------

    # 계산 연산자가 있는지 저장하는 변수입니다.
    has_operator = False

    # 사칙연산 기호를 하나씩 확인합니다.
    for operator in "+-*/":

        # 질문에 연산자가 있으면 계산식 후보로 표시합니다.
        if operator in question:
            has_operator = True
            break

    # 계산식 모양이고 실제 연산자도 있으면 계산으로 분류합니다.
    if CALC_PATTERN.match(question) and has_operator:
        return "calc"

    # --------------------------------------------------------
    # 3. 문서 범위 밖 질문인지 확인합니다.
    # --------------------------------------------------------

    # 범위 밖 단어를 하나씩 확인합니다.
    for word in OUT_OF_SCOPE_WORDS:

        # 질문에 범위 밖 단어가 포함되어 있으면 scope로 분류합니다.
        if word in question:
            return "scope"

    # 규칙으로 판단하지 못한 질문입니다.
    return None

# ============================================================
# 2차 분류: LLM 기반 분류
# ============================================================

# 프롬프트, LLM, 문자열 변환기를 연결합니다.
classify_chain = (
    CLASSIFY_PROMPT
    | _llm
    | StrOutputParser()
)

# LLM이 반환할 수 있는 정상적인 분류 결과입니다.
VALID_INTENTS = [
    "greeting",
    "calc",
    "scope",
    "document",
]

# ============================================================
# LangGraph 분류 노드
# ============================================================

def classifier_node(state):
    """
    질문을 분류하고 intent 값을 State에 저장합니다.

    처리 순서:
        1. 규칙으로 먼저 판단합니다.
        2. 판단하지 못하면 LLM 사용 여부를 확인합니다.
        3. LLM을 사용하지 않으면 document로 처리합니다.
        4. LLM 오류가 발생해도 document로 처리합니다.
    """

    # State에서 질문을 가져옵니다.
    question = state.get("question", "")

    # --------------------------------------------------------
    # 1단계: 규칙 기반 분류
    # --------------------------------------------------------

    # 비용이 들지 않는 규칙 분류를 먼저 실행합니다.
    intent = rule_classify(question)

    # 규칙으로 판단했다면 즉시 결과를 반환합니다.
    if intent is not None:
        return {
            "intent": intent,
            "log": [f"분류(규칙): {intent}"],
        }

    # --------------------------------------------------------
    # 2단계: LLM 분류 사용 여부 확인
    # --------------------------------------------------------

    # 설정값이 없으면 기본적으로 LLM을 사용하지 않습니다.
    use_llm = getattr(config, "USE_LLM_CLASSIFY", False)

    # LLM 분류를 사용하지 않으면 문서 질문으로 처리합니다.
    if use_llm == False:
        return {
            "intent": "document",
            "log": ["분류: 기본값 document"],
        }

    # --------------------------------------------------------
    # 3단계: LLM으로 분류
    # --------------------------------------------------------

    try:
        # LLM에게 질문을 전달합니다.
        result = classify_chain.invoke({
            "question": question,
        })

        # LLM의 결과를 소문자로 바꾸고 공백을 제거합니다.
        intent = result.strip().lower()

        # 허용되지 않은 결과는 document로 바꿉니다.
        if intent not in VALID_INTENTS:
            intent = "document"

        # LLM 분류 결과를 기록합니다.
        log_message = f"분류(LLM): {intent}"

    except Exception as error:
        # LLM 오류가 발생하면 안전하게 document로 처리합니다.
        intent = "document"

        # 오류 종류만 로그에 기록합니다.
        error_name = type(error).__name__
        log_message = f"분류 오류({error_name}) -> document"

    # 최종 분류 결과를 State에 저장합니다.
    return {
        "intent": intent,
        "log": [log_message],
    }

# ============================================================
# 단독 실행 테스트
# ============================================================

if __name__ == "__main__":

    # 테스트할 질문 목록입니다.
    test_questions = [
        "안녕하세요",
        "125 + 340 * 2",
        "오늘 날씨 어때요?",
        "환불은 며칠 이내인가요?",
    ]

    # 질문을 하나씩 테스트합니다.
    for question in test_questions:

        # 테스트용 State를 직접 만듭니다.
        state = {
            "question": question
        }

        # 분류 노드를 실행합니다.
        result = classifier_node(state)

        # 결과를 출력합니다.
        print(f"\n질문: {question}")
        print(f"분류: {result['intent']}")
        print(f"로그: {result['log']}")