From 9d375f8f1c5b2b78e26d7a59a742edd9574519df Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 25 Sep 2026 17:04:49 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BE=AA=E7=8E=AF=E8=B7=91=E6=BB=A1=E8=87=AA?= =?UTF-8?q?=E5=B7=B1=E5=A3=B0=E6=98=8E=E7=9A=84=E8=BD=AE=E6=95=B0=EF=BC=8C?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E8=A2=AB=E5=85=A8=E5=B1=80=E6=AD=A5=E6=95=B0?= =?UTF-8?q?=E4=B8=8A=E9=99=90=E5=8D=8A=E8=B7=AF=E6=8E=90=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 「测试 1」按校验提示改好之后仍然跑不完：while 循环配了 1005 轮，每轮是"循环节点 + 循环体"两步，而整次运行的步数上限（max_graph_steps）是 200——跑到 100 轮上下 LangGraph 抛 GraphRecursionError，报错是一段英文，叫人去调一个界面上碰不到的 recursion_limit。 图没写错，是平台自己的两个上限在打架：校验通过，跑到一半才死。 全局上限防的是没人把关的环；loop 节点的 max_iterations 是作者明说的轮数上限。现在循环 按自己的轮数另记一份预算（schema.loop_steps），加在全局上限之上，子图同理。取上界： 一个节点最多跑 Π(包住它的各层循环 max_iterations + 1) 次，嵌套相乘；done 出口之后的 节点只跑一次，不按轮数记账，否则预算虚高、等于没有上限。 没有 loop 节点把关的环（分支连回上游）照旧由全局上限拦住，报错改成中文、说清预算怎么 来的、该怎么改；子图里撞上限报子图自己的名字和步数——以前那段英文会被外层包进 NodeError 原样带出来。 实测：「测试 1」照提示改完，在真沙箱里 1000 轮跑满，25 秒，结果 501500（= 2+3+…+1001）。 测试：test_step_budget 8 条——预算的上界、嵌套相乘、done 之后不记账；全局上限压到 20 步 时循环照样跑满 30 轮；空转的环中文报错；子图里的循环有自己的预算、空转的子图报它自己的 名字。把 runner / 子图的预算换成 0，对应用例确实失败。README 补了常见问题，测试数更新为 470。 --- README.md | 7 +- backend/app/engine/nodes/multi.py | 38 ++++--- backend/app/engine/runner.py | 19 +++- backend/app/engine/schema.py | 45 ++++++++ backend/tests/test_step_budget.py | 164 ++++++++++++++++++++++++++++++ 5 files changed, 256 insertions(+), 17 deletions(-) create mode 100644 backend/tests/test_step_budget.py diff --git a/README.md b/README.md index c23a5193..acb6f717 100644 --- a/README.md +++ b/README.md @@ -437,12 +437,17 @@ agent 侧同样会收到实情，并被告知可直接查询 `information_schema 单次模型调用默认 300 秒没有输出即超时。多半是模型或工具没有响应；确实需要更久，调大 `AGENTLAB_MAX_RUN_SECONDS` / `AGENTLAB_MODEL_TIMEOUT_SECONDS`。设置页「运行环境」里能看到当前值。 +**运行报「走满了 N 步还没跑完」。** 循环按自己 `max_iterations` 声明的轮数另算步数预算， +配多少轮就能跑多少轮，不受全局上限约束。走到这条报错的，多半是分支连回上游、却没有 +loop 节点给它定轮数上限的环：用 loop 节点包住它。确实需要更多步，调大 +`AGENTLAB_MAX_GRAPH_STEPS`（默认 200）。 + --- ## 开发与验证 ```bash -# 后端：292 个测试 +# 后端：470 个测试 cd backend && pytest tests # 前端类型检查 diff --git a/backend/app/engine/nodes/multi.py b/backend/app/engine/nodes/multi.py index eb78f5a1..5c6fb5c8 100644 --- a/backend/app/engine/nodes/multi.py +++ b/backend/app/engine/nodes/multi.py @@ -6,6 +6,7 @@ import time from typing import Any from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage +from langgraph.errors import GraphRecursionError from langgraph.func import task from app.core.config import settings @@ -367,7 +368,7 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: 每一步；但它有独立的状态，不会污染父图的变量池。 """ from app.engine.compiler import compile_graph # 延迟导入，避免循环依赖 - from app.engine.schema import GraphSpec + from app.engine.schema import GraphSpec, loop_steps if ctx.run.depth >= _MAX_DEPTH: raise NodeError(ctx.node.id, f"子图嵌套超过 {_MAX_DEPTH} 层，已阻止") @@ -429,19 +430,28 @@ async def run_subgraph(state: GraphState, ctx: NodeContext) -> dict[str, Any]: compiled = compile_graph(sub_spec, sub_run) # 子图不带 checkpointer：它的断点由父图的 checkpoint 统一承载 app = compiled.compile() - result_state = await app.ainvoke( - { - "input": sub_input, - "vars": sub_input, - "nodes": {}, - "output": {}, - "messages": [], - "usage": {}, - "trail": [], - "loops": {}, - }, - {"recursion_limit": settings.max_graph_steps}, - ) + limit = settings.max_graph_steps + loop_steps(sub_spec) + try: + result_state = await app.ainvoke( + { + "input": sub_input, + "vars": sub_input, + "nodes": {}, + "output": {}, + "messages": [], + "usage": {}, + "trail": [], + "loops": {}, + }, + {"recursion_limit": limit}, + ) + except GraphRecursionError as e: + # 不接住的话，外层包装成 NodeError 时带的是 LangGraph 的英文原文 + raise NodeError( + ctx.node.id, + f"子图「{workflow.name}」走满了 {limit} 步还没跑完，已中止。多半是子图里有个环" + "在空转：分支连回了上游、却没有 loop 节点给它定轮数上限", + ) from e output = result_state.get("output") or {} result = { diff --git a/backend/app/engine/runner.py b/backend/app/engine/runner.py index 1520493d..bcb27a18 100644 --- a/backend/app/engine/runner.py +++ b/backend/app/engine/runner.py @@ -9,6 +9,7 @@ from datetime import datetime, timezone from typing import Any from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver +from langgraph.errors import GraphRecursionError from langgraph.types import Command from sqlalchemy import select, update @@ -19,7 +20,7 @@ from app.db.base import SessionLocal from app.db.models import Approval, Run, RunEvent, Workflow from app.engine.compiler import compile_graph, initial_state from app.engine.context import NodeError, RunContext -from app.engine.schema import GraphSpec, topology_of, validate_graph +from app.engine.schema import GraphSpec, loop_steps, topology_of, validate_graph logger = logging.getLogger(__name__) @@ -466,9 +467,11 @@ class RunManager: collection=collection, ) app = compile_graph(spec, run_ctx).compile(checkpointer=self.checkpointer) + # 循环按自己声明的轮数另记一份预算，全局上限只管没人把关的环 + extra_steps = loop_steps(spec) config = { "configurable": {"thread_id": run_id}, - "recursion_limit": settings.max_graph_steps, + "recursion_limit": settings.max_graph_steps + extra_steps, } interrupted = False @@ -520,6 +523,18 @@ class RunManager: error = f"{type(e).__name__}: {e}" logger.exception("run %s 失败", run_id) await self._emit(run_id, EventType.RUN_FAILED, data={"error": error}) + except GraphRecursionError: + status = "failed" + # LangGraph 的原文是英文，还叫人去调 recursion_limit——用户碰不到那个键。 + # 循环的轮数已经另算了预算，走到这里的多半是不归 loop 节点管的环 + error = ( + f"走满了 {settings.max_graph_steps + extra_steps} 步还没跑完，已中止" + f"（图最大步数 {settings.max_graph_steps}" + + (f"，另按循环轮数预留了 {extra_steps} 步" if extra_steps else "") + "）。" + "多半是有个环在空转：分支连回了上游、却没有 loop 节点给它定轮数上限。" + "用 loop 节点包住它；确实需要这么多步，调大 AGENTLAB_MAX_GRAPH_STEPS" + ) + await self._emit(run_id, EventType.RUN_FAILED, data={"error": error}) except Exception as e: # noqa: BLE001 status = "failed" # NodeError 的 message 本来就是写给人看的（"人工驳回：…"）， diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index 8f49c574..fe8d0ffc 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -211,6 +211,51 @@ def back_edges( return back +def loop_steps(spec: GraphSpec) -> int: + """图里的循环按各自声明的轮数跑满，最多要走多少步（LangGraph 的 superstep）。 + + 整次运行的步数上限（max_graph_steps）防的是没人把关的环；loop 节点自己带着 + max_iterations，那是作者明说的轮数上限。以前两者打架时全局值悄悄赢：「测试 1」 + 的 while 循环配了 1005 轮、每轮 2 步，跑到 100 轮上下就被 GraphRecursionError + 掐断——校验通过，跑到一半才死。runner 把这个数加在全局上限之上。 + + 取上界不求精确：一个节点最多跑 Π(包住它的各层循环的 max_iterations + 1) 次， + 逐个加起来。一步至少跑一个节点，所以步数只会比这少。 + """ + loops = [n for n in spec.nodes if n.type == NodeType.LOOP] + if not loops: + return 0 + succ: dict[str, list[str]] = defaultdict(list) + pred: dict[str, list[str]] = defaultdict(list) + for e in spec.edges: + succ[e.source].append(e.target) + pred[e.target].append(e.source) + + def reach(starts: Iterable[str], step: dict[str, list[str]], avoid: str) -> set[str]: + seen: set[str] = set() + stack = [s for s in starts if s != avoid] + while stack: + n = stack.pop() + if n not in seen: + seen.add(n) + stack.extend(x for x in step[n] if x != avoid) + return seen + + runs: dict[str, int] = defaultdict(lambda: 1) + for loop in loops: + try: + cap = int(loop.config.get("max_iterations", 10) or 10) + except (TypeError, ValueError): + cap = 10 # 运行到这个节点时同样会报错，这里不替它报 + # 循环体 = 从 body 出口出发、不经过循环节点本身、又能绕回它的那些节点。 + # 没标出口的边路由时按 default 走，body 也会走它 + entries = [e.target for e in spec.outgoing(loop.id) if e.sourceHandle != "done"] + body = reach(entries, succ, loop.id) & reach(pred[loop.id], pred, loop.id) + for n in body | {loop.id}: + runs[n] *= max(cap, 0) + 1 + return sum(runs.values()) + + def expression_fields(node: GraphNode) -> list[tuple[str, str]]: """这个节点上按表达式求值的字段：(给人看的名字, 原文)。 diff --git a/backend/tests/test_step_budget.py b/backend/tests/test_step_budget.py new file mode 100644 index 00000000..ab1aaec0 --- /dev/null +++ b/backend/tests/test_step_budget.py @@ -0,0 +1,164 @@ +"""循环跑满自己声明的轮数，不被全局步数上限半路掐断。 + +开发库里的「测试 1」：while 循环 `vars.state.count <= 1000`，max_iterations 配了 +1005。每轮是「循环节点 + 循环体」两步，而整次运行的步数上限（max_graph_steps） +是 200——跑到 100 轮上下 LangGraph 抛 GraphRecursionError，报错是一段英文， +叫人去调一个界面上根本碰不到的 recursion_limit。校验是通过的：这张图没写错， +是平台自己的两个上限在打架。 + +全局上限防的是没人把关的环；loop 节点的 max_iterations 是作者明说的轮数上限。 +所以循环按自己的轮数另记一份预算（loop_steps），全局上限照旧管其余的环。 +""" +from __future__ import annotations + +import asyncio + +import pytest + +from app.core.config import settings +from app.db.base import SessionLocal +from app.db.models import Run +from app.engine.runner import run_manager +from app.engine.schema import GraphSpec, loop_steps + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "data": {"label": nid, "config": config}} + + +def edge(source, target, handle=None): + return {"source": source, "target": target, "sourceHandle": handle} + + +def counting_loop(rounds: int, cap: int) -> dict: + """「测试 1」的骨架：while 数到 rounds，循环体一个节点。""" + return { + "nodes": [ + node("start", "input", fields=[]), + node("init", "transform", expression="0", assign_to="n"), + node("lp", "loop", mode="while", condition=f"vars.n < {rounds}", max_iterations=cap), + node("inc", "transform", expression="vars.n + 1", assign_to="n"), + node("out", "output", fields=[{"name": "结果", "value": "{{ vars.n }}"}]), + ], + "edges": [edge("start", "init"), edge("init", "lp"), edge("lp", "inc", "body"), + edge("inc", "lp"), edge("lp", "out", "done")], + } + + +def test_no_loop_no_extra_budget(): + spec = GraphSpec.model_validate({ + "nodes": [node("a", "input"), node("b", "output")], "edges": [edge("a", "b")]}) + assert loop_steps(spec) == 0 + + +def test_budget_covers_the_declared_rounds(): + # 1005 轮 × 每轮 2 步，外加收尾的一次判定 + assert loop_steps(GraphSpec.model_validate(counting_loop(1000, 1005))) >= 1005 * 2 + 1 + + +def test_nested_loops_multiply(): + """内层循环每一轮外层都要从头跑一遍，预算得乘起来，不是加起来。""" + spec = GraphSpec.model_validate({ + "nodes": [node("start", "input"), + node("outer", "loop", mode="foreach", items="[1,2,3]", max_iterations=3), + node("inner", "loop", mode="foreach", items="[1,2,3,4]", max_iterations=4), + node("work", "transform", expression="1"), + node("out", "output")], + "edges": [edge("start", "outer"), edge("outer", "inner", "body"), + edge("inner", "work", "body"), edge("work", "inner"), + edge("inner", "outer", "done"), edge("outer", "out", "done")], + }) + # 实际：外层 4 次判定 + 3 轮 ×（内层 5 次判定 + 4 次干活）= 31 步 + assert loop_steps(spec) >= 31 + + +def test_nodes_after_the_loop_are_not_charged_per_round(): + """done 出口之后的节点只跑一次，不该按轮数记账——否则预算虚高，等于没有上限。""" + spec = GraphSpec.model_validate(counting_loop(3, 3)) + assert loop_steps(spec) == (3 + 1) * 2 # 只有 lp 和 inc + + +@pytest.fixture +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +async def _finish(graph) -> Run: + run = await run_manager.start(graph=graph, input_payload={}) + for _ in range(600): + async with SessionLocal() as session: + row = await session.get(Run, run.id) + if row.status in ("succeeded", "failed", "cancelled", "interrupted"): + return row + await asyncio.sleep(0.05) + raise AssertionError(f"没跑完：{row.status}") + + +async def test_a_loop_runs_past_the_global_step_limit(engine_up, monkeypatch): + """全局上限压到 20 步，循环照样跑满它声明的 30 轮。""" + monkeypatch.setattr(settings, "max_graph_steps", 20) + row = await _finish(counting_loop(30, 35)) + assert row.status == "succeeded", row.error + assert row.output["结果"] == "30" + + +async def test_an_unguarded_cycle_still_stops_and_says_why(engine_up, monkeypatch): + """分支连回上游、没有 loop 节点把关的环，照旧由全局上限拦住——报错得是人话。""" + monkeypatch.setattr(settings, "max_graph_steps", 12) + row = await _finish({ + "nodes": [node("start", "input", fields=[]), + node("t", "transform", expression="1"), + node("br", "branch", cases=[{"key": "again", "condition": "true"}]), + node("out", "output", fields=[])], + "edges": [edge("start", "t"), edge("t", "br"), edge("br", "t", "again"), + edge("br", "out", "default")], + }) + assert row.status == "failed" + assert "走满了 12 步" in row.error and "loop 节点" in row.error, row.error + assert "recursion_limit" not in row.error + + +# -------------------------------------------------------------------------- +# 子图：自己的循环自己算预算；撞上限时报子图自己的名字和步数 +# -------------------------------------------------------------------------- + + +async def _parent_of(sub_graph: dict, name: str) -> dict: + from app.db.models import Workflow + + async with SessionLocal() as session: + wf = Workflow(name=name, graph=sub_graph) + session.add(wf) + await session.commit() + wf_id = wf.id + return { + "nodes": [node("start", "input", fields=[]), + node("sub", "subgraph", workflow_id=wf_id), + node("out", "output", fields=[{"name": "结果", "value": "{{ nodes.sub.output.结果 }}"}])], + "edges": [edge("start", "sub"), edge("sub", "out")], + } + + +async def test_a_loop_inside_a_subgraph_gets_its_own_budget(engine_up, monkeypatch): + monkeypatch.setattr(settings, "max_graph_steps", 20) + row = await _finish(await _parent_of(counting_loop(30, 35), "数到 30")) + assert row.status == "succeeded", row.error + assert row.output["结果"] == "30" + + +async def test_an_unguarded_cycle_inside_a_subgraph_says_which_one(engine_up, monkeypatch): + monkeypatch.setattr(settings, "max_graph_steps", 12) + spinning = { + "nodes": [node("start", "input", fields=[]), + node("t", "transform", expression="1"), + node("br", "branch", cases=[{"key": "again", "condition": "true"}]), + node("out", "output", fields=[])], + "edges": [edge("start", "t"), edge("t", "br"), edge("br", "t", "again"), + edge("br", "out", "default")], + } + row = await _finish(await _parent_of(spinning, "空转的子图")) + assert row.status == "failed" + assert "子图「空转的子图」走满了 12 步" in row.error, row.error + assert "recursion_limit" not in row.error