本地记忆+意图识别 (#250)

* 增加本地记忆功能，使用llm总结记忆 * update：增加统一非流式输出输出 * 增加意图识别内容，使用llm进行识别 * 初始化记忆模块 * 完善意图识别处理后的流程 * 通过使用function call实现意图识别 * update:优化意图识别的配置 * update:function call最优设置成doubao-pro-32k-functioncall-241028 --------- Co-authored-by: 玄凤科技 <eric230308@gmail.com> Co-authored-by: hrz <1710360675@qq.com>
2026-07-28 19:33:51 +08:00 · 2025-03-09 21:33:45 +08:00
parent 9c2b2a2dcc
commit 63f34e5a82
19 changed files with 858 additions and 108 deletions
@@ -0,0 +1,33 @@
+from abc import ABC, abstractmethod
+from typing import List, Dict
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+class IntentProviderBase(ABC):
+    def __init__(self, config):
+        self.config = config
+        self.intent_options = config.get("intent_options", {
+            "continue_chat": "继续聊天",
+            "end_chat": "结束聊天",
+            "play_music": "播放音乐"
+        })
+
+    def set_llm(self, llm):
+        self.llm = llm
+        logger.bind(tag=TAG).debug("Set LLM for intent provider")
+        
+    @abstractmethod
+    async def detect_intent(self, dialogue_history: List[Dict]) -> str:
+        """
+        检测用户最后一句话的意图
+        Args:
+            dialogue_history: 对话历史记录列表，每条记录包含role和content
+        Returns:
+            返回识别出的意图，格式为:
+            - "继续聊天"
+            - "结束聊天" 
+            - "播放音乐 歌名" 或 "随机播放音乐"
+        """
+        pass
@@ -0,0 +1,61 @@
+from typing import List, Dict
+from ..base import IntentProviderBase
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+
+
+class IntentProvider(IntentProviderBase):
+    def __init__(self, config):
+        super().__init__(config)
+        self.llm = None
+        self.promot = self.get_intent_system_prompt()
+
+    def get_intent_system_prompt(self) -> str:
+        """
+        根据配置的意图选项动态生成系统提示词
+        Returns:
+            格式化后的系统提示词
+        """
+        intent_list = []
+        for key, value in self.intent_options.items():
+            if key == "play_music":
+                intent_list.append(f"{value} [歌名]")
+            else:
+                intent_list.append(value)
+                
+        prompt = (
+            "你是一个意图识别助手。你需要根据和用户的对话记录，重点分析用户的最后一句话，判断用户意图属于以下哪一类：\n"
+            f"{', '.join(intent_list)}\n"
+            "如果是唱歌、听歌、播放音乐，请指定歌名，格式为'播放音乐 [识别出的歌名]'。\n"
+            "如果听不出具体歌名，可以返回'随机播放音乐'。\n"
+            "只需要返回意图结果的json，不要解释。"
+            "返回格式如下：\n"
+            "{intent: '用户意图'}"
+        )
+        return prompt
+    
+    async def detect_intent(self, dialogue_history: List[Dict]) -> str:
+        if not self.llm:
+            raise ValueError("LLM provider not set")
+
+        # 构建用户最后一句话的提示
+        msgStr = ""
+        for msg in dialogue_history:
+            if msg.role == "user":
+                msgStr += f"User: {msg.content}\n"
+            elif msg.role== "assistant":
+                msgStr += f"Assistant: {msg.content}\n"
+
+        user_prompt = f"请分析用户的意图：\n{msgStr}"
+        
+        # 使用LLM进行意图识别
+        intent = self.llm.response_no_stream(
+            system_prompt=self.promot,
+            user_prompt=user_prompt
+        )
+  
+        logger.bind(tag=TAG).info(f"Detected intent: {intent}")
+        return intent.strip()
@@ -0,0 +1,18 @@
+from ..base import IntentProviderBase
+from typing import List, Dict
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+class IntentProvider(IntentProviderBase):
+    async def detect_intent(self, dialogue_history: List[Dict]) -> str:
+        """
+        默认的意图识别实现，始终返回继续聊天
+        Args:
+            dialogue_history: 对话历史记录列表
+        Returns:
+            固定返回"继续聊天"
+        """
+        logger.bind(tag=TAG).debug("Using NoIntentProvider, always returning continue chat")
+        return self.intent_options["continue_chat"]
@@ -1,8 +1,38 @@
 from abc import ABC, abstractmethod
+from config.logger import setup_logging

+TAG = __name__
+logger = setup_logging()

 class LLMProviderBase(ABC):
    @abstractmethod
    def response(self, session_id, dialogue):
        """LLM response generator"""
        pass
+
+    def response_no_stream(self, system_prompt, user_prompt):
+        try:
+            # 构造对话格式
+            dialogue = [
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": user_prompt}
+            ]
+            result = ""
+            for part in self.response("", dialogue):
+                result += part
+            return result
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"Error in Ollama response generation: {e}")
+            return "【LLM服务响应异常】"
+    
+    def response_with_functions(self, session_id, dialogue, functions=None):
+        """
+        Default implementation for function calling (streaming)
+        This should be overridden by providers that support function calls
+        
+        Returns: generator that yields either text tokens or a special function call token
+        """
+        # For providers that don't support functions, just return regular response
+        for token in self.response(session_id, dialogue):
+            yield {"type": "content", "content": token}
@@ -1,5 +1,6 @@
 from config.logger import setup_logging
