update:强化小智提示词

main/xiaozhi-server/agent-base-prompt.txt

@@ -0,0 +1,51 @@
+<identity>
+{base_prompt}
+</identity>
+
+<emotion>
+你不是一个冰冷的机器，你可以从用户的上下文中感受到用户的情绪，作为懂人心的生活伙伴，请用有温度的回应照亮每个对话！
+- 1.合理穿插笑声（哈哈、嘿嘿、噗，但每句≤1次）
+- 2.惊讶时用夸张语气（“不会吧？！”、“这么神奇？！”）
+- 3.安慰时说暖心话（“别急嘛”、“有我呢”）
+</emotion>
+
+<communication_style>
+- 1.使用口语化表达（如“呀”、“呢”、“啦”等语气词）。
+- 2.避免书面语和学术腔调，禁用“根据资料显示”等机械表达
+- 3.模仿人类对话的轻微不完美（适当使用“嗯”、“啊”等思考词）
+- 3.由于用户语音是通过ASR识别，识别结果可能存在错别字，请结合上下文推断真实含义。
+- 4.绝对禁止使用 markdown、列表、标题等格式
+</communication_style>
+
+<speaker_recognition>
+- 1.当用户消息包含 [说话人: 姓名] 前缀时，表示系统已识别出说话人身份。
+- 2.请根据说话人的身份特征（如果之前有相关信息）来调整回应风格和内容。
+- 3.你可以称呼说话人的名字，并参考他们的特点进行个性化回应。
+</speaker_recognition>
+
+<tool_calling>
+你可以调用工具来响应用户的要求。遵循以下关于工具调用的规则：
+- 1.始终严格遵循指定的工具调用模式，并确保提供所有必要的参数。
+- 2.对话可能会引用不再可用的工具。切勿调用未明确提供的工具。
+- 3.在与用户交谈时，切勿提及工具名称。相反，只需用自然语言说出工具正在做什么。
+- 4.你尽可能需要通过工具调用获得更多信息，而不是问用户。
+- 5.你应该结合用户上下文需求，洞察用户的真实需求才去调用相关的指令，而不是为了调工具而调工具。
+- 6.【重要】如果是查询"现在的时间"、"今天的几号"、"今天的日期"、"今天农历是多少"、"今天农历日期"、"今天{local_address}的天气"，这些信息已经包含在`<context>`中，不需要调用工具，请直接根据context回复。
+- 7.如果是查询"其他日期的农历"（明天、昨天、具体日期）或"详细农历信息"（宜忌、八字、节气等），需要调用相应工具获取。
+- 8.除了基本时间、今日农历、{local_address}天气查询外，用户的其他要求都视为独立任务，即使内容相似也需重新调用工具，不要偷懒，不要使用历史消息糊弄用户。
+- 9.如果你不确定与用户请求相关的动作，不要猜测或编造答案。
+- 10.如果工具中包含camera、take_photo等相关工具，说明用户已经给你安装了摄像头，调用这些工具会让你具备拍照、描述所见物品等能力。如果没有，切勿调用。
+</tool_calling>
+
+<context>
+- 1.现在的时间：{current_time}
+- 2.今天的日期：{today_date}、{today_weekday}
+- 3.今天的农历日期：{lunar_date}
+- 4.当前用户所处城市
+{local_address}
+- 5.用户所处城市未来7天天气
+{weather_info}
+</context>
+
+<memory>
+</memory>

main/xiaozhi-server/config/config_loader.py

@@ -1,14 +1,9 @@
 import os
-import argparse
 import yaml
 from collections.abc import Mapping
 from config.manage_api_client import init_service, get_server_config, get_agent_models
 
 
-# 添加全局配置缓存
-_config_cache = None
-
-
 def get_project_dir():
     """获取项目根目录"""
     return os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/"
@@ -22,9 +17,12 @@ def read_config(config_path):
 
 def load_config():
     """加载配置文件"""
-    global _config_cache
-    if _config_cache is not None:
-        return _config_cache
+    from core.utils.cache.manager import cache_manager, CacheType
+
+    # 检查缓存
+    cached_config = cache_manager.get(CacheType.CONFIG, "main_config")
+    if cached_config is not None:
+        return cached_config
 
     default_config_path = get_project_dir() + "config.yaml"
     custom_config_path = get_project_dir() + "data/.config.yaml"
@@ -40,7 +38,9 @@ def load_config():
         config = merge_configs(default_config, custom_config)
     # 初始化目录
     ensure_directories(config)
-    _config_cache = config
+
+    # 缓存配置
+    cache_manager.set(CacheType.CONFIG, "main_config", config)
     return config
 
 

main/xiaozhi-server/core/connection.py

@@ -36,6 +36,7 @@ from config.config_loader import get_private_config_from_api
 from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType
 from config.logger import setup_logging, build_module_string, update_module_string
 from config.manage_api_client import DeviceNotFoundException, DeviceBindException
+from core.utils.prompt_manager import PromptManager
 
 
 TAG = __name__
