diff --git a/pipeline_service/agent_loop_v2.py b/pipeline_service/agent_loop_v2.py index 0c8de8d..d2a56fb 100644 --- a/pipeline_service/agent_loop_v2.py +++ b/pipeline_service/agent_loop_v2.py @@ -875,7 +875,7 @@ class AgentExecutor: mins = int(getattr(rr, 'mins', 0) or 0) except (TypeError, ValueError): mins = 0 - if mins >= 10: + if mins >= 20: flag = f"⚠️心跳超时{mins}分钟(疑似僵尸,将被回收重跑)" else: flag = f"已运行{mins}分钟" diff --git a/pipeline_service/init.py b/pipeline_service/init.py index 73f7da1..8c81e8b 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -549,12 +549,12 @@ def load_pipeline_service(): while True: try: async with poll_db.sqlorContext("pipeline") as sor: - # 回收僵尸 running 任务：进程崩溃/协程挂起遗留（心跳超时 10 分钟未更新）。 - # 否则任务永久卡 running，后续任务链断裂。 + # 回收僵尸 running 任务：进程崩溃/协程挂起遗留（心跳超时未更新）。 + # 阈值 20 分钟 > LLM 单轮最坏时长（3 次 × 300s 重试 ≈ 15 分钟），避免误杀慢任务。 await sor.sqlExe( "UPDATE pipeline_tasks SET state='submitted', claimed_by=NULL, updated_at=NOW() " "WHERE state='running' AND pipeline_id='role_task' " - "AND updated_at < (NOW() - INTERVAL 10 MINUTE)", {}) + "AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {}) recs = await sor.sqlExe( "SELECT id, tenant_id, role FROM pipeline_tasks " "WHERE state='submitted' AND pipeline_id='role_task' " @@ -601,7 +601,7 @@ def load_pipeline_service(): await sor.sqlExe( "UPDATE pipeline_tasks SET claimed_by=NULL, updated_at=NOW() " "WHERE state='review' AND claimed_by IS NOT NULL " - "AND updated_at < (NOW() - INTERVAL 10 MINUTE)", {}) + "AND updated_at < (NOW() - INTERVAL 20 MINUTE)", {}) recs = await sor.sqlExe( "SELECT id, tenant_id, role FROM pipeline_tasks " "WHERE state='review' AND claimed_by IS NULL " diff --git a/pipeline_service/llm_bridge.py b/pipeline_service/llm_bridge.py index 46c80ce..6efc055 100644 --- a/pipeline_service/llm_bridge.py +++ b/pipeline_service/llm_bridge.py @@ -65,6 +65,48 @@ async def _get_model_config(model_name: str = None) -> dict: return {} +# 瞬时错误重试：超时/连接错误/限流/服务端5xx 均重试；配置错误(无key)、鉴权/参数4xx 不重试 +_RETRYABLE_STATUS = (429, 500, 502, 503, 504) +_LLM_MAX_ATTEMPTS = 3 +# 大上下文（角色 agent 多轮累积）生成偏慢，180s 曾触发超时（空 err=），放宽到 5 分钟； +# connect 单独设 30s，连接建立失败能快速失败并重试，而不是干等 5 分钟。 +_LLM_TOTAL_TIMEOUT = 300 +_LLM_CONNECT_TIMEOUT = 30 + + +async def _post_chat_completion(url: str, headers: dict, payload: dict) -> dict: + """POST /chat/completions，带瞬时错误重试。返回解析后的 JSON dict。""" + import aiohttp + import asyncio + + last_exc = None + for attempt in range(_LLM_MAX_ATTEMPTS): + try: + async with aiohttp.ClientSession() as session: + async with session.post( + url, headers=headers, json=payload, + timeout=aiohttp.ClientTimeout(total=_LLM_TOTAL_TIMEOUT, connect=_LLM_CONNECT_TIMEOUT), + ) as resp: + if resp.status != 200: + text = await resp.text() + err = ValueError("LLM API error %d: %s" % (resp.status, text[:300])) + if resp.status in _RETRYABLE_STATUS and attempt < _LLM_MAX_ATTEMPTS - 1: + last_exc = err + await asyncio.sleep(2 * (attempt + 1)) + continue + raise err + return await resp.json() + except (asyncio.TimeoutError, aiohttp.ClientError) as e: + last_exc = e + if attempt < _LLM_MAX_ATTEMPTS - 1: + logger.warning("llm_bridge: 瞬时错误重试 %d/%d: %s", attempt + 1, _LLM_MAX_ATTEMPTS, e) + await asyncio.sleep(2 * (attempt + 1)) + continue + raise + + raise last_exc if last_exc else ValueError("LLM call failed") + + async def llm_call(prompt: str, model: str = None, temperature: float = 0.7) -> str: """Call LLM and return text response. @@ -114,16 +156,8 @@ async def llm_call(prompt: str, model: str = None, temperature: float = 0.7) -> } url = api_base.rstrip("/") + "/chat/completions" - async with aiohttp.ClientSession() as session: - async with session.post( - url, headers=headers, json=payload, - timeout=aiohttp.ClientTimeout(total=120) - ) as resp: - if resp.status != 200: - text = await resp.text() - raise ValueError(f"LLM API error {resp.status}: {text[:300]}") - data = await resp.json() - return data["choices"][0]["message"]["content"] + data = await _post_chat_completion(url, headers, payload) + return data["choices"][0]["message"]["content"] async def call_llm(tenant_id: str, prompt: str, model: str = None, temperature: float = 0.7) -> str: @@ -152,14 +186,8 @@ async def llm_call_msgs(messages: list, model: str = None, temperature: float = payload = {"model": model_id, "messages": messages, "temperature": temperature} url = api_base.rstrip("/") + "/chat/completions" - async with aiohttp.ClientSession() as session: - async with session.post(url, headers=headers, json=payload, - timeout=aiohttp.ClientTimeout(total=180)) as resp: - if resp.status != 200: - text = await resp.text() - raise ValueError(f"LLM API error {resp.status}: {text[:300]}") - data = await resp.json() - return data["choices"][0]["message"]["content"] + data = await _post_chat_completion(url, headers, payload) + return data["choices"][0]["message"]["content"] async def llm_call_msgs_native(messages: list, tools: list = None, model: str = None, temperature: float = 0.7) -> dict: @@ -191,16 +219,10 @@ async def llm_call_msgs_native(messages: list, tools: list = None, model: str = payload["tool_choice"] = "auto" url = api_base.rstrip("/") + "/chat/completions" - async with aiohttp.ClientSession() as session: - async with session.post(url, headers=headers, json=payload, - timeout=aiohttp.ClientTimeout(total=180)) as resp: - if resp.status != 200: - text = await resp.text() - raise ValueError(f"LLM API error {resp.status}: {text[:300]}") - data = await resp.json() - msg = data["choices"][0]["message"] - return { - "content": msg.get("content") or "", - "tool_calls": msg.get("tool_calls") or [], - } + data = await _post_chat_completion(url, headers, payload) + msg = data["choices"][0]["message"] + return { + "content": msg.get("content") or "", + "tool_calls": msg.get("tool_calls") or [], + }