"""Phase 3 이해층 검증 — InputInterpreter (자유 발화 NLU → 기대 입력 구조화). 원칙 검증: LLM 은 의도 분류·가격 표현 위치만 찾고, 숫자 계산은 결정론 파서가 한다. ① 한국어 가격 파서 결정론 ② choice 는 선택지 목록 검증 ③ price_text 는 원문 부분문자열 검증 ④ 실패/타임아웃 → None(원문 폴백) ⑤ 챗 플로우: 자유 발화로 분기·가격 입력이 진행된다 ⑥ LLM 미설정 시 기존(정형 입력) 동작 그대로 — 회귀 없음. """ import os import pytest from negotiation.chat.service.input_interpreter import ( InputInterpreter, InterpretedInput, parse_korean_price, ) _TENANTS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tenants") # ---- 결정론 한국어 가격 파서 ------------------------------------------------- @pytest.mark.parametrize("text,expected", [ ("10,500원", 10500), ("10500", 10500), ("1만 500원", 10500), ("1만500원", 10500), ("만원", 10000), ("1.5만", 15000), ("3만2천원", 32000), ("2억", 200_000_000), ("0", None), # 0 이하 무효 ("그건 어렵습니다", None), # 가격 아님 ("만원에 3개", None), # 잡문자 혼입 → 해석 불가(안전 폴백) ]) def test_parse_korean_price(text, expected): got = parse_korean_price(text) assert (got == expected) if expected is not None else (got is None) # ---- LLM 출력 검증 (환각 차단) ---------------------------------------------- def _fake(reply: dict): def call(messages): return reply return call @pytest.mark.asyncio async def test_choice_mapped_to_option(): nat = InputInterpreter(llm_call=_fake({"intent": "choice", "choice": "예", "price_text": None})) out = await nat.interpret("네 접니다, 말씀하세요", input_mode="yes_no", input_options=["예", "아니오"]) assert out == InterpretedInput(kind="choice", value="예", source="예") @pytest.mark.asyncio async def test_choice_outside_options_rejected(): nat = InputInterpreter(llm_call=_fake({"intent": "choice", "choice": "글쎄요", "price_text": None})) assert await nat.interpret("음...", input_mode="yes_no", input_options=["예", "아니오"]) is None @pytest.mark.asyncio async def test_price_span_verified_and_parsed_deterministically(): nat = InputInterpreter(llm_call=_fake({"intent": "price", "choice": None, "price_text": "1만 500원"})) out = await nat.interpret("저희 마진상 1만 500원까지는 맞춰드릴 수 있습니다", input_mode="price") assert out is not None and out.kind == "price" assert out.value == "10500" # 숫자는 결정론 파서 산출(LLM 계산 아님) assert out.source == "1만 500원" @pytest.mark.asyncio async def test_price_span_not_in_text_rejected(): """LLM 이 원문에 없는 가격 표현을 지어내면 폐기(환각 차단).""" nat = InputInterpreter(llm_call=_fake({"intent": "price", "choice": None, "price_text": "9,000원"})) assert await nat.interpret("만원이면 가능합니다", input_mode="price") is None @pytest.mark.asyncio async def test_unknown_and_failures_fall_back(): assert await InputInterpreter(llm_call=_fake({"intent": "unknown"})).interpret( "글쎄요 검토해 볼게요", input_mode="yes_no", input_options=["예", "아니오"]) is None def boom(messages): raise RuntimeError("LLM down") assert await InputInterpreter(llm_call=boom).interpret("네", input_mode="yes_no", input_options=["예"]) is None def slow(messages): import time time.sleep(0.5) return {"intent": "choice", "choice": "예"} nat = InputInterpreter(llm_call=slow, timeout_seconds=0.05) assert await nat.interpret("네", input_mode="yes_no", input_options=["예"]) is None # ---- 챗 플로우 E2E (fake LLM) ------------------------------------------------ def _routing_fake(messages): """단계별 fake — 기대 입력이 가격이면 price, 아니면 '예' choice 로 응답.""" user = messages[-1]["content"] if "가격(숫자)" in user: return {"intent": "price", "choice": None, "price_text": "1만 500원"} return {"intent": "choice", "choice": "예", "price_text": None} @pytest.mark.asyncio async def test_chat_flow_free_text_negotiation(db_engine): """자유 발화만으로 담당자확인 분기 + 가격 입력이 진행된다 (버튼 없는 '진짜 대화').""" from router.v1.chat.protocol import Req_Chat from services.chat_service import ChatService, reset_sessions from tenancy.config_loader import TenantConfigLoader from tenancy.registry import TenantEngineRegistry reset_sessions() reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0)) eng = await reg.get_engine("imarketkorea") eng.config.llm.enabled = True svc = ChatService() svc._interpreter = InputInterpreter(llm_call=_routing_fake) orig = InputInterpreter.available InputInterpreter.available = staticmethod(lambda: True) try: sid = None r = await svc.chat(eng, Req_Chat(session_id=sid)) # 서비스안내 sid = r.session_id r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="확인")) # 담당자확인 (fast path) assert r.step == "담당자확인" and r.interpreted_input is None # 자유 발화 → NLU 가 "예" 로 매핑 → 협상품목안내 r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="네 접니다, 말씀하세요")) assert r.step == "협상품목안내" assert r.interpreted_input == "예" r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="확인")) # 기존가격제시(price) # 자유 발화 가격 → span "1만 500원" → 결정론 파서 10500 → 가격 저장 후 확인 단계 r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="저희 마진상 1만 500원까지는 맞춰드릴 수 있습니다")) assert r.step == "가격협상_확인" assert r.interpreted_input == "10500" assert "10500" in r.script # 멘트 치환도 해석된 가격으로 finally: InputInterpreter.available = orig eng.config.llm.enabled = False @pytest.mark.asyncio async def test_chat_flow_without_llm_keeps_legacy_behavior(db_engine): """LLM 미설정(available=False)이면 자유 발화는 원문 그대로 엔진에 전달 — 기존 동작 회귀 없음.""" from router.v1.chat.protocol import Req_Chat from services.chat_service import ChatService, reset_sessions from tenancy.config_loader import TenantConfigLoader from tenancy.registry import TenantEngineRegistry reset_sessions() reg = TenantEngineRegistry(loader=TenantConfigLoader(tenants_dir=_TENANTS_DIR, cache_ttl_seconds=0)) eng = await reg.get_engine("imarketkorea") svc = ChatService() sid = None r = await svc.chat(eng, Req_Chat(session_id=sid)) sid = r.session_id r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="확인")) assert r.step == "담당자확인" # conftest 가드로 available=False → NLU 미동작, interpreted_input 없음 r = await svc.chat(eng, Req_Chat(session_id=sid, user_input="네 접니다")) assert r.interpreted_input is None