From 69f6641c276d03737167ce0a1c0d825c51c31e2f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EB=AF=BC=ED=97=8C?= Date: Thu, 9 Jul 2026 15:14:35 +0900 Subject: [PATCH] =?UTF-8?q?feat(lps):=20=EA=B2=80=EC=83=89=201=EA=B1=B4=20?= =?UTF-8?q?=EC=9B=90=EA=B0=80=20=EA=B3=84=EC=B8=A1=20=E2=80=94=20AI=20?= =?UTF-8?q?=ED=86=A0=ED=81=B0=C2=B7=EB=B9=84=EC=9A=A9=20+=20=EC=8B=9C?= =?UTF-8?q?=EA=B0=84=20+=20=ED=81=AC=EB=A1=A4=20=ED=8A=B8=EB=9E=98?= =?UTF-8?q?=ED=94=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 검색이 소모하는 리소스/비용/시간을 잡 단위로 집계해 result.metrics 로 적재(API/FE 노출). 지금까진 타임스탬프만 있고 실제 비용 동인(AI 토큰·대역폭)은 버려지고 있었다. - services/metrics.SearchMetrics: duration_ms + ai(calls/tokens/est_cost_usd, gpt-4o-mini 단가) + crawl(fetches/html_bytes/malls_crawled) + source_ms - AI 클라이언트: resp.usage 를 last_usage 로 노출(그동안 폐기하던 토큰) - 어댑터: last_bytes(처리 HTML 바이트) 노출 — naver/coupang/browser_base 공통 - handler: 각 fetch 타이밍+바이트, AI 호출 토큰을 metrics 로 누적 → 결과에 스냅샷 - FE: 작업 카드에 원가 4타일(소요/AI비용/토큰/크롤 트래픽) - 테스트 2종. ⚠️ html_bytes 는 대역폭 근사(오픈마켓 리소스 미차단분 제외=하한), CDP 정확화는 백로그 Co-Authored-By: Claude Fable 5 --- lps/docs/api.md | 9 +++- lps/services/ai/keyword.py | 2 + lps/services/ai/similarity.py | 3 ++ lps/services/metrics.py | 75 ++++++++++++++++++++++++++ lps/services/search/browser_base.py | 2 + lps/services/search/coupang/adapter.py | 2 + lps/services/search/naver/adapter.py | 3 ++ lps/tests/test_search_handler.py | 36 ++++++++++++- lps/worker/handlers.py | 52 ++++++++++++------ lps/worker_main.py | 1 + 10 files changed, 168 insertions(+), 17 deletions(-) create mode 100644 lps/services/metrics.py diff --git a/lps/docs/api.md b/lps/docs/api.md index f497f5c..c52dacf 100644 --- a/lps/docs/api.md +++ b/lps/docs/api.md @@ -81,11 +81,18 @@ curl -X POST localhost:9600/v1/lps/search -H 'Content-Type: application/json' \ "shipping_fee": 0, "shipping_type": "rocket" }, "top": [ /* 최저가 상위 N개 */ ], "sources": { "naver": {"count": 40}, "coupang": {"count": 40} }, - "stages": [ {"stage":"outlier","in":80,"out":76}, {"stage":"ai_match","in":76,"out":1}, {"stage":"top_n","in":1,"out":1} ] + "stages": [ {"stage":"outlier","in":80,"out":76}, {"stage":"ai_match","in":76,"out":1}, {"stage":"top_n","in":1,"out":1} ], + "metrics": { // 이 검색 1건이 쓴 리소스/비용/시간 + "duration_ms": 21500, + "ai": { "calls": 2, "prompt_tokens": 5200, "completion_tokens": 180, "est_cost_usd": 0.000888 }, + "crawl": { "fetches": 3, "html_bytes": 1560000, "malls_crawled": ["gmarket"] }, + "source_ms": { "naver": 480, "coupang": 12300, "gmarket": 8700 } + } } } ``` - `output.stages` = 각 단계에서 몇 건이 걸러졌는지(디버깅·품질 확인용). +- `output.metrics` = 검색 1건의 원가. `ai`(호출·토큰·추정 비용$), `crawl`(fetch 수·처리 HTML 바이트·크롤한 몰), `source_ms`(소스별 소요), `duration_ms`(전체). ※ `html_bytes`는 처리한 응답 본문 기준(대역폭 근사) — 오픈마켓은 리소스 미차단이라 실제 대역폭보다 작다(하한). - 없는 job_id/잘못된 형식 → `result.desc = "LPS_JOB_NOT_FOUND"`. > **⚠️ 가격의 의미 (배송비)** diff --git a/lps/services/ai/keyword.py b/lps/services/ai/keyword.py index f39eb4d..ed561b5 100644 --- a/lps/services/ai/keyword.py +++ b/lps/services/ai/keyword.py @@ -30,6 +30,7 @@ class KeywordGenerator: 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(계측용) async def generate(self, target: dict) -> Keywords: user = ( @@ -44,6 +45,7 @@ class KeywordGenerator: response_format=Keywords, temperature=0, ) + self.last_usage = getattr(resp, "usage", None) 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/services/ai/similarity.py b/lps/services/ai/similarity.py index 70af0ff..ea66a3b 100644 --- a/lps/services/ai/similarity.py +++ b/lps/services/ai/similarity.py @@ -37,9 +37,11 @@ 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 리스트 반환.""" + self.last_usage = None if not candidates: return [] @@ -60,6 +62,7 @@ class SimilarityJudge: response_format=JudgmentList, temperature=0, ) + self.last_usage = getattr(resp, "usage", None) parsed = resp.choices[0].message.parsed by_idx = {j.index: j for j in (parsed.judgments if parsed else [])} # 누락된 후보는 보수적으로 불일치 처리 diff --git a/lps/services/metrics.py b/lps/services/metrics.py new file mode 100644 index 0000000..09dfb4e --- /dev/null +++ b/lps/services/metrics.py @@ -0,0 +1,75 @@ +"""검색 1건의 리소스/비용/시간 계측(관측용). + +한 상품 검색이 소모하는 것을 잡 단위로 집계한다: + - 시간: 전체 소요 + 소스별 소요(ms) + - AI: 호출 수 + 토큰(prompt/completion) + 추정 비용($) + - 크롤: 외부 fetch 수 + 처리 HTML 바이트 + 크롤한 몰 + +핸들러가 각 호출을 타이밍하고, AI 클라이언트/어댑터가 노출하는 last_usage/last_bytes 를 읽어 누적한다. +결과는 job.result.metrics 로 적재돼 API/FE 에서 검색 원가를 확인할 수 있다. + +⚠️ HTML 바이트는 '처리한 응답 본문' 기준(대역폭 근사). 쿠팡은 리소스 차단이라 실제와 근접하지만, +오픈마켓(리소스 미차단)은 CSS/JS/이미지가 빠져 실제 대역폭보다 작다(하한). 정확 대역폭은 백로그(CDP). +""" + +import time + +# 모델별 1M 토큰당 단가(USD). input=prompt, output=completion. 모르면 0(비용 미추정). +_PRICING = { + "gpt-4o-mini": (0.150, 0.600), + "gpt-4o": (2.50, 10.00), + "gpt-4.1-mini": (0.40, 1.60), +} + + +def estimate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float: + inp, out = _PRICING.get(model, (0.0, 0.0)) + return round(prompt_tokens / 1_000_000 * inp + completion_tokens / 1_000_000 * out, 6) + + +class SearchMetrics: + """검색 1건의 누적 계측기. 스레드/코루틴 공유 X — 잡마다 새로 만든다.""" + + def __init__(self, model: str = ""): + self._start = time.monotonic() + self._model = model + self.ai_calls = 0 + self.ai_prompt = 0 + self.ai_completion = 0 + self.fetches = 0 + self.html_bytes = 0 + self.source_ms: dict[str, int] = {} + self.crawled_malls: list[str] = [] + + def add_ai(self, usage): + """OpenAI resp.usage(prompt_tokens/completion_tokens) 누적. None 이면 무시.""" + if not usage: + return + self.ai_calls += 1 + self.ai_prompt += getattr(usage, "prompt_tokens", 0) or 0 + self.ai_completion += getattr(usage, "completion_tokens", 0) or 0 + + def add_fetch(self, source: str, html_bytes: int, ms: int, crawl: bool = False): + """외부 fetch 1건(소스·바이트·소요) 누적. crawl=True 면 오픈마켓 폴백 크롤.""" + self.fetches += 1 + self.html_bytes += html_bytes or 0 + self.source_ms[source] = self.source_ms.get(source, 0) + ms + if crawl and source not in self.crawled_malls: + self.crawled_malls.append(source) + + def snapshot(self) -> dict: + return { + "duration_ms": int((time.monotonic() - self._start) * 1000), + "ai": { + "calls": self.ai_calls, + "prompt_tokens": self.ai_prompt, + "completion_tokens": self.ai_completion, + "est_cost_usd": estimate_cost(self._model, self.ai_prompt, self.ai_completion), + }, + "crawl": { + "fetches": self.fetches, + "html_bytes": self.html_bytes, + "malls_crawled": self.crawled_malls, + }, + "source_ms": self.source_ms, + } diff --git a/lps/services/search/browser_base.py b/lps/services/search/browser_base.py index db8ddda..17a468f 100644 --- a/lps/services/search/browser_base.py +++ b/lps/services/search/browser_base.py @@ -67,6 +67,7 @@ class BrowserSearchAdapter(SearchAdapter): self._lock = asyncio.Lock() self._ok = 0 self._blocked = 0 + self.last_bytes = 0 # 직전 search 의 처리 HTML 바이트(계측용) — 호출부가 await 직후 읽는다 # ---- 사이트별 훅 -------------------------------------------------- @abstractmethod @@ -149,6 +150,7 @@ class BrowserSearchAdapter(SearchAdapter): except Exception as ex: raise AdapterError(f"{self.source} 검색 실패: {ex}", source=self.source) from ex + self.last_bytes = len(html.encode("utf-8")) products = self._parse(html) if products: self._ok += 1 diff --git a/lps/services/search/coupang/adapter.py b/lps/services/search/coupang/adapter.py index 628d89b..3053334 100644 --- a/lps/services/search/coupang/adapter.py +++ b/lps/services/search/coupang/adapter.py @@ -69,6 +69,7 @@ class CoupangAdapter(SearchAdapter): self._lock = asyncio.Lock() self._ok = 0 self._blocked = 0 + self.last_bytes = 0 # 직전 search 의 처리 HTML 바이트(계측용) async def _route(self, route): if route.request.resource_type in _BLOCKED_RESOURCES: @@ -130,6 +131,7 @@ class CoupangAdapter(SearchAdapter): except Exception as ex: raise AdapterError(f"쿠팡 검색 실패: {ex}", source=self.source) from ex + self.last_bytes = len(html.encode("utf-8")) products = parse_search_html(html, source=self.source) if products: self._ok += 1 diff --git a/lps/services/search/naver/adapter.py b/lps/services/search/naver/adapter.py index 691f8e6..abee4a3 100644 --- a/lps/services/search/naver/adapter.py +++ b/lps/services/search/naver/adapter.py @@ -33,6 +33,7 @@ class NaverAdapter(SearchAdapter): self._timeout = timeout self._ok = 0 self._blocked = 0 + self.last_bytes = 0 # 직전 search 의 응답 바이트(계측용) def _headers(self) -> dict: cid, csec = self._keys[self._idx] @@ -45,6 +46,7 @@ class NaverAdapter(SearchAdapter): if not self._keys: raise AdapterError("네이버 API 키 없음(config.local.toml [NaverConfig].keys)", source=self.source) + self.last_bytes = 0 collected: list[dict] = [] async with httpx.AsyncClient(timeout=self._timeout) as client: start = 1 @@ -74,6 +76,7 @@ class NaverAdapter(SearchAdapter): await self._rl.wait() r = await client.get(_API, params=params, headers=self._headers()) if r.status_code == 200: + self.last_bytes += len(r.content) return r.json() if r.status_code in (429, 403): self._blocked += 1 diff --git a/lps/tests/test_search_handler.py b/lps/tests/test_search_handler.py index eceb26e..2ff668a 100644 --- a/lps/tests/test_search_handler.py +++ b/lps/tests/test_search_handler.py @@ -24,9 +24,15 @@ class FakeAdapter: return self._products +class _Usage: + def __init__(self, prompt, completion): + self.prompt_tokens, self.completion_tokens = prompt, completion + + class FakeJudge: - def __init__(self, predicate): + def __init__(self, predicate, usage=None): self._pred = predicate + self.last_usage = usage # 계측용(핸들러가 judge 후 읽음) async def judge(self, target, candidates): from services.ai.similarity import Judgment @@ -37,6 +43,7 @@ class FakeJudge: class FakeKeywordGen: def __init__(self, precise="", broad=""): self._p, self._b = precise, broad + self.last_usage = None async def generate(self, target): from services.ai.keyword import Keywords @@ -151,6 +158,33 @@ async def test_fallback_failure_is_isolated(): assert r["outcome"] == "found" and r["lowest"]["price"] == 9000 # 폴백 실패해도 정상 종료 +# ── 검색 원가 계측(metrics) ──────────────────────────────────────── +async def test_metrics_recorded_in_result(): + adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 1000), _np("naver", 2000)])} + adapters["naver"].last_bytes = 1234 + judge = FakeJudge(lambda c: True, usage=_Usage(500, 40)) + r = await build_search_handler(adapters, judge=judge, ai_model="gpt-4o-mini")(_job()) + m = r["metrics"] + assert m["ai"]["calls"] == 1 and m["ai"]["prompt_tokens"] == 500 and m["ai"]["completion_tokens"] == 40 + assert m["ai"]["est_cost_usd"] == round(500/1e6*0.15 + 40/1e6*0.60, 6) # gpt-4o-mini 단가 + assert m["crawl"]["fetches"] == 1 and m["crawl"]["html_bytes"] == 1234 + assert "naver" in m["source_ms"] and "duration_ms" in m + + +async def test_metrics_counts_fallback_crawl(): + adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 5000, mall="네이버")])} + st11 = FakeAdapter("st11", products=[_np("st11", 3000, mall="11번가")]) + st11.last_bytes = 9999 + r = await build_search_handler( + adapters, judge=FakeJudge(lambda c: True), + fallback_adapters={"st11": st11}, + )(_job()) + m = r["metrics"] + assert m["crawl"]["fetches"] == 2 # naver + st11 크롤 + assert "st11" in m["crawl"]["malls_crawled"] # 폴백 크롤 몰 기록 + assert m["crawl"]["html_bytes"] == 9999 # st11 바이트 포함 + + async def test_fallback_dedup_same_mall_keeps_lowest(): # 네이버 매칭에 G마켓 없음 → 크롤. 크롤 G마켓이 네이버 '네이버몰'보다 싸면 최저가 갱신. adapters = {"naver": FakeAdapter("naver", products=[_np("naver", 8000, mall="네이버")])} diff --git a/lps/worker/handlers.py b/lps/worker/handlers.py index 54ed8b2..13dccee 100644 --- a/lps/worker/handlers.py +++ b/lps/worker/handlers.py @@ -14,9 +14,11 @@ """ import asyncio +import time from common.enums import JobType from common.logger import LOG +from services.metrics import SearchMetrics from services.search.contract import SearchAdapter, NormalizedProduct from services.search.card_parser import canonical_mall, MALL_BY_SOURCE from services.search.util import parse_price @@ -52,6 +54,7 @@ def build_search_handler( neg_cache=None, history=None, fallback_adapters: dict[str, SearchAdapter] | None = None, + ai_model: str = "", ): """검색 핸들러 생성. judge: SimilarityJudge(같은 상품 판정) / keyword_gen: KeywordGenerator(정밀·광역 재검색어) / @@ -70,9 +73,21 @@ def build_search_handler( except Exception as ex: LOG.e_no_callstack(f"[history] 스냅샷 기록 실패(무시): {ex}") - 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) + async def _timed_search(adapter, query: str, source: str, metrics: SearchMetrics, crawl: bool): + """어댑터 검색 1건을 타이밍+바이트 계측하며 실행. 예외는 그대로 전파(호출부가 처리).""" + t0 = time.monotonic() + try: + res = await adapter.search(query, limit=limit) + metrics.add_fetch(source, getattr(adapter, "last_bytes", 0), int((time.monotonic() - t0) * 1000), crawl=crawl) + return res + except Exception: + metrics.add_fetch(source, 0, int((time.monotonic() - t0) * 1000), crawl=crawl) + raise + + async def _search_round(query: str, metrics: SearchMetrics): + """한 라운드: 모든 소스 동시 검색 → (products, per_source, tech_failed). 소스별 시간/바이트 계측.""" + results = await asyncio.gather(*[_timed_search(adapters[s], query, s, metrics, False) for s in use], + return_exceptions=True) products, per_source, tech_failed = [], {}, False for src, res in zip(use, results): if isinstance(res, Exception): @@ -84,15 +99,16 @@ def build_search_handler( per_source[src] = {"count": len(res)} return products, per_source, tech_failed - async def _match(target: dict, products: list, base_price): - """필터 → (있으면) AI 같은상품 판정 → 매칭 후보. 폴백 크롤 결과 판정에도 재사용.""" + async def _match(target: dict, products: list, base_price, metrics: SearchMetrics): + """필터 → (있으면) AI 같은상품 판정 → 매칭 후보. AI 토큰은 metrics 에 누적.""" candidates, _ = apply_filters(products, base_price=base_price) if judge is not None and candidates: verdicts = await judge.judge(target, candidates) + metrics.add_ai(judge.last_usage) candidates = [c for c, v in zip(candidates, verdicts) if v.is_match] return candidates - async def _enrich_with_fallback(target: dict, query: str, matched: list, base_price): + async def _enrich_with_fallback(target: dict, query: str, matched: list, base_price, metrics: SearchMetrics): """네이버가 커버 못 한 오픈마켓만 직접 크롤(폴백) → 같은상품 판정 후 병합. 사용자 규칙: '네이버로 그 몰 값 확보 성공 → 그 값, 실패(몰 없음) → 실사이트 크롤'.""" if not fallbacks: @@ -103,21 +119,22 @@ def build_search_handler( if mall in covered: # 네이버가 이미 그 몰 최저가 확보 → 크롤 생략 continue try: - crawled = await adapter.search(query, limit=limit) + crawled = await _timed_search(adapter, query, src, metrics, crawl=True) except Exception as ex: LOG.w(f"[fallback:{src}] 크롤 실패(무시): {type(ex).__name__}: {ex}") continue - hits = await _match(target, crawled, base_price) + hits = await _match(target, crawled, base_price, metrics) if hits: LOG.d(f"[fallback:{src}] 크롤 {len(crawled)}건 중 같은상품 {len(hits)}건 병합") matched = matched + hits return matched - async def _round_queries(base_query: str, target: dict): + async def _round_queries(base_query: str, target: dict, metrics: SearchMetrics): """라운드 쿼리 지연 생성: 원본 → (0매칭 시에만 LLM 호출로) 정밀 → 광역.""" yield ("original", base_query) if keyword_gen is not None: kw = await keyword_gen.generate(target) # 원본이 실패해 여기까지 온 경우에만 호출됨 + metrics.add_ai(keyword_gen.last_usage) seen = {base_query} for label, q in (("precise", kw.precise), ("broad", kw.broad)): q = (q or "").strip() @@ -136,23 +153,26 @@ def build_search_handler( 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 + metrics = SearchMetrics(ai_model) # 검색 1건의 리소스/비용/시간 계측 # 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": {}} + "rounds_tried": 0, "lowest": None, "top": [], "stages": [], "sources": {}, + "metrics": metrics.snapshot()} rounds_done = 0 last_stages, last_sources = [], {} - async for label, query in _round_queries(base_query, target): + async for label, query in _round_queries(base_query, target, metrics): if rounds_done >= max_rounds: break rounds_done += 1 - products, per_source, tech_failed = await _search_round(query) + products, per_source, tech_failed = await _search_round(query, metrics) candidates, stages = apply_filters(products, base_price=base_price) if judge is not None and candidates: verdicts = await judge.judge(target, candidates) + metrics.add_ai(judge.last_usage) 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 @@ -160,11 +180,12 @@ def build_search_handler( if candidates: # 찾음 → 오픈마켓 폴백 보강 후 종료 before = len(candidates) - candidates = await _enrich_with_fallback(target, query, candidates, base_price) + candidates = await _enrich_with_fallback(target, query, candidates, base_price, metrics) if len(candidates) > before: stages.append({"stage": "fallback_crawl", "in": before, "out": len(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) + result.update(outcome="found", query=query, round=label, rounds_tried=rounds_done, + sources=per_source, metrics=metrics.snapshot()) await _record_history(cache_key, job.get("job_id"), "found", candidates) return result @@ -175,7 +196,8 @@ def build_search_handler( 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) + result.update(outcome="not_found", query=base_query, rounds_tried=rounds_done, + sources=last_sources, metrics=metrics.snapshot()) await _record_history(cache_key, job.get("job_id"), "not_found", []) return result diff --git a/lps/worker_main.py b/lps/worker_main.py index e958714..d3e7749 100644 --- a/lps/worker_main.py +++ b/lps/worker_main.py @@ -57,6 +57,7 @@ async def main(concurrency: int = 1): adapters, judge=judge, keyword_gen=keyword_gen, neg_cache=NegativeCache(), history=PriceHistory(), fallback_adapters=fallback_adapters, + ai_model=openai_config.model, ) stop = asyncio.Event()