|
|
@@ -0,0 +1,140 @@
|
|
|
+"""生成 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
|