-import requests, json
+from openai import OpenAI
+import json
 from core.providers.llm.base import LLMProviderBase

 TAG = __name__
@@ -8,39 +9,73 @@ logger = setup_logging()

 class LLMProvider(LLMProviderBase):
    def __init__(self, config):
-
        self.model_name = config.get("model_name")
        self.base_url = config.get("base_url", "http://localhost:11434")
+        # Initialize OpenAI client with Ollama base URL
+        #如果没有v1，增加v1
+        if not self.base_url.endswith("/v1"):
+            self.base_url = f"{self.base_url}/v1"
+            
+        self.client = OpenAI(
+            base_url=self.base_url,
+            api_key="ollama"  # Ollama doesn't need an API key but OpenAI client requires one
+        )

-    def response(self, session_id, dialogue):        
+    def response(self, session_id, dialogue):
        try:
-            # Convert dialogue format to Ollama format
-            prompt = ""
-            for msg in dialogue:
-                if msg["role"] == "system":
-                    prompt += f"System: {msg['content']}\n"
-                elif msg["role"] == "user":
-                    prompt += f"User: {msg['content']}\n"
-                elif msg["role"] == "assistant":
-                    prompt += f"Assistant: {msg['content']}\n"
-
-            # Make request to Ollama API
-            response = requests.post(
-                f"{self.base_url}/api/generate",
-                json={
-                    "model": self.model_name,
-                    "prompt": prompt,
-                    "stream": True
-                },
+            responses = self.client.chat.completions.create(
+                model=self.model_name,
+                messages=dialogue,
                stream=True
            )
-
-            for line in response.iter_lines():
-                if line:
-                    json_response = json.loads(line)
-                    if "response" in json_response:
-                        yield json_response["response"]
+            
+            for chunk in responses:
+                try:
+                    delta = chunk.choices[0].delta if getattr(chunk, 'choices', None) else None
+                    content = delta.content if hasattr(delta, 'content') else ''
+                    if content:
+                        yield content
+                except Exception as e:
+                    logger.bind(tag=TAG).error(f"Error processing chunk: {e}")

        except Exception as e:
            logger.bind(tag=TAG).error(f"Error in Ollama response generation: {e}")
            yield "【Ollama服务响应异常】"
+
+    def response_with_functions(self, session_id, dialogue, functions=None):
+        try:
+            stream = self.client.chat.completions.create(
+                model=self.model_name,
+                messages=dialogue,
+                stream=True,
+                tools=functions,
+            )
+            
+            current_function_call = None
+            current_content = ""
+            
+            for chunk in stream:
+                delta = chunk.choices[0].delta
+                
+                if delta.content:
+                    current_content += delta.content
+                    yield {"type": "content", "content": delta.content}
+                
+                if delta.tool_calls:
+                    tool_call = delta.tool_calls[0]
+                    # Handle the function call data using proper attribute access
+                    if not current_function_call:
+                        current_function_call = {
+                            "function": {
+                                "name": tool_call.function.name,
+                                "arguments": tool_call.function.arguments
+                            }
+                        }
+            
+            if current_function_call:
+                logger.bind(tag=TAG).debug(f"ollama Function call detected: {current_function_call}")
+                yield {"type": "function_call", "function_call": current_function_call}
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"Error in Ollama function call: {e}")
+            yield {"type": "content", "content": f"【Ollama服务响应异常: {str(e)}】"}
@@ -43,3 +43,41 @@ class LLMProvider(LLMProviderBase):

        except Exception as e:
            logger.bind(tag=TAG).error(f"Error in response generation: {e}")
