fix(roundtable): rwfx 结构化抽取纯树·单父兜底，根治叙事多对一乱线

rwfx 全链是一棵树：除预期行为↔驱动因素 1:1 外其余皆 1:N，任何节点只有一个父节点、
绝无多对一/多对多/汇聚。但弱模型常把「叙事」连到一片驱动因素 → 汇聚乱线。

- extraction_spec：ChainSpec 加 single_parent 标志（rwfx=True，xdfx 有汇聚=False）；
  helper is_single_parent_chain / chain_level_rank；强化 rwfx ②「叙事只有一个父驱动因素」
  + 新增④「全链是树，每个非根节点恰一个 from」
- flow_extract：_collapse_to_single_parent 确定性兜底——纯树业务把每个节点的多父入边
  收敛到「紧邻上一层」的单一父节点（只删多余入边、不新增、不动节点），在 final 前执行；
  非纯树业务（xdfx）跳过
- dag-types/types：DAG 派活忠实本阶段目标——不擅自加数量上限/精简、不替席位加体例风格、
  逐条全部输出、不诱导脱离输入编造（前后台同步）
- Tests: tests/test_flow_extract_single_parent.py

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

frontend-web/src/roundtable-planning/lib/dag-types.ts

@@ -392,6 +392,13 @@ export function buildDagStageLeaderTask(args: {
     lines.push(
       `【本阶段目标】${goalText}`,
       "请据此目标为每个席位拆解任务，并要求他们**显式承接上一阶段的产出**来完成本阶段目标。",
+      "⚠️ 派活时**严格忠实于本阶段目标，不要自行新增目标里没有的「数量上限 / 只取最有代表性的 N 组 / 精简归纳成几条」等限制**，" +
+        "也不要照搬其它阶段的数量（如上一阶段筛了 3 组，不代表本阶段也只要 3 组）；" +
+        "目标没要求筛选 / 限定数量时，就要求席位**按其职责把每一条都完整输出**——" +
+        "若上一阶段有 N 条产出需逐条处理，就让席位**逐条、全部**产出（有多少条就出多少条），绝不替席位归纳 / 合并 / 只取代表性的几条。",
+      "⚠️ 也**不要替席位新增目标 / 席位职责之外的「输出体例 / 风格」要求**（如『写成连贯故事 / 要有场景感 / " +
+        "举具体城市景区为例 / 虚构时间地点情节』等）——派活只需转达本阶段目标 + 上一阶段产物 + 承接关系，" +
+        "**让席位严格按它自己的职责设定（SOUL）输出**，且**忠实基于上一阶段的真实产物做分析，不要诱导席位脱离输入去编造 / 虚构事例**。",
     );
   }
   lines.push(

offline-backend-20260512/backend/app/gateway/routers/flow_extract.py

@@ -36,7 +36,11 @@ from app.gateway.routers._roundtable_seed import (
     STRUCTURE_AGENT_ID,
     ensure_roundtable_functional_agents,
 )
-from deerflow.agents.roundtable_orchestrator.extraction_spec import build_structure_directive
+from deerflow.agents.roundtable_orchestrator.extraction_spec import (
+    build_structure_directive,
+    chain_level_rank,
+    is_single_parent_chain,
+)
 from deerflow.config import get_app_config
 from deerflow.config.agents_config import load_agent_config, load_agent_soul
 from deerflow.models.factory import create_chat_model
@@ -406,6 +410,64 @@ def _derive_layered_edges(nodes: list[dict[str, Any]]) -> list[dict[str, Any]] |
     return out
 
 
+def _collapse_to_single_parent(graph: dict[str, Any], business_code: str | None) -> dict[str, Any]:
+    """**纯树**业务链（rwfx）确定性兜底：每个非根节点最多保留一个父节点（一条入边）。
+
+    rwfx 全链是一棵树（除 1:1 的预期行为↔驱动因素，其余皆 1:N），任何节点都只有一个父节点、
+    绝无多对一/汇聚。但弱模型常给某节点（典型是「叙事」）连多个上层父节点 → 画出汇聚/乱线。
+    这里按「最近上一层」原则把每个节点的入边收敛到一条：在该节点的多个父节点里，优先保留 group
+    处于它**紧邻上一层**的那个（按业务链层级序 ``chain_level_rank``）；同层多个则保留模型给出的
+    第一个；都判断不了就保留第一个。根节点（无入边）不变。**只删多余入边、不新增**，不影响节点。
+    """
+    nodes = graph.get("nodes") or []
+    edges = graph.get("edges") or []
+    rank = chain_level_rank(business_code)  # 层级名 → 自上而下序号
+    group_by_id: dict[str, str] = {
+        n["id"]: _as_str(n.get("group")) for n in nodes if isinstance(n, dict) and n.get("id")
+    }
+    appear: dict[str, int] = {}
+    for n in nodes:
+        g = _as_str(n.get("group")) if isinstance(n, dict) else ""
+        if g and g not in appear:
+            appear[g] = len(appear)
+
+    def grank(nid: str) -> int:
+        g = group_by_id.get(nid, "")
+        if g in rank:
+            return rank[g]
+        return 1000 + appear.get(g, 0)  # 未在业务层级里的 group 排到最后，不抢「上一层」判定
+
+    by_to: dict[str, list[dict[str, Any]]] = {}
+    order: list[str] = []
+    for e in edges:
+        t = e.get("to")
+        if t not in by_to:
+            by_to[t] = []
+            order.append(t)
+        by_to[t].append(e)
+
+    kept: list[dict[str, Any]] = []
+    collapsed = 0
+    for t in order:
+        ins = by_to[t]
+        if len(ins) <= 1:
+            kept.extend(ins)
+            continue
+        tr = grank(t)
+        prev_layer = [e for e in ins if grank(_as_str(e.get("from"))) == tr - 1]
+        if prev_layer:
+            best = prev_layer[0]
+        else:
+            upper = [e for e in ins if grank(_as_str(e.get("from"))) < tr]
+            best = max(upper, key=lambda e: grank(_as_str(e.get("from")))) if upper else ins[0]
+        kept.append(best)
+        collapsed += len(ins) - 1
+    if collapsed:
+        logger.info("flow-extract: collapsed %d extra parent edge(s) to single-parent (tree chain %s)", collapsed, business_code)
+    graph["edges"] = kept
+    return graph
+
+
 # ── prompts ────────────────────────────────────────────────────────────────
 
 _SHAPE = (
@@ -735,6 +797,10 @@ async def extract_flow_stream(request: Request, payload: FlowExtractRequest = Bo
             if graph is None:
                 yield _frame({"event": "error", "detail": "未能从模型输出解析出流程图节点"})
                 return
+            # 纯树业务链（rwfx）确定性兜底：把任何节点的多父入边收敛到单一父节点，杜绝弱模型
+            # 违反 1:N 画出的汇聚/乱线（典型是叙事连了一片驱动因素）。非纯树业务（xdfx 有汇聚）跳过。
+            if is_single_parent_chain(payload.business_code):
+                graph = _collapse_to_single_parent(graph, payload.business_code)
             yield _frame({"event": "final", "graph": graph})
         except Exception as exc:  # noqa: BLE001
             logger.exception("flow-extract stream failed")

offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/extraction_spec.py

@@ -31,6 +31,10 @@ class ChainSpec:
     relations_note: str  # 层间关系说明（1:1 / 1:N / 多对一 / 分叉汇聚，复杂结构全写在这里）
     completeness_note: str = ""  # 「完整性」要求行；空 → 用通用「以上 N 层都要出节点」
     extra_note: str = ""  # 该业务额外补充（可空）
+    # 是否为**纯树**业务链：每个非根节点恰有一个父节点（全链 1:N + 1:1，绝无多对一/多对多/汇聚）。
+    # 为真时后端结构化抽取会**确定性兜底**把每个节点收敛到单一父节点（见 flow_extract._collapse_to_single_parent），
+    # 杜绝弱模型违反 1:N 画出乱线/汇聚。rwfx=True；xdfx 有「脆弱点→策略」多对一汇聚，为 False。
+    single_parent: bool = False
 
 
 # 节点命名 / 取数优先：各业务通用，单独抽出避免重复。
@@ -75,9 +79,11 @@ _RWFX = ChainSpec(
         "而要**按预期行为把驱动因素拆开 / 重新归纳**：给每条预期行为各配一个驱动因素节点（其内容综合提炼该条预期行为背后最核心的那个原因），"
         "拆分 / 提炼后务必让两层等量、严格一一对应；\n"
         "② **「驱动因素」→「叙事」是一对多**——**一个驱动因素对应多个叙事**（每个驱动因素抽出它名下的若干条叙事），"
-        "每条叙事的 from **连到它所属的那一个驱动因素**（叙事的上一层是驱动因素，不是预期行为）；"
-        "**绝不能把多个驱动因素汇聚连到同一个叙事节点**（那就把一对多画反成多对一了，是常见错误）；也不要把全部叙事合并成一个节点；\n"
-        "③ 其余相邻层级（任务→目的、目的→行为体、行为体→预期行为）均为**一对多**（上层一条可带下层多条，下层每条连其所属上层条目）。"
+        "每条叙事的 from **只连它所属的那一个驱动因素**（叙事的上一层是驱动因素，不是预期行为、不是行为体）；"
+        "**绝不能把多个驱动因素汇聚连到同一条叙事**（那是多对一，rwfx 链不存在），**一条叙事只能有一个父节点**；"
+        "若报告里某条叙事看似关联了多个行为体 / 多处驱动，**也只把它归到最直接驱动它的那一个驱动因素名下、只连这一个父节点**，不要给它连多个 from；也不要把全部叙事合并成一个节点；\n"
+        "③ 其余相邻层级（任务→目的、目的→行为体、行为体→预期行为）均为**一对多**（上层一条可带下层多条，下层每条连其所属上层条目）；\n"
+        "④ **全链是一棵树：除根「任务」外，每个节点有且仅有一个父节点（一个 from）**——任何节点都不会从多个上层节点汇入（无多对一 / 多对多 / 汇聚）。"
     ),
     completeness_note=(
         "以上 6 个层级都要出节点，且每层的**全部条目都要抽全**：\n"
@@ -87,8 +93,10 @@ _RWFX = ChainSpec(
         "  · **驱动因素（1:1 特例，按上面①处理）**：驱动因素层节点数 **= 预期行为层节点数**——每条预期行为各配一个核心驱动因素，"
         "**绝不能因为报告把多个原因挤在一处 / 留空，就只抽 1 个驱动因素去连多条预期行为（多对一，本次最严重的错）**，也不要抽得比预期行为多；务必两层等量、位次一一配对；\n"
         "  · **叙事**：报告若列了多个候选叙事方案（如「3 组候选叙事」），**全部**抽成叙事节点（含未选用的），不要只出一两个；"
-        "叙事层通常比驱动因素层**更宽**，每条叙事连到它所属/对接的那一个驱动因素——不要让多个驱动因素指向同一叙事。"
+        "叙事层通常比驱动因素层**更宽**，每条叙事连到它所属/对接的那一个驱动因素——不要让多个驱动因素指向同一叙事；"
+        "**每条叙事只有一个父驱动因素（单 from）**，绝不要把一条叙事连到多个驱动因素。"
     ),
+    single_parent=True,  # rwfx 是纯树：每个非根节点恰一个父节点（后端确定性兜底收敛）。
 )
 
 
@@ -140,6 +148,24 @@ def get_chain_spec(business_code: str | None) -> ChainSpec | None:
     return _SPECS.get(norm)
 
 
+def is_single_parent_chain(business_code: str | None) -> bool:
+    """该业务链是否为**纯树**（每个非根节点恰一个父节点，无多对一/多对多/汇聚）。
+
+    为真时结构化抽取会确定性兜底把每个节点收敛到单一父节点。rwfx=True；xdfx（有汇聚）/
+    未写实业务=False（保守，不强行收敛）。
+    """
+    spec = get_chain_spec(business_code)
+    return bool(spec and spec.single_parent)
+
+
+def chain_level_rank(business_code: str | None) -> dict[str, int]:
+    """该业务链的「层级名 → 自上而下序号」映射（用于确定性兜底判断「上一层」）。无写实规范 → 空。"""
+    spec = get_chain_spec(business_code)
+    if spec is None:
+        return {}
+    return {name: i for i, name in enumerate(spec.levels)}
+
+
 def build_structure_directive(business_code: str | None) -> str:
     """结构化抽取（flow-json）用的强制层级规范。
 

offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py

@@ -348,6 +348,13 @@ def build_dag_stage_leader_task(
         lines += [
             f"【本阶段目标】{goal_text}",
             "请据此目标为每个席位拆解任务，并要求他们显式承接上一阶段的产出来完成本阶段目标。",
+            "⚠️ 派活时**严格忠实于本阶段目标，不要自行新增目标里没有的「数量上限 / 只取最有代表性的 N 组 / 精简归纳成几条」等限制**，"
+            "也不要照搬其它阶段的数量（如上一阶段筛了 3 组，不代表本阶段也只要 3 组）；"
+            "目标没要求筛选 / 限定数量时，就要求席位**按其职责把每一条都完整输出**——"
+            "若上一阶段有 N 条产出需逐条处理，就让席位**逐条、全部**产出（有多少条就出多少条），绝不替席位归纳 / 合并 / 只取代表性的几条。",
+            "⚠️ 也**不要替席位新增目标 / 席位职责之外的「输出体例 / 风格」要求**（如『写成连贯故事 / 要有场景感 / "
+            "举具体城市景区为例 / 虚构时间地点情节』等）——派活只需转达本阶段目标 + 上一阶段产物 + 承接关系，"
+            "**让席位严格按它自己的职责设定（SOUL）输出**，且**忠实基于上一阶段的真实产物做分析，不要诱导席位脱离输入去编造/ 虚构事例**。",
         ]
     lines += [
         f"请你作为总控，通过 agent_orchestration 技能**分别**向上述{'这些席位' if is_parallel else '席位'}派活，"

offline-backend-20260512/backend/tests/test_flow_extract_single_parent.py

@@ -0,0 +1,78 @@
+"""rwfx 结构化抽取「纯树·单父节点」确定性兜底测试。
+
+rwfx 全链是一棵树（除预期行为↔驱动因素 1:1，其余皆 1:N），任何节点只有一个父节点、绝无
+多对一/汇聚。弱模型常给「叙事」连多个驱动因素 → 乱线。``_collapse_to_single_parent`` 兜底把
+每个节点的多父入边收敛到紧邻上一层的单一父节点。
+"""
+
+from __future__ import annotations
+
+from app.gateway.routers.flow_extract import _collapse_to_single_parent
+from deerflow.agents.roundtable_orchestrator.extraction_spec import (
+    chain_level_rank,
+    is_single_parent_chain,
+)
+
+
+def test_is_single_parent_chain() -> None:
+    assert is_single_parent_chain("6BF") is True
+    assert is_single_parent_chain("rwfx") is True
+    assert is_single_parent_chain("rwfx-A") is True  # 多智能体 -A 后缀也认
+    assert is_single_parent_chain("7BF") is False  # xdfx 有「脆弱点→策略」汇聚，非纯树
+    assert is_single_parent_chain("3Q") is False  # 未写实业务保守不收敛
+    assert is_single_parent_chain(None) is False
+
+
+def test_chain_level_rank_rwfx() -> None:
+    rank = chain_level_rank("6BF")
+    assert rank == {"任务": 0, "目的": 1, "行为体": 2, "预期行为": 3, "驱动因素": 4, "叙事": 5}
+
+
+def _rwfx_graph_with_bad_narrative() -> dict:
+    # 叙事 s1 错误地连了 2 个驱动因素(上一层) + 1 个行为体(更上层)，应被收敛成单父。
+    return {
+        "title": "t",
+        "nodes": [
+            {"id": "n1", "group": "任务"},
+            {"id": "m1", "group": "目的"},
+            {"id": "a1", "group": "行为体"},
+            {"id": "b1", "group": "预期行为"},
+            {"id": "b2", "group": "预期行为"},
+            {"id": "d1", "group": "驱动因素"},
+            {"id": "d2", "group": "驱动因素"},
+            {"id": "s1", "group": "叙事"},
+        ],
+        "edges": [
+            {"from": "n1", "to": "m1"},
+            {"from": "m1", "to": "a1"},
+            {"from": "a1", "to": "b1"},
+            {"from": "a1", "to": "b2"},
+            {"from": "b1", "to": "d1"},
+            {"from": "b2", "to": "d2"},
+            {"from": "d1", "to": "s1"},  # 叙事多父：驱动因素 d1
+            {"from": "d2", "to": "s1"},  # 叙事多父：驱动因素 d2
+            {"from": "a1", "to": "s1"},  # 叙事多父：行为体 a1（跨层）
+        ],
+    }
+
+
+def test_collapse_narrative_multi_parent_to_single() -> None:
+    out = _collapse_to_single_parent(_rwfx_graph_with_bad_narrative(), "6BF")
+    s1_parents = [e["from"] for e in out["edges"] if e["to"] == "s1"]
+    # 叙事只剩一个父；应保留紧邻上一层（驱动因素），不保留更上层的行为体；同层保留第一个。
+    assert s1_parents == ["d1"]
+    # 其余每个非根节点仍恰一个父（1:1 / 1:N 本就单父，不受影响）。
+    for nid in ("m1", "a1", "b1", "b2", "d1", "d2"):
+        assert len([e for e in out["edges"] if e["to"] == nid]) == 1
+    # 根节点无入边。
+    assert [e for e in out["edges"] if e["to"] == "n1"] == []
+
+
+def test_collapse_noop_when_already_tree() -> None:
+    # 每个节点本就单父 → 边数不变、内容不变。
+    graph = {
+        "nodes": [{"id": "n1", "group": "任务"}, {"id": "m1", "group": "目的"}, {"id": "a1", "group": "行为体"}],
+        "edges": [{"from": "n1", "to": "m1"}, {"from": "m1", "to": "a1"}],
+    }
+    out = _collapse_to_single_parent(graph, "6BF")
+    assert out["edges"] == [{"from": "n1", "to": "m1"}, {"from": "m1", "to": "a1"}]