269 lines
9.9 KiB
Python
269 lines
9.9 KiB
Python
"""
|
||
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="有效的角色列表"
|
||
)
|