+
+    def response_with_functions(self, session_id, dialogue, functions=None):
+        try:
+            stream = self.client.chat.completions.create(
+                model=self.model_name,
+                messages=dialogue,
+                stream=True,
+                tools=functions,
+            )
+            
+            current_function_call = None
+            current_content = ""
+            
+            for chunk in stream:
+                delta = chunk.choices[0].delta
+                
+                if delta.content:
+                    current_content += delta.content
+                    yield {"type": "content", "content": delta.content}
+                
+                if delta.tool_calls:
+                    tool_call = delta.tool_calls[0]
+                    # Handle the function call data using proper attribute access
+                    if not current_function_call:
+                        current_function_call = {
+                            "function": {
+                                "name": tool_call.function.name,
+                                "arguments": tool_call.function.arguments
+                            }
+                        }
+            
+            if current_function_call:
+                logger.bind(tag=TAG).debug(f"openai Function call detected: {current_function_call}")
+                yield {"type": "function_call", "function_call": current_function_call}
+                
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"Error in function call streaming: {e}")
+            yield {"type": "content", "content": f"【OpenAI服务响应异常: {e}】"}
@@ -8,6 +8,7 @@ class MemoryProviderBase(ABC):
    def __init__(self, config):
        self.config = config
        self.role_id = None
+        self.llm = None

    @abstractmethod
    async def save_memory(self, msgs):
@@ -19,5 +20,6 @@ class MemoryProviderBase(ABC):
        """Query memories for specific role based on similarity"""
        return "please implement query method"

