8.2-1提示词管理器

main/server/core/utils/prompt_manager.py

@@ -7,6 +7,8 @@ import os
 import cnlunar
 from typing import Dict, Any
 from config.logger import setup_logging
+from jinjia2 import Template
+
 
 TAG = __name__
 
@@ -53,11 +55,153 @@ class PromptManager:
         self.last_update_time = 0
 
         # 导入全局缓存管理器
-        from core.utils.cathe.manager import cache_manager, CacheType
+        from core.utils.cache.manager import cache_manager, CacheType
 
         self.cache_manager = cache_manager
         self.CacheType = CacheType
         
         self._load_base_template()
 
-    def 
+    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).info(f"成功从缓存中加载基础提示词模板")
+                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).info(f"成功从文件加载基础提示词模板")
+            else:
+                self.logger.bind(tag=TAG).error(f"未找到agent-base-prompt.txt文件")
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"加载基础提示词模板失败: {str(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:
+            return cached_device_prompt
+        else:
+            self.logger.bind(tag=TAG).info(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).info(f"缓存设备{device_id}的提示词")
+
+        self.logger.bind(tag=TAG).info(f"使用传入的提示词")
+        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
+            
+            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"获取位置信息失败: {str(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
+            
+            from plugins_func.functions.get_weather import get_weather
+            from plugins_func.register import ActionResponse
+
+            result = get_weather(conn, location=location, lang="zn_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"获取天气信息失败: {str(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"更新上下文信息失败: {str(e)}")
+
+    def build_enhanced_prompt(self, user_prompt: str, device_id: str = None, client_ip: str = None):
+        """构建增强提示词"""
+        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 ""
+            template = Template(self.base_prompt_template)
+            enhance_prompt = template.render(
+                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,
+                emojilist=EMOJI_List,
+            )
+            device_cache_key = f"device_prompt:{device_id}"
+            self.cache_manager.set(self.CacheType.DEVICE_PROMPT, device_cache_key, enhance_prompt)
+            self.logger.bind(tag=TAG).info(f"构建增强提示词成功")
+            return enhance_prompt
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"构建增强提示词失败: {str(e)}")
+            return user_prompt

main/server/core/utils/util.py

@@ -86,4 +86,67 @@ def get_local_ip() -> str:
         s.close()
         return local_ip
     except Exception as e:
-        return "127.0.0.1"
+        return "127.0.0.1"
+    
+
+def get_ip_info(ip_addr, logger):
+    try:
+        from core.utils.cathe.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
+        
+        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"获取IP信息失败: {str(e)}")
+        return {"city": "未知城市"}
+    
+def is_private_ip(ip_addr):
+    """
+    Check if an IP address is a private IP address (compatible with IPv4 and IPv6).
+
+    @param {string} ip_addr - The IP address to check.
+    @return {bool} True if the IP address is private, False otherwise.
+    """
+    try:
+        # Validate IPv4 or IPv6 address format
+        if not re.match(
+            r"^(\d{1,3}\.){3}\d{1,3}$|^([0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}$", ip_addr
+        ):
+            return False  # Invalid IP address format
+
+        # IPv4 private address ranges
+        if "." in ip_addr:  # IPv4 address
+            ip_parts = list(map(int, ip_addr.split(".")))
+            if ip_parts[0] == 10:
+                return True  # 10.0.0.0/8 range
+            elif ip_parts[0] == 172 and 16 <= ip_parts[1] <= 31:
+                return True  # 172.16.0.0/12 range
+            elif ip_parts[0] == 192 and ip_parts[1] == 168:
+                return True  # 192.168.0.0/16 range
+            elif ip_addr == "127.0.0.1":
+                return True  # Loopback address
+            elif ip_parts[0] == 169 and ip_parts[1] == 254:
+                return True  # Link-local address 169.254.0.0/16
+            else:
+                return False  # Not a private IPv4 address
+        else:  # IPv6 address
+            ip_addr = ip_addr.lower()
+            if ip_addr.startswith("fc00:") or ip_addr.startswith("fd00:"):
+                return True  # Unique Local Addresses (FC00::/7)
+            elif ip_addr == "::1":
+                return True  # Loopback address
+            elif ip_addr.startswith("fe80:"):
+                return True  # Link-local unicast addresses (FE80::/10)
+            else:
+                return False  # Not a private IPv6 address
+
+    except (ValueError, IndexError):
+        return False  # IP address format error or insufficient segments

main/server/plugins_func/functions/get_weather.py

