[REFACTOR] LLM 성능개선

llm-server/server.py

@@ -7,6 +7,7 @@
 from datetime import datetime
 import os
 import re
+import json
 
 load_dotenv()
 client = OpenAI(api_key=os.getenv("API_KEY"))
@@ -42,110 +43,227 @@ class SavingRecommendRequest(BaseModel):
     userRole: str
     mainBankName: str = None
 
+# classify 키워드
 category_keyword_map = {
     1: ["우아한형제들", "요기요", "배달의민족", "배민", "쿠팡이츠"],
-    2: ["스타벅스", "투썸", "컴포즈", "매머드", "커피", "카페", "베이커리", "메가커피", "이디야", "폴바셋"],
-    3: ["신세계백화점", "현대백화점", "롯데백화점", "올리브영", "무신사", "쿠팡", "11번가", "G마켓", "SSG.COM", "롯데온", "네이버스마트스토어"],
-    4: ["카카오T", "택시", "우버", "마카롱", "T 블루"],
-    5: ["CU", "씨유", "GS", "지에스", "세븐일레븐", "이마트24", "편의점", "미니스톱"],
-    6: ["CGV", "메가박스", "롯데시네마", "예스24", "인터파크", "도서", "공연", "문화", "티켓", "교보문고"],
-    7: ["맥주", "소주", "술집", "포차", "홈술", "와인", "역전할머니", "더부스"],
-    8: ["버스", "지하철", "교통카드", "티머니", "T머니", "코레일", "KTX", "공항철도", "카카오모빌리티"],
-    9: ["약국", "병원", "의원", "한의원", "치과", "안과", "피부과"],
-    10: ["이마트", "홈플러스", "마트", "다이소", "코스트코", "마켓컬리", "쿠팡 로켓프레시"],
-    11: ["김밥", "분식", "식당", "라멘", "파스타", "카츠", "맥도날드", "우동", "써브웨이", "칼국수", "버거", "스시", 
-         "롯데리아", "버거킹", "신전떡볶이", "백향목분식", "본도시락", "BHC", "교촌치킨", "미분당", "홍콩반점"]
+    2: ["스타벅스", "투썸", "컴포즈", "매머드", "커피", "카페", "베이커리","파리바게뜨","공차",
+        "이디야", "빽다방","쥬씨","배스킨라빈스","요거트","던킨","노티드","설빙","젤라또"],
+    4: ["카카오T", "택시", "우버","T블루"],
+    5: ["CU", "씨유", "GS25", "지에스25", "세븐일레븐", "이마트24", "미니스톱"],
+    6: ["CGV", "메가박스", "시네마", "예스24", "인터파크", "도서", "공연", "문화", "티켓", "문고", "알라딘","책방","씨어터","미술관"],
+    7: ["맥주", "소주", "술집", "포차", "와인", "호프","펍","주막"],
+    8: ["버스", "지하철", "교통카드", "티머니", "T머니", "코레일", "KTX", "공항철도"],
+    9: ["약국", "병원", "의원", "치과", "내과","외과", "이비인후과","안과", "피부과","정형외과"],
+    10: ["이마트", "홈플러스","마트", "다이소", "코스트코", "마켓컬리", "로켓프레시","안경","전기","수도","가스"],
+    11: ["김밥", "분식", "식당", "라멘", "파스타", "카츠", "맥도날드","KFC" ,"우동", "텐동","써브웨이","샤브","해화로",
+         "칼국수", "버거", "스시", "포케","롯데리아","맘스터치" ,"떡볶이", "도시락", "순대","육회","세종원","카레",
+         "치킨","통닭", "피자","해장","국수","국밥","토스트","감자탕", "갈비","브런치" ,"고기","한우","삼겹","프레퍼스",
+         "다이닝","타코","비스트로","레스토랑","반점","키친","그릴","정성","할머니","숯불","리필","원조","전통","샹츠마라"],
+    3: ["백화점", "아울렛","면세점","올리브영", "무신사", "에이블리","지그재그" ,"쿠팡", "11번가", "G마켓", 
+        "SSG", "롯데온", "네이버","나이스페이먼츠","ALIPAY","ARS","KCP"],
 }
 
+# 키워드 필터링 함수
 def keyword_filtering(desc: str) -> int:
     for id, keywords in category_keyword_map.items():
         for k in keywords:
             if k in desc:
                 return id 
     return -1
 
