Compare commits

...

8 Commits

Author SHA1 Message Date
Macbook db5623341a Bump chatbotApp submodule to include opinion like/dislike picker.
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-21 09:32:27 +09:00
Macbook d389266de6 Deploy: exdev sync 2026-08-21 09:18:44 2026-08-21 09:18:44 +09:00
Macbook 8d0a8bdb5c Deploy: exdev sync 2026-08-18 09:26:35 2026-08-18 09:26:35 +09:00
Macbook 88221b828e chore: chatbotApp-test 서브모듈 추가 및 git_fetch_all 대상 확장
KoreaExpressWayCorp/chatbotApp-test 를 submodule 로 등록하고,
git_fetch_all.sh 가 chatbot2.0-agent 브랜치로 GitHub 동기화·push 하도록 한다.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-14 10:09:53 +09:00
bkyungee c9f5915419 Revert "Deploy: exdev sync 2026-08-13 17:39:39"
This reverts commit 50c5039ec0.
2026-08-13 17:51:39 +09:00
bkyungee 50c5039ec0 Deploy: exdev sync 2026-08-13 17:39:39 2026-08-13 17:39:39 +09:00
bkyungee cd6a081a55 chore: chatbotApp 서브모듈 포인터를 44bbe25 로 갱신
아이폰에서 영업소 아이콘이 보이지 않던 문제를 고친 커밋을 가리키도록
루트가 기억하는 위치를 옮긴다. 이걸 갱신하지 않으면 루트를 clone 해서
git submodule update 로 받는 쪽은 수정 이전 버전을 받게 된다.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-13 17:22:32 +09:00
bkyungee 78d549194c fix: 원격 배포 rsync 제외 패턴이 소스를 잘못 걸러내던 문제 수정
tar 기반 rsync 대체 스크립트가 rsync 의 'node_modules/' 문법을 매칭하지
못해 node_modules 를 통째로 전송하다 메모리 부족으로 죽었다. 반대로
슬래시만 떼면 'build' 가 자바 패키지 kakao.response.build 까지 제외해
kakaoChatbotSkill 컴파일이 깨졌다. 선행 슬래시를 전송 루트 기준 고정으로
해석하도록 구분해 둘 다 해결한다.

