feat(query): activate selected creation families

acquisition/queries/axes.py

@@ -0,0 +1,32 @@
+"""Shared axes for formal creation-query families."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+TREES = ROOT / "scope_trees" / "trees_index.json"
+
+ACTIONS = ["构思", "策划", "组织", "撰写", "改编", "润色"]
+STAGES = ["灵感", "选题", "脚本"]
+KTYPE_SUFFIX = {"what": "有哪些", "why": "为什么", "how": "怎么做"}
+
+
+def _nonleaf_d4(
+    source_type: str,
+    limit: int,
+    under: str | None = None,
+    *,
+    tree_path: Path = TREES,
+) -> list[str]:
+    """Return level-4 non-leaf node names, preserving the legacy axis order."""
+    idx = json.loads(tree_path.read_text("utf-8"))
+    paths = {
+        n.get("path") or ""
+        for n in idx
+        if n.get("source_type") == source_type
+        and (not under or under in (n.get("path") or "").split("/"))
+    }
+    d4 = [p for p in paths if len([x for x in p.split("/") if x]) == 4]
+    nonleaf = sorted(p for p in d4 if any(o != p and o.startswith(p + "/") for o in paths))
+    return [p.split("/")[-1] for p in nonleaf][:limit]

acquisition/queries/builder.py

@@ -7,7 +7,7 @@ from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
-from acquisition.query import ACTIONS, STAGES as OLD_STAGES, _nonleaf_d4
+from acquisition.queries.axes import ACTIONS, STAGES, _nonleaf_d4
 from acquisition.queries.filter import filter_queries, prompt_version
 from acquisition.repositories.base import AcquisitionRepository
 from core.config import Settings
@@ -17,6 +17,7 @@ TREES = ROOT / "scope_trees" / "trees_index.json"
 KTYPE = ["怎么做", "有哪些", "为什么"]
 MODALITY = ["视频", "图文"]
 INTENT = ["灵感", "选题", "脚本"]
+DEFAULT_ACTIVE_FAMILY_KEYS = ("f1", "f2")
 
 
 @dataclass(frozen=True)
@@ -25,6 +26,7 @@ class QueryBuildOptions:
     batch_n: int = 30
     seed: int = 7
     dry: bool = False
+    active_family_keys: tuple[str, ...] = DEFAULT_ACTIVE_FAMILY_KEYS
 
 
 def _segs(path: str | None) -> list[str]:
@@ -90,8 +92,8 @@ def build_creation_query_batch(
     shi_batch = rng.sample(shi, min(opts.batch_n, len(shi)))
     xing_batch = rng.sample(xing, min(opts.batch_n, len(xing)))
     purpose_batch = rng.sample(purpose_pool, min(opts.batch_n, len(purpose_pool)))
-    f6_zy = _nonleaf_d4("作用", 10)
-    f6_stage_act = [(s, a) for s in OLD_STAGES for a in ACTIONS] + [("", "")]
+    f6_zy = _nonleaf_d4("作用", 10, tree_path=tree_path)
+    f6_stage_act = [(s, a) for s in STAGES for a in ACTIONS] + [("", "")]
 
     if not shi_batch or not xing_batch or not purpose_batch or not f6_zy:
         raise RuntimeError("scope tree does not contain enough creation query axes")
@@ -176,7 +178,7 @@ def build_creation_query_batch(
             "业务阶段": INTENT,
             "模态": MODALITY,
             "知识类型": KTYPE,
-            "阶段": OLD_STAGES,
+            "阶段": STAGES,
             "动作": ACTIONS,
             "作用": f6_zy,
         },
@@ -185,11 +187,18 @@ def build_creation_query_batch(
             "per": opts.per,
             "batch_n": opts.batch_n,
             "dry": opts.dry,
+            "active_family_keys": list(opts.active_family_keys),
             "query_filter_prompt_version": prompt_version(),
         },
         "families": [],
     }
-    for family in families:
+    family_by_key = {family["key"]: family for family in families}
+    unknown = [key for key in opts.active_family_keys if key not in family_by_key]
+    if unknown:
+        raise ValueError(f"unknown query family key(s): {', '.join(unknown)}")
+
+    for family_key in opts.active_family_keys:
+        family = family_by_key[family_key]
         name = " × ".join(family["axes"])
         seen: set[str] = set()
         items: list[dict[str, Any]] = []

acquisition/query.py

@@ -1,140 +0,0 @@
-"""生成 query 的 demo 引擎（只到「生成 query」为止：不搜索、不解构）。
-
-① 实质×创作阶段×需求点(LLM)  ② 形式×载体位置(LLM)  ③ 搜索词扩展(待接入真 sug)
-④ 多轴正交组合(机械)：实质×形式×阶段×动作×作用×知识类型
-取自 scope_trees 节点 + 人工定义轴；设计见 开发文档/query构造.md。
-"""
-from __future__ import annotations
-
-import json
-from pathlib import Path
-from typing import Optional
-
-from acquisition.suggest import suggest
-from core.config import Settings
-from core.llm import chat_json
-from core.prompts import load_prompt
-
-ROOT = Path(__file__).resolve().parent.parent
-TREES = ROOT / "scope_trees" / "trees_index.json"
-
-# ② 载体位置：载体 × 位置 的交叉（短视频/图片有封面，文章无封面）；剧本/小说/长文属于实质，不在此
-CARRIERS = ["短视频", "图片", "文章"]
-POSITIONS = ["开头", "中间", "收尾", "封面"]
-CARRIER_POS = [f"{c}{p}" for c in CARRIERS for p in POSITIONS if not (c == "文章" and p == "封面")]
-# 分组形式（给前端「查看全部」展示 载体 × 位置 的交叉）
-CARRIER_POS_GROUPED = {c: [p for p in POSITIONS if not (c == "文章" and p == "封面")] for c in CARRIERS}
-# ④ 需求点（人工拟定）：每个创作阶段下的细分需求，喂给 LLM 从中选 + 前端按钮展示
-DEMAND = {
-    "灵感": ["找方向", "找素材", "拆案例"],
-    "选题": ["选题", "爆款选题", "什么内容火"],
-    "脚本": ["开头钩子", "结构", "标题", "文案"],
-}
-# 制作屏蔽词：④ 过滤 + 前端标记（创作 vs 制作边界）
-BLOCK = ["剪辑", "调色", "参数", "导出", "软件", "生成", "插件", "渲染", "压制"]
-
-# ④ 多轴正交「组合」query（机械拼接）的人工定义轴
-ACTIONS = ["构思", "策划", "组织", "撰写", "改编", "润色"]            # 动作（待修改）
-STAGES = ["灵感", "选题", "脚本"]                                    # 阶段
-KTYPE_SUFFIX = {"what": "有哪些", "why": "为什么", "how": "怎么做"}   # 知识类型→句尾后缀
-MODALITIES = ["图文", "视频"]                                       # 知识模态（搜索筛选维度，不进 query 串）
-
-
-def sample_nodes(source_type: str, depths=(3, 4), limit: int = 20,
-                 under: Optional[str] = None) -> list[str]:
-    """读 trees_index.json，取某棵树的中层节点名（按出现序去重、封顶）。
-    under 给定时只取 path 含该分支的节点（如实质取 实质树/理念、手法取 形式树/架构）。"""
-    idx = json.loads(TREES.read_text("utf-8"))
-    out: list[str] = []
-    seen: set[str] = set()
-    for n in idx:
-        if n.get("source_type") != source_type:
-            continue
-        path = [x for x in (n.get("path") or "").split("/") if x]
-        if under and under not in path:
-            continue
-        if len(path) in depths:
-            name = n.get("name") or (path[-1] if path else "")
-            if name and name not in seen:
-                seen.add(name)
-                out.append(name)
-                if len(out) >= limit:
-                    break
-    return out
-
-
-def tactic2_form_llm(form_nodes: list[str], settings: Settings) -> list[dict]:
-    """② 形式树 × 载体位置 → LLM 正交生成自然 query（LLM 自行把书面形式标签理解成创作手法）。
-    返回 [{形式, 载体位置, query}]，供前端表格从左到右展示正交。1 次批量调用。"""
-    user = json.dumps({"形式树": form_nodes, "载体位置": CARRIER_POS, "屏蔽制作词": BLOCK}, ensure_ascii=False)
-    try:
-        res = chat_json(load_prompt("form_query_gen"), user, settings=settings, timeout=120)
-        rows = res.get("rows") or []
-    except Exception:
-        rows = []
-    return [{"形式": r.get("形式", ""), "载体位置": r.get("载体位置", ""), "query": r.get("query", "")}
-            for r in rows if isinstance(r, dict) and r.get("query")]
-
-
-def tactic3_suggest(seeds: list[dict], settings: Settings) -> list[dict]:
-    """③ 搜索词扩展（仅小红书）：每个种子 → keyword_v2 → 从相关帖挖候选搜索词。
-    seeds=[{query, 来源}]；来源标明种子出处（实质+意图 / 形式+需求词）。"""
-    from acquisition.crawler import RateLimiter
-    rl = RateLimiter(min_interval_seconds=1.0)
-    out = []
-    for s in seeds:
-        q, origin = s["query"], s.get("来源", "")
-        try:
-            cands = suggest(q, settings=settings, rate_limiter=rl, limit=12)
-        except Exception as exc:
-            cands = [f"(失败: {str(exc)[:40]})"]
-        out.append({"seed": q, "来源": origin, "候选": cands})
-    return out
-
-
-def tactic4_llm(topics: list[str], settings: Settings) -> list[dict]:
-    """④ LLM 正交清洗：实质 × 创作阶段 × 需求点(人工拟定 DEMAND) → 自然 query（滤制作）。
-    返回逐条带正交三轴的行：[{实质, 阶段, 需求点, query}]，供前端表格展示。1 次批量调用。"""
-    user = json.dumps({"实质": topics, "需求点表": DEMAND, "屏蔽制作词": BLOCK}, ensure_ascii=False)
-    try:
-        res = chat_json(load_prompt("query_gen"), user, settings=settings, timeout=120)
-        rows = res.get("rows") or []
-    except Exception:
-        rows = []
-    return [{"实质": r.get("实质", ""), "阶段": r.get("阶段", ""),
-             "需求点": r.get("需求点", ""), "query": r.get("query", "")}
-            for r in rows if isinstance(r, dict) and r.get("query")]
-
-
-def _nonleaf_d4(source_type: str, limit: int, under: Optional[str] = None) -> list[str]:
-    """取某棵树的【4级非叶子节点】名（实质 79 / 形式 47 / 作用 16…），按序采样封顶。
-    under 给定时只取该分支（如形式限 架构，避开 呈现 里的剪辑/后期等制作节点）。"""
-    idx = json.loads(TREES.read_text("utf-8"))
-    paths = {(n.get("path") or "") for n in idx if n.get("source_type") == source_type
-             and (not under or under in (n.get("path") or "").split("/"))}
-    d4 = [p for p in paths if len([x for x in p.split("/") if x]) == 4]
-    nonleaf = sorted(p for p in d4 if any(o != p and o.startswith(p + "/") for o in paths))
-    return [p.split("/")[-1] for p in nonleaf][:limit]
-
-
-def tactic_multiaxis(n: int = 36) -> list[dict]:
-    """④ 多轴正交组合（机械拼接）：实质×形式×阶段×动作×作用×知识类型 → 拼成「组合 query」。
-    实质/形式/作用 取自分类树（4级非叶子），阶段/动作/知识类型 人工定义；模态不进 query。
-    组合空间 ~320 万，这里 round-robin 取不同轴做【采样】，避免爆炸。无 LLM、纯机械。"""
-    sz = _nonleaf_d4("实质", 8)
-    xs = _nonleaf_d4("形式", 6, under="架构")   # 限创作手法（架构），避开呈现里的制作节点
-    zy = _nonleaf_d4("作用", 6)
-    ktypes = list(KTYPE_SUFFIX.items())                       # [(what,有哪些),…]
-    stage_act = [(s, a) for s in STAGES for a in ACTIONS] + [("", "")]  # +「无动作」变体
-    rows = []
-    for i in range(n):
-        s_ = sz[i % len(sz)]
-        f_ = xs[i % len(xs)]
-        st, ac = stage_act[i % len(stage_act)]
-        zy_ = zy[i % len(zy)]
-        kt, suf = ktypes[i % len(ktypes)]
-        seg = (st + ac) if ac else ""                         # 脚本撰写 / 空
-        parts = [s_, f_] + ([seg] if seg else []) + [zy_, suf]
-        rows.append({"实质": s_, "形式": f_, "阶段": st or "/", "动作": ac or "/",
-                     "作用": zy_, "知识类型": kt, "query": " ".join(parts)})
-    return rows

acquisition/suggest.py

@@ -1,91 +0,0 @@
-"""联想扩展：种子词 → 小红书 keyword_v2（返回相关帖）→ 从帖子标题/话题标签挖候选搜索词。
-
-⚠️ 小红书没有纯「搜索联想词」接口，所以这里是退一步：拿种子打 keyword_v2，从返回的
-【相关帖子】的 title + body_text 里的 #标签 + topic_list 挖候选词，不是搜索框下拉补全。
-（抖音只有 keyword 搜索接口、且与搜索共享强限流，已从 demo 移除，只保留小红书。）
-
-parse_suggest 纯函数可离线测；suggest 负责 HTTP（mirror search.py）。
-"""
-from __future__ import annotations
-
-import re
-from typing import Any, Optional
-from urllib.parse import urljoin
-
-import httpx
-
-from acquisition.crawler import RateLimiter
-from core.config import Settings
-
-SUGGEST_PATH = "/crawler/xiao_hong_shu/keyword_v2"
-_TAG = re.compile(r"#([^\s#＃]{2,20})")
-
-
-class SuggestError(RuntimeError):
-    pass
-
-
-def parse_suggest(response: Any, *, limit: int = 15) -> list[str]:
-    """从 keyword_v2 回包的相关帖挖候选搜索词：#标签/话题优先，标题兜底。去重截断。"""
-    if not isinstance(response, dict):
-        raise SuggestError("bad_response: not a dict")
-    if response.get("code") not in (0, "0"):
-        raise SuggestError(f"business_error: code={response.get('code')} msg={response.get('msg')}")
-    posts = ((response.get("data") or {}).get("data")) or []
-    tags: list[str] = []
-    titles: list[str] = []
-    for p in posts:
-        if not isinstance(p, dict):
-            continue
-        tags += _TAG.findall(f"{p.get('title', '')} {p.get('body_text', '')}")
-        for t in (p.get("topic_list") or []):
-            name = t.get("name") if isinstance(t, dict) else t
-            if name:
-                tags.append(str(name))
-        title = (p.get("title") or "").strip()
-        if title:
-            titles.append(title)
-    out: list[str] = []
-    seen: set[str] = set()
-    for c in tags + titles:          # 标签优先、标题兜底
-        c = c.strip()
-        if c and c not in seen:
-            seen.add(c)
-            out.append(c)
-            if len(out) >= limit:
-                break
-    return out
-
-
-def suggest(
-    keyword: str,
-    *,
-    content_type: str = "图文",
-    settings: Optional[Settings] = None,
-    http_client: Any = None,
-    rate_limiter: Optional[RateLimiter] = None,
-    env_file: str = ".env",
-    limit: int = 15,
-) -> list[str]:
-    """种子词 → 小红书候选搜索词列表（从相关帖挖）。失败抛 SuggestError。"""
-    settings = settings or Settings.from_env(env_file)
-    rate_limiter = rate_limiter or RateLimiter()
-    owns_client = http_client is None
-    client = http_client or httpx.Client()
-    try:
-        rate_limiter.wait("xhs_suggest")
-        url = urljoin(settings.aiddit_crawler_base_url, SUGGEST_PATH)
-        body = {"keyword": keyword, "content_type": content_type, "sort_type": "综合", "cursor": ""}
-        try:
-            resp = client.post(url, json=body, headers={"Content-Type": "application/json"},
-                               timeout=settings.crawler_timeout)
-            resp.raise_for_status()
-            data = resp.json()
-        except httpx.HTTPError as exc:
-            raise SuggestError(f"http_error: {exc}") from exc
-        except ValueError as exc:
-            raise SuggestError("bad_json") from exc
-        return parse_suggest(data, limit=limit)
-    finally:
-        if owns_client:
-            client.close()

