diff --git a/lps/common/database/model/models.py b/lps/common/database/model/models.py index 80f70ea..2be9127 100644 --- a/lps/common/database/model/models.py +++ b/lps/common/database/model/models.py @@ -50,3 +50,19 @@ class job(MAIN_BASE): postgresql_where=text("status IN (1, 2) AND dedupe_key IS NOT NULL"), ), ) + + +class search_negative(MAIN_BASE): + """네거티브 캐시 — '검색해도 없더라'를 TTL 동안 기억해 재검색 낭비를 막는다. + until 이 지나면 자동 무효(재도전 허용 — 나중에 입고될 수 있으므로).""" + + @staticmethod + def DBType(): + return DBType.MAIN.value + + __tablename__ = "search_negative" + + key = Column(String(300), primary_key=True) # 보통 product_code(없으면 query) + until = Column(DateTime(timezone=True), nullable=False) # 이 시각까지 not_found 로 간주 + reason = Column(String(200), nullable=True) # 종료 사유 메모(관측) + created_at = Column(DateTime(timezone=True), nullable=False, server_default=text("now()")) diff --git a/lps/crud/negative_cache.py b/lps/crud/negative_cache.py new file mode 100644 index 0000000..f657045 --- /dev/null +++ b/lps/crud/negative_cache.py @@ -0,0 +1,47 @@ +"""네거티브 캐시 CRUD — not_found 결론을 TTL 동안 기억. + +같은 상품 재요청이 즉시 재검색(브라우저+API+LLM 비용)하는 것을 막는다. +TTL 만료 후에는 다시 검색 허용(입고 가능성). key 는 보통 product_code.""" + +from sqlalchemy import text + +from common.database.db_session_manager import DB_SESSION_MNG +from common.enums import DBType, DBWRType + + +class NegativeCache: + DB = DBType.MAIN.value + + async def is_negative(self, key: str) -> bool: + """key 가 아직 유효한 not_found 로 캐시돼 있으면 True.""" + if not key: + return False + s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_READ.value) + try: + row = (await s.execute( + text("SELECT 1 FROM search_negative WHERE key = :k AND until > now()"), + {"k": key}, + )).first() + return row is not None + finally: + await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_READ.value) + + async def put(self, key: str, ttl_sec: int = 86400, reason: str = "not_found"): + """key 를 ttl_sec 동안 not_found 로 기록(upsert).""" + if not key: + return + sql = text(""" + INSERT INTO search_negative (key, until, reason) + VALUES (:k, now() + make_interval(secs => :ttl), :r) + ON CONFLICT (key) DO UPDATE + SET until = EXCLUDED.until, reason = EXCLUDED.reason, created_at = now() + """) + s = await DB_SESSION_MNG.start_session(self.DB, DBWRType.DB_WRITE.value) + try: + await s.execute(sql, {"k": key, "ttl": ttl_sec, "r": reason[:200]}) + await s.commit() + except Exception: + await s.rollback() + raise + finally: + await DB_SESSION_MNG.end_session(self.DB, DBWRType.DB_WRITE.value) diff --git a/lps/services/ai/keyword.py b/lps/services/ai/keyword.py new file mode 100644 index 0000000..b1e8e35 --- /dev/null +++ b/lps/services/ai/keyword.py @@ -0,0 +1,50 @@ +"""LLM 검색어 생성 — 재정제 라운드용 정밀/광역 쿼리 생성. + +원본 검색어가 0매칭일 때 사용한다: + - precise: 모델명/규격을 반영한 정밀 검색어("스탠리 텀블러" → "스탠리 퀜처 887ml") → 매칭률↑ + - broad: 핵심 상품 명사만 남긴 광역 검색어 → 정밀도 0건이면 폭넓게 최종 확인 +레퍼런스 keyword_maker 의 규칙(모델명 우선, 일반 카테고리어 제거)을 structured output 으로 이식. +""" + +import os + +from openai import AsyncOpenAI +from pydantic import BaseModel, Field + +from common.logger import LOG + +_SYSTEM = ( + "너는 커머스 검색어 생성기다. 상품 정보를 받아 쇼핑몰 검색창에 넣을 한국어 검색어 2개를 만든다.\n" + "- precise: 모델명(있으면 최우선)과 핵심 규격(용량/사이즈/개입 등)을 포함한 정밀 검색어. " + "동일 상품을 정확히 찾기 위함. 일반 카테고리 단어만 나열하지 말 것.\n" + "- broad: 핵심 상품 명사(브랜드+제품군) 위주의 광역 검색어. precise 가 0건일 때 폭넓게 찾기 위함.\n" + "검색어에 따옴표/특수문자/불필요한 수식어를 넣지 말 것." +) + + +class Keywords(BaseModel): + precise: str = Field(description="모델/규격 포함 정밀 검색어") + broad: str = Field(description="핵심 명사 위주 광역 검색어") + + +class KeywordGenerator: + 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 generate(self, target: dict) -> Keywords: + user = ( + f"상품명: {target.get('product_name', '')}\n" + f"모델: {target.get('model', '')}\n" + f"규격: {target.get('specification', '')}\n" + f"제조사/브랜드: {target.get('company', '')}" + ) + resp = await self._client.beta.chat.completions.parse( + model=self._model, + messages=[{"role": "system", "content": _SYSTEM}, {"role": "user", "content": user}], + response_format=Keywords, + temperature=0, + ) + kw = resp.choices[0].message.parsed or Keywords(precise="", broad="") + LOG.d(f"[ai] 검색어 생성 precise={kw.precise!r} broad={kw.broad!r}") + return kw diff --git a/lps/tests/test_negative_cache.py b/lps/tests/test_negative_cache.py new file mode 100644 index 0000000..7a9d6e6 --- /dev/null +++ b/lps/tests/test_negative_cache.py @@ -0,0 +1,36 @@ +"""네거티브 캐시 CRUD 테스트 (실 lps_db).""" + +import pytest_asyncio +from sqlalchemy import text + +from crud.negative_cache import NegativeCache + + +@pytest_asyncio.fixture +async def nc(db_engine): + async with db_engine.begin() as conn: + await conn.execute(text("TRUNCATE search_negative")) + return NegativeCache() + + +async def test_put_then_is_negative(nc): + assert await nc.is_negative("P1") is False + await nc.put("P1", ttl_sec=3600) + assert await nc.is_negative("P1") is True + + +async def test_expired_is_not_negative(nc): + await nc.put("P2", ttl_sec=-1) # 이미 만료 + assert await nc.is_negative("P2") is False + + +async def test_upsert_refreshes_ttl(nc): + await nc.put("P3", ttl_sec=-1) # 만료 상태 + assert await nc.is_negative("P3") is False + await nc.put("P3", ttl_sec=3600) # 갱신 → 유효 + assert await nc.is_negative("P3") is True + + +async def test_empty_key_is_noop(nc): + await nc.put("", ttl_sec=3600) + assert await nc.is_negative("") is False diff --git a/lps/tests/test_search_handler.py b/lps/tests/test_search_handler.py index 26a85ee..9f3da8d 100644 --- a/lps/tests/test_search_handler.py +++ b/lps/tests/test_search_handler.py @@ -1,4 +1,4 @@ -"""다중 소스 검색 핸들러 테스트 — 병합·소스별 실패 격리·전체 실패 시 잡 실패 (fake 어댑터).""" +"""검색 핸들러 테스트 — 병합·실패격리·AI판정·재정제 루프·not_found·네거티브 캐시 (fake 의존성).""" import pytest @@ -8,59 +8,23 @@ from worker.handlers import build_search_handler class FakeAdapter: - def __init__(self, source, products=None, fail=False): + def __init__(self, source, by_query=None, products=None, fail=False): self.source = source + self._by_query = by_query # {query: [products]} self._products = products or [] self._fail = fail + self.calls = [] async def search(self, query, limit=40): + self.calls.append(query) if self._fail: raise AdapterError("boom", source=self.source, blocked=True) + if self._by_query is not None: + return self._by_query.get(query, []) return self._products -def _np(source, price): - return NormalizedProduct(source=source, name=f"{source}-{price}", price=price) - - -def _job(): - return {"job_type": JobType.SEARCH.value, "attempts": 1, "payload": {"product_name": "x"}} - - -async def test_merges_and_ranks_across_sources(): - adapters = { - "coupang": FakeAdapter("coupang", [_np("coupang", 3000), _np("coupang", 1000)]), - "naver": FakeAdapter("naver", [_np("naver", 2000), _np("naver", 500)]), - } - r = await build_search_handler(adapters, top_n=3)(_job()) - assert r["lowest"]["price"] == 500 and r["lowest"]["source"] == "naver" - assert [p["price"] for p in r["top"]] == [500, 1000, 2000] - assert r["sources"]["coupang"]["count"] == 2 and r["sources"]["naver"]["count"] == 2 - - -async def test_isolates_single_source_failure(): - adapters = { - "coupang": FakeAdapter("coupang", fail=True), - "naver": FakeAdapter("naver", [_np("naver", 900)]), - } - r = await build_search_handler(adapters)(_job()) - assert "error" in r["sources"]["coupang"] # 실패 격리 - assert r["sources"]["naver"]["count"] == 1 - assert r["lowest"]["price"] == 900 # 성공 소스로 결과 산출 - - -async def test_all_sources_fail_raises(): - adapters = { - "coupang": FakeAdapter("coupang", fail=True), - "naver": FakeAdapter("naver", fail=True), - } - with pytest.raises(RuntimeError): - await build_search_handler(adapters)(_job()) - - class FakeJudge: - """가격 조건으로 매칭을 흉내내는 판정기(실제 OpenAI 호출 없음).""" - def __init__(self, predicate): self._pred = predicate @@ -70,10 +34,89 @@ class FakeJudge: for i, c in enumerate(candidates)] +class FakeKeywordGen: + def __init__(self, precise="", broad=""): + self._p, self._b = precise, broad + + async def generate(self, target): + from services.ai.keyword import Keywords + return Keywords(precise=self._p, broad=self._b) + + +class FakeNegCache: + def __init__(self, negative=False): + self._neg = negative + self.puts = [] + + async def is_negative(self, key): + return self._neg + + async def put(self, key, ttl_sec=86400, reason="x"): + self.puts.append(key) + + +def _np(source, price): + return NormalizedProduct(source=source, name=f"{source}-{price}", price=price) + + +def _job(**payload): + payload.setdefault("product_name", "x") + return {"job_type": JobType.SEARCH.value, "attempts": 1, "payload": payload} + + +# ── 병합 / 실패격리 / AI 판정 (round 0) ───────────────────────────── +async def test_merges_and_ranks_across_sources(): + adapters = { + "coupang": FakeAdapter("coupang", products=[_np("coupang", 3000), _np("coupang", 1000)]), + "naver": FakeAdapter("naver", products=[_np("naver", 2000), _np("naver", 500)]), + } + r = await build_search_handler(adapters, top_n=3)(_job()) + assert r["outcome"] == "found" and r["lowest"]["price"] == 500 + assert [p["price"] for p in r["top"]] == [500, 1000, 2000] + + +async def test_isolates_single_source_failure_but_still_found(): + adapters = {"coupang": FakeAdapter("coupang", fail=True), "naver": FakeAdapter("naver", products=[_np("naver", 900)])} + r = await build_search_handler(adapters)(_job()) + assert r["outcome"] == "found" and r["lowest"]["price"] == 900 + assert "error" in r["sources"]["coupang"] + + 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 + adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 1000), _np("naver", 2000), _np("naver", 3000)])} + r = await build_search_handler(adapters, judge=FakeJudge(lambda c: c.price == 2000))(_job()) + assert [p["price"] for p in r["top"]] == [2000] + assert next(s for s in r["stages"] if s["stage"] == "ai_match")["out"] == 1 + + +# ── 재정제 루프 ──────────────────────────────────────────────────── +async def test_refines_to_precise_query_when_original_empty(): + adapters = {"naver": FakeAdapter("naver", by_query={"스탠리 퀜처 887ml": [_np("naver", 40000)]})} # 원본은 0건 + kw = FakeKeywordGen(precise="스탠리 퀜처 887ml", broad="스탠리 텀블러") + r = await build_search_handler(adapters, keyword_gen=kw)(_job(product_name="스탠리 텀블러")) + assert r["outcome"] == "found" and r["round"] == "precise" and r["rounds_tried"] == 2 + assert r["lowest"]["price"] == 40000 + + +async def test_not_found_after_all_rounds_and_caches(): + adapters = {"naver": FakeAdapter("naver", by_query={})} # 어떤 쿼리든 0건 + kw = FakeKeywordGen(precise="P", broad="B") + neg = FakeNegCache() + r = await build_search_handler(adapters, keyword_gen=kw, neg_cache=neg)(_job(product_code="PC1", product_name="없는상품")) + assert r["outcome"] == "not_found" and r["rounds_tried"] == 3 + assert r["lowest"] is None and r["top"] == [] + assert neg.puts == ["PC1"] # 네거티브 캐시에 기록 + + +async def test_negative_cache_short_circuits(): + adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 100)])} + neg = FakeNegCache(negative=True) + r = await build_search_handler(adapters, neg_cache=neg)(_job(product_code="PC9")) + assert r["outcome"] == "not_found" and r["cached"] is True + assert adapters["naver"].calls == [] # 재검색 안 함 + + +async def test_technical_failure_with_zero_match_raises(): + adapters = {"coupang": FakeAdapter("coupang", fail=True), "naver": FakeAdapter("naver", by_query={})} + with pytest.raises(RuntimeError): + await build_search_handler(adapters)(_job()) # 0매칭 + 차단 → 기술 재시도 diff --git a/lps/worker/handlers.py b/lps/worker/handlers.py index 4936ed9..bc8ccb3 100644 --- a/lps/worker/handlers.py +++ b/lps/worker/handlers.py @@ -1,8 +1,16 @@ """잡 핸들러 — job_type 별 처리. 현재는 SEARCH(검색)만. -검색 핸들러: 여러 소스 어댑터를 동시 검색 → 병합 → 코어 파이프라인(필터·이상치·top-N 최저가). -소스별 실패는 격리한다(한 소스가 죽어도 나머지로 결과 산출). 모든 소스 실패 시에만 잡 실패(재시도). -AI 유사도 판정은 파이프라인 슬롯에 키 준비 시 결합한다. +검색 핸들러 = 한정된 재정제 루프 + 명시적 outcome: + 0. 네거티브 캐시 확인(최근 not_found면 즉시 반환) + 각 라운드(원본 → 정밀(LLM) → 광역, 최대 max_rounds): + 소스 동시 검색 → 필터 → 이상치 → AI 같은상품 판정 + ├ 매칭 있음 → DONE(outcome=found), 조기 종료 + ├ 0매칭 + 기술적 실패(차단/예외) 있음 → raise → 큐가 잡 전체 백오프 재시도(→소진 시 DEAD) + └ 0매칭 + 소스 정상 → 다음 라운드 + 라운드 소진 → DONE(outcome=not_found) + 네거티브 캐시 기록 + +두 재시도 축을 분리한다: 기술적(큐 attempts/백오프) ≠ 검색어(refine 라운드, 유한). +'못 찾음'은 정상 종료(DONE)지 dead-letter 가 아니다. """ import asyncio @@ -20,52 +28,87 @@ def build_search_handler( limit: int = 40, top_n: int = 5, judge=None, + keyword_gen=None, + max_rounds: int = 3, + neg_cache=None, ): - """검색 핸들러 생성. adapters = {source: SearchAdapter}. sources 미지정 시 전체 사용. - judge(SimilarityJudge) 주입 시 필터 뒤·top-N 앞에 '같은 상품' AI 판정을 끼운다(없으면 생략).""" + """검색 핸들러 생성. + judge: SimilarityJudge(같은 상품 판정) / keyword_gen: KeywordGenerator(정밀·광역 재검색어) / + neg_cache: NegativeCache(TTL not_found 캐시). 모두 선택 — 없으면 해당 단계 생략.""" use = list(sources) if sources else list(adapters.keys()) + async def _search_round(query: str): + """한 라운드: 모든 소스 동시 검색 → (products, per_source, tech_failed).""" + results = await asyncio.gather(*[adapters[s].search(query, limit=limit) for s in use], return_exceptions=True) + products, per_source, tech_failed = [], {}, False + for src, res in zip(use, results): + if isinstance(res, Exception): + tech_failed = True + per_source[src] = {"error": f"{type(res).__name__}: {res}"} + LOG.w(f"[{src}] 검색 실패: {type(res).__name__}: {res}") + else: + products.extend(res) + per_source[src] = {"count": len(res)} + return products, per_source, tech_failed + + async def _round_queries(base_query: str, target: dict): + """라운드 쿼리 지연 생성: 원본 → (0매칭 시에만 LLM 호출로) 정밀 → 광역.""" + yield ("original", base_query) + if keyword_gen is not None: + kw = await keyword_gen.generate(target) # 원본이 실패해 여기까지 온 경우에만 호출됨 + seen = {base_query} + for label, q in (("precise", kw.precise), ("broad", kw.broad)): + q = (q or "").strip() + if q and q not in seen: + seen.add(q) + yield (label, q) + async def handler(job: dict) -> dict: if job["job_type"] != JobType.SEARCH.value: raise ValueError(f"unsupported job_type: {job['job_type']}") payload = job.get("payload") or {} - query = (payload.get("product_name") or "").strip() - if not query: + base_query = (payload.get("product_name") or "").strip() + if not base_query: raise ValueError("empty product_name") + target = {k: payload.get(k, "") for k in ("product_name", "model", "specification", "company")} base_price = parse_price(payload.get("price")) + cache_key = payload.get("product_code") or base_query - # 소스 동시 검색 (실패는 예외로 수거해 격리) - results = await asyncio.gather( - *[adapters[s].search(query, limit=limit) for s in use], - return_exceptions=True, - ) + # 0) 네거티브 캐시 — 최근 not_found면 재검색 생략 + if neg_cache is not None and await neg_cache.is_negative(cache_key): + return {"outcome": "not_found", "cached": True, "query": base_query, + "rounds_tried": 0, "lowest": None, "top": [], "stages": [], "sources": {}} - products = [] - per_source: dict[str, dict] = {} - for src, res in zip(use, results): - if isinstance(res, Exception): - LOG.w(f"[{src}] 검색 실패: {type(res).__name__}: {res}") - per_source[src] = {"error": f"{type(res).__name__}: {res}"} - else: - products.extend(res) - per_source[src] = {"count": len(res)} + rounds_done = 0 + last_stages, last_sources = [], {} + async for label, query in _round_queries(base_query, target): + if rounds_done >= max_rounds: + break + rounds_done += 1 - if not products and all("error" in v for v in per_source.values()): - raise RuntimeError(f"모든 소스 검색 실패: {per_source}") # 잡 실패 → 재시도 + products, per_source, tech_failed = await _search_round(query) + candidates, stages = apply_filters(products, base_price=base_price) + if judge is not None and candidates: + 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 + last_stages, last_sources = stages, per_source - # 필터(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 + if candidates: # 찾음 → 조기 종료 + result = rank_result(candidates, len(products), stages, top_n) + result.update(outcome="found", query=query, round=label, rounds_tried=rounds_done, sources=per_source) + return result - result = rank_result(candidates, len(products), stages, top_n) - result["query"] = query - result["sources"] = per_source # 소스별 건수/에러 (관측) + if tech_failed: # 0매칭인데 소스가 죽어 있었음 → '없음'이라 단정 불가 → 기술 재시도 + raise RuntimeError(f"기술적 실패로 0매칭(round={label}) — 잡 재시도: {per_source}") + + # 모든 라운드 클린 0매칭 → 정상 not_found 종료 + if neg_cache is not None: + await neg_cache.put(cache_key, reason=f"not_found after {rounds_done} rounds") + result = rank_result([], 0, last_stages, top_n) + result.update(outcome="not_found", query=base_query, rounds_tried=rounds_done, sources=last_sources) return result return handler diff --git a/lps/worker_main.py b/lps/worker_main.py index 793b700..702fdb8 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -11,9 +11,11 @@ import os from common.logger import LOG from config.server_configs import web_server_config from crud.job_crud import JobQueue +from crud.negative_cache import NegativeCache from services.search.coupang.adapter import CoupangAdapter from services.search.naver.adapter import NaverAdapter from services.ai.similarity import SimilarityJudge +from services.ai.keyword import KeywordGenerator from worker.handlers import build_search_handler from worker.notify import JobListener from worker.runner import Worker, run_reaper @@ -25,10 +27,12 @@ async def main(concurrency: int = 1): queue = JobQueue() # 쿠팡(브라우저, 무거움) + 네이버(오픈API, 가벼움) 동시 검색 → 병합 최저가 adapters = {"coupang": CoupangAdapter(headless=False), "naver": NaverAdapter()} - # 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) + # OPENAI_API_KEY 있으면 '같은 상품' AI 판정 + 재검색어 생성 활성화 + has_openai = bool(os.environ.get("OPENAI_API_KEY")) + judge = SimilarityJudge() if has_openai else None + keyword_gen = KeywordGenerator() if has_openai else None + LOG.i(f"AI(판정+검색어생성): {'ON' if has_openai else 'OFF(키 없음)'}") + handler = build_search_handler(adapters, judge=judge, keyword_gen=keyword_gen, neg_cache=NegativeCache()) stop = asyncio.Event() listeners: list[JobListener] = []