@@ -73,7 +74,6 @@ class ConnectionHandler:
         self.headers = None
         self.device_id = None
         self.client_ip = None
-        self.client_ip_info = {}
         self.prompt = None
         self.welcome_msg = None
         self.max_output_size = 0
@@ -149,6 +149,9 @@ class ConnectionHandler:
         # {"mcp":true} 表示启用MCP功能
         self.features = None
 
+        # 初始化提示词管理器
+        self.prompt_manager = PromptManager(config, self.logger)
+
     async def handle_connection(self, ws):
         try:
             # 获取并验证headers
@@ -325,12 +328,15 @@ class ConnectionHandler:
                 self.config.get("selected_module", {})
             )
             update_module_string(self.selected_module_str)
-            """初始化组件"""
+
+            """快速初始化系统提示词"""
             if self.config.get("prompt") is not None:
-                self.prompt = self.config["prompt"]
-                self.change_system_prompt(self.prompt)
+                user_prompt = self.config["prompt"]
+                # 使用快速提示词进行初始化
+                prompt = self.prompt_manager.get_quick_prompt(user_prompt)
+                self.change_system_prompt(prompt)
                 self.logger.bind(tag=TAG).info(
-                    f"初始化组件: prompt成功 {self.prompt[:50]}..."
+                    f"快速初始化组件: prompt成功 {prompt[:50]}..."
                 )
 
             """初始化本地组件"""
@@ -355,9 +361,22 @@ class ConnectionHandler:
             self._initialize_intent()
             """初始化上报线程"""
             self._init_report_threads()
+            """更新系统提示词"""
+            self._init_prompt_enhancement()
+
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"实例化组件失败: {e}")
 
+    def _init_prompt_enhancement(self):
+        # 更新上下文信息
+        self.prompt_manager.update_context_info(self, self.client_ip)
+        enhanced_prompt = self.prompt_manager.build_enhanced_prompt(
+            self.config["prompt"], self.device_id, self.client_ip
+        )
+        if enhanced_prompt:
+            self.change_system_prompt(enhanced_prompt)
+            self.logger.bind(tag=TAG).info("系统提示词已增强更新")
+
     def _init_report_threads(self):
         """初始化ASR和TTS上报线程"""
         if not self.read_config_from_api or self.need_bind:
@@ -758,8 +777,11 @@ class ConnectionHandler:
                 )
             )
         self.llm_finish_task = True
+        # 使用lambda延迟计算，只有在DEBUG级别时才执行get_llm_dialogue()
         self.logger.bind(tag=TAG).debug(
-            json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False)
+            lambda: json.dumps(
+                self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False
+            )
         )
 
         return True

main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py

@@ -16,10 +16,11 @@ class IntentProvider(IntentProviderBase):
         super().__init__(config)
         self.llm = None
         self.promot = ""
-        # 添加缓存管理
-        self.intent_cache = {}  # 缓存意图识别结果
-        self.cache_expiry = 600  # 缓存有效期10分钟
-        self.cache_max_size = 100  # 最多缓存100个意图
+        # 导入全局缓存管理器
+        from core.utils.cache.manager import cache_manager, CacheType
+
+        self.cache_manager = cache_manager
+        self.CacheType = CacheType
         self.history_count = 4  # 默认使用最近4条对话记录
 
     def get_intent_system_prompt(self, functions_list: str) -> str:
@@ -102,27 +103,6 @@ class IntentProvider(IntentProviderBase):
         )
         return prompt
 
