Merge pull request #21 from hanjuhn/main

refactor: 관세 예측 기능 코드 리펙토링

Author: dongyeon1031

core/shared/router/intent_router.py

@@ -6,8 +6,6 @@
 from core.shared.constants import (
     TARIFF_SESSION_KEYWORDS,
     NUMBER_SELECTION_PATTERNS,
-    TARIFF_PREDICTION_KEYWORDS,
-    CUSTOMS_TRACKING_KEYWORDS,
     INTENT_CLASSIFICATION_PROMPT,
     INTENT_TYPES,
     DEFAULT_INTENT
@@ -16,11 +14,25 @@
 def intent_router(state: CustomsAgentState) -> CustomsAgentState:
     """사용자 쿼리의 의도를 분류합니다."""
 
+    # 현재 쿼리가 숫자 선택인지 확인
+    current_query = state["query"].strip()
+    is_number_selection = False
+    
+    # 숫자 선택 패턴 확인 (1번, 2번, 3번, 1, 2, 3 등)
+    for pattern in NUMBER_SELECTION_PATTERNS:
+        if re.match(pattern, current_query):
+            is_number_selection = True
+            break
+    
     # 이전 대화에서 관세 예측 중인지 확인
     messages = state.get("messages", [])
     is_in_tariff_session = False
 
-    # 최근 메시지들을 확인하여 관세 예측 세션 중인지 판단
+    # 1. state의 intent 필드 확인 (가장 중요)
+    if state.get("intent") == "tariff_prediction":
+        is_in_tariff_session = True
+    
+    # 2. 최근 메시지들을 확인하여 관세 예측 세션 중인지 판단
     for msg in messages[-5:]:  # 최근 5개 메시지 확인
         if hasattr(msg, 'content') and isinstance(msg.content, str):
             content = msg.content.lower()
@@ -29,7 +41,7 @@ def intent_router(state: CustomsAgentState) -> CustomsAgentState:
                 is_in_tariff_session = True
                 break
 
-    # 이전 의도가 tariff_prediction이었는지도 확인
+    # 3. 이전 의도가 tariff_prediction이었는지도 확인
     if messages and len(messages) > 0:
         last_msg = messages[-1]
         if hasattr(last_msg, 'content') and isinstance(last_msg.content, str):
@@ -43,23 +55,6 @@ def intent_router(state: CustomsAgentState) -> CustomsAgentState:
         print(state)
         return state
 
-    # 현재 쿼리가 숫자 선택인지 확인
-    current_query = state["query"].strip()
-    is_number_selection = False
-    
-    # 숫자 선택 패턴 확인 (1번, 2번, 3번, 1, 2, 3 등)
-    for pattern in NUMBER_SELECTION_PATTERNS:
-        if re.match(pattern, current_query):
-            is_number_selection = True
-            break
-    
-    # 관세 예측 세션 중이고 숫자 선택이면 무조건 tariff_prediction
-    if is_in_tariff_session and is_number_selection:
-        state["intent"] = "tariff_prediction"  # type: ignore
-        state["messages"].append(AIMessage(content=f"의도 분류 완료: {state['intent']} (세션 연속성 유지)"))
-        print(state)
-        return state
-    
     # 숫자 선택이지만 관세 예측 세션이 아닌 경우에도 tariff_prediction으로 분류
     # (HS 코드 직접 입력 등의 경우)
     if is_number_selection:
@@ -68,22 +63,7 @@ def intent_router(state: CustomsAgentState) -> CustomsAgentState:
         print(state)
         return state
 
-    # 관세 예측 관련 키워드가 있으면 무조건 tariff_prediction으로 분류
-    current_query_lower = current_query.lower()
-    if any(keyword in current_query_lower for keyword in TARIFF_PREDICTION_KEYWORDS):
-        state["intent"] = "tariff_prediction"  # type: ignore
-        state["messages"].append(AIMessage(content=f"의도 분류 완료: {state['intent']} (관세 예측 키워드 감지)"))
-        print(state)
-        return state
-    
-    # 운송장/배송 관련 키워드가 있으면 customs_tracking으로 분류 (배송은 제외)
-    if any(keyword in current_query_lower for keyword in CUSTOMS_TRACKING_KEYWORDS):
-        state["intent"] = "customs_tracking"  # type: ignore
-        state["messages"].append(AIMessage(content=f"의도 분류 완료: {state['intent']} (배송 추적 키워드 감지)"))
-        print(state)
-        return state
-    
-    # 일반적인 의도 분류 (LLM 사용)
+    # LLM 기반 의도 분류를 수행
     llm = get_llm()
 
     try:
@@ -99,6 +79,6 @@ def intent_router(state: CustomsAgentState) -> CustomsAgentState:
         intent = DEFAULT_INTENT  # 오류 시에도 tariff_prediction으로 분류
 
     state["intent"] = intent  # type: ignore
-    state["messages"].append(AIMessage(content=f"의도 분류 완료: {intent}"))
+    state["messages"].append(AIMessage(content=f"의도 분류 완료: {intent} (LLM 분류)"))
     print(state)
     return state

core/tariff_prediction/agent/step_api.py

@@ -5,41 +5,34 @@
 from core.tariff_prediction.tools.calculate_tariff_amount import calculate_tariff_amount
 from core.tariff_prediction.tools.parse_tariff_result import parse_tariff_result
 from core.shared.utils.llm import get_llm
+from core.tariff_prediction.constants import LLM_PROMPT_TEMPLATES, STEP_API
 
 def tariff_prediction_step_api(req: TariffPredictionRequest) -> TariffPredictionResponse:
     step = req.step
-    # Step 자동 분류: step이 'auto'이거나 비어 있으면 LLM으로 분류
-    if not step or step == 'auto':
+    if not step or step == STEP_API['AUTO_STEP']:
         llm = get_llm()
-        step_prompt = f"""
-        다음 사용자 입력이 관세 예측 플로우의 어떤 단계에 해당하는지 분류하세요.
-        - 상품 설명 입력: input
-        - HS6 코드 선택: hs6_select
-        - HS10 코드 선택 및 관세 계산: hs10_select
-        반드시 input, hs6_select, hs10_select 중 하나로만 답하세요.
-        사용자 입력: {req.product_description or req.hs6_code or req.hs10_code or ''}
-        """
+        user_input = req.product_description or req.hs6_code or req.hs10_code or ''
+        step_prompt = LLM_PROMPT_TEMPLATES['step_classification'].format(user_input=user_input)
         step_result = llm.invoke([{"role": "system", "content": step_prompt}])
         step = str(getattr(step_result, 'content', step_result)).strip()
-    # Step별 분기
-    if step == "input":
+    if step == STEP_API['INPUT_STEP']:
         # 상품 설명 → HS6 후보 예측
         hs6_result = get_hs_classification(req.product_description)
         hs6_candidates = parse_hs6_result(hs6_result)
         return TariffPredictionResponse(
-            step="hs6_select",
+            step=STEP_API['HS6_SELECT_STEP'],
             hs6_candidates=hs6_candidates,
-            message="상품에 해당하는 HS6 코드를 선택해 주세요."
+            message=STEP_API['HS6_SELECTION_MESSAGE']
         )
-    elif step == "hs6_select":
+    elif step == STEP_API['HS6_SELECT_STEP']:
         # HS6 코드 → HS10 후보 추출
         hs10_candidates = generate_hs10_candidates(req.hs6_code)
         return TariffPredictionResponse(
-            step="hs10_select",
+            step=STEP_API['HS10_SELECT_STEP'],
             hs10_candidates=hs10_candidates,
-            message="HS10 코드 후보를 선택해 주세요."
+            message=STEP_API['HS10_SELECTION_MESSAGE']
         )
-    elif step == "hs10_select":
+    elif step == STEP_API['HS10_SELECT_STEP']:
         # HS10 코드, 국가, 가격 등 입력받아 관세 계산
         result = calculate_tariff_amount.invoke({
             "product_code": req.hs10_code,
@@ -49,30 +42,27 @@ def tariff_prediction_step_api(req: TariffPredictionRequest) -> TariffPrediction
             "shipping_cost": req.shipping_cost,
             "situation": req.scenario
         })
-        
+    
         # 결과를 문자열로 변환
         result_str = str(result)
 
-        # 에러 메시지인지 확인 (에러 메시지는 보통 짧고 특정 키워드를 포함)
-        if result_str.startswith("오류") or result_str.startswith("Error") or "실패" in result_str or "오류" in result_str:
-            # 에러 메시지
+        if any(keyword in result_str for keyword in STEP_API['ERROR_KEYWORDS']):
             return TariffPredictionResponse(
-                step="result",
+                step=STEP_API['RESULT_STEP'],
                 calculation_result=None,
                 message=result_str
             )
         else:
-            # 성공적인 결과 - 예쁘게 포맷팅
             parsed_result = parse_tariff_result(result_str)
             formatted_result = parsed_result['formatted_result']
 
             return TariffPredictionResponse(
-                step="result",
-                calculation_result=parsed_result,  # 딕셔너리 형태로 전달
-                message=formatted_result  # 포맷팅된 결과를 message에 전달
+                step=STEP_API['RESULT_STEP'],
+                calculation_result=parsed_result, 
+                message=formatted_result
             )
     else:
         return TariffPredictionResponse(
-            step="hs6_select",
-            message="잘못된 요청입니다. 상품 설명을 입력해 주세요."
+            step=STEP_API['HS6_SELECT_STEP'],
+            message=STEP_API['DEFAULT_ERROR_MESSAGE']
         )