From b0a86b972014c0a30a3c2971bb74e43f23cb56ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Thu, 9 Jul 2026 08:44:57 +0900 Subject: [PATCH] =?UTF-8?q?feat(lps):=20AI=20=EC=9C=A0=EC=82=AC=EB=8F=84?= =?UTF-8?q?=20=ED=8C=90=EC=A0=95=20=E2=80=94=20'=EA=B0=99=EC=9D=80=20?= =?UTF-8?q?=EC=83=81=ED=92=88'=20=EB=A7=A4=EC=B9=AD=EC=9C=BC=EB=A1=9C=20?= =?UTF-8?q?=EC=95=A1=EC=84=B8=EC=84=9C=EB=A6=AC=20=EC=98=A4=EC=97=BC=20?= =?UTF-8?q?=ED=95=B4=EA=B2=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 파이프라인의 비워둔 슬롯(이상치 뒤·top-N 앞)에 OpenAI 유사도 판정을 결합. "스탠리 텀블러" 검색 시 빨대마개·커버 등 호환 액세서리가 최저가로 올라오던 문제를 해결한다(기계적 최저가 → 같은 상품 최저가). - ai/similarity: SimilarityJudge(OpenAI structured output). 액세서리/부품/다른규격 불일치 판별 - pipeline/core: apply_filters + rank_result 로 분리(AI 를 그 사이에 끼움), run_price_pipeline 동작 불변 - handler: judge 주입 시 ai_match STAGE 추가(필터 후 후보만 판정 → 토큰 절약), 미주입 시 생략 - worker_main: OPENAI_API_KEY 있으면 판정 ON - requirements: openai / tests: fake judge 필터링 검증 → 전체 32/32 - 라이브: '스탠리 퀜처 887ml' → ai_match(60→15) → 실제 텀블러 top-6(액세서리 제거) Co-Authored-By: Claude Opus 4.8 (1M context) --- lps/requirements.txt | 1 + lps/services/ai/similarity.py | 69 +++++++++++++++++++++ lps/services/pipeline/core.py | 101 ++++++++++++++++++------------- lps/tests/test_search_handler.py | 21 +++++++ lps/worker/handlers.py | 24 ++++++-- lps/worker_main.py | 6 +- 6 files changed, 175 insertions(+), 47 deletions(-) create mode 100644 lps/services/ai/similarity.py diff --git a/lps/requirements.txt b/lps/requirements.txt index 0b424ac..4bb610e 100644 --- a/lps/requirements.txt +++ b/lps/requirements.txt @@ -14,3 +14,4 @@ selectolax==0.4.10 # 빠른 C 파서(쿠팡 HTML). bs4 대비 최대 30배 patchright # 스텔스 Playwright 포크. 쿠팡 Akamai JS 챌린지 통과(실제 Chrome, channel=chrome) # ※ nodriver 는 Python 3.14 소스인코딩 버그로 미채택 → Patchright 로 대체 # ※ 실행엔 시스템 Google Chrome 필요(로컬) / 배포 이미지엔 chromium 설치 필요 +openai # AI 유사도 판정(같은 상품 매칭) — OPENAI_API_KEY(.env). structured output 사용 diff --git a/lps/services/ai/similarity.py b/lps/services/ai/similarity.py new file mode 100644 index 0000000..3867a03 --- /dev/null +++ b/lps/services/ai/similarity.py @@ -0,0 +1,69 @@ +"""AI 유사도 판정 — 검색 결과가 '찾는 상품과 동일한 상품'인지 OpenAI 로 판별. + +파이프라인(이상치 제거 뒤, top-N 앞) 슬롯에 결합한다. 규칙 고정 파싱이 취약한 문제를 +LLM 판단으로 대체 — 액세서리/호환부품/다른 상품/명백히 다른 규격을 걸러 최저가 오염을 막는다. +structured output(Pydantic)으로 정규식 파싱 없이 안정적으로 결과를 받는다. +""" + +import os + +from openai import AsyncOpenAI +from pydantic import BaseModel, Field + +from common.logger import LOG +from services.search.contract import NormalizedProduct + +_SYSTEM = ( + "너는 최저가 비교 시스템의 상품 매칭기다. 검색 결과 후보가 '찾는 상품과 동일한 상품'인지 판별하라.\n" + "규칙:\n" + "- 액세서리·호환부품·부속품(빨대마개·뚜껑·커버·거치대·스트랩·보호필름 등)은 불일치(false).\n" + "- 다른 종류/브랜드/모델은 불일치. 모델명이 주어지면 모델 일치를 우선한다.\n" + "- 용량·규격이 명백히 다르면 불일치.\n" + "- 판매자·색상·포장(개수/박스)·사은품 차이는 동일 상품으로 본다.\n" + "- 확신이 낮으면 score 를 낮게 준다." +) + + +class Judgment(BaseModel): + index: int = Field(description="후보 번호(1부터)") + is_match: bool = Field(description="찾는 상품과 동일 상품이면 true") + score: int = Field(description="동일 확신도 0~100") + + +class JudgmentList(BaseModel): + judgments: list[Judgment] + + +class SimilarityJudge: + def __init__(self, model: str = "gpt-4o-mini", api_key: str | None = None): + self._model = model + self._client = AsyncOpenAI(api_key=api_key or os.environ.get("OPENAI_API_KEY")) + + async def judge(self, target: dict, candidates: list[NormalizedProduct]) -> list[Judgment]: + """후보별 동일상품 여부 판정. candidates 와 같은 순서/길이로 Judgment 리스트 반환.""" + if not candidates: + return [] + + lines = "\n".join(f"{i + 1}. {c.name} ({c.price}원)" for i, c in enumerate(candidates)) + user = ( + f"[찾는 상품]\n" + f"상품명: {target.get('product_name', '')}\n" + f"모델: {target.get('model', '')}\n" + f"규격: {target.get('specification', '')}\n" + f"제조사/브랜드: {target.get('company', '')}\n\n" + f"[검색 결과 후보]\n{lines}\n\n" + f"각 후보가 찾는 상품과 동일 상품인지 index 별로 판별하라." + ) + + resp = await self._client.beta.chat.completions.parse( + model=self._model, + messages=[{"role": "system", "content": _SYSTEM}, {"role": "user", "content": user}], + response_format=JudgmentList, + temperature=0, + ) + parsed = resp.choices[0].message.parsed + by_idx = {j.index: j for j in (parsed.judgments if parsed else [])} + # 누락된 후보는 보수적으로 불일치 처리 + out = [by_idx.get(i + 1, Judgment(index=i + 1, is_match=False, score=0)) for i in range(len(candidates))] + LOG.d(f"[ai] 판정 {len(candidates)}건 중 매칭 {sum(1 for j in out if j.is_match)}건") + return out diff --git a/lps/services/pipeline/core.py b/lps/services/pipeline/core.py index 169aa13..f85e30f 100644 --- a/lps/services/pipeline/core.py +++ b/lps/services/pipeline/core.py @@ -12,6 +12,59 @@ from services.pipeline.filters import keep_only_mall, filter_out_malls, filter_b from services.pipeline.outliers import remove_price_outliers +def apply_filters( + products: list[NormalizedProduct], + *, + base_price: int | None = None, + keep_mall: str | None = None, + banned_malls=(), + band_tolerance: float = 0.7, + remove_outliers: bool = True, +) -> tuple[list[NormalizedProduct], list[dict]]: + """mall → 가격밴드 → 이상치(IQR) 순으로 후보를 좁힌다. (후보, STAGE로그) 반환. + AI 유사도 판정은 이 뒤(top-N 앞)에 결합한다.""" + stages: list[dict] = [] + cur = list(products) + + def stage(name: str, before: list, after: list): + stages.append({"stage": name, "in": len(before), "out": len(after)}) + + if keep_mall: + before = cur + cur = keep_only_mall(cur, keep_mall) + stage("keep_mall", before, cur) + elif banned_malls: + before = cur + cur = filter_out_malls(cur, banned_malls) + stage("filter_out_malls", before, cur) + + if base_price: + before = cur + cur = filter_by_price_band(cur, base_price, band_tolerance) + stage("price_band", before, cur) + + if remove_outliers: + before = cur + cur, _ = remove_price_outliers(cur) + stage("outlier", before, cur) + + return cur, stages + + +def rank_result(products: list[NormalizedProduct], total_found: int, stages: list[dict], top_n: int = 5) -> dict: + """최저가순 정렬 + top-N + 결과 봉투 조립(top_n STAGE 포함).""" + ranked = sorted(products, key=lambda p: p.price) + top = ranked[:top_n] + stages = stages + [{"stage": "top_n", "in": len(ranked), "out": len(top)}] + return { + "total_found": total_found, + "kept": len(ranked), + "lowest": top[0].model_dump() if top else None, + "top": [p.model_dump() for p in top], + "stages": stages, + } + + def run_price_pipeline( products: list[NormalizedProduct], *, @@ -22,45 +75,9 @@ def run_price_pipeline( remove_outliers: bool = True, top_n: int = 5, ) -> dict: - stages: list[dict] = [] - cur = list(products) - - def stage(name: str, before: list, after: list): - stages.append({"stage": name, "in": len(before), "out": len(after)}) - - # 1) mall 필터 (쿠팡 단독이면 사실상 no-op, 네이버 결합 시 오픈마켓 정리용) - if keep_mall: - before = cur - cur = keep_only_mall(cur, keep_mall) - stage("keep_mall", before, cur) - elif banned_malls: - before = cur - cur = filter_out_malls(cur, banned_malls) - stage("filter_out_malls", before, cur) - - # 2) 가격 밴드 (요청 현재가 기준 targeted 컷) - if base_price: - before = cur - cur = filter_by_price_band(cur, base_price, band_tolerance) - stage("price_band", before, cur) - - # 3) 이상치 제거 (IQR) - if remove_outliers: - before = cur - cur, _ = remove_price_outliers(cur) - stage("outlier", before, cur) - - # (AI 유사도 판정 슬롯 — 키 준비 시 여기) - - # 4) 최저가순 정렬 + top-N - ranked = sorted(cur, key=lambda p: p.price) - top = ranked[:top_n] - stage("top_n", ranked, top) - - return { - "total_found": len(products), - "kept": len(ranked), - "lowest": top[0].model_dump() if top else None, - "top": [p.model_dump() for p in top], - "stages": stages, - } + """AI 없는 동기 파이프라인(테스트/기본 경로). apply_filters + rank_result 조합.""" + cur, stages = apply_filters( + products, base_price=base_price, keep_mall=keep_mall, + banned_malls=banned_malls, band_tolerance=band_tolerance, remove_outliers=remove_outliers, + ) + return rank_result(cur, len(products), stages, top_n) diff --git a/lps/tests/test_search_handler.py b/lps/tests/test_search_handler.py index a538550..26a85ee 100644 --- a/lps/tests/test_search_handler.py +++ b/lps/tests/test_search_handler.py @@ -56,3 +56,24 @@ async def test_all_sources_fail_raises(): } with pytest.raises(RuntimeError): await build_search_handler(adapters)(_job()) + + +class FakeJudge: + """가격 조건으로 매칭을 흉내내는 판정기(실제 OpenAI 호출 없음).""" + + def __init__(self, predicate): + self._pred = predicate + + async def judge(self, target, candidates): + from services.ai.similarity import Judgment + return [Judgment(index=i + 1, is_match=self._pred(c), score=100 if self._pred(c) else 0) + for i, c in enumerate(candidates)] + + +async def test_ai_judge_filters_non_matches(): + adapters = {"naver": FakeAdapter("naver", [_np("naver", 1000), _np("naver", 2000), _np("naver", 3000)])} + judge = FakeJudge(lambda c: c.price == 2000) # 2000 만 '같은 상품' + r = await build_search_handler(adapters, judge=judge)(_job()) + assert [p["price"] for p in r["top"]] == [2000] # 비매칭 제거됨 + ai_stage = next(s for s in r["stages"] if s["stage"] == "ai_match") + assert ai_stage["in"] == 3 and ai_stage["out"] == 1 diff --git a/lps/worker/handlers.py b/lps/worker/handlers.py index 94bdda0..4936ed9 100644 --- a/lps/worker/handlers.py +++ b/lps/worker/handlers.py @@ -11,11 +11,18 @@ from common.enums import JobType from common.logger import LOG from services.search.contract import SearchAdapter from services.search.util import parse_price -from services.pipeline.core import run_price_pipeline +from services.pipeline.core import apply_filters, rank_result -def build_search_handler(adapters: dict[str, SearchAdapter], sources: list[str] | None = None, limit: int = 40, top_n: int = 5): - """검색 핸들러 생성. adapters = {source: SearchAdapter}. sources 미지정 시 전체 사용.""" +def build_search_handler( + adapters: dict[str, SearchAdapter], + sources: list[str] | None = None, + limit: int = 40, + top_n: int = 5, + judge=None, +): + """검색 핸들러 생성. adapters = {source: SearchAdapter}. sources 미지정 시 전체 사용. + judge(SimilarityJudge) 주입 시 필터 뒤·top-N 앞에 '같은 상품' AI 판정을 끼운다(없으면 생략).""" use = list(sources) if sources else list(adapters.keys()) async def handler(job: dict) -> dict: @@ -47,7 +54,16 @@ def build_search_handler(adapters: dict[str, SearchAdapter], sources: list[str] if not products and all("error" in v for v in per_source.values()): raise RuntimeError(f"모든 소스 검색 실패: {per_source}") # 잡 실패 → 재시도 - result = run_price_pipeline(products, base_price=base_price, top_n=top_n) + # 필터(mall·밴드·이상치) → [AI 유사도 판정] → top-N 최저가 + candidates, stages = apply_filters(products, base_price=base_price) + if judge is not None and candidates: + target = {k: payload.get(k, "") for k in ("product_name", "model", "specification", "company")} + verdicts = await judge.judge(target, candidates) + matched = [c for c, v in zip(candidates, verdicts) if v.is_match] + stages.append({"stage": "ai_match", "in": len(candidates), "out": len(matched)}) + candidates = matched + + result = rank_result(candidates, len(products), stages, top_n) result["query"] = query result["sources"] = per_source # 소스별 건수/에러 (관측) return result diff --git a/lps/worker_main.py b/lps/worker_main.py index dc6cf6d..793b700 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -13,6 +13,7 @@ from config.server_configs import web_server_config from crud.job_crud import JobQueue from services.search.coupang.adapter import CoupangAdapter from services.search.naver.adapter import NaverAdapter +from services.ai.similarity import SimilarityJudge from worker.handlers import build_search_handler from worker.notify import JobListener from worker.runner import Worker, run_reaper @@ -24,7 +25,10 @@ async def main(concurrency: int = 1): queue = JobQueue() # 쿠팡(브라우저, 무거움) + 네이버(오픈API, 가벼움) 동시 검색 → 병합 최저가 adapters = {"coupang": CoupangAdapter(headless=False), "naver": NaverAdapter()} - handler = build_search_handler(adapters) + # OPENAI_API_KEY 있으면 '같은 상품' AI 판정 활성화(없으면 기계적 최저가만) + judge = SimilarityJudge() if os.environ.get("OPENAI_API_KEY") else None + LOG.i(f"AI 유사도 판정: {'ON' if judge else 'OFF(키 없음)'}") + handler = build_search_handler(adapters, judge=judge) stop = asyncio.Event() listeners: list[JobListener] = []