#! /usr/bin/env python # -*- coding: utf-8 -*- # vim:fenc=utf-8 import random from enum import Enum from typing import Dict, List, Optional, Tuple, Any from datetime import datetime import time import textwrap import pymysql.cursors import cozepy from sqlalchemy.orm import sessionmaker, Session from pqai_agent import configs from pqai_agent.data_models.agent_push_record import AgentPushRecord from pqai_agent.logging_service import logger from pqai_agent.database import MySQLManager from pqai_agent import chat_service, prompt_templates from pqai_agent.history_dialogue_service import HistoryDialogueService from pqai_agent.chat_service import ChatServiceType from pqai_agent.mq_message import MessageType, MqMessage from pqai_agent.toolkit.lark_alert_for_human_intervention import LarkAlertForHumanIntervention from pqai_agent.toolkit.lark_sheet_record_for_human_intervention import LarkSheetRecordForHumanIntervention from pqai_agent.user_manager import UserManager from pqai_agent.utils import prompt_utils 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(int, Enum): INITIALIZED = 0 GREETING = 1 # 问候状态 CHITCHAT = 2 # 闲聊状态 CLARIFICATION = 3 # 澄清状态 FAREWELL = 4 # 告别状态 HUMAN_INTERVENTION = 5 # 人工介入状态 MESSAGE_AGGREGATING = 6 # 等待消息状态 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 DialogueStateChangeType(int, Enum): STATE = 0 INTERACTION_TIME = 1 DIALOGUE_HISTORY = 2 class DialogueStateChange: def __init__(self, event_type: DialogueStateChangeType,old: Any, new: Any): self.event_type = event_type self.old = old self.new = new class DialogueStateCache: def __init__(self): self.config = configs.get() self.db = MySQLManager(self.config['storage']['agent_state']['mysql']) self.table = self.config['storage']['agent_state']['table'] def get_state(self, staff_id: str, user_id: str) -> Tuple[DialogueState, DialogueState]: query = f"SELECT current_state, previous_state FROM {self.table} WHERE staff_id=%s AND user_id=%s" data = self.db.select(query, pymysql.cursors.DictCursor, (staff_id, user_id)) if not data: logger.warning(f"staff[{staff_id}], user[{user_id}]: agent state not found") state = DialogueState.INITIALIZED previous_state = DialogueState.INITIALIZED self.set_state(staff_id, user_id, state, previous_state) else: state = DialogueState(data[0]['current_state']) previous_state = DialogueState(data[0]['previous_state']) return state, previous_state def set_state(self, staff_id: str, user_id: str, state: DialogueState, previous_state: DialogueState): if self.config.get('debug_flags', {}).get('disable_database_write', False): return query = f"INSERT INTO {self.table} (staff_id, user_id, current_state, previous_state)" \ f" VALUES (%s, %s, %s, %s) " \ f"ON DUPLICATE KEY UPDATE current_state=%s, previous_state=%s" rows = self.db.execute(query, (staff_id, user_id, state.value, previous_state.value, state.value, previous_state.value)) logger.debug("staff[{}], user[{}]: set state: {}, previous state: {}, rows affected: {}" .format(staff_id, user_id, state, previous_state, rows)) class DialogueManager: def __init__(self, staff_id: str, user_id: str, user_manager: UserManager, state_cache: DialogueStateCache, AgentDBSession: sessionmaker[Session]): config = configs.get() self.staff_id = staff_id self.user_id = user_id self.user_manager = user_manager self.state_cache = state_cache self.current_state = DialogueState.GREETING self.previous_state = DialogueState.INITIALIZED # 目前实际仅用作调试，拼装prompt时使用history_dialogue_service获取 self.dialogue_history = [] self.user_profile = self.user_manager.get_user_profile(user_id) self.staff_profile = self.user_manager.get_staff_profile(staff_id) # FIXME: 交互时间和对话记录都涉及到回滚 self.last_interaction_time_ms = 0 self.last_active_interaction_time_sec = 0 self.human_intervention_triggered = False self.vector_memory = DummyVectorMemoryManager(user_id) self.message_aggregation_sec = config.get('agent_behavior', {}).get('message_aggregation_sec', 5) self.unprocessed_messages = [] self.history_dialogue_service = HistoryDialogueService( config['storage']['history_dialogue']['api_base_url'] ) self.AgentDBSession = AgentDBSession self._recover_state() # 由于本地状态管理过于复杂，引入事务机制做状态回滚 self._uncommited_state_change = [] @staticmethod def get_time_context(current_hour=None) -> TimeContext: """获取当前时间上下文""" if not current_hour: current_hour = datetime.now().hour if 5 <= current_hour < 7: return TimeContext.EARLY_MORNING elif 7 <= current_hour < 11: return TimeContext.MORNING elif 11 <= 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 is_valid(self): if not self.staff_profile.get('name', None) and not self.staff_profile.get('agent_name', None): return False return True def refresh_profile(self): self.staff_profile = self.user_manager.get_staff_profile(self.staff_id) def _recover_state(self): self.current_state, self.previous_state = self.state_cache.get_state(self.staff_id, self.user_id) # 从数据库恢复对话状态 minutes_to_get = 5 * 24 * 60 self.dialogue_history = self.history_dialogue_service.get_dialogue_history( self.staff_id, self.user_id, minutes_to_get) if self.dialogue_history: self.last_interaction_time_ms = self.dialogue_history[-1]['timestamp'] if self.current_state == DialogueState.MESSAGE_AGGREGATING: # 需要恢复未处理对话，找到dialogue_history中最后未处理的user消息 for entry in reversed(self.dialogue_history): if entry['role'] == 'user': self.unprocessed_messages.append(entry['content']) break else: # 默认设置 self.last_interaction_time_ms = int(time.time() * 1000) - minutes_to_get * 60 * 1000 with self.AgentDBSession() as session: # 读取数据库中的最后一次交互时间 query = session.query(AgentPushRecord).filter( AgentPushRecord.staff_id == self.staff_id, AgentPushRecord.user_id == self.user_id ).order_by(AgentPushRecord.timestamp.desc()).first() if query: self.last_active_interaction_time_sec = query.timestamp fmt_time = datetime.fromtimestamp(self.last_interaction_time_ms / 1000).strftime("%Y-%m-%d %H:%M:%S") logger.debug(f"staff[{self.staff_id}], user[{self.user_id}]: state: {self.current_state.name}, last_interaction: {fmt_time}") def update_interaction_time(self, timestamp_ms: int): self._uncommited_state_change.append(DialogueStateChange( DialogueStateChangeType.INTERACTION_TIME, self.last_interaction_time_ms, timestamp_ms )) self.last_interaction_time_ms = timestamp_ms def append_dialogue_history(self, message: Dict): self._uncommited_state_change.append(DialogueStateChange( DialogueStateChangeType.DIALOGUE_HISTORY, None, 1 )) self.dialogue_history.append(message) def persist_state(self): """持久化对话状态，只有当前状态处理成功后才应该做持久化""" self.commit() config = configs.get() if config.get('debug_flags', {}).get('disable_database_write', False): return self.state_cache.set_state(self.staff_id, self.user_id, self.current_state, self.previous_state) def rollback_state(self): logger.info(f"staff[{self.staff_id}], user[{self.user_id}]: reverse state") for entry in reversed(self._uncommited_state_change): if entry.event_type == DialogueStateChangeType.STATE: self.current_state, self.previous_state = entry.old elif entry.event_type == DialogueStateChangeType.INTERACTION_TIME: self.last_interaction_time_ms = entry.old elif entry.event_type == DialogueStateChangeType.DIALOGUE_HISTORY: self.dialogue_history.pop() else: logger.error(f"unimplemented type: [{entry.event_type}]") self._uncommited_state_change.clear() def commit(self): self._uncommited_state_change.clear() def do_state_change(self, state: DialogueState): state_backup = (self.current_state, self.previous_state) if self.current_state == DialogueState.MESSAGE_AGGREGATING: # MESSAGE_AGGREGATING不能成为previous_state，仅使用state_backup做回退 self.current_state = state else: self.previous_state = self.current_state self.current_state = state self._uncommited_state_change.append(DialogueStateChange( DialogueStateChangeType.STATE, state_backup, (self.current_state, self.previous_state) )) def update_state(self, message: MqMessage) -> Tuple[bool, Optional[str]]: """根据用户消息更新对话状态，并返回是否需要发起回复 及下一条需处理的用户消息""" message_text = message.content message_ts = message.sendTime # 如果当前已经是人工介入状态，根据消息类型决定保持/退出 if self.current_state == DialogueState.HUMAN_INTERVENTION: if message.type == MessageType.HUMAN_INTERVENTION_END: self.resume_from_human_intervention() # 恢复状态，但无需Agent产生回复 return False, None else: self.append_dialogue_history({ "role": "user", "content": message_text, "timestamp": int(time.time() * 1000), "state": self.current_state.name }) return False, message_text if message.type == MessageType.ENTER_HUMAN_INTERVENTION: logger.info(f"staff[{self.staff_id}], user[{self.user_id}]: human intervention triggered") self.do_state_change(DialogueState.HUMAN_INTERVENTION) return False, None # 检查是否处于消息聚合状态 if self.current_state == DialogueState.MESSAGE_AGGREGATING: # 收到的是特殊定时触发的空消息，且在聚合中，且已经超时，继续处理 if message.type == MessageType.AGGREGATION_TRIGGER: if message_ts - self.last_interaction_time_ms > self.message_aggregation_sec * 1000: logger.debug(f"staff[{self.staff_id}], user[{self.user_id}]: exit aggregation waiting") else: logger.debug(f"staff[{self.staff_id}], user[{self.user_id}]: continue aggregation waiting") return False, message_text else: # 非空消息，更新最后交互时间，保持消息聚合状态 if message_text: self.unprocessed_messages.append(message_text) self.update_interaction_time(message_ts) return False, message_text else: if message.type == MessageType.AGGREGATION_TRIGGER: # 未在聚合状态中，收到的聚合触发消息为过时消息，不应当处理 logger.warning(f"staff[{self.staff_id}], user[{self.user_id}]: received {message.type} in state {self.current_state}") return False, None if message.type == MessageType.HUMAN_INTERVENTION_END: # 未在人工介入状态中，收到的人工介入结束事件为过时消息，不应当处理 logger.warning(f"staff[{self.staff_id}], user[{self.user_id}]: received {message.type} in state {self.current_state}") return False, None if message.type != MessageType.AGGREGATION_TRIGGER and self.message_aggregation_sec > 0: # 收到有内容的用户消息，切换到消息聚合状态 self.do_state_change(DialogueState.MESSAGE_AGGREGATING) self.unprocessed_messages.append(message_text) # 更新最后交互时间 if message_text: self.update_interaction_time(message_ts) return False, message_text # 获得未处理的聚合消息，并清空未处理队列 if message_text: self.unprocessed_messages.append(message_text) if self.unprocessed_messages: message_text = '\n'.join(self.unprocessed_messages) self.unprocessed_messages.clear() # 实际上这里message_text并不会被最终送入LLM，只是用来做状态判断 # 根据消息内容和当前状态确定新状态 new_state = self._determine_state_from_message(message_text) # 更新状态 self.do_state_change(new_state) if message_text: self.update_interaction_time(message_ts) self.append_dialogue_history({ "role": "user", "content": message_text, "timestamp": message_ts, "state": self.current_state.name }) return True, message_text def _determine_state_from_message(self, message_text: Optional[str]) -> DialogueState: """根据消息内容确定对话状态""" if not message_text: logger.warning(f"staff[{self.staff_id}], user[{self.user_id}]: empty message") return self.current_state # 简单的规则-关键词匹配 message_lower = message_text.lower() # 问候检测 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 _send_alert(self, alert_type: str, reason: Optional[str] = None) -> None: time_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") staff_info = f"{self.staff_profile.get('name', '未知')}[{self.staff_id}]" user_info = f"{self.user_profile.get('nickname', '未知')}[{self.user_id}]" alert_message = f""" {alert_type}告警 员工: {staff_info} 用户: {user_info} 时间: {time_str} 原因：{reason if reason else "未知"} 最近对话: """ alert_message = textwrap.dedent(alert_message).strip() # 添加最近的对话记录 recent_dialogues = self.dialogue_history[-5:] dialogue_to_send = [] role_map = {'assistant': '客服', 'user': '用户'} for dialogue in recent_dialogues: if not dialogue['content']: continue role = dialogue['role'] if role not in role_map: continue dialogue_to_send.append(f"[{role_map[role]}]{dialogue['content']}") alert_message += '\n'.join(dialogue_to_send) if alert_type == '人工介入': ack_url = "http://ai-wechat-hook.piaoquantv.com/manage/insertEvent?" \ f"sender={self.user_id}&receiver={self.staff_id}&type={MessageType.HUMAN_INTERVENTION_END.value}&content=OPERATION" else: ack_url = None LarkAlertForHumanIntervention().send_lark_alert_for_human_intervention(alert_message, ack_url) if alert_type == '人工介入': LarkSheetRecordForHumanIntervention().send_lark_sheet_record_for_human_intervention( staff_info, user_info, '\n'.join(dialogue_to_send), reason ) def resume_from_human_intervention(self) -> None: """从人工介入状态恢复""" if self.current_state == DialogueState.HUMAN_INTERVENTION: self.do_state_change(DialogueState.CHITCHAT) def generate_response(self, llm_response: str) -> Optional[str]: """ 处理LLM的响应，更新对话状态和对话历史。 注意：所有的LLM响应都必须经过这个函数来处理！ :param llm_response: :return: """ if '<人工介入>' in llm_response: reason = llm_response.replace('<人工介入>', '') logger.warning(f'staff[{self.staff_id}], user[{self.user_id}]: human intervention triggered, reason: {reason}') self.do_state_change(DialogueState.HUMAN_INTERVENTION) self._send_alert('人工介入', reason) return None if '<结束>' in llm_response or '<负向情绪结束>' in llm_response: logger.warning(f'staff[{self.staff_id}], user[{self.user_id}]: conversation ended') self.do_state_change(DialogueState.FAREWELL) if '<负向情绪结束>' in llm_response: self._send_alert("用户负向情绪") return None """根据当前状态处理LLM响应，如果处于人工介入状态则返回None""" # 如果处于人工介入状态，不生成回复 if self.current_state == DialogueState.HUMAN_INTERVENTION: return None # 记录响应到对话历史 message_ts = int(time.time() * 1000) self.append_dialogue_history({ "role": "assistant", "type": MessageType.TEXT, "content": llm_response, "timestamp": message_ts, "state": self.current_state.name }) self.update_interaction_time(message_ts) return llm_response def generate_multimodal_response(self, item: Dict) -> Optional[Dict]: """ 处理LLM的多模态响应，更新对话状态和对话历史。 注意：所有的LLM多模态响应都必须经过这个函数来处理！ :param item: 包含多模态内容的字典 :return: None """ if self.current_state == DialogueState.HUMAN_INTERVENTION: return None raw_type = item.get("type", "text") if isinstance(raw_type, str): item["type"] = MessageType.from_str(raw_type) if item["type"] == MessageType.TEXT: if '<人工介入>' in item["content"]: reason = item["content"].replace('<人工介入>', '') logger.warning(f'staff[{self.staff_id}], user[{self.user_id}]: human intervention triggered, reason: {reason}') self.do_state_change(DialogueState.HUMAN_INTERVENTION) self._send_alert('人工介入', reason) return None if '<结束>' in item["content"] or '<负向情绪结束>' in item["content"]: logger.warning(f'staff[{self.staff_id}], user[{self.user_id}]: conversation ended') self.do_state_change(DialogueState.FAREWELL) if '<负向情绪结束>' in item["content"]: self._send_alert("用户负向情绪") return None # 记录响应到对话历史 message_ts = int(time.time() * 1000) self.append_dialogue_history({ "role": "assistant", "type": item["type"], "content": item["content"], "timestamp": message_ts, "state": self.current_state.name }) self.update_interaction_time(message_ts) return item def _get_hours_since_last_interaction(self, precision: int = -1): time_diff = (time.time() * 1000) - self.last_interaction_time_ms hours_passed = time_diff / 1000 / 3600 if precision >= 0: return round(hours_passed, precision) return hours_passed def update_last_active_interaction_time(self, timestamp_sec: int): # 只需更新本地时间，重启时可从数据库恢复 self.last_active_interaction_time_sec = timestamp_sec 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_time_context() # 根据用户交互频率偏好设置不同的阈值 interaction_frequency = self.user_profile.get("interaction_frequency", "medium") if interaction_frequency == 'stopped': return False # 设置不同偏好的交互时间阈值（小时） thresholds = { "low": 48, "medium": 24, "high": 12 } threshold = thresholds.get(interaction_frequency, 24) #FIXME 05-21 临时策略，两次主动发起至少48小时 if time.time() - self.last_active_interaction_time_sec < 2 * 24 * 3600: logger.debug(f"staff[{self.staff_id}], user[{self.user_id}]: last active interaction time too short") return False if hours_passed < threshold: return False # 根据时间上下文决定主动交互的状态 if self.is_time_suitable_for_active_conversation(time_context): return True return False @staticmethod def is_time_suitable_for_active_conversation(time_context=None) -> bool: if configs.get_env() == 'dev': return True if not time_context: time_context = DialogueManager.get_time_context() if time_context in [TimeContext.MORNING, TimeContext.NOON, TimeContext.AFTERNOON]: 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_time_context() # 刷新用户画像 self.user_profile = self.user_manager.get_user_profile(self.user_id) # 刷新员工画像（不一定需要） self.staff_profile = self.user_manager.get_staff_profile(self.staff_id) # 员工画像添加前缀，避免冲突，实现Coze Prompt模板的平滑升级 legacy_staff_profile = {} for key in self.staff_profile: legacy_staff_profile[f'agent_{key}'] = self.staff_profile[key] current_datetime = datetime.now() context = { "current_state": self.current_state.name, "previous_state": self.previous_state.name, "current_time_period": time_context.description, "current_hour": current_datetime.hour, "current_time": current_datetime.strftime("%H:%M:%S"), "current_date": current_datetime.strftime("%Y-%m-%d"), "current_datetime": current_datetime.strftime("%Y-%m-%d %H:%M:%S"), "last_interaction_interval": self._get_hours_since_last_interaction(2), "if_first_interaction": True if self.previous_state == DialogueState.INITIALIZED else False, "if_active_greeting": False if user_message else True, "formatted_staff_profile": prompt_utils.format_agent_profile(self.staff_profile), "formatted_user_profile": prompt_utils.format_user_profile(self.user_profile), **self.user_profile, **legacy_staff_profile } # 获取长期记忆 relevant_memories = self.vector_memory.retrieve_relevant_memories(user_message) context["long_term_memory"] = { "relevant_conversations": relevant_memories } return context @staticmethod def _select_prompt(state): state_to_prompt_map = { DialogueState.GREETING: prompt_templates.GENERAL_GREETING_PROMPT, DialogueState.CHITCHAT: prompt_templates.CHITCHAT_PROMPT_COZE, DialogueState.FAREWELL: prompt_templates.GENERAL_GREETING_PROMPT } return state_to_prompt_map[state] @staticmethod def _select_coze_bot(state, dialogue: List[Dict], multimodal=False): state_to_bot_map = { DialogueState.GREETING: '7486112546798780425', DialogueState.CHITCHAT: '7491300566573301770', DialogueState.FAREWELL: '7491300566573301770', } if multimodal: state_to_bot_map = { DialogueState.GREETING: '7496772218198900770', DialogueState.CHITCHAT: '7495692989504438308', DialogueState.FAREWELL: '7491300566573301770', } return state_to_bot_map[state] @staticmethod def need_multimodal_model(dialogue: List[Dict], max_message_to_use: int = 10): # 当前仅为简单实现 recent_messages = dialogue[-max_message_to_use:] ret = False for entry in recent_messages: if entry.get('type') in (MessageType.IMAGE_GW, MessageType.IMAGE_QW, MessageType.GIF): ret = True break return ret 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} @staticmethod def compose_chat_messages_openai_compatible(dialogue_history, current_time, multimodal=False): messages = [] for entry in dialogue_history: role = entry['role'] msg_type = entry.get('type', MessageType.TEXT) fmt_time = DialogueManager.format_timestamp(entry['timestamp']) if msg_type in (MessageType.IMAGE_GW, MessageType.IMAGE_QW, MessageType.GIF): if multimodal: messages.append({ "role": role, "content": [ {"type": "image_url", "image_url": {"url": entry["content"]}} ] }) else: logger.warning("Image in non-multimodal mode") messages.append({ "role": role, "content": "[{}] {}".format(fmt_time, '[图片]') }) else: messages.append({ "role": role, "content": '[{}] {}'.format(fmt_time, entry["content"]) }) # 添加一条前缀用于 约束时间场景 msg_prefix = '[{}]'.format(current_time) messages.append({'role': 'assistant', 'content': msg_prefix}) return messages @staticmethod def compose_chat_messages_coze(dialogue_history, current_time, staff_id, user_id): messages = [] # 如果system后的第1条消息不为user，需要在最开始补一条user消息，否则会吞assistant消息 if len(dialogue_history) > 0 and dialogue_history[0]['role'] != 'user': fmt_time = DialogueManager.format_timestamp(dialogue_history[0]['timestamp']) messages.append(cozepy.Message.build_user_question_text(f'[{fmt_time}] ')) # coze最后一条消息必须为user，且可能吞掉连续的user消息，故强制增加一条空消息（可参与合并） dialogue_history.append({ 'role': 'user', 'content': ' ', 'timestamp': int(datetime.strptime(current_time, '%Y-%m-%d %H:%M:%S').timestamp() * 1000), }) # 将连续的同一角色的消息做聚合，避免coze吞消息 messages_to_aggr = [] objects_to_aggr = [] last_message_role = None for entry in dialogue_history: if not entry['content']: logger.warning("staff[{}], user[{}], role[{}]: empty content in dialogue history".format( staff_id, user_id, entry['role'] )) continue role = entry['role'] if role != last_message_role: if objects_to_aggr: if last_message_role != 'user': pass else: text_message = '\n'.join(messages_to_aggr) object_string_list = [] for object_entry in objects_to_aggr: # FIXME: 其它消息类型的支持 object_string_list.append(cozepy.MessageObjectString.build_image(file_url=object_entry['content'])) object_string_list.append(cozepy.MessageObjectString.build_text(text_message)) messages.append(cozepy.Message.build_user_question_objects(object_string_list)) elif messages_to_aggr: aggregated_message = '\n'.join(messages_to_aggr) messages.append(DialogueManager.build_chat_message( last_message_role, aggregated_message, ChatServiceType.COZE_CHAT)) objects_to_aggr = [] messages_to_aggr = [] last_message_role = role if entry.get('type', MessageType.TEXT) in (MessageType.IMAGE_GW, MessageType.IMAGE_QW, MessageType.GIF): # 多模态消息必须用特殊的聚合方式，一个object_string数组中只能有一个文字消息，但可以有多个图片 if role == 'user': objects_to_aggr.append(entry) else: logger.warning("staff[{}], user[{}]: unsupported message type [{}] in assistant role".format( staff_id, user_id, entry['type'] )) else: messages_to_aggr.append(DialogueManager.format_dialogue_content(entry)) # 如果有未聚合的object消息，需要特殊处理 if objects_to_aggr: if last_message_role != 'user': pass else: text_message = '\n'.join(messages_to_aggr) object_string_list = [] for object_entry in objects_to_aggr: # FIXME: 其它消息类型的支持 object_string_list.append(cozepy.MessageObjectString.build_image(file_url=object_entry['content'])) object_string_list.append(cozepy.MessageObjectString.build_text(text_message)) messages.append(cozepy.Message.build_user_question_objects(object_string_list)) elif messages_to_aggr: aggregated_message = '\n'.join(messages_to_aggr) messages.append(DialogueManager.build_chat_message( last_message_role, aggregated_message, ChatServiceType.COZE_CHAT)) # 从末尾开始往前遍历，如果assistant曾经回复“无法回答”，则清除当前消息和前一条用户消息 idx = len(messages) - 1 while idx >= 0: if messages[idx].role == 'assistant' and '无法回答' in messages[idx].content: messages.pop(idx) idx -= 1 if idx >= 0: messages.pop(idx) idx -= 1 else: idx -= 1 return messages def build_active_greeting_config(self, user_tags: List[str]): # FIXME: 这里的抽象不好，短期支持人为配置实验 # 由于产运要求，指定使用GPT-4o模型 chat_config = {'user_id': self.user_id, 'model_name': chat_service.OPENAI_MODEL_GPT_4o} prompt_context = self.get_prompt_context(None) current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') system_message = {'role': 'system', 'content': 'You are a helpful AI assistant.'} # TODO: 随机选择一个prompt 或 带策略选择 或根据用户标签选择 # TODO：需要区分用户是否有历史交互、是否发送过相似内容 greeting_prompts = [ prompt_templates.GREETING_WITH_IMAGE_GAME, prompt_templates.GREETING_WITH_NAME_POETRY, prompt_templates.GREETING_WITH_AVATAR_STORY ] # 默认随机选择 selected_prompt = greeting_prompts[random.randint(0, len(greeting_prompts) - 1)] # 实验配置 tag_to_greeting_map = { '04W4-AA-1': prompt_templates.GREETING_WITH_NAME_POETRY, '04W4-AA-2': prompt_templates.GREETING_WITH_AVATAR_STORY, '04W4-AA-3': prompt_templates.GREETING_WITH_INTEREST_QUERY, '04W4-AA-4': prompt_templates.GREETING_WITH_CALENDAR, } for tag in user_tags: if tag in tag_to_greeting_map: selected_prompt = tag_to_greeting_map[tag] prompt = selected_prompt.format(**prompt_context) user_message = {'role': 'user', 'content': prompt} messages = [system_message, user_message] if selected_prompt in ( prompt_templates.GREETING_WITH_AVATAR_STORY, prompt_templates.GREETING_WITH_INTEREST_QUERY, ): messages.append({ "role": 'user', "content": [ {"type": "image_url", "image_url": {"url": self.user_profile['avatar']}} ] }) chat_config['use_multimodal_model'] = True chat_config['messages'] = messages return chat_config def build_chat_configuration( self, user_message: Optional[str] = None, chat_service_type: ChatServiceType = ChatServiceType.OPENAI_COMPATIBLE, overwrite_context: Optional[Dict] = None ) -> Dict: """ 参数: user_message: 当前用户消息，如果是主动交互则为None 返回: 消息列表 """ dialogue_history = self.history_dialogue_service.get_dialogue_history(self.staff_id, self.user_id) logger.debug("staff[{}], user[{}], recent dialogue_history: {}".format( self.staff_id, self.user_id, dialogue_history[-20:] )) messages = [] config = { 'user_id': self.user_id } prompt_context = self.get_prompt_context(user_message) if overwrite_context: prompt_context.update(overwrite_context) # FIXME(zhoutian): time in string type current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') if overwrite_context and 'current_time' in overwrite_context: current_time = overwrite_context.get('current_time') need_multimodal = self.need_multimodal_model(dialogue_history) config['use_multimodal_model'] = need_multimodal if chat_service_type == ChatServiceType.OPENAI_COMPATIBLE: system_message = self._create_system_message(prompt_context) messages.append(system_message) messages.extend(self.compose_chat_messages_openai_compatible(dialogue_history, current_time, need_multimodal)) elif chat_service_type == ChatServiceType.COZE_CHAT: dialogue_history = dialogue_history[-95:] # Coze最多支持100条，还需要附加系统消息 messages = self.compose_chat_messages_coze(dialogue_history, current_time, self.staff_id, self.user_id) 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, dialogue_history, need_multimodal) #FIXME(zhoutian): 临时报警 if user_message and not messages: logger.error(f"staff[{self.staff_id}], user[{self.user_id}]: inconsistency in messages") config['messages'] = messages return config @staticmethod def format_timestamp(timestamp_ms): return datetime.fromtimestamp(timestamp_ms / 1000).strftime("%Y-%m-%d %H:%M:%S") @staticmethod def format_dialogue_content(dialogue_entry): fmt_time = DialogueManager.format_timestamp(dialogue_entry['timestamp']) content = '[{}] {}'.format(fmt_time, dialogue_entry['content']) return content @staticmethod def build_chat_message(role, content, chat_service_type: ChatServiceType): if chat_service_type == ChatServiceType.COZE_CHAT: if role == 'user': return cozepy.Message.build_user_question_text(content) elif role == 'assistant': return cozepy.Message.build_assistant_answer(content) else: return {'role': role, 'content': content} if __name__ == '__main__': state_cache = DialogueStateCache() state_cache.set_state('1688854492669990', '7881302581935903', DialogueState.CHITCHAT, DialogueState.GREETING)