feat(decode): add framing scoping payload service

decode_content/framing.py

@@ -0,0 +1,138 @@
+"""Knowledge particle framing and type-shape cleanup."""
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from core.llm import chat_json as default_chat_json
+from core.models import Post
+from decode_content.contracts import BUSINESS_STAGES, CREATION_STAGES, WHAT_KINDS, SkillContract, load_contract
+from decode_content.gates import ChatJsonFn, how_gate, why_refute_gate
+
+
+CSTAGE = set(CREATION_STAGES)
+KINDS = set(WHAT_KINDS)
+KIND_FIX = {"多维度关系": "多维关系", "多维": "多维关系", "集合": "子集", "序列型": "序列"}
+
+
+def frame_knowledges(
+    post: Post,
+    read_text: str,
+    *,
+    contract: SkillContract | None = None,
+    chat_json_fn: ChatJsonFn = default_chat_json,
+) -> list[dict[str, Any]]:
+    contract = contract or load_contract()
+    user = (
+        f"原帖标题：{post.title or '(无)'}\n\n读懂后的完整内容：\n{read_text}\n\n"
+        "按上面规则拆颗+判类型+成形+轻标签。作用域字段一律留空 []。"
+        "只输出 JSON：{\"knowledges\":[ ... 见模板 ... ]}"
+    )
+    return chat_json_fn(contract.phase1, user, timeout=120).get("knowledges") or []
+
+
+def reshape_nonhow(
+    knowledge: dict[str, Any],
+    *,
+    contract: SkillContract | None = None,
+    chat_json_fn: ChatJsonFn = default_chat_json,
+) -> list[dict[str, Any]]:
+    contract = contract or load_contract()
+    body = f"目标：{knowledge.get('purpose', '')}\n" + "\n".join(
+        f"- 输入：{step.get('input', '')}｜方法：{step.get('directive', '')}｜产出：{step.get('output', '')}"
+        for step in knowledge.get("steps", [])
+    )
+    user = (
+        "【下面这块原被误判为 how 工序，实为「离散构成 / 原理」，请只拆成 What/Why 主颗——"
+        "每个'是什么/分几类'的构成块拆一颗 What，背后的原理/标准拆一颗 Why；"
+        "不要 how、不要组件颗，parent 一律 null。作用域字段留空 []。】\n\n"
+        f"原标题：{knowledge.get('title', '')}\n{body}\n\n"
+        "只输出 JSON：{\"knowledges\":[ ... 仅 what/why，见模板 ... ]}"
+    )
+    try:
+        out = chat_json_fn(contract.phase1, user, timeout=120).get("knowledges") or []
+    except Exception:
+        out = []
+    reshaped: list[dict[str, Any]] = []
+    for idx, item in enumerate(out, 1):
+        if item.get("type") == "how":
+            continue
+        item["id"] = f"{knowledge.get('id', 'k')}r{idx}"
+        item["role"], item["parent"] = "主", None
+        reshaped.append(item)
+    return reshaped or [knowledge]
+
+
+def fix_fake_hows(
+    knowledges: list[dict[str, Any]],
+    *,
+    contract: SkillContract | None = None,
+    chat_json_fn: ChatJsonFn = default_chat_json,
+) -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    for knowledge in knowledges:
+        if knowledge.get("type") == "how" and len(knowledge.get("steps", [])) >= 2:
+            gate = how_gate(knowledge, chat_json_fn=chat_json_fn)
+            if not gate.passed:
+                reshaped = reshape_nonhow(knowledge, contract=contract, chat_json_fn=chat_json_fn)
+                for item in reshaped:
+                    item.setdefault("metadata", {})["how_gate_reason"] = gate.reason
+                out.extend(reshaped)
+                continue
+        out.append(knowledge)
+    return out
+
+
+def drop_fake_whys(
+    knowledges: list[dict[str, Any]],
+    *,
+    chat_json_fn: ChatJsonFn = default_chat_json,
+) -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    for knowledge in knowledges:
+        if knowledge.get("type") == "why":
+            gate = why_refute_gate(knowledge, chat_json_fn=chat_json_fn)
+            if not gate.passed:
+                verdict = gate.details.get("verdict")
+                if verdict == "废话":
+                    continue
+                if verdict in {"实为what", "实为how"}:
+                    knowledge["inferred"] = True
+                    knowledge["inferred_reason"] = f"why闸疑似{verdict}：{gate.reason}"
+        out.append(knowledge)
+    return out
+
+
+def normalize_knowledge_shapes(knowledges: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    how_ids = {k.get("id") for k in knowledges if k.get("type") == "how"}
+    for knowledge in knowledges:
+        knowledge["业务阶段"] = [
+            stage for stage in (knowledge.get("业务阶段") or []) if stage in BUSINESS_STAGES
+        ]
+        if knowledge.get("type") == "what" and knowledge.get("kind") not in KINDS:
+            knowledge["kind"] = KIND_FIX.get((knowledge.get("kind") or "").strip(), knowledge.get("kind"))
+        for step in knowledge.get("steps", []):
+            if step.get("创作阶段") not in CSTAGE:
+                step["创作阶段"] = None
+        if knowledge.get("role") == "组件" and (knowledge.get("parent") or {}).get("how_id") not in how_ids:
+            knowledge["role"] = "主"
+            knowledge["parent"] = None
+    return knowledges
+
+
+def frame_and_clean(
+    post: Post,
+    read_text: str,
+    *,
+    contract: SkillContract | None = None,
+    chat_json_fn: ChatJsonFn = default_chat_json,
+) -> list[dict[str, Any]]:
+    contract = contract or load_contract()
+    knowledges = frame_knowledges(post, read_text, contract=contract, chat_json_fn=chat_json_fn)
+    knowledges = fix_fake_hows(knowledges, contract=contract, chat_json_fn=chat_json_fn)
+    knowledges = drop_fake_whys(knowledges, chat_json_fn=chat_json_fn)
+    return normalize_knowledge_shapes(knowledges)
+
+
+def knowledge_debug_json(knowledges: list[dict[str, Any]]) -> str:
+    return json.dumps(knowledges, ensure_ascii=False)

decode_content/gates.py

@@ -0,0 +1,123 @@
+"""Creation gates used before and during deep decode."""
+from __future__ import annotations
+
+import json
+from typing import Any, Callable
+
+from core.llm import chat_json as default_chat_json
+from core.prompts import load_prompt
+from decode_content.models import GateResult
+
+
+GATE_ADMIT = load_prompt("gate_admit")
+GATE_REFUTE = load_prompt("gate_refute")
+GATE_TIEBREAK = load_prompt("gate_tiebreak")
+GATE_HOW_ADMIT = load_prompt("gate_how_admit")
+GATE_HOW_REFUTE = load_prompt("gate_how_refute")
+GATE_HOW_TIEBREAK = load_prompt("gate_how_tiebreak")
+GATE_WHY_REFUTE = load_prompt("gate_why_refute")
+
+ChatJsonFn = Callable[..., dict[str, Any]]
+
+
+def vote_bool(
+    system: str,
+    user: str,
+    key: str,
+    *,
+    on_fail: bool,
+    chat_json_fn: ChatJsonFn = default_chat_json,
+    timeout: int = 90,
+) -> bool:
+    try:
+        return bool(chat_json_fn(system, user, timeout=timeout).get(key))
+    except Exception:
+        return on_fail
+
+
+def creation_gate(read_text: str, *, chat_json_fn: ChatJsonFn = default_chat_json) -> GateResult:
+    """Decide whether the read content is reusable creation knowledge."""
+
+    v_admit = vote_bool(GATE_ADMIT, read_text, "in_scope", on_fail=True, chat_json_fn=chat_json_fn)
+    v_refute = not vote_bool(GATE_REFUTE, read_text, "out_of_scope", on_fail=False, chat_json_fn=chat_json_fn)
+    if v_admit == v_refute:
+        return GateResult(
+            passed=v_admit,
+            reason=f"admit={v_admit}/refute={v_refute} 一致",
+            details={"admit": v_admit, "refute": v_refute},
+        )
+    v_tie = vote_bool(GATE_TIEBREAK, read_text, "in_scope", on_fail=False, chat_json_fn=chat_json_fn)
+    return GateResult(
+        passed=v_tie,
+        reason=f"admit={v_admit}/refute={v_refute} 分歧→裁决={v_tie}",
+        details={"admit": v_admit, "refute": v_refute, "tiebreak": v_tie},
+    )
+
+
+def chain_signals(steps: list[dict[str, Any]]) -> list[str]:
+    sigs: list[str] = []
+    outs = [(s.get("output") or "") for s in steps]
+    independent_count = 0
+    for idx, step in enumerate(steps):
+        if idx == 0:
+            continue
+        step_input = step.get("input") or ""
+        if not (("←" in step_input) or any(o and o[:4] in step_input for o in outs[:idx])):
+            independent_count += 1
+    if independent_count >= 2:
+        sigs.append(f"{independent_count} 个后步的 input 未指向前步产出（各自起头）")
+    for i in range(len(outs)):
+        for j in range(i + 1, len(outs)):
+            left, right = outs[i], outs[j]
+            if left and right and (left in right or right in left):
+                sigs.append(f"步骤{i + 1}与{j + 1}产出近义（{left} / {right}）")
+                break
+    return sigs
+
+
+def how_gate(knowledge: dict[str, Any], *, chat_json_fn: ChatJsonFn = default_chat_json) -> GateResult:
+    payload = json.dumps(
+        {
+            "purpose": knowledge.get("purpose"),
+            "steps": [
+                {
+                    "input": step.get("input"),
+                    "方法": (step.get("directive") or "")[:300],
+                    "产出": step.get("output"),
+                }
+                for step in knowledge.get("steps", [])
+            ],
+            "代码信号": chain_signals(knowledge.get("steps", [])),
+        },
+        ensure_ascii=False,
+    )
+    v_admit = vote_bool(GATE_HOW_ADMIT, payload, "is_real_how", on_fail=True, chat_json_fn=chat_json_fn)
+    v_refute = not vote_bool(GATE_HOW_REFUTE, payload, "is_fake", on_fail=False, chat_json_fn=chat_json_fn)
+    if v_admit == v_refute:
+        return GateResult(
+            passed=v_admit,
+            reason=f"admit={v_admit}/refute={v_refute} 一致",
+            details={"admit": v_admit, "refute": v_refute},
+        )
+    v_tie = vote_bool(GATE_HOW_TIEBREAK, payload, "is_real_how", on_fail=False, chat_json_fn=chat_json_fn)
+    return GateResult(
+        passed=v_tie,
+        reason=f"admit={v_admit}/refute={v_refute} 分歧→裁决={v_tie}",
+        details={"admit": v_admit, "refute": v_refute, "tiebreak": v_tie},
+    )
+
+
+def why_refute_gate(knowledge: dict[str, Any], *, chat_json_fn: ChatJsonFn = default_chat_json) -> GateResult:
+    payload = json.dumps({"阐述": knowledge.get("阐述")}, ensure_ascii=False)
+    try:
+        result = chat_json_fn(GATE_WHY_REFUTE, payload, timeout=90)
+    except Exception:
+        return GateResult(passed=True, reason="API错误→保留", details={"not_why": False})
+    not_why = bool(result.get("not_why"))
+    verdict = str(result.get("verdict") or "")
+    reason = str(result.get("reason") or "")
+    return GateResult(
+        passed=not not_why,
+        reason=reason,
+        details={"not_why": not_why, "verdict": verdict},
+    )

decode_content/models.py

@@ -0,0 +1,146 @@
+"""Formal decode-content domain models."""
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Literal
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+JsonDict = dict[str, Any]
+ParticleType = Literal["what", "how", "why"]
+
+
+class DecodeModel(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+
+class ReadCard(DecodeModel):
+    index: int
+    kind: str = "image"
+    url: str | None = None
+    timestamp: float | None = None
+    start: float | None = None
+    end: float | None = None
+    content: str = ""
+
+
+class ReadResult(DecodeModel):
+    text: str = ""
+    cards: list[ReadCard] = Field(default_factory=list)
+    media: JsonDict = Field(default_factory=dict)
+    is_empty: bool = False
+
+
+class GateResult(DecodeModel):
+    passed: bool
+    reason: str = ""
+    details: JsonDict = Field(default_factory=dict)
+
+
+class HowStep(DecodeModel):
+    input: str = ""
+    directive: str = ""
+    output: str = ""
+    scopes: list[JsonDict] = Field(default_factory=list)
+
+
+class KnowledgeParticle(DecodeModel):
+    id: UUID | None = None
+    decode_result_id: UUID | None = None
+    item_id: UUID | None = None
+    parent_particle_id: UUID | None = None
+    particle_type: ParticleType
+    title: str
+    business_stage: str | None = None
+    creation_stage: str | None = None
+    content: JsonDict = Field(default_factory=dict)
+    sort_order: int = 0
+    status: str = "draft"
+    metadata: JsonDict = Field(default_factory=dict)
+    error_message: str | None = None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+class ScopeResult(DecodeModel):
+    id: UUID | None = None
+    particle_id: UUID | None = None
+    item_id: UUID | None = None
+    scope_type: str
+    scope_value: str
+    is_reused: bool | None = None
+    matched_scope_id: str | None = None
+    confidence: float | None = None
+    evidence: JsonDict = Field(default_factory=dict)
+    status: str = "draft"
+    error_message: str | None = None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+class PayloadDraft(DecodeModel):
+    id: UUID | None = None
+    particle_id: UUID | None = None
+    item_id: UUID | None = None
+    payload: JsonDict = Field(default_factory=dict)
+    review_status: str = "pending"
+    ingest_ready: bool = False
+    status: str = "draft"
+    error_message: str | None = None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+class IngestRecord(DecodeModel):
+    id: UUID | None = None
+    payload_draft_id: UUID | None = None
+    target_system: str
+    target_id: str | None = None
+    status: str = "pending"
+    attempt_count: int = 0
+    response_payload: JsonDict = Field(default_factory=dict)
+    error_message: str | None = None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+class DecodeJob(DecodeModel):
+    id: UUID | None = None
+    item_id: UUID
+    status: str = "pending"
+    attempt_count: int = 0
+    metadata: JsonDict = Field(default_factory=dict)
+    error_message: str | None = None
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+class DecodeResult(DecodeModel):
+    id: UUID | None = None
+    decode_job_id: UUID | None = None
+    item_id: UUID
+    read_result: JsonDict = Field(default_factory=dict)
+    gate_result: JsonDict = Field(default_factory=dict)
+    framing_result: JsonDict = Field(default_factory=dict)
+    status: str = "draft"
+    error_message: str | None = None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+
+class ContractSnapshot(DecodeModel):
+    id: UUID | None = None
+    contract_name: str
+    contract_type: str
+    version_label: str | None = None
+    content_hash: str | None = None
+    source_path: str | None = None
+    snapshot: JsonDict = Field(default_factory=dict)
+    created_at: datetime | None = None
+
+
+Knowledge = KnowledgeParticle

decode_content/payloads.py

@@ -0,0 +1,135 @@
+"""Deterministic payload assembly for decoded creation knowledge."""
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from decode_content.contracts import DIM_ATTRIBUTE_BY_TYPE
+
+
+TYPE2ATTR = DIM_ATTRIBUTE_BY_TYPE
+
+
+def _post_value(post: Any, key: str, default: Any = "") -> Any:
+    if isinstance(post, dict):
+        return post.get(key, default)
+    return getattr(post, key, default)
+
+
+def build_content(knowledge: dict[str, Any]) -> str | dict[str, Any]:
+    t = knowledge.get("type")
+    if t == "how":
+        lines = [f"目标：{knowledge.get('purpose', '')}"]
+        for idx, step in enumerate(knowledge.get("steps", []), 1):
+            intent = step.get("intent") or step.get("purpose") or ""
+            label = f"步骤{idx}" + (f"（目的：{intent}）" if intent else "")
+            lines.extend(
+                [
+                    label,
+                    f"  输入：{step.get('input', '')}",
+                    f"  指引：{step.get('directive', '')}",
+                    f"  产出：{step.get('output', '')}",
+                ]
+            )
+        return "\n".join(lines)
+    if t == "what":
+        return {
+            "name": knowledge.get("title") or "",
+            "kind": knowledge.get("kind"),
+            "维度拆分规则": knowledge.get("维度拆分规则") if knowledge.get("kind") == "子集" else None,
+            "body": [
+                {
+                    "item_name": item.get("item_name", ""),
+                    "item_desc": item.get("item_desc", ""),
+                    "作用域": item.get("作用域") or [],
+                }
+                for item in (knowledge.get("body") or [])
+            ],
+        }
+    return knowledge.get("阐述") or knowledge.get("desc") or ""
+
+
+def _payload_content(knowledge: dict[str, Any]) -> str:
+    content = build_content(knowledge)
+    return content if isinstance(content, str) else json.dumps(content, ensure_ascii=False)
+
+
+def _add_scopes(scopes: list[dict[str, str]], seen: set[tuple[str, str]], items: list[dict[str, Any]]) -> None:
+    for item in items:
+        scope_type = item.get("scope_type")
+        value = item.get("value")
+        if not (scope_type and value):
+            continue
+        key = (scope_type, value)
+        if key in seen:
+            continue
+        seen.add(key)
+        scopes.append({"scope_type": scope_type, "value": value})
+
+
+def collect_payload_scopes(knowledge: dict[str, Any]) -> list[dict[str, str]]:
+    scopes: list[dict[str, str]] = []
+    seen: set[tuple[str, str]] = set()
+    if knowledge.get("type") == "how":
+        for step in knowledge.get("steps", []):
+            _add_scopes(scopes, seen, step.get("作用域", []))
+    else:
+        _add_scopes(scopes, seen, knowledge.get("作用域", []))
+    return scopes
+
+
+def build_payload(post: Any, knowledge: dict[str, Any], how_titles: dict[str, str] | None = None) -> dict[str, Any]:
+    how_titles = how_titles or {}
+    t = knowledge.get("type")
+    ext = [
+        {"key": "业务阶段", "type": "str", "value": stage}
+        for stage in (knowledge.get("业务阶段") or [])
+    ]
+    if t == "what" and knowledge.get("概要"):
+        ext.append({"key": "概要", "type": "str", "value": knowledge["概要"]})
+    if t == "how":
+        creation_stages: list[str] = []
+        seen_stages: set[str] = set()
+        for step in knowledge.get("steps", []):
+            stage = step.get("创作阶段")
+            if stage and stage not in seen_stages:
+                seen_stages.add(stage)
+                creation_stages.append(stage)
+        ext.extend({"key": "创作阶段", "type": "str", "value": stage} for stage in creation_stages)
+        ext.extend(
+            {"key": "动作", "type": "str", "value": step["动作"]}
+            for step in knowledge.get("steps", [])
+            if step.get("动作")
+        )
+    if knowledge.get("role") == "组件" and knowledge.get("parent"):
+        parent = knowledge["parent"]
+        ext.append(
+            {
+                "key": "出自",
+                "type": "str",
+                "value": f"{how_titles.get(parent.get('how_id'), parent.get('how_id'))} 第{parent.get('step')}步",
+            }
+        )
+    return {
+        "source": {
+            "id": _post_value(post, "id"),
+            "source_type": "post",
+            "title": _post_value(post, "title") or "",
+            "author": _post_value(post, "author_name") or "",
+            "source_metadata": {
+                "platform": _post_value(post, "platform"),
+                "url": _post_value(post, "url"),
+            },
+        },
+        "title": knowledge.get("title"),
+        "content": _payload_content(knowledge),
+        "dim_creations": ["创作"],
+        "dim_attributes": [TYPE2ATTR.get(t, "how工序")],
+        "scopes": collect_payload_scopes(knowledge),
+        "custom_ext": ext,
+    }
+
+
+def build_payloads(post: Any, knowledges: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    how_titles = {k.get("id"): k.get("title") for k in knowledges if k.get("type") == "how"}
+    return [build_payload(post, knowledge, how_titles) for knowledge in knowledges]

decode_content/repository.py

@@ -0,0 +1,107 @@
+"""Repository contracts for decode-content state."""
+from __future__ import annotations
+
+from typing import Any, Protocol
+from uuid import UUID
+
+from decode_content.models import (
+    ContractSnapshot,
+    DecodeJob,
+    DecodeResult,
+    IngestRecord,
+    KnowledgeParticle,
+    PayloadDraft,
+    ScopeResult,
+)
+
+
+class DecodeRepository(Protocol):
+    """Persistence boundary for the formal single-item decode workflow."""
+
+    def create_decode_job(
+        self,
+        *,
+        item_id: UUID,
+        status: str = "pending",
+        metadata: dict[str, Any] | None = None,
+    ) -> DecodeJob:
+        ...
+
+    def save_decode_result(
+        self,
+        *,
+        item_id: UUID,
+        decode_job_id: UUID | None = None,
+        read_result: dict[str, Any] | None = None,
+        gate_result: dict[str, Any] | None = None,
+        framing_result: dict[str, Any] | None = None,
+        status: str = "draft",
+    ) -> DecodeResult:
+        ...
+
+    def get_decode_result_for_item(self, item_id: UUID) -> DecodeResult:
+        ...
+
+    def save_knowledge_particle(
+        self,
+        *,
+        item_id: UUID,
+        particle_type: str,
+        title: str,
+        decode_result_id: UUID | None = None,
+        parent_particle_id: UUID | None = None,
+        content: dict[str, Any] | None = None,
+        status: str = "draft",
+    ) -> KnowledgeParticle:
+        ...
+
+    def save_scope_result(
+        self,
+        *,
+        scope_type: str,
+        scope_value: str,
+        particle_id: UUID | None = None,
+        item_id: UUID | None = None,
+        evidence: dict[str, Any] | None = None,
+        status: str = "draft",
+    ) -> ScopeResult:
+        ...
+
+    def save_payload_draft(
+        self,
+        *,
+        payload: dict[str, Any],
+        particle_id: UUID | None = None,
+        item_id: UUID | None = None,
+        review_status: str = "pending",
+        ingest_ready: bool = False,
+        status: str = "draft",
+    ) -> PayloadDraft:
+        ...
+
+    def list_payload_drafts(self, item_id: UUID | None = None) -> list[PayloadDraft]:
+        ...
+
+    def save_ingest_record(
+        self,
+        *,
+        payload_draft_id: UUID | None = None,
+        target_system: str,
+        target_id: str | None = None,
+        status: str = "pending",
+        response_payload: dict[str, Any] | None = None,
+        error_message: str | None = None,
+    ) -> IngestRecord:
+        ...
+
+    def save_contract_snapshot(
+        self,
+        *,
+        contract_name: str,
+        contract_type: str,
+        version_label: str | None = None,
+        content_hash: str | None = None,
+        source_path: str | None = None,
+        snapshot: dict[str, Any] | None = None,
+    ) -> ContractSnapshot:
+        ...

decode_content/scoping.py

@@ -0,0 +1,171 @@
+"""Scope candidate generation, normalization, and linking."""
+from __future__ import annotations
+
+import json
+import re
+from typing import Any, Protocol
+
+from core.llm import chat_json as default_chat_json
+from core.prompts import load_prompt
+from decode_content.contracts import REUSE_THRESHOLD, SCOPE_TYPE_CN, SkillContract, load_contract
+from decode_content.gates import ChatJsonFn
+
+
+NORMALIZE = load_prompt("normalize_scope")
+SRC2CN = SCOPE_TYPE_CN
+STRIP_VERBS = ("寻找", "定位", "推导", "核验", "提取", "挖掘", "捕捉", "识别", "梳理", "归纳", "判断", "验证", "确认", "复盘")
+SCOPE_CONN = re.compile(r"\s*[和与、，,/／&]\s*")
+
+
+class ScopeLinker(Protocol):
+    def link(self, value: str, *, source_type: str, top_k: int = 3) -> list[dict[str, Any]]:
+        ...
+
+
+def slim_knowledges(knowledges: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    slim: list[dict[str, Any]] = []
+    for knowledge in knowledges:
+        item = {"id": knowledge.get("id"), "type": knowledge.get("type"), "title": knowledge.get("title")}
+        if knowledge.get("type") == "how":
+            item["steps"] = [
+                {
+                    "id": step.get("id"),
+                    "input": step.get("input"),
+                    "directive": (step.get("directive") or "")[:500],
+                    "output": step.get("output"),
+                }
+                for step in knowledge.get("steps", [])
+            ]
+        else:
+            item["内容"] = {
+                key: knowledge.get(key)
+                for key in ("概要", "维度拆分规则", "body", "阐述")
+                if knowledge.get(key)
+            }
+        slim.append(item)
+    return slim
+
+
+def scope_candidates(
+    knowledges: list[dict[str, Any]],
+    *,
+    contract: SkillContract | None = None,
+    chat_json_fn: ChatJsonFn = default_chat_json,
+) -> list[dict[str, Any]]:
+    contract = contract or load_contract()
+    user = (
+        "给下面每颗知识标作用域候选（how 逐步：每个 step 一组；what/why 颗级：整颗一组）。\n"
+        "每类平铺列出所有相关值（同类可多条、全部对等、不选主次）；先脑内摊正交、收敛近义；意图通常 1 个。\n"
+        "只输出 JSON：{\"scopes\":[{\"knowledge_id\":\"k1\",\"step_id\":\"s1\",\"items\":["
+        "{\"scope_type\":\"substance\",\"value\":\"实质值1\"},{\"scope_type\":\"substance\",\"value\":\"实质值2\"},"
+        "{\"scope_type\":\"form\",\"value\":\"形式值1\"},{\"scope_type\":\"intent\",\"value\":\"意图值\"}]},"
+        "{\"knowledge_id\":\"k2\",\"step_id\":null,\"items\":[...]}]}\n\n"
+        + json.dumps(slim_knowledges(knowledges), ensure_ascii=False)
+    )
+    return chat_json_fn(contract.phase2, user, timeout=120).get("scopes") or []
+
+
+def strip_verb_tail(value: str) -> str:
+    if not value or len(value) < 3:
+        return value
+    original = value
+    for verb in STRIP_VERBS:
+        if value.startswith(verb) and len(value) - len(verb) >= 2:
+            value = value[len(verb):]
+            break
+    for verb in STRIP_VERBS:
+        if value.endswith(verb) and len(value) - len(verb) >= 2:
+            value = value[: -len(verb)]
+            break
+    return value or original
+
+
+def nounify_scopes(
+    scopes: list[dict[str, Any]],
+    *,
+    chat_json_fn: ChatJsonFn = default_chat_json,
+) -> list[dict[str, Any]]:
+    vals = sorted(
+        {
+            item["value"]
+            for scope in scopes
+            for item in (scope.get("items") or [])
+            if item.get("value") and item.get("scope_type") != "intent"
+        }
+    )
+    mapping: dict[str, str] = {}
+    if vals:
+        try:
+            mapping = chat_json_fn(NORMALIZE, json.dumps(vals, ensure_ascii=False), timeout=90).get("映射") or {}
+        except Exception:
+            mapping = {}
+    for scope in scopes:
+        for item in scope.get("items") or []:
+            value = item.get("value")
+            if not value or item.get("scope_type") == "intent":
+                continue
+            item["value"] = strip_verb_tail(mapping.get(value) or value)
+    return scopes
+
+
+def link_scope(linker: ScopeLinker, scope_type: str, value: str) -> dict[str, Any]:
+    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) or 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": hit["name"], "score": hit["score"], "path": hit.get("path", "")}
+            for hit in hits
+            if hit.get("name") is not None
+        ],
+    }
+
+
+def split_scope_items(items: list[dict[str, Any]] | None) -> list[dict[str, str]]:
+    out: list[dict[str, str]] = []
+    for item in items or []:
+        scope_type = item.get("scope_type")
+        value = item.get("value")
+        if not (scope_type and value):
+            continue
+        parts = [part.strip() for part in SCOPE_CONN.split(value) if part.strip()]
+        for part in (parts if len(parts) > 1 else [value]):
+            out.append({"scope_type": scope_type, "value": part})
+    return out
+
+
+def apply_scopes(
+    knowledges: list[dict[str, Any]],
+    scopes: list[dict[str, Any]],
+    linker: ScopeLinker | None = None,
+) -> None:
+    by_k = {knowledge.get("id"): knowledge for knowledge in knowledges}
+    for scope in scopes:
+        knowledge = by_k.get(scope.get("knowledge_id"))
+        if not knowledge:
+            continue
+        if linker is None:
+            linked = [
+                {"scope_type": item["scope_type"], "value": item["value"], "candidate": item["value"], "link": "候选"}
+                for item in split_scope_items(scope.get("items"))
+            ]
+        else:
+            linked = [
+                link_scope(linker, item["scope_type"], item["value"])
+                for item in split_scope_items(scope.get("items"))
+            ]
+        if knowledge.get("type") == "how" and scope.get("step_id"):
+            for step in knowledge.get("steps", []):
+                if step.get("id") == scope["step_id"]:
+                    step["作用域"] = linked
+        else:
+            knowledge["作用域"] = (knowledge.get("作用域") or []) + linked

decode_content/service.py

@@ -0,0 +1,273 @@
+"""Formal single-item decode orchestration."""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from typing import Any, Callable
+from uuid import UUID
+
+from core.config import Settings
+from core.llm import chat_json as default_chat_json
+from core.models import Post
+from decode_content.contracts import SkillContract, load_contract
+from decode_content.framing import frame_and_clean
+from decode_content.gates import ChatJsonFn, creation_gate
+from decode_content.models import DecodeResult, GateResult, PayloadDraft, ReadResult
+from decode_content.payloads import build_payloads
+from decode_content.readers.service import read_post
+from decode_content.repository import DecodeRepository
+from decode_content.scoping import ScopeLinker, apply_scopes, nounify_scopes, scope_candidates
+
+
+@dataclass
+class DecodeWorkflowOutput:
+    item_id: UUID
+    read_result: ReadResult
+    gate_result: GateResult
+    knowledges: list[dict[str, Any]] = field(default_factory=list)
+    payloads: list[dict[str, Any]] = field(default_factory=list)
+    decode_result: DecodeResult | None = None
+    payload_drafts: list[PayloadDraft] = field(default_factory=list)
+    status: str = "draft"
+
+
+def slugify(text: str) -> str:
+    s = re.sub(r"[^\w一-鿿]+", "-", text or "").strip("-").lower()
+    return s or "run"
+
+
+class DecodeService:
+    """Decode one creation candidate without depending on web files or SQLite."""
+
+    def __init__(
+        self,
+        *,
+        settings: Settings | None = None,
+        repository: DecodeRepository | None = None,
+        contract: SkillContract | None = None,
+        scope_linker: ScopeLinker | None = None,
+        chat_json_fn: ChatJsonFn | None = None,
+        reader: Callable[[Post], ReadResult] | None = None,
+    ) -> None:
+        self.settings = settings
+        self.repository = repository
+        self.contract = contract or load_contract()
+        self.scope_linker = scope_linker
+        self.chat_json_fn = chat_json_fn or default_chat_json
+        self.reader = reader
+
+    def _read(self, post: Post) -> ReadResult:
+        if self.reader is not None:
+            return self.reader(post)
+        return read_post(post, settings=self.settings)
+
+    def decode_post(
+        self,
+        *,
+        item_id: UUID,
+        post: Post,
+        read_result: ReadResult | None = None,
+    ) -> DecodeWorkflowOutput:
+        repo = self.repository
+        job = repo.create_decode_job(item_id=item_id, status="running") if repo else None
+        self._save_contract_snapshots()
+        read = read_result or self._read(post)
+        if read.is_empty:
+            gate = GateResult(passed=False, reason="读懂结果为空", details={"is_empty": True})
+            decode_result = self._save_decode_result(
+                item_id=item_id,
+                job_id=getattr(job, "id", None),
+                read=read,
+                gate=gate,
+                knowledges=[],
+                status="skipped",
+            )
+            return DecodeWorkflowOutput(
+                item_id=item_id,
+                read_result=read,
+                gate_result=gate,
+                decode_result=decode_result,
+                status="skipped",
+            )
+
+        gate = creation_gate(read.text, chat_json_fn=self.chat_json_fn)
+        if not gate.passed:
+            decode_result = self._save_decode_result(
+                item_id=item_id,
+                job_id=getattr(job, "id", None),
+                read=read,
+                gate=gate,
+                knowledges=[],
+                status="rejected",
+            )
+            return DecodeWorkflowOutput(
+                item_id=item_id,
+                read_result=read,
+                gate_result=gate,
+                decode_result=decode_result,
+                status="rejected",
+            )
+
+        knowledges = frame_and_clean(
+            post,
+            read.text,
+            contract=self.contract,
+            chat_json_fn=self.chat_json_fn,
+        )
+        scopes = scope_candidates(knowledges, contract=self.contract, chat_json_fn=self.chat_json_fn)
+        scopes = nounify_scopes(scopes, chat_json_fn=self.chat_json_fn)
+        apply_scopes(knowledges, scopes, self.scope_linker)
+        payloads = build_payloads(post, knowledges)
+        decode_result = self._save_decode_result(
+            item_id=item_id,
+            job_id=getattr(job, "id", None),
+            read=read,
+            gate=gate,
+            knowledges=knowledges,
+            status="decoded",
+        )
+        payload_drafts = self._save_particles_and_payloads(
+            item_id=item_id,
+            decode_result=decode_result,
+            knowledges=knowledges,
+            payloads=payloads,
+        )
+        return DecodeWorkflowOutput(
+            item_id=item_id,
+            read_result=read,
+            gate_result=gate,
+            knowledges=knowledges,
+            payloads=payloads,
+            decode_result=decode_result,
+            payload_drafts=payload_drafts,
+            status="decoded",
+        )
+
+    def _save_contract_snapshots(self) -> None:
+        if self.repository is None:
+            return
+        for snapshot in self.contract.snapshots():
+            self.repository.save_contract_snapshot(
+                contract_name=snapshot.contract_name,
+                contract_type=snapshot.contract_type,
+                version_label=snapshot.version_label,
+                content_hash=snapshot.content_hash,
+                source_path=snapshot.source_path,
+                snapshot=snapshot.snapshot,
+            )
+
+    def _save_decode_result(
+        self,
+        *,
+        item_id: UUID,
+        job_id: UUID | None,
+        read: ReadResult,
+        gate: GateResult,
+        knowledges: list[dict[str, Any]],
+        status: str,
+    ) -> DecodeResult:
+        framing_result = {"knowledges": knowledges, "contract_hash": self.contract.content_hash}
+        if self.repository is None:
+            return DecodeResult(
+                item_id=item_id,
+                decode_job_id=job_id,
+                read_result=read.model_dump(mode="json"),
+                gate_result=gate.model_dump(mode="json"),
+                framing_result=framing_result,
+                status=status,
+            )
+        return self.repository.save_decode_result(
+            item_id=item_id,
+            decode_job_id=job_id,
+            read_result=read.model_dump(mode="json"),
+            gate_result=gate.model_dump(mode="json"),
+            framing_result=framing_result,
+            status=status,
+        )
+
+    def _save_particles_and_payloads(
+        self,
+        *,
+        item_id: UUID,
+        decode_result: DecodeResult,
+        knowledges: list[dict[str, Any]],
+        payloads: list[dict[str, Any]],
+    ) -> list[PayloadDraft]:
+        if self.repository is None:
+            return [PayloadDraft(item_id=item_id, payload=payload) for payload in payloads]
+        drafts: list[PayloadDraft] = []
+        for knowledge, payload in zip(knowledges, payloads):
+            particle = self.repository.save_knowledge_particle(
+                item_id=item_id,
+                decode_result_id=decode_result.id,
+                particle_type=knowledge.get("type") or "how",
+                title=knowledge.get("title") or "",
+                content=knowledge,
+                status="draft",
+            )
+            for scope in payload.get("scopes") or []:
+                source_scope = self._scope_source(knowledge, scope)
+                self.repository.save_scope_result(
+                    item_id=item_id,
+                    particle_id=particle.id,
+                    scope_type=scope["scope_type"],
+                    scope_value=scope["value"],
+                    evidence={k: v for k, v in source_scope.items() if k not in {"scope_type", "value"}},
+                    status="draft",
+                )
+            drafts.append(
+                self.repository.save_payload_draft(
+                    item_id=item_id,
+                    particle_id=particle.id,
+                    payload=payload,
+                    review_status="pending",
+                    ingest_ready=False,
+                    status="draft",
+                )
+            )
+        return drafts
+
+    @staticmethod
+    def _scope_source(knowledge: dict[str, Any], scope: dict[str, Any]) -> dict[str, Any]:
+        source_scopes = list(knowledge.get("作用域", []))
+        for step in knowledge.get("steps", []):
+            source_scopes.extend(step.get("作用域", []))
+        return next(
+            (
+                item
+                for item in source_scopes
+                if item.get("scope_type") == scope["scope_type"] and item.get("value") == scope["value"]
+            ),
+            scope,
+        )
+
+
+def decode_post(
+    *,
+    item_id: UUID,
+    post: Post,
+    settings: Settings | None = None,
+    repository: DecodeRepository | None = None,
+    scope_linker: ScopeLinker | None = None,
+    chat_json_fn: ChatJsonFn | None = None,
+    reader: Callable[[Post], ReadResult] | None = None,
+    read_result: ReadResult | None = None,
+) -> DecodeWorkflowOutput:
+    service = DecodeService(
+        settings=settings,
+        repository=repository,
+        scope_linker=scope_linker,
+        chat_json_fn=chat_json_fn,
+        reader=reader,
+    )
+    return service.decode_post(item_id=item_id, post=post, read_result=read_result)
+
+
+def decode_item(
+    item_id: UUID,
+    *,
+    load_post: Callable[[UUID], Post],
+    service: DecodeService | None = None,
+) -> DecodeWorkflowOutput:
+    svc = service or DecodeService()
+    return svc.decode_post(item_id=item_id, post=load_post(item_id))

scripts/decompose.py

@@ -1,4 +1,9 @@
-"""创作知识解构引擎 v2：一帖 → N 颗（how/what/why，含组件颗）→ frameworks.json + payloads.json。
+"""LEGACY/debug 创作知识解构引擎 v2：一帖 → N 颗 → local web JSON.
+
+正式 decode 链路已经迁到 `decode_content/`，包括合同、reader、gate、framing、
+scoping、payload 和 service。本脚本仅保留给旧 demo/调试，仍会写
+`web/frameworks*.json`、`web/payloads*.json` 和 `web/runs/*.json`，不要作为
+正式云端 pipeline 入口。
 
 编排全流程，把 skill 的 phase 文档当 prompt 喂给 LLM（skill 是唯一真源）：
   ① 读懂：图文帖→extractor 读图；视频帖→video_extract 下载 mp4+原生整段提炼（base64→Gemini）
@@ -20,9 +25,9 @@ import time
 from pathlib import Path
 
 from core.config import Settings
-from creation_knowledge.integrations import video_extract
+from decode_content.readers import video as video_extract
 from acquisition.crawler import fetch_post_detail, parse_detail_response
-from creation_knowledge.integrations.extractor import BailianExtractor
+from decode_content.readers.imgtext import BailianExtractor
 from core.llm import chat_json
 from acquisition.oss import to_oss, upload_stream
 from acquisition.search import search_keyword

scripts/run_decode_content.py

@@ -0,0 +1,41 @@
+"""Run the formal single-item decode workflow.
+
+This CLI is intentionally thin. Production loading should come from the formal
+candidate item repository; fixture loading is kept only for local smoke checks.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+from uuid import UUID, uuid4
+
+from acquisition.crawler import parse_detail_response
+from core.config import Settings
+from decode_content.service import DecodeService
+
+
+ROOT = Path(__file__).resolve().parent.parent
+
+
+def _load_fixture(path: Path):
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    return parse_detail_response(payload, fallback_content_id=path.stem.removeprefix("xhs_case_"))
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Decode one creation candidate through decode_content.")
+    parser.add_argument("--fixture", type=Path, help="local fixture JSON for smoke runs")
+    parser.add_argument("--item-id", help="candidate item UUID; generated when omitted for fixture smoke")
+    args = parser.parse_args()
+    if not args.fixture:
+        parser.error("formal DB item loading is wired in pipeline Step 6; pass --fixture for local smoke")
+    item_id = UUID(args.item_id) if args.item_id else uuid4()
+    post = _load_fixture(args.fixture if args.fixture.is_absolute() else ROOT / args.fixture)
+    service = DecodeService(settings=Settings.from_env())
+    output = service.decode_post(item_id=item_id, post=post)
+    print(json.dumps({"item_id": str(item_id), "status": output.status, "payload_count": len(output.payloads)}, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()