128 lines
4.8 KiB
Python
128 lines
4.8 KiB
Python
"""
|
|
app/services/speaker_classifier.py
|
|
-------------------------------------
|
|
LLM 기반 단일 분류 화자 판별 서비스.
|
|
입력: 무전 텍스트 1줄 (str)
|
|
출력: Literal["관제", "열차", "불명"] 중 하나
|
|
|
|
[설계 원칙]
|
|
- 1단계: dictionary.py의 CALLSIGNS_CONTROL/TRAIN 목록으로 빠른 문자열 매칭
|
|
- 2단계: 단 1번의 LLM 호출로 JSON {"speaker":"관제"|"열차"|"불명"} 수신
|
|
- 3단계: 파싱 실패·LLM 미활성·예외 → "불명" 반환 (절대 크래시 없음)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import logging
|
|
from typing import Literal
|
|
|
|
from app.core.dictionary import CALLSIGNS_CONTROL, CALLSIGNS_TRAIN
|
|
|
|
logger = logging.getLogger("uvicorn.error")
|
|
|
|
SpeakerLabel = Literal["관제", "열차", "불명"]
|
|
|
|
# ── 시스템 프롬프트: 반드시 JSON 단답만 허용 ─────────────────────────────────
|
|
_SYS_PROMPT = (
|
|
"당신은 철도 무전 화자 분류기입니다. "
|
|
"입력된 무전 문장이 '관제사(관제)'의 발언인지 '기관사/차량(열차)'의 발언인지 판별하세요. "
|
|
"반드시 JSON 형태로만 답변하세요: "
|
|
'{"speaker": "관제"} 또는 {"speaker": "열차"} 또는 {"speaker": "불명"} '
|
|
"다른 텍스트, 설명, 생각 과정은 절대 포함하지 마세요. /no_think"
|
|
)
|
|
|
|
|
|
def _heuristic(text: str) -> SpeakerLabel | None:
|
|
"""
|
|
dictionary.py의 호출 부호 목록으로 빠른 1차 분류.
|
|
확실하지 않으면 None 반환 → 2단계 LLM 호출.
|
|
"""
|
|
for sign in CALLSIGNS_CONTROL:
|
|
if sign in text:
|
|
logger.debug(f"[SPKCLS] 관제 휴리스틱 히트: '{sign}'")
|
|
return "관제"
|
|
for sign in CALLSIGNS_TRAIN:
|
|
if sign in text:
|
|
logger.debug(f"[SPKCLS] 열차 휴리스틱 히트: '{sign}'")
|
|
return "열차"
|
|
|
|
# 정규식 추가 패턴: "XXX열차" (3~4자리 숫자 + 열차)
|
|
if re.search(r"\d{3,4}\s*열차", text):
|
|
return "열차"
|
|
|
|
return None
|
|
|
|
|
|
def classify_speaker(text: str, context: str = "") -> SpeakerLabel:
|
|
"""
|
|
무전 텍스트 1줄을 받아 화자를 분류한다.
|
|
1단계: 휴리스틱 (dictionary 호출 부호 + 정규식)
|
|
2단계: LLM JSON 단답 추론 (max_tokens=32, temperature=0.0)
|
|
3단계: 파싱 실패 시 "불명" 반환 — 절대 크래시 없음
|
|
|
|
Args:
|
|
text: 분류 대상 무전 문장
|
|
context: 직전 맥락 문장 (선택, LLM 정확도 향상)
|
|
|
|
Returns:
|
|
"관제" | "열차" | "불명"
|
|
"""
|
|
if not text or not text.strip():
|
|
return "불명"
|
|
|
|
# ── 1단계: 빠른 휴리스틱 ──────────────────────────────────────────────────
|
|
result = _heuristic(text)
|
|
if result is not None:
|
|
return result
|
|
|
|
# ── 2단계: LLM JSON 단답 추론 ────────────────────────────────────────────
|
|
try:
|
|
from app.services.llm_service import llm_service
|
|
model = llm_service._model
|
|
if model is None:
|
|
logger.debug("[SPKCLS] LLM 미활성 → 불명 반환")
|
|
return "불명"
|
|
|
|
ctx_line = f"맥락: {context}\n" if context.strip() else ""
|
|
user_msg = f"{ctx_line}무전: {text.strip()}"
|
|
|
|
prompt = (
|
|
f"<|im_start|>system\n{_SYS_PROMPT}<|im_end|>\n"
|
|
f"<|im_start|>user\n{user_msg}<|im_end|>\n"
|
|
f"<|im_start|>assistant\n"
|
|
)
|
|
|
|
out = model(
|
|
prompt,
|
|
max_tokens=32, # JSON 한 줄이므로 최소 토큰
|
|
temperature=0.0, # 결정론적 출력
|
|
stop=["<|im_end|>", "<|im_start|>", "\n\n"],
|
|
echo=False,
|
|
)
|
|
raw: str = out["choices"][0]["text"].strip()
|
|
|
|
# <think>...</think> 블록 제거 (Qwen3 계열)
|
|
raw = re.sub(r"<think>.*?</think>", "", raw, flags=re.DOTALL).strip()
|
|
logger.debug(f"[SPKCLS] LLM raw: {raw!r}")
|
|
|
|
# JSON 파싱: {"speaker": "관제"} 형태
|
|
m = re.search(r'\{[^{}]*"speaker"\s*:\s*"([^"]+)"[^{}]*\}', raw)
|
|
if m:
|
|
label = m.group(1).strip()
|
|
if label in ("관제", "열차", "불명"):
|
|
logger.debug(f"[SPKCLS] LLM 판별: '{text[:30]}' → {label}")
|
|
return label # type: ignore[return-value]
|
|
|
|
# 직접 단어 포함 Fallback
|
|
if "관제" in raw:
|
|
return "관제"
|
|
if "열차" in raw:
|
|
return "열차"
|
|
|
|
logger.warning(f"[SPKCLS] JSON 파싱 실패 → 불명 반환. raw={raw!r}")
|
|
return "불명"
|
|
|
|
except Exception as e:
|
|
logger.warning(f"[SPKCLS] LLM 호출 예외 → 불명 반환: {e}")
|
|
return "불명"
|