mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:31:58 +00:00
驳回不重试、不被「出错时继续」吞掉:人的决定照总方案 X9 收成被驳回
为什么:NodeRejected 是 NodeError 的子类,compiler._execute 的通用 except 照常接住它。节点配了失败重试的,驳回之后 会重试这个节点——再开一次待办、把同一件事再问一遍人;配了 on_error=continue 的,驳回被当成节点失败吞掉,图接着往下 走,最后以 succeeded 结束,通知、定时推送、发布单都看不到这次驳回。这和 X9 定下的「驳回 = cancelled + end_reason= rejected」直接矛盾。「失败重试」「出错时继续」是给故障准备的,驳回不是故障。 做了什么: - _execute 在通用 except 之前单独接住 NodeRejected:不重试,直接跳出;容错分支多一条「不是驳回」才走。节点这一步 照旧记一条 node.failed(实例进度、时间线、存量判断都按它收尾这一步),再原样交给 runner 收成被驳回。 - NodeRejected 的说明改成现在的样子(以前写的是 on_error=continue 照旧认它)。 原有行为变了:配了重试或出错时继续的关卡,驳回后不再重问、不再接着走。发布说明里要写明。 测试:test_run_rejected 加五种关卡 ×(出错时继续、失败重试)的参数化,断言终态是 cancelled + rejected、没有重试日志、 驳回之后没往下走、节点只有一条 node.failed。先跑:出错时继续的五条以 succeeded 结束,失败重试的五条又停回 interrupted (重问了人),10 条失败;改完通过。
This commit is contained in:
@@ -10,7 +10,7 @@ from langgraph.graph import END, START, StateGraph
|
||||
|
||||
from app.core.errors import describe_exception, raw_detail
|
||||
from app.core.events import EventType
|
||||
from app.engine.context import NodeContext, NodeError, RunContext
|
||||
from app.engine.context import NodeContext, NodeError, NodeRejected, RunContext
|
||||
from app.engine.expressions import open_template_scope, template_scope
|
||||
from app.engine.nodes import (
|
||||
challenge, control, dual_check, human, io, knowledge, llm, merge, metrics, multi, report, tools,
|
||||
@@ -144,6 +144,13 @@ def _wrap(
|
||||
raise
|
||||
except asyncio.CancelledError:
|
||||
raise
|
||||
except NodeRejected as e:
|
||||
# 人的决定(驳回、超时驳回),不是故障(总方案 X9):「失败重试」「出错时继续」都是给故障准备的。重试只会把
|
||||
# 同一件事再问一遍人;on_error=continue 会把驳回吞掉接着往下走,最后以 succeeded 结束,通知、定时推送、
|
||||
# 发布单都看不到这次驳回。所以不重试、不被容错接住;节点这一步照旧记一条 node.failed(时间线、实例进度
|
||||
# 按它收尾这一步),交给 runner 收成 cancelled + end_reason=rejected
|
||||
last_error = e
|
||||
break
|
||||
except Exception as e: # noqa: BLE001
|
||||
# NodeError 以前在这里直接 raise,于是节点自己抛的错既不重试、
|
||||
# 也不发 node.failed、更不被 on_error=continue 接住。而 NodeError
|
||||
@@ -168,8 +175,8 @@ def _wrap(
|
||||
ctx.emit(EventType.NODE_FAILED, error=message, duration_ms=elapsed, node_id=node.id,
|
||||
detail=_detail(last_error))
|
||||
|
||||
# 容错模式:记下错误继续往下走,而不是让整张图挂掉
|
||||
if node.config.get("on_error") == "continue":
|
||||
# 容错模式:记下错误继续往下走,而不是让整张图挂掉。驳回不算错误,不走这里(见上)
|
||||
if node.config.get("on_error") == "continue" and not isinstance(last_error, NodeRejected):
|
||||
return {
|
||||
"nodes": {node.id: {"error": message, "failed": True}},
|
||||
"trail": [{"node_id": node.id, "error": message, "ts": time.time()}],
|
||||
|
||||
@@ -119,7 +119,8 @@ class NodeRejected(NodeError):
|
||||
驳回过的试运行说成「试运行失败了」,记录页还给「继续运行」——接着跑只会读到同一个驳回答复、再停一次。现在 runner
|
||||
单独接住它,收成 cancelled + end_reason=rejected,run.cancelled 带上在哪一步、谁、结论和备注,消费方都读这一个字段。
|
||||
|
||||
是 NodeError 的子类:节点失败的那一套(node.failed、on_error=continue)照旧认它;原话(str)照旧写给人看,
|
||||
是 NodeError 的子类:节点这一步照旧记 node.failed(时间线、实例进度按它收尾);但不重试、不被 on_error=continue
|
||||
接住(engine/compiler._execute:那两项是给故障准备的,驳回被吞掉运行就以 succeeded 结束了)。原话(str)照旧写给人看,
|
||||
Run.error 和以前一字不差。kind 见 REJECTION_KINDS;decision 是 reject 或 expired_reject(同待办的结论)。"""
|
||||
|
||||
def __init__(self, node_id: str, message: str, *, kind: str, decision: str = "reject",
|
||||
|
||||
@@ -212,3 +212,28 @@ async def test_a_run_still_queued_at_restart_gets_its_run_failed_event():
|
||||
assert row.status == "failed" and row.started_at is None and row.finished_at is None
|
||||
[failed] = await _events(run_id, "run.failed")
|
||||
assert failed.seq == 3 and failed.data == {"error": row.error}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kind", list(GATES))
|
||||
@pytest.mark.parametrize("tolerance", [{"on_error": "continue"}, {"retries": 1, "retry_backoff": 0.01}])
|
||||
async def test_a_rejection_is_neither_retried_nor_swallowed_by_on_error(kind, tolerance):
|
||||
"""「失败重试」「出错时继续」是给故障准备的。驳回是人的决定:重试只会把同一件事再问一遍人,on_error=continue 会把
|
||||
驳回吞掉、接着往下走、最后以 succeeded 结束——通知、定时推送、发布单都看不到这次驳回(总方案 X9)。"""
|
||||
gate, said = GATES[kind]
|
||||
gate = {**gate, "data": {**gate["data"], "config": {**gate["data"]["config"], **tolerance}}}
|
||||
run = await run_manager.start(graph=chain(START, gate, OUT), input_payload={"q": "x"})
|
||||
await _wait(run.id, ("interrupted",))
|
||||
await run_manager.wait_idle(run.id)
|
||||
await run_manager.resume(run.id, {"approved": False, "note": "不对"}, actor="张工")
|
||||
final = await _wait(run.id, ("cancelled", "failed", "succeeded", "interrupted"))
|
||||
if final.status == "interrupted":
|
||||
# 重试会再问一遍人:又停在同一个关卡上
|
||||
await run_manager.wait_idle(run.id)
|
||||
final = await _wait(run.id, ("cancelled", "failed", "succeeded", "interrupted"))
|
||||
assert (final.status, final.end_reason) == ("cancelled", "rejected"), (final.status, final.error)
|
||||
assert final.error == said and final.error_node_id == "gate"
|
||||
events = await _events(run.id)
|
||||
assert not [e for e in events if e.type == "log" and (e.data or {}).get("code") == "node_retry"], "驳回不该重试"
|
||||
assert not [e for e in events if e.type == "node.finished" and e.node_id == "out"], "驳回之后不该接着往下走"
|
||||
# 节点这一步照旧留一条 node.failed(实例进度、时间线按它收尾这一步),只有一条
|
||||
assert len([e for e in events if e.type == "node.failed" and e.node_id == "gate"]) == 1
|
||||
|
||||
Reference in New Issue
Block a user