-    def set_role_id(self, role_id: str):
-        self.role_id = role_id
+    def init_memory(self, role_id, llm):
+        self.role_id = role_id    
+        self.llm = llm
@@ -0,0 +1,156 @@
+from ..base import MemoryProviderBase, logger
+import time
+import json
+import os
+import yaml
+from core.utils.util import get_project_dir
+
+short_term_memory_prompt = """
+# 时空记忆编织者
+
+## 核心使命
+构建可生长的动态记忆网络，在有限空间内保留关键信息的同时，智能维护信息演变轨迹
+根据对话记录，总结user的重要信息，以便在未来的对话中提供更个性化的服务
+
+## 记忆法则
+### 1. 三维度记忆评估（每次更新必执行）
+| 维度       | 评估标准                  | 权重分 |
+|------------|---------------------------|--------|
+| 时效性     | 信息新鲜度（按对话轮次） | 40%    |
+| 情感强度   | 含💖标记/重复提及次数     | 35%    |
+| 关联密度   | 与其他信息的连接数量      | 25%    |
+
+### 2. 动态更新机制
+**名字变更处理示例：**
+原始记忆："曾用名": ["张三"], "现用名": "张三丰"
+触发条件：当检测到「我叫X」「称呼我Y」等命名信号时
+操作流程：
+1. 将旧名移入"曾用名"列表
+2. 记录命名时间轴："2024-02-15 14:32:启用张三丰"
+3. 在记忆立方追加：「从张三到张三丰的身份蜕变」
+
+### 3. 空间优化策略
+- **信息压缩术**：用符号体系提升密度
+  - ✅"张三丰[北/软工/🐱]"
+  - ❌"北京软件工程师，养猫"
+- **淘汰预警**：当总字数≥900时触发
+  1. 删除权重分<60且3轮未提及的信息
+  2. 合并相似条目（保留时间戳最近的）
+
+## 记忆结构
+输出格式必须为可解析的json字符串，不需要解释、注释和说明，保存记忆时仅从对话提取信息，不要混入示例内容
+```json
+{
+  "时空档案": {
+    "身份图谱": {
+      "现用名": "",
+      "特征标记": [] 
+    },
+    "记忆立方": [
+      {
+        "事件": "入职新公司",
+        "时间戳": "2024-03-20",
+        "情感值": 0.9,
+        "关联项": ["下午茶"],
+        "保鲜期": 30 
+      }
+    ]
+  },
+  "关系网络": {
+    "高频话题": {"职场": 12},
+    "暗线联系": [""]
+  },
+  "待响应": {
+    "紧急事项": ["需立即处理的任务"], 
+    "潜在关怀": ["可主动提供的帮助"]
+  },
+  "高光语录": [
+    "最打动人心的瞬间，强烈的情感表达，user的原话"
+  ]
+}
+```
+"""
+
+def extract_json_data(json_code):
+    start = json_code.find("```json")
+    # 从start开始找到下一个```结束
+    end = json_code.find("```", start+1)
+    #print("start:", start, "end:", end)
+    if start == -1 or end == -1:
+        try:
+            jsonData = json.loads(json_code)
+            return json_code
+        except Exception as e:
+            print("Error:", e)
+        return ""
+    jsonData = json_code[start+7:end]
+    return jsonData
+
+TAG = __name__
+
+class MemoryProvider(MemoryProviderBase):
+    def __init__(self, config):
+        super().__init__(config)
+        self.short_momery = ""
+        self.memory_path = get_project_dir() + 'data/.memory.yaml'
+        self.load_memory()
+
+    def init_memory(self, role_id, llm):
+        super().init_memory(role_id, llm)
+        self.load_memory()
+    
+    def load_memory(self):
+        all_memory = {}
+        if os.path.exists(self.memory_path):
+            with open(self.memory_path, 'r', encoding='utf-8') as f:
+                all_memory = yaml.safe_load(f) or {}
+        if self.role_id in all_memory:
+            self.short_momery = all_memory[self.role_id]
+    
+    def save_memory_to_file(self):
+        all_memory = {}
+        if os.path.exists(self.memory_path):
+              with open(self.memory_path, 'r', encoding='utf-8') as f:
+                  all_memory = yaml.safe_load(f) or {}
+        all_memory[self.role_id] = self.short_momery
+        with open(self.memory_path, 'w', encoding='utf-8') as f:
+            yaml.dump(all_memory, f, allow_unicode=True)
+        
+    async def save_memory(self, msgs):
+        if self.llm is None:
+            logger.bind(tag=TAG).error("LLM is not set for memory provider")
+            return None
+        
+        if len(msgs) < 2:
+            return None
+        
+        msgStr = ""
+        for msg in msgs:
+            if msg.role == "user":
+                msgStr += f"User: {msg.content}\n"
+            elif msg.role== "assistant":
+                msgStr += f"Assistant: {msg.content}\n"
+        if len(self.short_momery) > 0:
+            msgStr+="历史记忆：\n"
+            msgStr+=self.short_momery
+        
+        #当前时间
+        time_str = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
+        msgStr += f"当前时间：{time_str}"
+
+        result = self.llm.response_no_stream(short_term_memory_prompt, msgStr)
+ 
+        json_str = extract_json_data(result)
+        try:
+            json_data = json.loads(json_str) # 检查json格式是否正确
+            self.short_momery = json_str
+        except Exception as e:
+            print("Error:", e)
+        
+        self.save_memory_to_file()
+        logger.bind(tag=TAG).info(f"Save memory successful - Role: {self.role_id}")
+
+        return self.short_momery
+    
+    async def query_memory(self, query: str)-> str:
+        return self.short_momery
@@ -0,0 +1,18 @@
+'''
+不使用记忆，可以选择此模块
+'''
+from ..base import MemoryProviderBase, logger
+
+TAG = __name__
+
+class MemoryProvider(MemoryProviderBase):
+    def __init__(self, config):
+        super().__init__(config)
+      
+    async def save_memory(self, msgs):
+        logger.bind(tag=TAG).debug("nomem mode: No memory saving is performed.")
+        return None
+
+    async def query_memory(self, query: str)-> str:
+        logger.bind(tag=TAG).debug("nomem mode: No memory query is performed.")
+        return ""