feat(concurrency): trace 一致性 hash 亲和 + 透传 stream 优化 (#20)

配合 pricebot 单机多进程:透传层按 trace_id 把同一次比价/领券的所有帧
亲和到同一 pricebot 实例,使进程内内存维护 state、无需 Redis。本仓改动:

- 一致性 hash: 新增 app/core/pricebot_router.py(ketama 环,1000 虚拟节点)
  + pick_pricebot(trace_id),md5 纯函数、跨 worker 一致;实测负载偏差 4%、
  6→7 扩容只重映射 ~13%(取模会 ~85%)。
- 实例列表: app/core/config.py 加 PRICEBOT_INSTANCES(逗号分隔)+
  pricebot_instances property;留空 fallback 单实例,向后兼容。
- 透传亲和 + stream: api/v1/compare.py + coupon.py 改为读原始字节、只
  loads 一次拿 trace_id 选实例、转发原始 bytes 不重序列化(省 ~半透传 CPU,
  单 worker 即可扛高并发,避开多 worker 破坏短信内存表的坑);覆盖
  intent/recognize、intent/step、price/step、coupon/step 四端点。
- .env.example 加 PRICEBOT_INSTANCES + PRICEBOT_COMPARE_TIMEOUT_SEC。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: pure <pure@192.168.0.104>
Reviewed-on: #20

app/api/v1/compare.py

@@ -18,6 +18,7 @@ pricebot 协议文档:
 """
 from __future__ import annotations
 
+import json
 import logging
 from typing import Any
 
@@ -25,6 +26,7 @@ import httpx
 from fastapi import APIRouter, HTTPException, Request, status
 
 from app.core.config import settings
+from app.core.pricebot_router import pick_pricebot
 
 logger = logging.getLogger("shagua.compare")
 
@@ -38,25 +40,35 @@ async def _passthrough(request: Request, upstream_path: str) -> dict[str, Any]:
     打日志。比价单帧是大上下文 / 逐帧 LLM,超时用 PRICEBOT_COMPARE_TIMEOUT_SEC(60s,
     比领券的 30s 长)。
     """
+    # 读原始字节,避免"反序列化→再序列化"的双重 JSON(省 ~一半透传 CPU,让 app-server
+    # 单 worker 也扛得住高并发)。只 json.loads 一次拿 trace_id 做亲和 + 打日志,转发时
+    # 直接发原始 bytes(content=raw),不重新 dumps。
+    raw = await request.body()
     try:
-        body = await request.json()
+        meta = json.loads(raw)
     except Exception as e:
         raise HTTPException(status_code=400, detail=f"invalid json body: {e}") from e
+    if not isinstance(meta, dict):
+        meta = {}
 
-    url = f"{settings.PRICEBOT_BASE_URL.rstrip('/')}{upstream_path}"
+    # 按 trace_id 一致性 hash 选 pricebot 实例(同一比价的所有帧落同一进程,内存维护 state)
+    base = pick_pricebot(meta.get("trace_id"))
+    url = f"{base.rstrip('/')}{upstream_path}"
     timeout = settings.PRICEBOT_COMPARE_TIMEOUT_SEC
 
     logger.info(
         "compare %s device_id=%s trace_id=%s step=%s",
         upstream_path,
-        body.get("device_id"),
-        body.get("trace_id"),
-        body.get("step"),
+        meta.get("device_id"),
+        meta.get("trace_id"),
+        meta.get("step"),
     )
 
     try:
         async with httpx.AsyncClient(timeout=timeout) as client:
-            resp = await client.post(url, json=body)
+            resp = await client.post(
+                url, content=raw, headers={"Content-Type": "application/json"}
+            )
     except httpx.RequestError as e:
         logger.error("[pricebot] request failed: %s", e)
         raise HTTPException(

app/api/v1/coupon.py

@@ -10,6 +10,7 @@ pricebot 协议文档:
 """
 from __future__ import annotations
 
+import json
 import logging
 from typing import Any
 
@@ -17,6 +18,7 @@ import httpx
 from fastapi import APIRouter, HTTPException, Request, status
 
 from app.core.config import settings
+from app.core.pricebot_router import pick_pricebot
 
 logger = logging.getLogger("shagua.coupon")
 
@@ -33,24 +35,33 @@ async def coupon_step(
     - 透传: 不做 schema 校验,pricebot 自己校验
     - 失败: 网络不可达 / pricebot 5xx → 502 + 友好 message
     """
+    # 读原始字节,避免"反序列化→再序列化"的双重 JSON(省透传 CPU)。只 loads 一次拿
+    # trace_id 做亲和 + 打日志,转发时直接发原始 bytes,不重新 dumps。
+    raw = await request.body()
     try:
-        body = await request.json()
+        meta = json.loads(raw)
     except Exception as e:
         raise HTTPException(status_code=400, detail=f"invalid json body: {e}") from e
+    if not isinstance(meta, dict):
+        meta = {}
 
-    url = f"{settings.PRICEBOT_BASE_URL.rstrip('/')}/api/coupon/step"
+    # 按 trace_id 一致性 hash 选 pricebot 实例(同一领券任务的所有帧落同一进程)
+    base = pick_pricebot(meta.get("trace_id"))
+    url = f"{base.rstrip('/')}/api/coupon/step"
     timeout = settings.PRICEBOT_REQUEST_TIMEOUT_SEC
 
     logger.info(
         "coupon_step device_id=%s trace_id=%s step=%s",
-        body.get("device_id"),
-        body.get("trace_id"),
-        body.get("step"),
+        meta.get("device_id"),
+        meta.get("trace_id"),
+        meta.get("step"),
     )
 
     try:
         async with httpx.AsyncClient(timeout=timeout) as client:
-            resp = await client.post(url, json=body)
+            resp = await client.post(
+                url, content=raw, headers={"Content-Type": "application/json"}
+            )
     except httpx.RequestError as e:
         logger.error("[pricebot] request failed: %s", e)
         raise HTTPException(