M4 四环节：筛选/拆分/解构/组装 + LLM 客户端 + 提示词

- integrations/llm.py chat_json：OpenRouter claude-sonnet-4.5，JSON self-repair 重试
- jsonio.py 共用 JSON 提取；extractor 改用之
- stages/{screen,split,deconstruct,assemble}；prompts/{screen,split,deconstruct}.txt(v1)
- tests/test_stages.py(离线5)；scripts/smoke_stages.py(真机)
- 离线 16 passed；真机 claude 全链路跑通(含 JSON self-repair)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

creation_knowledge/integrations/extractor.py

@@ -8,12 +8,12 @@
 """
 from __future__ import annotations
 
-import json
 from typing import Any, Callable, Mapping, Optional
 
 import httpx
 
 from creation_knowledge.config import load_env_file
+from creation_knowledge.jsonio import extract_json_object, to_bool
 from creation_knowledge.models import ExtractedContent, Post
 
 DEFAULT_MODEL = "google/gemini-3-flash-preview"
@@ -43,25 +43,6 @@ class ExtractorError(RuntimeError):
     pass
 
 
-def _strip_to_json(text: str) -> dict:
-    """从模型输出里取出 JSON 对象（容忍 ```json fences 或前后多余文本）。"""
-    s = text.strip()
-    if s.startswith("```"):
-        s = s.split("```", 2)[1] if s.count("```") >= 2 else s.strip("`")
-        if s.lstrip().lower().startswith("json"):
-            s = s.lstrip()[4:]
-    start, end = s.find("{"), s.rfind("}")
-    if start == -1 or end == -1 or end < start:
-        raise ExtractorError(f"no json object in model output: {text[:120]!r}")
-    return json.loads(s[start : end + 1])
-
-
-def _to_bool(value: Any) -> bool:
-    if isinstance(value, bool):
-        return value
-    return str(value).strip().lower() in ("1", "true", "yes", "是")
-
-
 class GeminiExtractor:
     def __init__(
         self,
@@ -125,12 +106,12 @@ class GeminiExtractor:
                 )
                 resp.raise_for_status()
                 content = resp.json()["choices"][0]["message"]["content"]
-                data = _strip_to_json(content)
+                data = extract_json_object(content)
                 return ExtractedContent(
                     text=str(data.get("text") or ""),
                     from_image=str(data.get("from_image") or ""),
                     from_video=str(data.get("from_video") or ""),
-                    is_empty=_to_bool(data.get("is_empty")),
+                    is_empty=to_bool(data.get("is_empty")),
                 )
             except httpx.HTTPError as exc:
                 last_exc = exc

creation_knowledge/integrations/llm.py

@@ -0,0 +1,81 @@
+"""文本 LLM：判断 / 拆分 / 解构用，走 OpenRouter /chat/completions（默认 claude-sonnet-4.5）。
+
+只做一件事：给 system + user，返回解析好的 JSON dict。HTTP/鉴权风格同 extractor。
+"""
+from __future__ import annotations
+
+from typing import Any, Callable, Optional
+
+import httpx
+
+from creation_knowledge.config import Settings
+from creation_knowledge.jsonio import extract_json_object
+
+
+class LLMError(RuntimeError):
+    pass
+
+
+def chat_json(
+    system: str,
+    user: str,
+    *,
+    model: Optional[str] = None,
+    settings: Optional[Settings] = None,
+    http_post: Callable[..., Any] = httpx.post,
+    env_file: str = ".env",
+    timeout: float = 60.0,
+) -> dict:
+    """调一次对话，强约束输出 JSON，返回解析后的 dict。带一次重试。"""
+    settings = settings or Settings.from_env(env_file)
+    api_key = settings.openrouter_api_key
+    if not api_key:
+        raise LLMError("missing OPENROUTER_API_KEY")
+    model = model or settings.llm_model
+    messages = [
+        {"role": "system", "content": system + (
+            "\n只输出一个严格合法的 JSON 对象，不要解释或 markdown。"
+            "字符串值要写在一行内，内部的换行写成 \\n、双引号写成 \\\"，不要出现裸换行或裸双引号。"
+        )},
+        {"role": "user", "content": user},
+    ]
+    url = f"{settings.openrouter_base_url.rstrip('/')}/chat/completions"
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    last_exc: Optional[Exception] = None
+    # 最多 3 轮：HTTP 错误重试；JSON 不合法则把坏输出喂回去做 self-repair
+    for attempt in range(3):
+        try:
+            resp = http_post(url, headers=headers,
+                             json={"model": model, "messages": messages}, timeout=timeout)
+            resp.raise_for_status()
+            content = resp.json()["choices"][0]["message"]["content"]
+        except httpx.HTTPError as exc:
+            last_exc = exc
+            if attempt < 2:
+                continue
+            raise LLMError(f"llm_http_error: {exc}") from exc
+        except (KeyError, IndexError, TypeError) as exc:
+            raise LLMError(f"llm_response_invalid: {exc}") from exc
+        try:
+            return extract_json_object(content)
+        except ValueError as exc:
+            last_exc = exc
+            messages = messages + [
+                {"role": "assistant", "content": content},
+                {"role": "user", "content": (
+                    "上面的输出不是严格合法的 JSON。请只重新输出严格合法的 JSON，"
+                    "字符串内的换行写成 \\n、双引号写成 \\\"，不要任何解释或 markdown。"
+                )},
+            ]
+    raise LLMError(f"llm_json_unrepairable: {last_exc}")
+
+
+# 默认对话器类型：(system, user) -> dict。stage 可注入假实现做离线测试。
+ChatFn = Callable[[str, str], dict]
+
+
+def default_chat(env_file: str = ".env", model: Optional[str] = None) -> ChatFn:
+    settings = Settings.from_env(env_file)
+    return lambda system, user: chat_json(
+        system, user, model=model, settings=settings
+    )

