董宏凯andClaude Opus 4.8
b1b1ff98f5
深链接会商（taskId+goPath，businessCode）下按所选业务链完整层级强制校正 结构化抽取(flow-json) 与 方案总结报告： - 新增 harness 层 extraction_spec（app+harness 共用，单份领域知识两角色用法）： build_structure_directive / build_summary_directive 按 businessCode 取 · rwfx(6BF)：线性 任务→目的→行为体→预期行为→驱动因素→叙事；预期行为↔驱动因素 1:1、余 1:N · xdfx(7BF)：分叉汇聚 DAG，根=行动；行动→受众目标/关键传播节点(1:N)→脆弱点(1:N)→策略(多对一) · 其余业务通用规范（强约束完整性+命名+取数、不套固定模板）；business_code 空→零副作用 - 结构化抽取：flow_extract._hierarchy_hint + _validate_messages 用 structure_directive 权威覆盖默认层级（根治"无论什么业务都按目的→行为体抽/停在行为体"）；前端 streamFlowExtract 传 business_code（源 ingestBusinessCode） - 方案总结报告：前台 multi_agent._special_run(summary) + 后台 build_summary_prompt 注入 business 完整性核对（StartJobRequest/StartParams 透传） - 报告产物策略：.md 报告文件保持完整、干净、无需修改（不写 待补充/不足/⚠️）； 「核对说明」（意图对齐+业务链完整性+不足/可补充）改为只在对话回复正文、且每次必出 - 种子 roundtable-structure SOUL 层级顺序改 spec-driven、去 rwfx 写死示例 Tests: test_roundtable_extraction_spec / test_roundtable_summary_prompt Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
108 lines
4.7 KiB
Python
108 lines
4.7 KiB
Python
"""按 businessCode 的「业务链抽取规范」+ 其在 flow_extract / summary 的注入测试。
|
|
|
|
深链接会商模式下：结构化抽取(flow_extract)按业务完整层级抽全（根治"无论什么业务都按目的→行为体抽"）、
|
|
方案总结报告按业务链完整性核对。规范单份、两角色用法；仅 business_code 非空时注入。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from deerflow.agents.roundtable_orchestrator.extraction_spec import (
|
|
build_structure_directive,
|
|
build_summary_directive,
|
|
get_chain_spec,
|
|
)
|
|
|
|
|
|
# ── 规范本身 ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_rwfx_structure_directive_has_full_six_levels_and_relations():
|
|
out = build_structure_directive("6BF")
|
|
for level in ("任务", "目的", "行为体", "预期行为", "驱动因素", "叙事"):
|
|
assert level in out
|
|
assert "一一对应" in out # 预期行为↔驱动因素 1:1
|
|
assert "一对多" in out
|
|
# 不含步骤/阶段名 的命名约束
|
|
assert "步骤" in out and "阶段" in out
|
|
assert "优先" in out and "方案总结报告" in out # 取数优先
|
|
|
|
|
|
def test_xdfx_structure_directive_is_branching_dag_with_many_to_one():
|
|
out = build_structure_directive("7BF")
|
|
for node in ("行动", "受众目标", "关键传播节点", "脆弱点", "策略"):
|
|
assert node in out, node
|
|
assert "分叉汇聚" in out
|
|
assert "多对一" in out
|
|
# 根是「行动」而非「任务」
|
|
assert "根节点 = 用户传入的「行动」" in out
|
|
# 7BF / xdfx / xdfx-A 都命中同一规范
|
|
for code in ("7BF", "xdfx", "XDFX-A"):
|
|
assert get_chain_spec(code) is not None and get_chain_spec(code).label == "xdfx", code
|
|
|
|
|
|
def test_business_code_normalization_handles_suffix_and_case():
|
|
# rwfx-A / 小写 / 含 -A 后缀都应命中 rwfx 规范
|
|
for code in ("rwfx", "RWFX", "rwfx-A", "6bf", "6BF"):
|
|
spec = get_chain_spec(code)
|
|
assert spec is not None and spec.label == "rwfx", code
|
|
|
|
|
|
def test_unknown_business_gets_generic_directive_not_rwfx_template():
|
|
out = build_structure_directive("3Q")
|
|
assert out # 非空（深链接业务仍注入通用规范）
|
|
assert "不要套用" in out # 明确不要套固定模板
|
|
# 通用规范不应写死 rwfx 的「预期行为/驱动因素」层级
|
|
assert "预期行为" not in out
|
|
|
|
|
|
def test_no_business_code_returns_empty():
|
|
assert build_structure_directive("") == ""
|
|
assert build_structure_directive(None) == ""
|
|
assert build_summary_directive("") == ""
|
|
assert build_summary_directive(None) == ""
|
|
|
|
|
|
def test_rwfx_summary_directive_checks_completeness():
|
|
out = build_summary_directive("6BF")
|
|
assert "完整性核对" in out
|
|
assert "叙事" in out and "驱动因素" in out
|
|
assert "待补充" in out # 缺层走护栏标注
|
|
|
|
|
|
# ── flow_extract 层级提示注入（结构化抽取） ───────────────────────────────────
|
|
|
|
|
|
def test_flow_extract_hierarchy_hint_uses_spec_when_business_code():
|
|
from app.gateway.routers.flow_extract import FlowExtractRequest, _hierarchy_hint
|
|
|
|
req = FlowExtractRequest(report_md="x", business_code="6BF")
|
|
hint = _hierarchy_hint(req)
|
|
# 业务规范权威注入：含完整层级 + 覆盖默认假设的措辞
|
|
assert "预期行为" in hint and "驱动因素" in hint and "叙事" in hint
|
|
assert "强制遵守" in hint
|
|
|
|
# 无 business_code → 回退原行为（不含业务规范块）
|
|
req2 = FlowExtractRequest(report_md="x", chain_levels=["目的", "行为体"])
|
|
hint2 = _hierarchy_hint(req2)
|
|
assert "强制遵守" not in hint2
|
|
assert "目的" in hint2 # 仍用 chain_levels
|
|
|
|
|
|
# ── build_summary_prompt 业务完整性注入（方案总结报告） ────────────────────────
|
|
|
|
|
|
def test_build_summary_prompt_injects_business_completeness():
|
|
from deerflow.agents.roundtable_orchestrator.engine import build_summary_prompt
|
|
from deerflow.agents.roundtable_orchestrator.types import SeatRef
|
|
|
|
agents = [SeatRef("seat-a", "情报")]
|
|
out = build_summary_prompt(
|
|
"目标X", agents, {"seat-a": "交付"}, "共识", business_code="6BF"
|
|
)
|
|
assert "完整性核对" in out
|
|
assert "预期行为" in out and "驱动因素" in out
|
|
# 无 business_code → 不注入业务完整性块（但意图对齐块仍在）
|
|
out2 = build_summary_prompt("目标X", agents, {"seat-a": "交付"}, "共识")
|
|
assert "完整性核对" not in out2
|
|
assert "意图对齐与缺口核对" in out2
|