-def classify_category(desc):
-    filtered = keyword_filtering(desc)
-    if filtered != -1:
-        return filtered
-
-    prompt = f'''
-    소비내역: {desc}
-    소비내역을 아래 카테고리 번호 중 하나로 분류하세요. 설명없이 카테고리 번호만 출력하세요.
-    (카페/간식:2, 쇼핑:3, 택시:4, 편의점:5, 문화(영화관, 티켓, 공연, 스포츠):6,
-    술/유흥:7, 대중교통:8, 의료(병원/약국):9, 생활(마트/생활/주거):10, 식비:11, 기타:12)
-    우아한형제들과 요기요만 1로 분류하세요.:
-    '''
+# 한 번에 여러 건을 분류하는 LLM 함수
+def classify_batch_llm(items: List[Dict]) -> Dict[int, int]:
+
+    system_prompt = (
+        "너는 영수증의 상호명을 정확히 하나의 카테고리로 분류하는 분류기다. "
+        "반드시 제공된 ENUM(키) 중 하나만 선택한다. "
+        "입력은 [expenditure_id:int, description:str]의 JSON 배열이다. "
+        '반드시 JSON 객체 하나로만 응답하고, "results" 키 아래에 '
+        "[expenditure_id:int, category_id:int] 쌍의 배열만 담아라. "
+        '예: {"results": [[1, 11], [2, 2], ...]}'
+    )
+
+    # LLM에 보내는 inputs를 [id, desc]의 compact 배열로 구성
+    compact_inputs = [[it["expenditure_id"], it["description"]] for it in items]
+
+    user_payload = {
+        "inputs": compact_inputs,
+        "enum": {
+            "2": "카페/간식/디저트",
+            "3": "쇼핑",
+            "5": "편의점",
+            "6": "문화(영화,공연,도서,전시)",
+            "7": "술/유흥",
+            "8": "대중교통",
+            "9": "의료",
+            "10": "생활(마트,주거,통신,이발)",
+            "11": "식비(식당,음식점)",
+            "12": "기타"
+        },
+        "rules": [
+            "브랜드/장소가 명확하면 해당 카테고리 우선(예: 스타벅스=2).",
+            "애매하면 문자열 주요 키워드만 분석하여, 대한민국 기준으로 가장 가능성 높은 카테고리로 추정.",
+            "전혀 예측할 수 없는 경우에만 불가피하게 12로 지정.(최대한 피한다.)"
+        ],
+        "output_schema": {"results": [["int", "int"]]}
+    }
+
     response = client.chat.completions.create(
         model="gpt-4o",
-        messages=[{"role": "user", "content": prompt}],
-        temperature=0
+        temperature=0,
+        top_p=0,
+        response_format={"type": "json_object"},
+        messages=[
+            {"role": "system", "content": system_prompt},
+            {"role": "user", "content": json.dumps(user_payload, ensure_ascii=False)},
+        ]
     )
-    return response.choices[0].message.content.strip()
 
-async def classify_category_async(desc):
-    loop = asyncio.get_event_loop()
-    return await loop.run_in_executor(None, lambda: classify_category(desc))
+    raw = response.choices[0].message.content
+    try:
+        data = json.loads(raw)
+        pairs = data.get("results", [])
+    except Exception:
+        # 비정상 출력 시 [id, 12] 쌍 배열로 방어
+        pairs = [[it["expenditure_id"], 12] for it in items]
+
+    # CHANGED: [id, cat] → {id: cat} 매핑으로 변환 (반환 타입은 기존과 동일)
+    out: Dict[int, int] = {}
+    for p in pairs:
+        try:
+            eid, cid = int(p[0]), int(p[1])
+        except Exception:
+            # 방어적 캐스팅
+            try:
+                eid = int(p[0])
+            except Exception:
+                continue
+            cid = 12
+        out[eid] = cid
+
+    return out
+
+# 동시 배치 수 제한을 위한 세마포어
+_SEMAPHORE = asyncio.Semaphore(16)  # 병렬 갯수
+_BATCH_SIZE = 16                    # 배치 크기
+
+# 배치 호출을 비동기로 감싸기 (블로킹 SDK 호출을 스레드로 우회)
+async def _classify_batch_llm_async(items: List[Dict]) -> Dict[int, int]:
+    async with _SEMAPHORE:
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(None, lambda: classify_batch_llm(items))
 
 # 분류 API
 @app.post("/classify")
 async def classify(list: ExpsList):
-    tasks = [classify_category_async(e.description) for e in list.exps]
-    categories = await asyncio.gather(*tasks)
-    results = [
-        {"expenditure_id": e.expenditure_id, "category_id": int(c)}
-        for e, c in zip(list.exps, categories)
-    ]
-    return {"results": results}
+    # 1) 키워드 필터로 즉시 결정 가능한 것 선별
+    decided: Dict[int, int] = {}
+    undecided: List[Dict] = []
 
