diff --git a/.gitignore b/.gitignore index 7e3f1de..c60e2a5 100644 --- a/.gitignore +++ b/.gitignore @@ -17,3 +17,7 @@ test CLAUDE.md .claude/ backup_file/ +AGENTS.md +PLANS.md +SKILL.md +CHECKLIST.md \ No newline at end of file diff --git a/LLM/OSS/formatter.py b/LLM/OSS/formatter.py index 92092db..b29cf2b 100644 --- a/LLM/OSS/formatter.py +++ b/LLM/OSS/formatter.py @@ -5,6 +5,8 @@ from core.settings import get_settings from LLM.OSS.modes import CONTACT_INTENT_RE +from LLM.OSS.postprocess.context import PostProcessContext +from LLM.OSS.postprocess.synonym_table import DORM_SYNONYM_RULES, POLICY_SYNONYM_RULES settings = get_settings() @@ -251,22 +253,23 @@ def expand_synonyms(user_text: str) -> list[str]: def expand_policy_synonyms(user_text: str) -> list[str]: text = user_text or "" synonyms: list[str] = [] - if "복학" in text or "휴복학" in text or "휴·복학" in text or "휴 학" in text: - synonyms += ["복학", "휴학", "휴복학", "휴·복학", "휴학/복학"] - if "휴학" in text: - for word in ["휴학", "복학", "휴복학", "휴·복학", "휴학/복학"]: - if word not in synonyms: - synonyms.append(word) - if "학적" in text: - synonyms += ["학적", "학적변동", "휴학", "복학", "재입학", "자퇴", "전과"] + for rule in POLICY_SYNONYM_RULES: + if any(trigger in text for trigger in rule["triggers"]): + for word in rule["synonyms"]: + if word not in synonyms: + synonyms.append(word) return synonyms def expand_dorm_synonyms(user_text: str) -> list[str]: text = user_text or "" - if any(keyword in text for keyword in ["기숙사", "생활관", "학생생활관", "사생", "입사", "퇴사", "입실", "퇴실", "생활관비"]): - return ["기숙사", "생활관", "학생생활관", "입사", "퇴사", "입사신청", "생활관비", "생활관 안내", "생활관 규정"] - return [] + synonyms: list[str] = [] + for rule in DORM_SYNONYM_RULES: + if any(trigger in text for trigger in rule["triggers"]): + for word in rule["synonyms"]: + if word not in synonyms: + synonyms.append(word) + return synonyms def ensure_layout_unknown(url: str) -> str: @@ -457,6 +460,31 @@ def _parse_bullets_and_pick( def one_sentence_from_sub_answer(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: if not sub_answer: return "요청하신 정보를 찾지 못했습니다.", None + ctx = build_contact_context(user_text, sub_answer) + return render_contact_response(ctx) + + +def matches_hint_label(label: str, hint: Optional[dict[str, object]]) -> bool: + return _hint_matches_line(label, "", hint) + + +def extract_contact_fields( + user_text: str, + sub_answer: str, + hint: Optional[dict[str, object]] = None, +) -> tuple[Optional[str], Optional[str], Optional[str]]: + return _parse_bullets_and_pick(user_text, sub_answer, hint) + + +def extract_fallback_phone( + user_text: str, + sub_answer: str, + hint: Optional[dict[str, object]] = None, +) -> Optional[str]: + return _fallback_phone_from_sub_answer(user_text, sub_answer, hint) + + +def build_contact_context(user_text: str, sub_answer: str, mode: str = "fast") -> PostProcessContext: hint = detect_dept_hint(user_text) label, phone, url = _parse_bullets_and_pick(user_text, sub_answer, hint) contact_intent = bool(CONTACT_INTENT_RE.search(user_text or "")) @@ -467,148 +495,215 @@ def one_sentence_from_sub_answer(user_text: str, sub_answer: str) -> tuple[str, if contact_intent and hint and label and not _hint_matches_line(label, "", hint): label = None - if label and phone: - return f"{label} 전화번호는 {phone}입니다.", url - if contact_intent and phone: - return f"요청하신 부서 담당자 전화번호는 {phone}입니다.", url - if contact_intent and not phone: - return "담당자 연락처를 바로 찾지 못했습니다. 학과(또는 부서) 풀네임으로 다시 입력해 주세요.", url - if label and url: - return f"{label} 정보는 {url}에서 확인할 수 있습니다.", url + return PostProcessContext( + user_text=user_text, + mode=mode, + sub_answer=sub_answer, + label=label, + phone=phone, + url=url, + hint=hint, + first_line=_first_line_or_default(sub_answer, "요청하신 정보를 찾지 못했습니다."), + contact_intent=contact_intent, + ) - first = sub_answer.strip().splitlines()[0].lstrip("- ").strip() - text = first if first.endswith(("다.", "요.")) else (first + "." if first else "요청하신 정보를 찾지 못했습니다.") - return text, url +def render_contact_response(ctx: PostProcessContext) -> tuple[str, Optional[str]]: + if ctx.label and ctx.phone: + return f"{ctx.label} 전화번호는 {ctx.phone}입니다.", ctx.url + if ctx.contact_intent and ctx.phone: + return f"요청하신 부서 담당자 전화번호는 {ctx.phone}입니다.", ctx.url + if ctx.contact_intent and not ctx.phone: + return "담당자 연락처를 바로 찾지 못했습니다. 학과(또는 부서) 풀네임으로 다시 입력해 주세요.", ctx.url + if ctx.label and ctx.url: + return f"{ctx.label} 정보는 {ctx.url}에서 확인할 수 있습니다.", ctx.url + return ctx.first_line, ctx.url -def one_sentence_topic(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + +def _build_default_info_text(user_text: str, suffix: str = "") -> str: + return f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}{suffix}에서 확인할 수 있습니다." + + +def _first_line_or_default(sub_answer: str, default_text: str) -> str: if not sub_answer: - return f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}에서 확인할 수 있습니다.", ORG_HOMEPAGE_URL - items = list(TITLE_URL_PAT.finditer(sub_answer)) - if not items: - first = sub_answer.strip().splitlines()[0].lstrip("- ").strip() - text = first if first.endswith(("다.", "요.")) else (first + "." if first else f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}에서 확인할 수 있습니다.") - return text, None + return default_text + first = sub_answer.strip().splitlines()[0].lstrip("- ").strip() + if not first: + return default_text + return first if first.endswith(("다.", "요.")) else first + "." - hint = detect_dept_hint(user_text) - compact_query = re.sub(r"\s+", "", user_text) + +def _collect_title_url_candidates( + user_text: str, + sub_answer: str, + scorer, +) -> list[tuple[float, str, str]]: candidates: list[tuple[float, str, str]] = [] - for item in items: + for item in TITLE_URL_PAT.finditer(sub_answer or ""): title = (item.group("title") or "").strip() url = (item.group("url") or "").strip() if SAFETY_URL_RE.search(url) or SAFETY_TITLE_RE.search(title): continue - score = 0.0 - if GOOD_URL_RE.search(url): - score += 3.0 - if BAD_URL_RE.search(url): - score -= 4.0 - if re.sub(r"\s+", "", title) == compact_query: - score += 6.0 - if INTRO_WORD_RE.search(title): - score += 3.0 - if CONTACT_WORD_RE.search(title): - score -= 5.0 - tokens = [token for token in re.findall(r"[가-힣A-Za-z0-9]{2,}", user_text)] - if any(token in title for token in tokens): - score += 1.0 - if hint: - if hint["path_base"] and hint["path_base"] in url: - score += 8.0 - if any(alias in title for alias in hint["aliases"]): - score += 4.0 - candidates.append((score, title, url)) - - if not candidates: - return f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}에서 확인할 수 있습니다.", ORG_HOMEPAGE_URL + candidates.append((scorer(user_text, title, url), title, url)) candidates.sort(key=lambda item: item[0], reverse=True) - best_score, best_title, best_url = candidates[0] - if best_score <= 0: - return f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}에서 확인할 수 있습니다.", ORG_HOMEPAGE_URL - best_url = ensure_layout_unknown(best_url) - return f"‘{user_text}’ 관련 정보는 ‘{best_title}’ 페이지({best_url})에서 확인할 수 있습니다.", best_url + return candidates -def one_sentence_policy(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: - if not sub_answer: - return f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}의 학사안내에서 확인할 수 있습니다.", ORG_HOMEPAGE_URL - items = list(TITLE_URL_PAT.finditer(sub_answer)) - if not items: - first = sub_answer.strip().splitlines()[0].lstrip("- ").strip() - text = first if first.endswith(("다.", "요.")) else (first + "." if first else f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}의 학사안내에서 확인할 수 있습니다.") - return text, None +def _topic_candidate_score(user_text: str, title: str, url: str) -> float: + hint = detect_dept_hint(user_text) + compact_query = re.sub(r"\s+", "", user_text) + score = 0.0 + if GOOD_URL_RE.search(url): + score += 3.0 + if BAD_URL_RE.search(url): + score -= 4.0 + if re.sub(r"\s+", "", title) == compact_query: + score += 6.0 + if INTRO_WORD_RE.search(title): + score += 3.0 + if CONTACT_WORD_RE.search(title): + score -= 5.0 + tokens = [token for token in re.findall(r"[가-힣A-Za-z0-9]{2,}", user_text)] + if any(token in title for token in tokens): + score += 1.0 + if hint: + if hint["path_base"] and hint["path_base"] in url: + score += 8.0 + if any(alias in title for alias in hint["aliases"]): + score += 4.0 + return score + +def _policy_candidate_score(user_text: str, title: str, url: str) -> float: synonyms = expand_policy_synonyms(user_text) tokens = [token for token in re.findall(r"[가-힣A-Za-z0-9·]{2,}", user_text)] - candidates: list[tuple[float, str, str]] = [] - for item in items: - title = (item.group("title") or "").strip() - url = (item.group("url") or "").strip() - if SAFETY_URL_RE.search(url) or SAFETY_TITLE_RE.search(title): - continue - score = 0.0 - if GOOD_URL_RE.search(url): - score += 3.5 - if BAD_URL_RE.search(url): - score -= 8.0 - if any(keyword in title for keyword in synonyms): - score += 7.0 - if any(keyword in title for keyword in ("휴학", "복학", "휴·복학", "휴복학", "학적", "학사안내")): - score += 3.0 - if any(token in title for token in tokens): - score += 1.0 - if CONTACT_WORD_RE.search(title): - score -= 3.0 - candidates.append((score, title, url)) + score = 0.0 + if GOOD_URL_RE.search(url): + score += 3.5 + if BAD_URL_RE.search(url): + score -= 8.0 + if any(keyword in title for keyword in synonyms): + score += 7.0 + if any(keyword in title for keyword in ("휴학", "복학", "휴·복학", "휴복학", "학적", "학사안내")): + score += 3.0 + if any(token in title for token in tokens): + score += 1.0 + if CONTACT_WORD_RE.search(title): + score -= 3.0 + return score + + +def _dorm_candidate_score(user_text: str, title: str, url: str) -> float: + synonyms = expand_dorm_synonyms(user_text) + tokens = [token for token in re.findall(r"[가-힣A-Za-z0-9·]{2,}", user_text)] + score = 0.0 + if GOOD_URL_RE.search(url): + score += 3.5 + if BAD_URL_RE.search(url): + score -= 8.0 + if any(keyword in title for keyword in synonyms): + score += 8.0 + if any(keyword in title for keyword in ("학생생활관", "생활관", "기숙사", "입사", "생활관비", "생활관 안내")): + score += 3.0 + if any(token in title for token in tokens): + score += 1.0 + if CONTACT_WORD_RE.search(title): + score -= 3.0 + return score + + +def _best_title_url_candidate( + user_text: str, + sub_answer: str, + default_text: str, + scorer, + prefer_good_url: bool = False, + require_positive_score: bool = False, +) -> tuple[Optional[str], Optional[str], str]: + items = list(TITLE_URL_PAT.finditer(sub_answer or "")) + if not sub_answer: + return None, None, default_text + if not items: + return None, None, _first_line_or_default(sub_answer, default_text) + candidates = _collect_title_url_candidates(user_text, sub_answer, scorer) if not candidates: - return f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}의 학사안내에서 확인할 수 있습니다.", ORG_HOMEPAGE_URL - candidates.sort(key=lambda item: item[0], reverse=True) - preferred = [candidate for candidate in candidates if GOOD_URL_RE.search(candidate[2])] or candidates - _, best_title, best_url = preferred[0] - best_url = ensure_layout_unknown(best_url) - return f"‘{user_text}’ 관련 공식 안내는 ‘{best_title}’ 페이지({best_url})에서 확인할 수 있습니다.", best_url + return None, None, default_text + + selected = candidates + if prefer_good_url: + selected = [candidate for candidate in candidates if GOOD_URL_RE.search(candidate[2])] or candidates + + best_score, best_title, best_url = selected[0] + if require_positive_score and best_score <= 0: + return None, None, default_text + return best_title, best_url, _first_line_or_default(sub_answer, default_text) + + +def one_sentence_topic(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + title, url, first_line = extract_topic_candidate(user_text, sub_answer) + default_text = _build_default_info_text(user_text) + if not sub_answer: + return default_text, ORG_HOMEPAGE_URL + if title and url: + best_url = ensure_layout_unknown(url) + return f"‘{user_text}’ 관련 정보는 ‘{title}’ 페이지({best_url})에서 확인할 수 있습니다.", best_url + return first_line, None + + +def extract_topic_candidate(user_text: str, sub_answer: str) -> tuple[Optional[str], Optional[str], str]: + default_text = _build_default_info_text(user_text) + return _best_title_url_candidate( + user_text, + sub_answer, + default_text, + scorer=_topic_candidate_score, + require_positive_score=True, + ) + + +def one_sentence_policy(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + title, url, first_line = extract_policy_candidate(user_text, sub_answer) + default_text = _build_default_info_text(user_text, "의 학사안내") + if not sub_answer: + return default_text, ORG_HOMEPAGE_URL + if title and url: + best_url = ensure_layout_unknown(url) + return f"‘{user_text}’ 관련 공식 안내는 ‘{title}’ 페이지({best_url})에서 확인할 수 있습니다.", best_url + return first_line, None + + +def extract_policy_candidate(user_text: str, sub_answer: str) -> tuple[Optional[str], Optional[str], str]: + default_text = _build_default_info_text(user_text, "의 학사안내") + return _best_title_url_candidate( + user_text, + sub_answer, + default_text, + scorer=_policy_candidate_score, + prefer_good_url=True, + ) def one_sentence_dorm(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + title, url, first_line = extract_dorm_candidate(user_text, sub_answer) + default_text = _build_default_info_text(user_text, "의 생활관 안내") if not sub_answer: - return f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}의 생활관 안내에서 확인할 수 있습니다.", ORG_HOMEPAGE_URL - items = list(TITLE_URL_PAT.finditer(sub_answer)) - if not items: - first = sub_answer.strip().splitlines()[0].lstrip("- ").strip() - text = first if first.endswith(("다.", "요.")) else (first + "." if first else f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}의 생활관 안내에서 확인할 수 있습니다.") - return text, None + return default_text, ORG_HOMEPAGE_URL + if title and url: + best_url = ensure_layout_unknown(url) + return f"‘{user_text}’ 관련 공식 안내는 ‘{title}’ 페이지({best_url})에서 확인할 수 있습니다.", best_url + return first_line, None - synonyms = expand_dorm_synonyms(user_text) - tokens = [token for token in re.findall(r"[가-힣A-Za-z0-9·]{2,}", user_text)] - candidates: list[tuple[float, str, str]] = [] - for item in items: - title = (item.group("title") or "").strip() - url = (item.group("url") or "").strip() - if SAFETY_URL_RE.search(url) or SAFETY_TITLE_RE.search(title): - continue - score = 0.0 - if GOOD_URL_RE.search(url): - score += 3.5 - if BAD_URL_RE.search(url): - score -= 8.0 - if any(keyword in title for keyword in synonyms): - score += 8.0 - if any(keyword in title for keyword in ("학생생활관", "생활관", "기숙사", "입사", "생활관비", "생활관 안내")): - score += 3.0 - if any(token in title for token in tokens): - score += 1.0 - if CONTACT_WORD_RE.search(title): - score -= 3.0 - candidates.append((score, title, url)) - if not candidates: - return f"‘{user_text}’ 관련 정보는 {ORG_HOMEPAGE_LABEL}의 생활관 안내에서 확인할 수 있습니다.", ORG_HOMEPAGE_URL - candidates.sort(key=lambda item: item[0], reverse=True) - preferred = [candidate for candidate in candidates if GOOD_URL_RE.search(candidate[2])] or candidates - _, best_title, best_url = preferred[0] - best_url = ensure_layout_unknown(best_url) - return f"‘{user_text}’ 관련 공식 안내는 ‘{best_title}’ 페이지({best_url})에서 확인할 수 있습니다.", best_url +def extract_dorm_candidate(user_text: str, sub_answer: str) -> tuple[Optional[str], Optional[str], str]: + default_text = _build_default_info_text(user_text, "의 생활관 안내") + return _best_title_url_candidate( + user_text, + sub_answer, + default_text, + scorer=_dorm_candidate_score, + prefer_good_url=True, + ) def _extract_url_from_text(text: str) -> Optional[str]: @@ -626,6 +721,15 @@ def _extract_url_from_text(text: str) -> Optional[str]: def one_sentence_grad(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: if not sub_answer: return f"졸업 관련 정보는 {GRAD_PAGE_URL}에서 확인할 수 있습니다.", GRAD_PAGE_URL + summary, source_url = extract_grad_summary(user_text, sub_answer) + if "확인할 수 있습니다." in summary: + return summary, source_url + return f"{summary} 자세한 내용은 {source_url}에서 확인할 수 있습니다.", source_url + + +def extract_grad_summary(user_text: str, sub_answer: str) -> tuple[str, str]: + if not sub_answer: + return "졸업 관련 정보를 찾지 못했습니다.", ensure_layout_unknown(GRAD_PAGE_URL) lines = [re.sub(r"\*\*", "", line.strip().lstrip("- ").strip()) for line in sub_answer.splitlines() if line.strip()] ask_two_year = bool(re.search(r"2\s*년제", user_text)) @@ -642,6 +746,4 @@ def one_sentence_grad(user_text: str, sub_answer: str) -> tuple[str, Optional[st summary = preferred[0] if preferred else (lines[0] if lines else "졸업 관련 정보를 찾지 못했습니다.") source_url = _extract_url_from_text(sub_answer) or ensure_layout_unknown(GRAD_PAGE_URL) - if "확인할 수 있습니다." in summary: - return summary, source_url - return f"{summary} 자세한 내용은 {source_url}에서 확인할 수 있습니다.", source_url + return summary, source_url diff --git a/LLM/OSS/postprocess/__init__.py b/LLM/OSS/postprocess/__init__.py new file mode 100644 index 0000000..4234ebf --- /dev/null +++ b/LLM/OSS/postprocess/__init__.py @@ -0,0 +1,3 @@ +from LLM.OSS.postprocess.engine import run_postprocess + +__all__ = ["run_postprocess"] diff --git a/LLM/OSS/postprocess/context.py b/LLM/OSS/postprocess/context.py new file mode 100644 index 0000000..d7ff10b --- /dev/null +++ b/LLM/OSS/postprocess/context.py @@ -0,0 +1,31 @@ +from dataclasses import dataclass +from typing import Optional + + +@dataclass +class PostProcessContext: + user_text: str + mode: str + sub_answer: str + label: Optional[str] = None + phone: Optional[str] = None + url: Optional[str] = None + hint: Optional[dict[str, object]] = None + first_line: str = "" + contact_intent: bool = False + + @property + def has_label(self) -> bool: + return bool(self.label) + + @property + def has_phone(self) -> bool: + return bool(self.phone) + + @property + def has_url(self) -> bool: + return bool(self.url) + + @property + def has_sub_answer(self) -> bool: + return bool((self.sub_answer or "").strip()) diff --git a/LLM/OSS/postprocess/engine.py b/LLM/OSS/postprocess/engine.py new file mode 100644 index 0000000..45873f3 --- /dev/null +++ b/LLM/OSS/postprocess/engine.py @@ -0,0 +1,126 @@ +from typing import Optional + +from core.settings import get_settings +from LLM.OSS import formatter +from LLM.OSS.postprocess.context import PostProcessContext +from LLM.OSS.postprocess.message_table import MESSAGES +from LLM.OSS.postprocess.registry import MODE_PIPELINES +from LLM.OSS.postprocess.rules_table import CONTACT_RESPONSE_RULES + + +settings = get_settings() + + +def _format_message(template_key: str, **values: object) -> str: + template = MESSAGES[template_key] + payload = { + "org_homepage_label": settings.org_homepage_label, + "org_homepage_url": settings.org_homepage_url, + "grad_page_url": settings.grad_page_url, + **values, + } + return template.format(**payload) + + +def _first_line(sub_answer: str) -> str: + first = (sub_answer or "").strip().splitlines()[0].lstrip("- ").strip() if (sub_answer or "").strip() else "" + if not first: + return MESSAGES["not_found"] + return first if first.endswith(("다.", "요.")) else first + "." + + +def _build_sub_answer_context(mode: str, user_text: str, sub_answer: str) -> PostProcessContext: + ctx = formatter.build_contact_context(user_text, sub_answer, mode=mode) + if not ctx.first_line: + ctx.first_line = _first_line(sub_answer) + return ctx + + +def _conditions(ctx: PostProcessContext) -> dict[str, bool]: + return { + "contact_intent": ctx.contact_intent, + "has_label": ctx.has_label, + "has_phone": ctx.has_phone, + "has_url": ctx.has_url, + "has_sub_answer": ctx.has_sub_answer, + "not_has_phone": not ctx.has_phone, + } + + +def _apply_contact_rules(ctx: PostProcessContext) -> Optional[str]: + flags = _conditions(ctx) + rules = sorted(CONTACT_RESPONSE_RULES, key=lambda item: item["priority"]) + for rule in rules: + if ctx.mode not in rule["modes"]: + continue + if all(flags.get(condition, False) for condition in rule["conditions"]): + return _format_message(rule["template_key"], label=ctx.label, phone=ctx.phone, url=ctx.url, user_text=ctx.user_text) + return None + + +def _run_sub_answer(mode: str, user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + if not sub_answer: + return MESSAGES["not_found"], None + ctx = _build_sub_answer_context(mode, user_text, sub_answer) + text = _apply_contact_rules(ctx) + if text: + return text, ctx.url + return ctx.first_line, ctx.url + + +def _run_topic(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + title, url, first_line = formatter.extract_topic_candidate(user_text, sub_answer) + if not sub_answer: + return _format_message("topic_default", user_text=user_text), settings.org_homepage_url + if title and url: + url = formatter.ensure_layout_unknown(url) + return _format_message("topic_page", user_text=user_text, title=title, url=url), url + return first_line, None + + +def _run_policy(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + title, url, first_line = formatter.extract_policy_candidate(user_text, sub_answer) + if not sub_answer: + return _format_message("policy_default", user_text=user_text), settings.org_homepage_url + if title and url: + url = formatter.ensure_layout_unknown(url) + return _format_message("official_page", user_text=user_text, title=title, url=url), url + return first_line, None + + +def _run_dorm(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + title, url, first_line = formatter.extract_dorm_candidate(user_text, sub_answer) + if not sub_answer: + return _format_message("dorm_default", user_text=user_text), settings.org_homepage_url + if title and url: + url = formatter.ensure_layout_unknown(url) + return _format_message("official_page", user_text=user_text, title=title, url=url), url + return first_line, None + + +def _run_grad(user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + summary, url = formatter.extract_grad_summary(user_text, sub_answer) + if not sub_answer: + return _format_message("grad_default"), settings.grad_page_url + if "확인할 수 있습니다." in summary: + return summary, url + return _format_message("grad_with_url", summary=summary, url=url), url + + +PROCESSORS = { + "sub_answer": _run_sub_answer, + "topic": _run_topic, + "policy": _run_policy, + "dorm": _run_dorm, + "grad": _run_grad, +} + + +def run_postprocess(mode: str, user_text: str, sub_answer: str) -> tuple[str, Optional[str]]: + config = MODE_PIPELINES.get(mode) + if not config: + return formatter.one_sentence_from_sub_answer(user_text, sub_answer) + processor = PROCESSORS[config["processor"]] + if config["processor"] == "sub_answer": + return processor(mode, user_text, sub_answer) + return processor(user_text, sub_answer) diff --git a/LLM/OSS/postprocess/message_table.py b/LLM/OSS/postprocess/message_table.py new file mode 100644 index 0000000..f427280 --- /dev/null +++ b/LLM/OSS/postprocess/message_table.py @@ -0,0 +1,14 @@ +MESSAGES = { + "not_found": "요청하신 정보를 찾지 못했습니다.", + "contact_with_label_and_phone": "{label} 전화번호는 {phone}입니다.", + "contact_with_phone_only": "요청하신 부서 담당자 전화번호는 {phone}입니다.", + "contact_retry_full_name": "담당자 연락처를 바로 찾지 못했습니다. 학과(또는 부서) 풀네임으로 다시 입력해 주세요.", + "topic_default": "‘{user_text}’ 관련 정보는 {org_homepage_label}에서 확인할 수 있습니다.", + "policy_default": "‘{user_text}’ 관련 정보는 {org_homepage_label}의 학사안내에서 확인할 수 있습니다.", + "dorm_default": "‘{user_text}’ 관련 정보는 {org_homepage_label}의 생활관 안내에서 확인할 수 있습니다.", + "topic_page": "‘{user_text}’ 관련 정보는 ‘{title}’ 페이지({url})에서 확인할 수 있습니다.", + "official_page": "‘{user_text}’ 관련 공식 안내는 ‘{title}’ 페이지({url})에서 확인할 수 있습니다.", + "label_with_url": "{label} 정보는 {url}에서 확인할 수 있습니다.", + "grad_default": "졸업 관련 정보는 {grad_page_url}에서 확인할 수 있습니다.", + "grad_with_url": "{summary} 자세한 내용은 {url}에서 확인할 수 있습니다.", +} diff --git a/LLM/OSS/postprocess/registry.py b/LLM/OSS/postprocess/registry.py new file mode 100644 index 0000000..68cb5fd --- /dev/null +++ b/LLM/OSS/postprocess/registry.py @@ -0,0 +1,17 @@ +MODE_PIPELINES = { + "fast": { + "processor": "sub_answer", + }, + "policy": { + "processor": "policy", + }, + "dorm": { + "processor": "dorm", + }, + "grad": { + "processor": "grad", + }, + "topic": { + "processor": "topic", + }, +} diff --git a/LLM/OSS/postprocess/rules_table.py b/LLM/OSS/postprocess/rules_table.py new file mode 100644 index 0000000..2d85910 --- /dev/null +++ b/LLM/OSS/postprocess/rules_table.py @@ -0,0 +1,30 @@ +CONTACT_RESPONSE_RULES = [ + { + "name": "contact_with_label_and_phone", + "modes": {"fast", "policy", "dorm", "grad"}, + "conditions": ("contact_intent", "has_label", "has_phone"), + "template_key": "contact_with_label_and_phone", + "priority": 10, + }, + { + "name": "contact_with_phone_only", + "modes": {"fast", "policy", "dorm", "grad"}, + "conditions": ("contact_intent", "has_phone"), + "template_key": "contact_with_phone_only", + "priority": 20, + }, + { + "name": "contact_retry_full_name", + "modes": {"fast", "policy", "dorm", "grad"}, + "conditions": ("contact_intent", "not_has_phone"), + "template_key": "contact_retry_full_name", + "priority": 30, + }, + { + "name": "label_with_url", + "modes": {"fast"}, + "conditions": ("has_label", "has_url"), + "template_key": "label_with_url", + "priority": 40, + }, +] diff --git a/LLM/OSS/postprocess/synonym_table.py b/LLM/OSS/postprocess/synonym_table.py new file mode 100644 index 0000000..7bb4d4b --- /dev/null +++ b/LLM/OSS/postprocess/synonym_table.py @@ -0,0 +1,21 @@ +POLICY_SYNONYM_RULES = ( + { + "triggers": ("복학", "휴복학", "휴·복학", "휴 학"), + "synonyms": ("복학", "휴학", "휴복학", "휴·복학", "휴학/복학"), + }, + { + "triggers": ("휴학",), + "synonyms": ("휴학", "복학", "휴복학", "휴·복학", "휴학/복학"), + }, + { + "triggers": ("학적",), + "synonyms": ("학적", "학적변동", "휴학", "복학", "재입학", "자퇴", "전과"), + }, +) + +DORM_SYNONYM_RULES = ( + { + "triggers": ("기숙사", "생활관", "학생생활관", "사생", "입사", "퇴사", "입실", "퇴실", "생활관비"), + "synonyms": ("기숙사", "생활관", "학생생활관", "입사", "퇴사", "입사신청", "생활관비", "생활관 안내", "생활관 규정"), + }, +) diff --git a/LLM/OSS/service.py b/LLM/OSS/service.py index 3dcf108..86ace6e 100644 --- a/LLM/OSS/service.py +++ b/LLM/OSS/service.py @@ -14,11 +14,6 @@ from core.settings import get_settings from LLM.OSS.formatter import ( dept_clarification_message, - one_sentence_dorm, - one_sentence_from_sub_answer, - one_sentence_grad, - one_sentence_policy, - one_sentence_topic, render_chatty_schedule, scrub_non_contact, ) @@ -32,6 +27,7 @@ looks_like_schedule, looks_like_topic, ) +from LLM.OSS.postprocess import run_postprocess from LLM.rule_book.graph import run_rule_book from LLM.sub_model.query_index import build_answer from LLM.sub_model.schedule_index import schedule_search @@ -249,7 +245,7 @@ def cache_and_return(response: dict) -> dict: return cache_and_return({"engine": "fast", "text": clarification}) sub_answer = call_submodel(user_text) - text, url = one_sentence_from_sub_answer(user_text, sub_answer) + text, url = run_postprocess("fast", user_text, sub_answer) response = {"engine": "fast", "text": text} if url: response["url"] = url @@ -257,7 +253,7 @@ def cache_and_return(response: dict) -> dict: if mode == "policy": sub_answer = call_submodel(user_text) - text, url = one_sentence_policy(user_text, sub_answer) + text, url = run_postprocess("policy", user_text, sub_answer) response = {"engine": "policy", "text": text} if url: response["url"] = url @@ -265,7 +261,7 @@ def cache_and_return(response: dict) -> dict: if mode == "dorm": sub_answer = call_submodel(user_text) - text, url = one_sentence_dorm(user_text, sub_answer) + text, url = run_postprocess("dorm", user_text, sub_answer) response = {"engine": "dorm", "text": text} if url: response["url"] = url @@ -273,7 +269,7 @@ def cache_and_return(response: dict) -> dict: if mode == "grad": sub_answer = call_submodel(user_text) - text, url = one_sentence_grad(user_text, sub_answer) + text, url = run_postprocess("grad", user_text, sub_answer) response = {"engine": "grad", "text": text} if url: response["url"] = url @@ -281,7 +277,7 @@ def cache_and_return(response: dict) -> dict: if mode == "topic": sub_answer = call_submodel(user_text) - text, url = one_sentence_topic(user_text, sub_answer) + text, url = run_postprocess("topic", user_text, sub_answer) response = {"engine": "topic", "text": text} if url: response["url"] = url @@ -312,9 +308,9 @@ def cache_and_return(response: dict) -> dict: return response if sub_answer: if looks_like_topic(user_text): - text, _ = one_sentence_topic(user_text, sub_answer) + text, _ = run_postprocess("topic", user_text, sub_answer) else: - text, _ = one_sentence_from_sub_answer(user_text, sub_answer) + text, _ = run_postprocess("fast", user_text, sub_answer) output = text else: output = "잘 이해하지 못했어요. 다시 질문해주세요." @@ -336,7 +332,7 @@ def cache_and_return(response: dict) -> dict: temperature=0.2, ) if not fused: - text, _ = one_sentence_topic(user_text, sub_answer) + text, _ = run_postprocess("topic", user_text, sub_answer) fused = text if looks_like_topic(user_text) else "좋아요, 무엇을 이야기해 볼까요?" latency = int((time.monotonic() - start) * 1000)