50 lines
1.5 KiB
Python
50 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any, Protocol
|
|
|
|
|
|
class ModerationError(RuntimeError):
|
|
"""Base error for content-moderation failures."""
|
|
|
|
|
|
class ModerationUnavailableError(ModerationError):
|
|
"""Provider could not return a verdict (down / timeout / not implemented).
|
|
|
|
The orchestration layer converts this into a fail-closed BLOCK or a
|
|
fail-open PASS depending on ``MODERATION_FAIL_CLOSED`` policy.
|
|
"""
|
|
|
|
|
|
class ModerationConfigError(ModerationError):
|
|
"""Provider is misconfigured (missing credentials / biz type / endpoint)."""
|
|
|
|
|
|
class Verdict(str, Enum):
|
|
PASS = "pass" # 放行
|
|
REVIEW = "review" # 建议人工复审（是否等同拦截由 policy 决定）
|
|
BLOCK = "block" # 拦截
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ModerationResult:
|
|
"""Normalized moderation verdict, decoupled from any vendor's response shape.
|
|
|
|
NOTE: ``raw`` is only for debugging in-process; never persist it verbatim
|
|
(may contain the moderated content / vendor internals). Compliance logging
|
|
stores a summary (verdict + labels + request_id), not ``raw``.
|
|
"""
|
|
|
|
verdict: Verdict
|
|
labels: tuple[str, ...] = ()
|
|
score: int | None = None
|
|
keywords: tuple[str, ...] = ()
|
|
request_id: str = ""
|
|
raw: Any = field(default=None, repr=False, compare=False)
|
|
|
|
|
|
class ModerationProvider(Protocol):
|
|
def moderate_text(self, text: str, *, biz_type: str = "") -> ModerationResult:
|
|
...
|