prompts/form_query_gen.txt

@@ -1,15 +0,0 @@
-你在为「创作知识」做 query 正交生成。用户消息给一组【形式树】节点（创作的形式/手法分析标签，偏书面，如 条目列举 / 反差错位 / 戏剧张力 / 细节描绘）、一组【载体位置】（载体 × 位置 的交叉，如 短视频开头 / 图片封面 / 文章收尾）、一组【屏蔽制作词】。
-
-对每个 形式，挑 3-5 个**最贴切**的载体位置，各生成一条创作者真会在小红书/抖音搜的短句 query：
-- 先把书面的形式标签理解成创作者口语会说的手法（如 条目列举→列点/清单、反差错位→反差/反转、戏剧张力→冲突/张力、细节描绘→画面感），再结合该载体位置（哪个载体的哪个部分）生成自然口语 query；
-  例：形式『反差错位』×『短视频开头』→「短视频开头怎么用反差抓住人」；『条目列举』×『图片封面』→「图文封面怎么用清单感更吸睛」；『戏剧张力』×『文章收尾』→「文章结尾怎么留张力」；
-- 不是所有形式都适配所有载体位置——只生成讲得通的组合，跳过别扭的；
-- 只要「创作」（怎么想 / 怎么写 / 怎么呈现），**剔除「制作」**：含屏蔽制作词（剪辑/调色/参数/导出/软件…）或本质是「用工具产出成品」的，一律不要。
-
-每条 query 都要保留它的「形式」来源和「载体位置」。
-
-只输出一个 JSON 对象，rows 逐条列出：
-{"rows": [
-  {"形式":"<形式树节点>","载体位置":"<载体位置>","query":"<生成的搜索词>"},
-  ...
-]}

