157 lines
6.1 KiB
Python
157 lines
6.1 KiB
Python
"""管理员知识发布编排：客服运行期只读，本模块仅供显式发布工具调用。"""
|
|
|
|
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)
|