mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 08:31:59 +00:00
汇合节点只跑一次：长短支路不再把它触发两遍，循环的 done 出口也不再被提前放行
多条边汇入同一个节点时，编译器对每条入边各 add_edge 一次。LangGraph 对这种 接法的语义是"任一上游跑完就触发一次"：两条支路一样长时，两次触发落在同一步 被合并，看不出问题；一长一短时，汇合节点先拿短支路的结果跑一次，长支路到了 再跑一次——模型调用、工具副作用、下游全部翻倍。循环体里更糟：每轮汇合两次， 循环节点就多推进一格，done 出口在循环跑到一半时就放行了，成果节点跑了 3 次。 README 写的"自动等待汇聚"只在支路等长时成立。 修法是给汇合节点加 defer（等图里其余待执行的任务都跑完再跑一次），而不是 屏障 add_edge([a, b], j)：分支之后的二选一汇合，没走的那条永远不会到，屏障 会一直等，而 LangGraph 没有可执行任务时会安静结束——汇合节点和下游就被无声 跳过。代价写在 _join_nodes 的注释和 README 里：它也会等图里无关、恰好在跑的支路。 汇合 = 两个以上前向上游，循环回边不算。回边识别因此从 layout.py 挪到 schema.back_edges，加了 roots 参数：执行语义要从入口起 DFS，否则循环体里的 前向边可能被误判成回边；排版照旧按节点顺序起算，行为不变（check-canvas-layout 全过）。 新增 test_fan_in.py 5 例：长短支路、循环体内汇合（修前分别跑了 2 次和 4 次）； 二选一汇合、被分支剪掉一路的汇合（守着别把修法做成屏障）；回边不算汇合。 292 → 297 passed。内置模板里 ③④ 的成果节点是二选一汇合，行为不变。
This commit is contained in:
ThearchyHelios
@@ -150,7 +150,9 @@ flowchart LR
|
||||
{{ vars.items | json }} 过滤器：json / compact / upper / length / first …
|
||||
```
|
||||
|
||||
连线即并行：单个节点连出多条边构成 fan-out，多条边汇入同一节点时自动等待汇聚。
|
||||
连线即并行：单个节点连出多条边构成 fan-out，多条边汇入同一节点时自动等待汇聚——
|
||||
汇合节点等图里其余待执行的任务都跑完再执行一次，支路一长一短也不会被触发两遍；
|
||||
分支之后的二选一汇合不会去等没走的那条。代价是它也会等图里与它无关、恰好还在跑的支路。
|
||||
|
||||
自动排版按拓扑分层：并行的支路落在同一列，层内按中位数启发式排序以减少交叉，
|
||||
分支节点的出口按 yes/no 的声明顺序自上而下，循环的回边不参与分层（否则循环体会
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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"}
|
||||
Reference in New Issue
Block a user