harness_zbt/backend/app/harness/schemas.py

"""Agent 与双端前端之间的稳定结构化协议。"""

from dataclasses import dataclass, field
from typing import Annotated, Any, Literal

from pydantic import BaseModel, ConfigDict, Field


class StrictModel(BaseModel):
    """拒绝未声明字段，防止模型生成内容悄悄穿透协议。"""

    model_config = ConfigDict(extra="forbid")


class ProductPlanView(StrictModel):
    id: str
    code: str
    name: str
    summary: str = ""
    premium_cents: int = Field(ge=0)
    coverage_amount_cents: int = Field(ge=0)
    min_age: int = Field(ge=0, le=120)
    max_age: int = Field(ge=0, le=120)


class ProductView(StrictModel):
    product_id: str
    product_code: str
    name: str
    category: str
    summary: str
    product_version_id: str
    plans: list[ProductPlanView]


class ProductRecommendationsBlock(StrictModel):
    type: Literal["product_recommendations"] = "product_recommendations"
    version: Literal["1.0"] = "1.0"
    title: str = "为你匹配的保障方案"
    items: list[ProductView] = Field(max_length=2)


class QuoteView(StrictModel):
    eligible: bool
    quote_id: str | None = None
    product_id: str | None = None
    product_name: str
    plan_id: str | None = None
    plan_name: str
    premium_cents: int | None = None
    currency: str = "CNY"
    expires_at: str | None = None
    reason: str | None = None


class QuoteBlock(StrictModel):
    type: Literal["quote"] = "quote"
    version: Literal["1.0"] = "1.0"
    title: str = "保费测算结果"
    quote: QuoteView


class BusinessListBlock(StrictModel):
    type: Literal["business_list"] = "business_list"
    version: Literal["1.0"] = "1.0"
    title: str
    entity: Literal["order", "policy", "product", "attribution"]
    items: list[dict[str, Any]]
    total: int = Field(ge=0)


class MetricItem(StrictModel):
    label: str
    value: int | float | str
    unit: str = ""
    trend: str | None = None


class MetricsBlock(StrictModel):
    type: Literal["metrics"] = "metrics"
    version: Literal["1.0"] = "1.0"
    title: str
    items: list[MetricItem]


UiBlock = Annotated[
    ProductRecommendationsBlock | QuoteBlock | BusinessListBlock | MetricsBlock,
    Field(discriminator="type"),
]


class AgentAction(StrictModel):
    """前端可执行动作；类型和负载都由 Harness 控制。"""

    type: Literal[
        "open_enrollment",
        "open_orders",
        "open_policies",
        "open_admin_orders",
        "open_admin_policies",
        "open_attribution",
        "none",
    ]
    label: str
    payload: dict[str, Any] = Field(default_factory=dict)
    confirmation_required: bool = False


class AgentStructuredOutput(StrictModel):
    """模型最终输出；业务数据必须来自 Tool，不允许模型自行填充。"""

    message: str = Field(min_length=1, max_length=1200)
    selected_product_codes: list[str] = Field(default_factory=list, max_length=2)
    suggested_action: AgentAction | None = None


class HarnessToolResult(StrictModel):
    """单个 Tool 的统一结果包络。"""

    summary: str
    data: dict[str, Any] = Field(default_factory=dict)
    blocks: list[UiBlock] = Field(default_factory=list)
    actions: list[AgentAction] = Field(default_factory=list)


@dataclass(frozen=True)
class AgentReply:
    """Agent Kernel 对应用层返回的结果。

    ``cards`` 沿用第一版 H5 字段名，内容已经升级为受校验的 UI Block。
    """

    text: str
    cards: list[dict[str, Any]]
    trace_id: str | None
    trace_url: str | None = None
    actions: list[dict[str, Any]] = field(default_factory=list)
    invoked_tools: tuple[str, ...] = ()