diff --git a/pipeline_opportunity/opp_normalize.py b/pipeline_opportunity/opp_normalize.py index 963a096..0c8739c 100644 --- a/pipeline_opportunity/opp_normalize.py +++ b/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:]