yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: LLM 调用加瞬时错误重试(3次) + 超时 180s→300s(connect 30s)；心跳回收阈值 10→20 分钟防误杀慢任务

Browse Source 
- llm_bridge: 抽出 _post_chat_completion，超时/连接错误/429/5xx 重试3次退避；total=300 connect=30
- init.py: 僵尸回收阈值对齐 LLM 最坏单轮时长(3×300s≈15min)，20分钟
- agent_loop_v2: diagnose 心跳超时判定 10→20 分钟
This commit is contained in:
ymq 2026-08-15 00:46:30 +08:00
parent 00a7e77629
commit 8c0d63efb1
3 changed files with 57 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pipeline_service/agent_loop_v2.py
View File

@ -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}分钟"

8
pipeline_service/init.py
View File

@ -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 "

82
pipeline_service/llm_bridge.py
View File

@ -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 [],
}