creation_knowledge/jsonio.py

@@ -0,0 +1,22 @@
+"""从 LLM 文本输出里稳健地取出 JSON。各 integration / stage 共用，避免重复。"""
+from __future__ import annotations
+
+import json
+from typing import Any
+
+
+def extract_json_object(text: str) -> dict:
+    """取第一个 { 到最后一个 } 的切片解析为 dict。
+
+    天然容忍 ```json fences、前后多余说明文字（它们都在大括号之外）。
+    """
+    start, end = text.find("{"), text.rfind("}")
+    if start == -1 or end == -1 or end < start:
+        raise ValueError(f"no json object in output: {text[:120]!r}")
+    return json.loads(text[start : end + 1])
+
+
+def to_bool(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    return str(value).strip().lower() in ("1", "true", "yes", "是")

creation_knowledge/prompts.py

@@ -0,0 +1,13 @@
+"""提示词加载：从 prompts/<name>.txt 读取模板，带版本号（便于 A/B 与审计回溯）。"""
+from __future__ import annotations
+
+from pathlib import Path
+
+PROMPTS_DIR = Path(__file__).resolve().parent.parent / "prompts"
+
+# 提示词版本：改提示词时手动 +1，写进 ck 记录便于回溯
+PROMPT_VERSION = "v1"
+
+
+def load_prompt(name: str) -> str:
+    return (PROMPTS_DIR / f"{name}.txt").read_text(encoding="utf-8")

creation_knowledge/stages/__init__.py

@@ -0,0 +1,7 @@
+"""四个环节：screen / split / deconstruct / assemble。"""
+from creation_knowledge.stages.assemble import build_ingest_payload
+from creation_knowledge.stages.deconstruct import deconstruct_item
+from creation_knowledge.stages.screen import screen_post
+from creation_knowledge.stages.split import split_post
+
+__all__ = ["screen_post", "split_post", "deconstruct_item", "build_ingest_payload"]

creation_knowledge/stages/_common.py

@@ -0,0 +1,38 @@
+"""stage 共用小工具。"""
+from __future__ import annotations
+
+from typing import Optional
+
+from creation_knowledge.models import ExtractedContent, KnowledgeType, Post
+
+VALID_TYPES = {"what", "why", "how"}
+_EMPTY = {"", "null", "none", "无", "n/a", "na"}
+
+
+def content_for_llm(post: Post, content: ExtractedContent) -> str:
+    """喂给 LLM 的内容 = 多模态提取结果（不是 body_text）。"""
+    parts = []
+    if content.text:
+        parts.append(content.text)
+    if content.from_image:
+        parts.append("【图片】" + content.from_image)
+    if content.from_video:
+        parts.append("【视频】" + content.from_video)
+    return "\n".join(parts) or post.body_text
+
+
+def norm_text(value) -> Optional[str]:
+    """把 'null' / '' / '无' 之类归一成 None。"""
+    if value is None:
+        return None
+    s = str(value).strip()
+    return None if s.lower() in _EMPTY else s
+
+
+def norm_types(values) -> list[KnowledgeType]:
+    out: list[KnowledgeType] = []
+    for v in values or []:
+        t = str(v).strip().lower()
+        if t in VALID_TYPES and t not in out:
+            out.append(t)  # type: ignore[arg-type]
+    return out

creation_knowledge/stages/assemble.py

@@ -0,0 +1,51 @@
+"""组装：把帖子+知识片段+解构，组装成 ingest 请求体（对齐 创作知识-重构设计.md §6）。
+
+纯函数，无 LLM、无 IO。同一帖子拆出的多条知识复用同一 source.id。
+"""
+from __future__ import annotations
+
+import json
+
+from creation_knowledge.models import (
+    Deconstruction,
+    IngestPayload,
+    KnowledgeItem,
+    Post,
+)
+
+
+def build_ingest_payload(
+    post: Post, item: KnowledgeItem, deco: Deconstruction
+) -> IngestPayload:
+    content = json.dumps(
+        {"what": item.what, "why": item.why, "how": item.how},
+        ensure_ascii=False,
+    )
+    custom_ext: list[dict] = []
+    if item.evidence:
+        custom_ext.append(
+            {"key": "原文证据", "type": "str", "value": " / ".join(item.evidence)}
+        )
+    if deco.stage_reason:
+        custom_ext.append(
+            {"key": "阶段判断理由", "type": "str", "value": deco.stage_reason}
+        )
+    if deco.scope_reason:
+        custom_ext.append(
+            {"key": "作用域判断理由", "type": "str", "value": deco.scope_reason}
+        )
+    return IngestPayload(
+        source={
+            "id": post.id,
+            "source_type": "post",
+            "title": post.title,
+            "author": post.author_name,
+            "source_metadata": {"platform": post.platform, "url": post.url},
+        },
+        title=item.title,
+        content=content,
+        dim_attributes=list(item.knowledge_types),
+        dim_creations=list(deco.stages),
+        scopes=[{"scope_type": s.scope_type, "value": s.value} for s in deco.scopes],
+        custom_ext=custom_ext,
+    )

creation_knowledge/stages/deconstruct.py

@@ -0,0 +1,38 @@
+"""解构：给知识片段补上阶段（灵感/选题/脚本）和作用域（五棵树）。"""
+from __future__ import annotations
+
+from typing import Optional
+
+from creation_knowledge.integrations.llm import ChatFn, default_chat
+from creation_knowledge.models import Deconstruction, KnowledgeItem, Scope
+from creation_knowledge.prompts import load_prompt
+
+SYSTEM = "你是严谨的创作知识解构器，按阶段和五棵分类树归类。"
+
+VALID_STAGES = {"灵感", "选题", "脚本"}
+VALID_SCOPES = {"substance", "form", "feeling", "effect", "intent"}
+
+
+def deconstruct_item(
+    item: KnowledgeItem,
+    *,
+    chat: Optional[ChatFn] = None,
+    env_file: str = ".env",
+) -> Deconstruction:
+    chat = chat or default_chat(env_file)
+    user = load_prompt("deconstruct").format(
+        item=item.model_dump_json(indent=2)
+    )
+    data = chat(SYSTEM, user)
+    stages = [s for s in (data.get("stages") or []) if s in VALID_STAGES]
+    scopes = [
+        Scope(scope_type=s["scope_type"], value=str(s["value"]).strip())
+        for s in (data.get("scopes") or [])
+        if isinstance(s, dict) and s.get("scope_type") in VALID_SCOPES and s.get("value")
+    ]
+    return Deconstruction(
+        stages=stages,
+        scopes=scopes,
+        stage_reason=str(data.get("stage_reason") or ""),
+        scope_reason=str(data.get("scope_reason") or ""),
+    )

creation_knowledge/stages/screen.py

@@ -0,0 +1,33 @@
+"""筛选：判断帖子是否值得提取为创作知识。"""
+from __future__ import annotations
+
+from typing import Optional
+
+from creation_knowledge.integrations.llm import ChatFn, default_chat
+from creation_knowledge.jsonio import to_bool
+from creation_knowledge.models import ExtractedContent, Post, ScreeningResult
+from creation_knowledge.prompts import load_prompt
+from creation_knowledge.stages._common import content_for_llm
+
+SYSTEM = "你是严谨的创作知识筛选器，只认能指导内容创作的知识，作品本身不算。"
+
+
+def screen_post(
+    post: Post,
+    content: ExtractedContent,
+    *,
+    chat: Optional[ChatFn] = None,
+    env_file: str = ".env",
+) -> ScreeningResult:
+    chat = chat or default_chat(env_file)
+    user = load_prompt("screen").format(
+        title=post.title or "(无)",
+        topics="、".join(post.topic_list) or "(无)",
+        content=content_for_llm(post, content) or "(空)",
+    )
+    data = chat(SYSTEM, user)
+    return ScreeningResult(
+        passed=to_bool(data.get("passed")),
+        score=int(data.get("score") or 0),
+        reason=str(data.get("reason") or ""),
+    )

creation_knowledge/stages/split.py

@@ -0,0 +1,47 @@
+"""拆分：把帖子拆成一个或多个 What/Why/How 知识片段。"""
+from __future__ import annotations
+
+from typing import Optional
+
+from creation_knowledge.integrations.llm import ChatFn, default_chat
+from creation_knowledge.models import ExtractedContent, KnowledgeItem, Post
+from creation_knowledge.prompts import load_prompt
+from creation_knowledge.stages._common import content_for_llm, norm_text, norm_types
+
+SYSTEM = "你是严谨的创作知识拆分器，原文有什么提什么，绝不编造。"
+
+
+def _to_item(raw: dict) -> Optional[KnowledgeItem]:
+    what = norm_text(raw.get("what"))
+    why = norm_text(raw.get("why"))
+    how = norm_text(raw.get("how"))
+    # knowledge_types 以非空字段为准（模型给的若不一致，用实际有内容的对齐）
+    types = norm_types(raw.get("knowledge_types"))
+    actual = [t for t, v in (("what", what), ("why", why), ("how", how)) if v]
+    types = [t for t in types if t in actual] or actual
+    if not types:
+        return None  # 三个都空，丢弃
+    title = norm_text(raw.get("title")) or (what or why or how or "")[:20]
+    evidence = [e for e in (raw.get("evidence") or []) if norm_text(e)]
+    return KnowledgeItem(
+        title=title, knowledge_types=types, what=what, why=why, how=how,
+        evidence=[str(e).strip() for e in evidence],
+    )
+
+
+def split_post(
+    post: Post,
+    content: ExtractedContent,
+    *,
+    chat: Optional[ChatFn] = None,
+    env_file: str = ".env",
+) -> list[KnowledgeItem]:
+    chat = chat or default_chat(env_file)
+    user = load_prompt("split").format(
+        title=post.title or "(无)",
+        topics="、".join(post.topic_list) or "(无)",
+        content=content_for_llm(post, content) or "(空)",
+    )
+    data = chat(SYSTEM, user)
+    items = [_to_item(raw) for raw in (data.get("items") or [])]
+    return [it for it in items if it is not None]

prompts/deconstruct.txt

@@ -0,0 +1,22 @@
+你在做"创作知识"解构：给一个知识片段补上【阶段】和【作用域】。
+
+阶段（创作 = 灵感 + 选题 + 脚本，可多选）：
+- 灵感：帮助发现方向、素材、切口、洞察
+- 选题：帮助判断写什么、拍什么、从哪个角度切入
+- 脚本：帮助组织表达顺序、文案、镜头、结构
+
+作用域（五棵分类树，可多选；value 要填具体标签，不是大类名）：
+- substance（实质）：内容真正讲的对象、主题、事实、问题
+- form（形式）：结构、模板、公式、表达形式
+- feeling（感受）：用户情绪、体验、氛围、感知
+- effect（作用）：解决什么创作问题、带来什么效果
+- intent（意图）：创作者目的、传播目的、商业目标
+
+知识片段：
+{item}
+
+输出 JSON：
+{{"stages": ["灵感" 和/或 "选题" 和/或 "脚本"],
+  "scopes": [{{"scope_type": "substance/form/feeling/effect/intent", "value": "具体标签"}}],
+  "stage_reason": "为什么是这些阶段",
+  "scope_reason": "为什么是这些作用域"}}

prompts/screen.txt

@@ -0,0 +1,15 @@
+你在做"创作知识"筛选：判断一篇帖子是否值得提取为能指导内容创作的知识。
+
+判断依据的是【多模态提取后的内容】，不是原始正文。通过需同时满足：
+1. 不是空内容（多模态提取后仍有有效内容）。
+2. 和"内容创作"相关（教人怎么选题/写脚本/起号/表达等），而不是一篇作品本身（如一段文案、一张图的出图提示词、一条纯展示内容）。
+3. 至少包含一点 What（是什么/有哪些）/ Why（为什么有效）/ How（怎么做）。
+4. 至少能拆出一个知识片段。
+
+帖子标题：{title}
+话题：{topics}
+多模态提取内容：
+{content}
+
+输出 JSON：
+{{"passed": true 或 false, "score": 0-10 的整数, "reason": "一句话理由"}}

prompts/split.txt

@@ -0,0 +1,25 @@
+你在做"创作知识"拆分：把一篇帖子拆成一个或多个知识片段。
+
+知识类型只有三种：
+- what：是什么 / 有哪些 / 由什么组成（类型、分类、要素、清单、特征、模板）
+- why：为什么这样做 / 为什么有效（原理、底层逻辑、机制、依据）
+- how：具体怎么做 / 怎么用（方法、步骤、技巧、流程、公式框架、实操）
+
+规则：
+1. 原文（含图片/视频提取内容）有哪个就提哪个，没有的填 null，绝不为结构完整而编造。
+2. knowledge_types 必须与非空的 what/why/how 完全一致。
+3. 若 what/why/how 在讲同一个知识对象，放进同一个片段；若有多个独立知识对象，拆成多个片段。
+4. evidence 填原文中支持该片段的句子（可多条）。
+
+帖子标题：{title}
+话题：{topics}
+多模态提取内容：
+{content}
+
+输出 JSON：
+{{"items": [
+  {{"title": "片段标题",
+    "knowledge_types": ["what" 和/或 "why" 和/或 "how"],
+    "what": "内容或 null", "why": "内容或 null", "how": "内容或 null",
+    "evidence": ["支撑句", "..."]}}
+]}}

scripts/smoke_stages.py

@@ -0,0 +1,55 @@
+"""M4 真实验收（省 Gemini 费）：用 海狸 帖的富文本 body_text 当提取内容，
+真调 claude-sonnet-4.5 跑 screen/split/deconstruct/assemble，打印结果。
+
+只产生文本 LLM 费用，不调 Gemini。须在能联网调 OpenRouter 的云端跑。
+用法：python scripts/smoke_stages.py [env_file] [content_id]
+"""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+from creation_knowledge.integrations.crawler import parse_detail_response
+from creation_knowledge.integrations.llm import default_chat
+from creation_knowledge.models import ExtractedContent
+from creation_knowledge.stages import (
+    build_ingest_payload,
+    deconstruct_item,
+    screen_post,
+    split_post,
+)
+
+FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures"
+
+
+def main() -> int:
+    args = sys.argv[1:]
+    env_file = args[0] if args and args[0].endswith(".env") else ".env"
+    cid = next((a for a in args if not a.endswith(".env")), "67e4bdf50000000006028a59")
+
+    resp = json.loads((FIXTURES / f"xhs_case_{cid}.json").read_text("utf-8"))
+    post = parse_detail_response(resp, fallback_content_id=cid)
+    # 用富文本 body_text 当已提取内容，避免再调 Gemini
+    content = ExtractedContent(text=post.body_text, is_empty=not post.body_text.strip())
+    chat = default_chat(env_file)
+
+    print(f"==== {post.id}  {post.title} ====\n")
+    scr = screen_post(post, content, chat=chat)
+    print(f"[screen] passed={scr.passed} score={scr.score} reason={scr.reason}")
+    if not scr.passed:
+        return 0
+
+    items = split_post(post, content, chat=chat)
+    print(f"[split] {len(items)} 个知识片段")
+    for i, item in enumerate(items, 1):
+        print(f"\n  片段{i}: {item.title}  types={item.knowledge_types}")
+        deco = deconstruct_item(item, chat=chat)
+        print(f"    stages={deco.stages}  scopes={[(s.scope_type, s.value) for s in deco.scopes]}")
+        payload = build_ingest_payload(post, item, deco)
+        print(f"    ingest.content={payload.content[:120]}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_stages.py

@@ -0,0 +1,103 @@
+"""M4 离线测试：注入假 chat，逐环节断言；assemble 纯函数对齐设计文档 §6。"""
+from __future__ import annotations
+
+import json
+
+from creation_knowledge.models import (
+    Deconstruction,
+    ExtractedContent,
+    KnowledgeItem,
+    Post,
+    Scope,
+)
+from creation_knowledge.stages import (
+    build_ingest_payload,
+    deconstruct_item,
+    screen_post,
+    split_post,
+)
+
+
+def _post() -> Post:
+    return Post(
+        id="xhs_abc", url="https://www.xiaohongshu.com/explore/abc", content_id="abc",
+        title="只要学会这几样，写短视频脚本真的不难", platform="xiaohongshu",
+        author_name="海狸教自媒体运营", topic_list=["短视频"],
+    )
+
+
+def _content() -> ExtractedContent:
+    return ExtractedContent(
+        text="短视频脚本包含标题、拍摄地点、分镜等要素",
+        from_image="九宫格图讲了脚本6要素", is_empty=False,
+    )
+
+
+def test_screen_passed():
+    r = screen_post(_post(), _content(),
+                    chat=lambda s, u: {"passed": True, "score": 8, "reason": "含明确 How"})
+    assert r.passed and r.score == 8
+
+
+def test_screen_feeds_extracted_not_body():
+    seen = {}
+
+    def chat(s, u):
+        seen["u"] = u
+        return {"passed": True, "score": 7, "reason": "x"}
+
+    screen_post(_post(), _content(), chat=chat)
+    assert "九宫格" in seen["u"]  # 多模态提取内容进了 prompt，而非空 body_text
+
+
+def test_split_consistency_and_drop_empty():
+    def chat(s, u):
+        return {"items": [
+            {"title": "脚本要素", "knowledge_types": ["what", "how", "why"],
+             "what": "脚本含6要素", "why": "null", "how": "逐个填要素",
+             "evidence": ["原句"]},
+            {"title": "空", "knowledge_types": ["what"],
+             "what": "null", "why": "null", "how": "null", "evidence": []},
+        ]}
+
+    items = split_post(_post(), _content(), chat=chat)
+    assert len(items) == 1  # 三个都空的片段被丢弃
+    it = items[0]
+    assert it.why is None  # "null" 归一为 None
+    assert set(it.knowledge_types) == {"what", "how"}  # 与非空字段对齐，why 被剔除
+
+
+def test_deconstruct_filters_invalid():
+    def chat(s, u):
+        return {"stages": ["选题", "脚本", "乱写"],
+                "scopes": [{"scope_type": "form", "value": "操作流程"},
+                           {"scope_type": "bad", "value": "x"},
+                           {"scope_type": "effect", "value": ""}],
+                "stage_reason": "r1", "scope_reason": "r2"}
+
+    d = deconstruct_item(KnowledgeItem(title="t", knowledge_types=["how"], how="做法"),
+                         chat=chat)
+    assert d.stages == ["选题", "脚本"]  # 非法阶段过滤
+    assert [s.scope_type for s in d.scopes] == ["form"]  # 非法 scope / 空 value 过滤
+
+
+def test_assemble_matches_design_doc():
+    item = KnowledgeItem(title="评论区选题法", knowledge_types=["what", "how"],
+                         what="从评论提取痛点", why=None, how="收集高赞评论改写",
+                         evidence=["原句a"])
+    deco = Deconstruction(
+        stages=["选题", "脚本"],
+        scopes=[Scope(scope_type="form", value="公式框架"),
+                Scope(scope_type="effect", value="生成选题")],
+        stage_reason="既选题又脚本", scope_reason="形式+作用")
+    p = build_ingest_payload(_post(), item, deco)
+
+    assert p.source["id"] == "xhs_abc"
+    assert p.source["source_type"] == "post"
+    assert p.source["source_metadata"]["platform"] == "xiaohongshu"
+    assert p.dim_attributes == ["what", "how"]
+    assert p.dim_creations == ["选题", "脚本"]
+    assert {"scope_type": "form", "value": "公式框架"} in p.scopes
+    assert json.loads(p.content) == {"what": "从评论提取痛点", "why": None,
+                                     "how": "收集高赞评论改写"}
+    assert "原文证据" in [e["key"] for e in p.custom_ext]