"""创作知识解构引擎 v2：一帖 → N 颗（how/what/why，含组件颗）→ frameworks.json + payloads.json。 编排全流程，把 skill 的 phase 文档当 prompt 喂给 LLM（skill 是唯一真源）： ① 读懂：extractor 读图 → 完整文字（vision, OpenRouter） ② 判颗+类型闸+三lane成形+轻标签：system = phase1-frame.md ③ 作用域：system = phase2-scope.md → 候选 → scope_link 回扣（火山） ⑤ 组装：代码 → 每颗一个 ingest payload（按类型分拼） 须在云端跑（OpenRouter vision + 火山 可达）。用法：PYTHONPATH=. python scripts/decompose.py """ from __future__ import annotations import json from pathlib import Path from creation_knowledge.integrations.crawler import parse_detail_response from creation_knowledge.integrations.extractor import GeminiExtractor from creation_knowledge.integrations.llm import chat_json from scripts.scope_link import ScopeLinker ROOT = Path(__file__).resolve().parent.parent FIX = ROOT / "tests" / "fixtures" SKILL = ROOT / "创作知识提取-skill" PHASE1 = (SKILL / "extraction" / "phase1-frame.md").read_text(encoding="utf-8") PHASE2 = (SKILL / "extraction" / "phase2-scope.md").read_text(encoding="utf-8") CIDS = ["699308fa0000000016009697", "698481e1000000000a02a7c1", "67e2e39b0000000003028ff0", "680659e8000000001a007a11", "67e4bdf50000000006028a59"] SRC2CN = {"substance": "实质", "form": "形式", "feeling": "感受", "effect": "作用", "intent": "意图"} TYPE2ATTR = {"how": "how工序", "what": "what构成", "why": "why原理"} REUSE_THRESHOLD = 0.90 # ---------- ① 读懂 ---------- def read_post(cid: str, extractor: GeminiExtractor): resp = json.loads((FIX / f"xhs_case_{cid}.json").read_text("utf-8")) post = parse_detail_response(resp, fallback_content_id=cid) if not post.url: post.url = f"https://www.xiaohongshu.com/explore/{cid}" ec = extractor.extract(post) parts = [ec.text] if ec.from_image: parts.append("【图片要点】\n" + ec.from_image) parts += [f"【卡片{c.index}】{c.content}" for c in ec.cards if c.content] return post, "\n\n".join(p for p in parts if p) # ---------- ② 判颗+成形+轻标签 ---------- def shape(post, read: str) -> list[dict]: user = (f"原帖标题：{post.title or '(无)'}\n\n读懂后的完整内容：\n{read}\n\n" "按上面规则拆颗+判类型+成形+轻标签。作用域字段一律留空 []。" "只输出 JSON：{\"knowledges\":[ ... 见模板 ... ]}") out = chat_json(PHASE1, user, timeout=120) return out.get("knowledges") or [] # ---------- ③ 作用域候选 + 回扣 ---------- def _slim(knowledges: list[dict]) -> list[dict]: slim = [] for k in knowledges: e = {"id": k.get("id"), "type": k.get("type"), "title": k.get("title")} if k.get("type") == "how": e["steps"] = [{"id": s.get("id"), "intent": s.get("intent"), "directive": (s.get("directive") or "")[:500], "output": s.get("output")} for s in k.get("steps", [])] else: e["内容"] = {x: k.get(x) for x in ("界定", "构成", "主张", "依据", "对创作的影响") if k.get(x)} slim.append(e) return slim def scope_candidates(knowledges: list[dict]) -> dict: user = ("给下面每颗知识标作用域候选（how 逐步：每个 step 一组；what/why 颗级：整颗一组）。\n" "只输出 JSON：{\"scopes\":[{\"knowledge_id\":\"k1\",\"step_id\":\"s1\",\"items\":[{\"scope_type\":\"substance\",\"value\":\"赛道共识\"}]}," "{\"knowledge_id\":\"k2\",\"step_id\":null,\"items\":[...]}]}\n\n" + json.dumps(_slim(knowledges), ensure_ascii=False)) out = chat_json(PHASE2, user, timeout=120) return out.get("scopes") or [] def link_scope(linker: ScopeLinker, scope_type: str, value: str) -> dict: try: hits = linker.link(value, source_type=SRC2CN.get(scope_type, scope_type), top_k=3) except Exception: hits = [] top = hits[0] if hits else {} score = float(top.get("score", 0.0)) reuse = score >= REUSE_THRESHOLD and top.get("name") return {"scope_type": scope_type, "value": top["name"] if reuse else value, "candidate": value, "link": "复用" if reuse else "新建", "score": round(score, 4), "top": [{"name": h["name"], "score": h["score"], "path": h.get("path", "")} for h in hits]} def apply_scopes(knowledges: list[dict], scopes: list[dict], linker: ScopeLinker) -> None: by_k = {k.get("id"): k for k in knowledges} for sc in scopes: k = by_k.get(sc.get("knowledge_id")) if not k: continue linked = [link_scope(linker, it["scope_type"], it["value"]) for it in (sc.get("items") or []) if it.get("scope_type") and it.get("value")] if k.get("type") == "how" and sc.get("step_id"): for s in k.get("steps", []): if s.get("id") == sc["step_id"]: s["作用域"] = linked else: k["作用域"] = (k.get("作用域") or []) + linked # ---------- ⑤ 组装 payload ---------- def build_content(k: dict) -> str: t = k.get("type") if t == "how": lines = [f"目标：{k.get('purpose','')}"] for i, s in enumerate(k.get("steps", []), 1): lines += [f"步骤{i}（目的：{s.get('intent','')}）", f" 指引：{s.get('directive','')}", f" 产出：{s.get('output','')}"] return "\n".join(lines) if t == "what": lines = [f"界定：{k.get('界定','')}", "构成："] lines += [f"- {c.get('要素','')}：{c.get('说明','')}" for c in k.get("构成", [])] return "\n".join(lines) return f"主张：{k.get('主张','')}\n依据：{k.get('依据','')}\n对创作的影响：{k.get('对创作的影响','')}" def build_payload(post, k: dict, how_titles: dict | None = None) -> dict: how_titles = how_titles or {} t = k.get("type") scopes, seen = [], set() def add(lst): for sc in lst: key = (sc["scope_type"], sc["value"]) if key not in seen: seen.add(key); scopes.append({"scope_type": sc["scope_type"], "value": sc["value"]}) if t == "how": for s in k.get("steps", []): add(s.get("作用域", [])) else: add(k.get("作用域", [])) ext = [{"key": "业务阶段", "type": "str", "value": v} for v in (k.get("业务阶段") or [])] if t == "how": cs, cseen = [], set() for s in k.get("steps", []): c = s.get("创作阶段") if c and c not in cseen: cseen.add(c); cs.append(c) ext += [{"key": "创作阶段", "type": "str", "value": v} for v in cs] ext += [{"key": "动作", "type": "str", "value": s["动作"]} for s in k.get("steps", []) if s.get("动作")] if k.get("role") == "组件" and k.get("parent"): p = k["parent"] ext.append({"key": "出自", "type": "str", "value": f"{how_titles.get(p.get('how_id'), p.get('how_id'))} 第{p.get('step')}步"}) return {"source": {"id": f"xhs_{post.content_id}", "source_type": "post", "title": post.title or "", "author": post.author_name or "", "source_metadata": {"platform": post.platform, "url": post.url}}, "title": k.get("title"), "content": build_content(k), "dim_creations": ["创作"], "dim_attributes": [TYPE2ATTR.get(t, "how工序")], "scopes": scopes, "custom_ext": ext} def main() -> None: extractor = GeminiExtractor.from_env() linker = ScopeLinker() posts_out, payloads = [], [] for cid in CIDS: print(f"\n=== {cid[:8]} ===") post, read = read_post(cid, extractor) print(f" ① 读懂 {len(read)} 字") knowledges = shape(post, read) # 守卫：组件颗 parent 必须指向同帖一个 how，否则降级为 orphan 主颗 how_ids = {k.get("id") for k in knowledges if k.get("type") == "how"} how_titles = {k.get("id"): k.get("title") for k in knowledges if k.get("type") == "how"} for k in knowledges: if k.get("role") == "组件" and (k.get("parent") or {}).get("how_id") not in how_ids: k["role"] = "主" k["parent"] = None print(f" ② {len(knowledges)} 颗：" + ", ".join(f"{k.get('type')}/{k.get('role')}" for k in knowledges)) apply_scopes(knowledges, scope_candidates(knowledges), linker) print(" ③⑤ 作用域回扣 + 组装") posts_out.append({"post_id": cid, "title": post.title or "", "platform": post.platform, "url": post.url, "img_base": f"/data/demo/xiaohongshu/xhs_{cid}", "img_count": len(post.image_urls), "knowledges": knowledges}) payloads += [build_payload(post, k, how_titles) for k in knowledges] (ROOT / "web/frameworks.json").write_text( json.dumps({"count": len(posts_out), "posts": posts_out}, ensure_ascii=False, indent=1), encoding="utf-8") (ROOT / "web/payloads.json").write_text( json.dumps(payloads, ensure_ascii=False, indent=2), encoding="utf-8") print(f"\nwrote {len(posts_out)} posts, {len(payloads)} payloads") if __name__ == "__main__": main()