prompts/query_gen.txt

@@ -1,15 +0,0 @@
-你在为「创作知识」做 query 正交生成。用户消息给【实质】、【需求点表】(按创作阶段分好的需求点，人工拟定)、【屏蔽制作词】。
-
-对每个实质，沿「创作阶段 × 需求点」正交，生成口语化的搜索 query：
-- 阶段和需求点**只能用「需求点表」里给的**，不要自己另造；
-- 每个实质尽量覆盖多个 (阶段 × 需求点) 组合；
-- 像真人搜的短句（如「历史视频选题怎么找」「健身脚本开头怎么写」「美食爆款选题有哪些」），带上该实质词；
-- 只要「创作」（怎么想 / 怎么选 / 怎么写），**剔除「制作」**：含屏蔽制作词（剪辑/调色/参数/导出/软件…）或本质是「用工具产出成品」的，一律不要。
-
-为每条 query 标注它的【阶段】和【需求点】（正交两轴，取自需求点表）。
-
-只输出一个 JSON 对象，rows 逐条列出（每条含正交三轴 + 生成的 query）：
-{"rows": [
-  {"实质":"<实质>","阶段":"<需求点表里的阶段>","需求点":"<该阶段下的需求点>","query":"<生成的搜索词>"},
-  ...
-]}

