""" Metadata schemas for API request/response. [AC-MRS-01, AC-MRS-07] 元数据职责分层相关的 Pydantic Schema """ from __future__ import annotations from datetime import datetime from enum import Enum from typing import Any from pydantic import BaseModel, Field class FieldRole(str, Enum): """ [AC-MRS-01] 字段角色枚举 用于标识元数据字段的职责分层 """ RESOURCE_FILTER = "resource_filter" SLOT = "slot" PROMPT_VAR = "prompt_var" ROUTING_SIGNAL = "routing_signal" class ExtractStrategy(str, Enum): """ [AC-MRS-07] 槽位值提取策略 """ RULE = "rule" LLM = "llm" USER_INPUT = "user_input" class SlotValueSource(str, Enum): """ [AC-MRS-09] 槽位值来源 """ USER_CONFIRMED = "user_confirmed" RULE_EXTRACTED = "rule_extracted" LLM_INFERRED = "llm_inferred" DEFAULT = "default" VALID_FIELD_ROLES = [role.value for role in FieldRole] class MetadataFieldDefinitionResponse(BaseModel): """[AC-MRS-01] 元数据字段定义响应""" id: str = Field(..., description="字段定义 ID") field_key: str = Field(..., description="字段键名") label: str = Field(..., description="字段显示名称") type: str = Field(..., description="字段类型") required: bool = Field(default=False, description="是否必填") options: list[str] | None = Field(default=None, description="选项列表") default_value: Any | None = Field(default=None, description="默认值") scope: list[str] = Field(default_factory=list, description="适用范围") is_filterable: bool = Field(default=True, description="是否可用于过滤") is_rank_feature: bool = Field(default=False, description="是否用于排序特征") field_roles: list[str] = Field( default_factory=list, description="[AC-MRS-01] 字段角色列表" ) status: str = Field(..., description="字段状态") version: int = Field(default=1, description="版本号") created_at: datetime = Field(..., description="创建时间") updated_at: datetime = Field(..., description="更新时间") class Config: from_attributes = True class MetadataFieldDefinitionCreateRequest(BaseModel): """[AC-MRS-01,02,03] 创建元数据字段定义请求""" field_key: str = Field(..., min_length=1, max_length=64, description="字段键名") label: str = Field(..., min_length=1, max_length=64, description="字段显示名称") type: str = Field(default="string", description="字段类型") required: bool = Field(default=False, description="是否必填") options: list[str] | None = Field(default=None, description="选项列表") default_value: Any | None = Field(default=None, description="默认值") scope: list[str] = Field( default_factory=lambda: ["kb_document"], description="适用范围" ) is_filterable: bool = Field(default=True, description="是否可用于过滤") is_rank_feature: bool = Field(default=False, description="是否用于排序特征") field_roles: list[str] = Field( default_factory=list, description="[AC-MRS-01] 字段角色列表,可选值: resource_filter, slot, prompt_var, routing_signal" ) status: str = Field(default="draft", description="字段状态") def validate_field_roles(self) -> list[str]: """验证 field_roles 中的角色值是否有效""" invalid_roles = [r for r in self.field_roles if r not in VALID_FIELD_ROLES] if invalid_roles: raise ValueError( f"Invalid field_roles: {invalid_roles}. " f"Valid roles are: {VALID_FIELD_ROLES}" ) return self.field_roles class MetadataFieldDefinitionUpdateRequest(BaseModel): """[AC-MRS-01] 更新元数据字段定义请求""" label: str | None = Field(default=None, min_length=1, max_length=64) type: str | None = Field(default=None, description="字段类型") required: bool | None = None options: list[str] | None = None default_value: Any | None = None scope: list[str] | None = None is_filterable: bool | None = None is_rank_feature: bool | None = None usage_description: str | None = Field(default=None, description="用途说明") field_roles: list[str] | None = Field( default=None, description="[AC-MRS-01] 字段角色列表" ) status: str | None = None def validate_field_roles(self) -> list[str] | None: """验证 field_roles 中的角色值是否有效""" if self.field_roles is None: return None invalid_roles = [r for r in self.field_roles if r not in VALID_FIELD_ROLES] if invalid_roles: raise ValueError( f"Invalid field_roles: {invalid_roles}. " f"Valid roles are: {VALID_FIELD_ROLES}" ) return self.field_roles class SlotDefinitionResponse(BaseModel): """[AC-MRS-07,08] 槽位定义响应""" id: str = Field(..., description="槽位定义 ID") slot_key: str = Field(..., description="槽位键名") type: str = Field(..., description="槽位类型") required: bool = Field(default=False, description="是否必填槽位") # [AC-MRS-07-UPGRADE] 保留旧字段用于兼容 extract_strategy: str | None = Field( default=None, description="[兼容字段] 单提取策略,已废弃" ) # [AC-MRS-07-UPGRADE] 新增策略链字段 extract_strategies: list[str] | None = Field( default=None, description="[AC-MRS-07-UPGRADE] 提取策略链:有序数组,元素为 rule/llm/user_input" ) validation_rule: str | None = Field(default=None, description="校验规则") ask_back_prompt: str | None = Field(default=None, description="追问提示语模板") default_value: dict[str, Any] | None = Field(default=None, description="默认值") linked_field_id: str | None = Field(default=None, description="关联的元数据字段 ID") created_at: datetime = Field(..., description="创建时间") updated_at: datetime = Field(..., description="更新时间") class Config: from_attributes = True class SlotDefinitionCreateRequest(BaseModel): """[AC-MRS-07,08] 创建槽位定义请求""" slot_key: str = Field(..., min_length=1, max_length=100, description="槽位键名") type: str = Field(default="string", description="槽位类型") required: bool = Field(default=False, description="是否必填槽位") # [AC-MRS-07-UPGRADE] 支持策略链 extract_strategies: list[str] | None = Field( default=None, description="[AC-MRS-07-UPGRADE] 提取策略链:有序数组,元素为 rule/llm/user_input,按顺序执行直到成功" ) # [AC-MRS-07-UPGRADE] 保留旧字段用于兼容 extract_strategy: str | None = Field( default=None, description="[兼容字段] 单提取策略,已废弃,请使用 extract_strategies" ) validation_rule: str | None = Field(default=None, description="校验规则") ask_back_prompt: str | None = Field(default=None, description="追问提示语模板") default_value: dict[str, Any] | None = Field(default=None, description="默认值") linked_field_id: str | None = Field(default=None, description="关联的元数据字段 ID") class SlotDefinitionUpdateRequest(BaseModel): """[AC-MRS-07] 更新槽位定义请求""" type: str | None = None required: bool | None = None # [AC-MRS-07-UPGRADE] 支持策略链 extract_strategies: list[str] | None = Field( default=None, description="[AC-MRS-07-UPGRADE] 提取策略链:有序数组,元素为 rule/llm/user_input,按顺序执行直到成功" ) # [AC-MRS-07-UPGRADE] 保留旧字段用于兼容 extract_strategy: str | None = Field( default=None, description="[兼容字段] 单提取策略,已废弃,请使用 extract_strategies" ) validation_rule: str | None = None ask_back_prompt: str | None = None default_value: dict[str, Any] | None = None linked_field_id: str | None = None class SlotValueResponse(BaseModel): """[AC-MRS-09] 运行时槽位值响应""" key: str = Field(..., description="槽位键名") value: Any = Field(..., description="槽位值") source: str = Field( default="default", description="来源: user_confirmed/rule_extracted/llm_inferred/default" ) confidence: float = Field( default=1.0, ge=0.0, le=1.0, description="置信度 0.0~1.0" ) updated_at: datetime = Field(..., description="最后更新时间") class SlotWithFieldDefinitionResponse(BaseModel): """[AC-MRS-10] 槽位定义与关联字段定义响应""" slot_definition: SlotDefinitionResponse | None = Field( default=None, description="槽位定义" ) field_definition: MetadataFieldDefinitionResponse | None = Field( default=None, description="关联的元数据字段定义" ) class GetFieldsByRoleQuery(BaseModel): """[AC-MRS-04,05] 按角色查询字段定义请求""" role: str = Field(..., description="字段角色") def validate_role(self) -> str: """验证角色值是否有效""" if self.role not in VALID_FIELD_ROLES: raise ValueError( f"[AC-MRS-05] Invalid role '{self.role}'. " f"Valid roles are: {VALID_FIELD_ROLES}" ) return self.role class GetSlotsByRoleQuery(BaseModel): """[AC-MRS-10] 按角色获取槽位定义请求""" role: str = Field(default="slot", description="字段角色,默认为 slot") class InvalidRoleErrorResponse(BaseModel): """[AC-MRS-05] 无效角色错误响应""" error: str = Field(default="INVALID_ROLE", description="错误码") message: str = Field(..., description="错误消息") valid_roles: list[str] = Field( default_factory=lambda: VALID_FIELD_ROLES, description="有效的角色列表" )