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

配合 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>
2026-06-07 02:29:17 +08:00
6 changed files with 144 additions and 91 deletions
@@ -41,12 +41,18 @@ MT_CPS_APP_SECRET=
 # 默认渠道追踪标识(sid),用于区分不同 app 的 CPS 数据
 MT_CPS_DEFAULT_SID=sgbjia

-# ===== Pricebot 上游 (领券业务透传目标) =====
-# 客户端调本服务的 /api/v1/coupon/step,我们透传到 pricebot-backend 的 /api/coupon/step。
+# ===== Pricebot 上游 (领券/比价业务透传目标) =====
+# 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。
 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。
 PRICEBOT_BASE_URL=http://localhost:8000
+# 【多实例】单机多进程部署时,填逗号分隔的实例列表(端口与 pricebot 集群对齐),透传层按
+# trace_id 一致性 hash 选实例 → 同一比价所有帧落同一进程(进程内维护 state,无需 Redis)。
+# 留空 = 单实例(用上面的 PRICEBOT_BASE_URL)。详见 pricebot-backend/docs/并发部署设计.md
+# PRICEBOT_INSTANCES=http://127.0.0.1:8001,http://127.0.0.1:8002,http://127.0.0.1:8003,http://127.0.0.1:8004,http://127.0.0.1:8005,http://127.0.0.1:8006
 # 领券单帧最多 wait 6s,加网络往返,30s 兜底
 PRICEBOT_REQUEST_TIMEOUT_SEC=30
+# 比价(intent/recognize + price/step)透传超时:大上下文 LLM + 逐帧 LLM,给 60s
+PRICEBOT_COMPARE_TIMEOUT_SEC=60

 # ===== CORS =====
 # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类
