dialogue_manager.py 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649
  1. #! /usr/bin/env python
  2. # -*- coding: utf-8 -*-
  3. # vim:fenc=utf-8
  4. from enum import Enum, auto
  5. from typing import Dict, List, Optional, Tuple, Any
  6. from datetime import datetime
  7. import time
  8. from logging_service import logger
  9. import pymysql.cursors
  10. import configs
  11. import cozepy
  12. from database import MySQLManager
  13. from history_dialogue_service import HistoryDialogueService
  14. from chat_service import ChatServiceType
  15. from message import MessageType, Message
  16. from user_manager import UserManager
  17. from prompt_templates import *
  18. class DummyVectorMemoryManager:
  19. def __init__(self, user_id):
  20. pass
  21. def add_to_memory(self, conversation):
  22. pass
  23. def retrieve_relevant_memories(self, query, k=3):
  24. return []
  25. class DialogueState(int, Enum):
  26. INITIALIZED = 0
  27. GREETING = 1 # 问候状态
  28. CHITCHAT = 2 # 闲聊状态
  29. CLARIFICATION = 3 # 澄清状态
  30. FAREWELL = 4 # 告别状态
  31. HUMAN_INTERVENTION = 5 # 人工介入状态
  32. MESSAGE_AGGREGATING = 6 # 等待消息状态
  33. class TimeContext(Enum):
  34. EARLY_MORNING = "清晨" # 清晨 (5:00-7:59)
  35. MORNING = "上午" # 上午 (8:00-11:59)
  36. NOON = "中午" # 中午 (12:00-13:59)
  37. AFTERNOON = "下午" # 下午 (14:00-17:59)
  38. EVENING = "晚上" # 晚上 (18:00-21:59)
  39. NIGHT = "深夜" # 夜晚 (22:00-4:59)
  40. def __init__(self, description):
  41. self.description = description
  42. class DialogueStateCache:
  43. def __init__(self):
  44. config = configs.get()
  45. self.db = MySQLManager(config['storage']['agent_state']['mysql'])
  46. self.table = config['storage']['agent_state']['table']
  47. def get_state(self, staff_id: str, user_id: str) -> Tuple[DialogueState, DialogueState]:
  48. query = f"SELECT current_state, previous_state FROM {self.table} WHERE staff_id=%s AND user_id=%s"
  49. data = self.db.select(query, pymysql.cursors.DictCursor, (staff_id, user_id))
  50. if not data:
  51. logger.warning(f"staff[{staff_id}], user[{user_id}]: agent state not found")
  52. state = DialogueState.INITIALIZED
  53. previous_state = DialogueState.INITIALIZED
  54. self.set_state(staff_id, user_id, state, previous_state)
  55. else:
  56. state = DialogueState(data[0]['current_state'])
  57. previous_state = DialogueState(data[0]['previous_state'])
  58. return state, previous_state
  59. def set_state(self, staff_id: str, user_id: str, state: DialogueState, previous_state: DialogueState):
  60. query = f"INSERT INTO {self.table} (staff_id, user_id, current_state, previous_state)" \
  61. f" VALUES (%s, %s, %s, %s) " \
  62. f"ON DUPLICATE KEY UPDATE current_state=%s, previous_state=%s"
  63. rows = self.db.execute(query, (staff_id, user_id, state.value, previous_state.value, state.value, previous_state.value))
  64. logger.debug("staff[{}], user[{}]: set state: {}, previous state: {}, rows affected: {}"
  65. .format(staff_id, user_id, state, previous_state, rows))
  66. class DialogueManager:
  67. def __init__(self, staff_id: str, user_id: str, user_manager: UserManager, state_cache: DialogueStateCache):
  68. config = configs.get()
  69. self.staff_id = staff_id
  70. self.user_id = user_id
  71. self.user_manager = user_manager
  72. self.state_cache = state_cache
  73. self.current_state = DialogueState.GREETING
  74. self.previous_state = DialogueState.INITIALIZED
  75. # 用于消息处理失败时回滚
  76. self.state_backup = (DialogueState.INITIALIZED, DialogueState.INITIALIZED)
  77. # 目前实际仅用作调试,拼装prompt时使用history_dialogue_service获取
  78. self.dialogue_history = []
  79. self.user_profile = self.user_manager.get_user_profile(user_id)
  80. self.staff_profile = self.user_manager.get_staff_profile(staff_id)
  81. # FIXME(zhoutian): last_interaction_time也需要回滚
  82. self.last_interaction_time = 0
  83. self.consecutive_clarifications = 0
  84. self.complex_request_counter = 0
  85. self.human_intervention_triggered = False
  86. self.vector_memory = DummyVectorMemoryManager(user_id)
  87. self.message_aggregation_sec = config.get('agent_behavior', {}).get('message_aggregation_sec', 5)
  88. self.unprocessed_messages = []
  89. self.history_dialogue_service = HistoryDialogueService(
  90. config['storage']['history_dialogue']['api_base_url']
  91. )
  92. self._recover_state()
  93. def _recover_state(self):
  94. self.current_state, self.previous_state = self.state_cache.get_state(self.staff_id, self.user_id)
  95. # 从数据库恢复对话状态
  96. self.dialogue_history = self.history_dialogue_service.get_dialogue_history(self.staff_id, self.user_id)
  97. if self.dialogue_history:
  98. self.last_interaction_time = self.dialogue_history[-1]['timestamp']
  99. else:
  100. # 默认设置为24小时前
  101. self.last_interaction_time = int(time.time() * 1000) - 24 * 3600 * 1000
  102. time_for_read = datetime.fromtimestamp(self.last_interaction_time / 1000).strftime("%Y-%m-%d %H:%M:%S")
  103. logger.debug(f"staff[{self.staff_id}], user[{self.user_id}]: state: {self.current_state.name}, last_interaction: {time_for_read}")
  104. def persist_state(self):
  105. """持久化对话状态,只有当前状态处理成功后才应该做持久化"""
  106. config = configs.get()
  107. if config.get('debug_flags', {}).get('disable_database_write', False):
  108. return
  109. self.state_cache.set_state(self.staff_id, self.user_id, self.current_state, self.previous_state)
  110. def rollback_state(self):
  111. logger.debug("staff[{}], user[{}]: rollback state: {}, previous state: {}".format(
  112. self.staff_id, self.user_id, self.state_backup, self.current_state
  113. ))
  114. self.current_state, self.previous_state = self.state_backup
  115. @staticmethod
  116. def get_time_context(current_hour=None) -> TimeContext:
  117. """获取当前时间上下文"""
  118. if not current_hour:
  119. current_hour = datetime.now().hour
  120. if 5 <= current_hour < 8:
  121. return TimeContext.EARLY_MORNING
  122. elif 8 <= current_hour < 12:
  123. return TimeContext.MORNING
  124. elif 12 <= current_hour < 14:
  125. return TimeContext.NOON
  126. elif 14 <= current_hour < 18:
  127. return TimeContext.AFTERNOON
  128. elif 18 <= current_hour < 22:
  129. return TimeContext.EVENING
  130. else:
  131. return TimeContext.NIGHT
  132. def do_state_change(self, state: DialogueState):
  133. self.state_backup = (self.current_state, self.previous_state)
  134. if self.current_state == DialogueState.MESSAGE_AGGREGATING:
  135. # MESSAGE_AGGREGATING不能成为previous_state,仅使用state_backup做回退
  136. self.current_state = state
  137. else:
  138. self.previous_state = self.current_state
  139. self.current_state = state
  140. def update_state(self, message: Message) -> Tuple[bool, Optional[str]]:
  141. """根据用户消息更新对话状态,并返回是否需要发起回复 及下一条需处理的用户消息"""
  142. message_text = message.content
  143. message_ts = message.sendTime
  144. # 如果当前已经是人工介入状态,保持该状态
  145. if self.current_state == DialogueState.HUMAN_INTERVENTION:
  146. # 记录对话历史,但不改变状态
  147. self.dialogue_history.append({
  148. "role": "user",
  149. "content": message_text,
  150. "timestamp": int(time.time() * 1000),
  151. "state": self.current_state.name
  152. })
  153. return False, message_text
  154. # 检查是否处于消息聚合状态
  155. if self.current_state == DialogueState.MESSAGE_AGGREGATING:
  156. # 收到的是特殊定时触发的空消息,且在聚合中,且已经超时,继续处理
  157. if message.type == MessageType.AGGREGATION_TRIGGER \
  158. and message_ts - self.last_interaction_time > self.message_aggregation_sec * 1000:
  159. logger.debug("user_id: {}, last interaction time: {}".format(
  160. self.user_id, datetime.fromtimestamp(self.last_interaction_time / 1000)))
  161. else:
  162. # 非空消息,更新最后交互时间,保持消息聚合状态
  163. if message_text:
  164. self.unprocessed_messages.append(message_text)
  165. self.last_interaction_time = message_ts
  166. return False, message_text
  167. else:
  168. if message.type == MessageType.AGGREGATION_TRIGGER:
  169. # 未在聚合状态中,收到的聚合触发消息为过时消息,不应当处理
  170. logger.warning(f"staff[{self.staff_id}], user[{self.user_id}]: received {message.type} in state {self.current_state}")
  171. return False, None
  172. if message.type != MessageType.AGGREGATION_TRIGGER and self.message_aggregation_sec > 0:
  173. # 收到有内容的用户消息,切换到消息聚合状态
  174. self.do_state_change(DialogueState.MESSAGE_AGGREGATING)
  175. self.unprocessed_messages.append(message_text)
  176. # 更新最后交互时间
  177. if message_text:
  178. self.last_interaction_time = message_ts
  179. return False, message_text
  180. # 获得未处理的聚合消息,并清空未处理队列
  181. if message_text:
  182. self.unprocessed_messages.append(message_text)
  183. if self.unprocessed_messages:
  184. message_text = '\n'.join(self.unprocessed_messages)
  185. self.unprocessed_messages.clear()
  186. # 根据消息内容和当前状态确定新状态
  187. new_state = self._determine_state_from_message(message_text)
  188. # 处理连续澄清的情况
  189. if new_state == DialogueState.CLARIFICATION:
  190. self.consecutive_clarifications += 1
  191. # FIXME(zhoutian): 规则过于简单
  192. if self.consecutive_clarifications >= 10000:
  193. new_state = DialogueState.HUMAN_INTERVENTION
  194. # self._trigger_human_intervention("连续多次澄清请求")
  195. else:
  196. self.consecutive_clarifications = 0
  197. # 更新状态
  198. self.do_state_change(new_state)
  199. if message_text:
  200. self.last_interaction_time = message_ts
  201. self.dialogue_history.append({
  202. "role": "user",
  203. "content": message_text,
  204. "timestamp": message_ts,
  205. "state": self.current_state.name
  206. })
  207. return True, message_text
  208. def _determine_state_from_message(self, message_text: Optional[str]) -> DialogueState:
  209. """根据消息内容确定对话状态"""
  210. if not message_text:
  211. return self.current_state
  212. # 简单的规则-关键词匹配
  213. message_lower = message_text.lower()
  214. # 判断是否是复杂请求
  215. # FIXME(zhoutian): 规则过于简单
  216. # complex_request_keywords = ["帮我", "怎么办", "我需要", "麻烦你", "请帮助", "急", "紧急"]
  217. # if any(keyword in message_lower for keyword in complex_request_keywords):
  218. # self.complex_request_counter += 1
  219. #
  220. # # 如果检测到困难请求且计数达到阈值,触发人工介入
  221. # if self.complex_request_counter >= 1:
  222. # # self._trigger_human_intervention("检测到复杂请求")
  223. # return DialogueState.HUMAN_INTERVENTION
  224. # else:
  225. # # 如果不是复杂请求,重置计数器
  226. # self.complex_request_counter = 0
  227. # 问候检测
  228. greeting_keywords = ["你好", "早上好", "中午好", "晚上好", "嗨", "在吗"]
  229. if any(keyword in message_lower for keyword in greeting_keywords):
  230. return DialogueState.GREETING
  231. # 告别检测
  232. farewell_keywords = ["再见", "拜拜", "晚安", "明天见", "回头见"]
  233. if any(keyword in message_lower for keyword in farewell_keywords):
  234. return DialogueState.FAREWELL
  235. # 澄清请求
  236. clarification_keywords = ["没明白", "不明白", "没听懂", "不懂", "什么意思", "再说一遍"]
  237. if any(keyword in message_lower for keyword in clarification_keywords):
  238. return DialogueState.CLARIFICATION
  239. # 默认为闲聊状态
  240. return DialogueState.CHITCHAT
  241. def _trigger_human_intervention(self, reason: str) -> None:
  242. """触发人工介入"""
  243. if not self.human_intervention_triggered:
  244. self.human_intervention_triggered = True
  245. # 记录人工介入事件
  246. event = {
  247. "timestamp": int(time.time() * 1000),
  248. "reason": reason,
  249. "dialogue_context": self.dialogue_history[-10:]
  250. }
  251. # 更新用户资料中的人工介入历史
  252. if "human_intervention_history" not in self.user_profile:
  253. self.user_profile["human_intervention_history"] = []
  254. self.user_profile["human_intervention_history"].append(event)
  255. self.user_manager.save_user_profile(self.user_id, self.user_profile)
  256. # 发送告警
  257. self._send_human_intervention_alert(reason)
  258. def _send_human_intervention_alert(self, reason: str) -> None:
  259. alert_message = f"""
  260. 人工介入告警
  261. 用户ID: {self.user_id}
  262. 用户昵称: {self.user_profile.get("nickname", "未知")}
  263. 时间: {int(time.time() * 1000)}
  264. 原因: {reason}
  265. 最近对话:
  266. """
  267. # 添加最近的对话记录
  268. recent_dialogues = self.dialogue_history[-10:]
  269. for dialogue in recent_dialogues:
  270. alert_message += f"\n{dialogue['role']}: {dialogue['content']}"
  271. # TODO(zhoutian): 实现发送告警的具体逻辑
  272. logger.warning(alert_message)
  273. def resume_from_human_intervention(self) -> None:
  274. """从人工介入状态恢复"""
  275. if self.current_state == DialogueState.HUMAN_INTERVENTION:
  276. self.current_state = DialogueState.GREETING
  277. self.human_intervention_triggered = False
  278. self.consecutive_clarifications = 0
  279. self.complex_request_counter = 0
  280. # 记录恢复事件
  281. self.dialogue_history.append({
  282. "role": "system",
  283. "content": "已从人工介入状态恢复到自动对话",
  284. "timestamp": int(time.time() * 1000),
  285. "state": self.current_state.name
  286. })
  287. def generate_response(self, llm_response: str) -> Optional[str]:
  288. """根据当前状态处理LLM响应,如果处于人工介入状态则返回None"""
  289. # 如果处于人工介入状态,不生成回复
  290. if self.current_state == DialogueState.HUMAN_INTERVENTION:
  291. return None
  292. # 记录响应到对话历史
  293. current_ts = int(time.time() * 1000)
  294. self.dialogue_history.append({
  295. "role": "assistant",
  296. "content": llm_response,
  297. "timestamp": current_ts,
  298. "state": self.current_state.name
  299. })
  300. self.last_interaction_time = current_ts
  301. return llm_response
  302. def _get_hours_since_last_interaction(self, precision: int = -1):
  303. time_diff = (time.time() * 1000) - self.last_interaction_time
  304. hours_passed = time_diff / 1000 / 3600
  305. if precision >= 0:
  306. return round(hours_passed, precision)
  307. return hours_passed
  308. def should_initiate_conversation(self) -> bool:
  309. """判断是否应该主动发起对话"""
  310. # 如果处于人工介入状态,不应主动发起对话
  311. if self.current_state == DialogueState.HUMAN_INTERVENTION:
  312. return False
  313. hours_passed = self._get_hours_since_last_interaction()
  314. # 获取当前时间上下文
  315. time_context = self.get_time_context()
  316. # 根据用户交互频率偏好设置不同的阈值
  317. interaction_frequency = self.user_profile.get("interaction_frequency", "medium")
  318. if interaction_frequency == 'stopped':
  319. return False
  320. # 设置不同偏好的交互时间阈值(小时)
  321. thresholds = {
  322. "low": 24, # 低频率:一天一次
  323. "medium": 12, # 中频率:半天一次
  324. "high": 6 # 高频率:大约6小时一次
  325. }
  326. threshold = thresholds.get(interaction_frequency, 12)
  327. if hours_passed < threshold:
  328. return False
  329. # 根据时间上下文决定主动交互的状态
  330. if time_context in [TimeContext.MORNING,
  331. TimeContext.NOON, TimeContext.AFTERNOON]:
  332. self.previous_state = self.current_state
  333. self.current_state = DialogueState.GREETING
  334. return True
  335. return False
  336. def is_in_human_intervention(self) -> bool:
  337. """检查是否处于人工介入状态"""
  338. return self.current_state == DialogueState.HUMAN_INTERVENTION
  339. def get_prompt_context(self, user_message) -> Dict:
  340. # 获取当前时间上下文
  341. time_context = self.get_time_context()
  342. # 刷新用户画像
  343. self.user_profile = self.user_manager.get_user_profile(self.user_id)
  344. # 刷新员工画像(不一定需要)
  345. self.staff_profile = self.user_manager.get_staff_profile(self.staff_id)
  346. context = {
  347. "user_profile": self.user_profile,
  348. "current_state": self.current_state.name,
  349. "previous_state": self.previous_state.name,
  350. "current_time_period": time_context.description,
  351. "current_hour": datetime.now().hour,
  352. "last_interaction_interval": self._get_hours_since_last_interaction(2),
  353. "if_first_interaction": True if self.previous_state == DialogueState.INITIALIZED else False,
  354. "if_active_greeting": False if user_message else True,
  355. **self.user_profile,
  356. **self.staff_profile
  357. }
  358. # 获取长期记忆
  359. relevant_memories = self.vector_memory.retrieve_relevant_memories(user_message)
  360. context["long_term_memory"] = {
  361. "relevant_conversations": relevant_memories
  362. }
  363. return context
  364. @staticmethod
  365. def _select_prompt(state):
  366. state_to_prompt_map = {
  367. DialogueState.GREETING: GENERAL_GREETING_PROMPT,
  368. DialogueState.CHITCHAT: CHITCHAT_PROMPT_COZE,
  369. DialogueState.FAREWELL: GENERAL_GREETING_PROMPT
  370. }
  371. return state_to_prompt_map[state]
  372. @staticmethod
  373. def _select_coze_bot(state, dialogue: List[Dict], multimodal=False):
  374. state_to_bot_map = {
  375. DialogueState.GREETING: '7486112546798780425',
  376. DialogueState.CHITCHAT: '7491300566573301770',
  377. DialogueState.FAREWELL: '7491300566573301770',
  378. }
  379. if multimodal:
  380. state_to_bot_map = {
  381. DialogueState.GREETING: '7496772218198900770',
  382. DialogueState.CHITCHAT: '7495692989504438308',
  383. DialogueState.FAREWELL: '7491300566573301770',
  384. }
  385. return state_to_bot_map[state]
  386. @staticmethod
  387. def need_multimodal_model(dialogue: List[Dict], max_message_to_use: int = 10):
  388. # 当前仅为简单实现
  389. recent_messages = dialogue[-max_message_to_use:]
  390. ret = False
  391. for entry in recent_messages:
  392. if entry.get('type') in (MessageType.IMAGE_GW, MessageType.IMAGE_QW):
  393. ret = True
  394. break
  395. return ret
  396. def _create_system_message(self, prompt_context):
  397. prompt_template = self._select_prompt(self.current_state)
  398. prompt = prompt_template.format(**prompt_context)
  399. return {'role': 'system', 'content': prompt}
  400. @staticmethod
  401. def compose_chat_messages_openai_compatible(dialogue_history, current_time, multimodal=False):
  402. messages = []
  403. for entry in dialogue_history:
  404. role = entry['role']
  405. msg_type = entry.get('type', MessageType.TEXT)
  406. fmt_time = DialogueManager.format_timestamp(entry['timestamp'])
  407. if msg_type in (MessageType.IMAGE_GW, MessageType.IMAGE_QW):
  408. if multimodal:
  409. messages.append({
  410. "role": role,
  411. "content": [
  412. {"type": "image_url", "image_url": {"url": entry["content"]}}
  413. ]
  414. })
  415. else:
  416. logger.warning("Image in non-multimodal mode")
  417. messages.append({
  418. "role": role,
  419. "content": "[{}] {}".format(fmt_time, '[图片]')
  420. })
  421. else:
  422. messages.append({
  423. "role": role,
  424. "content": '[{}] {}'.format(fmt_time, entry["content"])
  425. })
  426. # 添加一条前缀用于 约束时间场景
  427. msg_prefix = '[{}]'.format(current_time)
  428. messages.append({'role': 'assistant', 'content': msg_prefix})
  429. return messages
  430. @staticmethod
  431. def compose_chat_messages_coze(dialogue_history, current_time, staff_id, user_id):
  432. messages = []
  433. # 如果system后的第1条消息不为user,需要在最开始补一条user消息,否则会吞assistant消息
  434. if len(dialogue_history) > 0 and dialogue_history[0]['role'] != 'user':
  435. fmt_time = DialogueManager.format_timestamp(dialogue_history[0]['timestamp'])
  436. messages.append(cozepy.Message.build_user_question_text(f'[{fmt_time}] '))
  437. # coze最后一条消息必须为user,且可能吞掉连续的user消息,故强制增加一条空消息(可参与合并)
  438. dialogue_history.append({
  439. 'role': 'user',
  440. 'content': ' ',
  441. 'timestamp': int(datetime.strptime(current_time, '%Y-%m-%d %H:%M:%S').timestamp() * 1000),
  442. })
  443. # 将连续的同一角色的消息做聚合,避免coze吞消息
  444. messages_to_aggr = []
  445. objects_to_aggr = []
  446. last_message_role = None
  447. for entry in dialogue_history:
  448. if not entry['content']:
  449. logger.warning("staff[{}], user[{}], role[{}]: empty content in dialogue history".format(
  450. staff_id, user_id, entry['role']
  451. ))
  452. continue
  453. role = entry['role']
  454. if role != last_message_role:
  455. if objects_to_aggr:
  456. if last_message_role != 'user':
  457. pass
  458. else:
  459. text_message = '\n'.join(messages_to_aggr)
  460. object_string_list = []
  461. for object_entry in objects_to_aggr:
  462. # FIXME: 其它消息类型的支持
  463. object_string_list.append(cozepy.MessageObjectString.build_image(file_url=object_entry['content']))
  464. object_string_list.append(cozepy.MessageObjectString.build_text(text_message))
  465. messages.append(cozepy.Message.build_user_question_objects(object_string_list))
  466. elif messages_to_aggr:
  467. aggregated_message = '\n'.join(messages_to_aggr)
  468. messages.append(DialogueManager.build_chat_message(
  469. last_message_role, aggregated_message, ChatServiceType.COZE_CHAT))
  470. objects_to_aggr = []
  471. messages_to_aggr = []
  472. last_message_role = role
  473. if entry.get('type', MessageType.TEXT) in (MessageType.IMAGE_GW, MessageType.IMAGE_QW):
  474. # 多模态消息必须用特殊的聚合方式,一个object_string数组中只能有一个文字消息,但可以有多个图片
  475. if role == 'user':
  476. objects_to_aggr.append(entry)
  477. else:
  478. logger.warning("staff[{}], user[{}]: unsupported message type [{}] in assistant role".format(
  479. staff_id, user_id, entry['type']
  480. ))
  481. else:
  482. messages_to_aggr.append(DialogueManager.format_dialogue_content(entry))
  483. # 如果有未聚合的object消息,需要特殊处理
  484. if objects_to_aggr:
  485. if last_message_role != 'user':
  486. pass
  487. else:
  488. text_message = '\n'.join(messages_to_aggr)
  489. object_string_list = []
  490. for object_entry in objects_to_aggr:
  491. # FIXME: 其它消息类型的支持
  492. object_string_list.append(cozepy.MessageObjectString.build_image(file_url=object_entry['content']))
  493. object_string_list.append(cozepy.MessageObjectString.build_text(text_message))
  494. messages.append(cozepy.Message.build_user_question_objects(object_string_list))
  495. elif messages_to_aggr:
  496. aggregated_message = '\n'.join(messages_to_aggr)
  497. messages.append(DialogueManager.build_chat_message(
  498. last_message_role, aggregated_message, ChatServiceType.COZE_CHAT))
  499. return messages
  500. def build_chat_configuration(
  501. self,
  502. user_message: Optional[str] = None,
  503. chat_service_type: ChatServiceType = ChatServiceType.OPENAI_COMPATIBLE,
  504. overwrite_context: Optional[Dict] = None
  505. ) -> Dict:
  506. """
  507. 参数:
  508. user_message: 当前用户消息,如果是主动交互则为None
  509. 返回:
  510. 消息列表
  511. """
  512. dialogue_history = self.history_dialogue_service.get_dialogue_history(self.staff_id, self.user_id)
  513. logger.debug("staff[{}], user[{}], dialogue_history: {}".format(
  514. self.staff_id, self.user_id, dialogue_history
  515. ))
  516. messages = []
  517. config = {
  518. 'user_id': self.user_id
  519. }
  520. prompt_context = self.get_prompt_context(user_message)
  521. if overwrite_context:
  522. prompt_context.update(overwrite_context)
  523. # FIXME(zhoutian): time in string type
  524. current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
  525. if overwrite_context and 'current_time' in overwrite_context:
  526. current_time = overwrite_context.get('current_time')
  527. need_multimodal = self.need_multimodal_model(dialogue_history)
  528. config['use_multimodal_model'] = need_multimodal
  529. if chat_service_type == ChatServiceType.OPENAI_COMPATIBLE:
  530. system_message = self._create_system_message(prompt_context)
  531. messages.append(system_message)
  532. messages.extend(self.compose_chat_messages_openai_compatible(dialogue_history, current_time, need_multimodal))
  533. elif chat_service_type == ChatServiceType.COZE_CHAT:
  534. dialogue_history = dialogue_history[-95:] # Coze最多支持100条,还需要附加系统消息
  535. messages = self.compose_chat_messages_coze(dialogue_history, current_time, self.staff_id, self.user_id)
  536. custom_variables = {}
  537. for k, v in prompt_context.items():
  538. custom_variables[k] = str(v)
  539. custom_variables.pop('user_profile', None)
  540. config['custom_variables'] = custom_variables
  541. config['bot_id'] = self._select_coze_bot(self.current_state, dialogue_history, need_multimodal)
  542. #FIXME(zhoutian): 临时报警
  543. if user_message and not messages:
  544. logger.error(f"staff[{self.staff_id}], user[{self.user_id}]: inconsistency in messages")
  545. config['messages'] = messages
  546. return config
  547. @staticmethod
  548. def format_timestamp(timestamp_ms):
  549. return datetime.fromtimestamp(timestamp_ms / 1000).strftime("%Y-%m-%d %H:%M:%S")
  550. @staticmethod
  551. def format_dialogue_content(dialogue_entry):
  552. fmt_time = DialogueManager.format_timestamp(dialogue_entry['timestamp'])
  553. content = '[{}] {}'.format(fmt_time, dialogue_entry['content'])
  554. return content
  555. @staticmethod
  556. def build_chat_message(role, content, chat_service_type: ChatServiceType):
  557. if chat_service_type == ChatServiceType.COZE_CHAT:
  558. if role == 'user':
  559. return cozepy.Message.build_user_question_text(content)
  560. elif role == 'assistant':
  561. return cozepy.Message.build_assistant_answer(content)
  562. else:
  563. return {'role': role, 'content': content}
  564. if __name__ == '__main__':
  565. state_cache = DialogueStateCache()
  566. state_cache.set_state('1688854492669990', '7881302581935903', DialogueState.CHITCHAT, DialogueState.GREETING)