You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
130 lines
4.3 KiB
Python
130 lines
4.3 KiB
Python
# backend/app/automation/nl_parser.py
|
|
# 자연어 한 문장 → 규칙. LLM generate_json + 휴리스틱 폴백. 골든 예시 3건 결정적.
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from dataclasses import dataclass
|
|
|
|
from ..llm.provider import LLMProvider, get_provider
|
|
|
|
|
|
@dataclass
|
|
class ParsedRule:
|
|
matched: bool
|
|
name: str
|
|
cat: str # mail|cal|focus|life
|
|
trigger: str
|
|
cond: str | None
|
|
action: str
|
|
model: str
|
|
confidence: float
|
|
|
|
|
|
# 원본 auto-data.js examples 골든 매핑(텍스트 정확 일치 → 결정적 미리보기)
|
|
EXAMPLES = {
|
|
"출장 전날엔 저녁 일정 비워줘": ParsedRule(
|
|
True,
|
|
"출장 전날 저녁 비우기",
|
|
"cal",
|
|
"출장 전날이 되면",
|
|
"18시 이후 일정이 있으면",
|
|
"다른 날로 옮기자고 제안",
|
|
"example",
|
|
0.95,
|
|
),
|
|
"뉴스레터는 모아서 저녁에 보여줘": ParsedRule(
|
|
True,
|
|
"뉴스레터는 저녁에",
|
|
"mail",
|
|
"뉴스레터 메일 도착",
|
|
None,
|
|
"받은편지함 건너뛰고 18:30 다이제스트로",
|
|
"example",
|
|
0.95,
|
|
),
|
|
"운동을 3일 거르면 산책 잡아줘": ParsedRule(
|
|
True,
|
|
"산책 리마인더",
|
|
"life",
|
|
"운동 기록 3일 연속 없음",
|
|
"저녁에 빈 시간이 있으면",
|
|
"30분 산책 블록 제안",
|
|
"example",
|
|
0.95,
|
|
),
|
|
}
|
|
|
|
# 카테고리 키워드(휴리스틱)
|
|
CAT_RE = [
|
|
("mail", re.compile(r"(메일|뉴스레터|받은편지함|회신|발송|스팸)")),
|
|
("cal", re.compile(r"(일정|회의|미팅|캘린더|블록|비워|출장|약속)")),
|
|
("focus", re.compile(r"(집중|딥 ?워크|방해 금지|알림 보류|리포트 준비)")),
|
|
("life", re.compile(r"(운동|산책|영수증|결제|구독|수면|건강|지출|카드)")),
|
|
]
|
|
|
|
NL_SYSTEM = (
|
|
"너는 한국어 비서 '아리'의 자동화 규칙 파서다. 사용자의 한 문장을 "
|
|
"{trigger(언제), cond(조건; 없으면 null), action(무엇을), cat, name} 으로 분해한다. "
|
|
"cat 은 mail|cal|focus|life 중 하나. name 은 8자 내외 한국어 규칙 이름. "
|
|
"보내기·결제·삭제·전달 같은 위험 동작은 'cond'에 확인 단서를 남겨라."
|
|
)
|
|
|
|
|
|
def _build_prompt(text: str) -> str:
|
|
return (
|
|
f'문장: "{text}"\n\n'
|
|
"아래 JSON 스키마로만 답하라:\n"
|
|
"{\n"
|
|
' "name": "규칙 이름(한국어, 8자 내외)",\n'
|
|
' "cat": "mail|cal|focus|life",\n'
|
|
' "trigger": "언제(트리거)",\n'
|
|
' "cond": "조건 또는 null",\n'
|
|
' "action": "무엇을(동작)"\n'
|
|
"}"
|
|
)
|
|
|
|
|
|
def _heuristic(text: str) -> ParsedRule:
|
|
t = (text or "").strip()
|
|
if not t:
|
|
return ParsedRule(False, "", "cal", "", None, "", "heuristic", 0.0)
|
|
cat = "cal"
|
|
for c, rx in CAT_RE:
|
|
if rx.search(t):
|
|
cat = c
|
|
break
|
|
cond = None
|
|
m = re.search(r"(.+?(되면|도착|없으면|이면|거르면|걸리면))\s*(.*)", t)
|
|
if m:
|
|
trigger, action = m.group(1).strip(), (m.group(3).strip() or t)
|
|
else:
|
|
trigger, action = t, t
|
|
name = (action[:8] or t[:8]).strip()
|
|
return ParsedRule(True, name, cat, trigger, cond, action, "heuristic", 0.55)
|
|
|
|
|
|
def parse_rule(text: str, provider: LLMProvider | None = None) -> ParsedRule:
|
|
t = (text or "").strip()
|
|
if not t: # 빈 입력은 LLM 호출 없이 결정적으로 미매칭
|
|
return _heuristic(t)
|
|
if t in EXAMPLES: # 골든: 데모 결정성 보장
|
|
return EXAMPLES[t]
|
|
prov = provider or get_provider()
|
|
try:
|
|
data = prov.generate_json(NL_SYSTEM + "\n" + _build_prompt(t), schema={"cat": "str"})
|
|
cat = data.get("cat") if data.get("cat") in ("mail", "cal", "focus", "life") else None
|
|
if cat and data.get("action"):
|
|
return ParsedRule(
|
|
True,
|
|
str(data.get("name") or data["action"][:8]).strip(),
|
|
cat,
|
|
str(data.get("trigger", t)).strip(),
|
|
(str(data["cond"]).strip() if data.get("cond") else None),
|
|
str(data["action"]).strip(),
|
|
f"llm:{getattr(prov, 'model', prov.name)}",
|
|
float(data.get("confidence", 0.8)),
|
|
)
|
|
except Exception:
|
|
pass
|
|
return _heuristic(t) # 폴백
|