@@ -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(
@@ -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(
@@ -135,15 +135,28 @@ class Settings(BaseSettings):
        """回调开关打开且验签密钥已配,才接受发奖回调。"""
        return bool(self.PANGLE_CALLBACK_ENABLED and self.PANGLE_REWARD_SECRET)

-    # ===== Pricebot 上游 (领券业务透传目标) =====
+    # ===== Pricebot 上游 (领券/比价业务透传目标) =====
    # pricebot-backend 默认跑在 8000。/api/v1/coupon/step 会透传到这里的 /api/coupon/step
    PRICEBOT_BASE_URL: str = "http://localhost:8000"
+    # 多实例(单机多进程)时填逗号分隔的实例列表,例如:
+    #   PRICEBOT_INSTANCES=http://127.0.0.1:8001,http://127.0.0.1:8002,http://127.0.0.1:8003
+    # 透传层按 trace_id 一致性 hash 选实例(见 app/core/pricebot_router.py),保证同一次
+    # 比价/领券的所有帧落同一 pricebot 进程,从而用进程内内存维护 session/coordinator,
+    # 无需 Redis。留空 → 退回单实例 [PRICEBOT_BASE_URL],零改动兼容。
+    PRICEBOT_INSTANCES: str = ""
    # 领券一帧最多 wait 6s,加网络往返,timeout 给 30s 比较稳
    PRICEBOT_REQUEST_TIMEOUT_SEC: int = 30
    # 比价(intent/recognize + price/step)透传超时:意图识别是大上下文 LLM、
    # price/step 每帧也是 LLM,可能 >30s;对齐客户端 agent ApiClient 的 60s 读超时。
    PRICEBOT_COMPARE_TIMEOUT_SEC: int = 60

+    @property
+    def pricebot_instances(self) -> list[str]:
+        """pricebot 上游实例列表。空 → 单实例兜底 [PRICEBOT_BASE_URL]。"""
+        if not self.PRICEBOT_INSTANCES.strip():
+            return [self.PRICEBOT_BASE_URL]
+        return [u.strip() for u in self.PRICEBOT_INSTANCES.split(",") if u.strip()]
+
    # ===== 媒体文件(用户头像上传)=====
    # 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。
    # 生产可改由 nginx 直接 serve MEDIA_ROOT,绕过应用进程。
@@ -1,91 +1,20 @@
 """统一日志配置。

-业务代码用 `logger = logging.getLogger("shagua.xxx")` 即可, 本模块在 main.py 启动时调一次。
-
- 控制台(stdout): 人类可读文本, 给 systemd / 本地看。
- 文件 `logs/app-server.log`: 单行 JSON, 供阿里云 SLS/Logtail 采集(JSON 模式零正则);
-  异常栈作为字段内嵌不换行 → 每条日志一行。
- 环境变量:
-  - LOG_JSON_CONSOLE=1   控制台也输出 JSON
-  - LOG_DIR / LOG_FILE   改落盘路径(默认 logs/app-server.log)
-    ⚠️ admin 子进程(app.admin.main, 独立进程)应设不同 LOG_FILE, 避免与主进程争抢同一轮转文件
-  - LOG_SERVICE_NAME     JSON 里的 service 字段(默认 app-server)
+业务代码用 `logger = logging.getLogger("shagua.xxx")` 即可,本模块在 main.py 启动时调一次。
 """
 from __future__ import annotations

-import json
 import logging
-import os
 import sys
-from datetime import datetime
-from logging.handlers import RotatingFileHandler
-from pathlib import Path
-
-
-class JsonFormatter(logging.Formatter):
-    """把 LogRecord 序列化成单行 JSON(SLS/Logtail 友好)。异常栈内嵌为字段, 整条仍是一行。"""
-    def __init__(self, service: str = "app-server"):
-        super().__init__()
-        self.service = service
-
-    def format(self, record: logging.LogRecord) -> str:
-        dt = datetime.fromtimestamp(record.created)
-        data = {
-            "time": dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{int(record.msecs):03d}",
-            "level": record.levelname,
-            "service": self.service,
-            "logger": record.name,
-            "func": record.funcName,
-            "line": record.lineno,
-            "message": record.getMessage(),
-        }
-        if record.exc_info:
-            data["exception"] = self.formatException(record.exc_info)
-        if record.stack_info:
-            data["stack"] = self.formatStack(record.stack_info)
-        return json.dumps(data, ensure_ascii=False, default=str)
-
-
-_CONFIGURED = False


 def setup_logging(debug: bool = False) -> None:
-    global _CONFIGURED
-    if _CONFIGURED:
-        return
-
    level = logging.DEBUG if debug else logging.INFO
-    service = os.getenv("LOG_SERVICE_NAME", "app-server")
-
-    root = logging.getLogger()
-    root.setLevel(level)
-    # 清掉 basicConfig/uvicorn 可能预置的 root handler, 避免重复输出
-    for h in list(root.handlers):
-        root.removeHandler(h)
-
-    # 控制台: 默认文本(systemd/本地看); LOG_JSON_CONSOLE=1 时输出 JSON
-    console = logging.StreamHandler(sys.stdout)
-    if os.getenv("LOG_JSON_CONSOLE") == "1":
-        console.setFormatter(JsonFormatter(service))
-    else:
-        console.setFormatter(
-            logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")
-        )
-    root.addHandler(console)
-
-    # 文件: 单行 JSON, 供 Logtail 采集(自动轮转, 单文件 10MB, 保留 5 个)
-    log_file = os.getenv("LOG_FILE") or str(
-        Path(os.getenv("LOG_DIR", "logs")) / "app-server.log"
+    logging.basicConfig(
+        level=level,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+        stream=sys.stdout,
    )
-    Path(log_file).parent.mkdir(parents=True, exist_ok=True)
-    file_handler = RotatingFileHandler(
-        log_file, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8",
-    )
-    file_handler.setFormatter(JsonFormatter(service))
-    root.addHandler(file_handler)
-
    # 第三方库降噪
    logging.getLogger("httpx").setLevel(logging.WARNING)
    logging.getLogger("httpcore").setLevel(logging.WARNING)
-
-    _CONFIGURED = True
@@ -0,0 +1,82 @@
+"""pricebot 上游实例选择：按 trace_id 一致性 hash(ketama 风格)选实例。
+
+单机多进程下,保证同一 trace_id(一次比价/领券的所有帧)落同一 pricebot 进程,
+从而用进程内内存维护 session/coordinator,不需要 Redis。
+
+为什么用一致性 hash 而非简单取模(crc32 % N):
+  加/减实例时,取模会让几乎所有 trace 重新映射(扩缩容全量中断进行中的比价);
+  一致性 hash(虚拟节点)只重映射约 1/N 的 trace,扩缩容对存量冲击最小。
+  ⚠️ 但内存态下,被重映射的那 1/N trace 仍会丢失 session(状态没外置),
+     所以扩缩容仍建议挑低峰。
+
+hash 用 md5(纯函数),app-server 多 worker / 多实例算出的环一致,亲和不会因
+app-server 自身水平扩展而被破坏。
+"""
+from __future__ import annotations
+
+import bisect
+import hashlib
+from typing import Dict, List, Optional
+
+from app.core.config import settings
+
+# 每个真实实例在 hash 环上的虚拟节点数。越多分布越均匀。实例数少时(6~10)需要更多
+# 虚拟节点才均匀:实测 150→负载偏差~16%、1000→~6%。环只在启动/实例变更时构建一次,
+# 1000×N 个点的构建与 bisect 查找开销都可忽略。
+_VNODES_PER_NODE = 1000
+
+
+def _hash(s: str) -> int:
+    """取 md5 前 8 hex(32-bit)做环坐标。确定性、跨进程一致。"""
+    return int(hashlib.md5(s.encode("utf-8")).hexdigest()[:8], 16)
+
+
+class _HashRing:
+    """ketama 风格一致性 hash 环。不可变,实例列表变化时整体重建(见 _get_ring)。"""
+
+    def __init__(self, nodes: List[str]):
+        self._keys: List[int] = []                # 升序的环坐标
+        self._key_to_node: Dict[int, str] = {}
+        for node in nodes:
+            for v in range(_VNODES_PER_NODE):
+                h = _hash(f"{node}#{v}")
+                # 极小概率撞坐标,撞了跳过该虚拟节点(不影响正确性,仅少一个 vnode)
+                if h not in self._key_to_node:
+                    self._key_to_node[h] = node
+                    self._keys.append(h)
+        self._keys.sort()
+
+    def pick(self, key: str) -> str:
+        h = _hash(key)
+        idx = bisect.bisect(self._keys, h)
+        if idx == len(self._keys):
+            idx = 0                                # 环回绕
+        return self._key_to_node[self._keys[idx]]
+
+
+# 按实例列表缓存环,列表变了(改 PRICEBOT_INSTANCES + 重启)才重建。
+_ring: Optional[_HashRing] = None
+_ring_nodes: tuple = ()
+
+
+def _get_ring(nodes: List[str]) -> _HashRing:
+    global _ring, _ring_nodes
+    key = tuple(nodes)
+    if _ring is None or key != _ring_nodes:
+        _ring = _HashRing(nodes)
+        _ring_nodes = key
+    return _ring
+
+
+def pick_pricebot(trace_id: Optional[str]) -> str:
+    """按 trace_id 选 pricebot 实例 base url。
+
+    - 实例列表来自 settings.pricebot_instances(空则单实例兜底 [PRICEBOT_BASE_URL])
+    - trace_id 缺失/空 或 单实例 → 直接返回第一个,不进环
+    """
+    nodes = settings.pricebot_instances
+    if len(nodes) == 1 or not trace_id:
+        return nodes[0]
+    # str() 防御:协议保证 trace_id 是 str,但万一传入非 str(如 int)也不至于在
+    # _hash 的 .encode() 处炸,确定性地选到实例。
+    return _get_ring(nodes).pick(str(trace_id))