Zdao032
135 lines
5.0 KiB
Python
135 lines
5.0 KiB
Python
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 |