배포 중 자동 커밋에 딸려 들어갔던 일회용 진단 스크립트도 함께 제거한다.
2026-08-13 15:07:36 +09:00
16 changed files with 1334 additions and 201 deletions
+4
View File
@@ -14,3 +14,7 @@
path = kakaoChatbotSkill-chatbot2.0-agent path = kakaoChatbotSkill-chatbot2.0-agent
url = https://github.com/KoreaExpressWayCorp/kakaoChatbotSkill.git url = https://github.com/KoreaExpressWayCorp/kakaoChatbotSkill.git
branch = chatbot2.0-agent branch = chatbot2.0-agent
[submodule "chatbotApp-test"]
path = chatbotApp-test
url = https://github.com/KoreaExpressWayCorp/chatbotApp-test.git
branch = chatbot2.0-agent
+1
Submodule chatbotApp-test added at 20835ab277
+4 -2
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env bash #!/usr/bin/env bash
# 4개 submodule: 로컬 변경 → 공식 GitHub chatbot2.0-agent commit + push # 5개 submodule: 로컬 변경 → 공식 GitHub chatbot2.0-agent commit + push
# #
# ★ 공식 GitHub(KoreaExpressWayCorp) 반영은 이 스크립트로만 수행한다. # ★ 공식 GitHub(KoreaExpressWayCorp) 반영은 이 스크립트로만 수행한다.
# start-all-server.sh / push-to-gitea.sh 는 개발서버(Gitea/exdev)만 다룬다. # start-all-server.sh / push-to-gitea.sh 는 개발서버(Gitea/exdev)만 다룬다.
@@ -19,6 +19,7 @@
# 대상 (.gitmodules) — 모두 branch=chatbot2.0-agent (main 사용 안 함): # 대상 (.gitmodules) — 모두 branch=chatbot2.0-agent (main 사용 안 함):
# chatbotApi-chatbot2.0-agent # chatbotApi-chatbot2.0-agent
# chatbotApp-chatbot2.0-agent # chatbotApp-chatbot2.0-agent
# chatbotApp-test
# exAiChatBot-chatbot2.0-agent # exAiChatBot-chatbot2.0-agent
# kakaoChatbotSkill-chatbot2.0-agent # kakaoChatbotSkill-chatbot2.0-agent
set -euo pipefail set -euo pipefail
@@ -38,6 +39,7 @@ TARGET_BRANCH="chatbot2.0-agent"
TARGET_PATHS=( TARGET_PATHS=(
"chatbotApi-chatbot2.0-agent" "chatbotApi-chatbot2.0-agent"
"chatbotApp-chatbot2.0-agent" "chatbotApp-chatbot2.0-agent"
"chatbotApp-test"
"exAiChatBot-chatbot2.0-agent" "exAiChatBot-chatbot2.0-agent"
"kakaoChatbotSkill-chatbot2.0-agent" "kakaoChatbotSkill-chatbot2.0-agent"
) )
@@ -487,4 +489,4 @@ elif $DO_PUSH; then
else else
log "완료 — ${TARGET_BRANCH}: fetch/merge$($DO_COMMIT && echo ' → commit' || true) (push 생략)" log "완료 — ${TARGET_BRANCH}: fetch/merge$($DO_COMMIT && echo ' → commit' || true) (push 생략)"
fi fi
log "monorepo 포인터 갱신: git add *-chatbot2.0-agent && git status" log "monorepo 포인터 갱신: git add *-chatbot2.0-agent chatbotApp-test && git status"
-16
View File
@@ -1,16 +0,0 @@
#!/usr/bin/env bash
ssh -o BatchMode=yes -o ConnectTimeout=15 aidev@chatbot.exdev.co.kr 'bash -s' <<'EOSSH'
export PATH="/opt/homebrew/bin:/opt/homebrew/sbin:/usr/local/bin:/Applications/Docker.app/Contents/Resources/bin:$PATH"
echo "=== listening ports ==="
for p in 16000 16001 16002 27017 1521 28012 28013 8086 8083 8092 443; do
if lsof -iTCP:$p -sTCP:LISTEN >/dev/null 2>&1; then echo "UP :$p"; else echo "DOWN :$p"; fi
done
echo "=== docker ps ==="
docker ps --format '{{.Names}}\t{{.Status}}' 2>&1 | head -20
echo "=== docker compose plugin ==="
docker compose version 2>&1 | head -2
echo "=== public-ai timestamp ==="
ls -la /Users/aidev/dev/exAichatbot_agent/chatbotApp-chatbot2.0-agent/public-ai/index.html 2>/dev/null || echo "(missing)"
echo "=== HTTPS UI ==="
curl -s -o /dev/null -w 'https=%{http_code}\n' -m 10 https://chatbot.exdev.co.kr/chatbot/ai/ 2>&1
EOSSH
+7 -1
View File
@@ -105,7 +105,13 @@ TAR_EXCLUDES=()
for pat in "${EXCLUDES[@]}"; do for pat in "${EXCLUDES[@]}"; do
# tar 의 --exclude 는 'node_modules/' 처럼 슬래시로 끝나면 디렉터리에 매칭되지 않는다. # tar 의 --exclude 는 'node_modules/' 처럼 슬래시로 끝나면 디렉터리에 매칭되지 않는다.
# (rsync 문법 그대로 넘기면 node_modules 가 통째로 전송돼 메모리 부족으로 죽는다) # (rsync 문법 그대로 넘기면 node_modules 가 통째로 전송돼 메모리 부족으로 죽는다)
TAR_EXCLUDES+=(--exclude "${pat%/}") pat="${pat%/}"
if [[ "${pat}" == /* ]]; then
# rsync 의 선행 슬래시 = 전송 루트 기준 고정. tar 멤버명은 './' 로 시작한다.
TAR_EXCLUDES+=(--anchored --exclude ".${pat}")
else
TAR_EXCLUDES+=(--no-anchored --exclude "${pat}")
fi
done done
ssh -o BatchMode=yes "${HOST}" "mkdir -p '${REMOTE_PATH}'" ssh -o BatchMode=yes "${HOST}" "mkdir -p '${REMOTE_PATH}'"
-19
View File
@@ -1,19 +0,0 @@
#!/usr/bin/env bash
export PATH="/opt/homebrew/bin:/opt/homebrew/sbin:/usr/local/bin:/Applications/Docker.app/Contents/Resources/bin:/usr/bin:/bin:/usr/sbin:/sbin"
cd /Users/aidev/dev/exAichatbot_agent
git config core.autocrlf false
git config core.eol lf
python3 - << 'PYEOF'
import os
for d in ['kakaoChatbotSkill-chatbot2.0-agent','chatbotApi-chatbot2.0-agent']:
f = d+'/gradlew'
if os.path.exists(f):
data=open(f,'rb').read().replace(b'\r\n',b'\n').replace(b'\r',b'\n')
open(f,'wb').write(data)
os.chmod(f,0o755)
print('fixed:', f)
PYEOF
bash server-dev/start-all-server.sh --no-pull --no-push --skip-desktop
+526
View File
@@ -0,0 +1,526 @@
#!/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())
+626
View File
@@ -0,0 +1,626 @@
#!/usr/bin/env python3
"""간편조회 기능 다양 질문 테스트 → test.csv
- domain-query: 간편조회 UI와 동일 경로 (다수 케이스, 빠름)
- /ask: 실제 고객 자연어 질문 (기능당 소수)
사용:
PYTHONUNBUFFERED=1 python3 scripts/quick_inquiry_nlu_test.py \\
--base http://127.0.0.1:8083/web --out test.csv
"""
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
# 운영 HTTPS 등에서 로컬 CA 이슈 시 사용
_SSL_CTX = ssl.create_default_context()
def one_line(text: str, limit: int = 280) -> 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 get_json(url: str, timeout: float) -> Tuple[Optional[int], Any, float]:
t0 = time.time()
try:
with urllib.request.urlopen(url, timeout=timeout, context=_SSL_CTX) as resp:
return resp.status, json.loads(resp.read().decode("utf-8", errors="replace")), time.time() - t0
except Exception as e:
return None, str(e), time.time() - t0
def domain_cases_prod_light() -> List[Dict[str, Any]]:
"""운영용 — 기능당 1~2건만 (부하 최소화)."""
light: List[Tuple[str, str, str, Dict[str, str]]] = [
("미납조회", "FARE_UNPAID", "야 내 차 12가3456인데 미납 있어?", {"carNo": "12가3456"}),
("환불조회", "FARE_REFUND", "잘못 낸 톨비 환불받을 수 있어? 12가3456", {"carNo": "12가3456"}),
("구간통행요금", "FARE_SEARCH", "서울에서 부산까지 톨비 얼마 나와?", {"fromIc": "서울", "toIc": "부산"}),
("구간통행요금", "FARE_SEARCH", "판교에서 신갈 가면 요금이 어떻게 돼?", {"fromIc": "판교", "toIc": "신갈"}),
("구간통행시간", "TROAD_TRAVEL_TIME", "서울에서 부산까지 지금 가면 얼마나 걸려?",
{"fromIc": "서울", "toIc": "부산", "carType": "1"}),
("구간통행시간", "TROAD_TRAVEL_TIME", "판교에서 신갈 지금 막혀? 몇 분 걸려?",
{"fromIc": "판교", "toIc": "신갈", "carType": "1"}),
("차단계획", "CONSTRUCTION_PLAN", "경부선 오늘 공사해서 막히는 데 있어?", {"routeName": "경부선"}),
("차단계획", "CONSTRUCTION_PLAN", "영동선 차로 막는 일정 있나?", {"routeName": "영동선"}),
("영업소정보", "IC_TEL", "판교 톨게이트 전화번호 좀", {"icName": "판교"}),
("영업소정보", "IC_TEL", "기흥 톨게이트 주소랑 전화 좀", {"icName": "기흥"}),
("긴급구조기관", "EMERGENCY_HELP_ORG", "경부선에서 사고 나면 어디로 연락해?",
{"listMode": "true", "routeName": "경부선"}),
("휴게소음식", "REST_AREA_FOOD_LIST", "죽전휴게소에서 뭐 먹을 거 있어?", {"restAreaName": "죽전"}),
("휴게소브랜드매장", "BRAND_SHOP", "죽전에 편의점이나 카페 있어?", {"restAreaName": "죽전"}),
("휴게소시설", "REST_AREA_FACILITY", "죽전휴게소에 수유실 있나?", {"restAreaName": "죽전"}),
("휴게소주유", "GASSTATION", "죽전휴게소 기름값 얼마야?", {"restAreaName": "죽전"}),
("휴게소주유", "GASSTATION", "경부선에서 기름 제일 싼 데 어디야?",
{"compareMode": "cheapest", "fuelType": "휘발유", "routeName": "경부선"}),
("휴게소충전소", "REST_AREA_CHARGING", "죽전에서 전기차 충전돼?",
{"restAreaName": "죽전", "chargingType": "all"}),
("휴게소충전소", "REST_AREA_CHARGING", "강릉휴게소에 충전소 있어?",
{"restAreaName": "강릉", "chargingType": "all"}),
("휴게소청년창업매장", "REST_AREA_YOUTH_SHOP", "죽전휴게소에 청년들 장사하는 가게 있어?",
{"restAreaName": "죽전"}),
("가까운휴게소", "REST_AREA_NEAREST", "판교 근처 휴게소 어디 있어?", {"icName": "판교"}),
("노선혼잡", "TROAD_CONGESTION", "경부선 지금 막히는 구간 있어?", {"routeName": "경부선"}),
]
return [
{
"mode": "domain-query",
"feature": f,
"expected_intent": intent,
"question": q,
"params": params,
"note": "운영 경량 domain-query",
}
for f, intent, q, params in light
]
def extract_answer(body: Any) -> str:
if not isinstance(body, dict):
return one_line(str(body))
for key in ("answer", "message", "text", "statusMsg", "llmSummary", "formattedAnswer"):
val = body.get(key)
if isinstance(val, str) and val.strip():
return one_line(val)
for nest_key in ("domainData", "data", "rawData"):
nest = body.get(nest_key)
if isinstance(nest, dict):
for key in ("llmSummary", "statusMsg", "answer", "message"):
val = nest.get(key)
if isinstance(val, str) and val.strip():
return one_line(val)
# list summary
lst = nest.get("list") or nest.get("items")
if isinstance(lst, list) and lst:
return one_line(f"건수={len(lst)} 예시={json.dumps(lst[0], ensure_ascii=False)}")
if body.get("needsRestAreaForm") or body.get("needsIcTelForm"):
return one_line(str(body.get("statusMsg") or "폼 유도 응답"))
return one_line(json.dumps(body, ensure_ascii=False))
def intent_of(body: Any, fallback: str = "") -> str:
if not isinstance(body, dict):
return fallback
return str(body.get("intentType") or fallback)
def status_of(body: Any) -> str:
if not isinstance(body, dict):
return ""
if "status" in body:
return str(body.get("status"))
dd = body.get("domainData")
if isinstance(dd, dict) and "status" in dd:
return str(dd.get("status"))
return ""
INTENT_OK = {
"FARE_UNPAID": {"FARE_UNPAID"},
"FARE_REFUND": {"FARE_REFUND"},
"FARE_SEARCH": {"FARE_SEARCH"},
"TROAD_TRAVEL_TIME": {"TROAD_TRAVEL_TIME"},
"TROAD_BREAKING_NEWS": {"TROAD_BREAKING_NEWS", "TRAFFIC_NEWS"},
"TRAFFIC_NEWS": {"TRAFFIC_NEWS", "TROAD_BREAKING_NEWS"},
"CONSTRUCTION_PLAN": {"CONSTRUCTION_PLAN"},
"IC_TEL": {"IC_TEL"},
"EMERGENCY_HELP_ORG": {"EMERGENCY_HELP_ORG"},
"REST_AREA_FOOD_LIST": {"REST_AREA_FOOD_LIST"},
"BRAND_SHOP": {"BRAND_SHOP"},
"REST_AREA_FACILITY": {"REST_AREA_FACILITY"},
"GASSTATION": {"GASSTATION"},
"REST_AREA_CHARGING": {"REST_AREA_CHARGING"},
"REST_AREA_YOUTH_SHOP": {"REST_AREA_YOUTH_SHOP"},
"REST_AREA_NEAREST": {"REST_AREA_NEAREST"},
"TROAD_CONGESTION": {"TROAD_CONGESTION"},
}
def judge(expected: str, actual: str, status: str, answer: str, http_st: Optional[int]) -> Tuple[str, str]:
if http_st != 200:
return "FAIL", f"HTTP {http_st}"
natural = len(answer or "") >= 6 and "Traceback" not in (answer or "")
intent_ok = (actual or "") in INTENT_OK.get(expected, {expected}) or actual == expected
st_true = status in ("True", "true", "1")
infra = any(
h in (answer or "")
for h in ("일시적인 장애", "Failed to obtain", "Listener refused", "연결")
)
if intent_ok and st_true and natural:
return "PASS", "의도·응답 정상"
if intent_ok and natural and infra:
return "PARTIAL", "의도 일치, 인프라/DB 제약 안내"
if intent_ok and natural:
return "PARTIAL", "의도 일치, 데이터없음/안내성 응답"
if natural and actual:
return "PARTIAL", f"의도상이 expected={expected} actual={actual}"
if natural:
return "PARTIAL", "응답만 존재"
return "FAIL", f"의도/응답 불량 expected={expected} actual={actual or '-'}"
# ---- case builders ----
def domain_cases() -> List[Dict[str, Any]]:
"""간편조회 domain-query — 질문은 구어체, 호출은 폼/직조회 params."""
rows: List[Dict[str, Any]] = []
def add(feature: str, intent: str, question: str, params: Dict[str, str], note: str = "") -> None:
rows.append(
{
"mode": "domain-query",
"feature": feature,
"expected_intent": intent,
"question": question,
"params": params,
"note": note,
}
)
# 미납/환불
add("미납조회", "FARE_UNPAID", "야 내 차 12가3456인데 미납 있어?", {"carNo": "12가3456"}, "Oracle 미연결 시 PARTIAL 가능")
add("미납조회", "FARE_UNPAID", "하이패스 미납금 좀 봐줘 서울12가3456", {"carNo": "서울12가3456"}, "Oracle 미연결 시 PARTIAL 가능")
add("미납조회", "FARE_UNPAID", "톨비 밀린 거 있어요? 차량번호 123가4567요", {"carNo": "123가4567"}, "Oracle 미연결 시 PARTIAL 가능")
add("미납조회", "FARE_UNPAID", "98나1234 미납통행료 나와요?", {"carNo": "98나1234"}, "Oracle 미연결 시 PARTIAL 가능")
add("환불조회", "FARE_REFUND", "12가3456으로 환불받을 돈 있어요?", {"carNo": "12가3456"}, "Oracle 미연결 시 PARTIAL 가능")
add("환불조회", "FARE_REFUND", "잘못 낸 톨비 환불 내역 좀 확인해줘 서울12가3456", {"carNo": "서울12가3456"}, "Oracle 미연결 시 PARTIAL 가능")
add("환불조회", "FARE_REFUND", "과오납 환불금 나왔나? 123가4567", {"carNo": "123가4567"}, "Oracle 미연결 시 PARTIAL 가능")
add("환불조회", "FARE_REFUND", "하이패스 환불받을 거 있는지 봐줘 98나1234", {"carNo": "98나1234"}, "Oracle 미연결 시 PARTIAL 가능")
# 통행요금
fare_qs = [
("서울에서 부산까지 톨비 얼마 나와?", "서울", "부산"),
("판교에서 신갈 가면 요금이 어떻게 돼?", "판교", "신갈"),
("판교IC에서 신갈IC까지 하이패스 얼마야?", "판교IC", "신갈IC"),
("판교 영업소에서 수원까지 통행료 좀", "판교영업소", "수원"),
("대전에서 대구 가는 데 톨비 얼마쯤 해?", "대전", "대구"),
("천안에서 남이IC까지 요금 알려줄래?", "천안", "남이"),
("서울 톨게이트에서 기흥까지 얼마야?", "서울영업소", "기흥"),
("김포에서 인천 쪽 고속도로 요금 궁금해", "김포", "인천"),
]
for q, a, b in fare_qs:
add("구간통행요금", "FARE_SEARCH", q, {"fromIc": a, "toIc": b})
# 통행시간
time_qs = [
("서울에서 부산까지 지금 가면 얼마나 걸려?", "서울", "부산"),
("판교에서 신갈 지금 막혀? 몇 분 걸려?", "판교", "신갈"),
("대전에서 대구까지 지금 소요시간 어때?", "대전", "대구"),
("서울IC에서 부산IC 지금 가면 몇 시간이야?", "서울IC", "부산IC"),
("기흥에서 천안까지 지금 얼마나 걸릴까?", "기흥", "천안"),
]
for q, a, b in time_qs:
add("구간통행시간", "TROAD_TRAVEL_TIME", q, {"fromIc": a, "toIc": b, "carType": "1"})
# 차단계획
for q, route in [
("경부선 오늘 공사해서 막히는 데 있어?", "경부선"),
("영동선 차로 막는 일정 있나?", "영동선"),
("서해안선 공사구간 좀 알려줘", "서해안선"),
("호남선 이번 주 차단계획 있어?", "호남선"),
("경부고속도로 공사한다던데 어디야?", "경부고속도로"),
("중부내륙선 차로통제 있어?", "중부내륙선"),
]:
add("차단계획", "CONSTRUCTION_PLAN", q, {"routeName": route})
# 영업소
for q, ic in [
("판교 톨게이트 전화번호 좀", "판교"),
("판교IC 전화 어떻게 걸어?", "판교IC"),
("판교 영업소 연락처 알려줘", "판교영업소"),
("서울 영업소 번호 뭐야?", "서울"),
("기흥 톨게이트 주소랑 전화 좀", "기흥"),
("부산IC 영업소 정보 있어?", "부산IC"),
("신갈 영업소 전화 받나?", "신갈"),
("경부선 판교 쪽 영업소 번호 알려줘", "경부선판교"),
]:
add("영업소정보", "IC_TEL", q, {"icName": ic})
# 긴급구조
for q, route in [
("경부선에서 사고 나면 어디로 연락해?", "경부선"),
("영동선 긴급구조 가능한 데 어디야?", "영동선"),
("서해안선 긴급구조기관 좀 알려줘", "서해안선"),
]:
add("긴급구조기관", "EMERGENCY_HELP_ORG", q, {"listMode": "true", "routeName": route})
add("긴급구조기관", "EMERGENCY_HELP_ORG", "판교 근처 고속도로 긴급구조대 있어?", {"icName": "판교"})
# 휴게소
rest_sets = [
("죽전", "죽전휴게소에서 뭐 먹을 거 있어?", "죽전에 편의점이나 카페 있어?",
"죽전휴게소에 수유실 있나?", "죽전휴게소 기름값 얼마야?",
"죽전에서 전기차 충전돼?", "죽전휴게소에 청년들 장사하는 가게 있어?"),
("안성", "안성휴게소 밥 뭐 파나?", "안성휴게소에 브랜드 매장 뭐 있어?",
"안성휴게소 샤워실 쓸 수 있어?", "안성 휘발유 얼마야?",
"안성휴게소 전기차 충전소 있어?", "안성에 청년창업 매장 있어?"),
("김천", "김천휴게소에서 뭐 먹지?", "김천휴게소 입점 매장 뭐 있어?",
"김천휴게소 편의시설 뭐 있어?", "김천 경유값 어때?",
"김천휴게소에 수소차 충전해?", "김천에 청년몰 있어?"),
("금강", "금강휴게소 메뉴 추천해줘", "금강휴게소에 카페 있어?",
"금강휴게소 시설 뭐 있어?", "금강휴게소 주유소 있어?",
"금강에서 전기차 충전 가능해?", "금강휴게소 청년창업매장 있나?"),
("죽전휴게소", "죽전 휴게소 식당 뭐 있어?", "죽전 휴게소에 쇼핑할 데 있어?",
"죽전 휴게소에 애완동물 시설 있어?", "죽전 휴게소 기름 넣어?",
"죽전 휴게소 EV 충전돼?", "죽전 휴게소 청년가게 있어?"),
("강릉", "강릉휴게소에서 뭐 먹을까?", "강릉휴게소 매장 뭐 있어?",
"강릉휴게소 편의시설 알려줘", "강릉휴게소 주유소 유가 궁금해",
"강릉휴게소 충전소 있어?", "강릉에 청년창업 매장 있어?"),
]
for name, q_food, q_brand, q_fac, q_gas, q_chg, q_youth in rest_sets:
add("휴게소음식", "REST_AREA_FOOD_LIST", q_food, {"restAreaName": name})
add("휴게소브랜드매장", "BRAND_SHOP", q_brand, {"restAreaName": name})
add("휴게소시설", "REST_AREA_FACILITY", q_fac, {"restAreaName": name})
add("휴게소주유", "GASSTATION", q_gas, {"restAreaName": name})
add("휴게소충전소", "REST_AREA_CHARGING", q_chg, {"restAreaName": name, "chargingType": "all"})
add("휴게소청년창업매장", "REST_AREA_YOUTH_SHOP", q_youth, {"restAreaName": name})
add("휴게소주유", "GASSTATION", "경부선에서 기름 제일 싼 데 어디야?",
{"compareMode": "cheapest", "fuelType": "휘발유", "routeName": "경부선"})
add("휴게소주유", "GASSTATION", "고속도로에서 경유 제일 싼 곳 알려줘",
{"compareMode": "cheapest", "fuelType": "경유"})
add("휴게소충전소", "REST_AREA_CHARGING", "경부선에 전기차 충전되는 휴게소 어디 있어?",
{"listMode": "true", "routeName": "경부선", "chargingType": "electric"})
add("휴게소충전소", "REST_AREA_CHARGING", "강릉휴게소충전소 있어?",
{"restAreaName": "강릉휴게소충전소", "chargingType": "all"})
for q, ic in [
("판교 근처 휴게소 어디 있어?", "판교"),
("서울 톨게이트 지나서 바로 나오는 휴게소 어디야?", "서울"),
("기흥IC 근처에 쉴 데 있어?", "기흥IC"),
]:
add("가까운휴게소", "REST_AREA_NEAREST", q, {"icName": ic})
for q, route in [
("경부선 지금 막히는 구간 있어?", "경부선"),
("영동선 어디가 막혀?", "영동선"),
("서해안선 정체 심한 데 알려줘", "서해안선"),
]:
add("노선혼잡", "TROAD_CONGESTION", q, {"routeName": route})
return rows
def ask_cases() -> List[Dict[str, Any]]:
"""자연어 /ask — 일반 이용자 구어체 질문."""
pairs = [
("미납조회", "FARE_UNPAID", [
"야 내 차 12가3456인데 미납 있어?",
"하이패스 미납금 나왔나 봐줘 서울12가3456",
"톨비 밀린 거 있어요? 45다6789요",
]),
("환불조회", "FARE_REFUND", [
"잘못 낸 톨비 환불받을 수 있어? 12가3456",
"하이패스 환불금 들어왔나? 서울33하1122",
]),
("구간통행요금", "FARE_SEARCH", [
"서울에서 부산까지 톨비 얼마 나와?",
"판교에서 신갈 가면 요금이 어떻게 돼?",
"판교IC에서 신갈IC까지 하이패스 얼마야?",
"대전에서 대구 가는 데 톨비 얼마쯤 해?",
]),
("구간통행시간", "TROAD_TRAVEL_TIME", [
"서울에서 부산까지 지금 가면 얼마나 걸려?",
"판교에서 신갈 지금 막혀? 몇 분 걸려?",
"지금 대전에서 대구까지 가면 몇 시간이야?",
]),
("노선별교통속보", "TROAD_BREAKING_NEWS", [
"고속도로에 사고 났다는데 어디야?",
"경부선에 사고 났어?",
"영동선 지금 교통상황 어때?",
]),
("차단계획", "CONSTRUCTION_PLAN", [
"경부선 오늘 공사해서 막히는 데 있어?",
"영동선 차로 막는 일정 있나?",
"서해안선 공사구간 좀 알려줘",
]),
("영업소정보", "IC_TEL", [
"판교 톨게이트 전화번호 좀",
"기흥 톨게이트 주소랑 전화 좀",
"판교 영업소 연락처 알려줘",
]),
("긴급구조기관", "EMERGENCY_HELP_ORG", [
"경부선에서 사고 나면 어디로 연락해?",
"고속도로에서 사고 났는데 가까운 구조대 어디야?",
]),
("휴게소음식", "REST_AREA_FOOD_LIST", [
"죽전휴게소에서 뭐 먹을 거 있어?",
"안성휴게소 밥 뭐 파나?",
]),
("휴게소브랜드매장", "BRAND_SHOP", [
"죽전에 편의점이나 카페 있어?",
"김천휴게소에 브랜드 매장 뭐 있어?",
]),
("휴게소시설", "REST_AREA_FACILITY", [
"죽전휴게소에 수유실 있나?",
"안성휴게소 샤워실 쓸 수 있어?",
]),
("휴게소주유", "GASSTATION", [
"죽전휴게소 기름값 얼마야?",
"경부선에서 기름 제일 싼 데 어디야?",
"고속도로에서 휘발유 제일 싼 곳 알려줘",
]),
("휴게소충전소", "REST_AREA_CHARGING", [
"죽전에서 전기차 충전돼?",
"강릉휴게소에 충전소 있어?",
"김천휴게소에서 수소차 충전해?",
]),
("휴게소청년창업매장", "REST_AREA_YOUTH_SHOP", [
"죽전휴게소에 청년들 장사하는 가게 있어?",
"고속도로에 청년창업 매장 있는 휴게소 어디야?",
]),
("가까운휴게소", "REST_AREA_NEAREST", [
"판교 근처 휴게소 어디 있어?",
"서울 톨게이트 지나서 바로 나오는 휴게소 어디야?",
]),
("노선혼잡", "TROAD_CONGESTION", [
"경부선 지금 막히는 구간 있어?",
"영동선 어디가 막혀?",
]),
]
rows: List[Dict[str, Any]] = []
for feature, intent, qs in pairs:
for q in qs:
rows.append(
{
"mode": "ask",
"feature": feature,
"expected_intent": intent,
"question": q,
"params": {},
"note": "구어체 자연어 채팅",
}
)
return rows
FIELDNAMES = [
"tested_at",
"mode",
"feature",
"expected_intent",
"question",
"params",
"http_status",
"elapsed_sec",
"actual_intent",
"api_status",
"answer",
"result",
"reason",
"route_type",
"note",
]
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="http://127.0.0.1:8083/web")
ap.add_argument("--out", default="test.csv")
ap.add_argument("--timeout-domain", type=float, default=45.0)
ap.add_argument("--timeout-ask", type=float, default=90.0)
ap.add_argument("--sleep", type=float, default=0.15)
ap.add_argument("--sleep-ask", type=float, default=0.0, help="ask 전용 sleep(0이면 --sleep)")
ap.add_argument("--skip-ask", action="store_true")
ap.add_argument("--skip-domain", action="store_true")
ap.add_argument("--limit", type=int, default=0)
ap.add_argument(
"--profile",
choices=["full", "prod"],
default="full",
help="prod=기능당 소수 + 구어체(운영 부하 최소화)",
)
ap.add_argument("--insecure", action="store_true", help="SSL 검증 비활성화")
args = ap.parse_args()
if args.insecure:
_SSL_CTX = ssl._create_unverified_context()
base = args.base.rstrip("/")
domain_url = base + "/domain-query"
ask_url = base + "/ask"
traffic_url = base + "/traffic-news"
started = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
sleep_ask = args.sleep_ask if args.sleep_ask > 0 else max(args.sleep, 1.5)
# reset out
open(args.out, "w", encoding="utf-8-sig").close()
header_needed = True
results: List[str] = []
def emit(row: Dict[str, str]) -> None:
nonlocal header_needed
append_row(args.out, row, header_needed)
header_needed = False
results.append(row["result"])
print(
f"{row['result']:7} [{row['mode']:12}] {row['feature']:14} "
f"{row['expected_intent']:22} → {row['actual_intent'] or '-':22} | {row['question'][:42]}",
flush=True,
)
# traffic-news GET
http_st, body, elapsed = get_json(traffic_url, args.timeout_domain)
ans = extract_answer(body) if isinstance(body, (dict, list)) else one_line(str(body))
if isinstance(body, list):
ans = one_line(f"list={len(body)}")
ok = http_st == 200
else:
ok = http_st == 200 and (
(
isinstance(body, dict)
and (
body.get("status") is True
or "list" in body
or "items" in body
or "routes" in body
)
)
or isinstance(body, list)
)
emit(
{
"tested_at": started,
"mode": "domain-api",
"feature": "노선별교통속보",
"expected_intent": "TRAFFIC_NEWS",
"question": "오늘 고속도로에 사고나 통제 소식 있어?",
"params": "",
"http_status": str(http_st or ""),
"elapsed_sec": f"{elapsed:.2f}",
"actual_intent": "TRAFFIC_NEWS",
"api_status": "True" if ok else "False",
"answer": ans,
"result": "PASS" if ok else "FAIL",
"reason": "GET /traffic-news" if ok else "GET 실패",
"route_type": "GET",
"note": "간편조회 UI 전용",
}
)
suite: List[Dict[str, Any]] = []
if not args.skip_domain:
suite.extend(domain_cases_prod_light() if args.profile == "prod" else domain_cases())
if not args.skip_ask:
# prod: 기능당 질문 최대 2개로 축소
ask = ask_cases()
if args.profile == "prod":
by_feat: Dict[str, int] = {}
trimmed: List[Dict[str, Any]] = []
for c in ask:
n = by_feat.get(c["feature"], 0)
if n < 2:
trimmed.append(c)
by_feat[c["feature"]] = n + 1
ask = trimmed
suite.extend(ask)
if args.limit > 0:
suite = suite[: args.limit]
total = len(suite)
for i, case in enumerate(suite, 1):
mode = case["mode"]
print(f"({i}/{total}) …", flush=True)
if mode == "domain-query":
http_st, body, elapsed = post_json(
domain_url,
{
"intentType": case["expected_intent"],
"params": case["params"],
"botId": f"qi-dom-{i}",
},
args.timeout_domain,
)
actual = intent_of(body, case["expected_intent"])
# domain-query often echoes intent in request; keep expected if missing
if not actual:
actual = case["expected_intent"]
else:
http_st, body, elapsed = post_json(
ask_url,
{"query": case["question"], "botId": f"qi-ask-{i}"},
args.timeout_ask,
)
actual = intent_of(body)
status = status_of(body)
answer = extract_answer(body)
route_type = ""
if isinstance(body, dict):
route_type = str(body.get("routeType") or mode)
result, reason = judge(case["expected_intent"], actual, status, answer, http_st)
emit(
{
"tested_at": started,
"mode": mode,
"feature": case["feature"],
"expected_intent": case["expected_intent"],
"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,
"api_status": status,
"answer": answer,
"result": result,
"reason": reason,
"route_type": route_type,
"note": case.get("note") or "",
}
)
time.sleep(sleep_ask if mode == "ask" else args.sleep)
c = Counter(results)
print("---", flush=True)
print(f"saved={args.out} total={len(results)} {dict(c)}", flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+2 -1
View File
@@ -4,12 +4,13 @@
## GitHub upstream (submodule) ## GitHub upstream (submodule)
4개 서비스 소스는 GitHub `KoreaExpressWayCorp` 저장소를 **git submodule**로 연결합니다. 5개 소스는 GitHub `KoreaExpressWayCorp` 저장소를 **git submodule**로 연결합니다.
| monorepo 경로 | GitHub | 브랜치 | | monorepo 경로 | GitHub | 브랜치 |
|---------------|--------|--------| |---------------|--------|--------|
| `chatbotApp-chatbot2.0-agent` | https://github.com/KoreaExpressWayCorp/chatbotApp.git | `chatbot2.0-agent` | | `chatbotApp-chatbot2.0-agent` | https://github.com/KoreaExpressWayCorp/chatbotApp.git | `chatbot2.0-agent` |
| `chatbotApi-chatbot2.0-agent` | https://github.com/KoreaExpressWayCorp/chatbotApi.git | `chatbot2.0-agent` | | `chatbotApi-chatbot2.0-agent` | https://github.com/KoreaExpressWayCorp/chatbotApi.git | `chatbot2.0-agent` |
| `chatbotApp-test` | https://github.com/KoreaExpressWayCorp/chatbotApp-test.git | `chatbot2.0-agent` |
| `exAiChatBot-chatbot2.0-agent` | https://github.com/KoreaExpressWayCorp/exAiChatBot.git | `chatbot2.0-agent` | | `exAiChatBot-chatbot2.0-agent` | https://github.com/KoreaExpressWayCorp/exAiChatBot.git | `chatbot2.0-agent` |
| `kakaoChatbotSkill-chatbot2.0-agent` | https://github.com/KoreaExpressWayCorp/kakaoChatbotSkill.git | `chatbot2.0-agent` | | `kakaoChatbotSkill-chatbot2.0-agent` | https://github.com/KoreaExpressWayCorp/kakaoChatbotSkill.git | `chatbot2.0-agent` |
+6 -4
View File
@@ -23,16 +23,18 @@ for path in "${SUBMODULES[@]}"; do
local_src="${ROOT_DIR}/${path}" local_src="${ROOT_DIR}/${path}"
[[ -d "${local_src}" ]] || { echo "ERROR: ${local_src} 없음"; exit 1; } [[ -d "${local_src}" ]] || { echo "ERROR: ${local_src} 없음"; exit 1; }
log ">>> rsync ${path}" log ">>> rsync ${path}"
# build/dist/logs 는 자바 패키지명(예: kakao.response.build)과 충돌하므로
# 선행 슬래시로 프로젝트 최상위에만 적용한다.
rsync -az --delete \ rsync -az --delete \
--exclude 'node_modules/' \ --exclude 'node_modules/' \
--exclude 'build/' \
--exclude 'dist/' \
--exclude '.gradle/' \ --exclude '.gradle/' \
--exclude 'qdrant_data/' \ --exclude 'qdrant_data/' \
--exclude 'logs/' \
--exclude '.next/' \ --exclude '.next/' \
--exclude 'web/out/' \
--exclude '.git/' \ --exclude '.git/' \
--exclude '/build/' \
--exclude '/dist/' \
--exclude '/logs/' \
--exclude '/web/out/' \
"${local_src}/" "${EXDEV_SSH}:${REMOTE_WT}/${path}/" "${local_src}/" "${EXDEV_SSH}:${REMOTE_WT}/${path}/"
done done