Create-knowledge-find-decode/creation_knowledge/stages/split.py

"""拆分：把帖子拆成一个或多个 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 Evidence, 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_int(value) -> Optional[int]:
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


def _to_evidence(raw) -> Optional[Evidence]:
    """证据可能是字符串（旧）或 {text, card}（新）。"""
    if isinstance(raw, dict):
        text = norm_text(raw.get("text"))
        return Evidence(text=text, card=_to_int(raw.get("card"))) if text else None
    text = norm_text(raw)
    return Evidence(text=str(raw).strip()) if text else None


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 (_to_evidence(x) for x in (raw.get("evidence") or [])) if e]
    source_cards = [n for n in (_to_int(x) for x in (raw.get("source_cards") or [])) if n is not None]
    return KnowledgeItem(
        title=title, knowledge_types=types, what=what, why=why, how=how,
        source_cards=source_cards, evidence=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]