-    def clean_cache(self):
-        """清理过期缓存"""
-        now = time.time()
-        # 找出过期键
-        expired_keys = [
-            k
-            for k, v in self.intent_cache.items()
-            if now - v["timestamp"] > self.cache_expiry
-        ]
-        for key in expired_keys:
-            del self.intent_cache[key]
-
-        # 如果缓存太大，移除最旧的条目
-        if len(self.intent_cache) > self.cache_max_size:
-            # 按时间戳排序并保留最新的条目
-            sorted_items = sorted(
-                self.intent_cache.items(), key=lambda x: x[1]["timestamp"]
-            )
-            for key, _ in sorted_items[: len(sorted_items) - self.cache_max_size]:
-                del self.intent_cache[key]
-
     def replyResult(self, text: str, original_text: str):
         llm_result = self.llm.response_no_stream(
             system_prompt=text,
@@ -145,21 +125,16 @@ class IntentProvider(IntentProviderBase):
         logger.bind(tag=TAG).debug(f"使用意图识别模型: {model_info}")
 
         # 计算缓存键
-        cache_key = hashlib.md5(text.encode()).hexdigest()
+        cache_key = hashlib.md5((conn.device_id + text).encode()).hexdigest()
 
         # 检查缓存
-        if cache_key in self.intent_cache:
-            cache_entry = self.intent_cache[cache_key]
-            # 检查缓存是否过期
-            if time.time() - cache_entry["timestamp"] <= self.cache_expiry:
-                cache_time = time.time() - total_start_time
-                logger.bind(tag=TAG).debug(
-                    f"使用缓存的意图: {cache_key} -> {cache_entry['intent']}, 耗时: {cache_time:.4f}秒"
-                )
-                return cache_entry["intent"]
-
-        # 清理缓存
-        self.clean_cache()
+        cached_intent = self.cache_manager.get(self.CacheType.INTENT, cache_key)
+        if cached_intent is not None:
+            cache_time = time.time() - total_start_time
+            logger.bind(tag=TAG).debug(
+                f"使用缓存的意图: {cache_key} -> {cached_intent}, 耗时: {cache_time:.4f}秒"
+            )
+            return cached_intent
 
         if self.promot == "":
             functions = conn.func_handler.get_functions()
@@ -259,10 +234,7 @@ class IntentProvider(IntentProviderBase):
                     conn.dialogue.dialogue = clean_history
 
                 # 添加到缓存
-                self.intent_cache[cache_key] = {
-                    "intent": intent,
-                    "timestamp": time.time(),
-                }
+                self.cache_manager.set(self.CacheType.INTENT, cache_key, intent)
 
                 # 后处理时间
                 postprocess_time = time.time() - postprocess_start_time
@@ -272,10 +244,7 @@ class IntentProvider(IntentProviderBase):
                 return intent
             else:
                 # 添加到缓存
-                self.intent_cache[cache_key] = {
-                    "intent": intent,
-                    "timestamp": time.time(),
-                }
+                self.cache_manager.set(self.CacheType.INTENT, cache_key, intent)
 
                 # 后处理时间
                 postprocess_time = time.time() - postprocess_start_time

main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py

@@ -51,13 +51,13 @@ class ServerPluginExecutor(ToolExecutor):
         tools = {}
 
         # 获取必要的函数
-        necessary_functions = ["handle_exit_intent", "get_time", "get_lunar"]
+        necessary_functions = ["handle_exit_intent", "get_lunar"]
 
         # 获取配置中的函数
         config_functions = self.config["Intent"][
             self.config["selected_module"]["Intent"]
         ].get("functions", [])
-        
+
         # 转换为列表
         if not isinstance(config_functions, list):
             try:

main/xiaozhi-server/core/utils/cache/config.py

@@ -0,0 +1,58 @@
+"""
+缓存配置管理
+"""
+
+from enum import Enum
+from typing import Dict, Any, Optional
+from dataclasses import dataclass
+from .strategies import CacheStrategy
+
+
+class CacheType(Enum):
+    """缓存类型枚举"""
+
+    LOCATION = "location"
+    WEATHER = "weather"
+    LUNAR = "lunar"
+    INTENT = "intent"
+    IP_INFO = "ip_info"
+    CONFIG = "config"
+    DEVICE_PROMPT = "device_prompt"
+
+
+@dataclass
+class CacheConfig:
+    """缓存配置类"""
+
+    strategy: CacheStrategy = CacheStrategy.TTL
+    ttl: Optional[float] = 300  # 默认5分钟
+    max_size: Optional[int] = 1000  # 默认最大1000条
+    cleanup_interval: float = 60  # 清理间隔（秒）
+
+    @classmethod
+    def for_type(cls, cache_type: CacheType) -> "CacheConfig":
+        """根据缓存类型返回预设配置"""
+        configs = {
+            CacheType.LOCATION: cls(
+                strategy=CacheStrategy.TTL, ttl=None, max_size=1000  # 手动失效
+            ),
+            CacheType.IP_INFO: cls(
+                strategy=CacheStrategy.TTL, ttl=86400, max_size=1000  # 24小时
+            ),
+            CacheType.WEATHER: cls(
+                strategy=CacheStrategy.TTL, ttl=28800, max_size=1000  # 8小时
+            ),
+            CacheType.LUNAR: cls(
+                strategy=CacheStrategy.TTL, ttl=2592000, max_size=365  # 30天过期
+            ),
+            CacheType.INTENT: cls(
+                strategy=CacheStrategy.TTL_LRU, ttl=600, max_size=1000  # 10分钟
+            ),
+            CacheType.CONFIG: cls(
+                strategy=CacheStrategy.FIXED_SIZE, ttl=None, max_size=20  # 手动失效
+            ),
+            CacheType.DEVICE_PROMPT: cls(
+                strategy=CacheStrategy.TTL, ttl=None, max_size=1000  # 手动失效
+            ),
+        }
+        return configs.get(cache_type, cls())

main/xiaozhi-server/core/utils/cache/manager.py

@@ -0,0 +1,216 @@
+"""
+全局缓存管理器
+"""
+
+import time
+import threading
+from typing import Any, Optional, Dict
+from collections import OrderedDict
+from .strategies import CacheStrategy, CacheEntry
+from .config import CacheConfig, CacheType
+
+
+class GlobalCacheManager:
+    """全局缓存管理器"""
+
+    def __init__(self):
+        self._logger = None
+        self._caches: Dict[str, Dict[str, CacheEntry]] = {}
+        self._configs: Dict[str, CacheConfig] = {}
+        self._locks: Dict[str, threading.RLock] = {}
+        self._global_lock = threading.RLock()
+        self._last_cleanup = time.time()
+        self._stats = {"hits": 0, "misses": 0, "evictions": 0, "cleanups": 0}
+
+    @property
+    def logger(self):
+        """延迟初始化 logger 以避免循环导入"""
+        if self._logger is None:
+            from config.logger import setup_logging
+
+            self._logger = setup_logging()
+        return self._logger
+
+    def _get_cache_name(self, cache_type: CacheType, namespace: str = "") -> str:
+        """生成缓存名称"""
+        if namespace:
+            return f"{cache_type.value}:{namespace}"
+        return cache_type.value
+
+    def _get_or_create_cache(
+        self, cache_name: str, config: CacheConfig
+    ) -> Dict[str, CacheEntry]:
+        """获取或创建缓存空间"""
+        with self._global_lock:
+            if cache_name not in self._caches:
+                self._caches[cache_name] = (
+                    OrderedDict()
+                    if config.strategy in [CacheStrategy.LRU, CacheStrategy.TTL_LRU]
+                    else {}
+                )
+                self._configs[cache_name] = config
+                self._locks[cache_name] = threading.RLock()
+            return self._caches[cache_name]
+
+    def set(
+        self,
+        cache_type: CacheType,
+        key: str,
+        value: Any,
+        ttl: Optional[float] = None,
+        namespace: str = "",
+    ) -> None:
+        """设置缓存值"""
+        cache_name = self._get_cache_name(cache_type, namespace)
+        config = self._configs.get(cache_name) or CacheConfig.for_type(cache_type)
+        cache = self._get_or_create_cache(cache_name, config)
+
+        # 使用配置的TTL或传入的TTL
+        effective_ttl = ttl if ttl is not None else config.ttl
+
+        with self._locks[cache_name]:
+            # 创建缓存条目
+            entry = CacheEntry(value=value, timestamp=time.time(), ttl=effective_ttl)
+
+            # 处理不同策略
+            if config.strategy in [CacheStrategy.LRU, CacheStrategy.TTL_LRU]:
+                # LRU策略：如果已存在则移动到末尾
+                if key in cache:
+                    del cache[key]
+                cache[key] = entry
+
+                # 检查大小限制
+                if config.max_size and len(cache) > config.max_size:
+                    # 移除最旧的条目
+                    oldest_key = next(iter(cache))
+                    del cache[oldest_key]
+                    self._stats["evictions"] += 1
+
+            else:
+                cache[key] = entry
+
+                # 检查大小限制
+                if config.max_size and len(cache) > config.max_size:
+                    # 简单策略：随机移除一个条目
+                    victim_key = next(iter(cache))
+                    del cache[victim_key]
+                    self._stats["evictions"] += 1
+
+        # 定期清理过期条目
+        self._maybe_cleanup(cache_name)
+
+    def get(
+        self, cache_type: CacheType, key: str, namespace: str = ""
+    ) -> Optional[Any]:
+        """获取缓存值"""
+        cache_name = self._get_cache_name(cache_type, namespace)
+
+        if cache_name not in self._caches:
+            self._stats["misses"] += 1
+            return None
+
+        cache = self._caches[cache_name]
+        config = self._configs[cache_name]
+
+        with self._locks[cache_name]:
+            if key not in cache:
+                self._stats["misses"] += 1
+                return None
+
+            entry = cache[key]
+
+            # 检查过期
+            if entry.is_expired():
+                del cache[key]
+                self._stats["misses"] += 1
+                return None
+
+            # 更新访问信息
+            entry.touch()
+
+            # LRU策略：移动到末尾
+            if config.strategy in [CacheStrategy.LRU, CacheStrategy.TTL_LRU]:
+                del cache[key]
+                cache[key] = entry
+
+            self._stats["hits"] += 1
+            return entry.value
+
+    def delete(self, cache_type: CacheType, key: str, namespace: str = "") -> bool:
+        """删除缓存条目"""
+        cache_name = self._get_cache_name(cache_type, namespace)
+
+        if cache_name not in self._caches:
+            return False
+
+        cache = self._caches[cache_name]
+
+        with self._locks[cache_name]:
+            if key in cache:
+                del cache[key]
+                return True
+            return False
+
+    def clear(self, cache_type: CacheType, namespace: str = "") -> None:
+        """清空指定缓存"""
+        cache_name = self._get_cache_name(cache_type, namespace)
+
+        if cache_name not in self._caches:
+            return
+
+        with self._locks[cache_name]:
+            self._caches[cache_name].clear()
+
+    def invalidate_pattern(
+        self, cache_type: CacheType, pattern: str, namespace: str = ""
+    ) -> int:
+        """按模式失效缓存条目"""
+        cache_name = self._get_cache_name(cache_type, namespace)
+
+        if cache_name not in self._caches:
+            return 0
+
+        cache = self._caches[cache_name]
+        deleted_count = 0
+
+        with self._locks[cache_name]:
+            keys_to_delete = [key for key in cache.keys() if pattern in key]
+            for key in keys_to_delete:
+                del cache[key]
+                deleted_count += 1
+
+        return deleted_count
+
+    def _cleanup_expired(self, cache_name: str) -> int:
+        """清理过期条目"""
+        if cache_name not in self._caches:
+            return 0
+
+        cache = self._caches[cache_name]
+        deleted_count = 0
+
+        with self._locks[cache_name]:
+            expired_keys = [key for key, entry in cache.items() if entry.is_expired()]
+            for key in expired_keys:
+                del cache[key]
+                deleted_count += 1
+
+        return deleted_count
+
+    def _maybe_cleanup(self, cache_name: str):
+        """定期清理检查"""
+        config = self._configs.get(cache_name)
+        if not config:
+            return
+
+        now = time.time()
+        if now - self._last_cleanup > config.cleanup_interval:
+            self._last_cleanup = now
+            deleted = self._cleanup_expired(cache_name)
+            if deleted > 0:
+                self._stats["cleanups"] += 1
+                self.logger.debug(f"清理缓存 {cache_name}: 删除 {deleted} 个过期条目")
+
+
+# 创建全局缓存管理器实例
+cache_manager = GlobalCacheManager()

main/xiaozhi-server/core/utils/cache/strategies.py

@@ -0,0 +1,43 @@
+"""
+缓存策略和数据结构定义
+"""
+
+import time
+from enum import Enum
+from typing import Any, Optional
+from dataclasses import dataclass
+
+
+class CacheStrategy(Enum):
+    """缓存策略枚举"""
+
+    TTL = "ttl"  # 基于时间过期
+    LRU = "lru"  # 最近最少使用
+    FIXED_SIZE = "fixed_size"  # 固定大小
+    TTL_LRU = "ttl_lru"  # TTL + LRU混合策略
+
+
+@dataclass
+class CacheEntry:
+    """缓存条目数据结构"""
+
+    value: Any
+    timestamp: float
+    ttl: Optional[float] = None  # 生存时间（秒）
+    access_count: int = 0
+    last_access: float = None
+
+    def __post_init__(self):
+        if self.last_access is None:
+            self.last_access = self.timestamp
+
+    def is_expired(self) -> bool:
+        """检查是否过期"""
+        if self.ttl is None:
+            return False
+        return time.time() - self.timestamp > self.ttl
+
+    def touch(self):
+        """更新访问时间和计数"""
+        self.last_access = time.time()
+        self.access_count += 1

main/xiaozhi-server/core/utils/dialogue.py

@@ -1,4 +1,5 @@
 import uuid
+import re
 from typing import List, Dict
 from datetime import datetime
 
@@ -74,9 +75,12 @@ class Dialogue:
         )
 
         if system_message:
-            enhanced_system_prompt = (
-                f"{system_message.content}\n\n"
-                f"以下是用户的历史记忆：\n```\n{memory_str}\n```"
+            # 使用正则表达式匹配 <memory> 标签，不管中间有什么内容
+            enhanced_system_prompt = re.sub(
+                r"<memory>.*?</memory>",
+                f"<memory>\n{memory_str}\n</memory>",
+                system_message.content,
+                flags=re.DOTALL,
             )
             dialogue.append({"role": "system", "content": enhanced_system_prompt})
 

main/xiaozhi-server/core/utils/prompt_manager.py

@@ -0,0 +1,219 @@
+"""
+系统提示词管理器模块
+负责管理和更新系统提示词，包括快速初始化和异步增强功能
+"""
+
+import os
+import cnlunar
+from typing import Dict, Any
+from config.logger import setup_logging
+
+TAG = __name__
+
+WEEKDAY_MAP = {
+    "Monday": "星期一",
+    "Tuesday": "星期二",
+    "Wednesday": "星期三",
+    "Thursday": "星期四",
+    "Friday": "星期五",
+    "Saturday": "星期六",
+    "Sunday": "星期日",
+}
+
+
+class PromptManager:
+    """系统提示词管理器，负责管理和更新系统提示词"""
+
+    def __init__(self, config: Dict[str, Any], logger=None):
+        self.config = config
+        self.logger = logger or setup_logging()
+        self.base_prompt_template = None
+        self.last_update_time = 0
+
+        # 导入全局缓存管理器
+        from core.utils.cache.manager import cache_manager, CacheType
+
+        self.cache_manager = cache_manager
+        self.CacheType = CacheType
+
+        self._load_base_template()
+
+    def _load_base_template(self):
+        """加载基础提示词模板"""
+        try:
+            template_path = "agent-base-prompt.txt"
+            cache_key = f"prompt_template:{template_path}"
+
+            # 先从缓存获取
+            cached_template = self.cache_manager.get(self.CacheType.CONFIG, cache_key)
+            if cached_template is not None:
+                self.base_prompt_template = cached_template
+                self.logger.bind(tag=TAG).debug("从缓存加载基础提示词模板")
+                return
+
+            # 缓存未命中，从文件读取
+            if os.path.exists(template_path):
+                with open(template_path, "r", encoding="utf-8") as f:
+                    template_content = f.read()
+
+                # 存入缓存（CONFIG类型默认不自动过期，需要手动失效）
+                self.cache_manager.set(
+                    self.CacheType.CONFIG, cache_key, template_content
+                )
+                self.base_prompt_template = template_content
+                self.logger.bind(tag=TAG).debug("成功加载基础提示词模板并缓存")
+            else:
+                self.logger.bind(tag=TAG).warning("未找到agent-base-prompt.txt文件")
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"加载提示词模板失败: {e}")
+
+    def get_quick_prompt(self, user_prompt: str, device_id: str = None) -> str:
+        """快速获取系统提示词（使用用户配置）"""
+        device_cache_key = f"device_prompt:{device_id}"
+        cached_device_prompt = self.cache_manager.get(
+            self.CacheType.DEVICE_PROMPT, device_cache_key
+        )
+        if cached_device_prompt is not None:
+            self.logger.bind(tag=TAG).debug(f"使用设备 {device_id} 的缓存提示词")
+            return cached_device_prompt
+        else:
+            self.logger.bind(tag=TAG).debug(
+                f"设备 {device_id} 无缓存提示词，使用传入的提示词"
+            )
+
+        # 使用传入的提示词并缓存（如果有设备ID）
+        if device_id:
+            device_cache_key = f"device_prompt:{device_id}"
+            self.cache_manager.set(self.CacheType.CONFIG, device_cache_key, user_prompt)
+            self.logger.bind(tag=TAG).debug(f"设备 {device_id} 的提示词已缓存")
+
+        self.logger.bind(tag=TAG).info(f"使用快速提示词: {user_prompt[:50]}...")
+        return user_prompt
+
+    def _get_current_time_info(self) -> tuple:
+        """获取当前时间信息"""
+        from datetime import datetime
+
+        now = datetime.now()
+        current_time = now.strftime("%H:%M")
+        today_date = now.strftime("%Y-%m-%d")
+        today_weekday = WEEKDAY_MAP[now.strftime("%A")]
+        today_lunar = cnlunar.Lunar(now, godType="8char")
+        lunar_date = "%s年%s%s\n" % (
+            today_lunar.lunarYearCn,
+            today_lunar.lunarMonthCn[:-1],
+            today_lunar.lunarDayCn,
+        )
+
+        return current_time, today_date, today_weekday, lunar_date
+
+    def _get_location_info(self, client_ip: str) -> str:
+        """获取位置信息"""
+        try:
+            # 先从缓存获取
+            cached_location = self.cache_manager.get(self.CacheType.LOCATION, client_ip)
+            if cached_location is not None:
+                return cached_location
+
+            # 缓存未命中，调用API获取
+            from core.utils.util import get_ip_info
+
+            ip_info = get_ip_info(client_ip, self.logger)
+            city = ip_info.get("city", "未知位置")
+            location = f"{city}"
+
+            # 存入缓存
+            self.cache_manager.set(self.CacheType.LOCATION, client_ip, location)
+            return location
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"获取位置信息失败: {e}")
+            return "未知位置"
+
+    def _get_weather_info(self, conn, location: str) -> str:
+        """获取天气信息"""
+        try:
+            # 先从缓存获取
+            cached_weather = self.cache_manager.get(self.CacheType.WEATHER, location)
+            if cached_weather is not None:
+                return cached_weather
+
+            # 缓存未命中，调用get_weather函数获取
+            from plugins_func.functions.get_weather import get_weather
+            from plugins_func.register import ActionResponse
+
+            # 调用get_weather函数
+            result = get_weather(conn, location=location, lang="zh_CN")
+            if isinstance(result, ActionResponse):
+                weather_report = result.result
+                self.cache_manager.set(self.CacheType.WEATHER, location, weather_report)
+                return weather_report
+            return "天气信息获取失败"
+
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"获取天气信息失败: {e}")
+            return "天气信息获取失败"
+
+    def update_context_info(self, conn, client_ip: str):
+        """同步更新上下文信息"""
+        try:
+            # 获取位置信息（使用全局缓存）
+            local_address = self._get_location_info(client_ip)
+            # 获取天气信息（使用全局缓存）
+            self._get_weather_info(conn, local_address)
+            self.logger.bind(tag=TAG).info(f"上下文信息更新完成")
+
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"更新上下文信息失败: {e}")
+
+    def build_enhanced_prompt(
+        self, user_prompt: str, device_id: str, client_ip: str = None
+    ) -> str:
+        """构建增强的系统提示词"""
+        if not self.base_prompt_template:
+            return user_prompt
+
+        try:
+            # 获取最新的时间信息（不缓存）
+            current_time, today_date, today_weekday, lunar_date = (
+                self._get_current_time_info()
+            )
+
+            # 获取缓存的上下文信息
+            local_address = ""
+            weather_info = ""
+
+            if client_ip:
+                # 获取位置信息（从全局缓存）
+                local_address = (
+                    self.cache_manager.get(self.CacheType.LOCATION, client_ip) or ""
+                )
+
+                # 获取天气信息（从全局缓存）
+                if local_address:
+                    weather_info = (
+                        self.cache_manager.get(self.CacheType.WEATHER, local_address)
+                        or ""
+                    )
+
+            # 替换模板变量
+            enhanced_prompt = self.base_prompt_template.format(
+                base_prompt=user_prompt,
+                current_time=current_time,
+                today_date=today_date,
+                today_weekday=today_weekday,
+                lunar_date=lunar_date,
+                local_address=local_address,
+                weather_info=weather_info,
+            )
+            device_cache_key = f"device_prompt:{device_id}"
+            self.cache_manager.set(
+                self.CacheType.DEVICE_PROMPT, device_cache_key, enhanced_prompt
+            )
+            self.logger.bind(tag=TAG).info(
+                f"构建增强提示词成功，长度: {len(enhanced_prompt)}"
+            )
+            return enhanced_prompt
+
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"构建增强提示词失败: {e}")
+            return user_prompt

