feat(lps): 검색 1건 원가 계측 — AI 토큰·비용 + 시간 + 크롤 트래픽

검색이 소모하는 리소스/비용/시간을 잡 단위로 집계해 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 <noreply@anthropic.com>

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"`.
 
 > **⚠️ 가격의 의미 (배송비)**

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

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 [])}
         # 누락된 후보는 보수적으로 불일치 처리

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,
+        }

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

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

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

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="네이버")])}

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):
+    async def _timed_search(adapter, query: str, source: str, metrics: SearchMetrics, crawl: bool):
-        """한 라운드: 모든 소스 동시 검색 → (products, per_source, tech_failed)."""
+        """어댑터 검색 1건을 타이밍+바이트 계측하며 실행. 예외는 그대로 전파(호출부가 처리)."""
-        results = await asyncio.gather(*[adapters[s].search(query, limit=limit) for s in use], return_exceptions=True)
+        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):
+    async def _match(target: dict, products: list, base_price, metrics: SearchMetrics):
-        """필터 → (있으면) AI 같은상품 판정 → 매칭 후보. 폴백 크롤 결과 판정에도 재사용."""
+        """필터 → (있으면) 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
 

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()