Files
exAichatbot_agent/scripts/quick_inquiry_branch_test.py
T
2026-08-18 09:26:35 +09:00

527 lines
21 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""간편조회 파라미터 분기(폼/부분선택/상세/위치/별칭) 구어체 테스트 → test.csv
운영 예:
PYTHONUNBUFFERED=1 python3 scripts/quick_inquiry_branch_test.py \\
--base https://chatbot.ex.co.kr/chatbot --insecure \\
--out test.csv --sleep-ask 2.0 --sleep-domain 1.2
"""
from __future__ import annotations
import argparse
import csv
import json
import re
import ssl
import time
import urllib.error
import urllib.request
from collections import Counter
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple
_SSL_CTX = ssl.create_default_context()
FIELDNAMES = [
"tested_at",
"feature",
"param_tier", # none | partial | full | alias | location
"expected_mode", # form | detail | clarify | list | location
"expected_intent",
"mode", # ask | domain-query
"question",
"params",
"http_status",
"elapsed_sec",
"actual_intent",
"actual_mode",
"api_status",
"route_type",
"form_flags",
"answer",
"result",
"reason",
"note",
]
def one_line(text: str, limit: int = 260) -> str:
t = re.sub(r"\s+", " ", (text or "").strip())
return (t[: limit - 1] + "…") if len(t) > limit else t
def post_json(url: str, body: Dict[str, Any], timeout: float) -> Tuple[Optional[int], Any, float]:
t0 = time.time()
data = json.dumps(body, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(
url, data=data, headers={"Content-Type": "application/json; charset=utf-8"}
)
try:
with urllib.request.urlopen(req, timeout=timeout, context=_SSL_CTX) as resp:
raw = resp.read().decode("utf-8", errors="replace")
try:
return resp.status, json.loads(raw), time.time() - t0
except json.JSONDecodeError:
return resp.status, raw, time.time() - t0
except urllib.error.HTTPError as e:
try:
payload = e.read().decode("utf-8", errors="replace")
try:
return e.code, json.loads(payload), time.time() - t0
except Exception:
return e.code, payload, time.time() - t0
except Exception:
return e.code, str(e), time.time() - t0
except Exception as e:
return None, str(e), time.time() - t0
def extract_answer(body: Any) -> str:
if not isinstance(body, dict):
return one_line(str(body))
for key in ("answer", "message", "text", "statusMsg", "llmSummary"):
val = body.get(key)
if isinstance(val, str) and val.strip():
return one_line(val)
dd = body.get("domainData")
if isinstance(dd, dict):
for key in ("llmSummary", "statusMsg", "answer"):
val = dd.get(key)
if isinstance(val, str) and val.strip():
return one_line(val)
return one_line(json.dumps(body, ensure_ascii=False))
def form_flags(body: Any) -> str:
flags = []
if not isinstance(body, dict):
return ""
dd = body.get("domainData") if isinstance(body.get("domainData"), dict) else {}
raw = body.get("rawData") if isinstance(body.get("rawData"), dict) else {}
raw_dd = raw.get("domainData") if isinstance(raw.get("domainData"), dict) else {}
for src in (body, dd, raw, raw_dd):
for k, v in src.items():
if str(k).startswith("needs") and v:
flags.append(f"{k}={v}")
if src.get("needsClarification"):
flags.append("needsClarification=True")
if src.get("listMode"):
flags.append(f"listMode={src.get('listMode')}")
# dedupe
seen = set()
out = []
for f in flags:
if f not in seen:
seen.add(f)
out.append(f)
return ";".join(out)
def detect_mode(body: Any, answer: str) -> str:
"""응답을 form/detail/clarify/list/location 으로 분류."""
if not isinstance(body, dict):
return "unknown"
flags = form_flags(body)
route = str(body.get("routeType") or "")
dd = body.get("domainData") if isinstance(body.get("domainData"), dict) else {}
ans = answer or ""
# 구체 데이터(전화·요금 등)가 있으면 form 플래그보다 detail 우선
if re.search(r"0\d{1,2}-\d{3,4}-\d{4}", ans) or "연락처입니다" in ans:
return "detail"
if route == "clarify" or "needsClarification=True" in flags:
return "clarify"
if "위치정보" in ans or ("동의" in ans and "위치" in ans):
return "location"
if any(p in ans for p in ("선택하면", "선택해 주세요", "알려주세요", "어느 노선", "어느 방향", "차량번호가 필요")):
if any(x in flags for x in ("needsRestAreaForm=True", "needsIcTelForm=True")):
return "form"
return "clarify"
if any(x in flags for x in ("needsRestAreaForm=True", "needsIcTelForm=True")):
return "form"
lst = dd.get("list") or dd.get("items")
if isinstance(lst, list) and len(lst) > 0:
return "list"
if any(
k in ans
for k in ("전화", "원", "km/h", "메뉴", "유가", "충전", "구조", "공사", "정체", "서행", "주소")
):
return "detail"
if body.get("status") is True and len(ans) > 15:
return "detail"
return "unknown"
def scenarios() -> List[Dict[str, Any]]:
"""기능 × 파라미터 단계 × 구어체/별칭/위치."""
rows: List[Dict[str, Any]] = []
def ask(
feature: str,
tier: str,
expected_mode: str,
intent: str,
question: str,
note: str = "",
) -> None:
rows.append(
{
"mode": "ask",
"feature": feature,
"param_tier": tier,
"expected_mode": expected_mode,
"expected_intent": intent,
"question": question,
"params": {},
"note": note,
}
)
def domain(
feature: str,
tier: str,
expected_mode: str,
intent: str,
question: str,
params: Dict[str, str],
note: str = "",
) -> None:
rows.append(
{
"mode": "domain-query",
"feature": feature,
"param_tier": tier,
"expected_mode": expected_mode,
"expected_intent": intent,
"question": question,
"params": params,
"note": note,
}
)
# ---- 정체/혼잡 ----
ask("노선혼잡", "none", "clarify", "TROAD_CONGESTION", "고속도로 정체구간 알려줘")
ask("노선혼잡", "none", "clarify", "TROAD_CONGESTION", "지금 어디가 막혀?")
ask("노선혼잡", "partial", "detail", "TROAD_CONGESTION", "경부선에 정체구간 있어?")
ask("노선혼잡", "partial", "detail", "TROAD_CONGESTION", "영동선 지금 막혀?")
ask("노선혼잡", "alias", "detail", "TROAD_CONGESTION", "경부고속도로 정체 어때?")
ask(
"노선혼잡",
"location",
"location",
"TROAD_CONGESTION",
"이 근처에 정체되는 구간 있어?",
"GPS 동의/근처 영업소 파생 가능",
)
domain(
"노선혼잡",
"full",
"detail",
"TROAD_CONGESTION",
"경부선 정체 상세",
{"routeName": "경부선"},
)
# ---- 영업소 정보 ----
ask("영업소정보", "none", "form", "IC_TEL", "영업소 전화번호 좀 알려줘")
ask("영업소정보", "full", "detail", "IC_TEL", "판교 톨게이트 전화번호 좀")
ask("영업소정보", "alias", "detail", "IC_TEL", "동김천 톨게이트 번호 뭐야?")
ask("영업소정보", "alias", "detail", "IC_TEL", "동김천ic 연락처 알려줘")
ask("영업소정보", "alias", "detail", "IC_TEL", "동감천 톨게이트 전화 있어?") # 오표기
ask("영업소정보", "alias", "detail", "IC_TEL", "판교영업소 대표번호")
ask(
"영업소정보",
"location",
"location",
"IC_TEL",
"여기 근처 영업소 전화번호 알려줘",
"위치기반 폼/상세",
)
domain("영업소정보", "none", "form", "IC_TEL", "영업소 조회 폼", {})
domain("영업소정보", "full", "detail", "IC_TEL", "동김천 상세", {"icName": "동김천"})
# ---- 통행요금 ----
ask("구간통행요금", "none", "clarify", "FARE_SEARCH", "톨비 얼마야?")
ask("구간통행요금", "partial", "clarify", "FARE_SEARCH", "서울에서 출발하면 요금 얼마야?")
ask("구간통행요금", "full", "detail", "FARE_SEARCH", "서울에서 부산까지 톨비 얼마 나와?")
ask("구간통행요금", "alias", "detail", "FARE_SEARCH", "판교IC에서 신갈 톨게이트까지 요금 어때?")
ask("구간통행요금", "alias", "detail", "FARE_SEARCH", "판교 영업소에서 수원까지 하이패스 얼마야?")
# ---- 통행시간 ----
ask("구간통행시간", "none", "clarify", "TROAD_TRAVEL_TIME", "지금 고속도로 얼마나 걸려?")
ask("구간통행시간", "full", "detail", "TROAD_TRAVEL_TIME", "서울에서 부산까지 지금 가면 얼마나 걸려?")
ask("구간통행시간", "alias", "detail", "TROAD_TRAVEL_TIME", "판교에서 신갈 지금 막혀? 몇 분 걸려?")
# ---- 교통속보 ----
ask("노선별교통속보", "none", "clarify", "TROAD_BREAKING_NEWS", "고속도로에 사고 났다는데 어디야?")
ask("노선별교통속보", "partial", "detail", "TROAD_BREAKING_NEWS", "경부선에 사고 났어?")
ask("노선별교통속보", "alias", "detail", "TROAD_BREAKING_NEWS", "영동고속도로 교통속보 있어?")
# ---- 차단계획 ----
ask("차단계획", "none", "clarify", "CONSTRUCTION_PLAN", "고속도로 공사하는 데 있어?")
ask("차단계획", "partial", "detail", "CONSTRUCTION_PLAN", "경부선 오늘 공사해서 막히는 데 있어?")
ask("차단계획", "alias", "detail", "CONSTRUCTION_PLAN", "경부고속도로 차로 막는 일정 있나?")
domain(
"차단계획",
"partial",
"detail",
"CONSTRUCTION_PLAN",
"영동선 차단계획",
{"routeName": "영동선"},
)
# ---- 긴급구조 ----
ask("긴급구조기관", "none", "form", "EMERGENCY_HELP_ORG", "고속도로에서 사고 나면 어디로 연락해?")
ask("긴급구조기관", "partial", "form", "EMERGENCY_HELP_ORG", "경부선 긴급구조기관 알려줘")
ask("긴급구조기관", "alias", "form", "EMERGENCY_HELP_ORG", "경부고속도로 긴급구조대 어디야?")
domain("긴급구조기관", "none", "form", "EMERGENCY_HELP_ORG", "긴급구조 폼", {})
domain(
"긴급구조기관",
"partial",
"form",
"EMERGENCY_HELP_ORG",
"경부선 선택 폼",
{"routeName": "경부선"},
)
# ---- 휴게소 음식/브랜드/시설 ----
ask("휴게소음식", "none", "clarify", "REST_AREA_FOOD_LIST", "휴게소에서 뭐 먹지?")
ask("휴게소음식", "full", "detail", "REST_AREA_FOOD_LIST", "죽전휴게소에서 뭐 먹을 거 있어?")
ask("휴게소음식", "alias", "detail", "REST_AREA_FOOD_LIST", "죽전 휴게소 밥 뭐 파나?")
ask("휴게소브랜드매장", "none", "clarify", "BRAND_SHOP", "휴게소에 카페 있어?")
ask("휴게소브랜드매장", "full", "detail", "BRAND_SHOP", "김천휴게소에 브랜드 매장 뭐 있어?")
ask("휴게소브랜드매장", "alias", "detail", "BRAND_SHOP", "안성 휴게소에 편의점 있나?")
ask("휴게소시설", "none", "clarify", "REST_AREA_FACILITY", "휴게소에 수유실 있어?")
ask("휴게소시설", "full", "detail", "REST_AREA_FACILITY", "죽전휴게소에 수유실 있나?")
ask("휴게소시설", "alias", "detail", "REST_AREA_FACILITY", "안성SA 샤워실 쓸 수 있어?")
# ---- 주유 ----
ask("휴게소주유", "none", "form", "GASSTATION", "고속도로 기름값 얼마야?")
ask("휴게소주유", "full", "detail", "GASSTATION", "죽전휴게소 기름값 얼마야?")
ask("휴게소주유", "partial", "detail", "GASSTATION", "경부선에서 기름 제일 싼 데 어디야?")
ask("휴게소주유", "alias", "detail", "GASSTATION", "안성 휴게소 휘발유값 어때?")
domain("휴게소주유", "none", "form", "GASSTATION", "주유 폼", {})
# ---- 충전 ----
ask("휴게소충전소", "none", "form", "REST_AREA_CHARGING", "고속도로에 전기차 충전소 있어?")
ask("휴게소충전소", "full", "detail", "REST_AREA_CHARGING", "죽전에서 전기차 충전돼?")
ask("휴게소충전소", "partial", "form", "REST_AREA_CHARGING", "경부선에 충전되는 휴게소 어디 있어?")
ask("휴게소충전소", "alias", "detail", "REST_AREA_CHARGING", "강릉휴게소충전소 있어?")
ask("휴게소충전소", "alias", "detail", "REST_AREA_CHARGING", "김천 휴게소에서 수소차 충전해?")
domain("휴게소충전소", "none", "form", "REST_AREA_CHARGING", "충전 폼", {})
domain(
"휴게소충전소",
"partial",
"form",
"REST_AREA_CHARGING",
"경부선 충전 폼",
{"routeName": "경부선"},
)
# ---- 청년창업 ----
ask("휴게소청년창업매장", "none", "form", "REST_AREA_YOUTH_SHOP", "고속도로에 청년창업 매장 있어?")
ask("휴게소청년창업매장", "full", "detail", "REST_AREA_YOUTH_SHOP", "죽전휴게소에 청년들 장사하는 가게 있어?")
ask(
"휴게소청년창업매장",
"alias",
"detail",
"REST_AREA_YOUTH_SHOP",
"안성 휴게소 청년몰 있나?",
)
# ---- 가까운 휴게소 ----
ask("가까운휴게소", "partial", "detail", "REST_AREA_NEAREST", "판교 근처 휴게소 어디 있어?")
ask(
"가까운휴게소",
"location",
"location",
"REST_AREA_NEAREST",
"여기 제일 가까운 휴게소 어디야?",
"GPS 경로",
)
ask("가까운휴게소", "alias", "detail", "REST_AREA_NEAREST", "기흥IC 근처에 쉴 데 있어?")
# ---- 미납/환불 (파라미터=차량번호) ----
ask("미납조회", "none", "clarify", "FARE_UNPAID", "미납요금 조회해줘")
ask("미납조회", "full", "detail", "FARE_UNPAID", "야 내 차 12가3456인데 미납 있어?")
ask("환불조회", "none", "clarify", "FARE_REFUND", "톨비 환불받을 거 있어?")
ask("환불조회", "full", "detail", "FARE_REFUND", "하이패스 환불금 들어왔나? 서울33하1122")
return rows
def mode_match(expected: str, actual: str) -> bool:
if expected == actual:
return True
# form↔clarify 는 둘 다 '추가 선택 유도'로 허용
if {expected, actual} <= {"form", "clarify"}:
return True
# location 질문은 form/clarify/detail 도 허용(클라이언트 GPS 없을 때)
if expected == "location" and actual in {"location", "form", "clarify", "detail"}:
return True
# list 는 detail 상위 개념으로 허용
if expected == "detail" and actual in {"detail", "list"}:
return True
if expected == "list" and actual in {"list", "form", "detail"}:
return True
return False
def intent_ok(expected: str, actual: str) -> bool:
if not actual:
return False
if actual == expected:
return True
aliases = {
"TROAD_BREAKING_NEWS": {"TROAD_BREAKING_NEWS", "TRAFFIC_NEWS"},
"TROAD_CONGESTION": {"TROAD_CONGESTION"},
"BRAND_SHOP": {"BRAND_SHOP", "REST_AREA_FACILITY"}, # 편의점 질문은 시설로도 갈 수 있음
}
return actual in aliases.get(expected, {expected})
def judge(
expected_intent: str,
actual_intent: str,
expected_mode: str,
actual_mode: str,
http_st: Optional[int],
answer: str,
) -> Tuple[str, str]:
if http_st != 200:
return "FAIL", f"HTTP {http_st}"
natural = len(answer or "") >= 6
i_ok = intent_ok(expected_intent, actual_intent)
m_ok = mode_match(expected_mode, actual_mode)
if i_ok and m_ok and natural:
return "PASS", "의도·응답모드 적합"
if i_ok and natural:
return "PARTIAL", f"모드상이 expected={expected_mode} actual={actual_mode}"
if m_ok and natural:
return "PARTIAL", f"의도상이 expected={expected_intent} actual={actual_intent or '-'}"
if natural:
return "PARTIAL", "응답만 존재"
return "FAIL", "응답 불량"
def append_row(path: str, row: Dict[str, str], write_header: bool) -> None:
with open(path, "a", newline="", encoding="utf-8-sig") as f:
w = csv.DictWriter(f, fieldnames=FIELDNAMES)
if write_header:
w.writeheader()
w.writerow(row)
f.flush()
def main() -> int:
global _SSL_CTX
ap = argparse.ArgumentParser()
ap.add_argument("--base", default="https://chatbot.ex.co.kr/chatbot")
ap.add_argument("--out", default="test.csv")
ap.add_argument("--timeout-ask", type=float, default=90.0)
ap.add_argument("--timeout-domain", type=float, default=45.0)
ap.add_argument("--sleep-ask", type=float, default=2.0)
ap.add_argument("--sleep-domain", type=float, default=1.2)
ap.add_argument("--insecure", action="store_true")
ap.add_argument("--limit", type=int, default=0)
ap.add_argument("--skip-domain", action="store_true")
ap.add_argument("--skip-ask", action="store_true")
args = ap.parse_args()
if args.insecure:
_SSL_CTX = ssl._create_unverified_context()
base = args.base.rstrip("/")
ask_url = base + "/ask"
domain_url = base + "/domain-query"
started = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
open(args.out, "w", encoding="utf-8-sig").close()
header = True
results: List[str] = []
suite = scenarios()
if args.skip_ask:
suite = [c for c in suite if c["mode"] != "ask"]
if args.skip_domain:
suite = [c for c in suite if c["mode"] != "domain-query"]
if args.limit > 0:
suite = suite[: args.limit]
total = len(suite)
for i, case in enumerate(suite, 1):
print(f"({i}/{total}) {case['mode']} {case['feature']} [{case['param_tier']}] {case['question'][:40]}", flush=True)
if case["mode"] == "ask":
http_st, body, elapsed = post_json(
ask_url,
{"query": case["question"], "botId": f"branch-ask-{i}"},
args.timeout_ask,
)
sleep = args.sleep_ask
else:
http_st, body, elapsed = post_json(
domain_url,
{
"intentType": case["expected_intent"],
"params": case["params"],
"botId": f"branch-dom-{i}",
},
args.timeout_domain,
)
sleep = args.sleep_domain
answer = extract_answer(body)
actual_intent = ""
api_status = ""
route_type = ""
if isinstance(body, dict):
actual_intent = str(body.get("intentType") or "")
api_status = str(body.get("status"))
route_type = str(body.get("routeType") or "")
if case["mode"] == "domain-query" and not actual_intent:
actual_intent = case["expected_intent"]
actual_mode = detect_mode(body, answer)
result, reason = judge(
case["expected_intent"],
actual_intent,
case["expected_mode"],
actual_mode,
http_st,
answer,
)
row = {
"tested_at": started,
"feature": case["feature"],
"param_tier": case["param_tier"],
"expected_mode": case["expected_mode"],
"expected_intent": case["expected_intent"],
"mode": case["mode"],
"question": case["question"],
"params": json.dumps(case.get("params") or {}, ensure_ascii=False),
"http_status": str(http_st or ""),
"elapsed_sec": f"{elapsed:.2f}",
"actual_intent": actual_intent,
"actual_mode": actual_mode,
"api_status": api_status,
"route_type": route_type,
"form_flags": form_flags(body),
"answer": answer,
"result": result,
"reason": reason,
"note": case.get("note") or "",
}
append_row(args.out, row, header)
header = False
results.append(result)
print(f" → {result:7} intent={actual_intent or '-':22} mode={actual_mode:8} | {answer[:70]}", flush=True)
time.sleep(sleep)
print("---", flush=True)
print(f"saved={args.out} total={len(results)} {dict(Counter(results))}", flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())