312 lines
15 KiB
Python
312 lines
15 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
File: DataModels.py
|
|
Author: oyyz
|
|
Date: 2025-05-13
|
|
Description: 提取和分类的数据模型
|
|
"""
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from typing import List, Optional, Dict, Tuple, Union, Any
|
|
from enum import Enum
|
|
|
|
class SoftwareName(str, Enum):
|
|
"""软件名称枚举类"""
|
|
D3 = "配网工程计价通D3软件"
|
|
C1 = "新型储能电站建设计价通C1软件"
|
|
Z1 = "西藏电力工程计价通Z1软件"
|
|
T1 = "技改检修工程计价通T1软件"
|
|
T1_LIST = "技改检修清单计价通T1软件"
|
|
MAIN = "主网电力建设计价通软件"
|
|
UNKNOWN = "" # 未知
|
|
|
|
# 软件别名映射
|
|
ALIASES = {
|
|
D3: "别名包括：配网D3、D3软件、配网工程软件等其他类似称呼",
|
|
C1: "别名包括：储能C1、C1软件、储能电站软件、储能软件等其他类似称呼",
|
|
Z1: "别名包括：西藏Z1、Z1软件、西藏电力软件等其他类似称呼",
|
|
T1: "别名包括：技改T1、T1软件、技改检修软件等其他类似称呼",
|
|
T1_LIST: "别名包括：技改清单T1、T1清单软件、技改检修清单软件等其他类似称呼",
|
|
MAIN: "别名包括：主网软件、电力建设软件、主网建设软件、博微电力建设计价通等其他类似称呼"
|
|
}
|
|
|
|
# 构建别名到标准名称的映射
|
|
def build_alias_mapping() -> Dict[str, SoftwareName]:
|
|
"""构建从别名到标准软件名称的映射字典"""
|
|
alias_map = {}
|
|
|
|
# 配网工程计价通D3软件的别名映射
|
|
alias_map["配网D3"] = SoftwareName.D3
|
|
alias_map["D3软件"] = SoftwareName.D3
|
|
alias_map["配网工程软件"] = SoftwareName.D3
|
|
alias_map["配网软件"] = SoftwareName.D3
|
|
|
|
# 新型储能电站建设计价通C1软件的别名映射
|
|
alias_map["储能C1"] = SoftwareName.C1
|
|
alias_map["C1软件"] = SoftwareName.C1
|
|
alias_map["储能电站软件"] = SoftwareName.C1
|
|
alias_map["储能软件"] = SoftwareName.C1
|
|
|
|
# 西藏电力工程计价通Z1软件的别名映射
|
|
alias_map["西藏Z1"] = SoftwareName.Z1
|
|
alias_map["Z1软件"] = SoftwareName.Z1
|
|
alias_map["西藏电力软件"] = SoftwareName.Z1
|
|
|
|
# 技改检修工程计价通T1软件的别名映射
|
|
alias_map["技改T1"] = SoftwareName.T1
|
|
alias_map["T1软件"] = SoftwareName.T1
|
|
alias_map["技改检修软件"] = SoftwareName.T1
|
|
|
|
# 技改检修清单计价通T1软件的别名映射
|
|
alias_map["技改清单T1"] = SoftwareName.T1_LIST
|
|
alias_map["T1清单软件"] = SoftwareName.T1_LIST
|
|
alias_map["技改检修清单软件"] = SoftwareName.T1_LIST
|
|
|
|
# 主网电力建设计价通软件的别名映射
|
|
alias_map["主网软件"] = SoftwareName.MAIN
|
|
alias_map["电力建设软件"] = SoftwareName.MAIN
|
|
alias_map["主网建设软件"] = SoftwareName.MAIN
|
|
alias_map["博微电力建设计价通"] = SoftwareName.MAIN
|
|
alias_map["主网计价通"] = SoftwareName.MAIN
|
|
alias_map["主网计价通软件"] = SoftwareName.MAIN
|
|
alias_map["计价通软件"] = SoftwareName.MAIN
|
|
alias_map["电力计价通软件"] = SoftwareName.MAIN
|
|
alias_map["计价通"] = SoftwareName.MAIN
|
|
# 添加标准名称映射
|
|
alias_map[SoftwareName.D3.value] = SoftwareName.D3
|
|
alias_map[SoftwareName.C1.value] = SoftwareName.C1
|
|
alias_map[SoftwareName.Z1.value] = SoftwareName.Z1
|
|
alias_map[SoftwareName.T1.value] = SoftwareName.T1
|
|
alias_map[SoftwareName.T1_LIST.value] = SoftwareName.T1_LIST
|
|
alias_map[SoftwareName.MAIN.value] = SoftwareName.MAIN
|
|
|
|
return alias_map
|
|
|
|
# 全局别名映射字典
|
|
SOFTWARE_NAME_ALIAS_MAP = build_alias_mapping()
|
|
|
|
# 定义输出模型
|
|
class Term(BaseModel):
|
|
name: str = Field(description="专业名词")
|
|
synonymous: List[str] = Field(description="同义词列表")
|
|
description: str = Field(description="描述信息", default="")
|
|
|
|
def __hash__(self):
|
|
return hash(self.name)
|
|
|
|
def __eq__(self, other):
|
|
if isinstance(other, Term):
|
|
return self.name == other.name
|
|
return False
|
|
|
|
class TermList(BaseModel):
|
|
terms: List[Term] = Field(description="专业名词列表")
|
|
|
|
class Classification(BaseModel):
|
|
vertical_classification:str = Field(description="垂直领域一级分类")
|
|
sub_classification:str = Field(description="一级分类下的二级分类")
|
|
|
|
class QueryRewrite(BaseModel):
|
|
rewrite:str = Field(description="问题改写")
|
|
|
|
##########################槽位模型###########################
|
|
class SlotBase(BaseModel):
|
|
"""槽位基础模型"""
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
"""检查必填槽位是否都存在"""
|
|
raise NotImplementedError("子类必须实现check_required_slots方法")
|
|
|
|
@field_validator('software_name', mode='before', check_fields=False)
|
|
@classmethod
|
|
def validate_software_name(cls, v):
|
|
"""验证并转换软件名称，支持别名"""
|
|
if v is None or v == "":
|
|
return ""
|
|
|
|
# 如果已经是枚举类型，直接返回其值
|
|
if isinstance(v, SoftwareName):
|
|
return v.value
|
|
|
|
# 如果是字符串，尝试转换
|
|
if isinstance(v, str):
|
|
# 直接匹配枚举值
|
|
for software in SoftwareName:
|
|
if v == software.value:
|
|
return software.value
|
|
|
|
# 尝试通过别名匹配
|
|
if v in SOFTWARE_NAME_ALIAS_MAP:
|
|
return SOFTWARE_NAME_ALIAS_MAP[v].value
|
|
|
|
# 如果无法匹配，返回原值
|
|
return v
|
|
|
|
return v
|
|
|
|
# 1. 软件问题
|
|
# 1.1 软件功能
|
|
class SoftwareFunctionSlots(SlotBase):
|
|
software_name: str = Field(default="", description="软件名称")
|
|
function_name: str = Field(default="", description="具体功能名称")
|
|
operation: str = Field(default="", description="用户操作意图(如何使用功能、功能入口、功能使用场景)")
|
|
project_type: Optional[str] = Field(default="单工程", description="工程类型(单工程、多工程、批次工程), 未明确提及则默认下是(单工程)")
|
|
software_version: Optional[str] = Field(default="", description="软件版本")
|
|
operation_steps: Optional[str] = Field(default="", description="操作步骤描述")
|
|
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
"""检查必填槽位是否都存在"""
|
|
if self.project_type is None or len(self.project_type) == 0:
|
|
self.project_type="单工程"
|
|
missing_slots = {}
|
|
if not self.software_name:
|
|
missing_slots["software_name"] = f"{SoftwareFunctionSlots.model_fields['software_name'].description}，可选值：{', '.join([name.value for name in SoftwareName if name not in [SoftwareName.UNKNOWN, SoftwareName.ALIASES]])}"
|
|
if not self.function_name:
|
|
missing_slots["function_name"] = SoftwareFunctionSlots.model_fields["function_name"].description
|
|
if not self.operation:
|
|
missing_slots["operation"] = SoftwareFunctionSlots.model_fields["operation"].description
|
|
return len(missing_slots) == 0, missing_slots
|
|
|
|
# 1.2 故障排查
|
|
class SoftwareTroubleShootingSlots(SlotBase):
|
|
software_name: str = Field(default="", description="软件名称")
|
|
function_name: str = Field(default="", description="具体功能名称/操作描述")
|
|
error_message: str = Field(default="", description="报错信息/异常现象")
|
|
software_version: Optional[str] = Field(default="", description="软件版本")
|
|
os_version: Optional[str] = Field(default="", description="操作系统及版本")
|
|
reproduction_steps: Optional[str] = Field(default="", description="故障重现步骤")
|
|
project_type: Optional[str] = Field(default="单工程", description="工程类型(单工程、多工程、批次工程)")
|
|
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
"""检查必填槽位是否都存在"""
|
|
missing_slots = {}
|
|
if not self.software_name:
|
|
missing_slots["software_name"] = f"{SoftwareTroubleShootingSlots.model_fields['software_name'].description}，可选值：{', '.join([name.value for name in SoftwareName if name not in [SoftwareName.UNKNOWN, SoftwareName.ALIASES]])}"
|
|
if not self.function_name:
|
|
missing_slots["function_name"] = SoftwareTroubleShootingSlots.model_fields["function_name"].description
|
|
if not self.error_message:
|
|
missing_slots["error_message"] = SoftwareTroubleShootingSlots.model_fields["error_message"].description
|
|
return len(missing_slots) == 0, missing_slots
|
|
|
|
# 2. 业务问题
|
|
# 2.1 专业咨询
|
|
class ProfessionalConsultingSlots(SlotBase):
|
|
scene_subject: str = Field(default="", description="场景主体")
|
|
business_scene: str = Field(default="", description="业务场景描述")
|
|
software_name: Optional[str] = Field(default="", description="软件名称")
|
|
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
"""检查必填槽位是否都存在"""
|
|
missing_slots = {}
|
|
if not self.scene_subject:
|
|
missing_slots["scene_subject"] = ProfessionalConsultingSlots.model_fields["scene_subject"].description
|
|
if not self.business_scene:
|
|
missing_slots["business_scene"] = ProfessionalConsultingSlots.model_fields["business_scene"].description
|
|
return len(missing_slots) == 0, missing_slots
|
|
|
|
# 2.2 数据问题
|
|
class DataProblemSlots(SlotBase):
|
|
expense_type: str = Field(default="", description="费用(数据)项、费用类型")
|
|
operation_purpose: str = Field(default="", description="操作目的")
|
|
software_name: Optional[str] = Field(default="", description="软件名称")
|
|
project_type: Optional[str] = Field(default="", description="工程类型")
|
|
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
"""检查必填槽位是否都存在"""
|
|
missing_slots = {}
|
|
if not self.expense_type:
|
|
missing_slots["expense_type"] = DataProblemSlots.model_fields["expense_type"].description
|
|
if not self.operation_purpose:
|
|
missing_slots["operation_purpose"] = DataProblemSlots.model_fields["operation_purpose"].description
|
|
return len(missing_slots) == 0, missing_slots
|
|
|
|
# 3. 安装下载注册
|
|
# 3.1 后缀名咨询
|
|
class FileExtensionConsultingSlots(SlotBase):
|
|
file_extension: str = Field(default="", description="文件后缀名")
|
|
operation_purpose: str = Field(default="", description="操作目的(了解对应软件,对应工程)")
|
|
file_source: Optional[str] = Field(default="", description="文件来源场景")
|
|
related_software: Optional[str] = Field(default="", description="相关软件名称")
|
|
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
"""检查必填槽位是否都存在"""
|
|
missing_slots = {}
|
|
if not self.file_extension:
|
|
missing_slots["file_extension"] = FileExtensionConsultingSlots.model_fields["file_extension"].description
|
|
if not self.operation_purpose:
|
|
missing_slots["operation_purpose"] = FileExtensionConsultingSlots.model_fields["operation_purpose"].description
|
|
return len(missing_slots) == 0, missing_slots
|
|
|
|
# 3.2 软件锁类
|
|
class SoftwareLockSlots(SlotBase):
|
|
lock_type: str = Field(default="", description="锁类型")
|
|
operation_purpose: str = Field(default="", description="操作目的")
|
|
lock_number: Optional[str] = Field(default="", description="软件锁编号/注册号")
|
|
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
"""检查必填槽位是否都存在"""
|
|
missing_slots = {}
|
|
if not self.lock_type:
|
|
missing_slots["lock_type"] = SoftwareLockSlots.model_fields["lock_type"].description
|
|
if not self.operation_purpose:
|
|
missing_slots["operation_purpose"] = SoftwareLockSlots.model_fields["operation_purpose"].description
|
|
return len(missing_slots) == 0, missing_slots
|
|
|
|
# 3.3 安装下载类
|
|
class InstallationDownloadSlots(SlotBase):
|
|
|
|
software_name: str = Field(default="", description="软件/插件名称，与file_name二选一")
|
|
file_name: str = Field(default="", description="文件名，与software_name二选一")
|
|
operation_stage: str = Field(default="", description="操作阶段(下载、安装等)")
|
|
os_version: Optional[str] = Field(default="", description="操作系统版本")
|
|
package_source: Optional[str] = Field(default="", description="安装包来源/版本号")
|
|
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
"""检查必填槽位是否都存在"""
|
|
missing_slots = {}
|
|
if not self.software_name and not self.file_name:
|
|
missing_slots["software_name"] = f"{InstallationDownloadSlots.model_fields['software_name'].description},"
|
|
f"可选值：{', '.join([name.value for name in SoftwareName if name not in [SoftwareName.UNKNOWN, SoftwareName.ALIASES]])}"
|
|
missing_slots["file_name"] = InstallationDownloadSlots.model_fields["file_name"].description
|
|
if not self.operation_stage:
|
|
missing_slots["operation_stage"] = InstallationDownloadSlots.model_fields["operation_stage"].description
|
|
return len(missing_slots) == 0, missing_slots
|
|
|
|
# 3.4 问题排查类
|
|
class ProblemDiagnosisSlots(SlotBase):
|
|
error_message: str = Field(default="", description="报错信息/异常现象")
|
|
software_name: Optional[str] = Field(default="", description="软件名称")
|
|
os_version: Optional[str] = Field(default="", description="操作系统版本")
|
|
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
"""检查必填槽位是否都存在"""
|
|
missing_slots = {}
|
|
if not self.error_message:
|
|
missing_slots["error_message"] = ProblemDiagnosisSlots.model_fields["error_message"].description
|
|
return len(missing_slots) == 0, missing_slots
|
|
|
|
class OtherSlots(SlotBase):
|
|
"""其他类型槽位"""
|
|
content_type: str = Field(default="", description="内容类型(必填)")
|
|
intent: Optional[str] = Field(default="", description="用户意图(选填)")
|
|
|
|
def check_required_slots(self) -> Tuple[bool, Dict[str, str]]:
|
|
missing_slots = {}
|
|
if not self.content_type:
|
|
missing_slots["content_type"] = OtherSlots.model_fields["content_type"].description
|
|
return len(missing_slots) == 0, missing_slots
|
|
|
|
class IntentAndSlotResult(BaseModel):
|
|
"""意图槽位填充结果"""
|
|
classification: Classification
|
|
slots: Union[
|
|
SoftwareFunctionSlots,
|
|
SoftwareTroubleShootingSlots,
|
|
ProfessionalConsultingSlots,
|
|
DataProblemSlots,
|
|
FileExtensionConsultingSlots,
|
|
SoftwareLockSlots,
|
|
InstallationDownloadSlots,
|
|
ProblemDiagnosisSlots,
|
|
OtherSlots
|
|
] |