group_fqcd_jr/app/service/knowledge_publication_service.py

"""管理员知识发布编排：客服运行期只读，本模块仅供显式发布工具调用。"""

from collections import defaultdict
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Protocol

from app.core.knowledge_contracts import ALLOWED_KNOWLEDGE_COLLECTIONS

VECTOR_DIMENSION = 1024


class KnowledgePublicationError(RuntimeError):
    """发布前置条件、阶段写入或补偿失败时的明确错误。"""


@dataclass(frozen=True)
class KnowledgePublicationRecord:
    """预检清单中一条已经审核、可供管理员发布的公开知识。"""

    qa_id: str
    milvus_collection: str
    retrieval_text: str
    title: str
    snippet: str
    tags: tuple[str, ...]
    version: str
    metadata: Mapping[str, object]


@dataclass(frozen=True)
class KnowledgePublicationResult:
    """仅返回可审计的业务编号和数据库主键映射，不返回正文或密钥。"""

    knowledge_ids: dict[str, int]
    collections: tuple[str, ...]


class KnowledgeEmbedder(Protocol):
    async def embed(self, text: str) -> list[float]: ...


class KnowledgePublicationStore(Protocol):
    async def stage(self, records: tuple[KnowledgePublicationRecord, ...]) -> dict[str, int]: ...

    async def publish(self, knowledge_ids: tuple[int, ...], reviewer_id: int) -> None: ...

    async def disable(self, knowledge_ids: tuple[int, ...]) -> None: ...


class KnowledgeVectorPublisher(Protocol):
    async def upsert(self, collection: str, records: tuple[dict[str, object], ...]) -> None: ...

    async def delete(self, collection: str, knowledge_ids: tuple[str, ...]) -> None: ...


class KnowledgePublicationService:
    """把发布动作拆为可补偿阶段，任何中断都不能让未索引知识对客可见。"""

    def __init__(
        self,
        embedder: KnowledgeEmbedder,
        store: KnowledgePublicationStore,
        vectors: KnowledgeVectorPublisher,
    ) -> None:
        self._embedder = embedder
        self._store = store
        self._vectors = vectors

    async def publish(
        self, records: Sequence[KnowledgePublicationRecord], *, reviewer_id: int
    ) -> KnowledgePublicationResult:
        immutable_records = tuple(records)
        self._validate(immutable_records, reviewer_id)
        embeddings = await self._embeddings(immutable_records)
        knowledge_ids = await self._store.stage(immutable_records)
        self._validate_staged_ids(immutable_records, knowledge_ids)
        payloads = self._payloads(immutable_records, embeddings, knowledge_ids)
        try:
            for collection, collection_payloads in payloads.items():
                await self._vectors.upsert(collection, tuple(collection_payloads))
        except Exception as exc:
            await self._compensate(payloads, tuple(knowledge_ids.values()))
            raise KnowledgePublicationError("向量写入失败，知识保持未发布状态") from exc
        await self._store.publish(tuple(knowledge_ids.values()), reviewer_id)
        return KnowledgePublicationResult(
            knowledge_ids=knowledge_ids,
            collections=tuple(payloads),
        )

    @staticmethod
    def _validate(records: tuple[KnowledgePublicationRecord, ...], reviewer_id: int) -> None:
        if reviewer_id <= 0:
            raise KnowledgePublicationError("reviewer_id 必须是正整数")
        if not records:
            raise KnowledgePublicationError("没有可发布的公开知识")
        qa_ids = [record.qa_id for record in records]
        if len(qa_ids) != len(set(qa_ids)):
            raise KnowledgePublicationError("发布清单存在重复 qa_id")
        for record in records:
            if record.milvus_collection not in ALLOWED_KNOWLEDGE_COLLECTIONS:
                raise KnowledgePublicationError("发布清单包含未授权集合")
            if not record.retrieval_text.strip():
                raise KnowledgePublicationError(f"{record.qa_id}: 检索文本不能为空")

    async def _embeddings(
        self, records: tuple[KnowledgePublicationRecord, ...]
    ) -> dict[str, list[float]]:
        embeddings: dict[str, list[float]] = {}
        for record in records:
            vector = await self._embedder.embed(record.retrieval_text)
            if len(vector) != VECTOR_DIMENSION:
                raise KnowledgePublicationError(
                    f"{record.qa_id}: 向量维度必须为 {VECTOR_DIMENSION}"
                )
            embeddings[record.qa_id] = vector
        return embeddings

    @staticmethod
    def _validate_staged_ids(
        records: tuple[KnowledgePublicationRecord, ...], knowledge_ids: dict[str, int]
    ) -> None:
        expected = {record.qa_id for record in records}
        if set(knowledge_ids) != expected or any(value <= 0 for value in knowledge_ids.values()):
            raise KnowledgePublicationError("MySQL 暂存结果与发布清单不一致")

    @staticmethod
    def _payloads(
        records: tuple[KnowledgePublicationRecord, ...],
        embeddings: dict[str, list[float]],
        knowledge_ids: dict[str, int],
    ) -> dict[str, list[dict[str, object]]]:
        payloads: dict[str, list[dict[str, object]]] = defaultdict(list)
        for record in records:
            payloads[record.milvus_collection].append({
                "knowledge_id": str(knowledge_ids[record.qa_id]),
                "embedding": embeddings[record.qa_id],
                "title": record.title,
                "snippet": record.snippet,
                "tags": list(record.tags),
                "version": record.version,
            })
        return dict(payloads)

    async def _compensate(
        self, payloads: dict[str, list[dict[str, object]]], knowledge_ids: tuple[int, ...]
    ) -> None:
        for collection, items in payloads.items():
            try:
                await self._vectors.delete(
                    collection, tuple(str(item["knowledge_id"]) for item in items)
                )
            except Exception:
                # MySQL 行仍会被停用，因此清理失败的残余向量无法对客返回。
                pass
        await self._store.disable(knowledge_ids)