From 68e5adaa28d0d6e06806a68548adf6350b7f6f24 Mon Sep 17 00:00:00 2001 From: MerCry Date: Thu, 5 Mar 2026 17:11:48 +0800 Subject: [PATCH] feat: add Pydantic schemas for metadata field roles and slot definitions [AC-MRS-01,07] --- ai-service/app/schemas/__init__.py | 39 +++++ ai-service/app/schemas/metadata.py | 242 +++++++++++++++++++++++++++++ 2 files changed, 281 insertions(+) create mode 100644 ai-service/app/schemas/__init__.py create mode 100644 ai-service/app/schemas/metadata.py diff --git a/ai-service/app/schemas/__init__.py b/ai-service/app/schemas/__init__.py new file mode 100644 index 0000000..f5cc2bb --- /dev/null +++ b/ai-service/app/schemas/__init__.py @@ -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", +] diff --git a/ai-service/app/schemas/metadata.py b/ai-service/app/schemas/metadata.py new file mode 100644 index 0000000..0e8db84 --- /dev/null +++ b/ai-service/app/schemas/metadata.py @@ -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="有效的角色列表" + )