"""ChatEngine — 대화 step 전이 엔진 (동기 순수 로직, P7 슬라이스). Chat_server 의 step 체계(서비스안내→담당자확인→협상품목안내→가격협상→협상완료/실패→협상종료)와 조건 분기(check_wildcard_entry/price_match/iteration_limit)·와일드카드 진입을 우리 구현으로 재작성. DB/정책(카드선택·학습)은 여기 두지 않는다 — ChatService(async)가 StepView 의 신호를 보고 처리한다. """ import re from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from negotiation.chat.service.script_repository import ScriptRepository MAX_ROUNDS = 3 _PRICE_MODES = ("price",) _CHOICE_MODES = ("yes_no", "confirm", "delivery_type") # 프론트는 표시용 문자열로 가격을 보낸다(예: "530,000원"). 천단위 콤마·통화기호("원")·공백 등 # 숫자 외 문자를 제거하고 파싱한다. (콤마만 지우면 "원" 때문에 float() 가 실패해 가격 입력이 # 영영 저장되지 않고 같은 step 에 머무는 버그가 났었다.) _PRICE_CLEAN_RE = re.compile(r"[^\d.]") def _parse_price(user_input: Any) -> Optional[float]: cleaned = _PRICE_CLEAN_RE.sub("", str(user_input if user_input is not None else "")) if not cleaned or cleaned == ".": return None try: price = float(cleaned) except ValueError: return None return price if price > 0 else None @dataclass class ChatSession: session_id: str tenant_id: str company_id: str rq_type: str = "재협상" step: str = "시작" context: Dict[str, Any] = field(default_factory=dict) used_action_ids: set = field(default_factory=set) action_space_size: int = 0 # 카드 소진 판정용 (ChatService 가 주입) ended: bool = False @dataclass class StepView: step: str script: str input_mode: str input_options: List[str] chat_end: bool client_step: Optional[str] = None needs_card_selection: bool = False # 가격협상(에이전트 카운터) → UCB 카드선택+학습 outcome: Optional[str] = None # 협상완료="success" / 협상실패="failure" → 종료보상 wildcard: Optional[str] = None # 발동한 와일드카드 key error: Optional[str] = None class ChatEngine: def __init__(self, scripts_repo: ScriptRepository, rq_type: str = "재협상"): self.repo = scripts_repo self.rq_type = rq_type self.scripts = scripts_repo.load_scripts(rq_type) self.step_map = scripts_repo.client_step_mapping() # ---- public -------------------------------------------------------- def start(self, session: ChatSession) -> StepView: first = self._default_next(self.scripts.get("시작", {})) or "서비스안내" return self._render(session, first) def advance(self, session: ChatSession, user_input: Optional[str]) -> StepView: if session.ended: return self._error(session, "협상이 종료되었습니다. 새 협상을 시작하세요.") node = self.scripts.get(session.step, {}) mode = node.get("next_input_mode", "null") if mode in _PRICE_MODES: price = _parse_price(user_input) if price is None: return self._error(session, "가격을 숫자로 입력해 주세요.") session.context["input_price"] = price session.context["round"] = session.context.get("round", 0) + 1 nxt = self._default_next(node) elif mode in _CHOICE_MODES: nxt = self._choice_next(node, user_input, session) else: nxt = self._default_next(node) nxt = self._resolve(nxt, session) return self._render(session, nxt) # ---- transition ---------------------------------------------------- def _default_next(self, node: dict) -> Optional[str]: ns = node.get("next_step") if isinstance(ns, dict): return ns.get("default") or next(iter(ns.values()), None) return ns def _choice_next(self, node: dict, choice: Optional[str], session: ChatSession): ns = node.get("next_step") or {} if not isinstance(ns, dict): return ns val = ns.get(choice) if val is None: val = ns.get("default") or next(iter(ns.values()), None) return val def _resolve(self, nxt, session: ChatSession) -> Optional[str]: """조건 리스트 평가 + 가격협상_와일드 가상스텝 → 실제 와일드카드 key 로 해석.""" if isinstance(nxt, list): nxt = self._eval_conditions(nxt, session) if nxt == "가격협상_와일드": nxt = self._pick_wildcard(session) return nxt def _eval_conditions(self, conds: List[dict], session: ChatSession) -> Optional[str]: """KT 구매자 관점 조건 평가. - 협력사 제시가 ≤ anchor → 우선협상(협상완료). - anchor 살짝 초과(≤ anchor*1.05) + 와일드카드 미사용 → 와일드카드로 인하 압박. - 설정 카드(action_space) 모두 소진 → 협상실패. - 그 외 → 가격협상(카드 1장 플레이 후 재제안). """ ctx = session.context price = ctx.get("input_price", 0) anchor = ctx.get("anchor_price", 0) cards_used = len(session.used_action_ids) cards_total = session.action_space_size or 0 for c in conds: cond = c.get("condition") ok = False if cond == "check_wildcard_entry": ok = (not ctx.get("wildcard_used")) and anchor < price <= anchor * 1.05 elif cond == "check_is_supplier_type_c": ok = False # 공급사 유형 미보유 (PoC 단순화) elif cond == "check_price_match": # = 우선협상: 제시가가 앵커가 이하 ok = anchor > 0 and price <= anchor elif cond == "check_iteration_limit": # = 카드 소진 ok = cards_total > 0 and cards_used >= cards_total elif cond == "default": ok = True if ok: return c.get("next") return "가격협상" def _pick_wildcard(self, session: ChatSession) -> str: """앵커가 살짝 초과 구간에서 인하 압박 카드 선택. - 앵커가에 아주 근접(≤ anchor*1.02): 1% 인하 요청(wild_card_1pct) → 앵커가 이하로 유도. - 그 외: 목표 매입가 맞춰달라(wild_card_budget). """ ctx = session.context price = ctx.get("input_price", 0) anchor = ctx.get("anchor_price", 0) ctx["wildcard_used"] = True if anchor > 0 and price <= anchor * 1.02: ctx["offer_1pct"] = int(round(price * 0.99)) # 1% 인하가 return "wild_card_1pct" return "wild_card_budget" # ---- render -------------------------------------------------------- def _vars(self, session: ChatSession) -> Dict[str, Any]: ctx = session.context out = {} if "input_price" in ctx: out["input_price"] = int(ctx["input_price"]) if "target_price" in ctx: out["target"] = int(ctx["target_price"]) if "offer_1pct" in ctx: out["offer_1pct"] = int(ctx["offer_1pct"]) return out def _render(self, session: ChatSession, step_key: Optional[str]) -> StepView: if not step_key or step_key not in self.scripts: return self._error(session, f"다음 단계를 찾을 수 없습니다: {step_key}") node = self.scripts[step_key] session.step = step_key session.ended = bool(node.get("chat_end")) outcome = "success" if step_key == "협상완료" else "failure" if step_key == "협상실패" else None return StepView( step=step_key, script=self.repo.format_script(node.get("script", ""), self._vars(session)), input_mode=node.get("next_input_mode", "null"), input_options=node.get("input_options", []), chat_end=bool(node.get("chat_end")), client_step=self.step_map.get(step_key, step_key), needs_card_selection=(step_key == "가격협상"), outcome=outcome, wildcard=step_key if step_key.startswith("wild_card_") else None, ) def _error(self, session: ChatSession, msg: str) -> StepView: node = self.scripts.get(session.step, {}) return StepView( step=session.step, script=node.get("script", ""), input_mode=node.get("next_input_mode", "null"), input_options=node.get("input_options", []), chat_end=session.ended, client_step=self.step_map.get(session.step, session.step), error=msg, )