scripts/build_creation_demo.py

@@ -9,6 +9,11 @@ from __future__ import annotations
 import argparse
 import json
 from pathlib import Path
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
 
 from acquisition.queries.builder import (
     QueryBuildOptions,
@@ -29,6 +34,12 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser.add_argument("--batch-n", type=int, default=30)
     parser.add_argument("--seed", type=int, default=7)
     parser.add_argument("--dry", action="store_true", help="Skip LLM query filtering")
+    parser.add_argument(
+        "--family",
+        action="append",
+        dest="families",
+        help="Query family key to activate; repeat to enable multiple families",
+    )
     parser.add_argument("--export-json", type=Path, help="Optional local review export")
     parser.add_argument("--persist", action="store_true", help="Persist to formal PG")
     parser.add_argument("--name", default="creation-demo")
@@ -46,6 +57,7 @@ def _summary(generated: dict) -> dict:
     )
     return {
         "family_count": len(families),
+        "family_keys": [family.get("key") for family in families],
         "query_count": total,
         "kept_count": kept,
         "metadata": generated.get("metadata") or {},
@@ -63,6 +75,7 @@ def main(argv: list[str] | None = None) -> int:
             batch_n=args.batch_n,
             seed=args.seed,
             dry=args.dry,
+            active_family_keys=tuple(args.families) if args.families else ("f1", "f2"),
         ),
     )
     summary = _summary(generated)