@@ -0,0 +1,216 @@
+import requests
+from bs4 import BeautifulSoup
+from config.logger import setup_logging
+from plugins_func.register import register_function,ToolType, ActionResponse, Action
+from core.utils.util import get_ip_info
+
+
+TAG = __name__
+logger = setup_logging()
+
+
+GET_WEATHER_FUNCTION_DESC = {
+    "type": "function",
+    "function": {
+        "name": "get_weather",
+        "description": (
+            "获取某个地点的天气，用户应提供一个位置，比如用户说杭州天气，参数为：杭州。"
+            "如果用户说的是省份，默认用省会城市。如果用户说的不是省份或城市而是一个地名，默认用该地所在省份的省会城市。"
+            "如果用户没有指明地点，说“天气怎么样”，”今天天气如何“，location参数为空"
+        ),
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "location": {
+                    "type": "string",
+                    "description": "地点名，例如杭州。可选参数，如果不提供则不传",
+                },
+                "lang": {
+                    "type": "string",
+                    "description": "返回用户使用的语言code，例如zh_CN/zh_HK/en_US/ja_JP等，默认zh_CN",
+                },
+            },
+            "required": ["lang"],
+        },
+    },
+}
+
+HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+        "(KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36"
+    )
+}
+
+# 天气代码 https://dev.qweather.com/docs/resource/icons/#weather-icons
+WEATHER_CODE_MAP = {
+    "100": "晴",
+    "101": "多云",
+    "102": "少云",
+    "103": "晴间多云",
+    "104": "阴",
+    "150": "晴",
+    "151": "多云",
+    "152": "少云",
+    "153": "晴间多云",
+    "300": "阵雨",
+    "301": "强阵雨",
+    "302": "雷阵雨",
+    "303": "强雷阵雨",
+    "304": "雷阵雨伴有冰雹",
+    "305": "小雨",
+    "306": "中雨",
+    "307": "大雨",
+    "308": "极端降雨",
+    "309": "毛毛雨/细雨",
+    "310": "暴雨",
+    "311": "大暴雨",
+    "312": "特大暴雨",
+    "313": "冻雨",
+    "314": "小到中雨",
+    "315": "中到大雨",
+    "316": "大到暴雨",
+    "317": "暴雨到大暴雨",
+    "318": "大暴雨到特大暴雨",
+    "350": "阵雨",
+    "351": "强阵雨",
+    "399": "雨",
+    "400": "小雪",
+    "401": "中雪",
+    "402": "大雪",
+    "403": "暴雪",
+    "404": "雨夹雪",
+    "405": "雨雪天气",
+    "406": "阵雨夹雪",
+    "407": "阵雪",
+    "408": "小到中雪",
+    "409": "中到大雪",
+    "410": "大到暴雪",
+    "456": "阵雨夹雪",
+    "457": "阵雪",
+    "499": "雪",
+    "500": "薄雾",
+    "501": "雾",
+    "502": "霾",
+    "503": "扬沙",
+    "504": "浮尘",
+    "507": "沙尘暴",
+    "508": "强沙尘暴",
+    "509": "浓雾",
+    "510": "强浓雾",
+    "511": "中度霾",
+    "512": "重度霾",
+    "513": "严重霾",
+    "514": "大雾",
+    "515": "特强浓雾",
+    "900": "热",
+    "901": "冷",
+    "999": "未知",
+}
+
+def fetch_city_info(location, api_key, api_host):
+    url = f"https://{api_host}/geo/v2/city/lookup?key={api_key}&location={location}&lang=zh"
+    response = requests.get(url, headers=HEADERS).json()
+    return response.get("location", [])[0] if response.get("location") else None
+
+
+def fetch_weather_page(url):
+    response = requests.get(url, headers=HEADERS)
+    return BeautifulSoup(response.text, "html.parser") if response.ok else None
+
+
+def parse_weather_info(soup):
+    city_name = soup.select_one("h1.c-submenu__location").get_text(strip=True)
+
+    current_abstract = soup.select_one(".c-city-weather-current .current-abstract")
+    current_abstract = (
+        current_abstract.get_text(strip=True) if current_abstract else "未知"
+    )
+
+    current_basic = {}
+    for item in soup.select(
+        ".c-city-weather-current .current-basic .current-basic___item"
+    ):
+        parts = item.get_text(strip=True, separator=" ").split(" ")
+        if len(parts) == 2:
+            key, value = parts[1], parts[0]
+            current_basic[key] = value
+
+    temps_list = []
+    for row in soup.select(".city-forecast-tabs__row")[:7]:  # 取前7天的数据
+        date = row.select_one(".date-bg .date").get_text(strip=True)
+        weather_code = (
+            row.select_one(".date-bg .icon")["src"].split("/")[-1].split(".")[0]
+        )
+        weather = WEATHER_CODE_MAP.get(weather_code, "未知")
+        temps = [span.get_text(strip=True) for span in row.select(".tmp-cont .temp")]
+        high_temp, low_temp = (temps[0], temps[-1]) if len(temps) >= 2 else (None, None)
+        temps_list.append((date, weather, high_temp, low_temp))
+
+    return city_name, current_abstract, current_basic, temps_list
+
+
+@register_function("get_weather", GET_WEATHER_FUNCTION_DESC, ToolType.SYSTEM_CTL)
+def get_weather(conn, location: str = None, lang: str = "zh_CN"):
+    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
+
+    if not location:
+        if client_ip:
+            cached_ip_info = cache_manager.get(CacheType.IP_INFO, client_ip)
+            if cached_ip_info:
+                location = cached_ip_info.get("city")
+            else:
+                ip_info = get_ip_info(client_ip, logger)
+                if ip_info:
+                    location = ip_info.get("city")
+                    cache_manager.set(CacheType.IP_INFO, client_ip, ip_info)
+        else:
+            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(
+            Action.REQLLM, f"未找到相关的城市: {location}，请确认地点是否正确", None
+        )
+    soup = fetch_weather_page(city_info["fxLink"])
+    if not soup:
+        return ActionResponse(Action.REQLLM, None, "请求失败")
+    city_name, current_abstract, current_basic, temps_list = parse_weather_info(soup)
+
+    weather_report = f"您查询的位置是：{city_name}\n\n当前天气: {current_abstract}\n"
+
+    # 添加有效的当前天气参数
+    if current_basic:
+        weather_report += "详细参数：\n"
+        for key, value in current_basic.items():
+            if value != "0":  # 过滤无效值
+                weather_report += f"  · {key}: {value}\n"
+
+    # 添加7天预报
+    weather_report += "\n未来7天预报：\n"
+    for date, weather, high, low in temps_list:
+        weather_report += f"{date}: {weather}，气温 {low}~{high}\n"
+
+    # 提示语
+    weather_report += "\n（如需某一天的具体天气，请告诉我日期）"
+
+    # 缓存完整的天气报告
+    cache_manager.set(CacheType.WEATHER, weather_cache_key, weather_report)
+
+    return ActionResponse(Action.REQLLM, weather_report, None)

