From eba62858415c8c5448f993aacb47613c5ef99aa2 Mon Sep 17 00:00:00 2001 From: yumoqing Date: Sat, 12 Sep 2026 11:53:24 +0800 Subject: [PATCH] =?UTF-8?q?fix(normalize):=20LLM=E5=AE=A2=E6=88=B7?= =?UTF-8?q?=E7=AB=AF=E7=A1=AC=E8=B6=85=E6=97=B6=E2=80=94=E2=80=94llm=5Fcal?= =?UTF-8?q?l=E7=9A=84=5Ftimeout=E5=8F=AA=E9=80=8F=E4=BC=A0=E6=B2=BB?= =?UTF-8?q?=E7=90=86=E7=AB=AF=E7=82=B9,=E5=AE=A2=E6=88=B7=E7=AB=AFaiohttp?= =?UTF-8?q?=E6=97=A0=E8=B6=85=E6=97=B6=E8=87=B4=E4=B8=8A=E6=B8=B8=E6=8C=82?= =?UTF-8?q?=E8=B5=B7=E6=97=A0=E9=99=90=E7=AD=89(=E5=AE=9E=E6=B5=8Batomize?= =?UTF-8?q?=E5=8D=A16=E5=88=86=E9=92=9Fatoms=3D0);=5Fllm=E5=8C=85=E8=A3=85?= =?UTF-8?q?asyncio.wait=5Ffor(120s)+=E6=8C=87=E6=95=B0=E9=80=80=E9=81=BF?= =?UTF-8?q?=E9=87=8D=E8=AF=952=E6=AC=A1,=E4=B8=89=E5=A4=84=E8=B0=83?= =?UTF-8?q?=E7=94=A8=E7=BB=9F=E4=B8=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pipeline_opportunity/opp_normalize.py | 39 ++++++++++++++++++--------- 1 file changed, 27 insertions(+), 12 deletions(-) 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:]