"""创作知识 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 import httpx from core.config import Settings ROOT = Path(__file__).resolve().parent.parent TREES = ROOT / "scope_trees" / "trees_index.json" FILTER_PROMPT = ROOT / "acquisition" / "query_filter.txt" # 筛选词在 acquisition/ OUT = ROOT / "data" / "queries" / "creation_demo.json" PER = 20 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 _filter(queries, settings): """把一批原串喂 query_filter.txt(LLM 只做 keep/排除)。返回 [{keep,reason}] 对齐顺序。""" user = json.dumps([{"idx": i, "query": q} for i, q in enumerate(queries)], ensure_ascii=False) api = settings.openrouter_base_url.rstrip("/") + "/chat/completions" headers = {"Authorization": f"Bearer {settings.openrouter_api_key}", "Content-Type": "application/json"} body = {"model": settings.llm_model, "messages": [ {"role": "system", "content": FILTER_PROMPT.read_text("utf-8")}, {"role": "user", "content": user}], "response_format": {"type": "json_object"}} try: resp = httpx.post(api, headers=headers, json=body, timeout=120) resp.raise_for_status() txt = resp.json()["choices"][0]["message"]["content"] # query_filter 要求输出数组；有的模型会包一层 {"result":[...]}，都兜住 data = json.loads(txt) arr = data if isinstance(data, list) else next((v for v in data.values() if isinstance(v, list)), []) by = {d.get("idx"): d for d in arr if isinstance(d, dict)} return [{"keep": bool(by.get(i, {}).get("keep", True)), "reason": str(by.get(i, {}).get("reason", ""))[:50]} for i in range(len(queries))] except Exception as exc: return [{"keep": True, "reason": f"筛选失败:{str(exc)[:30]}"} for _ in queries] 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)}") def pick(seq): return rng.choice(seq) # 每家族：生成器 + 用到的轴(给前端标列) # 阶段=脊柱,每族必带;模态+知识类型固定收尾;前面配 实质/形式/目的 之一或组合 families = [ {"key": "f1", "name": "实质 × 阶段", "axes": ["实质", "阶段", "模态", "知识类型"], "gen": lambda: {"parts": {"实质": pick(SHI), "阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, {"key": "f2", "name": "形式 × 阶段", "axes": ["形式", "阶段", "模态", "知识类型"], "gen": lambda: {"parts": {"形式": pick(XING), "阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, {"key": "f3", "name": "实质 × 形式 × 阶段", "axes": ["实质", "形式", "阶段", "模态", "知识类型"], "gen": lambda: {"parts": {"实质": pick(SHI), "形式": pick(XING), "阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, {"key": "f4", "name": "(作用/感受/意图) × 阶段", "axes": ["作用/感受/意图", "阶段", "模态", "知识类型"], "gen": lambda: {"parts": {"目的": pick(POOL), "阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, {"key": "f5", "name": "纯阶段", "axes": ["阶段", "模态", "知识类型"], "gen": lambda: {"parts": {"阶段": pick(INTENT), "模态": pick(MODALITY), "知识类型": pick(KTYPE)}}}, ] # 各部件按固定顺序拼成原串（内容维度在前，阶段+模态+知识类型固定收尾） order = ["实质", "形式", "目的", "阶段", "模态", "知识类型"] # 阶段值存「分组结构」(STAGE_INTENT)，前端按 灵感/选题/脚本 分组+缩进展示；其余轴是扁平数组 out = {"axis_values": {"实质": SHI, "形式": XING, "目的池": POOL, "阶段": STAGE_INTENT, "模态": MODALITY, "知识类型": KTYPE}, "families": []} for fam in families: seen, items = set(), [] while len(items) < PER and len(seen) < PER * 40: parts = fam["gen"]()["parts"] q = " ".join(parts[k] for k in order if k in parts) if q in seen: continue seen.add(q) items.append({"query": q, "parts": parts}) verdicts = _filter([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"[{fam['name']}] 生成 {len(items)} 条, 筛后保留 {kept}") out["families"].append({"key": fam["key"], "name": fam["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()