main/server/plugins_func/register.py

@@ -54,4 +54,73 @@ class DeviceTypeRegister:
     """
     
     def __init__(self):
-        self.type_functions = {}   # type_signature -> {func_name: FunctionItem}
+        self.type_functions = {}   # type_signature -> {func_name: FunctionItem}
+
+    def generate_device_type_id(self, descripter):
+        properties = sorted(descripter["properties"].keys())
+        method = sorted(descripter["methods"].keys())
+        type_signature = (
+            f"{descripter['name']}:{','.join(properties)}:{','.join(method)}"
+        )
+        return type_signature
+    
+    def get_device_functions(self,type_id):
+        return self.type_functions_get(type_id,{})
+    
+    def register_device_type(self, type_id, functions):
+        if type_id not in self.type_functions:
+            self.type_functions[type_id] = functions
+
+
+
+# 初始化函数注册字典
+all_function_register = {}
+
+
+def register_function(name, desc, type=None):
+    """注册函数到函数字典的装饰器"""
+
+    def decorator(func):
+        logger.bind(tag=TAG).debug(f"设备函数{name}已加载")
+        return func
+    
+    return decorator
+
+
+class FunctionRegistry:
+    def __init__(self):
+        self.function_register = {}
+        self.logger = setup_logging()
+
+    def register_function(self, name, func_item=None):
+        # 如果提供了func_item，直接注册
+        if func_item:
+            self.function_register[name] = func_item
+            self.logger.bind(tag=TAG).debug(f"函数{name}直接注册成功")
+            return func_item
+        
+        # 否则从all_function_register中获取
+        func = all_function_register.get(name)
+        if not func:
+            self.logger.bind(tag=TAG).error(f"函数{name}未找到")
+            return None
+        self.function_register[name] = func
+        self.logger.bind(tag=TAG).debug(f"函数{name}从all_function_register中获取并注册成功")
+        return func
+    
+    def unregister_function(self, name):
+        if name not in self.function_register:
+            self.logger.bind(tag=TAG).error(f"函数{name}未找到")
+            return None
+        self.function_register.pop(name,None)
+        self.logger.bind(tag=TAG).debug(f"函数{name}注销成功")
+        return True
+    
+    def get_function(self, name):
+        return self.function_register.get(name)
+     
+    def get_all_functions(self):
+        return self.function_register
+    
+    def get_all_function_desc(self):
+        return [func.description for _ , func in self.function_register.items()]