#! /usr/bin/env python # -*- coding: utf-8 -*- # vim:fenc=utf-8 from enum import Enum, auto from typing import Dict, List, Optional, Tuple, Any from datetime import datetime import time import logging import cozepy from chat_service import ChatServiceType from message import MessageType, Message # from vector_memory_manager import VectorMemoryManager from structured_memory_manager import StructuredMemoryManager from user_manager import UserManager from prompt_templates import * # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(funcName)s[%(lineno)d] - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) class DummyVectorMemoryManager: def __init__(self, user_id): pass def add_to_memory(self, conversation): pass def retrieve_relevant_memories(self, query, k=3): return [] class DialogueState(Enum): GREETING = auto() # 问候状态 CHITCHAT = auto() # 闲聊状态 CLARIFICATION = auto() # 澄清状态 FAREWELL = auto() # 告别状态 HUMAN_INTERVENTION = auto() # 人工介入状态 MESSAGE_AGGREGATING = auto() # 等待消息状态 class TimeContext(Enum): EARLY_MORNING = "清晨" # 清晨 (5:00-7:59) MORNING = "上午" # 上午 (8:00-11:59) NOON = "中午" # 中午 (12:00-13:59) AFTERNOON = "下午" # 下午 (14:00-17:59) EVENING = "晚上" # 晚上 (18:00-21:59) NIGHT = "深夜" # 夜晚 (22:00-4:59) def __init__(self, description): self.description = description class DialogueManager: def __init__(self, staff_id: str, user_id: str, user_manager: UserManager): self.staff_id = staff_id self.user_id = user_id self.user_manager = user_manager self.current_state = DialogueState.GREETING self.previous_state = None self.dialogue_history = [] self.user_profile = self.user_manager.get_user_profile(user_id) self.last_interaction_time = 0 self.consecutive_clarifications = 0 self.complex_request_counter = 0 self.human_intervention_triggered = False self.vector_memory = DummyVectorMemoryManager(user_id) self.message_aggregation_sec = 5 self.unprocessed_messages = [] def get_current_time_context(self) -> TimeContext: """获取当前时间上下文""" current_hour = datetime.now().hour if 5 <= current_hour < 8: return TimeContext.EARLY_MORNING elif 8 <= current_hour < 12: return TimeContext.MORNING elif 12 <= current_hour < 14: return TimeContext.NOON elif 14 <= current_hour < 18: return TimeContext.AFTERNOON elif 18 <= current_hour < 22: return TimeContext.EVENING else: return TimeContext.NIGHT def update_state(self, message: Message) -> Tuple[DialogueState, Optional[str]]: """根据用户消息更新对话状态,并返回下一条需处理的用户消息""" message_text = message.content message_ts = message.sendTime # 如果当前已经是人工介入状态,保持该状态 if self.current_state == DialogueState.HUMAN_INTERVENTION: # 记录对话历史,但不改变状态 self.dialogue_history.append({ "role": "user", "content": message_text, "timestamp": int(time.time() * 1000), "state": self.current_state.name }) return self.current_state, message_text # 检查是否处于消息聚合状态 if self.current_state == DialogueState.MESSAGE_AGGREGATING: # 收到的是特殊定时触发的空消息,且在聚合中,且已经超时,恢复之前状态,继续处理 if message.type == MessageType.AGGREGATION_TRIGGER \ and message_ts - self.last_interaction_time > self.message_aggregation_sec * 1000: logging.debug("user_id: {}, last interaction time: {}".format( self.user_id, datetime.fromtimestamp(self.last_interaction_time / 1000))) self.current_state = self.previous_state else: # 非空消息,更新最后交互时间,保持消息聚合状态 if message_text: self.unprocessed_messages.append(message_text) self.last_interaction_time = message_ts return self.current_state, message_text elif message.type != MessageType.AGGREGATION_TRIGGER and self.message_aggregation_sec > 0: # 收到有内容的用户消息,切换到消息聚合状态 self.previous_state = self.current_state self.current_state = DialogueState.MESSAGE_AGGREGATING self.unprocessed_messages.append(message_text) # 更新最后交互时间 if message_text: self.last_interaction_time = message_ts return self.current_state, message_text # 保存前一个状态 self.previous_state = self.current_state # 检查是否长时间未交互(超过3小时) if self._get_hours_since_last_interaction() > 3: self.current_state = DialogueState.GREETING self.dialogue_history = [] # 重置对话历史 self.consecutive_clarifications = 0 # 重置澄清计数 self.complex_request_counter = 0 # 重置复杂请求计数 # 获得未处理的聚合消息,并清空未处理队列 if message_text: self.unprocessed_messages.append(message_text) if self.unprocessed_messages: message_text = '\n'.join(self.unprocessed_messages) self.unprocessed_messages.clear() # 根据消息内容和当前状态确定新状态 new_state = self._determine_state_from_message(message_text) # 处理连续澄清的情况 if new_state == DialogueState.CLARIFICATION: self.consecutive_clarifications += 1 if self.consecutive_clarifications >= 2: new_state = DialogueState.HUMAN_INTERVENTION # self._trigger_human_intervention("连续多次澄清请求") else: self.consecutive_clarifications = 0 # 更新状态 self.current_state = new_state # 更新最后交互时间 if message_text: self.last_interaction_time = message_ts # 记录对话历史 if message_text: self.dialogue_history.append({ "role": "user", "content": message_text, "timestamp": message_ts, "state": self.current_state.name }) return self.current_state, message_text def _determine_state_from_message(self, message_text: Optional[str]) -> DialogueState: """根据消息内容确定对话状态""" if not message_text: return self.current_state # 简单的规则-关键词匹配 message_lower = message_text.lower() # 判断是否是复杂请求 complex_request_keywords = ["帮我", "怎么办", "我需要", "麻烦你", "请帮助", "急", "紧急"] if any(keyword in message_lower for keyword in complex_request_keywords): self.complex_request_counter += 1 # 如果检测到困难请求且计数达到阈值,触发人工介入 if self.complex_request_counter >= 1: # self._trigger_human_intervention("检测到复杂请求") return DialogueState.HUMAN_INTERVENTION else: # 如果不是复杂请求,重置计数器 self.complex_request_counter = 0 # 问候检测 greeting_keywords = ["你好", "早上好", "中午好", "晚上好", "嗨", "在吗"] if any(keyword in message_lower for keyword in greeting_keywords): return DialogueState.GREETING # 告别检测 farewell_keywords = ["再见", "拜拜", "晚安", "明天见", "回头见"] if any(keyword in message_lower for keyword in farewell_keywords): return DialogueState.FAREWELL # 澄清请求 clarification_keywords = ["没明白", "不明白", "没听懂", "不懂", "什么意思", "再说一遍"] if any(keyword in message_lower for keyword in clarification_keywords): return DialogueState.CLARIFICATION # 默认为闲聊状态 return DialogueState.CHITCHAT def _trigger_human_intervention(self, reason: str) -> None: """触发人工介入""" if not self.human_intervention_triggered: self.human_intervention_triggered = True # 记录人工介入事件 event = { "timestamp": int(time.time() * 1000), "reason": reason, "dialogue_context": self.dialogue_history[-5:] if len(self.dialogue_history) >= 5 else self.dialogue_history } # 更新用户资料中的人工介入历史 if "human_intervention_history" not in self.user_profile: self.user_profile["human_intervention_history"] = [] self.user_profile["human_intervention_history"].append(event) self.user_manager.save_user_profile(self.user_profile) # 发送告警 self._send_human_intervention_alert(reason) def _send_human_intervention_alert(self, reason: str) -> None: alert_message = f""" 人工介入告警 用户ID: {self.user_id} 用户昵称: {self.user_profile.get("nickname", "未知")} 时间: {int(time.time() * 1000)} 原因: {reason} 最近对话: """ # 添加最近的对话记录 recent_dialogues = self.dialogue_history[-5:] if len(self.dialogue_history) >= 5 else self.dialogue_history for dialogue in recent_dialogues: alert_message += f"\n{dialogue['role']}: {dialogue['content']}" # TODO(zhoutian): 实现发送告警的具体逻辑 logger.warning(alert_message) def resume_from_human_intervention(self) -> None: """从人工介入状态恢复""" if self.current_state == DialogueState.HUMAN_INTERVENTION: self.current_state = DialogueState.GREETING self.human_intervention_triggered = False self.consecutive_clarifications = 0 self.complex_request_counter = 0 # 记录恢复事件 self.dialogue_history.append({ "role": "system", "content": "已从人工介入状态恢复到自动对话", "timestamp": int(time.time() * 1000), "state": self.current_state.name }) def generate_response(self, llm_response: str) -> Optional[str]: """根据当前状态处理LLM响应,如果处于人工介入状态则返回None""" # 如果处于人工介入状态,不生成回复 if self.current_state == DialogueState.HUMAN_INTERVENTION: return None # 记录响应到对话历史 current_ts = int(time.time() * 1000) self.dialogue_history.append({ "role": "assistant", "content": llm_response, "timestamp": current_ts, "state": self.current_state.name }) self.last_interaction_time = current_ts return llm_response def _get_hours_since_last_interaction(self, precision: int = -1): time_diff = (time.time() * 1000) - self.last_interaction_time hours_passed = time_diff / 1000 / 3600 if precision >= 0: return round(hours_passed, precision) return hours_passed def should_initiate_conversation(self) -> bool: """判断是否应该主动发起对话""" # 如果处于人工介入状态,不应主动发起对话 if self.current_state == DialogueState.HUMAN_INTERVENTION: return False hours_passed = self._get_hours_since_last_interaction() # 获取当前时间上下文 time_context = self.get_current_time_context() # 根据用户交互频率偏好设置不同的阈值 interaction_frequency = self.user_profile.get("interaction_frequency", "medium") # 设置不同偏好的交互时间阈值(小时) thresholds = { "low": 24, # 低频率:一天一次 "medium": 12, # 中频率:半天一次 "high": 6 # 高频率:大约6小时一次 } threshold = thresholds.get(interaction_frequency, 12) # 如果足够时间已经过去 if hours_passed >= threshold: # 根据时间上下文决定主动交互的状态 if time_context in [TimeContext.EARLY_MORNING, TimeContext.MORNING, TimeContext.NOON, TimeContext.AFTERNOON, TimeContext.EVENING]: return True return False def is_in_human_intervention(self) -> bool: """检查是否处于人工介入状态""" return self.current_state == DialogueState.HUMAN_INTERVENTION def get_prompt_context(self, user_message) -> Dict: # 获取当前时间上下文 time_context = self.get_current_time_context() context = { "user_profile": self.user_profile, "current_state": self.current_state.name, "previous_state": self.previous_state.name if self.previous_state else None, "current_time_period": time_context.description, # "dialogue_history": self.dialogue_history[-10:], "last_interaction_interval": self._get_hours_since_last_interaction(2), "if_first_interaction": False, "if_active_greeting": False if user_message else True, **self.user_profile } # 获取长期记忆 relevant_memories = self.vector_memory.retrieve_relevant_memories(user_message) context["long_term_memory"] = { "relevant_conversations": relevant_memories } return context def _select_prompt(self, state): state_to_prompt_map = { DialogueState.GREETING: GENERAL_GREETING_PROMPT, DialogueState.CHITCHAT: GENERAL_GREETING_PROMPT, } return state_to_prompt_map[state] def _select_coze_bot(self, state): state_to_bot_map = { DialogueState.GREETING: '7479005417885417487', DialogueState.CHITCHAT: '7479005417885417487' } return state_to_bot_map[state] def _create_system_message(self, prompt_context): prompt_template = self._select_prompt(self.current_state) prompt = prompt_template.format(**prompt_context) return {'role': 'system', 'content': prompt} def build_chat_configuration( self, user_message: Optional[str] = None, chat_service_type: ChatServiceType = ChatServiceType.OPENAI_COMPATIBLE ) -> Dict: """ 参数: user_message: 当前用户消息,如果是主动交互则为None 返回: 消息列表 """ dialogue_history = self.dialogue_history[-10:] \ if len(self.dialogue_history) > 10 \ else self.dialogue_history messages = [] config = {} prompt_context = self.get_prompt_context(user_message) if chat_service_type == ChatServiceType.OPENAI_COMPATIBLE: system_message = self._create_system_message(prompt_context) messages.append(system_message) for entry in dialogue_history: role = entry['role'] messages.append({ "role": role, "content": entry["content"] }) elif chat_service_type == ChatServiceType.COZE_CHAT: for entry in dialogue_history: role = entry['role'] if role == 'user': messages.append(cozepy.Message.build_user_question_text(entry["content"])) elif role == 'assistant': messages.append(cozepy.Message.build_assistant_answer(entry['content'])) custom_variables = {} for k, v in prompt_context.items(): custom_variables[k] = str(v) custom_variables.pop('user_profile', None) config['custom_variables'] = custom_variables config['bot_id'] = self._select_coze_bot(self.current_state) if not user_message: messages.append(cozepy.Message.build_user_question_text('请开始对话')) config['messages'] = messages return config