Add history_dialogue_service and use in dialogue_manager

dialogue_manager.py

@@ -7,8 +7,9 @@ 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
@@ -54,6 +55,8 @@ class TimeContext(Enum):
 
 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
@@ -68,6 +71,9 @@ class DialogueManager:
         self.vector_memory = DummyVectorMemoryManager(user_id)
         self.message_aggregation_sec = 5
         self.unprocessed_messages = []
+        self.history_dialogue_service = HistoryDialogueService(
+            config['storage']['history_dialogue']['api_base_url']
+        )
 
     def get_current_time_context(self) -> TimeContext:
         """获取当前时间上下文"""
@@ -218,7 +224,7 @@ class DialogueManager:
             event = {
                 "timestamp": int(time.time() * 1000),
                 "reason": reason,
-                "dialogue_context": self.dialogue_history[-5:] if len(self.dialogue_history) >= 5 else self.dialogue_history
+                "dialogue_context": self.history_dialogue_service.get_dialogue_history(self.staff_id, self.user_id, 5)
             }
 
             # 更新用户资料中的人工介入历史
@@ -226,7 +232,7 @@ class DialogueManager:
                 self.user_profile["human_intervention_history"] = []
 
             self.user_profile["human_intervention_history"].append(event)
-            self.user_manager.save_user_profile(self.user_profile)
+            self.user_manager.save_user_profile(self.user_id, self.user_profile)
 
             # 发送告警
             self._send_human_intervention_alert(reason)
@@ -242,7 +248,7 @@ class DialogueManager:
         """
 
         # 添加最近的对话记录
-        recent_dialogues = self.dialogue_history[-5:] if len(self.dialogue_history) >= 5 else self.dialogue_history
+        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']}"
 
@@ -381,9 +387,10 @@ class DialogueManager:
         返回:
             消息列表
         """
-        dialogue_history = self.dialogue_history[-10:] \
-            if len(self.dialogue_history) > 10 \
-            else self.dialogue_history
+        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 = {}
 
@@ -399,6 +406,11 @@ class DialogueManager:
                 })
         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"]))

history_dialogue_service.py

@@ -0,0 +1,50 @@
+#! /usr/bin/env python
+# -*- coding: utf-8 -*-
+# vim:fenc=utf-8
+
+import requests
+import logging
+import json
+
+import configs
+
+
+class HistoryDialogueService:
+    def __init__(self, base_url: str):
+        self.base_url = base_url
+
+    def get_dialogue_history(self, staff_id: str, user_id: str, max_count: int = 10):
+        url = f"{self.base_url}?sender={staff_id}&receiver={user_id}"
+        response = requests.post(url, headers={
+            'Content-Type': 'application/json'
+        })
+        if response.status_code != 200:
+            raise Exception("Request error [{}]: {}".format(response.status_code, response.text))
+        data = response.json()
+        if not data.get('success', False):
+            raise Exception("Error in response: {}".format(data.get('message', 'no message returned')))
+        data = data.get('data', [])
+        ret = []
+        for record in data:
+            sender = record.get('sender')
+            if sender == user_id:
+                role = 'user'
+            elif sender == staff_id:
+                role = 'assistant'
+            else:
+                logging.warning("Unknown sender in dialogue history: {}".format(sender))
+                continue
+            ret.append({
+                'role': role,
+                'content': record.get('content', ''),
+                'timestamp': record.get('sendtime', 0)
+            })
+        ret = sorted(ret, key=lambda x: x['timestamp'])
+        return ret[-max_count:] if len(ret) > max_count else ret
+
+
+if __name__ == '__main__':
+    api_url = configs.get()['storage']['history_dialogue']['api_base_url']
+    service = HistoryDialogueService(api_url)
+    resp = service.get_dialogue_history(staff_id='1688854492669990', user_id='7881301263964433')
+    print(resp)