scripts/build_family6.py

@@ -1,52 +0,0 @@
-"""第六家族(最老正交方案去掉实质)：形式 × 阶段 × 动作 × 作用 × 知识类型 → 组合 query，过 query_filter。
-
-沿用 query.py 的老轴：形式/作用 取分类树(4级非叶子)，阶段=灵感/选题/脚本，动作=构思/策划/组织/
-撰写/改编/润色(+无动作变体)，知识类型=怎么做/有哪些/为什么。query 形如「叙事组织 脚本撰写 趣味互动 有哪些」。
-只产 query + LLM 筛选，打印看效果。用法：PYTHONPATH=. CK_ENV_FILE=.env python scripts/build_family6.py [n]
-"""
-from __future__ import annotations
-
-import sys
-
-from acquisition.query import ACTIONS, KTYPE_SUFFIX, STAGES, _nonleaf_d4
-from acquisition.query_filter import filter_queries
-from core.config import Settings
-
-N = int(sys.argv[1]) if len(sys.argv) > 1 else 24
-
-
-def build(n: int) -> list[dict]:
-    xs = _nonleaf_d4("形式", 10, under="架构")          # 创作手法(架构支)
-    zy = _nonleaf_d4("作用", 10)                          # 作用
-    ktypes = list(KTYPE_SUFFIX.items())                   # [(what,有哪些),…]
-    stage_act = [(s, a) for s in STAGES for a in ACTIONS] + [("", "")]  # +「无动作」变体(=老方案的 /)
-    rows = []
-    for i in range(n):
-        f_ = xs[i % len(xs)]
-        st, ac = stage_act[i % len(stage_act)]
-        zy_ = zy[i % len(zy)]
-        kt, suf = ktypes[i % len(ktypes)]
-        seg = (st + ac) if ac else ""                     # 脚本撰写 / 空(动作=/ 时连阶段一起省)
-        parts = [f_] + ([seg] if seg else []) + [zy_, suf]
-        rows.append({"形式": f_, "阶段": st or "/", "动作": ac or "/", "作用": zy_,
-                     "知识类型": kt, "query": " ".join(parts)})
-    return rows
-
-
-def main() -> None:
-    settings = Settings.from_env()
-    rows = build(N)
-    verdicts = filter_queries([r["query"] for r in rows], settings)
-    kept = 0
-    print(f"第六家族 形式×阶段×动作×作用×知识类型(去实质)：{len(rows)} 条\n")
-    print(f"{'留/扔':<4} {'query':<34} 轴(形式/阶段/动作/作用)         理由")
-    for r, v in zip(rows, verdicts):
-        kept += 1 if v["keep"] else 0
-        mark = "✓留" if v["keep"] else "✕扔"
-        axes = f"{r['形式']}/{r['阶段']}/{r['动作']}/{r['作用']}"
-        print(f"{mark:<4} {r['query']:<34} {axes:<22} {v['reason']}")
-    print(f"\n保留 {kept} / {len(rows)}（排除 {len(rows) - kept}）")
-
-
-if __name__ == "__main__":
-    main()

