diff --git a/README.md b/README.md index 03e015ca..6bf9b342 100644 --- a/README.md +++ b/README.md @@ -150,7 +150,9 @@ flowchart LR {{ vars.items | json }} 过滤器：json / compact / upper / length / first … ``` -连线即并行：单个节点连出多条边构成 fan-out，多条边汇入同一节点时自动等待汇聚。 +连线即并行：单个节点连出多条边构成 fan-out，多条边汇入同一节点时自动等待汇聚—— +汇合节点等图里其余待执行的任务都跑完再执行一次，支路一长一短也不会被触发两遍； +分支之后的二选一汇合不会去等没走的那条。代价是它也会等图里与它无关、恰好还在跑的支路。 自动排版按拓扑分层：并行的支路落在同一列，层内按中位数启发式排序以减少交叉， 分支节点的出口按 yes/no 的声明顺序自上而下，循环的回边不参与分层（否则循环体会 diff --git a/backend/app/engine/compiler.py b/backend/app/engine/compiler.py index 76252a7f..aeeccddc 100644 --- a/backend/app/engine/compiler.py +++ b/backend/app/engine/compiler.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio import time +from collections import defaultdict from typing import Any, Awaitable, Callable from langgraph.errors import GraphBubbleUp @@ -10,7 +11,7 @@ from langgraph.graph import END, START, StateGraph from app.core.events import EventType from app.engine.context import NodeContext, NodeError, RunContext from app.engine.nodes import control, human, io, knowledge, llm, metrics, multi, tools -from app.engine.schema import GraphNode, GraphSpec, NodeType +from app.engine.schema import GraphNode, GraphSpec, NodeType, back_edges from app.engine.state import GraphState NodeRunner = Callable[[GraphState, NodeContext], Awaitable[dict[str, Any]]] @@ -238,6 +239,35 @@ def _make_router( return route, path_map +def _join_nodes(spec: GraphSpec, node_map: dict[str, GraphNode]) -> set[str]: + """有两个以上上游、要等汇齐了才能跑的节点。 + + 每条入边各 add_edge 一次时，LangGraph 的语义是"任何一个上游跑完就触发 + 一次"。两条支路一样长，两次触发落在同一步里被合并，看不出问题；一长一短， + 汇合节点就先拿着短的那边跑一次、长的到了再跑一次——模型调用和工具副作用 + 都翻倍，下游跟着重跑；循环体里的汇合还会让循环节点多推进一格，把 done + 出口提前放行。 + + 不用屏障（add_edge([a, b], j)）：分支之后的二选一汇合，没走的那条永远 + 不会到，屏障会一直等，而 LangGraph 没有可执行任务时会安静地结束——汇合 + 节点和它下游的一切就这么被跳过了，连报错都没有。defer 的语义是"等图里 + 其余待执行的任务都跑完，再跑这一次"：该到的都到齐了，没走的也不会被等。 + 代价是它也会等图里与它无关、恰好还在跑的支路。 + + 循环回边不算：循环节点的入边是"入口 + 回边"，那不是汇合。 + """ + entries = [n.id for n in spec.entry_nodes()] + back = back_edges(node_map, spec.edges, roots=entries) + upstream: dict[str, set[str]] = defaultdict(set) + for e in spec.edges: + if e.source not in node_map or e.target not in node_map: + continue + if (e.source, e.target, e.sourceHandle or "") in back: + continue + upstream[e.target].add(e.source) + return {node_id for node_id, sources in upstream.items() if len(sources) >= 2} + + def compile_graph(spec: GraphSpec, run_ctx: RunContext) -> StateGraph: """把图定义编译成未 compile 的 StateGraph。 @@ -246,9 +276,10 @@ def compile_graph(spec: GraphSpec, run_ctx: RunContext) -> StateGraph: """ builder = StateGraph(GraphState) node_map = spec.node_map() + joins = _join_nodes(spec, node_map) for node in spec.nodes: - builder.add_node(node.id, _wrap(node, run_ctx)) + builder.add_node(node.id, _wrap(node, run_ctx), defer=node.id in joins) # 入口 entries = spec.entry_nodes() diff --git a/backend/app/engine/layout.py b/backend/app/engine/layout.py index 14a52993..c079ac99 100644 --- a/backend/app/engine/layout.py +++ b/backend/app/engine/layout.py @@ -25,7 +25,7 @@ from __future__ import annotations from collections import defaultdict, deque from statistics import median -from app.engine.schema import GraphNode, GraphSpec, NodeType +from app.engine.schema import GraphNode, GraphSpec, NodeType, back_edges #: 画布节点的宽度。前端 NodeCard 的 `width: 238` 是写死的，这里必须跟着它—— #: 排版算出来的走廊是给连线用的，宽度估错了走廊就白留。 @@ -85,42 +85,6 @@ def _dedupe_edges(spec: GraphSpec, nodes: dict[str, GraphNode]) -> list: return out -def _back_edges(nodes: dict[str, GraphNode], edges: list) -> set[tuple[str, str, str]]: - """DFS 森林里的回边。剔掉它们，剩下的必然是无环的。 - - 迭代式而不是递归：一张几百个节点的图不该因为递归深度挂掉。 - """ - adj: dict[str, list] = defaultdict(list) - for e in edges: - adj[e.source].append(e) - - WHITE, GREY, BLACK = 0, 1, 2 - state = dict.fromkeys(nodes, WHITE) - back: set[tuple[str, str, str]] = set() - - for root in nodes: - if state[root] != WHITE: - continue - stack: list[tuple[str, int]] = [(root, 0)] - state[root] = GREY - while stack: - node_id, i = stack[-1] - out = adj[node_id] - if i >= len(out): - state[node_id] = BLACK - stack.pop() - continue - stack[-1] = (node_id, i + 1) - edge = out[i] - target = edge.target - if state[target] == GREY: - back.add((edge.source, edge.target, edge.sourceHandle or "")) - elif state[target] == WHITE: - state[target] = GREY - stack.append((target, 0)) - return back - - def _ranks( nodes: dict[str, GraphNode], edges: list, back: set[tuple[str, str, str]], ) -> dict[str, int]: @@ -390,7 +354,7 @@ def auto_layout( # 自环不参与分层：它不跨层，留着只会让"自己的前驱是自己"这种噪音渗进排序 edges = [e for e in edges if e.source != e.target] - back = _back_edges(nodes, edges) + back = back_edges(nodes, edges) rank = _ranks(nodes, edges, back) preds: dict[str, list[str]] = defaultdict(list) diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index 07023364..661f2c3b 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -1,7 +1,8 @@ from __future__ import annotations +from collections import defaultdict from enum import StrEnum -from typing import Any, Literal +from typing import Any, Iterable, Literal from pydantic import BaseModel, Field, field_validator, model_validator @@ -167,6 +168,48 @@ def topology_of(spec: GraphSpec) -> tuple[tuple, tuple]: ) +def back_edges( + node_ids: Iterable[str], edges: Iterable[GraphEdge], *, roots: Iterable[str] = (), +) -> set[tuple[str, str, str]]: + """DFS 森林里的回边，以 (source, target, sourceHandle) 表示。剔掉它们，剩下的必然无环。 + + 哪条边算"回边"取决于 DFS 从哪里出发。排版不在乎，按节点顺序起算就行； + 执行语义在乎——得从入口（roots）起算，否则循环体里的一条普通前向边可能被 + 误判成回边。两端有一端不在 node_ids 里的悬空边直接忽略。 + + 迭代式而不是递归：一张几百个节点的图不该因为递归深度挂掉。 + """ + WHITE, GREY, BLACK = 0, 1, 2 + state = dict.fromkeys(node_ids, WHITE) + adj: dict[str, list[GraphEdge]] = defaultdict(list) + for e in edges: + if e.source in state and e.target in state: + adj[e.source].append(e) + + back: set[tuple[str, str, str]] = set() + for root in [*roots, *state]: + if state.get(root) != WHITE: + continue + stack: list[tuple[str, int]] = [(root, 0)] + state[root] = GREY + while stack: + node_id, i = stack[-1] + out = adj[node_id] + if i >= len(out): + state[node_id] = BLACK + stack.pop() + continue + stack[-1] = (node_id, i + 1) + edge = out[i] + target = edge.target + if state[target] == GREY: + back.add((edge.source, edge.target, edge.sourceHandle or "")) + elif state[target] == WHITE: + state[target] = GREY + stack.append((target, 0)) + return back + + def validate_graph(spec: GraphSpec) -> ValidationResult: """编译前的静态检查。错误会挡住运行，警告只在画布上提示。""" result = ValidationResult() diff --git a/backend/tests/test_fan_in.py b/backend/tests/test_fan_in.py new file mode 100644 index 00000000..8df4f2f5 --- /dev/null +++ b/backend/tests/test_fan_in.py @@ -0,0 +1,221 @@ +"""汇合节点只跑一次：等该到的都到了，再动手。 + +多条边汇入同一个节点时，编译器以前对每条入边各调一次 add_edge。LangGraph +对这种接法的语义是"任何一条入边的上游跑完，就触发一次"——两条支路一样长时， +两次触发落在同一步里被合并成一次执行，看起来一切正常；一长一短时，汇合节点 +先拿着短支路的结果跑一次（长支路的数据还是旧的），长支路到了再跑一次。模型 +调用、工具副作用全部翻倍，下游也跟着跑两遍，而第一次的半截结果已经进了事件流。 + +修法不能是一律改成屏障（add_edge([a, b], j)）：分支之后的"二选一"汇合， +没走的那条永远不会到，屏障会一直等下去——而 LangGraph 在没有可执行任务时 +会安静地结束运行，汇合节点和它下游的一切就这么被跳过了，连报错都没有。 +所以这里守两头：长短支路只跑一次，分支剪掉的那条不许把汇合节点饿死。 +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from app.db.base import SessionLocal +from app.db.models import Run, RunEvent +from app.engine.compiler import compile_graph +from app.engine.context import RunContext +from app.engine.runner import run_manager +from app.engine.schema import GraphSpec + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +def node(nid, ntype, **config): + return { + "id": nid, "type": ntype, "position": {"x": 0, "y": 0}, + "data": {"label": nid, "config": config}, + } + + +def text(nid, template, assign_to=None): + cfg = {"mode": "template", "template": template} + if assign_to: + cfg["assign_to"] = assign_to + return node(nid, "transform", **cfg) + + +async def _run(graph, *, timeout=20.0): + run = await run_manager.start(graph=graph, input_payload={"question": "随便"}) + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + async with SessionLocal() as session: + row = await session.get(Run, run.id) + if row and row.status in ("succeeded", "failed", "cancelled"): + return row + await asyncio.sleep(0.05) + raise AssertionError(f"运行 {run.id} 没有在 {timeout}s 内结束") + + +async def _starts(run_id: str) -> list[str]: + async with SessionLocal() as session: + rows = list((await session.execute( + RunEvent.__table__.select().where(RunEvent.run_id == run_id) + .order_by(RunEvent.seq) + )).mappings()) + return [r["node_id"] for r in rows if r["type"] == "node.started" and r["node_id"]] + + +async def test_unequal_branches_join_once_with_both_results(): + """一长一短两条支路汇合：汇合节点只跑一次，而且拿到的是两边的最终结果。""" + graph = { + "nodes": [ + node("start", "input", fields=[{"name": "question"}]), + text("a1", "A", assign_to="a"), + text("a2", "{{ vars.a }}!", assign_to="a"), + text("b", "B", assign_to="b"), + text("j", "{{ vars.a }}+{{ vars.b }}", assign_to="joined"), + node("out", "output", fields=[{"name": "结果", "value": "{{ vars.joined }}"}]), + ], + "edges": [ + {"source": "start", "target": "a1"}, + {"source": "a1", "target": "a2"}, + {"source": "a2", "target": "j"}, + {"source": "start", "target": "b"}, + {"source": "b", "target": "j"}, + {"source": "j", "target": "out"}, + ], + } + run = await _run(graph) + assert run.status == "succeeded", run.error + + starts = await _starts(run.id) + assert starts.count("j") == 1, f"汇合节点跑了 {starts.count('j')} 次：{starts}" + assert starts.count("out") == 1, f"下游跟着重跑了：{starts}" + assert run.output["结果"] == "A!+B" + + +async def test_branch_merge_does_not_wait_for_the_road_not_taken(): + """分支之后的二选一汇合：没走的那条永远不会到，不能一直等它。""" + graph = { + "nodes": [ + node("start", "input", fields=[{"name": "question"}]), + node("br", "branch", cases=[{"key": "yes", "condition": "true"}]), + text("x", "走了 yes", assign_to="road"), + text("y", "走了 default", assign_to="road"), + text("j", "汇合：{{ vars.road }}", assign_to="joined"), + node("out", "output", fields=[{"name": "结果", "value": "{{ vars.joined }}"}]), + ], + "edges": [ + {"source": "start", "target": "br"}, + {"source": "br", "target": "x", "sourceHandle": "yes"}, + {"source": "br", "target": "y", "sourceHandle": "default"}, + {"source": "x", "target": "j"}, + {"source": "y", "target": "j"}, + {"source": "j", "target": "out"}, + ], + } + run = await _run(graph) + assert run.status == "succeeded", run.error + + starts = await _starts(run.id) + assert "y" not in starts + assert starts.count("j") == 1, starts + assert run.output["结果"] == "汇合：走了 yes" + + +async def test_join_still_runs_when_one_feeder_is_pruned_by_a_branch(): + """一条支路被分支整个剪掉：汇合节点照样要跑，下游不能被安静地跳过。""" + graph = { + "nodes": [ + node("start", "input", fields=[{"name": "question"}]), + text("c", "C", assign_to="c"), + node("br", "branch", cases=[{"key": "yes", "condition": "false"}]), + text("x", "X", assign_to="x"), + text("j", "只有 {{ vars.c }}", assign_to="joined"), + node("out", "output", fields=[{"name": "结果", "value": "{{ vars.joined }}"}]), + ], + "edges": [ + {"source": "start", "target": "c"}, + {"source": "start", "target": "br"}, + # default 出口没接线：分支判 default 就在这里结束，x 永远不会跑 + {"source": "br", "target": "x", "sourceHandle": "yes"}, + {"source": "c", "target": "j"}, + {"source": "x", "target": "j"}, + {"source": "j", "target": "out"}, + ], + } + run = await _run(graph) + assert run.status == "succeeded", run.error + + starts = await _starts(run.id) + assert "x" not in starts + assert starts.count("j") == 1, starts + assert run.output["结果"] == "只有 C" + + +async def test_join_inside_a_loop_runs_once_per_iteration(): + """循环体里的长短汇合：每一轮汇合一次，不多不少。""" + graph = { + "nodes": [ + node("start", "input", fields=[{"name": "question"}]), + node("lp", "loop", mode="foreach", items="[1, 2]", max_iterations=5), + text("p", "第 {{ vars.item }} 轮"), + text("q1", "长"), + text("q2", "长长"), + text("r", "短"), + text("j", "合"), + node("out", "output", fields=[{"name": "结果", "value": "完"}]), + ], + "edges": [ + {"source": "start", "target": "lp"}, + {"source": "lp", "target": "p", "sourceHandle": "body"}, + {"source": "p", "target": "q1"}, + {"source": "q1", "target": "q2"}, + {"source": "p", "target": "r"}, + {"source": "q2", "target": "j"}, + {"source": "r", "target": "j"}, + {"source": "j", "target": "lp"}, + {"source": "lp", "target": "out", "sourceHandle": "done"}, + ], + } + run = await _run(graph) + assert run.status == "succeeded", run.error + + starts = await _starts(run.id) + assert starts.count("p") == 2, starts + assert starts.count("j") == 2, f"两轮应该各汇合一次：{starts}" + + +def test_loop_back_edge_does_not_make_the_loop_node_a_join(): + """循环节点的入边是"入口 + 回边"，这不是汇合，不该被推迟。 + + 推迟只会让它白等图里其他还在跑的支路；真正需要等的是有两条以上 + 前向入边的节点。 + """ + graph = GraphSpec.model_validate({ + "nodes": [ + node("start", "input", fields=[{"name": "question"}]), + node("lp", "loop", mode="foreach", items="[1]"), + text("body", "x"), + text("a", "a"), + text("b", "b"), + text("j", "j"), + node("out", "output", fields=[]), + ], + "edges": [ + {"source": "start", "target": "lp"}, + {"source": "lp", "target": "body", "sourceHandle": "body"}, + {"source": "body", "target": "lp"}, + {"source": "lp", "target": "a", "sourceHandle": "done"}, + {"source": "lp", "target": "b", "sourceHandle": "done"}, + {"source": "a", "target": "j"}, + {"source": "b", "target": "j"}, + {"source": "j", "target": "out"}, + ], + }) + builder = compile_graph(graph, RunContext(run_id="t", thread_id="t", spec=graph)) + deferred = {nid for nid, spec in builder.nodes.items() if spec.defer} + assert deferred == {"j"}