main/xiaozhi-server/core/utils/util.py

@@ -96,11 +96,23 @@ def is_private_ip(ip_addr):
 
 def get_ip_info(ip_addr, logger):
     try:
+        # 导入全局缓存管理器
+        from core.utils.cache.manager import cache_manager, CacheType
+
+        # 先从缓存获取
+        cached_ip_info = cache_manager.get(CacheType.IP_INFO, ip_addr)
+        if cached_ip_info is not None:
+            return cached_ip_info
+
+        # 缓存未命中，调用API
         if is_private_ip(ip_addr):
             ip_addr = ""
         url = f"https://whois.pconline.com.cn/ipJson.jsp?json=true&ip={ip_addr}"
         resp = requests.get(url).json()
         ip_info = {"city": resp.get("city")}
+
+        # 存入缓存
+        cache_manager.set(CacheType.IP_INFO, ip_addr, ip_info)
         return ip_info
     except Exception as e:
         logger.bind(tag=TAG).error(f"Error getting client ip info: {e}")

main/xiaozhi-server/plugins_func/functions/get_time.py

@@ -2,59 +2,27 @@ from datetime import datetime
 import cnlunar
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
 
-# 添加星期映射字典
-WEEKDAY_MAP = {
-    "Monday": "星期一",
-    "Tuesday": "星期二",
-    "Wednesday": "星期三",
-    "Thursday": "星期四",
-    "Friday": "星期五",
-    "Saturday": "星期六",
-    "Sunday": "星期日",
-}
-
-get_time_function_desc = {
-    "type": "function",
-    "function": {
-        "name": "get_time",
-        "description": "获取今天日期或者当前时间信息",
-        "parameters": {"type": "object", "properties": {}, "required": []},
-    },
-}
-
-
-@register_function("get_time", get_time_function_desc, ToolType.WAIT)
-def get_time():
-    """
-    获取当前的日期时间信息
-    """
-    now = datetime.now()
-    current_time = now.strftime("%H:%M:%S")
-    current_date = now.strftime("%Y-%m-%d")
-    current_weekday = WEEKDAY_MAP[now.strftime("%A")]
-    response_text = (
-        f"当前日期: {current_date}，当前时间: {current_time}， {current_weekday}"
-    )
-
-    return ActionResponse(Action.REQLLM, response_text, None)
-
-
 get_lunar_function_desc = {
     "type": "function",
     "function": {
         "name": "get_lunar",
         "description": (
-            "用于获取今天的阴历/农历和黄历信息。"
+            "用于具体日期的阴历/农历和黄历信息。"
             "用户可以指定查询内容，如：阴历日期、天干地支、节气、生肖、星座、八字、宜忌等。"
             "如果没有指定查询内容，则默认查询干支年和农历日期。"
+            "对于'今天农历是多少'、'今天农历日期'这样的基本查询，请直接使用context中的信息，不要调用此工具。"
         ),
         "parameters": {
             "type": "object",
             "properties": {
+                "date": {
+                    "type": "string",
+                    "description": "要查询的日期，格式为YYYY-MM-DD，例如2024-01-01。如果不提供，则使用当前日期",
+                },
                 "query": {
                     "type": "string",
                     "description": "要查询的内容，例如阴历日期、天干地支、节日、节气、生肖、星座、八字、宜忌等",
-                }
+                },
             },
             "required": [],
         },
@@ -63,23 +31,41 @@ get_lunar_function_desc = {
 
 
 @register_function("get_lunar", get_lunar_function_desc, ToolType.WAIT)
-def get_lunar(query=None):
+def get_lunar(date=None, query=None):
     """
     用于获取当前的阴历/农历，和天干地支、节气、生肖、星座、八字、宜忌等黄历信息
     """
-    now = datetime.now()
-    current_time = now.strftime("%H:%M:%S")
+    from core.utils.cache.manager import cache_manager, CacheType
+
+    # 如果提供了日期参数，则使用指定日期；否则使用当前日期
+    if date:
+        try:
+            now = datetime.strptime(date, "%Y-%m-%d")
+        except ValueError:
+            return ActionResponse(
+                Action.REQLLM,
+                f"日期格式错误，请使用YYYY-MM-DD格式，例如：2024-01-01",
+                None,
+            )
+    else:
+        now = datetime.now()
+
     current_date = now.strftime("%Y-%m-%d")
-    current_weekday = WEEKDAY_MAP[now.strftime("%A")]
 
     # 如果 query 为 None，则使用默认文本
     if query is None:
         query = "默认查询干支年和农历日期"
+
+    # 尝试从缓存获取农历信息
+    lunar_cache_key = f"lunar_info_{current_date}"
+    cached_lunar_info = cache_manager.get(CacheType.LUNAR, lunar_cache_key)
+    if cached_lunar_info:
+        return ActionResponse(Action.REQLLM, cached_lunar_info, None)
+
     response_text = f"根据以下信息回应用户的查询请求，并提供与{query}相关的信息：\n"
 
     lunar = cnlunar.Lunar(now, godType="8char")
     response_text += (
-        f"当前公历日期: {current_date}，当前时间: {current_time}，{current_weekday}\n"
         "农历信息：\n"
         "%s年%s%s\n" % (lunar.lunarYearCn, lunar.lunarMonthCn[:-1], lunar.lunarDayCn)
         + "干支: %s年 %s月 %s日\n" % (lunar.year8Char, lunar.month8Char, lunar.day8Char)
@@ -135,4 +121,7 @@ def get_lunar(query=None):
         + "(默认返回干支年和农历日期；仅在要求查询宜忌信息时才返回本日宜忌)"
     )
 
+    # 缓存农历信息
+    cache_manager.set(CacheType.LUNAR, lunar_cache_key, response_text)
+
     return ActionResponse(Action.REQLLM, response_text, None)

main/xiaozhi-server/plugins_func/functions/get_weather.py

@@ -151,20 +151,44 @@ def parse_weather_info(soup):
 
 @register_function("get_weather", GET_WEATHER_FUNCTION_DESC, ToolType.SYSTEM_CTL)
 def get_weather(conn, location: str = None, lang: str = "zh_CN"):
-    api_host = conn.config["plugins"]["get_weather"].get("api_host", "mj7p3y7naa.re.qweatherapi.com")
-    api_key = conn.config["plugins"]["get_weather"].get("api_key", "a861d0d5e7bf4ee1a83d9a9e4f96d4da")
+    from core.utils.cache.manager import cache_manager, CacheType
+
+    api_host = conn.config["plugins"]["get_weather"].get(
+        "api_host", "mj7p3y7naa.re.qweatherapi.com"
+    )
+    api_key = conn.config["plugins"]["get_weather"].get(
+        "api_key", "a861d0d5e7bf4ee1a83d9a9e4f96d4da"
+    )
     default_location = conn.config["plugins"]["get_weather"]["default_location"]
     client_ip = conn.client_ip
+
     # 优先使用用户提供的location参数
     if not location:
         # 通过客户端IP解析城市
         if client_ip:
-            # 动态解析IP对应的城市信息
-            ip_info = get_ip_info(client_ip, logger)
-            location = ip_info.get("city") if ip_info and "city" in ip_info else None
+            # 先从缓存获取IP对应的城市信息
+            cached_ip_info = cache_manager.get(CacheType.IP_INFO, client_ip)
+            if cached_ip_info:
+                location = cached_ip_info.get("city")
+            else:
+                # 缓存未命中，调用API获取
+                ip_info = get_ip_info(client_ip, logger)
+                if ip_info:
+                    cache_manager.set(CacheType.IP_INFO, client_ip, ip_info)
+                    location = ip_info.get("city")
+
+            if not location:
+                location = default_location
         else:
-            # 若IP解析失败或无IP，使用默认位置
+            # 若无IP，使用默认位置
             location = default_location
+    # 尝试从缓存获取完整天气报告
+    weather_cache_key = f"full_weather_{location}_{lang}"
+    cached_weather_report = cache_manager.get(CacheType.WEATHER, weather_cache_key)
+    if cached_weather_report:
+        return ActionResponse(Action.REQLLM, cached_weather_report, None)
+
+    # 缓存未命中，获取实时天气数据
     city_info = fetch_city_info(location, api_key, api_host)
     if not city_info:
         return ActionResponse(
@@ -192,4 +216,7 @@ def get_weather(conn, location: str = None, lang: str = "zh_CN"):
     # 提示语
     weather_report += "\n（如需某一天的具体天气，请告诉我日期）"
 
+    # 缓存完整的天气报告
+    cache_manager.set(CacheType.WEATHER, weather_cache_key, weather_report)
+
     return ActionResponse(Action.REQLLM, weather_report, None)