scripts/build_query_demo.py

@@ -1,51 +0,0 @@
-"""四种打法生成 query 的 demo：只到「生成 query」，写 data/queries/demo.json（不落库）。
-
-用法：PYTHONPATH=. python scripts/build_query_demo.py
-③ 联想 + ④ LLM 走真实接口，须在能访问 crawler.aiddit.com / OpenRouter 的环境跑。
-"""
-from __future__ import annotations
-
-import json
-import time
-from pathlib import Path
-
-from acquisition.query import (
-    ACTIONS, BLOCK, CARRIER_POS_GROUPED, DEMAND, KTYPE_SUFFIX, MODALITIES, STAGES,
-    sample_nodes, tactic2_form_llm, tactic4_llm, tactic_multiaxis,
-)
-from core.config import Settings
-
-ROOT = Path(__file__).resolve().parent.parent
-OUT = ROOT / "data" / "queries" / "demo.json"
-
-
-def main() -> None:
-    settings = Settings.from_env()
-    topics = sample_nodes("实质", depths=(3,), limit=12, under="理念")   # 实质：理念分支(事件/知识/情感…)
-    forms = sample_nodes("形式", depths=(4,), limit=20, under="架构")    # 形式：架构分支(叙事/修辞/创意…)
-    print(f"实质 {len(topics)}: {topics}")
-    print(f"形式 {len(forms)}: {forms}")
-
-    t2 = tactic2_form_llm(forms, settings)
-    # ③ 暂留空：小红书/抖音无真·搜索联想接口（keyword_v2 返回的是帖子不是联想词），待接入真 sug 再补
-    print(f"① LLM 实质: {topics[:5]}")
-    t4 = tactic4_llm(topics[:5], settings)
-    t5 = tactic_multiaxis(n=36)   # ④ 机械多轴组合（无 LLM）
-
-    data = {
-        "generated_at": int(time.time()),
-        "block": BLOCK,
-        "tactic1": {"name": "实质 × 创作阶段 × 需求", "需求点": DEMAND, "items": t4},
-        "tactic2": {"name": "形式 + 载体位置", "载体位置": CARRIER_POS_GROUPED, "items": t2},
-        "tactic3": {"name": "搜索词sug扩展", "items": [],
-                    "待接入": "小红书/抖音都没有真正的「输入框搜索联想」接口；keyword_v2 名字像联想、实际返回的是帖子。此页暂空，待接入真 sug 接口后再补。"},
-        "tactic4": {"name": "多轴正交组合", "items": t5,
-                    "人工轴": {"阶段": STAGES, "动作": ACTIONS, "知识类型": KTYPE_SUFFIX, "模态": MODALITIES}},
-    }
-    OUT.parent.mkdir(parents=True, exist_ok=True)
-    OUT.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
-    print(f"\n① {len(t4)} 条  ② {len(t2)} 条  ③ 待接入  ④ {len(t5)} 条(机械组合) → {OUT}")
-
-
-if __name__ == "__main__":
-    main()

