fix(normalize): LLM客户端硬超时——llm_call的_timeout只透传治理端点,客户端aiohttp无超时致上游挂起无限等(实测atomize卡6分钟atoms=0);_llm包装asyncio.wait_for(120s)+指数退避重试2次,三处调用统一

pipeline_opportunity/opp_normalize.py

@@ -23,6 +23,30 @@ from .opp_mining import _cos, _centroid, demand_vec_id
 
 logger = logging.getLogger("pipeline.opp_normalize")
 
+# 客户端硬超时（秒）：llm_call 的 _timeout 只透传给治理端点，客户端 aiohttp 无超时——
+# 上游挂起时客户端会无限等（实测 atomize 卡 6 分钟 atoms=0）。wait_for 快速失败+重试。
+LLM_CLIENT_TIMEOUT = 120
+LLM_RETRIES = 2
+
+
+async def _llm(prompt, ctx, timeout=LLM_CLIENT_TIMEOUT):
+    """llm_call 客户端超时包装：wait_for + 指数退避重试，全失败抛 RuntimeError。"""
+    from pipeline_service.llm_bridge import llm_call
+    last = None
+    for a in range(LLM_RETRIES):
+        try:
+            return await asyncio.wait_for(
+                llm_call(prompt, purpose="utility", timeout=timeout,
+                         org_id=ctx.get("org_id") or "0",
+                         user_id=ctx.get("user_id") or "",
+                         session_id=ctx.get("session_id") or ""),
+                timeout=timeout + 10)
+        except Exception as e:
+            last = e
+            if a < LLM_RETRIES - 1:
+                await asyncio.sleep(2 ** a)
+    raise RuntimeError("LLM 调用失败(%d次): %s" % (LLM_RETRIES, str(last)[:150]))
+
 P = {
     "opp_norm_atom_batch": "20",     # 原子化每批文档数
     "opp_norm_tau2": "0.85",         # 归一化相似度触发阈值
@@ -95,10 +119,7 @@ async def _atomize(sor, snaps, cfg, ctx):
             "不要解释、不要 markdown 代码块。")
         async with sem:
             try:
-                raw = await llm_call(prompt, purpose="utility", timeout=90,
-                                     org_id=ctx.get("org_id") or "0",
-                                     user_id=ctx.get("user_id") or "",
-                                     session_id=ctx.get("session_id") or "")
+                raw = await _llm(prompt, ctx, timeout=90)
                 raw = (raw or "").strip()
                 if raw.startswith("```"):
                     raw = raw.strip("`")
@@ -195,10 +216,7 @@ async def _normalize(sor, atoms, project_id, cfg, ctx):
                 "（编号从1起，对应上面列表顺序）。只输出 JSON。")
             async with sem:
                 try:
-                    raw = await llm_call(prompt, purpose="utility", timeout=90,
-                                         org_id=ctx.get("org_id") or "0",
-                                         user_id=ctx.get("user_id") or "",
-                                         session_id=ctx.get("session_id") or "")
+                    raw = await _llm(prompt, ctx, timeout=90)
                     raw = (raw or "").strip().strip("`")
                     if raw.lower().startswith("json"):
                         raw = raw[4:]
@@ -272,10 +290,7 @@ async def _domains(sor, cluster_id, org_id, std_rows, ctx):
         "（如「人员管理」「工时与考勤」「合同管理」「报表统计」）。\n" + names +
         "\n\n只输出 JSON 数组 [{\"domain\": \"域名\", \"nos\": [编号...]}]，覆盖全部编号，不解释。")
     try:
-        raw = await llm_call(prompt, purpose="utility", timeout=90,
-                             org_id=ctx.get("org_id") or "0",
-                             user_id=ctx.get("user_id") or "",
-                             session_id=ctx.get("session_id") or "")
+        raw = await _llm(prompt, ctx, timeout=90)
         raw = (raw or "").strip().strip("`")
         if raw.lower().startswith("json"):
             raw = raw[4:]