Simon
ead5f83bee
dependencies.py 拆分为 dependencies/ 包; 新增 vision/ragflow_ingestion/neo4j 客户端与 h5_ai_task; alembic 045 图置信度迁移; 响应契约统一收尾。
159 lines
5.7 KiB
Python
159 lines
5.7 KiB
Python
# =============================================================================
|
||
# 企微IT智能服务台 — 枚举集中定义
|
||
# =============================================================================
|
||
# 说明：所有知识库迭代相关的枚举类型集中定义在此，避免散落在各模块。
|
||
# 包括：受众类型、建议状态、图同步状态、来源类型、关系类型。
|
||
# =============================================================================
|
||
|
||
from enum import Enum
|
||
|
||
|
||
class AudienceEnum(str, Enum):
|
||
"""知识受众枚举 — 决定知识条目面向哪类用户。
|
||
|
||
D8 硬约束：当前仅两类，第三类（管理运营KB）由后续 P2 扩展。
|
||
|
||
Values:
|
||
employee_quick_reply: 员工快捷回复 KB（面向普通员工）
|
||
engineer_workguide: 工程师作业指导 KB（面向 IT 工程师/坐席）
|
||
"""
|
||
|
||
employee_quick_reply = "employee_quick_reply" # 员工快捷回复 KB
|
||
engineer_workguide = "engineer_workguide" # 工程师作业指导 KB
|
||
|
||
|
||
class SuggestionStatusEnum(str, Enum):
|
||
"""知识建议审批状态枚举 — 五态 + 终止态。
|
||
|
||
D7 硬约束：默认 pending（非自动采纳），审批通过后流转 applied→graph_synced。
|
||
|
||
状态流转路径：
|
||
pending → queued → approved → applied → graph_synced
|
||
pending → approved → applied → graph_synced (内联审批直达)
|
||
pending → rejected (驳回)
|
||
pending → expired (超时)
|
||
approved → applied → graph_synced (采纳后写图)
|
||
|
||
Values:
|
||
pending: 待审核（初始状态）
|
||
queued: 已入队列（会话关闭后未处理的提案）
|
||
approved: 已通过（训练师审批通过）
|
||
rejected: 已驳回
|
||
applied: 已应用（KB 条目已落库）
|
||
graph_synced: 图已同步（Neo4j 写图完成，最终态）
|
||
expired: 已过期（超时未处理）
|
||
"""
|
||
|
||
pending = "pending" # 待审核（初始状态）
|
||
queued = "queued" # 已入独立队列
|
||
approved = "approved" # 已通过
|
||
rejected = "rejected" # 已驳回
|
||
applied = "applied" # 已应用到 KB
|
||
graph_synced = "graph_synced" # Neo4j 图已同步（最终态）
|
||
expired = "expired" # 已过期
|
||
|
||
|
||
class GraphSyncStatusEnum(str, Enum):
|
||
"""图同步状态枚举 — 追踪知识条目与 Neo4j 图的同步状态。
|
||
|
||
Values:
|
||
pending: 待同步（KB 已落库但图未写入）
|
||
synced: 已同步（图写入成功）
|
||
failed: 同步失败（进入重试队列）
|
||
"""
|
||
|
||
pending = "pending" # 待同步
|
||
synced = "synced" # 已同步
|
||
failed = "failed" # 同步失败
|
||
|
||
|
||
class SourceTypeEnum(str, Enum):
|
||
"""建议来源类型枚举 — 追踪知识建议的生成通道。
|
||
|
||
对应三条输入通道：
|
||
A: conversation/annotation/ai_uncertain → 会话自动生成
|
||
B: manual → 训练师手动录入
|
||
C: document_ragflow → RAGFlow 文档 ETL
|
||
merge: 去重合并产生 → 知识合并去重
|
||
|
||
Values:
|
||
annotation: 标注数据分析
|
||
conversation: 会话数据分析
|
||
ai_uncertain: AI 不确定回复
|
||
manual: 训练师手动录入（通道 B）
|
||
document_ragflow: RAGFlow 文档处理（通道 C）
|
||
merge: 知识合并去重（任务3 P2）
|
||
"""
|
||
|
||
annotation = "annotation" # 标注数据分析
|
||
conversation = "conversation" # 会话数据分析
|
||
ai_uncertain = "ai_uncertain" # AI 不确定回复
|
||
manual = "manual" # 训练师手动录入（通道 B）
|
||
document_ragflow = "document_ragflow" # RAGFlow 文档 ETL（通道 C）
|
||
merge = "merge" # 知识合并去重（任务3 P2）
|
||
|
||
|
||
class RelationTypeEnum(str, Enum):
|
||
"""图关系类型枚举 — 对齐复杂场景重构 v1.1。
|
||
|
||
命名映射约定（§8.1）：
|
||
LEADS_TO → 完全对齐
|
||
RELATES_TO → CAN_JUMP_TO 语义简化统一
|
||
CAN_JUMP_TO → 保留兼容（等同于 RELATES_TO {type:"jump"}）
|
||
|
||
Values:
|
||
LEADS_TO: 引导关系（Issue→Action 或 Issue→Issue）
|
||
RELATES_TO: 关联关系（双向，含子类型）
|
||
CAN_JUMP_TO: 跳转关系（非线性跳转支持）
|
||
"""
|
||
|
||
LEADS_TO = "LEADS_TO" # 引导关系
|
||
RELATES_TO = "RELATES_TO" # 关联关系
|
||
CAN_JUMP_TO = "CAN_JUMP_TO" # 跳转关系
|
||
|
||
|
||
# =============================================================================
|
||
# 合法状态转换表（用于审批状态机校验）
|
||
# =============================================================================
|
||
|
||
# key: 当前状态, value: 允许转换到的目标状态集合
|
||
ALLOWED_STATUS_TRANSITIONS: dict = {
|
||
SuggestionStatusEnum.pending: {
|
||
SuggestionStatusEnum.queued,
|
||
SuggestionStatusEnum.approved,
|
||
SuggestionStatusEnum.rejected,
|
||
SuggestionStatusEnum.expired,
|
||
},
|
||
SuggestionStatusEnum.queued: {
|
||
SuggestionStatusEnum.approved,
|
||
SuggestionStatusEnum.rejected,
|
||
SuggestionStatusEnum.expired,
|
||
},
|
||
SuggestionStatusEnum.approved: {
|
||
SuggestionStatusEnum.applied,
|
||
SuggestionStatusEnum.rejected,
|
||
},
|
||
SuggestionStatusEnum.applied: {
|
||
SuggestionStatusEnum.graph_synced,
|
||
},
|
||
SuggestionStatusEnum.graph_synced: set(), # 终态，不可再转换
|
||
SuggestionStatusEnum.rejected: set(), # 终态
|
||
SuggestionStatusEnum.expired: set(), # 终态
|
||
}
|
||
|
||
|
||
def is_valid_transition(
|
||
current: SuggestionStatusEnum, target: SuggestionStatusEnum
|
||
) -> bool:
|
||
"""检查状态转换是否合法。
|
||
|
||
Args:
|
||
current: 当前状态
|
||
target: 目标状态
|
||
|
||
Returns:
|
||
bool: 转换合法返回 True
|
||
"""
|
||
allowed = ALLOWED_STATUS_TRANSITIONS.get(current, set())
|
||
return target in allowed
|