Merge pull request #11 from hanjuhn/main

chore: 관세 예측 프로세스 step으로 분기

Author: dongyeon1031

core/tariff_prediction/agent/step_api.py

@@ -0,0 +1,68 @@
+from core.tariff_prediction.dto.tariff_request import TariffPredictionRequest
+from core.tariff_prediction.dto.tariff_response import TariffPredictionResponse
+from core.tariff_prediction.tools.get_hs_classification import get_hs_classification
+from core.tariff_prediction.tools.parse_hs_results import parse_hs6_result, generate_hs10_candidates
+from core.tariff_prediction.tools.calculate_tariff_amount import calculate_tariff_amount
+from core.shared.utils.llm import get_llm
+
+def tariff_prediction_step_api(req: TariffPredictionRequest) -> TariffPredictionResponse:
+    step = req.stepㄴ
+    # Step 자동 분류: step이 'auto'이거나 비어 있으면 LLM으로 분류
+    if not step or step == 'auto':
+        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 ''}
+        """
+        step_result = llm.invoke([{"role": "system", "content": step_prompt}])
+        step = str(getattr(step_result, 'content', step_result)).strip()
+    # Step별 분기
+    if step == "input":
+        # 상품 설명 → HS6 후보 예측
+        hs6_result = get_hs_classification(req.product_description)
+        hs6_candidates = parse_hs6_result(hs6_result)
+        return TariffPredictionResponse(
+            step="hs6_select",
+            hs6_candidates=hs6_candidates,
+            message="상품에 해당하는 HS6 코드를 선택해 주세요."
+        )
+    elif step == "hs6_select":
+        # HS6 코드 → HS10 후보 추출
+        hs10_candidates = generate_hs10_candidates(req.hs6_code)
+        return TariffPredictionResponse(
+            step="hs10_select",
+            hs10_candidates=hs10_candidates,
+            message="해당하는 HS10 코드를 선택해 주세요."
+        )
+    elif step == "hs10_select":
+        # HS10 코드, 국가, 가격 등 입력받아 관세 계산
+        result = calculate_tariff_amount.invoke({
+            "product_code": req.hs10_code,
+            "value": req.price,
+            "origin_country": req.origin_country,
+            "item_count": req.quantity,
+            "shipping_cost": req.shipping_cost,
+            "situation": req.scenario
+        })
+        if isinstance(result, str):
+            # 에러 메시지
+            return TariffPredictionResponse(
+                step="result",
+                calculation_result=None,
+                message=result
+            )
+        else:
+            return TariffPredictionResponse(
+                step="result",
+                calculation_result=result,
+                message="관세 계산 결과입니다."
+            )
+    else:
+        return TariffPredictionResponse(
+            step="input",
+            message="잘못된 요청입니다. 상품 설명을 입력해 주세요."
+        ) 

core/tariff_prediction/dto/tariff_request.py

@@ -1,11 +1,13 @@
 from pydantic import BaseModel
-from typing import Optional
+from typing import Optional, Literal
 
 class TariffPredictionRequest(BaseModel):
-    """관세 예측 요청 DTO"""
-    product_description: str
-    price: Optional[float] = None
-    quantity: Optional[int] = 1
-    origin_country: Optional[str] = None
-    shipping_cost: Optional[float] = 0
-    scenario: Optional[str] = "해외직구"  # 해외직구, 해외체류 중 구매, 해외배송 
+    step: Literal["input", "hs6_select", "hs10_select"]  # 현재 단계
+    product_description: Optional[str] = None  # 상품 설명 (input 단계)
+    hs6_code: Optional[str] = None             # 선택한 HS6 코드 (hs6_select 단계)
+    hs10_code: Optional[str] = None            # 선택한 HS10 코드 (hs10_select 단계)
+    origin_country: Optional[str] = None       # 원산지 국가 (hs10_select 단계)
+    price: Optional[float] = None              # 상품 가격 (hs10_select 단계)
+    quantity: Optional[int] = 1                # 수량 (hs10_select 단계)
+    shipping_cost: Optional[float] = 0         # 배송비 (hs10_select 단계)
+    scenario: Optional[str] = "해외직구"         # 시나리오 (hs10_select 단계) 

core/tariff_prediction/dto/tariff_response.py

@@ -1,5 +1,5 @@
 from pydantic import BaseModel
-from typing import List, Optional
+from typing import List, Optional, Dict, Any, Literal
 
 class HSClassificationResult(BaseModel):
     """HS 코드 분류 결과"""
@@ -25,9 +25,8 @@ class ExchangeRateResult(BaseModel):
     date: str
 
 class TariffPredictionResponse(BaseModel):
-    """관세 예측 응답 DTO"""
-    product_description: str
-    hs_classifications: List[HSClassificationResult]
-    tariff_calculation: Optional[TariffCalculationResult] = None
-    exchange_rate: Optional[ExchangeRateResult] = None
-    total_response: str 
+    step: Literal["hs6_select", "hs10_select", "result"]  # 다음 단계
+    hs6_candidates: Optional[List[Dict[str, Any]]] = None     # HS6 후보 리스트 (input 단계 응답)
+    hs10_candidates: Optional[List[Dict[str, Any]]] = None    # HS10 후보 리스트 (hs6_select 단계 응답)
+    calculation_result: Optional[Dict[str, Any]] = None       # 관세 계산 결과 (hs10_select 단계 응답)
+    message: Optional[str] = None                             # 안내/에러 메시지 

core/tariff_prediction/tools/parse_hs_results.py

@@ -58,7 +58,7 @@ def generate_hs10_candidates(hs6_code: str) -> List[Dict]:
         candidates = []
         for _, row in matching_rows.iterrows():
             candidates.append({
-                'code': row['HS10'],
+                'code': str(row['HS10']),  # 항상 문자열로 변환
                 'description': row['한글품목명']
             })