from datetime import datetime from enum import Enum from dataclasses import dataclass, field from typing import List, Dict, Optional, Any @dataclass class QuestionInfo: """用户问句信息""" question_raw: str # 原始问句:用户咨询问题源输入 final_processed: str # 最终处理:经过一系列处理后形成的最终问句的信息存储 # mediate_processing: None # 中间临时处理:形如停用词处理、文本纠错等中间步骤暂存的临时存储信息 # ================================================= @dataclass class DialogType(str, Enum): IN_PROGRESS = "应答中" OPENING = "开头" CLOSING = "结尾" COMPLETION = "追问" @dataclass class DialogInfo: """当前对话信息""" dialog_type: DialogType # 对话类型:应答中、开头、结尾、追问 timestamp: str # datetime类型 # ================================================= class QuestionType(str, Enum): WHEN = "when" WHERE = "where" WHO = "who" WHAT = "what" WHY = "why" HOW = "how" HOW_MUCH = "how much" # @dataclass # class InheritedInfo: # """继承信息容器""" # business_entity: Optional[str] = None # 业务实体词 # intent_keyword: Optional[str] = None # 意图词 # domain: Optional[str] = None # 领域 # intent: Optional[str] = None # 意图 # def get_priority_info(self): # """按优先级获取继承信息""" # return next(filter(None, [self.business_entity, self.intent_keyword, self.domain, self.intent]), None) # 关于下载安装 # 查询功能(询问软件功能或用途):【新型储能计价通C1】可以做批次工程吗 # 后缀名问题(专属子意图) # 操作指导(询问具体使用方法):【西藏造价软件Z1】怎么查看人工机械费用 # 故障排查(解决软件错误问题):【技改检修2016软件】初始化失败了 # 费用计算(专属意图,复杂计算推理) # 下载与安装(询问软件的下载和安装):【配网D3软件】程序包发我;【配网D3软件】怎么安装 @dataclass class NLUInfo: """语义理解信息""" vertical_category: str # 垂直开放分类:开放聊天助手、软件使用助手 intent_category: Optional[str] = None # 意图分类:查询功能、操作指导、故障排查、费用输出、下载安装 domain_category: Optional[str] = None # 领域分类:西藏造价软件Z1, 新型储能计价通C1, 技改检修计价通T1 previous_intent: Optional[str] = None # 上文意图 previous_domain: Optional[str] = None # 上文领域 intent_results: Optional[str] = None # 意图识别结果列表: Ai模型预测(待定) domain_results: Optional[str] = None # 领域识别结果列表: Ai模型预测(待定) tokens: Optional[str] = None # 分词结果 emotion: Optional[str] = None # 情绪识别 question_type: Optional[QuestionType] = None # 问句类型结果:5W2H中的一种 inherited_info: Optional[str] = None # 用户继承信息:一次对话只能从上下文继承一种信息(优先度:业务实体词>意图词>领域>意图) retrieve_keywords: Optional[str] = None # 测试 keys rewrite: Optional[str] = None # 建议构有向无环图 @dataclass class ScenInfo: """场景信息""" scene_name: str # 场景名称 process_status: None # 场景流程状态 slots: Dict[str, Any] = field(default_factory=dict) # 槽位信息 # ================================================= # @dataclass # class AnswerInfo: # """系统应答信息""" # answer_result: str # 一次对话的答案结果 # transfer_human: bool=False # 是否转人工 # ================================================= @dataclass class TalkInfo: Q: str # 用户输入 A: str # 机器输出 @dataclass class ChatRecord: """历史记录""" conversation: List[TalkInfo] = field(default_factory=list) # ================================================= @dataclass class DSTManager: """对话状态管理器,整合各类信息""" question_info: QuestionInfo # 用户问句信息 dialog_info: DialogInfo # 当前对话信息 nlu_info: NLUInfo # 语义理解信息 scen_info: ScenInfo # 场景信息 chat_record: ChatRecord # 历史对话记录 # def update_question(self, question_raw: str): # """更新用户问题并重置相关状态""" # self.question_info = QuestionInfo(question_raw=question_raw) # self.dialog_info.timestamp = datetime.now() # def generate_answer(self, answer: str, transfer_human: bool = False): # """生成系统应答信息""" # self.answer_info = AnswerInfo(answer_result=answer, transfer_human=transfer_human) # def inherit_previous_context(self): # """继承上一次对话的上下文信息""" # inherited = self.nlu_info.inherited_info.get_priority_info() # return inherited