"""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.cards.domain.tactics import compute_counter, tactic_for from negotiation.chat.service.script_repository import ScriptRepository MAX_ROUNDS = 3 # config 미주입 시 폴백 (규칙 정본은 tenant config negotiation.max_counter_rounds) _PRICE_MODES = ("price",) _CHOICE_MODES = ("yes_no", "confirm", "delivery_type") # 최종 타결/결렬 스텝. 재협상=협상완료(우선협상 타결), 재견적=결과제출(투찰확정). 둘 다 협상실패=결렬. # 이 스텝들은 chat_end=False(뒤에 협상종료가 옴)라, outcome 을 컨텍스트에 적재했다가 # 실제 종료(chat_end=협상종료) 시점에 확정 보고한다 → backend 가 chat_end 에서 DONE/REJECTED 를 옳게 가른다. _SUCCESS_STEPS = ("협상완료", "결과제출") _FAILURE_STEPS = ("협상실패",) # 카운터 제안(pending_counter_price) 수락으로 인정하는 선택 입력. _ACCEPT_INPUTS = ("예", "수락") # 프론트는 표시용 문자열로 가격을 보낸다(예: "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() # 결정 스택 규칙층(Phase 1): 와일드카드 진입 임계·라운드 상한을 테넌트 config 에서 읽는다. # (하드코딩 1.02/1.05/3 을 데이터화 — 고객사별로 튜닝 가능, 코드 수정 불필요) self.rules = scripts_repo.config.negotiation # ---- 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.pop("pending_counter_price", None) session.context["prev_partner_price"] = price # 협력사 첫 제시가 — 가격 수용률(첫 제시가 대비 양보율) 동적 계산의 기준값. session.context.setdefault("first_offer_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) # 카운터 수락 일반 메커니즘: 카드/와일드카드가 제시한 카운터가(pending_counter_price)를 # 협력사가 수락("예"/"수락")한 채 성공 스텝으로 전이하면 합의가 = 카운터가. # (구 offer_1pct 특수 분기의 일반화. 거절인데 성공 스텝으로 가는 경로 — 1% 거절 시 # 원 제시가 수락 종결 — 는 카운터를 버리고 기존 input_price 로 타결한다.) if mode in _CHOICE_MODES and nxt in _SUCCESS_STEPS: pending = session.context.pop("pending_counter_price", None) if pending and user_input in _ACCEPT_INPUTS: session.context["input_price"] = float(pending) 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 구매자 관점 조건 평가 (임계값은 config negotiation.* — 규칙층 데이터화). - 협력사 제시가 ≤ anchor → 우선협상(협상완료). - anchor 살짝 초과(≤ anchor×wildcard_entry_ratio) + 와일드카드 미사용 → 와일드카드로 인하 압박. - 설정 카드(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 > 0 and anchor < price and ( price <= anchor * self.rules.wildcard_1pct_ratio or (bool(ctx.get("allow_selected_wildcards", True)) and price <= anchor * self.rules.wildcard_entry_ratio) ) ) 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": # 협상 라운드 상한 또는 카드 소진 → 종결 국면 # round 는 매 가격입력마다 증가(기존가격제시=1). 카운터제안이 상한을 넘거나 카드가 # 소진되면 곧장 실패가 아니라 **종결 국면**으로 처리한다(실제 MD 협상 방식): # ① 아직 종결 전술을 안 썼으면 → 가격협상으로 보내되 force_closing 마킹 # (ChatService 가 종결 전술 — 중간값 절충/최후통첩 — 을 강제 발동) # ② 종결 전술까지 소진(closing_played)이면 → 최종 제시가 ≤ target 은 타결, # 초과는 결렬(협상실패) — "목표가 초과 타결 금지" 가드레일과 정합. counter_rounds = max(0, ctx.get("round", 0) - 1) exhausted = counter_rounds >= self.rules.max_counter_rounds or (cards_total > 0 and cards_used >= cards_total) if exhausted: target = ctx.get("target_price", 0) if not ctx.get("closing_played"): ctx["force_closing"] = True return "가격협상" return "협상완료" if (target > 0 and price <= target) else c.get("next") ok = False elif cond == "default": ok = True if ok: return c.get("next") return "가격협상" def _pick_wildcard(self, session: ChatSession) -> str: """앵커가에 아주 근접(≤ anchor×wildcard_1pct_ratio)한 구간에서만 1% 인하 요청(wild_card_1pct)으로 앵커가 이하로 유도한다. 그 외 구간은 일반 가격협상(카드 플레이)으로 돌린다. 과거 여기서 반환하던 '재원부족'(wild_card_budget) 하드코딩 카드는 제거했다 — 견적에서 실제 선택한 와일드카드(중간값 절충·목표가 선제안 등)와 매핑되지 않은 채 '와일드카드를 하나라도 골랐으면' 조건만으로 발동해, 선택하지도 않은 재원부족 멘트가 노출되는 오작동이 있었다. """ ctx = session.context price = ctx.get("input_price", 0) anchor = ctx.get("anchor_price", 0) if anchor > 0 and price <= anchor * self.rules.wildcard_1pct_ratio: # 와일드카드는 실제로 노출할 때만 '사용됨'으로 마킹한다 — 가격협상으로 돌아가는 # 경우에도 마킹하면 이후 라운드에서 정당한 1% 카드까지 억제된다. ctx["wildcard_used"] = True ctx["offer_1pct"] = int(round(price * 0.99)) # 1% 인하가 (멘트 변수) ctx["pending_counter_price"] = ctx["offer_1pct"] # 수락 시 이 가격으로 타결 return "wild_card_1pct" # 1.02 초과 ~ entry(1.05) 구간: 견적에서 선택한 와일드카드의 전술로 카운터 제시. # (구현 전에는 이 구간이 일반 가격협상으로 회귀해 선택형 WC 가 영영 발동하지 않던 갭.) if anchor > 0 and price <= anchor * self.rules.wildcard_entry_ratio: for number in (ctx.get("selected_wild_card_numbers") or []): counter = compute_counter(tactic_for(str(number)), ctx) if counter is not None: ctx["wildcard_used"] = True ctx["pending_counter_price"] = counter ctx["active_wild_card_number"] = str(number) return "wild_card_dynamic" return "가격협상" # ---- render -------------------------------------------------------- def vars_for(self, session: ChatSession) -> Dict[str, Any]: """스크립트 치환 변수. 가격협상_확인 멘트(할인율)·카드 스크립트가 공유한다.""" ctx = session.context out = {} if "input_price" in ctx: out["input_price"] = int(ctx["input_price"]) # 목표가/앵커가: 엔진 내부 파일 스크립트는 {target}/{anchor}, negodata 카드 에디터는 # {target_price}/{anchor_price}(variables.ts) 를 쓴다 — 양쪽 이름 모두 채워 치환 누락 방지. if "target_price" in ctx: out["target"] = out["target_price"] = int(ctx["target_price"]) if "anchor_price" in ctx: # anchoring_price = DB 시드 기본 카드/sessions 컬럼 표기, anchor_price = 카드 에디터 표기. out["anchor"] = out["anchor_price"] = out["anchoring_price"] = int(ctx["anchor_price"]) # 카드 에디터 카탈로그의 협력사명/상품명(partner_name·product_name) 치환. if ctx.get("partner_name"): out["partner_name"] = str(ctx["partner_name"]) if ctx.get("product_name"): out["product_name"] = str(ctx["product_name"]) if "offer_1pct" in ctx: out["offer_1pct"] = int(ctx["offer_1pct"]) # 인터넷 최저가: LPS 대표값(items.internet_lowest_price). 카드는 {internet_lowest_price}, # 라벨 매핑(variable_mapping.json)은 internet_min_price 를 쓰므로 target/anchor 처럼 양쪽 이름 모두 채운다. # 미수집(0/없음)이면 키를 만들지 않는다 — 원형 유지 → 허위 시장가 인용 방지(NGC-008 은 값 있을 때만 유효). ilp = ctx.get("internet_lowest_price") or 0 if ilp > 0: out["internet_lowest_price"] = out["internet_min_price"] = int(ilp) # 전술 카운터 변수(카드 시드 멘트의 가격 변수) — tactics.compute_counter 산식과 동일 정의. anchor = ctx.get("anchor_price") or 0 target = ctx.get("target_price") or 0 if "input_price" in ctx: out["prev_partner_price"] = int(ctx.get("prev_partner_price") or ctx["input_price"]) prev_customer = ctx.get("prev_customer_price") or anchor if prev_customer: out["prev_customer_price"] = int(prev_customer) if anchor and target: out["target_mid_price"] = int(round((anchor + target) / 2)) if prev_customer and "input_price" in ctx: out["middle_price"] = int(round((prev_customer + ctx["input_price"]) / 2)) if ctx.get("pending_counter_price"): out["counter_price"] = int(ctx["pending_counter_price"]) # 인하율 = (기존 공급가(상품단가) - 제시가) / 기존 공급가 * 100. 기존가 없으면 미표시(0.0). # 제시가가 기존가보다 높으면(인상 제시) 음수가 나오는데, "-1.3% 인하된 금액" 같은 # 모순 표현이 되므로 discount_rate 는 0 미만 금지하고, 인상/동일/인하를 구분한 # 문구는 discount_phrase 로 별도 제공한다(가격협상_확인 멘트가 사용). base = ctx.get("item_price") or 0 if base > 1 and "input_price" in ctx: rate = ((base - ctx["input_price"]) / base) * 100 out["discount_rate"] = f"{max(0.0, rate):.1f}" if rate >= 0.05: out["discount_phrase"] = f"기존 공급가 대비 약 **{rate:.1f}%** 인하된 금액입니다. " elif rate <= -0.05: out["discount_phrase"] = ( f"기존 공급가(**{int(base)}원**)보다 약 **{abs(rate):.1f}%** 높은 금액입니다. ") else: out["discount_phrase"] = "기존 공급가와 동일한 수준의 금액입니다. " else: out["discount_rate"] = "0.0" out["discount_phrase"] = "" return out def _vars(self, session: ChatSession) -> Dict[str, Any]: return self.vars_for(session) def render_step(self, session: ChatSession, step_key: str) -> StepView: """지정 스텝으로 전이·렌더 (공개) — ChatService 가 카드 카운터 제시 시 가격협상 → 가격협상_카운터로 스텝을 전환할 때 사용한다.""" return self._render(session, step_key) 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}") # 가드레일(최후 방어선): 구매자 대리는 목표가 초과로 절대 타결하지 않는다. # 카운터 클램프·종결 규칙이 정상이면 도달하지 않지만, 스크립트 편집 실수 등으로 # 성공 스텝에 초과가로 진입하면 결렬로 강제 전환한다. (재협상 흐름 한정) if step_key in _SUCCESS_STEPS and self.rq_type == "재협상": ctx = session.context target = ctx.get("target_price") or 0 if target > 0 and ctx.get("input_price", 0) > target: step_key = "협상실패" node = self.scripts[step_key] session.step = step_key chat_end = bool(node.get("chat_end")) session.ended = chat_end # 최종 성공/실패를 통과 시점에 기록하고, 실제 종료(chat_end) 시점에만 outcome 으로 확정 보고. # (중간 성공/실패 스텝에서 보고하면 backend 종료확정 타이밍(chat_end)과 어긋나고, # ChatService 종료학습도 두 번 도는 문제가 생긴다.) if step_key in _SUCCESS_STEPS: session.context["final_outcome"] = "success" elif step_key in _FAILURE_STEPS: session.context["final_outcome"] = "failure" outcome = session.context.get("final_outcome") if chat_end 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, )