From b0575a7d1b1cd908dd053e581bc6cdf96dc9b3db Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 14 Sep 2026 15:35:47 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BC=95=E6=93=8E=E5=9B=9B=E5=A4=84=EF=BC=9A?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E5=A4=84=E7=90=86=E3=80=81=E6=81=A2=E5=A4=8D?= =?UTF-8?q?=E8=AF=AD=E4=B9=89=E3=80=81=E5=B9=82=E8=BF=90=E7=AE=97=E7=82=B8?= =?UTF-8?q?=E5=BC=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 一条分支条件能把后端冻死 求值器放行 ast.Pow，唯一护栏是 AST 节点数 <= 200。`9**9**9**9` 只有 11 个节点， 却让 CPython 无限期算大整数。而求值是在事件循环里同步执行的（分支条件、 transform、口径卡、每个节点的 skip_if），没有线程池隔离也没有超时，asyncio 的 cancel 打断不了同步 CPU。给底数和指数加上界，超界报 ExpressionError。 2**10、2**0.5、10**-2 这些正常用法不受影响。 ## NodeError 绕过 retries / on_error / node.failed `except NodeError: raise` 排在重试与兜底之前，于是节点自己抛的错既不重试、 也不发 node.failed、更不被 on_error=continue 接住——而 NodeError 恰恰是所有 执行器表达失败的标准方式（缺必填输入、模型没配、口径卡算不出…）。同一张图 配着同样的 retries=1 + on_error=continue，1/0 会重试并继续，而 JSON 模板非法 直接整个 run 失败。两条路径现在行为一致，node.failed 也带上 node_id （前端靠它把出错节点从"转圈"收敛成"失败"）。 ## 没有 checkpoint 的运行被"恢复"成一份假结果 重启扫描把 running 和 queued 一律标成 interrupted，而 resume 不检查有没有 checkpoint 就发 Command。queued 的运行从没写过 checkpoint，LangGraph 会拿空 state 从 START 重跑整张图——run.input 完全没注入，用字段默认值算完还落成 succeeded。现在 queued 直接标 failed（没有进度可恢复，请重新发起），resume 先 aget_state 确认有断点，没有就 409 并说清为什么。 ## 并行人工节点：一次 resume 打爆整个 run 同一 superstep 有多个 interrupt 时，resume 仍只发标量 Command，LangGraph 直接 抛 RuntimeError；而所有 pending 审批已经被写成同一个回复。Approval.interrupt_id 和 ResumeIn.approval_id 这两个字段一直存在却从没被用过。现在按 interrupt_id 逐条收集：多条待办时必须指定 approval_id，回复的那条转 answered 继续等， 凑齐引擎在等的全部 interrupt 才发 Command(resume={iid: value})。 验证：两个并行 human 节点——不指定 approval_id 被 409 拒、回复第一条后仍 interrupted 且剩 1 条 pending、凑齐后 succeeded。单人审批与 decide 通道回归通过。 --- backend/app/api/runs.py | 18 ++++- backend/app/engine/compiler.py | 26 ++++-- backend/app/engine/expressions.py | 23 +++++- backend/app/engine/runner.py | 127 ++++++++++++++++++++++++++---- 4 files changed, 170 insertions(+), 24 deletions(-) diff --git a/backend/app/api/runs.py b/backend/app/api/runs.py index bf94bebc..2c36cd90 100644 --- a/backend/app/api/runs.py +++ b/backend/app/api/runs.py @@ -229,9 +229,17 @@ class ResumeIn(BaseModel): @router.post("/{run_id}/resume", response_model=RunOut) -async def resume_run(run_id: str, payload: ResumeIn) -> Run: +async def resume_run( + run_id: str, payload: ResumeIn, x_actor: str | None = Header(default=None) +) -> Run: try: - return await run_manager.resume(run_id, payload.response) + # approval_id 以前定义了却从没传下去——一次运行里有多条待审批时， + # 引擎无从知道这次回复的是哪一条 + return await run_manager.resume( + run_id, payload.response, + approval_id=payload.approval_id, + actor=(x_actor or "").strip() or None, + ) except KeyError as e: raise HTTPException(404, str(e)) from e except ValueError as e: @@ -443,7 +451,11 @@ async def decide( response["args"] = payload.args try: - return await run_manager.resume(approval.run_id, response) + return await run_manager.resume( + approval.run_id, response, + approval_id=approval.id, + actor=approval.resolved_by, + ) except ValueError as e: raise HTTPException(409, str(e)) from e diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index 25ea30be..2a06c831 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -114,23 +114,26 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita raise except asyncio.CancelledError: raise - except NodeError: - raise except Exception as e: # noqa: BLE001 + # NodeError 以前在这里直接 raise，于是节点自己抛的错既不重试、 + # 也不发 node.failed、更不被 on_error=continue 接住。而 NodeError + # 恰恰是所有执行器表达失败的标准方式（缺必填输入、模型没配、 + # 口径卡算不出…），等于这两个配置对最常见的报错全都不生效。 last_error = e if attempt < retries: ctx.emit( EventType.LOG, level="warn", - message=f"第 {attempt + 1} 次失败（{type(e).__name__}: {e}），准备重试", + message=f"第 {attempt + 1} 次失败（{_describe(e)}），准备重试", ) await asyncio.sleep(backoff * (2**attempt)) continue break elapsed = int((time.perf_counter() - started) * 1000) - message = f"{type(last_error).__name__}: {last_error}" - ctx.emit(EventType.NODE_FAILED, error=message, duration_ms=elapsed) + message = _describe(last_error) + # node_id 要带上：前端靠它把出错的节点从"转圈"收敛成"失败" + ctx.emit(EventType.NODE_FAILED, error=message, duration_ms=elapsed, node_id=node.id) # 容错模式：记下错误继续往下走，而不是让整张图挂掉 if node.config.get("on_error") == "continue": @@ -138,6 +141,10 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita "nodes": {node.id: {"error": message, "failed": True}}, "trail": [{"node_id": node.id, "error": message, "ts": time.time()}], } + # 已经是 NodeError 就别再套一层，否则消息会变成 + # "NodeError: NodeError: 真正的原因" + if isinstance(last_error, NodeError): + raise last_error raise NodeError(node.id, message) from last_error async def _entry(state: GraphState) -> dict[str, Any]: @@ -146,6 +153,15 @@ def _wrap(node: GraphNode, run_ctx: RunContext) -> Callable[[GraphState], Awaita return _entry +def _describe(error: BaseException | None) -> str: + """错误消息。NodeError 的 message 本来就是写给人看的，不用再前缀类型名。""" + if error is None: + return "未知错误" + if isinstance(error, NodeError): + return str(error) + return f"{type(error).__name__}: {error}" + + def _preview(updates: dict[str, Any], node_id: str) -> Any: """给前端卡片用的输出摘要，避免把整坨数据塞进事件流。""" payload = (updates.get("nodes") or {}).get(node_id) diff --git a/backend/app/engine/expressions.py b/backend/app/engine/expressions.py index def89713..7b2d3a8c 100644 --- a/backend/app/engine/expressions.py +++ b/backend/app/engine/expressions.py @@ -107,6 +107,27 @@ def render_deep(value: Any, ctx: dict[str, Any]) -> Any: # 安全表达式求值（分支条件用） # -------------------------------------------------------------------------- +# 幂运算的上界。`9**9**9**9` 只有 11 个 AST 节点，轻松过掉节点数护栏， +# 却会让 CPython 无限期地算一个天文数字的大整数 —— 而求值是在事件循环里 +# 同步执行的（分支条件、transform、口径卡、每个节点的 skip_if 都走这里）， +# 既没有线程池隔离也没有超时，asyncio 的 cancel 打断不了同步 CPU。 +# 一条画布上的分支条件就能把整个后端冻死。 +_MAX_POW_EXPONENT = 1024 +_MAX_POW_BASE = 1e15 + + +def _safe_pow(base: Any, exponent: Any) -> Any: + """带上界的幂运算：超界直接报错，而不是让进程算到天荒地老。""" + try: + if abs(exponent) > _MAX_POW_EXPONENT: + raise ExpressionError(f"指数超过上限 {_MAX_POW_EXPONENT}") + if abs(base) > _MAX_POW_BASE and exponent > 1: + raise ExpressionError(f"底数超过上限 {_MAX_POW_BASE:g}") + except TypeError as e: # 非数值类型交给 operator.pow 自己报 + raise ExpressionError(f"幂运算的操作数必须是数值：{e}") from e + return operator.pow(base, exponent) + + _BIN_OPS = { ast.Add: operator.add, ast.Sub: operator.sub, @@ -114,7 +135,7 @@ _BIN_OPS = { ast.Div: operator.truediv, ast.FloorDiv: operator.floordiv, ast.Mod: operator.mod, - ast.Pow: operator.pow, + ast.Pow: _safe_pow, } _CMP_OPS = { ast.Eq: operator.eq, diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 036ba404..c9cdfcdf 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -42,13 +42,21 @@ class RunManager: settings.ensure_dirs() self._cm = AsyncSqliteSaver.from_conn_string(str(settings.checkpoint_path)) self._checkpointer = await self._cm.__aenter__() - # 上次进程被强杀时留下的 running 记录，标成中断态而不是永远转圈 + # 上次进程被强杀时留下的记录。running 和 queued 要分开处理： + # running 至少跑过一步、有 checkpoint，可以续；queued 还在排队， + # 一个 checkpoint 都没写过，"恢复"它只会从 START 用默认值重跑一遍， + # 还丢掉 run.input——那不是恢复，是伪造一份结果。 async with SessionLocal() as session: await session.execute( update(Run) - .where(Run.status.in_(["running", "queued"])) + .where(Run.status == "running") .values(status="interrupted", error="服务重启，运行已挂起，可从断点恢复") ) + await session.execute( + update(Run) + .where(Run.status == "queued") + .values(status="failed", error="服务重启时这次运行还在排队，没有可恢复的进度，请重新发起") + ) await session.commit() async def shutdown(self) -> None: @@ -160,8 +168,23 @@ class RunManager: ) return run - async def resume(self, run_id: str, response: Any) -> Run: - """从人工介入的断点继续。""" + async def resume( + self, run_id: str, response: Any, *, approval_id: str | None = None, + actor: str | None = None, + ) -> Run: + """从人工介入的断点继续。 + + 两件事以前是错的： + + 一是不看有没有 checkpoint 就发 Command。重启扫描会把 queued（从没跑过、 + 没有任何 checkpoint）也标成 interrupted，对这种 run 发 Command，LangGraph + 会拿一个空 state 从 START 重跑整张图——run.input 完全没被注入，最后还落成 + succeeded。跑出来的是一份用默认值算的假结果。 + + 二是同一 superstep 里有多个 interrupt 时，只发标量 Command 会让 LangGraph + 直接抛 RuntimeError 打爆整个 run，而所有 pending 审批还都被写成同一个回复。 + 现在按 interrupt_id 逐条收集，凑齐了才继续，没凑齐就继续等下一个人回复。 + """ async with SessionLocal() as session: run = await session.get(Run, run_id) if not run: @@ -169,29 +192,103 @@ class RunManager: if run.status not in ("interrupted",): raise ValueError(f"当前状态是 {run.status}，无法恢复") spec = GraphSpec.model_validate(run.graph) - bus.set_seq(run_id, run.last_seq) - run.status = "running" - run.error = None + workflow_id = run.workflow_id - # 把这次恢复对应的待审批记录结掉，前端的待办列表才不会一直挂着 - pending = ( + # 先问引擎：这条线程到底停在哪、还等着哪些 interrupt + run_ctx = RunContext(run_id=run_id, thread_id=run_id, spec=spec, workflow_id=workflow_id) + app = compile_graph(spec, run_ctx).compile(checkpointer=self.checkpointer) + config = {"configurable": {"thread_id": run_id}} + snapshot = await app.aget_state(config) + waiting = list(getattr(snapshot, "interrupts", None) or []) + + if not waiting and not getattr(snapshot, "next", None): + # 没有断点可续。多半是重启时被标成 interrupted 的 queued 运行。 + raise ValueError( + "这次运行没有可恢复的断点（很可能重启前还没真正开始跑）。" + "请重新发起一次运行，而不是恢复——继续下去只会用默认值跑出一份假结果。" + ) + + answers: dict[str, Any] = {} + async with SessionLocal() as session: + pending = list(( await session.execute( select(Approval).where( Approval.run_id == run_id, Approval.status == "pending" ) ) - ).scalars() - for approval in pending: - approval.status = "resolved" - approval.response = _safe(response) if isinstance(response, dict) else {"value": _safe(response)} - approval.resolved_at = datetime.now(timezone.utc) + ).scalars()) + # 定位这次回复的是哪一条：显式指定优先，否则只在唯一 pending 时才敢猜 + target = None + if approval_id: + target = next((a for a in pending if a.id == approval_id), None) + if target is None: + raise ValueError(f"审批 {approval_id} 不在待处理列表里") + elif len(pending) == 1: + target = pending[0] + elif len(pending) > 1: + raise ValueError( + f"这次运行有 {len(pending)} 条待处理的人工介入，" + "恢复时必须指定 approval_id 说明回复的是哪一条" + ) + + now = datetime.now(timezone.utc) + if target is not None: + target.status = "answered" # 已回复，但还没提交给引擎 + target.response = ( + _safe(response) if isinstance(response, dict) else {"value": _safe(response)} + ) + target.resolved_at = now + if actor: + target.resolved_by = actor + await session.commit() + + # 凑齐没有？引擎还在等的每一个 interrupt 都得有答案 + answered = list(( + await session.execute( + select(Approval).where( + Approval.run_id == run_id, + Approval.status.in_(["answered", "pending"]), + ) + ) + ).scalars()) + by_iid = {a.interrupt_id: a for a in answered if a.interrupt_id} + missing = [] + for item in waiting: + iid = getattr(item, "id", None) + rec = by_iid.get(iid) + if rec is None or rec.status != "answered": + missing.append(iid) + continue + payload = rec.response or {} + answers[iid] = payload.get("value", payload) if isinstance(payload, dict) else payload + + if missing: + # 还差人没回。保持 interrupted，把已回复的那条留在 answered 上等齐 + await session.commit() + run = await session.get(Run, run_id) + return run + + # 齐了：正式落 resolved，然后驱动引擎 + for rec in answered: + if rec.status == "answered": + rec.status = "resolved" + run = await session.get(Run, run_id) + bus.set_seq(run_id, run.last_seq) + run.status = "running" + run.error = None await session.commit() await session.refresh(run) + # 单个 interrupt 用标量（LangGraph 两种都认），多个必须用 {interrupt_id: value} 映射 + command = ( + Command(resume=response) + if len(waiting) <= 1 + else Command(resume=answers) + ) await self._emit(run_id, EventType.RUN_RESUMED, data={"response": _safe(response)}) self._tasks[run_id] = asyncio.create_task( - self._drive(run_id, spec, Command(resume=response), workflow_id=run.workflow_id) + self._drive(run_id, spec, command, workflow_id=workflow_id) ) return run