From f63a1faf3b2b7b39bd3e408735aea3a30a421c45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=91=A3=E5=AE=8F=E5=87=AF?= Date: Sat, 20 Jun 2026 23:37:49 +0800 Subject: [PATCH] =?UTF-8?q?fix(roundtable):=20rwfx=20=E7=BB=93=E6=9E=84?= =?UTF-8?q?=E5=8C=96=E6=8A=BD=E5=8F=96=E7=BA=AF=E6=A0=91=C2=B7=E5=8D=95?= =?UTF-8?q?=E7=88=B6=E5=85=9C=E5=BA=95=EF=BC=8C=E6=A0=B9=E6=B2=BB=E5=8F=99?= =?UTF-8?q?=E4=BA=8B=E5=A4=9A=E5=AF=B9=E4=B8=80=E4=B9=B1=E7=BA=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../src/roundtable-planning/lib/dag-types.ts | 7 ++ .../app/gateway/routers/flow_extract.py | 68 +++++++++++++++- .../extraction_spec.py | 34 +++++++- .../agents/roundtable_orchestrator/types.py | 7 ++ .../tests/test_flow_extract_single_parent.py | 78 +++++++++++++++++++ 5 files changed, 189 insertions(+), 5 deletions(-) create mode 100644 offline-backend-20260512/backend/tests/test_flow_extract_single_parent.py diff --git a/frontend-web/src/roundtable-planning/lib/dag-types.ts b/frontend-web/src/roundtable-planning/lib/dag-types.ts index 6f1ae7b4..ae4e3557 100644 --- a/frontend-web/src/roundtable-planning/lib/dag-types.ts +++ b/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( diff --git a/offline-backend-20260512/backend/app/gateway/routers/flow_extract.py b/offline-backend-20260512/backend/app/gateway/routers/flow_extract.py index fd4017d7..b59a5046 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/flow_extract.py +++ b/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") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/extraction_spec.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/extraction_spec.py index 1ee3a665..24e7237f 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/extraction_spec.py +++ b/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）用的强制层级规范。 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py index 95c67566..e52d1182 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/roundtable_orchestrator/types.py +++ b/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 '席位'}派活，" diff --git a/offline-backend-20260512/backend/tests/test_flow_extract_single_parent.py b/offline-backend-20260512/backend/tests/test_flow_extract_single_parent.py new file mode 100644 index 00000000..b3757f56 --- /dev/null +++ b/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"}]