AgentCoreService/dialogue_manager.py

#! /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 configs
import cozepy
from history_dialogue_service import HistoryDialogueService

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):
        config = configs.get()

        self.staff_id = staff_id
        self.user_id = user_id
        self.user_manager = user_manager
        self.current_state = DialogueState.GREETING
        self.previous_state = None
        # 目前实际仅用作调试，拼装prompt时使用history_dialogue_service获取
        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 = config.get('message_aggregation_sec', 5)
        self.unprocessed_messages = []
        self.history_dialogue_service = HistoryDialogueService(
            config['storage']['history_dialogue']['api_base_url']
        )
        self._reset_interaction_time()

    def _reset_interaction_time(self):
        last_message = self.history_dialogue_service.get_dialogue_history(self.staff_id, self.user_id, 1)
        if last_message:
            self.last_interaction_time = last_message[0]['timestamp']
        else:
            # 默认设置为24小时前
            self.last_interaction_time = int(time.time() * 1000) - 24 * 3600 * 1000

    @staticmethod
    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.history_dialogue_service.get_dialogue_history(self.staff_id, self.user_id, 5)
            }

            # 更新用户资料中的人工介入历史
            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_id, 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.history_dialogue_service.get_dialogue_history(self.staff_id, self.user_id, 5)
        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()
        # 刷新用户画像
        self.user_profile = self.user_manager.get_user_profile(self.user_id)

        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.history_dialogue_service.get_dialogue_history(self.staff_id, self.user_id)
        logging.debug("staff[{}], user[{}], dialogue_history: {}".format(
            self.staff_id, self.user_id, 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:
                if not entry['content']:
                    logging.warning("staff[{}], user[{}], role[{}]: empty content in dialogue history".format(
                        self.staff_id, self.user_id, entry['role']
                    ))
                    continue
                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)
            #FIXME(zhoutian): 这种方法并不可靠，需要通过状态来判断
            if not user_message:
                messages.append(cozepy.Message.build_user_question_text('请开始对话'))
            #FIXME(zhoutian): 临时报警
            if user_message and not messages:
                logging.error(f"staff[{self.staff_id}], user[{self.user_id}]: inconsistency in messages")
        config['messages'] = messages

        return config