tests/test_legacy_cleanup_contract.py

@@ -23,6 +23,24 @@ LEGACY_RUNTIME_TOKENS = [
     "import creation_knowledge",
 ]
 
+LEGACY_QUERY_TOKENS = [
+    "from acquisition.query",
+    "import acquisition.query",
+    "tactic2_form_llm",
+    "tactic3_suggest",
+    "tactic4_llm",
+    "tactic_multiaxis",
+    "tactic1",
+    "tactic2",
+    "tactic3",
+    "tactic4",
+    "build_query_demo",
+    "build_family6",
+    "query_gen.txt",
+    "form_query_gen.txt",
+    "data/queries/demo.json",
+]
+
 
 def _active_source_files():
     for root in FORMAL_SOURCE_ROOTS:
@@ -42,6 +60,20 @@ def test_formal_sources_do_not_depend_on_archived_legacy_runtime():
     assert offenders == []
 
 
+def test_formal_sources_do_not_depend_on_legacy_query_tactics():
+    offenders: list[str] = []
+    script_roots = [Path("scripts")]
+    for root in FORMAL_SOURCE_ROOTS + script_roots:
+        for path in root.rglob("*"):
+            if path.is_file() and path.suffix in {".py", ".js", ".jsx", ".ts", ".tsx"}:
+                text = path.read_text(encoding="utf-8")
+                for token in LEGACY_QUERY_TOKENS:
+                    if token in text:
+                        offenders.append(f"{path}:{token}")
+
+    assert offenders == []
+
+
 def test_legacy_debug_decompose_entry_is_explicitly_labeled():
     text = Path("scripts/decompose.py").read_text(encoding="utf-8")
 

