From a03f7257a6dd9c75f149c724acff35bbaba903bf Mon Sep 17 00:00:00 2001 From: Bae Han Jun Date: Thu, 17 Jul 2025 17:25:51 +0900 Subject: [PATCH] =?UTF-8?q?chore:=20=EA=B4=80=EC=84=B8=20=EC=98=88?= =?UTF-8?q?=EC=B8=A1=20=ED=94=84=EB=A1=9C=EC=84=B8=EC=8A=A4=20step?= =?UTF-8?q?=EC=9C=BC=EB=A1=9C=20=EB=B6=84=EA=B8=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/tariff_prediction/agent/step_api.py | 68 +++++++++++++++++++ core/tariff_prediction/dto/tariff_request.py | 18 ++--- core/tariff_prediction/dto/tariff_response.py | 13 ++-- .../tools/parse_hs_results.py | 2 +- 4 files changed, 85 insertions(+), 16 deletions(-) create mode 100644 core/tariff_prediction/agent/step_api.py diff --git a/core/tariff_prediction/agent/step_api.py b/core/tariff_prediction/agent/step_api.py new file mode 100644 index 0000000..847ebb0 --- /dev/null +++ b/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="잘못된 요청입니다. 상품 설명을 입력해 주세요." + ) \ No newline at end of file diff --git a/core/tariff_prediction/dto/tariff_request.py b/core/tariff_prediction/dto/tariff_request.py index 8136824..1517113 100644 --- a/core/tariff_prediction/dto/tariff_request.py +++ b/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] = "해외직구" # 해외직구, 해외체류 중 구매, 해외배송 \ No newline at end of file + 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 단계) \ No newline at end of file diff --git a/core/tariff_prediction/dto/tariff_response.py b/core/tariff_prediction/dto/tariff_response.py index eda0e7d..0db8ad1 100644 --- a/core/tariff_prediction/dto/tariff_response.py +++ b/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 \ No newline at end of file + 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 # 안내/에러 메시지 \ No newline at end of file diff --git a/core/tariff_prediction/tools/parse_hs_results.py b/core/tariff_prediction/tools/parse_hs_results.py index 85e4502..899e68d 100644 --- a/core/tariff_prediction/tools/parse_hs_results.py +++ b/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['한글품목명'] })