-# 분석 API
-@app.post("/analysis")
-async def analysis(list: ExpsAnalytics):
-    simplified = [
-        {"category_id": e.category_id, 
-        "amount": e.amount, 
-        "expenditure_date": e.expenditure_date.strftime("%Y-%m-%d")
-        } for e in list.exps]
-
-    messages = [
-        {
-            "role": "system",
-            "content": """당신은 개인 금융 관리 전문가입니다. 사용자의 소비 패턴을 분석하여 과소비가 심한 카테고리를 찾아내는 것이 당신의 역할입니다.
+    for e in list.exps:
+        k = keyword_filtering(e.description or "")
+        if k != -1:
+            decided[e.expenditure_id] = k
+        else:
+            undecided.append({"expenditure_id": e.expenditure_id, "description": e.description})
 
-분석 기준:
-1. 최근 30일 vs 이전 30일 지출 증가율
-2. 사치성 소비 카테고리 가중치 (배달음식, 카페, 쇼핑, 택시, 편의점, 문화, 술/유흥)
-3. 절대 지출 금액의 크기
-4. 새롭게 등장한 소비 카테고리
+    # 2) 필터링 후 남은 것들
+
+    # 같은 description은 한 번만 LLM에 보내게끔 설정
+    by_desc: Dict[str, List[int]] = {}
+    for it in undecided:
+        d = it["description"]
+        by_desc.setdefault(d, []).append(it["expenditure_id"])
 
-출력 형식: 카테고리 번호 3개를 쉼표로 구분 (예: 1,3,7)
-1~7 범위의 숫자만 사용하세요."""
-        },
-        {
-            "role": "user",
-            "content": f"""다음 소비 데이터를 분석하여 과소비가 심한 카테고리 상위 3개를 선정하세요:
+    # 대표 id 하나만 들고 가서 LLM 호출
+    uniq_items = [{"expenditure_id": ids[0], "description": desc}
+                  for desc, ids in by_desc.items()]
+
+    # 배치 LLM 호출(세마포어로 동시성 제한)
+    tasks = []
+    for i in range(0, len(uniq_items), _BATCH_SIZE):
+        batch = uniq_items[i:i + _BATCH_SIZE]
+        tasks.append(_classify_batch_llm_async(batch))
 
-소비내역: {simplified}
+    results_llm: Dict[int, int] = {}
+    if tasks:
+        for chunk in await asyncio.gather(*tasks):
+            results_llm.update(chunk)
 
-카테고리 설명:
-1: 배달음식, 2: 카페/간식, 3: 쇼핑, 4: 택시, 5: 편의점
-6: 문화, 7: 술/유흥
+    # 대표 id의 예측을 동일 description의 나머지 id에 전파
+    for desc, ids in by_desc.items():
+        rep_id = ids[0]
+        pred = int(results_llm.get(rep_id, 12))
+        for eid in ids: # 대표 포함 전체에 동일 적용
+            results_llm[eid] = pred
 
-분석 조건:
-- 지난 60~30일 대비 최근 30일 지출이 급증한 카테고리
-- 배달음식(1), 카페(2), 쇼핑(3), 택시(4), 편의점(5), 문화(6), 술/유흥(7) 사치성 소비 우선
-- 절대 금액이 큰 카테고리 우선
-- 새로운 소비 패턴도 고려
+    # 3) 결합 결과 생성
+    final = []
+    for e in list.exps:
+        cid = decided.get(e.expenditure_id) or results_llm.get(e.expenditure_id) or 12  # 방어적 기본값
+        final.append({"expenditure_id": e.expenditure_id, "category_id": int(cid)})
 
-상위 3개 카테고리 번호만 쉼표로 구분하여 출력:"""
-        }
+    return {"results": final}
+
+# 분석 API
+@app.post("/analysis")
+async def analysis(list: ExpsAnalytics):
+
+    # compact 형태로 변환 ([category_id, amount, date])
+    simplified = [
+        [e.category_id, e.amount, e.expenditure_date.strftime("%Y-%m-%d")]  # CHANGED
+        for e in list.exps
     ]
 
+    SYS = (
+        "당신은 개인 금융 분석가. "
+        "user가 준 rules만 근거로 판단하고, 정수 배열로만 출력하라. "
+        "입력 데이터는 [category_id:int, amount:int, date:str] 배열 형식이다."
+    )
+
+    USER = {
+        "task":"과소비 카테고리 상위 3개 선정",
+        "rules":[
+            "최근 60~30일의 지출 대비 최근 30일의 지출 증가율을 최우선 고려.",
+            "증가율이 같다면 사치성 소비를 우선. ",
+            "절대 지출 금액 우선. ",
+            "최근 30일에 새로 등장한 카테고리는 가산점.",
+        ],
+        "categories":{
+            1: "배달음식", 2: "카페/간식", 3: "쇼핑", 4: "택시",
+            5: "편의점", 6: "문화", 7: "술/유흥",
+        },
+        "data":simplified,
+        "output": "중복없는 3개의 정수 배열(예: [1,3,7])"
+    }
+
     response = client.chat.completions.create(
         model="gpt-4o", 
-        messages=messages,
+        messages=[
+            {"role": "system", "content": SYS},
+            {"role": "user", "content": json.dumps(USER, ensure_ascii=False)},
+        ],
+        response_format={
+            "type": "json_schema",
+            "json_schema": {
+                "name": "top3",
+                "strict": True,
+                "schema": {
+                    "type": "object",  # 루트는 object
+                    "properties": {
+                        "results": {
+                            "type": "array",
+                            "items": {"type": "integer", "minimum": 1, "maximum": 7},
+                            "minItems": 3,
+                            "maxItems": 3,
+                        }
+                    },
+                    "required": ["results"],
+                    "additionalProperties": False
+                }
+            }
+        },
         temperature=0,
-        max_tokens=20
     )
 
     res = response.choices[0].message.content.strip()