"""创作知识 query 正交 demo：5 套家族机械正交 → LLM 只做排除(query_filter.txt) → 存 JSON。 不真实搜，只产 query 供前端看。轴严格取分类树的"创作支"： 实质 = 实质树·理念支(排除表象) 形式 = 形式树·架构支(排除呈现) 目的池 = 作用树 + 感受树 + 意图树 全部合一(随机取) 阶段意图 = 灵感/选题/脚本 展开成创作者真会搜的词(选题/开头/钩子/标题/封面/文案…)——脊柱，每族必带 模态 = 视频/图片 知识类型 = 怎么做/有哪些/为什么 脊柱(每条都带)：… 阶段意图 + 模态 + 知识类型；前面配 实质/形式/目的 之一或组合。 5 家族：① 实质×阶段 ② 形式×阶段 ③ 实质×形式×阶段 ④ 目的池×阶段 ⑤ 纯阶段，各 20 条。 每条原串过 query_filter.txt(keep/排除)，存 keep+reason 供前端展示。 用法：PYTHONPATH=. CK_ENV_FILE=.env python scripts/build_creation_demo.py """ from __future__ import annotations import json import random from pathlib import Path from acquisition.query import ACTIONS, _nonleaf_d4 # 第六家族(老正交)用的动作轴 + 4级非叶子取节点 from acquisition.query import STAGES as OLD_STAGES # 老的裸阶段 灵感/选题/脚本 from acquisition.query_filter import filter_queries # 共享筛选器(query_filter.txt) from core.config import Settings ROOT = Path(__file__).resolve().parent.parent TREES = ROOT / "scope_trees" / "trees_index.json" OUT = ROOT / "data" / "queries" / "creation_demo.json" PER = 20 BATCH_N = 16 # 全 demo 统一抽这么多个「实质 / 形式」，各族共用同一批，方便切页签比较 KTYPE = ["怎么做", "有哪些", "为什么"] MODALITY = ["视频", "图片"] # 被创作内容的形态(与教学帖本身格式无关),正交进所有家族 # 创作阶段意图轴(脊柱):灵感/选题/脚本 展开成创作者真会搜的词(query构造.md)。真实数据里 # "脚本"0次、但 开头/钩子/标题/封面/选题 各几十次——故用展开词,不用三个干阶段词。 STAGE_INTENT = { "灵感": ["找素材", "内容方向", "案例拆解", "灵感", "拆解", "复盘"], "选题": ["选题", "爆款选题", "选题方向"], "脚本": ["脚本", "文案", "开头", "钩子", "结构", "标题", "封面", "结尾"], } INTENT = [w for ws in STAGE_INTENT.values() for w in ws] # 扁平成一个池,随机取 def _segs(p): return [x for x in (p or "").split("/") if x] def _leaves(idx, source_type, under=None): """某树某支下的叶子节点名(没有更深子节点的=元素层)。under 限定分支。""" paths = [(_segs(n["path"]), n.get("name")) for n in idx if n.get("source_type") == source_type] if under: paths = [(s, nm) for s, nm in paths if under in s] allp = {"/".join(s) for s, _ in paths} out, seen = [], set() for s, nm in paths: if len(s) < 2: continue full = "/".join(s) is_leaf = not any(o != full and o.startswith(full + "/") for o in allp) name = nm or s[-1] if is_leaf and name and name not in seen: seen.add(name) out.append(name) return out def _nonleaf(idx, source_type, depths=(3, 4), under=None): """某树某支下、指定层级的【非叶子"类目"节点】(底下还有元素,不取元素本身)。 对齐制作侧取法:实质/形式 取 depth 3-4 的类目层,而非最深的元素层。""" paths = [(_segs(n["path"]), n.get("name")) for n in idx if n.get("source_type") == source_type] if under: paths = [(s, nm) for s, nm in paths if under in s] allp = {"/".join(s) for s, _ in paths} out, seen = [], set() for s, nm in paths: if len(s) not in depths: continue full = "/".join(s) is_nonleaf = any(o != full and o.startswith(full + "/") for o in allp) name = nm or (s[-1] if s else "") if is_nonleaf and name and name not in seen: seen.add(name) out.append(name) return out def main(): settings = Settings.from_env() rng = random.Random(7) idx = json.loads(TREES.read_text("utf-8")) SHI = _nonleaf(idx, "实质", depths=(3, 4), under="理念") # 类目层,非元素 XING = _nonleaf(idx, "形式", depths=(3, 4), under="架构") # 类目层,非元素 POOL = _leaves(idx, "作用") + _leaves(idx, "感受") + _leaves(idx, "意图") print(f"实质 {len(SHI)} / 形式 {len(XING)} / 目的池 {len(POOL)} / 业务阶段 {len(INTENT)}") # 全 demo 统一「一批实质 / 一批形式」——各家族都取同一批、且顺序一致，方便切页签横向比较 SHI_BATCH = rng.sample(SHI, min(BATCH_N, len(SHI))) XING_BATCH = rng.sample(XING, min(BATCH_N, len(XING))) print(f"统一批: 实质×{len(SHI_BATCH)} 形式×{len(XING_BATCH)}") def pick(seq): return rng.choice(seq) def shi(i): # 按 query 序号轮转,保证每族都覆盖整批、首次出现顺序一致 return SHI_BATCH[i % len(SHI_BATCH)] def xing(i): return XING_BATCH[i % len(XING_BATCH)] # 第六家族(最老正交方案·去实质)：形式 × 阶段 × 动作 × 作用 × 知识类型。 # 形式/作用 取分类树4级非叶子,阶段=灵感/选题/脚本,动作=ACTIONS(+无动作变体),query 形如「叙事组织 脚本撰写 趣味互动 有哪些」 F6_FORMS = _nonleaf_d4("形式", 10, under="架构") F6_ZY = _nonleaf_d4("作用", 10) F6_STAGE_ACT = [(s, a) for s in OLD_STAGES for a in ACTIONS] + [("", "")] # +「无动作」=老方案的 / def gen6(i): f_ = F6_FORMS[i % len(F6_FORMS)] st, ac = F6_STAGE_ACT[i % len(F6_STAGE_ACT)] zy_ = F6_ZY[i % len(F6_ZY)] suf = KTYPE[i % len(KTYPE)] seg = (st + ac) if ac else "" # 脚本撰写 / 空(动作=/ 时连阶段一起省) q = " ".join([f_] + ([seg] if seg else []) + [zy_, suf]) return {"parts": {"形式": f_, "阶段": st or "/", "动作": ac or "/", "作用": zy_, "知识类型": suf}, "query": q} # 每家族：生成器 + 用到的轴。家族名 = axes 用「×」连接，直接反映正交结构(见下方循环)。 # axes 顺序即前端列顺序；业务阶段=脊柱每族必带，模态+知识类型收尾 families = [ {"key": "f1", "axes": ["实质", "模态", "业务阶段", "知识类型"], "gen": lambda i: {"parts": {"实质": shi(i), "业务阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, {"key": "f2", "axes": ["形式", "模态", "业务阶段", "知识类型"], "gen": lambda i: {"parts": {"形式": xing(i), "业务阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, {"key": "f4", "axes": ["作用/感受/意图", "模态", "业务阶段", "知识类型"], "gen": lambda i: {"parts": {"目的": pick(POOL), "业务阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, {"key": "f3", "axes": ["实质", "形式", "模态", "业务阶段", "知识类型"], "gen": lambda i: {"parts": {"实质": shi(i), "形式": xing(i), "业务阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, {"key": "f5", "axes": ["模态", "业务阶段", "知识类型"], "gen": lambda i: {"parts": {"业务阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, {"key": "f6", "axes": ["形式", "阶段", "动作", "作用", "知识类型"], "gen": gen6}, ] # 各部件按固定顺序拼成原串（内容维度在前，模态贴题材后，业务阶段+知识类型收尾） order = ["实质", "形式", "目的", "模态", "业务阶段", "知识类型"] # 业务阶段值存「分组结构」(STAGE_INTENT)，前端按 灵感/选题/脚本 分组+缩进展示；其余轴是扁平数组 out = {"axis_values": {"实质": SHI, "形式": XING, "目的池": POOL, "业务阶段": STAGE_INTENT, "模态": MODALITY, "知识类型": KTYPE, "阶段": OLD_STAGES, "动作": ACTIONS, "作用": F6_ZY}, # 第六家族的老轴 "families": []} for fam in families: name = " × ".join(fam["axes"]) # 家族名直接反映正交结构 seen, items = set(), [] while len(items) < PER and len(seen) < PER * 40: g = fam["gen"](len(items)) # 序号轮转实质/形式批 parts = g["parts"] q = g.get("query") or " ".join(parts[k] for k in order if k in parts) # f6 自带 query 串 if q in seen: continue seen.add(q) items.append({"query": q, "parts": parts}) verdicts = filter_queries([it["query"] for it in items], settings) for it, v in zip(items, verdicts): it.update(v) kept = sum(1 for it in items if it["keep"]) print(f"[{name}] 生成 {len(items)} 条, 筛后保留 {kept}") out["families"].append({"key": fam["key"], "name": name, "axes": fam["axes"], "items": items}) OUT.parent.mkdir(parents=True, exist_ok=True) OUT.write_text(json.dumps(out, ensure_ascii=False, indent=1), encoding="utf-8") print(f"→ {OUT}") if __name__ == "__main__": main()