b2b5708a6a
A/B 실험(상품 8개)에서 2개가 매칭 0~1건으로 사실상 실패했다. 수집은 40건 정상, 가격도 정확했는데 판정이 못 찾았다. 파고드니 두 가지였다. **① 일괄 판정 붕괴(핵심)**: 후보 37건을 한 번에 넣으면 gpt-4o-mini 가 전 항목에 같은 점수(70)를 매기고 **전부 불일치**로 답한다. temperature=0 에서 3회 재현. 10건씩 쪼개면 같은 모델·같은 입력으로 12건이 매칭된다. 상품 3종에서 모두 재현(크리넥스 0→12, 다우니 8→23). → _BATCH=10 으로 쪼개 병렬 판정 후 전체 index 로 복원. 지연은 배치 1개분, 토큰은 합산. **② 프롬프트 규칙 충돌**: "포장(개수/박스) 차이는 동일 상품" vs "용량·규격이 다르면 불일치" 가 함께 있어 '30롤 1팩'과 '30롤 2팩'(=60롤)을 어느 쪽으로 볼지 정할 수 없었다. 최저가 관점에선 수량이 다르면 다른 상품이다(60롤 값을 30롤 최저가로 쓰면 왜곡). 종류 혼동도 실측에서 오탐을 만들었다 — '프라이팬'을 찾는데 볶음팬·웍팬이 매칭됐다. → 규칙을 '무시할 차이 / 불일치로 볼 차이'로 갈라 재작성(수량·종류 명시). 결과(캐시 후보 재판정): 크리넥스 30롤 매칭 0 → 5건, 최저가 **not_found → 15,500원** 다우니 4.1L 매칭 8 → 12건, 최저 21,890 유지 테팔 프라이팬 매칭 2 → 0건 (볶음팬·웍팬 오탐 제거 — 후보에 진짜 프라이팬이 없었다) 테스트 6건 추가(배치 분할 시 index 매핑·누락 보수처리·토큰 합산·빈 후보·상수 상한·규칙 충돌), 전체 237 passed.
112 lines
5.8 KiB
Python
112 lines
5.8 KiB
Python
"""AI 유사도 판정 — 검색 결과가 '찾는 상품과 동일한 상품'인지 OpenAI 로 판별.
|
|
|
|
파이프라인(이상치 제거 뒤, top-N 앞) 슬롯에 결합한다. 규칙 고정 파싱이 취약한 문제를
|
|
LLM 판단으로 대체 — 액세서리/호환부품/다른 상품/명백히 다른 규격을 걸러 최저가 오염을 막는다.
|
|
structured output(Pydantic)으로 정규식 파싱 없이 안정적으로 결과를 받는다.
|
|
"""
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
|
|
from openai import AsyncOpenAI
|
|
from pydantic import BaseModel, Field
|
|
|
|
from common.logger import LOG
|
|
from config.server_configs import openai_config
|
|
from services.search.contract import NormalizedProduct
|
|
|
|
# 규칙은 '무시할 차이'와 '불일치로 볼 차이'로 갈라 쓴다. 예전엔 "포장(개수/박스) 차이는 동일"과
|
|
# "용량·규격이 다르면 불일치"가 함께 있어 '30롤 1팩' vs '30롤 2팩'(=60롤)을 어느 쪽으로 볼지
|
|
# 모델이 정할 수 없었다. 최저가 관점에선 수량이 다르면 다른 상품이다(60롤 값을 30롤 최저가로
|
|
# 쓰면 왜곡). 종류 혼동(프라이팬/볶음팬/웍팬)도 실측에서 오탐을 만들어 명시한다.
|
|
_SYSTEM = (
|
|
"너는 최저가 비교 시스템의 상품 매칭기다. 후보가 '찾는 상품과 동일한 상품'인지 판별하라.\n"
|
|
"동일하다고 보는 차이(무시할 것):\n"
|
|
"- 판매자·스토어, 색상/향, 사은품·증정품, 배송 문구, 상품명 수식어(무형광·프리미엄 등)\n"
|
|
"불일치로 보는 차이:\n"
|
|
"- 종류가 다름: 프라이팬 / 볶음팬 / 웍팬 / 냄비는 서로 다른 종류다. 세트 구성품도 단품과 다르다\n"
|
|
"- 브랜드·모델이 다름(모델명이 주어지면 모델 일치를 우선)\n"
|
|
"- 용량·크기가 다름(ml·L·g·cm)\n"
|
|
"- 수량이 다름: 개입·롤수·팩수. 예) '30롤 1팩'과 '30롤 2팩'(=60롤)은 다른 상품이다\n"
|
|
"- 액세서리·호환부품(뚜껑·손잡이·거치대·리필 전용 부속 등)\n"
|
|
"확신이 낮으면 score 를 낮게 준다."
|
|
)
|
|
|
|
# 한 번에 판정할 후보 수. **크면 판정이 통째로 무너진다** — 실측(2026-08-05): 후보 37건을 일괄로
|
|
# 넣으면 gpt-4o-mini 가 전 항목에 같은 점수(70)를 매기고 전부 불일치로 답한다(3회 재현).
|
|
# 10건씩 나누면 같은 입력·같은 모델로 12건이 매칭됐다(0건 → 12건). 상품 3종에서 모두 재현.
|
|
_BATCH = 10
|
|
|
|
|
|
@dataclass
|
|
class _Usage:
|
|
"""배치 여러 개의 토큰을 합쳐 계측부(metrics.add_ai)에 하나로 넘긴다."""
|
|
prompt_tokens: int = 0
|
|
completion_tokens: int = 0
|
|
|
|
|
|
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 | None = None, api_key: str | None = None):
|
|
self._model = model or openai_config.model
|
|
self._client = AsyncOpenAI(api_key=api_key or openai_config.api_key)
|
|
self.last_usage = None # 직전 호출 토큰 usage(계측용) — 호출부가 await 직후 읽는다
|
|
|
|
async def judge(self, target: dict, candidates: list[NormalizedProduct]) -> list[Judgment]:
|
|
"""후보별 동일상품 여부 판정. candidates 와 같은 순서/길이로 Judgment 리스트 반환.
|
|
|
|
후보를 _BATCH 단위로 쪼개 **병렬**로 묻고 합친다(지연은 배치 1개분). 쪼개는 이유는
|
|
속도가 아니라 정확도다 — 위 _BATCH 주석의 실측 참고.
|
|
"""
|
|
self.last_usage = None
|
|
if not candidates:
|
|
return []
|
|
|
|
chunks = [candidates[i:i + _BATCH] for i in range(0, len(candidates), _BATCH)]
|
|
results = await asyncio.gather(*(self._judge_batch(target, ch) for ch in chunks))
|
|
|
|
out: list[Judgment] = []
|
|
prompt = completion = 0
|
|
for offset, (judgments, usage) in zip(range(0, len(candidates), _BATCH), results):
|
|
for k in range(len(chunks[offset // _BATCH])):
|
|
j = judgments.get(k + 1)
|
|
# 배치 안 번호(1..n)를 전체 번호로 되돌린다. 누락은 보수적으로 불일치.
|
|
out.append(Judgment(index=offset + k + 1,
|
|
is_match=bool(j and j.is_match), score=j.score if j else 0))
|
|
prompt += getattr(usage, "prompt_tokens", 0) or 0
|
|
completion += getattr(usage, "completion_tokens", 0) or 0
|
|
self.last_usage = _Usage(prompt, completion)
|
|
LOG.d(f"[ai] 판정 {len(candidates)}건({len(chunks)}배치) 중 매칭 {sum(1 for j in out if j.is_match)}건")
|
|
return out
|
|
|
|
async def _judge_batch(self, target: dict, candidates: list[NormalizedProduct]):
|
|
"""배치 1개 판정 → ({배치내 index: Judgment}, usage)."""
|
|
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
|
|
return {j.index: j for j in (parsed.judgments if parsed else [])}, getattr(resp, "usage", None)
|