|
|
@@ -0,0 +1,127 @@
|
|
|
+"""创作知识 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
|
|
|
+STAGES = ["灵感", "选题", "脚本"]
|
|
|
+KTYPE = ["怎么做", "有哪些", "为什么"]
|
|
|
+
|
|
|
+
|
|
|
+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 _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 = _leaves(idx, "实质", under="理念")
|
|
|
+ XING = _leaves(idx, "形式", under="架构")
|
|
|
+ POOL = _leaves(idx, "作用") + _leaves(idx, "感受") + _leaves(idx, "意图")
|
|
|
+ print(f"实质 {len(SHI)} / 形式 {len(XING)} / 目的池 {len(POOL)} / 阶段 {len(STAGES)}")
|
|
|
+
|
|
|
+ def pick(seq):
|
|
|
+ return rng.choice(seq)
|
|
|
+
|
|
|
+ # 每家族：生成器 + 用到的轴(给前端标列)
|
|
|
+ families = [
|
|
|
+ {"key": "f1", "name": "实质 × 形式", "axes": ["实质", "形式", "知识类型"],
|
|
|
+ "gen": lambda: {"parts": {"实质": pick(SHI), "形式": pick(XING), "知识类型": pick(KTYPE)}}},
|
|
|
+ {"key": "f2", "name": "实质 × 目的池", "axes": ["实质", "作用/感受/意图", "知识类型"],
|
|
|
+ "gen": lambda: {"parts": {"实质": pick(SHI), "目的": pick(POOL), "知识类型": pick(KTYPE)}}},
|
|
|
+ {"key": "f3", "name": "纯形式", "axes": ["形式", "知识类型"],
|
|
|
+ "gen": lambda: {"parts": {"形式": pick(XING), "知识类型": pick(KTYPE)}}},
|
|
|
+ {"key": "f4", "name": "形式 × 目的池", "axes": ["形式", "作用/感受/意图", "知识类型"],
|
|
|
+ "gen": lambda: {"parts": {"形式": pick(XING), "目的": pick(POOL), "知识类型": pick(KTYPE)}}},
|
|
|
+ {"key": "f5", "name": "创作阶段 × 实质", "axes": ["实质", "创作阶段", "知识类型"],
|
|
|
+ "gen": lambda: {"parts": {"实质": pick(SHI), "创作阶段": pick(STAGES), "知识类型": pick(KTYPE)}}},
|
|
|
+ ]
|
|
|
+ # 各部件按固定顺序拼成原串
|
|
|
+ order = ["实质", "形式", "目的", "创作阶段", "知识类型"]
|
|
|
+
|
|
|
+ out = {"axis_values": {"实质": SHI, "形式": XING, "目的池": POOL, "创作阶段": STAGES, "知识类型": 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()
|
SamLee