feat: 失败任务处理第二层——failed poller 自动重跑(最多3次)+报故障给用户
- pipeline_tasks 加 retry_count/last_error 列(启动时幂等 DDL) - role/pm agent 失败时记录 last_error(含异常类型) - handle_failed_task: 瞬时错误(超时/连接/5xx/429)重跑，永久错误或重试>=3次报故障(建问题通知用户，任务置 waiting) - init.py: 新增 failed poller 每60s扫描 failed 任务，role任务由pm上报、pm任务由cockpit上报
This commit is contained in:
ymq
parent
8c0d63efb1
commit
aa0dcc9c04
@ -661,7 +661,8 @@ async def role_agent_run(project_id, role, agent_id=None, model_name=None):
|
||||
try:
|
||||
resp = await llm_call_msgs_native(msgs, tools=tools_schema, model=model_name, temperature=0.4)
|
||||
except Exception as e:
|
||||
await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id})
|
||||
err_msg = f"{type(e).__name__}: {str(e)[:400]}"
|
||||
await sor.sqlExe("UPDATE pipeline_tasks SET state='failed', last_error=${e}$ WHERE id=${tid}$", {"e": err_msg, "tid": task_id})
|
||||
logger.error(f"role_agent_run llm failed: task={task_id} err={e}")
|
||||
return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
|
||||
|
||||
@ -842,7 +843,8 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
|
||||
try:
|
||||
raw = await llm_call_msgs(msgs, model=model_name, temperature=0.3)
|
||||
except Exception as e:
|
||||
await sor.sqlExe("UPDATE pipeline_tasks SET state='failed' WHERE id=${tid}$", {"tid": task_id})
|
||||
err_msg = f"{type(e).__name__}: {str(e)[:400]}"
|
||||
await sor.sqlExe("UPDATE pipeline_tasks SET state='failed', last_error=${e}$ WHERE id=${tid}$", {"e": err_msg, "tid": task_id})
|
||||
return {"status": "failed", "task_id": task_id, "error": str(e)[:200]}
|
||||
|
||||
act = _parse_agent_action(raw)
|
||||
@ -903,6 +905,78 @@ async def pm_review_run(project_id, agent_id=None, model_name=None):
|
||||
return {"status": "completed", "task_id": task_id, "comment": comment or "项目完成"}
|
||||
|
||||
|
||||
# ── 失败任务处理（第二层：PM/cockpit 判定重跑，最多3次，超限报故障给用户）──
|
||||
|
||||
_RETRY_HINTS = (
|
||||
"timeout", "timed out", "connection", "connect", "network", "unreachable",
|
||||
"rate limit", "429", "500", "502", "503", "504", "reset", "broken pipe",
|
||||
"eof", "temporar", "busy", "overloaded",
|
||||
)
|
||||
|
||||
|
||||
def _classify_failure(last_error: str) -> str:
|
||||
"""按失败原因分类：retry(瞬时，可重跑) / fault(永久，报故障)。
|
||||
|
||||
空错误通常是 asyncio.TimeoutError（str 为空），视为瞬时可重跑。
|
||||
"""
|
||||
err = (last_error or "").strip().lower()
|
||||
if not err:
|
||||
return "retry"
|
||||
for h in _RETRY_HINTS:
|
||||
if h in err:
|
||||
return "retry"
|
||||
return "fault"
|
||||
|
||||
|
||||
async def handle_failed_task(task_id: str, project_id: str) -> dict:
|
||||
"""处理一个失败任务：判定重跑或报故障。由 failed poller 周期调用。
|
||||
|
||||
- retry_count < 3 且失败原因判定为瞬时 → 重跑(retry_count+1, 回 submitted)
|
||||
- retry_count >= 3 或判定为永久错误 → 报故障(建问题通知用户，任务置 waiting)
|
||||
"""
|
||||
db = _get_db()
|
||||
async with db.sqlorContext("pipeline") as sor:
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT id, title, role, retry_count, last_error FROM pipeline_tasks "
|
||||
"WHERE id=${tid}$ AND state='failed'",
|
||||
{"tid": task_id})
|
||||
if not recs:
|
||||
return {"status": "idle", "task_id": task_id}
|
||||
t = recs[0]
|
||||
title = getattr(t, "title", "") or ""
|
||||
role = getattr(t, "role", "") or ""
|
||||
try:
|
||||
retry_count = int(getattr(t, "retry_count", 0) or 0)
|
||||
except (TypeError, ValueError):
|
||||
retry_count = 0
|
||||
last_error = getattr(t, "last_error", "") or ""
|
||||
|
||||
# 已重试 3 次仍未成功 → 报故障
|
||||
if retry_count >= 3:
|
||||
decision = "fault"
|
||||
else:
|
||||
decision = _classify_failure(last_error)
|
||||
|
||||
if decision == "retry":
|
||||
await sor.sqlExe(
|
||||
"UPDATE pipeline_tasks SET retry_count=retry_count+1, state='submitted', "
|
||||
"claimed_by=NULL, last_error=NULL, updated_at=NOW() "
|
||||
"WHERE id=${tid}$ AND state='failed'",
|
||||
{"tid": task_id})
|
||||
logger.info("failed task retry: task=%s role=%s attempt=%d", task_id, role, retry_count + 1)
|
||||
return {"status": "retry", "task_id": task_id, "attempt": retry_count + 1}
|
||||
|
||||
# 报故障：建问题通知用户，任务置 waiting 等待人工介入
|
||||
from .questions import agent_ask
|
||||
reporter = "cockpit" if role == "pm" else "pm"
|
||||
qid = await agent_ask(
|
||||
project_id, task_id, reporter,
|
||||
f"任务「{title}」已失败 {retry_count} 次，自动重试无法解决，请人工介入。失败原因：{last_error or '未知'}",
|
||||
context={"fault": True, "last_error": last_error, "retry_count": retry_count})
|
||||
logger.info("failed task fault: task=%s role=%s qid=%s", task_id, role, qid)
|
||||
return {"status": "fault", "task_id": task_id, "question_id": qid}
|
||||
|
||||
|
||||
async def role_agent_loop(project_id, role, agent_id=None, model_name=None, max_iterations=10):
|
||||
results = []
|
||||
for _ in range(max_iterations):
|
||||
|
||||
@ -38,7 +38,7 @@ from .questions import (
|
||||
get_question, list_questions, get_task_qna,
|
||||
)
|
||||
from .agent_loop import role_agent_run, role_agent_loop, agent_loop, run_agent_loop
|
||||
from .agent_loop import pm_review_run, pm_review_loop
|
||||
from .agent_loop import pm_review_run, pm_review_loop, handle_failed_task
|
||||
|
||||
MODULE_NAME = "pipeline_service"
|
||||
MODULE_VERSION = "3.4.0"
|
||||
@ -632,6 +632,46 @@ def load_pipeline_service():
|
||||
|
||||
add_startup(_pm_poller)
|
||||
|
||||
# Failed task poller: 失败任务自动重跑(最多3次)，超限报故障给用户
|
||||
async def _failed_poller(app):
|
||||
from sqlor.dbpools import DBPools as _DBP3
|
||||
fd_db = _DBP3()
|
||||
_fd_dispatched = set()
|
||||
|
||||
async def _fd_poll_loop():
|
||||
while True:
|
||||
try:
|
||||
async with fd_db.sqlorContext("pipeline") as sor:
|
||||
recs = await sor.sqlExe(
|
||||
"SELECT id, tenant_id FROM pipeline_tasks "
|
||||
"WHERE state='failed' AND pipeline_id='role_task' "
|
||||
"ORDER BY created_at ASC LIMIT 10",
|
||||
{})
|
||||
for rec in (recs or []):
|
||||
tid = getattr(rec, 'id', '')
|
||||
pid = getattr(rec, 'tenant_id', '')
|
||||
if tid and pid and tid not in _fd_dispatched:
|
||||
_fd_dispatched.add(tid)
|
||||
debug(f"failed_poller handling task={tid}")
|
||||
|
||||
async def _fd_dispatch(tid, pid):
|
||||
try:
|
||||
await handle_failed_task(tid, pid)
|
||||
except Exception as e:
|
||||
debug(f"failed_poller dispatch error: task={tid} err={e}")
|
||||
finally:
|
||||
_fd_dispatched.discard(tid)
|
||||
|
||||
asyncio.ensure_future(_fd_dispatch(tid, pid))
|
||||
except Exception as e:
|
||||
debug(f"failed_poller error: {e}")
|
||||
await asyncio.sleep(60)
|
||||
|
||||
asyncio.create_task(_fd_poll_loop())
|
||||
debug("failed poller started")
|
||||
|
||||
add_startup(_failed_poller)
|
||||
|
||||
# v2 DDL auto-init: create tables and columns at startup
|
||||
async def _init_v2_tables(app):
|
||||
from sqlor.dbpools import DBPools as _DBP2
|
||||
@ -686,6 +726,17 @@ def load_pipeline_service():
|
||||
_debug("sd_org_settings.agent_config column added")
|
||||
except Exception:
|
||||
pass
|
||||
# failed 任务重试计数 + 失败原因
|
||||
try:
|
||||
await sor.sqlExe("ALTER TABLE pipeline_tasks ADD COLUMN retry_count int NOT NULL DEFAULT 0", {})
|
||||
_debug("pipeline_tasks.retry_count column added")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
await sor.sqlExe("ALTER TABLE pipeline_tasks ADD COLUMN last_error text", {})
|
||||
_debug("pipeline_tasks.last_error column added")
|
||||
except Exception:
|
||||
pass
|
||||
except Exception as e:
|
||||
_debug(f"v2 DDL init: {e}")
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user