X-Financial/server/src/app/schemas/ontology.py

from __future__ import annotations

from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field

OntologyScenario = Literal[
    "expense",
    "accounts_receivable",
    "accounts_payable",
    "knowledge",
    "unknown",
]
OntologyIntent = Literal["query", "explain", "compare", "risk_check", "draft", "operate"]
OntologyPermissionLevel = Literal["read", "draft_write", "approval_required", "forbidden"]
OntologyParseStrategy = Literal["llm_primary", "rule_fallback"]


class OntologyEntity(BaseModel):
    model_config = ConfigDict(extra="forbid")

    type: str = Field(description="业务对象类型，例如 employee / customer / vendor。")
    value: str = Field(description="从原始问题中提取的对象值。")
    normalized_value: str = Field(description="标准化后的对象值。")
    role: str = Field(default="target", description="对象角色，例如 target / filter / threshold。")
    confidence: float = Field(default=0.0, ge=0.0, le=1.0, description="字段级置信度。")


class OntologyTimeRange(BaseModel):
    model_config = ConfigDict(extra="forbid")

    raw: str = Field(default="", description="命中的原始时间表达。")
    start_date: str | None = Field(default=None, description="ISO 格式起始日期。")
    end_date: str | None = Field(default=None, description="ISO 格式结束日期。")
    granularity: str | None = Field(
        default=None,
        description="day / week / month / quarter / year。",
    )


class OntologyMetric(BaseModel):
    model_config = ConfigDict(extra="forbid")

    name: str = Field(description="指标名，例如 amount / count / overdue。")
    aggregation: str | None = Field(default=None, description="sum / count / max 等聚合口径。")
    unit: str | None = Field(default=None, description="金额、数量等单位。")
    sort: str | None = Field(default=None, description="asc / desc 排序方向。")
    top_n: int | None = Field(default=None, ge=1, description="Top N 口径。")


class OntologyConstraint(BaseModel):
    model_config = ConfigDict(extra="forbid")

    field: str = Field(description="约束字段，例如 department / status / amount。")
    operator: str = Field(description="操作符，例如 = / > / < / desc。")
    value: str | int | float | bool = Field(description="约束值。")
    currency: str | None = Field(default=None, description="金额类约束使用的币种。")


class OntologyPermission(BaseModel):
    model_config = ConfigDict(extra="forbid")

    level: OntologyPermissionLevel = Field(default="read", description="动作权限等级。")
    allowed: bool = Field(default=True, description="是否可直接执行当前动作。")
    reason: str = Field(default="", description="权限判断原因。")


class OntologyFieldError(BaseModel):
    model_config = ConfigDict(extra="forbid")

    field: str = Field(description="发生问题的字段。")
    code: str = Field(description="错误码。")
    message: str = Field(description="面向前端展示的说明。")


class OntologyParseRequest(BaseModel):
    query: str = Field(min_length=1, description="自然语言问题。")
    user_id: str | None = Field(default=None, description="当前请求用户 ID。")
    context_json: dict[str, Any] = Field(
        default_factory=dict,
        description="用户上下文，例如角色、部门、是否管理员。",
    )


class OntologyParseResult(BaseModel):
    scenario: OntologyScenario = Field(default="unknown", description="业务场景。")
    intent: OntologyIntent = Field(default="query", description="用户意图。")
    entities: list[OntologyEntity] = Field(default_factory=list, description="业务对象列表。")
    time_range: OntologyTimeRange = Field(
        default_factory=OntologyTimeRange,
        description="时间范围。",
    )
    metrics: list[OntologyMetric] = Field(default_factory=list, description="指标解析结果。")
    constraints: list[OntologyConstraint] = Field(
        default_factory=list,
        description="过滤、阈值、排序等约束。",
    )
    risk_flags: list[str] = Field(default_factory=list, description="风险信号列表。")
    permission: OntologyPermission = Field(
        default_factory=OntologyPermission,
        description="权限结果。",
    )
    confidence: float = Field(default=0.0, ge=0.0, le=1.0, description="整体置信度。")
    missing_slots: list[str] = Field(default_factory=list, description="继续处理所缺少的关键槽位。")
    ambiguity: list[str] = Field(default_factory=list, description="当前识别中的潜在歧义。")
    parse_strategy: OntologyParseStrategy = Field(
        default="rule_fallback",
        description="本次语义解析使用的主策略。",
    )
    clarification_required: bool = Field(default=False, description="是否需要追问。")
    clarification_question: str | None = Field(default=None, description="推荐追问问题。")
    run_id: str = Field(description="关联的 AgentRun.run_id。")
    field_errors: list[OntologyFieldError] = Field(
        default_factory=list,
        description="字段级错误或提示。",
    )