249 lines
12 KiB
Python
249 lines
12 KiB
Python
"""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")
|
|
|
|
# 최종 타결/결렬 스텝. 재협상=협상완료(우선협상 타결), 재견적=결과제출(투찰확정). 둘 다 협상실패=결렬.
|
|
# 이 스텝들은 chat_end=False(뒤에 협상종료가 옴)라, outcome 을 컨텍스트에 적재했다가
|
|
# 실제 종료(chat_end=협상종료) 시점에 확정 보고한다 → backend 가 chat_end 에서 DONE/REJECTED 를 옳게 가른다.
|
|
_SUCCESS_STEPS = ("협상완료", "결과제출")
|
|
_FAILURE_STEPS = ("협상실패",)
|
|
|
|
# 프론트는 표시용 문자열로 가격을 보낸다(예: "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.setdefault("first_offer_price", price)
|
|
session.context["round"] = session.context.get("round", 0) + 1
|
|
nxt = self._default_next(node)
|
|
elif mode in _CHOICE_MODES:
|
|
# 와일드카드 1% 인하 제안을 수락("예")하면 합의가를 제안가(offer_1pct)로 확정한다.
|
|
# (멘트에만 쓰이던 offer_1pct 가 input_price 에 반영되지 않아, 요약/입찰가가
|
|
# 직전 제시가로 잡히던 버그 수정 — 수락 시 실제 합의가는 인하가다.)
|
|
if session.step == "wild_card_1pct" and user_input == "예" and session.context.get("offer_1pct"):
|
|
session.context["input_price"] = float(session.context["offer_1pct"])
|
|
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 > 0
|
|
and anchor < price
|
|
and (
|
|
price <= anchor * 1.02
|
|
or (bool(ctx.get("allow_selected_wildcards", True)) and 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": # 협상 라운드 상한 또는 카드 소진
|
|
# round 는 매 가격입력마다 증가(기존가격제시=1). 그 이후 카운터제안이 MAX_ROUNDS 회를
|
|
# 넘으면 종료한다. 카드선택(RL)이 실패(state ValueError)해도 used_action_ids 가 안 늘어
|
|
# 카드 소진 조건만으로는 종료되지 않으므로, 라운드 상한을 독립적으로 둬 무한 가격입력을 막는다.
|
|
# (선행 chat_server 의 `iteration >= 3` 와 동일한 안전장치.)
|
|
counter_rounds = max(0, ctx.get("round", 0) - 1)
|
|
ok = counter_rounds >= MAX_ROUNDS or (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"
|
|
if not ctx.get("allow_selected_wildcards", True):
|
|
return "가격협상"
|
|
return "wild_card_budget"
|
|
|
|
# ---- 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"])
|
|
if "target_price" in ctx:
|
|
out["target"] = int(ctx["target_price"])
|
|
if "anchor_price" in ctx:
|
|
out["anchor"] = int(ctx["anchor_price"])
|
|
if "offer_1pct" in ctx:
|
|
out["offer_1pct"] = int(ctx["offer_1pct"])
|
|
# 인하율 = (기존 공급가 - 제시가) / 기존 공급가 * 100. 기존가 없으면 미표시(0.0).
|
|
base = ctx.get("item_price") or 0
|
|
if base > 1 and "input_price" in ctx:
|
|
out["discount_rate"] = f"{((base - ctx['input_price']) / base) * 100:.1f}"
|
|
else:
|
|
out["discount_rate"] = "0.0"
|
|
return out
|
|
|
|
def _vars(self, session: ChatSession) -> Dict[str, Any]:
|
|
return self.vars_for(session)
|
|
|
|
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
|
|
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,
|
|
)
|