MerCry
/
ai-robot-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add Pydantic schemas for metadata field roles and slot definitions [AC-MRS-01,07]

This commit is contained in:
MerCry 2026-03-05 17:11:48 +08:00
parent 14d1737728
commit 68e5adaa28
2 changed files with 281 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
ai-service/app/schemas/__init__.py Normal file
View File

@ -0,0 +1,39 @@
"""
Schemas package for API request/response models.
"""
from app.schemas.metadata import (
FieldRole,
ExtractStrategy,
SlotValueSource,
MetadataFieldDefinitionResponse,
MetadataFieldDefinitionCreateRequest,
MetadataFieldDefinitionUpdateRequest,
SlotDefinitionResponse,
SlotDefinitionCreateRequest,
SlotDefinitionUpdateRequest,
SlotValueResponse,
SlotWithFieldDefinitionResponse,
GetFieldsByRoleQuery,
GetSlotsByRoleQuery,
InvalidRoleErrorResponse,
VALID_FIELD_ROLES,
)
__all__ = [
"FieldRole",
"ExtractStrategy",
"SlotValueSource",
"MetadataFieldDefinitionResponse",
"MetadataFieldDefinitionCreateRequest",
"MetadataFieldDefinitionUpdateRequest",
"SlotDefinitionResponse",
"SlotDefinitionCreateRequest",
"SlotDefinitionUpdateRequest",
"SlotValueResponse",
"SlotWithFieldDefinitionResponse",
"GetFieldsByRoleQuery",
"GetSlotsByRoleQuery",
"InvalidRoleErrorResponse",
"VALID_FIELD_ROLES",
]

242
ai-service/app/schemas/metadata.py Normal file
View File

@ -0,0 +1,242 @@
"""
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)
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
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="是否必填槽位")
extract_strategy: str | None = Field(default=None, description="提取策略")
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="是否必填槽位")
extract_strategy: str | None = Field(
default=None,
description="提取策略: 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")
class SlotDefinitionUpdateRequest(BaseModel):
"""[AC-MRS-07] 更新槽位定义请求"""
type: str | None = None
required: bool | None = None
extract_strategy: str | None = None
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="有效的角色列表"
)