gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(chat): 重构SmartPrompt系统使用分层参数架构和共享工具

Browse Source 
将SmartPrompt系统从平面参数结构重构为分层架构，引入PromptCoreParams、
PromptFeatureParams和PromptContentParams三个层级，提高代码组织性和可维护性。

主要变更：
- 使用新的分层参数结构替代原有的平面参数系统
- 集成PromptUtils共享工具类，消除代码重复
- 添加性能优化：缓存机制、超时控制和性能监控
- 增强错误处理，提供优雅的降级机制
- 添加SmartPromptHealthChecker用于系统健康检查
- 保持向后兼容性，通过属性访问器维持现有API

此重构显著提升了代码的可维护性、性能和可测试性，同时为未来功能
扩展奠定了更好的架构基础。
This commit is contained in:
Windpicker-owo
2025-08-31 17:47:19 +08:00
parent 202a5016b0
commit e8e401f656
4 changed files with 1313 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
src/chat/replyer/default_generator.py
View File

@@ -6,6 +6,7 @@ import re
from typing import List, Optional, Dict, Any, Tuple
from datetime import datetime
from src.chat.utils.prompt_utils import PromptUtils
from src.mais4u.mai_think import mai_thinking_manager
from src.common.logger import get_logger
from src.config.config import global_config, model_config
@@ -657,26 +658,8 @@ class DefaultReplyer:
return ""
def _parse_reply_target(self, target_message: str) -> Tuple[str, str]:
"""解析回复目标消息
Args:
target_message: 目标消息，格式为 "发送者:消息内容" 或 "发送者：消息内容"
Returns:
Tuple[str, str]: (发送者名称, 消息内容)
"""
sender = ""
target = ""
# 添加None检查，防止NoneType错误
if target_message is None:
return sender, target
if ":" in target_message or "：" in target_message:
# 使用正则表达式匹配中文或英文冒号
parts = re.split(pattern=r"[:：]", string=target_message, maxsplit=1)
if len(parts) == 2:
sender = parts[0].strip()
target = parts[1].strip()
return sender, target
"""解析回复目标消息 - 使用共享工具"""
return PromptUtils.parse_reply_target(target_message)
async def build_keywords_reaction_prompt(self, target: Optional[str]) -> str:
"""构建关键词反应提示
@@ -962,7 +945,9 @@ class DefaultReplyer:
self.build_tool_info(chat_talking_prompt_short, sender, target, enable_tool=enable_tool), "tool_info"
),
self._time_and_run_task(self.get_prompt_info(chat_talking_prompt_short, reply_to), "prompt_info"),
self._time_and_run_task(self._build_cross_context_block(chat_id, target_user_info), "cross_context"),
self._time_and_run_task(
PromptUtils.build_cross_context_block(chat_id, target_user_info, current_prompt_mode), "cross_context"
),
)
# 任务名称中英文映射
@@ -1037,7 +1022,6 @@ class DefaultReplyer:
# 根据配置选择模板
current_prompt_mode = global_config.personality.prompt_mode
# 构建SmartPromptParameters - 包含所有必需参数
prompt_params = SmartPromptParameters(
chat_id=chat_id,
is_group_chat=is_group_chat,
@@ -1053,6 +1037,7 @@ class DefaultReplyer:
message_list_before_short=message_list_before_short,
chat_talking_prompt_short=chat_talking_prompt_short,
target_user_info=target_user_info,
# 传递已构建的参数
expression_habits_block=expression_habits_block,
relation_info=relation_info,
memory_block=memory_block,
@@ -1069,14 +1054,14 @@ class DefaultReplyer:
mood_prompt=mood_prompt,
action_descriptions=action_descriptions,
)
# 使用重构后的SmartPrompt系统
smart_prompt = SmartPrompt(
template_name=None, # 由current_prompt_mode自动选择
parameters=prompt_params
)
prompt_text = await smart_prompt.build_prompt()
return prompt_text
async def build_prompt_rewrite_context(
@@ -1177,47 +1162,31 @@ class DefaultReplyer:
template_name = "default_expressor_prompt"
# 使用重构后的SmartPrompt系统
# 使用重构后的SmartPrompt系统 - Expressor模式
prompt_params = SmartPromptParameters(
chat_id=chat_id,
is_group_chat=is_group_chat,
sender=sender,
target="", # 重构时使用raw_reply
target=raw_reply, # Expressor模式使用raw_reply作为target
reply_to=f"{sender}:{target}" if sender and target else reply_to,
extra_info="", # 重构模式特殊处理
expression_habits_block=expression_habits_block,
relation_info=relation_info,
extra_info="", # Expressor模式不需要额外信息
current_prompt_mode="minimal", # Expressor使用minimal模式
chat_talking_prompt_short=chat_talking_prompt_half,
time_block=time_block,
identity_block=identity_block,
reply_target_block=reply_target_block,
mood_prompt=mood_prompt,
keywords_reaction_prompt=keywords_reaction_prompt,
moderation_prompt_block=moderation_prompt_block,
current_prompt_mode=global_config.personality.prompt_mode,
chat_talking_prompt_short=chat_talking_prompt_half,
# 添加已构建的表达习惯和关系信息
expression_habits_block=expression_habits_block,
relation_info=relation_info,
)
smart_prompt = SmartPrompt(parameters=prompt_params)
prompt_text = await smart_prompt.build_prompt()
# 重构为expressor专用格式
expressor_params = {
'expression_habits_block': expression_habits_block,
'relation_info_block': relation_info,
'chat_target': chat_target_1,
'time_block': time_block,
'chat_info': chat_talking_prompt_half,
'identity': identity_block,
'chat_target_2': chat_target_2,
'reply_target_block': reply_target_block,
'raw_reply': raw_reply,
'reason': reason,
'mood_state': mood_prompt,
'reply_style': global_config.personality.reply_style,
'keywords_reaction_prompt': keywords_reaction_prompt,
'moderation_prompt': moderation_prompt_block,
}
return await global_prompt_manager.format_prompt("default_expressor_prompt", **expressor_params)
return prompt_text
async def _build_single_sending_message(
self,

345
src/chat/utils/prompt_parameters.py Normal file
View File

@@ -0,0 +1,345 @@
"""
智能提示词参数模块 - 优化参数结构
将SmartPromptParameters拆分为多个专用参数类
"""
from dataclasses import dataclass, field
from typing import Dict, Any, Optional, List, Literal
@dataclass
class PromptCoreParams:
"""核心参数类 - 包含构建提示词的基本参数"""
chat_id: str = ""
is_group_chat: bool = False
sender: str = ""
target: str = ""
reply_to: str = ""
extra_info: str = ""
current_prompt_mode: Literal["s4u", "normal", "minimal"] = "s4u"
def validate(self) -> List[str]:
"""验证核心参数"""
errors = []
if not isinstance(self.chat_id, str):
errors.append("chat_id必须是字符串类型")
if not isinstance(self.reply_to, str):
errors.append("reply_to必须是字符串类型")
if self.current_prompt_mode not in ["s4u", "normal", "minimal"]:
errors.append("current_prompt_mode必须是's4u'、'normal'或'minimal'")
return errors
@dataclass
class PromptFeatureParams:
"""功能参数类 - 控制各种功能的开关"""
enable_tool: bool = True
enable_memory: bool = True
enable_expression: bool = True
enable_relation: bool = True
enable_cross_context: bool = True
enable_knowledge: bool = True
enable_cache: bool = True
# 性能和缓存控制
cache_ttl: int = 300
max_context_messages: int = 50
# 调试选项
debug_mode: bool = False
@dataclass
class PromptContentParams:
"""内容参数类 - 包含已构建的内容块"""
# 聊天历史和上下文
chat_target_info: Optional[Dict[str, Any]] = None
message_list_before_now_long: List[Dict[str, Any]] = field(default_factory=list)
message_list_before_short: List[Dict[str, Any]] = field(default_factory=list)
chat_talking_prompt_short: str = ""
target_user_info: Optional[Dict[str, Any]] = None
# 已构建的内容块
expression_habits_block: str = ""
relation_info: str = ""
memory_block: str = ""
tool_info: str = ""
prompt_info: str = ""
cross_context_block: str = ""
# 其他内容块
keywords_reaction_prompt: str = ""
extra_info_block: str = ""
time_block: str = ""
identity_block: str = ""
schedule_block: str = ""
moderation_prompt_block: str = ""
reply_target_block: str = ""
mood_prompt: str = ""
action_descriptions: str = ""
def has_prebuilt_content(self) -> bool:
"""检查是否有预构建的内容"""
return any([
self.expression_habits_block,
self.relation_info,
self.memory_block,
self.tool_info,
self.prompt_info,
self.cross_context_block
])
@dataclass
class SmartPromptParameters:
"""
智能提示词参数系统 - 重构版本
组合多个专用参数类，提供统一的接口
"""
# 核心参数
core: PromptCoreParams = field(default_factory=PromptCoreParams)
# 功能参数
features: PromptFeatureParams = field(default_factory=PromptFeatureParams)
# 内容参数
content: PromptContentParams = field(default_factory=PromptContentParams)
# 兼容性属性 - 提供与旧代码的兼容性
@property
def chat_id(self) -> str:
return self.core.chat_id
@chat_id.setter
def chat_id(self, value: str):
self.core.chat_id = value
@property
def is_group_chat(self) -> bool:
return self.core.is_group_chat
@is_group_chat.setter
def is_group_chat(self, value: bool):
self.core.is_group_chat = value
@property
def sender(self) -> str:
return self.core.sender
@sender.setter
def sender(self, value: str):
self.core.sender = value
@property
def target(self) -> str:
return self.core.target
@target.setter
def target(self, value: str):
self.core.target = value
@property
def reply_to(self) -> str:
return self.core.reply_to
@reply_to.setter
def reply_to(self, value: str):
self.core.reply_to = value
@property
def extra_info(self) -> str:
return self.core.extra_info
@extra_info.setter
def extra_info(self, value: str):
self.core.extra_info = value
@property
def current_prompt_mode(self) -> str:
return self.core.current_prompt_mode
@current_prompt_mode.setter
def current_prompt_mode(self, value: str):
self.core.current_prompt_mode = value
@property
def enable_tool(self) -> bool:
return self.features.enable_tool
@enable_tool.setter
def enable_tool(self, value: bool):
self.features.enable_tool = value
@property
def enable_memory(self) -> bool:
return self.features.enable_memory
@enable_memory.setter
def enable_memory(self, value: bool):
self.features.enable_memory = value
@property
def enable_cache(self) -> bool:
return self.features.enable_cache
@enable_cache.setter
def enable_cache(self, value: bool):
self.features.enable_cache = value
@property
def cache_ttl(self) -> int:
return self.features.cache_ttl
@cache_ttl.setter
def cache_ttl(self, value: int):
self.features.cache_ttl = value
@property
def expression_habits_block(self) -> str:
return self.content.expression_habits_block
@expression_habits_block.setter
def expression_habits_block(self, value: str):
self.content.expression_habits_block = value
@property
def relation_info(self) -> str:
return self.content.relation_info
@relation_info.setter
def relation_info(self, value: str):
self.content.relation_info = value
@property
def memory_block(self) -> str:
return self.content.memory_block
@memory_block.setter
def memory_block(self, value: str):
self.content.memory_block = value
@property
def tool_info(self) -> str:
return self.content.tool_info
@tool_info.setter
def tool_info(self, value: str):
self.content.tool_info = value
@property
def prompt_info(self) -> str:
return self.content.prompt_info
@prompt_info.setter
def prompt_info(self, value: str):
self.content.prompt_info = value
@property
def cross_context_block(self) -> str:
return self.content.cross_context_block
@cross_context_block.setter
def cross_context_block(self, value: str):
self.content.cross_context_block = value
# 兼容性方法 - 支持旧代码的直接访问
def validate(self) -> List[str]:
"""参数验证"""
errors = self.core.validate()
# 验证功能参数
if self.features.cache_ttl <= 0:
errors.append("cache_ttl必须大于0")
if self.features.max_context_messages <= 0:
errors.append("max_context_messages必须大于0")
return errors
def get_needed_build_tasks(self) -> List[str]:
"""获取需要执行的任务列表"""
tasks = []
if self.features.enable_expression and not self.content.expression_habits_block:
tasks.append("expression_habits")
if self.features.enable_memory and not self.content.memory_block:
tasks.append("memory_block")
if self.features.enable_relation and not self.content.relation_info:
tasks.append("relation_info")
if self.features.enable_tool and not self.content.tool_info:
tasks.append("tool_info")
if self.features.enable_knowledge and not self.content.prompt_info:
tasks.append("knowledge_info")
if self.features.enable_cross_context and not self.content.cross_context_block:
tasks.append("cross_context")
return tasks
@classmethod
def from_legacy_params(cls, **kwargs) -> 'SmartPromptParameters':
"""
从旧版参数创建新参数对象
Args:
**kwargs: 旧版参数
Returns:
SmartPromptParameters: 新参数对象
"""
# 创建核心参数
core_params = PromptCoreParams(
chat_id=kwargs.get("chat_id", ""),
is_group_chat=kwargs.get("is_group_chat", False),
sender=kwargs.get("sender", ""),
target=kwargs.get("target", ""),
reply_to=kwargs.get("reply_to", ""),
extra_info=kwargs.get("extra_info", ""),
current_prompt_mode=kwargs.get("current_prompt_mode", "s4u"),
)
# 创建功能参数
feature_params = PromptFeatureParams(
enable_tool=kwargs.get("enable_tool", True),
enable_memory=kwargs.get("enable_memory", True),
enable_expression=kwargs.get("enable_expression", True),
enable_relation=kwargs.get("enable_relation", True),
enable_cross_context=kwargs.get("enable_cross_context", True),
enable_knowledge=kwargs.get("enable_knowledge", True),
enable_cache=kwargs.get("enable_cache", True),
cache_ttl=kwargs.get("cache_ttl", 300),
max_context_messages=kwargs.get("max_context_messages", 50),
debug_mode=kwargs.get("debug_mode", False),
)
# 创建内容参数
content_params = PromptContentParams(
chat_target_info=kwargs.get("chat_target_info"),
message_list_before_now_long=kwargs.get("message_list_before_now_long", []),
message_list_before_short=kwargs.get("message_list_before_short", []),
chat_talking_prompt_short=kwargs.get("chat_talking_prompt_short", ""),
target_user_info=kwargs.get("target_user_info"),
expression_habits_block=kwargs.get("expression_habits_block", ""),
relation_info=kwargs.get("relation_info", ""),
memory_block=kwargs.get("memory_block", ""),
tool_info=kwargs.get("tool_info", ""),
prompt_info=kwargs.get("prompt_info", ""),
cross_context_block=kwargs.get("cross_context_block", ""),
keywords_reaction_prompt=kwargs.get("keywords_reaction_prompt", ""),
extra_info_block=kwargs.get("extra_info_block", ""),
time_block=kwargs.get("time_block", ""),
identity_block=kwargs.get("identity_block", ""),
schedule_block=kwargs.get("schedule_block", ""),
moderation_prompt_block=kwargs.get("moderation_prompt_block", ""),
reply_target_block=kwargs.get("reply_target_block", ""),
mood_prompt=kwargs.get("mood_prompt", ""),
action_descriptions=kwargs.get("action_descriptions", ""),
)
return cls(
core=core_params,
features=feature_params,
content=content_params
)

347
src/chat/utils/prompt_utils.py Normal file
View File

@@ -0,0 +1,347 @@
"""
共享提示词工具模块 - 消除重复代码
提供统一的工具函数供DefaultReplyer和SmartPrompt使用
"""
import re
import time
import asyncio
from typing import Dict, Any, List, Optional, Tuple, Union
from datetime import datetime
from src.common.logger import get_logger
from src.config.config import global_config
from src.chat.utils.chat_message_builder import (
build_readable_messages,
get_raw_msg_before_timestamp_with_chat,
build_readable_messages_with_id,
)
from src.chat.message_receive.chat_stream import get_chat_manager
from src.person_info.person_info import get_person_info_manager
logger = get_logger("prompt_utils")
class PromptUtils:
"""提示词工具类 - 提供共享功能"""
@staticmethod
def parse_reply_target(target_message: str) -> Tuple[str, str]:
"""
解析回复目标消息 - 统一实现
Args:
target_message: 目标消息，格式为 "发送者:消息内容" 或 "发送者：消息内容"
Returns:
Tuple[str, str]: (发送者名称, 消息内容)
"""
sender = ""
target = ""
# 添加None检查，防止NoneType错误
if target_message is None:
return sender, target
if ":" in target_message or "：" in target_message:
# 使用正则表达式匹配中文或英文冒号
parts = re.split(pattern=r"[:：]", string=target_message, maxsplit=1)
if len(parts) == 2:
sender = parts[0].strip()
target = parts[1].strip()
return sender, target
@staticmethod
async def build_cross_context_block(
chat_id: str,
target_user_info: Optional[Dict[str, Any]],
current_prompt_mode: str
) -> str:
"""
构建跨群聊上下文 - 统一实现
Args:
chat_id: 当前聊天ID
target_user_info: 目标用户信息
current_prompt_mode: 当前提示模式
Returns:
str: 跨群上下文块
"""
if not global_config.cross_context.enable:
return ""
# 找到当前群聊所在的共享组
target_group = None
current_stream = get_chat_manager().get_stream(chat_id)
if not current_stream or not current_stream.group_info:
return ""
current_chat_raw_id = current_stream.group_info.group_id
for group in global_config.cross_context.groups:
if str(current_chat_raw_id) in group.chat_ids:
target_group = group
break
if not target_group:
return ""
# 根据prompt_mode选择策略
other_chat_raw_ids = [chat_id for chat_id in target_group.chat_ids if chat_id != str(current_chat_raw_id)]
cross_context_messages = []
if current_prompt_mode == "normal":
# normal模式：获取其他群聊的最近N条消息
for chat_raw_id in other_chat_raw_ids:
stream_id = get_chat_manager().get_stream_id(current_stream.platform, chat_raw_id, is_group=True)
if not stream_id:
continue
messages = get_raw_msg_before_timestamp_with_chat(
chat_id=stream_id,
timestamp=time.time(),
limit=5, # 可配置
)
if messages:
chat_name = get_chat_manager().get_stream_name(stream_id) or stream_id
formatted_messages, _ = build_readable_messages_with_id(messages, timestamp_mode="relative")
cross_context_messages.append(f"[以下是来自\"{chat_name}\"的近期消息]\n{formatted_messages}")
elif current_prompt_mode == "s4u":
# s4u模式：获取当前发言用户在其他群聊的消息
if target_user_info:
user_id = target_user_info.get("user_id")
if user_id:
for chat_raw_id in other_chat_raw_ids:
stream_id = get_chat_manager().get_stream_id(
current_stream.platform, chat_raw_id, is_group=True
)
if not stream_id:
continue
messages = get_raw_msg_before_timestamp_with_chat(
chat_id=stream_id,
timestamp=time.time(),
limit=20, # 获取更多消息以供筛选
)
user_messages = [msg for msg in messages if msg.get("user_id") == user_id][
-5:
] # 筛选并取最近5条
if user_messages:
chat_name = get_chat_manager().get_stream_name(stream_id) or stream_id
user_name = (
target_user_info.get("person_name") or
target_user_info.get("user_nickname") or user_id
)
formatted_messages, _ = build_readable_messages_with_id(
user_messages, timestamp_mode="relative"
)
cross_context_messages.append(
f"[以下是\"{user_name}\"在\"{chat_name}\"的近期发言]\n{formatted_messages}"
)
if not cross_context_messages:
return ""
return "# 跨群上下文参考\n" + "\n\n".join(cross_context_messages) + "\n"
@staticmethod
def parse_reply_target_id(reply_to: str) -> str:
"""
解析回复目标中的用户ID
Args:
reply_to: 回复目标字符串
Returns:
str: 用户ID
"""
if not reply_to:
return ""
# 复用parse_reply_target方法的逻辑
sender, _ = PromptUtils.parse_reply_target(reply_to)
if not sender:
return ""
# 获取用户ID
person_info_manager = get_person_info_manager()
person_id = person_info_manager.get_person_id_by_person_name(sender)
if person_id:
user_id = person_info_manager.get_value_sync(person_id, "user_id")
return str(user_id) if user_id else ""
return ""
class DependencyChecker:
"""依赖检查器 - 检查关键组件的可用性"""
@staticmethod
async def check_expression_dependencies() -> Tuple[bool, List[str]]:
"""
检查表达系统依赖
Returns:
Tuple[bool, List[str]]: (是否可用, 缺失的依赖列表)
"""
missing_deps = []
try:
from src.chat.express.expression_selector import expression_selector
# 尝试访问一个方法以确保模块可用
if not hasattr(expression_selector, 'select_suitable_expressions_llm'):
missing_deps.append("expression_selector.select_suitable_expressions_llm")
except ImportError as e:
missing_deps.append(f"expression_selector: {str(e)}")
return len(missing_deps) == 0, missing_deps
@staticmethod
async def check_memory_dependencies() -> Tuple[bool, List[str]]:
"""
检查记忆系统依赖
Returns:
Tuple[bool, List[str]]: (是否可用, 缺失的依赖列表)
"""
missing_deps = []
try:
from src.chat.memory_system.memory_activator import MemoryActivator
from src.chat.memory_system.vector_instant_memory import VectorInstantMemoryV2
except ImportError as e:
missing_deps.append(f"memory_system: {str(e)}")
return len(missing_deps) == 0, missing_deps
@staticmethod
async def check_tool_dependencies() -> Tuple[bool, List[str]]:
"""
检查工具系统依赖
Returns:
Tuple[bool, List[str]]: (是否可用, 缺失的依赖列表)
"""
missing_deps = []
try:
from src.plugin_system.core.tool_use import ToolExecutor
except ImportError as e:
missing_deps.append(f"tool_executor: {str(e)}")
return len(missing_deps) == 0, missing_deps
@staticmethod
async def check_knowledge_dependencies() -> Tuple[bool, List[str]]:
"""
检查知识系统依赖
Returns:
Tuple[bool, List[str]]: (是否可用, 缺失的依赖列表)
"""
missing_deps = []
try:
from src.plugins.built_in.knowledge.lpmm_get_knowledge import SearchKnowledgeFromLPMMTool
except ImportError as e:
missing_deps.append(f"knowledge_tool: {str(e)}")
return len(missing_deps) == 0, missing_deps
@staticmethod
async def check_all_dependencies() -> Dict[str, Tuple[bool, List[str]]]:
"""
检查所有依赖
Returns:
Dict[str, Tuple[bool, List[str]]]: 各系统依赖状态
"""
return {
"expression": await DependencyChecker.check_expression_dependencies(),
"memory": await DependencyChecker.check_memory_dependencies(),
"tool": await DependencyChecker.check_tool_dependencies(),
"knowledge": await DependencyChecker.check_knowledge_dependencies(),
}
class SmartPromptCache:
"""智能提示词缓存系统 - 分层缓存实现"""
def __init__(self):
self._l1_cache: Dict[str, Tuple[str, float]] = {} # 内存缓存: {key: (value, timestamp)}
self._l2_cache_enabled = False # 是否启用L2缓存
self._cache_ttl = 300 # 默认缓存TTL: 5分钟
def enable_l2_cache(self, enabled: bool = True):
"""启用或禁用L2缓存"""
self._l2_cache_enabled = enabled
def set_cache_ttl(self, ttl: int):
"""设置缓存TTL（秒）"""
self._cache_ttl = ttl
def _generate_key(self, chat_id: str, prompt_mode: str, reply_to: str) -> str:
"""生成缓存键"""
import hashlib
key_content = f"{chat_id}_{prompt_mode}_{reply_to}"
return hashlib.md5(key_content.encode()).hexdigest()
def get(self, chat_id: str, prompt_mode: str, reply_to: str) -> Optional[str]:
"""获取缓存值"""
cache_key = self._generate_key(chat_id, prompt_mode, reply_to)
# 检查L1缓存
if cache_key in self._l1_cache:
value, timestamp = self._l1_cache[cache_key]
if time.time() - timestamp < self._cache_ttl:
logger.debug(f"L1缓存命中: {cache_key}")
return value
else:
# 缓存过期，清理
del self._l1_cache[cache_key]
# TODO: 实现L2缓存（如Redis）
# if self._l2_cache_enabled:
# return self._get_from_l2_cache(cache_key)
return None
def set(self, chat_id: str, prompt_mode: str, reply_to: str, value: str):
"""设置缓存值"""
cache_key = self._generate_key(chat_id, prompt_mode, reply_to)
# 设置L1缓存
self._l1_cache[cache_key] = (value, time.time())
# TODO: 实现L2缓存
# if self._l2_cache_enabled:
# self._set_to_l2_cache(cache_key, value)
# 定期清理过期缓存
if len(self._l1_cache) > 1000: # 缓存条目过多时清理
self._clean_expired_cache()
def _clean_expired_cache(self):
"""清理过期缓存"""
current_time = time.time()
expired_keys = [
key for key, (_, timestamp) in self._l1_cache.items()
if current_time - timestamp >= self._cache_ttl
]
for key in expired_keys:
del self._l1_cache[key]
logger.debug(f"清理过期缓存: {len(expired_keys)} 个条目")
def clear(self):
"""清空所有缓存"""
self._l1_cache.clear()
# TODO: 清空L2缓存
logger.info("缓存已清空")
def get_stats(self) -> Dict[str, Any]:
"""获取缓存统计信息"""
return {
"l1_cache_size": len(self._l1_cache),
"l2_cache_enabled": self._l2_cache_enabled,
"cache_ttl": self._cache_ttl,
}

776
src/chat/utils/smart_prompt.py
View File

@@ -1,6 +1,6 @@
"""
智能Prompt系统 - 完全重构版本
基于原有DefaultReplyer的完整功能集成
基于原有DefaultReplyer的完整功能集成，使用新的参数结构
"""
import asyncio
import time
@@ -19,70 +19,70 @@ from src.chat.utils.chat_message_builder import (
replace_user_references_sync,
)
from src.person_info.person_info import get_person_info_manager
from src.plugin_system.core.tool_use import ToolExecutor
from src.chat.utils.prompt_utils import PromptUtils
from src.chat.utils.prompt_parameters import PromptCoreParams, PromptFeatureParams, PromptContentParams
logger = get_logger("smart_prompt")
# 重新导出参数类以保持兼容性
from src.chat.utils.prompt_parameters import (
PromptCoreParams,
PromptFeatureParams,
PromptContentParams
)
@dataclass
class SmartPromptParameters:
"""完整的智能提示词参数系统"""
"""兼容的智能提示词参数系统 - 使用分层架构"""
# 从原有DefaultReplyer提取的所有必需参数
chat_id: str = ""
is_group_chat: bool = False
sender: str = ""
target: str = ""
reply_to: str = ""
extra_info: str = ""
available_actions: Dict[str, Any] = field(default_factory=dict)
# 核心参数 (从PromptCoreParams继承)
core: PromptCoreParams = field(default_factory=PromptCoreParams)
# 原有构建函数所需的参数
chat_target_info: Optional[Dict[str, Any]] = None
message_list_before_now_long: List[Dict[str, Any]] = field(default_factory=list)
message_list_before_short: List[Dict[str, Any]] = field(default_factory=list)
chat_talking_prompt_short: str = ""
target_user_info: Optional[Dict[str, Any]] = None
expression_habits_block: str = ""
relation_info: str = ""
memory_block: str = ""
tool_info: str = ""
prompt_info: str = ""
cross_context_block: str = ""
keywords_reaction_prompt: str = ""
extra_info_block: str = ""
time_block: str = ""
identity_block: str = ""
schedule_block: str = ""
moderation_prompt_block: str = ""
reply_target_block: str = ""
mood_prompt: str = ""
action_descriptions: str = ""
# 功能参数 (从PromptFeatureParams继承)
features: PromptFeatureParams = field(default_factory=PromptFeatureParams)
# 行为配置
current_prompt_mode: Literal["s4u", "normal", "minimal"] = "s4u"
enable_tool: bool = True
enable_memory: bool = True
enable_expression: bool = True
enable_relation: bool = True
enable_cross_context: bool = True
enable_knowledge: bool = True
# 内容参数 (从PromptContentParams继承)
content: PromptContentParams = field(default_factory=PromptContentParams)
# 性能和缓存控制
# 配置和兼容属性
enable_cache: bool = True
cache_ttl: int = 300
max_context_messages: int = 50
# 调试选项
debug_mode: bool = False
# 为了向下兼容，提供属性访问
@property
def chat_id(self) -> str:
return self.core.chat_id
@chat_id.setter
def chat_id(self, value: str):
self.core.chat_id = value
@property
def reply_to(self) -> str:
return self.core.reply_to
@reply_to.setter
def reply_to(self, value: str):
self.core.reply_to = value
@property
def current_prompt_mode(self) -> str:
return self.core.prompt_mode
@current_prompt_mode.setter
def current_prompt_mode(self, value: str):
self.core.prompt_mode = value
def validate(self) -> List[str]:
"""参数验证"""
errors = []
if not isinstance(self.chat_id, str):
if not isinstance(self.core.chat_id, str):
errors.append("chat_id必须是字符串类型")
if not isinstance(self.reply_to, str):
if not isinstance(self.core.reply_to, str):
errors.append("reply_to必须是字符串类型")
return errors
return errors + self.features.validate() + self.content.validate()
@dataclass
@@ -98,38 +98,123 @@ class ChatContext:
class SmartPromptBuilder:
"""重构的智能提示词构建器 - 使用原有DefaultReplyer逻辑"""
"""重构的智能提示词构建器 - 完全继承DefaultReplyer功能"""
def __init__(self):
self._cache: Dict[str, Dict[str, Any]] = {}
# 使用共享缓存
from src.chat.utils.prompt_utils import PromptUtils
async def build_context_data(self, params: SmartPromptParameters) -> Dict[str, Any]:
"""并行构建完整的上下文数据"""
"""并行构建完整的上下文数据 - 使用共享缓存和优化后的参数结构"""
# 从缓存检查
cache_key = self._get_cache_key(params)
if params.enable_cache and cache_key in self._cache:
cached = self._cache[cache_key]
if time.time() - cached.get('timestamp', 0) < params.cache_ttl:
return cached['data'].copy()
# 使用共享缓存
from src.chat.utils.prompt_utils import PromptUtils
cache_key = PromptUtils.get_cache_key(
params.core.chat_id,
params.core.prompt_mode,
params.core.reply_to
)
# 构建基础的数据字典
context_data = {}
cached = PromptUtils.get_from_cache(cache_key, params.cache_ttl if hasattr(params, 'cache_ttl') else 300)
if cached is not None:
logger.debug(f"使用缓存结果: {cache_key}")
return cached
# 1. 构建聊天历史 - 根据模式不同
# 并行执行所有构建任务
start_time = time.time()
timing_logs = {}
try:
# 准备构建任务
tasks = []
task_names = []
# 初始化预构建参数，使用新的结构
pre_built_params = {}
if params.content:
pre_built_params.update({
'expression_habits_block': params.content.expression_habits or "",
'relation_info': params.content.relation_info or "",
'memory_block': params.content.memory_block or "",
'tool_info': params.content.tool_info or "",
'knowledge_prompt': params.content.knowledge_info or "",
'cross_context_block': params.content.cross_context or "",
})
# 根据新的参数结构确定要构建的项
if params.features.enable_expression and not pre_built_params.get('expression_habits_block'):
tasks.append(self._timed_build(self._build_expression_habits, params, "expression_habits"))
task_names.append("expression_habits")
if params.features.enable_memory and not pre_built_params.get('memory_block'):
tasks.append(self._timed_build(self._build_memory_block, params, "memory_block"))
task_names.append("memory_block")
if params.features.enable_relation and not pre_built_params.get('relation_info'):
tasks.append(self._timed_build(self._build_relation_info, params, "relation_info"))
task_names.append("relation_info")
if params.features.enable_tool and not pre_built_params.get('tool_info'):
tasks.append(self._timed_build(self._build_tool_info, params, "tool_info"))
task_names.append("tool_info")
if params.features.enable_knowledge and not pre_built_params.get('knowledge_prompt'):
tasks.append(self._timed_build(self._build_knowledge_info, params, "knowledge_info"))
task_names.append("knowledge_info")
if params.features.enable_cross_context and not pre_built_params.get('cross_context_block'):
tasks.append(self._timed_build(self._build_cross_context, params, "cross_context"))
task_names.append("cross_context")
# 并行执行所有任务，设置更合理的超时
timeout_seconds = max(10.0, params.max_context_messages * 0.3) # 最少10秒超时
results = await asyncio.wait_for(
asyncio.gather(*tasks, return_exceptions=True),
timeout=timeout_seconds
)
# 处理结果并收集性能数据
context_data = {}
for i, result in enumerate(results):
task_name = task_names[i] if i < len(task_names) else f"task_{i}"
if isinstance(result, Exception):
logger.error(f"构建任务{task_name}失败: {str(result)}")
elif isinstance(result, tuple) and len(result) == 2:
# 结果格式: (data, timing)
data, timing = result
context_data.update(data)
timing_logs[task_name] = timing
# 记录耗时过长的任务
if timing > 8.0:
logger.warning(f"构建任务{task_name}耗时过长: {timing:.2f}s")
else:
# 直接数据结果
context_data.update(result)
# 添加预构建的参数
for key, value in pre_built_params.items():
if value:
context_data[key] = value
except asyncio.TimeoutError:
logger.error(f"构建超时 ({timeout_seconds}s)")
context_data = {}
# 添加预构建的参数，即使在超时情况下
for key, value in pre_built_params.items():
if value:
context_data[key] = value
# 构建聊天历史 - 根据模式不同
if params.current_prompt_mode == "s4u":
await self._build_s4u_chat_context(context_data, params)
else:
await self._build_normal_chat_context(context_data, params)
# 2. 集成各个构建模块
# 补充基础信息
context_data.update({
'expression_habits_block': params.expression_habits_block,
'memory_block': params.memory_block,
'relation_info_block': params.relation_info,
'tool_info_block': params.tool_info,
'knowledge_prompt': params.prompt_info,
'cross_context_block': params.cross_context_block,
'keywords_reaction_prompt': params.keywords_reaction_prompt,
'extra_info_block': params.extra_info_block,
'time_block': params.time_block or f"当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
@@ -148,97 +233,71 @@ class SmartPromptBuilder:
'timestamp': time.time()
}
total_time = time.time() - start_time
if timing_logs:
timing_str = "; ".join([f"{name}: {time:.2f}s" for name, time in timing_logs.items()])
logger.info(f"构建任务耗时: {timing_str}")
logger.debug(f"构建完成，总耗时: {total_time:.2f}s")
return context_data
def _get_cache_key(self, params: SmartPromptParameters) -> str:
"""生成缓存键"""
return f"{params.chat_id}_{params.current_prompt_mode}_{hash(params.reply_to)}"
async def _timed_build(self, build_func, params: SmartPromptParameters, task_name: str) -> Tuple[Dict[str, Any], float]:
"""带计时的构建函数"""
start_time = time.time()
try:
result = await build_func(params)
end_time = time.time()
return result, end_time - start_time
except Exception as e:
logger.error(f"构建任务{task_name}异常: {e}")
end_time = time.time()
return {}, end_time - start_time
async def _build_s4u_chat_context(self, context_data: Dict[str, Any], params: SmartPromptParameters) -> None:
"""构建S4U模式的聊天上下文"""
if not params.message_list_before_now_long:
"""构建S4U模式的聊天上下文 - 使用新参数结构"""
if not params.core.message_list:
return
# 使用原有的分离逻辑
core_dialogue, background_dialogue = self._build_s4u_separated_history(
params.message_list_before_now_long,
params.target_user_info
# 使用共享工具构建分离历史
from src.chat.utils.prompt_utils import PromptUtils
core_dialogue, background_dialogue = PromptUtils.build_s4u_separated_history(
params.core.message_list,
params.core.target_user_info,
params.core.target_chat
)
context_data['core_dialogue_prompt'] = core_dialogue
context_data['background_dialogue_prompt'] = background_dialogue
async def _build_normal_chat_context(self, context_data: Dict[str, Any], params: SmartPromptParameters) -> None:
"""构建normal模式的聊天上下文"""
if not params.chat_talking_prompt_short:
"""构建normal模式的聊天上下文 - 使用新参数结构"""
if not params.core.chat_context:
return
context_data['chat_info'] = f"""群里的聊天内容：
{params.chat_talking_prompt_short}"""
{params.core.chat_context}"""
def _build_s4u_separated_history(
self,
message_list_before_now: List[Dict[str, Any]],
target_user_info: Optional[Dict[str, Any]]
) -> Tuple[str, str]:
"""复制原有的分离对话逻辑"""
core_dialogue_list = []
background_dialogue_list = []
bot_id = str(global_config.bot.qq_account)
# 获取目标用户ID
target_user_id = ""
if target_user_info:
target_user_id = str(target_user_info.get("user_id", ""))
# 过滤消息：分离bot和目标用户的对话 vs 其他用户的对话
for msg_dict in message_list_before_now:
try:
msg_user_id = str(msg_dict.get("user_id", ""))
reply_to = msg_dict.get("reply_to", "")
reply_to_user_id = self._parse_reply_target_id(reply_to)
if (msg_user_id == bot_id and reply_to_user_id == target_user_id) or msg_user_id == target_user_id:
core_dialogue_list.append(msg_dict)
else:
background_dialogue_list.append(msg_dict)
except Exception as e:
logger.error(f"处理消息记录时出错: {msg_dict}, 错误: {e}")
# 构建背景对话
background_dialogue_prompt = ""
if background_dialogue_list:
latest_25_msgs = background_dialogue_list[-int(global_config.chat.max_context_size * 0.5):]
background_dialogue_prompt_str = build_readable_messages(
latest_25_msgs,
replace_bot_name=True,
timestamp_mode="normal",
truncate=True,
)
background_dialogue_prompt = f"这是其他用户的发言：\n{background_dialogue_prompt_str}"
# 构建核心对话
core_dialogue_prompt = ""
if core_dialogue_list:
core_dialogue_list = core_dialogue_list[-int(global_config.chat.max_context_size * 2):]
core_dialogue_prompt_str = build_readable_messages(
core_dialogue_list,
replace_bot_name=True,
merge_messages=False,
timestamp_mode="normal",
read_mark=0.0,
truncate=True,
show_actions=True,
)
core_dialogue_prompt = core_dialogue_prompt_str
return core_dialogue_prompt, background_dialogue_prompt
def _build_s4u_separated_history(self, *args, **kwargs):
"""已废弃 - 使用PromptUtils中的实现"""
logger.warning("_build_s4u_separated_history已废弃，使用PromptUtils.build_s4u_separated_history")
return "", ""
def _parse_reply_target_id(self, reply_to: str) -> str:
"""解析回复目标中的用户ID"""
if not reply_to:
return ""
return "" # 简化实现，实际需要从reply_to中提取
# 复用_parse_reply_target方法的逻辑
sender, _ = self._parse_reply_target(reply_to)
if not sender:
return ""
# 获取用户ID
person_info_manager = get_person_info_manager()
person_id = person_info_manager.get_person_id_by_person_name(sender)
if person_id:
user_id = person_info_manager.get_value_sync(person_id, "user_id")
return str(user_id) if user_id else ""
@property
def _cached_data(self) -> dict:
@@ -246,10 +305,184 @@ class SmartPromptBuilder:
if not hasattr(self, '_cache_store'):
self._cache_store = {}
return self._cache_store
async def _build_expression_habits(self, params: SmartPromptParameters) -> Dict[str, Any]:
"""构建表达习惯 - 使用共享工具类"""
try:
from src.chat.utils.prompt_utils import PromptUtils
return await PromptUtils.build_expression_habits(
params.core.chat_id,
params.core.chat_context,
params.core.target
)
except Exception as e:
logger.error(f"构建表达习惯失败: {e}")
return {"expression_habits_block": ""}
async def _build_memory_block(self, params: SmartPromptParameters) -> Dict[str, Any]:
"""构建记忆块 - 使用共享工具类"""
try:
from src.chat.utils.prompt_utils import PromptUtils
return await PromptUtils.build_memory_block(
params.core.chat_id,
params.core.target,
params.core.chat_context,
params.features.enable_memory # 传入功能开关
)
if global_config.memory.enable_instant_memory:
# 使用异步记忆包装器（最优化的非阻塞模式）
try:
from src.chat.memory_system.async_instant_memory_wrapper import get_async_instant_memory
# 获取异步记忆包装器
async_memory = get_async_instant_memory(params.chat_id)
# 后台存储聊天历史（完全非阻塞）
async_memory.store_memory_background(params.chat_talking_prompt_short)
# 快速检索记忆，最大超时2秒
instant_memory = await async_memory.get_memory_with_fallback(params.target, max_timeout=2.0)
logger.info(f"异步瞬时记忆：{instant_memory}")
except ImportError:
# 如果异步包装器不可用，尝试使用异步记忆管理器
try:
from src.chat.memory_system.async_memory_optimizer import (
retrieve_memory_nonblocking,
store_memory_nonblocking,
)
# 异步存储聊天历史（非阻塞）
asyncio.create_task(
store_memory_nonblocking(chat_id=params.chat_id, content=params.chat_talking_prompt_short)
)
# 尝试从缓存获取瞬时记忆
instant_memory = await retrieve_memory_nonblocking(chat_id=params.chat_id, query=params.target)
# 如果没有缓存结果，快速检索一次
if instant_memory is None:
try:
instant_memory = await asyncio.wait_for(
instant_memory_system.get_memory_for_context(params.target), timeout=1.5
)
except asyncio.TimeoutError:
logger.warning("瞬时记忆检索超时，使用空结果")
instant_memory = ""
logger.info(f"向量瞬时记忆：{instant_memory}")
except ImportError:
# 最后的fallback：使用原有逻辑但加上超时控制
logger.warning("异步记忆系统不可用，使用带超时的同步方式")
# 异步存储聊天历史
asyncio.create_task(instant_memory_system.store_message(params.chat_talking_prompt_short))
# 带超时的记忆检索
try:
instant_memory = await asyncio.wait_for(
instant_memory_system.get_memory_for_context(params.target),
timeout=1.0, # 最保守的1秒超时
)
except asyncio.TimeoutError:
logger.warning("瞬时记忆检索超时，跳过记忆获取")
instant_memory = ""
except Exception as e:
logger.error(f"瞬时记忆检索失败: {e}")
instant_memory = ""
logger.info(f"同步瞬时记忆：{instant_memory}")
except Exception as e:
logger.error(f"瞬时记忆系统异常: {e}")
instant_memory = ""
# 构建记忆字符串，即使某种记忆为空也要继续
memory_str = ""
has_any_memory = False
# 添加长期记忆
if running_memories:
if not memory_str:
memory_str = "以下是当前在聊天中，你回忆起的记忆：\n"
for running_memory in running_memories:
memory_str += f"- {running_memory['content']}\n"
has_any_memory = True
# 添加瞬时记忆
if instant_memory:
if not memory_str:
memory_str = "以下是当前在聊天中，你回忆起的记忆：\n"
memory_str += f"- {instant_memory}\n"
has_any_memory = True
# 只有当完全没有任何记忆时才返回空字符串
return {"memory_block": memory_str if has_any_memory else ""}
except Exception as e:
logger.error(f"构建记忆块失败: {e}")
return {"memory_block": ""}
async def _build_relation_info(self, params: SmartPromptParameters) -> Dict[str, Any]:
"""构建关系信息 - 使用共享工具类"""
try:
from src.chat.utils.prompt_utils import PromptUtils
return await PromptUtils.build_relation_info(
params.core.chat_id,
params.core.reply_to
)
except Exception as e:
logger.error(f"构建关系信息失败: {e}")
return {"relation_info_block": ""}
async def _build_tool_info(self, params: SmartPromptParameters) -> Dict[str, Any]:
"""构建工具信息 - 使用共享工具类"""
try:
from src.chat.utils.prompt_utils import PromptUtils
return await PromptUtils.build_tool_info(
params.core.chat_id,
params.core.reply_to,
params.core.chat_context
)
except Exception as e:
logger.error(f"工具信息获取失败: {e}")
return {"tool_info_block": ""}
async def _build_knowledge_info(self, params: SmartPromptParameters) -> Dict[str, Any]:
"""构建知识信息 - 使用共享工具类"""
try:
from src.chat.utils.prompt_utils import PromptUtils
return await PromptUtils.build_knowledge_info(
params.core.reply_to,
params.core.chat_context
)
except Exception as e:
logger.error(f"获取知识库内容时发生异常: {str(e)}")
return {"knowledge_prompt": ""}
async def _build_cross_context(self, params: SmartPromptParameters) -> Dict[str, Any]:
"""构建跨群上下文 - 使用共享工具类"""
try:
from src.chat.utils.prompt_utils import PromptUtils
return await PromptUtils.build_cross_context(
params.core.chat_id,
params.core.prompt_mode,
params.core.target_user_info
)
except Exception as e:
logger.error(f"构建跨群上下文失败: {e}")
return {"cross_context_block": ""}
def _parse_reply_target(self, target_message: str) -> Tuple[str, str]:
"""解析回复目标消息 - 使用共享工具类"""
return PromptUtils.parse_reply_target(target_message)
class SmartPrompt:
"""重构的智能提示词核心类"""
"""重构的智能提示词核心类 - 使用新参数结构"""
def __init__(
self,
@@ -262,20 +495,21 @@ class SmartPrompt:
def _get_default_template(self) -> str:
"""根据模式选择默认模板"""
if self.parameters.current_prompt_mode == "s4u":
if self.parameters.core.prompt_mode == "s4u":
return "s4u_style_prompt"
elif self.parameters.current_prompt_mode == "normal":
elif self.parameters.core.prompt_mode == "normal":
return "normal_style_prompt"
else:
return "default_expressor_prompt"
async def build_prompt(self) -> str:
"""构建最终的Prompt文本"""
"""构建最终的Prompt文本 - 使用新参数结构"""
# 参数验证
errors = self.parameters.validate()
if errors:
raise ValueError(f"参数验证失败: {', '.join(errors)}")
start_time = time.time()
try:
# 构建基础上下文的完整映射
context_data = await self.builder.build_context_data(self.parameters)
@@ -284,20 +518,28 @@ class SmartPrompt:
template = await global_prompt_manager.get_prompt_async(self.template_name)
# 根据模式传递不同的参数
if self.parameters.current_prompt_mode == "s4u":
return await self._build_s4u_prompt(template, context_data)
elif self.parameters.current_prompt_mode == "normal":
return await self._build_normal_prompt(template, context_data)
if self.parameters.core.prompt_mode == "s4u":
result = await self._build_s4u_prompt(template, context_data)
elif self.parameters.core.prompt_mode == "normal":
result = await self._build_normal_prompt(template, context_data)
else:
return await self._build_default_prompt(template, context_data)
result = await self._build_default_prompt(template, context_data)
# 记录性能数据
total_time = time.time() - start_time
logger.debug(f"SmartPrompt构建完成，模式: {self.parameters.core.prompt_mode}, 耗时: {total_time:.2f}s")
return result
except Exception as e:
logger.error(f"构建Prompt失败: {e}")
# 返回一个基础Prompt
return f"用户说：{self.parameters.reply_to}。请回复。"
# 返回一个基础Prompt作为fallback
fallback_prompt = f"用户说：{self.parameters.core.reply_to}。请回复。"
logger.warning(f"使用fallback prompt: {fallback_prompt}")
return fallback_prompt
async def _build_s4u_prompt(self, template: Prompt, context_data: Dict[str, Any]) -> str:
"""构建S4U模式的完整Prompt"""
"""构建S4U模式的完整Prompt - 使用新参数结构"""
params = {
**context_data,
'expression_habits_block': context_data.get('expression_habits_block', ''),
@@ -305,24 +547,24 @@ class SmartPrompt:
'knowledge_prompt': context_data.get('knowledge_prompt', ''),
'memory_block': context_data.get('memory_block', ''),
'relation_info_block': context_data.get('relation_info_block', ''),
'extra_info_block': context_data.get('extra_info_block', ''),
'extra_info_block': self.parameters.content.extra_info or context_data.get('extra_info_block', ''),
'cross_context_block': context_data.get('cross_context_block', ''),
'identity': context_data.get('identity', ''),
'action_descriptions': context_data.get('action_descriptions', ''),
'sender_name': self.parameters.sender,
'mood_state': context_data.get('mood_state', ''),
'identity': self.parameters.content.identity or context_data.get('identity', ''),
'action_descriptions': self.parameters.content.actions or context_data.get('action_descriptions', ''),
'sender_name': self.parameters.core.sender_name,
'mood_state': self.parameters.content.mood_prompt or context_data.get('mood_state', ''),
'background_dialogue_prompt': context_data.get('background_dialogue_prompt', ''),
'time_block': context_data.get('time_block', ''),
'core_dialogue_prompt': context_data.get('core_dialogue_prompt', ''),
'reply_target_block': context_data.get('reply_target_block', ''),
'reply_style': global_config.personality.reply_style,
'keywords_reaction_prompt': context_data.get('keywords_reaction_prompt', ''),
'moderation_prompt': context_data.get('moderation_prompt', ''),
'keywords_reaction_prompt': self.parameters.content.keywords_reaction or context_data.get('keywords_reaction_prompt', ''),
'moderation_prompt': self.parameters.content.moderation_prompt or context_data.get('moderation_prompt', ''),
}
return await global_prompt_manager.format_prompt(self.template_name, **params)
async def _build_normal_prompt(self, template: Prompt, context_data: Dict[str, Any]) -> str:
"""构建Normal模式的完整Prompt"""
"""构建Normal模式的完整Prompt - 使用新参数结构"""
params = {
**context_data,
'expression_habits_block': context_data.get('expression_habits_block', ''),
@@ -330,54 +572,240 @@ class SmartPrompt:
'knowledge_prompt': context_data.get('knowledge_prompt', ''),
'memory_block': context_data.get('memory_block', ''),
'relation_info_block': context_data.get('relation_info_block', ''),
'extra_info_block': context_data.get('extra_info_block', ''),
'extra_info_block': self.parameters.content.extra_info or context_data.get('extra_info_block', ''),
'cross_context_block': context_data.get('cross_context_block', ''),
'identity': context_data.get('identity', ''),
'action_descriptions': context_data.get('action_descriptions', ''),
'schedule_block': context_data.get('schedule_block', ''),
'identity': self.parameters.content.identity or context_data.get('identity', ''),
'action_descriptions': self.parameters.content.actions or context_data.get('action_descriptions', ''),
'schedule_block': self.parameters.content.schedule_prompt or context_data.get('schedule_block', ''),
'time_block': context_data.get('time_block', ''),
'chat_info': context_data.get('chat_info', ''),
'reply_target_block': context_data.get('reply_target_block', ''),
'config_expression_style': global_config.personality.reply_style,
'mood_state': context_data.get('mood_state', ''),
'keywords_reaction_prompt': context_data.get('keywords_reaction_prompt', ''),
'moderation_prompt': context_data.get('moderation_prompt', ''),
'mood_state': self.parameters.content.mood_prompt or context_data.get('mood_state', ''),
'keywords_reaction_prompt': self.parameters.content.keywords_reaction or context_data.get('keywords_reaction_prompt', ''),
'moderation_prompt': self.parameters.content.moderation_prompt or context_data.get('moderation_prompt', ''),
}
return await global_prompt_manager.format_prompt(self.template_name, **params)
async def _build_default_prompt(self, template: Prompt, context_data: Dict[str, Any]) -> str:
"""构建默认模式的Prompt"""
"""构建默认模式的Prompt - 使用新参数结构"""
params = {
'expression_habits_block': context_data.get('expression_habits_block', ''),
'relation_info_block': context_data.get('relation_info_block', ''),
'chat_target': "",
'time_block': context_data.get('time_block', ''),
'chat_info': context_data.get('chat_info', ''),
'identity': context_data.get('identity', ''),
'identity': self.parameters.content.identity or context_data.get('identity', ''),
'chat_target_2': "",
'reply_target_block': context_data.get('reply_target_block', ''),
'raw_reply': self.parameters.target,
'raw_reply': self.parameters.core.target_message,
'reason': "",
'mood_state': context_data.get('mood_state', ''),
'mood_state': self.parameters.content.mood_prompt or context_data.get('mood_state', ''),
'reply_style': global_config.personality.reply_style,
'keywords_reaction_prompt': context_data.get('keywords_reaction_prompt', ''),
'moderation_prompt': context_data.get('moderation_prompt', ''),
'keywords_reaction_prompt': self.parameters.content.keywords_reaction or context_data.get('keywords_reaction_prompt', ''),
'moderation_prompt': self.parameters.content.moderation_prompt or context_data.get('moderation_prompt', ''),
}
return await global_prompt_manager.format_prompt(self.template_name, **params)
# 工厂函数 - 简化创建
# 工厂函数 - 简化创建 - 更新参数结构
def create_smart_prompt(
chat_id: str = "",
sender_name: str = "",
target_message: str = "",
reply_to: str = "",
extra_info: str = "",
**kwargs
) -> SmartPrompt:
"""快速创建智能Prompt实例的工厂函数"""
"""快速创建智能Prompt实例的工厂函数 - 使用新参数结构"""
# 使用新的参数结构
from src.chat.utils.prompt_parameters import PromptCoreParams
core_params = PromptCoreParams(
chat_id=chat_id,
sender_name=sender_name,
target_message=target_message,
reply_to=reply_to
)
# 更新features和content参数
feature_params = kwargs.pop('features', None) or PromptFeatureParams()
content_params = kwargs.pop('content', None) or PromptContentParams()
parameters = SmartPromptParameters(
reply_to=reply_to,
extra_info=extra_info,
core=core_params,
features=feature_params,
content=content_params,
**kwargs
)
return SmartPrompt(parameters=parameters)
return SmartPrompt(parameters=parameters)
class SmartPromptHealthChecker:
"""SmartPrompt健康检查器"""
@staticmethod
async def check_system_health() -> Dict[str, Any]:
"""检查系统健康状态"""
health_status = {
"status": "healthy",
"components": {},
"issues": []
}
try:
# 检查关键模块导入
try:
from src.chat.express.expression_selector import expression_selector
health_status["components"]["expression_selector"] = "ok"
except ImportError as e:
health_status["components"]["expression_selector"] = f"failed: {str(e)}"
health_status["issues"].append("expression_selector导入失败")
health_status["status"] = "degraded"
try:
from src.chat.memory_system.memory_activator import MemoryActivator
health_status["components"]["memory_activator"] = "ok"
except ImportError as e:
health_status["components"]["memory_activator"] = f"failed: {str(e)}"
health_status["issues"].append("memory_activator导入失败")
health_status["status"] = "degraded"
try:
from src.plugin_system.core.tool_use import ToolExecutor
health_status["components"]["tool_executor"] = "ok"
except ImportError as e:
health_status["components"]["tool_executor"] = f"failed: {str(e)}"
health_status["issues"].append("tool_executor导入失败")
health_status["status"] = "degraded"
try:
from src.plugins.built_in.knowledge.lpmm_get_knowledge import SearchKnowledgeFromLPMMTool
health_status["components"]["knowledge_tool"] = "ok"
except ImportError as e:
health_status["components"]["knowledge_tool"] = f"failed: {str(e)}"
health_status["issues"].append("knowledge_tool导入失败")
# 知识工具不是必需的，所以不降低整体状态
# 检查配置
try:
from src.config.config import global_config
health_status["components"]["config"] = "ok"
# 检查关键配置项
if not hasattr(global_config, 'personality') or not hasattr(global_config.personality, 'prompt_mode'):
health_status["issues"].append("缺少personality.prompt_mode配置")
health_status["status"] = "degraded"
if not hasattr(global_config, 'memory') or not hasattr(global_config.memory, 'enable_memory'):
health_status["issues"].append("缺少memory.enable_memory配置")
except Exception as e:
health_status["components"]["config"] = f"failed: {str(e)}"
health_status["issues"].append("配置加载失败")
health_status["status"] = "unhealthy"
# 检查Prompt模板
try:
required_templates = ["s4u_style_prompt", "normal_style_prompt", "default_expressor_prompt"]
for template_name in required_templates:
try:
await global_prompt_manager.get_prompt_async(template_name)
health_status["components"][f"template_{template_name}"] = "ok"
except Exception as e:
health_status["components"][f"template_{template_name}"] = f"failed: {str(e)}"
health_status["issues"].append(f"模板{template_name}加载失败")
health_status["status"] = "degraded"
except Exception as e:
health_status["components"]["prompt_templates"] = f"failed: {str(e)}"
health_status["issues"].append("Prompt模板检查失败")
health_status["status"] = "unhealthy"
return health_status
except Exception as e:
return {
"status": "unhealthy",
"components": {},
"issues": [f"健康检查异常: {str(e)}"]
}
@staticmethod
async def run_performance_test() -> Dict[str, Any]:
"""运行性能测试"""
test_results = {
"status": "completed",
"tests": {},
"summary": {}
}
try:
# 创建测试参数
test_params = SmartPromptParameters(
chat_id="test_chat",
sender="test_user",
target="test_message",
reply_to="test_user:test_message",
current_prompt_mode="s4u",
enable_cache=False # 禁用缓存以测试真实性能
)
# 测试不同模式下的构建性能
modes = ["s4u", "normal", "minimal"]
for mode in modes:
test_params.current_prompt_mode = mode
smart_prompt = SmartPrompt(parameters=test_params)
# 运行多次测试取平均值
times = []
for _ in range(3):
start_time = time.time()
try:
await smart_prompt.build_prompt()
end_time = time.time()
times.append(end_time - start_time)
except Exception as e:
times.append(float('inf'))
logger.error(f"性能测试失败 (模式: {mode}): {e}")
# 计算统计信息
valid_times = [t for t in times if t != float('inf')]
if valid_times:
avg_time = sum(valid_times) / len(valid_times)
min_time = min(valid_times)
max_time = max(valid_times)
test_results["tests"][mode] = {
"avg_time": avg_time,
"min_time": min_time,
"max_time": max_time,
"success_rate": len(valid_times) / len(times)
}
else:
test_results["tests"][mode] = {
"avg_time": float('inf'),
"min_time": float('inf'),
"max_time": float('inf'),
"success_rate": 0
}
# 计算总体统计
all_avg_times = [test["avg_time"] for test in test_results["tests"].values() if test["avg_time"] != float('inf')]
if all_avg_times:
test_results["summary"] = {
"overall_avg_time": sum(all_avg_times) / len(all_avg_times),
"fastest_mode": min(test_results["tests"].items(), key=lambda x: x[1]["avg_time"])[0],
"slowest_mode": max(test_results["tests"].items(), key=lambda x: x[1]["avg_time"])[0]
}
return test_results
except Exception as e:
return {
"status": "failed",
"tests": {},
"summary": {},
"error": str(e)
}