tests/test_query_builder.py

@@ -2,8 +2,32 @@ from __future__ import annotations
 
 from uuid import uuid4
 
+import pytest
+
 from acquisition.domain import Query, QueryBatch
-from acquisition.queries.builder import persist_query_batch
+from acquisition.queries.builder import (
+    QueryBuildOptions,
+    build_creation_query_batch,
+    persist_query_batch,
+)
+from core.config import PgConfig, Settings
+
+
+def _settings() -> Settings:
+    return Settings(
+        pg=PgConfig(host="h", port=5432, user="u", password="p", database="d"),
+        aiddit_crawler_base_url="http://crawler.test",
+        crawler_timeout=30,
+        openrouter_timeout_seconds=90,
+        openrouter_model="m",
+        openrouter_base_url="http://openrouter.test",
+        openrouter_api_key="k",
+        llm_model="m",
+        max_cards=12,
+        frames_dir="f",
+        douyin_ratio="540p",
+        data_dir="",
+    )
 
 
 class FakeRepo:
@@ -20,10 +44,67 @@ class FakeRepo:
         return Query(id=uuid4(), **kwargs)
 
 
+def test_build_creation_query_batch_defaults_to_first_two_families():
+    generated = build_creation_query_batch(
+        _settings(),
+        options=QueryBuildOptions(per=2, batch_n=4, dry=True),
+    )
+
+    assert [family["key"] for family in generated["families"]] == ["f1", "f2"]
+    assert generated["metadata"]["active_family_keys"] == ["f1", "f2"]
+    assert sum(len(family["items"]) for family in generated["families"]) == 4
+
+
+def test_build_creation_query_batch_can_explicitly_enable_reserved_families():
+    all_keys = (
+        "f1",
+        "f2",
+        "f4",
+        "f3",
+        "f5",
+        "a_shi",
+        "a_xing",
+        "a_both",
+        "a_purpose",
+        "a_tail",
+        "b_shi",
+        "b_xing",
+        "b_both",
+        "b_purpose",
+        "b_tail",
+    )
+    generated = build_creation_query_batch(
+        _settings(),
+        options=QueryBuildOptions(
+            per=1,
+            batch_n=4,
+            dry=True,
+            active_family_keys=all_keys,
+        ),
+    )
+
+    assert [family["key"] for family in generated["families"]] == list(all_keys)
+    assert generated["metadata"]["active_family_keys"] == list(all_keys)
+
+
+def test_build_creation_query_batch_rejects_unknown_family_key():
+    with pytest.raises(ValueError, match="unknown query family"):
+        build_creation_query_batch(
+            _settings(),
+            options=QueryBuildOptions(
+                dry=True,
+                active_family_keys=("f1", "no_such_family"),
+            ),
+        )
+
+
 def test_persist_query_batch_writes_formal_batch_and_query_contract():
     repo = FakeRepo()
     generated = {
-        "metadata": {"query_filter_prompt_version": "abc123"},
+        "metadata": {
+            "query_filter_prompt_version": "abc123",
+            "active_family_keys": ["f1", "f2"],
+        },
         "families": [
             {
                 "key": "f1",
@@ -49,6 +130,7 @@ def test_persist_query_batch_writes_formal_batch_and_query_contract():
     assert count == 1
     assert repo.batch_kwargs["status"] == "ready"
     assert repo.batch_kwargs["target_platforms"] == ["xiaohongshu", "weixin", "douyin"]
+    assert repo.batch_kwargs["metadata"]["active_family_keys"] == ["f1", "f2"]
     row = repo.queries[0]
     assert row["batch_id"] == batch.id
     assert row["query_text"] == "反转 视频 脚本 怎么做"