933 lines
43 KiB
Python
933 lines
43 KiB
Python
import json
|
||
import time
|
||
import traceback
|
||
import random
|
||
import re
|
||
import contextlib
|
||
from typing import Dict, Optional, Tuple, List, TYPE_CHECKING, Union
|
||
from collections import OrderedDict
|
||
from rich.traceback import install
|
||
from datetime import datetime
|
||
from json_repair import repair_json
|
||
from src.llm_models.utils_model import LLMRequest
|
||
from src.config.config import global_config, model_config
|
||
from src.common.logger import get_logger
|
||
from src.chat.logger.plan_reply_logger import PlanReplyLogger
|
||
from src.common.data_models.info_data_model import ActionPlannerInfo
|
||
from src.prompt.prompt_manager import prompt_manager
|
||
from src.services.message_service import (
|
||
build_readable_messages_with_id,
|
||
replace_user_references,
|
||
get_messages_before_time_in_chat,
|
||
translate_pid_to_description,
|
||
)
|
||
from src.chat.utils.utils import get_chat_type_and_target_info, is_bot_self
|
||
from src.chat.planner_actions.action_manager import ActionManager
|
||
from src.chat.message_receive.chat_manager import chat_manager as _chat_manager
|
||
from src.core.types import ActionActivationType, ActionInfo, ComponentType
|
||
from src.core.component_registry import component_registry
|
||
from src.person_info.person_info import Person
|
||
|
||
if TYPE_CHECKING:
|
||
from src.common.data_models.info_data_model import TargetPersonInfo
|
||
from src.common.data_models.database_data_model import DatabaseMessages
|
||
|
||
logger = get_logger("planner")
|
||
|
||
install(extra_lines=3)
|
||
|
||
|
||
class ActionPlanner:
|
||
def __init__(self, chat_id: str, action_manager: ActionManager):
|
||
self.chat_id = chat_id
|
||
self.log_prefix = f"[{_chat_manager.get_session_name(chat_id) or chat_id}]"
|
||
self.action_manager = action_manager
|
||
# LLM规划器配置
|
||
self.planner_llm = LLMRequest(
|
||
model_set=model_config.model_task_config.planner, request_type="planner"
|
||
) # 用于动作规划
|
||
|
||
self.last_obs_time_mark = 0.0
|
||
|
||
self.plan_log: List[Tuple[str, float, Union[List[ActionPlannerInfo], str]]] = []
|
||
|
||
# 黑话缓存：使用 OrderedDict 实现 LRU，最多缓存10个
|
||
self.unknown_words_cache: OrderedDict[str, None] = OrderedDict()
|
||
self.unknown_words_cache_limit = 10
|
||
|
||
def find_message_by_id(
|
||
self, message_id: str, message_id_list: List[Tuple[str, "DatabaseMessages"]]
|
||
) -> Optional["DatabaseMessages"]:
|
||
# sourcery skip: use-next
|
||
"""
|
||
根据message_id从message_id_list中查找对应的原始消息
|
||
|
||
Args:
|
||
message_id: 要查找的消息ID
|
||
message_id_list: 消息ID列表，格式为[{'id': str, 'message': dict}, ...]
|
||
|
||
Returns:
|
||
找到的原始消息字典，如果未找到则返回None
|
||
"""
|
||
for item in message_id_list:
|
||
if item[0] == message_id:
|
||
return item[1]
|
||
return None
|
||
|
||
def _replace_message_ids_with_text(
|
||
self, text: Optional[str], message_id_list: List[Tuple[str, "DatabaseMessages"]]
|
||
) -> Optional[str]:
|
||
"""将文本中的 m+数字 消息ID替换为原消息内容，并添加双引号"""
|
||
if not text:
|
||
return text
|
||
|
||
id_to_message = dict(message_id_list)
|
||
|
||
# 匹配m后带2-4位数字，前后不是字母数字下划线
|
||
pattern = r"(?<![A-Za-z0-9_])m\d{2,4}(?![A-Za-z0-9_])"
|
||
|
||
matches = re.findall(pattern, text)
|
||
if matches:
|
||
available_ids = set(id_to_message.keys())
|
||
found_ids = set(matches)
|
||
missing_ids = found_ids - available_ids
|
||
if missing_ids:
|
||
logger.info(
|
||
f"{self.log_prefix}planner理由中引用的消息ID不在当前上下文中: {missing_ids}, 可用ID: {list(available_ids)[:10]}..."
|
||
)
|
||
logger.info(
|
||
f"{self.log_prefix}planner理由替换: 找到{len(matches)}个消息ID引用，其中{len(found_ids & available_ids)}个在上下文中"
|
||
)
|
||
|
||
def _replace(match: re.Match[str]) -> str:
|
||
msg_id = match.group(0)
|
||
message = id_to_message.get(msg_id)
|
||
if not message:
|
||
logger.warning(f"{self.log_prefix}planner理由引用 {msg_id} 未找到对应消息，保持原样")
|
||
return msg_id
|
||
|
||
msg_text = (message.processed_plain_text or "").strip()
|
||
if not msg_text:
|
||
logger.warning(f"{self.log_prefix}planner理由引用 {msg_id} 的消息内容为空，保持原样")
|
||
return msg_id
|
||
|
||
# 替换 [picid:xxx] 为 [图片：描述]
|
||
pic_pattern = r"\[picid:([^\]]+)\]"
|
||
|
||
def replace_pic_id(pic_match: re.Match) -> str:
|
||
pic_id = pic_match.group(1)
|
||
description = translate_pid_to_description(pic_id)
|
||
return f"[图片：{description}]"
|
||
|
||
msg_text = re.sub(pic_pattern, replace_pic_id, msg_text)
|
||
|
||
# 替换用户引用格式：回复<aaa:bbb> 和 @<aaa:bbb>
|
||
platform = (
|
||
getattr(message, "user_info", None)
|
||
and message.user_info.platform
|
||
or getattr(message, "chat_info", None)
|
||
and message.chat_info.platform
|
||
or "qq"
|
||
)
|
||
msg_text = replace_user_references(msg_text, platform, replace_bot_name=True)
|
||
|
||
# 替换单独的 <用户名:用户ID> 格式（replace_user_references 已处理回复<和@<格式）
|
||
# 匹配所有 <aaa:bbb> 格式，由于 replace_user_references 已经替换了回复<和@<格式，
|
||
# 这里匹配到的应该都是单独的格式
|
||
user_ref_pattern = r"<([^:<>]+):([^:<>]+)>"
|
||
|
||
def replace_user_ref(user_match: re.Match) -> str:
|
||
user_name = user_match.group(1)
|
||
user_id = user_match.group(2)
|
||
try:
|
||
# 检查是否是机器人自己
|
||
if user_id == global_config.bot.qq_account:
|
||
return f"{global_config.bot.nickname}(你)"
|
||
person = Person(platform=platform, user_id=user_id)
|
||
return person.person_name or user_name
|
||
except Exception:
|
||
# 如果解析失败，使用原始昵称
|
||
return user_name
|
||
|
||
msg_text = re.sub(user_ref_pattern, replace_user_ref, msg_text)
|
||
|
||
preview = msg_text if len(msg_text) <= 100 else f"{msg_text[:97]}..."
|
||
logger.info(f"{self.log_prefix}planner理由引用 {msg_id} -> 消息（{preview}）")
|
||
return f"消息（{msg_text}）"
|
||
|
||
return re.sub(pattern, _replace, text)
|
||
|
||
def _parse_single_action(
|
||
self,
|
||
action_json: dict,
|
||
message_id_list: List[Tuple[str, "DatabaseMessages"]],
|
||
current_available_actions: List[Tuple[str, ActionInfo]],
|
||
extracted_reasoning: str = "",
|
||
) -> List[ActionPlannerInfo]:
|
||
"""解析单个action JSON并返回ActionPlannerInfo列表"""
|
||
action_planner_infos = []
|
||
|
||
try:
|
||
action = action_json.get("action", "no_reply")
|
||
# 使用 extracted_reasoning（整体推理文本）作为 reasoning
|
||
if extracted_reasoning:
|
||
reasoning = self._replace_message_ids_with_text(extracted_reasoning, message_id_list)
|
||
if reasoning is None:
|
||
reasoning = extracted_reasoning
|
||
else:
|
||
reasoning = "未提供原因"
|
||
action_data = {key: value for key, value in action_json.items() if key not in ["action"]}
|
||
|
||
# 非no_reply动作需要target_message_id
|
||
target_message = None
|
||
|
||
target_message_id = action_json.get("target_message_id")
|
||
if target_message_id:
|
||
# 根据target_message_id查找原始消息
|
||
target_message = self.find_message_by_id(target_message_id, message_id_list)
|
||
if target_message is None:
|
||
logger.warning(f"{self.log_prefix}无法找到target_message_id '{target_message_id}' 对应的消息")
|
||
# 选择最新消息作为target_message
|
||
target_message = message_id_list[-1][1]
|
||
else:
|
||
target_message = message_id_list[-1][1]
|
||
logger.debug(f"{self.log_prefix}动作'{action}'缺少target_message_id，使用最新消息作为target_message")
|
||
|
||
if action != "no_reply" and target_message is not None and self._is_message_from_self(target_message):
|
||
logger.info(
|
||
f"{self.log_prefix}Planner选择了自己的消息 {target_message_id or target_message.message_id} 作为目标，强制使用 no_reply"
|
||
)
|
||
reasoning = f"目标消息 {target_message_id or target_message.message_id} 来自机器人自身，违反不回复自身消息规则。原始理由: {reasoning}"
|
||
action = "no_reply"
|
||
target_message = None
|
||
|
||
# 验证action是否可用
|
||
available_action_names = [action_name for action_name, _ in current_available_actions]
|
||
internal_action_names = ["no_reply", "reply", "wait_time"]
|
||
|
||
if action not in internal_action_names and action not in available_action_names:
|
||
logger.warning(
|
||
f"{self.log_prefix}LLM 返回了当前不可用或无效的动作: '{action}' (可用: {available_action_names})，将强制使用 'no_reply'"
|
||
)
|
||
reasoning = (
|
||
f"LLM 返回了当前不可用的动作 '{action}' (可用: {available_action_names})。原始理由: {reasoning}"
|
||
)
|
||
action = "no_reply"
|
||
|
||
# 创建ActionPlannerInfo对象
|
||
# 将列表转换为字典格式
|
||
available_actions_dict = dict(current_available_actions)
|
||
action_planner_infos.append(
|
||
ActionPlannerInfo(
|
||
action_type=action,
|
||
reasoning=reasoning,
|
||
action_data=action_data,
|
||
action_message=target_message,
|
||
available_actions=available_actions_dict,
|
||
action_reasoning=extracted_reasoning or None,
|
||
)
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"{self.log_prefix}解析单个action时出错: {e}")
|
||
# 将列表转换为字典格式
|
||
available_actions_dict = dict(current_available_actions)
|
||
action_planner_infos.append(
|
||
ActionPlannerInfo(
|
||
action_type="no_reply",
|
||
reasoning=f"解析单个action时出错: {e}",
|
||
action_data={},
|
||
action_message=None,
|
||
available_actions=available_actions_dict,
|
||
action_reasoning=extracted_reasoning or None,
|
||
)
|
||
)
|
||
|
||
return action_planner_infos
|
||
|
||
def _is_message_from_self(self, message: "DatabaseMessages") -> bool:
|
||
"""判断消息是否由机器人自身发送（支持多平台，包括 WebUI）"""
|
||
try:
|
||
return is_bot_self(message.user_info.platform or "", str(message.user_info.user_id))
|
||
except AttributeError:
|
||
logger.warning(f"{self.log_prefix}检测消息发送者失败，缺少必要字段")
|
||
return False
|
||
|
||
def _update_unknown_words_cache(self, new_words: List[str]) -> None:
|
||
"""
|
||
更新黑话缓存，将新的黑话加入缓存
|
||
|
||
Args:
|
||
new_words: 新提取的黑话列表
|
||
"""
|
||
for word in new_words:
|
||
if not isinstance(word, str):
|
||
continue
|
||
word = word.strip()
|
||
if not word:
|
||
continue
|
||
|
||
# 如果已存在，移到末尾（LRU）
|
||
if word in self.unknown_words_cache:
|
||
self.unknown_words_cache.move_to_end(word)
|
||
else:
|
||
# 添加新词
|
||
self.unknown_words_cache[word] = None
|
||
# 如果超过限制，移除最老的
|
||
if len(self.unknown_words_cache) > self.unknown_words_cache_limit:
|
||
self.unknown_words_cache.popitem(last=False)
|
||
logger.debug(f"{self.log_prefix}黑话缓存已满，移除最老的黑话")
|
||
|
||
def _merge_unknown_words_with_cache(self, new_words: Optional[List[str]]) -> List[str]:
|
||
"""
|
||
合并新提取的黑话和缓存中的黑话
|
||
|
||
Args:
|
||
new_words: 新提取的黑话列表（可能为None）
|
||
|
||
Returns:
|
||
合并后的黑话列表（去重）
|
||
"""
|
||
# 清理新提取的黑话
|
||
cleaned_new_words: List[str] = []
|
||
if new_words:
|
||
for word in new_words:
|
||
if isinstance(word, str):
|
||
if word := word.strip():
|
||
cleaned_new_words.append(word)
|
||
|
||
# 获取缓存中的黑话列表
|
||
cached_words = list(self.unknown_words_cache.keys())
|
||
|
||
# 合并并去重（保留顺序：新提取的在前，缓存的在后）
|
||
merged_words: List[str] = []
|
||
seen = set()
|
||
|
||
# 先添加新提取的
|
||
for word in cleaned_new_words:
|
||
if word not in seen:
|
||
merged_words.append(word)
|
||
seen.add(word)
|
||
|
||
# 再添加缓存的（如果不在新提取的列表中）
|
||
for word in cached_words:
|
||
if word not in seen:
|
||
merged_words.append(word)
|
||
seen.add(word)
|
||
|
||
return merged_words
|
||
|
||
def _process_unknown_words_cache(self, actions: List[ActionPlannerInfo]) -> None:
|
||
"""
|
||
处理黑话缓存逻辑：
|
||
1. 检查是否有 reply action 提取了 unknown_words
|
||
2. 如果没有提取，移除最老的1个
|
||
3. 如果缓存数量大于5，移除最老的2个
|
||
4. 对于每个 reply action，合并缓存和新提取的黑话
|
||
5. 更新缓存
|
||
|
||
Args:
|
||
actions: 解析后的动作列表
|
||
"""
|
||
# 先检查缓存数量，如果大于5，移除最老的2个
|
||
if len(self.unknown_words_cache) > 5:
|
||
# 移除最老的2个
|
||
removed_count = 0
|
||
for _ in range(2):
|
||
if len(self.unknown_words_cache) > 0:
|
||
self.unknown_words_cache.popitem(last=False)
|
||
removed_count += 1
|
||
if removed_count > 0:
|
||
logger.debug(f"{self.log_prefix}缓存数量大于5，移除最老的{removed_count}个缓存")
|
||
|
||
# 检查是否有 reply action 提取了 unknown_words
|
||
has_extracted_unknown_words = False
|
||
for action in actions:
|
||
if action.action_type == "reply":
|
||
action_data = action.action_data or {}
|
||
unknown_words = action_data.get("unknown_words")
|
||
if unknown_words and isinstance(unknown_words, list) and len(unknown_words) > 0:
|
||
has_extracted_unknown_words = True
|
||
break
|
||
|
||
# 如果当前 plan 的 reply 没有提取，移除最老的1个
|
||
if not has_extracted_unknown_words and len(self.unknown_words_cache) > 0:
|
||
self.unknown_words_cache.popitem(last=False)
|
||
logger.debug(f"{self.log_prefix}当前 plan 的 reply 没有提取黑话，移除最老的1个缓存")
|
||
|
||
# 对于每个 reply action，合并缓存和新提取的黑话
|
||
for action in actions:
|
||
if action.action_type == "reply":
|
||
action_data = action.action_data or {}
|
||
new_words = action_data.get("unknown_words")
|
||
|
||
# 合并新提取的和缓存的黑话列表
|
||
if merged_words := self._merge_unknown_words_with_cache(new_words):
|
||
action_data["unknown_words"] = merged_words
|
||
logger.debug(
|
||
f"{self.log_prefix}合并黑话：新提取 {len(new_words) if new_words else 0} 个，"
|
||
f"缓存 {len(self.unknown_words_cache)} 个，合并后 {len(merged_words)} 个"
|
||
)
|
||
else:
|
||
# 如果没有合并后的黑话，移除 unknown_words 字段
|
||
action_data.pop("unknown_words", None)
|
||
|
||
# 更新缓存（将新提取的黑话加入缓存）
|
||
if new_words:
|
||
self._update_unknown_words_cache(new_words)
|
||
|
||
async def plan(
|
||
self,
|
||
available_actions: Dict[str, ActionInfo],
|
||
loop_start_time: float = 0.0,
|
||
force_reply_message: Optional["DatabaseMessages"] = None,
|
||
) -> List[ActionPlannerInfo]:
|
||
# sourcery skip: use-named-expression
|
||
"""
|
||
规划器 (Planner): 使用LLM根据上下文决定做出什么动作。
|
||
"""
|
||
plan_start = time.perf_counter()
|
||
|
||
# 获取聊天上下文
|
||
message_list_before_now = get_messages_before_time_in_chat(
|
||
chat_id=self.chat_id,
|
||
timestamp=time.time(),
|
||
limit=int(global_config.chat.max_context_size * 0.6),
|
||
filter_intercept_message_level=1,
|
||
)
|
||
message_id_list: list[Tuple[str, "DatabaseMessages"]] = []
|
||
chat_content_block, message_id_list = build_readable_messages_with_id(
|
||
messages=message_list_before_now,
|
||
timestamp_mode="normal_no_YMD",
|
||
read_mark=self.last_obs_time_mark,
|
||
truncate=True,
|
||
show_actions=True,
|
||
)
|
||
|
||
message_list_before_now_short = message_list_before_now[-int(global_config.chat.max_context_size * 0.3) :]
|
||
chat_content_block_short, message_id_list_short = build_readable_messages_with_id(
|
||
messages=message_list_before_now_short,
|
||
timestamp_mode="normal_no_YMD",
|
||
truncate=False,
|
||
show_actions=False,
|
||
)
|
||
|
||
self.last_obs_time_mark = time.time()
|
||
|
||
# 获取必要信息
|
||
is_group_chat, chat_target_info, current_available_actions = self.get_necessary_info()
|
||
|
||
# 应用激活类型过滤
|
||
filtered_actions = self._filter_actions_by_activation_type(available_actions, chat_content_block_short)
|
||
|
||
logger.debug(f"{self.log_prefix}过滤后有{len(filtered_actions)}个可用动作")
|
||
|
||
prompt_build_start = time.perf_counter()
|
||
# 构建包含所有动作的提示词
|
||
prompt, message_id_list = await self.build_planner_prompt(
|
||
is_group_chat=is_group_chat,
|
||
chat_target_info=chat_target_info,
|
||
current_available_actions=filtered_actions,
|
||
chat_content_block=chat_content_block,
|
||
message_id_list=message_id_list,
|
||
)
|
||
prompt_build_ms = (time.perf_counter() - prompt_build_start) * 1000
|
||
|
||
# 调用LLM获取决策
|
||
reasoning, actions, llm_raw_output, llm_reasoning, llm_duration_ms = await self._execute_main_planner(
|
||
prompt=prompt,
|
||
message_id_list=message_id_list,
|
||
filtered_actions=filtered_actions,
|
||
available_actions=available_actions,
|
||
loop_start_time=loop_start_time,
|
||
)
|
||
|
||
# 如果有强制回复消息，确保回复该消息
|
||
if force_reply_message:
|
||
# 检查是否已经有回复该消息的 action
|
||
has_reply_to_force_message = any(
|
||
action.action_type == "reply"
|
||
and action.action_message
|
||
and action.action_message.message_id == force_reply_message.message_id
|
||
for action in actions
|
||
)
|
||
|
||
# 如果没有回复该消息，强制添加回复 action
|
||
if not has_reply_to_force_message:
|
||
# 移除所有 no_reply action（如果有）
|
||
actions = [a for a in actions if a.action_type != "no_reply"]
|
||
|
||
# 创建强制回复 action
|
||
available_actions_dict = dict(current_available_actions)
|
||
force_reply_action = ActionPlannerInfo(
|
||
action_type="reply",
|
||
reasoning="用户提及了我，必须回复该消息",
|
||
action_data={"loop_start_time": loop_start_time},
|
||
action_message=force_reply_message,
|
||
available_actions=available_actions_dict,
|
||
action_reasoning=None,
|
||
)
|
||
# 将强制回复 action 放在最前面
|
||
actions.insert(0, force_reply_action)
|
||
logger.info(f"{self.log_prefix} 检测到强制回复消息，已添加回复动作")
|
||
|
||
logger.info(
|
||
f"{self.log_prefix}Planner:{reasoning}。选择了{len(actions)}个动作: {' '.join([a.action_type for a in actions])}"
|
||
)
|
||
|
||
self.add_plan_log(reasoning, actions)
|
||
|
||
try:
|
||
PlanReplyLogger.log_plan(
|
||
chat_id=self.chat_id,
|
||
prompt=prompt,
|
||
reasoning=reasoning,
|
||
raw_output=llm_raw_output,
|
||
raw_reasoning=llm_reasoning,
|
||
actions=actions,
|
||
timing={
|
||
"prompt_build_ms": round(prompt_build_ms, 2),
|
||
"llm_duration_ms": round(llm_duration_ms, 2) if llm_duration_ms is not None else None,
|
||
"total_plan_ms": round((time.perf_counter() - plan_start) * 1000, 2),
|
||
"loop_start_time": loop_start_time,
|
||
},
|
||
extra=None,
|
||
)
|
||
except Exception:
|
||
logger.exception(f"{self.log_prefix}记录plan日志失败")
|
||
|
||
return actions
|
||
|
||
def add_plan_log(self, reasoning: str, actions: List[ActionPlannerInfo]):
|
||
self.plan_log.append((reasoning, time.time(), actions))
|
||
if len(self.plan_log) > 20:
|
||
self.plan_log.pop(0)
|
||
|
||
def add_plan_excute_log(self, result: str):
|
||
self.plan_log.append(("", time.time(), result))
|
||
if len(self.plan_log) > 20:
|
||
self.plan_log.pop(0)
|
||
|
||
def get_plan_log_str(self, max_action_records: int = 2, max_execution_records: int = 5) -> str:
|
||
"""
|
||
获取计划日志字符串
|
||
|
||
Args:
|
||
max_action_records: 显示多少条最新的action记录，默认2
|
||
max_execution_records: 显示多少条最新执行结果记录，默认8
|
||
|
||
Returns:
|
||
格式化的日志字符串
|
||
"""
|
||
action_records = []
|
||
execution_records = []
|
||
|
||
# 从后往前遍历，收集最新的记录
|
||
for reasoning, timestamp, content in reversed(self.plan_log):
|
||
if isinstance(content, list) and all(isinstance(action, ActionPlannerInfo) for action in content):
|
||
if len(action_records) < max_action_records:
|
||
action_records.append((reasoning, timestamp, content, "action"))
|
||
elif len(execution_records) < max_execution_records:
|
||
execution_records.append((reasoning, timestamp, content, "execution"))
|
||
|
||
# 合并所有记录并按时间戳排序
|
||
all_records = action_records + execution_records
|
||
all_records.sort(key=lambda x: x[1]) # 按时间戳排序
|
||
|
||
plan_log_str = ""
|
||
|
||
# 按时间顺序添加所有记录
|
||
for reasoning, timestamp, content, record_type in all_records:
|
||
time_str = datetime.fromtimestamp(timestamp).strftime("%H:%M:%S")
|
||
if record_type == "action":
|
||
# plan_log_str += f"{time_str}:{reasoning}|你使用了{','.join([action.action_type for action in content])}\n"
|
||
plan_log_str += f"{time_str}:{reasoning}\n"
|
||
else:
|
||
plan_log_str += f"{time_str}:你执行了action:{content}\n"
|
||
|
||
return plan_log_str
|
||
|
||
async def build_planner_prompt(
|
||
self,
|
||
is_group_chat: bool,
|
||
chat_target_info: Optional["TargetPersonInfo"],
|
||
current_available_actions: Dict[str, ActionInfo],
|
||
message_id_list: List[Tuple[str, "DatabaseMessages"]],
|
||
chat_content_block: str = "",
|
||
interest: str = "",
|
||
) -> tuple[str, List[Tuple[str, "DatabaseMessages"]]]:
|
||
"""构建 Planner LLM 的提示词 (获取模板并填充数据)"""
|
||
try:
|
||
actions_before_now_block = self.get_plan_log_str()
|
||
|
||
# 构建聊天上下文描述
|
||
chat_context_description = "你现在正在一个群聊中"
|
||
|
||
# 构建动作选项块
|
||
action_options_block = await self._build_action_options_block(current_available_actions)
|
||
|
||
# 其他信息
|
||
moderation_prompt_block = "请不要输出违法违规内容，不要输出色情，暴力，政治相关内容，如有敏感内容，请规避。"
|
||
time_block = f"当前时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
||
bot_name = global_config.bot.nickname
|
||
bot_nickname = (
|
||
f",也有人叫你{','.join(global_config.bot.alias_names)}" if global_config.bot.alias_names else ""
|
||
)
|
||
name_block = f"你的名字是{bot_name}{bot_nickname}，请注意哪些是你自己的发言。"
|
||
|
||
# 根据 think_mode 配置决定 reply action 的示例 JSON
|
||
# 在 JSON 中直接作为 action 参数携带 unknown_words
|
||
if global_config.chat.think_mode == "classic":
|
||
reply_action_example = ""
|
||
if global_config.chat.llm_quote:
|
||
reply_action_example += (
|
||
"5.如果要明确回复消息，使用quote，如果消息不多不需要明确回复，设置quote为false\n"
|
||
)
|
||
reply_action_example += (
|
||
'{{"action":"reply", "target_message_id":"消息id(m+数字)", "unknown_words":["词语1","词语2"]'
|
||
)
|
||
if global_config.chat.llm_quote:
|
||
reply_action_example += ', "quote":"如果需要引用该message，设置为true"'
|
||
reply_action_example += "}"
|
||
else:
|
||
reply_action_example = (
|
||
"5.think_level表示思考深度，0表示该回复不需要思考和回忆，1表示该回复需要进行回忆和思考\n"
|
||
)
|
||
if global_config.chat.llm_quote:
|
||
reply_action_example += (
|
||
"6.如果要明确回复消息，使用quote，如果消息不多不需要明确回复，设置quote为false\n"
|
||
)
|
||
reply_action_example += (
|
||
'{{"action":"reply", "think_level":数值等级(0或1), '
|
||
'"target_message_id":"消息id(m+数字)", '
|
||
'"unknown_words":["词语1","词语2"]'
|
||
)
|
||
if global_config.chat.llm_quote:
|
||
reply_action_example += ', "quote":"如果需要引用该message，设置为true"'
|
||
reply_action_example += "}"
|
||
|
||
planner_prompt_template = prompt_manager.get_prompt("planner")
|
||
planner_prompt_template.add_context("time_block", time_block)
|
||
planner_prompt_template.add_context("chat_context_description", chat_context_description)
|
||
planner_prompt_template.add_context("chat_content_block", chat_content_block)
|
||
planner_prompt_template.add_context("actions_before_now_block", actions_before_now_block)
|
||
planner_prompt_template.add_context("action_options_text", action_options_block)
|
||
planner_prompt_template.add_context("moderation_prompt", moderation_prompt_block)
|
||
planner_prompt_template.add_context("name_block", name_block)
|
||
planner_prompt_template.add_context("interest", interest)
|
||
planner_prompt_template.add_context("plan_style", global_config.personality.plan_style)
|
||
planner_prompt_template.add_context("reply_action_example", reply_action_example)
|
||
prompt = await prompt_manager.render_prompt(planner_prompt_template)
|
||
|
||
return prompt, message_id_list
|
||
except Exception as e:
|
||
logger.error(f"构建 Planner 提示词时出错: {e}")
|
||
logger.error(traceback.format_exc())
|
||
return "构建 Planner Prompt 时出错", []
|
||
|
||
def get_necessary_info(self) -> Tuple[bool, Optional["TargetPersonInfo"], Dict[str, ActionInfo]]:
|
||
"""
|
||
获取 Planner 需要的必要信息
|
||
"""
|
||
is_group_chat = True
|
||
is_group_chat, chat_target_info = get_chat_type_and_target_info(self.chat_id)
|
||
logger.debug(f"{self.log_prefix}获取到聊天信息 - 群聊: {is_group_chat}, 目标信息: {chat_target_info}")
|
||
|
||
current_available_actions_dict = self.action_manager.get_using_actions()
|
||
|
||
# 获取完整的动作信息
|
||
all_registered_actions: Dict[str, ActionInfo] = component_registry.get_components_by_type( # type: ignore
|
||
ComponentType.ACTION
|
||
)
|
||
current_available_actions = {}
|
||
for action_name in current_available_actions_dict:
|
||
if action_name in all_registered_actions:
|
||
current_available_actions[action_name] = all_registered_actions[action_name]
|
||
else:
|
||
logger.warning(f"{self.log_prefix}使用中的动作 {action_name} 未在已注册动作中找到")
|
||
|
||
return is_group_chat, chat_target_info, current_available_actions
|
||
|
||
def _filter_actions_by_activation_type(
|
||
self, available_actions: Dict[str, ActionInfo], chat_content_block: str
|
||
) -> Dict[str, ActionInfo]:
|
||
"""根据激活类型过滤动作"""
|
||
filtered_actions = {}
|
||
|
||
for action_name, action_info in available_actions.items():
|
||
if action_info.activation_type == ActionActivationType.NEVER:
|
||
logger.debug(f"{self.log_prefix}动作 {action_name} 设置为 NEVER 激活类型，跳过")
|
||
continue
|
||
elif action_info.activation_type == ActionActivationType.ALWAYS:
|
||
filtered_actions[action_name] = action_info
|
||
elif action_info.activation_type == ActionActivationType.RANDOM:
|
||
if random.random() < action_info.random_activation_probability:
|
||
filtered_actions[action_name] = action_info
|
||
elif action_info.activation_type == ActionActivationType.KEYWORD:
|
||
if action_info.activation_keywords:
|
||
for keyword in action_info.activation_keywords:
|
||
if keyword in chat_content_block:
|
||
filtered_actions[action_name] = action_info
|
||
break
|
||
else:
|
||
logger.warning(f"{self.log_prefix}未知的激活类型: {action_info.activation_type}，跳过处理")
|
||
|
||
return filtered_actions
|
||
|
||
async def _build_action_options_block(self, current_available_actions: Dict[str, ActionInfo]) -> str:
|
||
"""构建动作选项块"""
|
||
if not current_available_actions:
|
||
return ""
|
||
|
||
action_options_block = ""
|
||
for action_name, action_info in current_available_actions.items():
|
||
# 构建参数文本
|
||
param_text = ""
|
||
if action_info.action_parameters:
|
||
param_text = "\n"
|
||
for param_name, param_description in action_info.action_parameters.items():
|
||
param_text += f' "{param_name}":"{param_description}"\n'
|
||
param_text = param_text.rstrip("\n")
|
||
|
||
# 构建要求文本
|
||
require_text = "\n".join(f"- {require_item}" for require_item in action_info.action_require)
|
||
|
||
parallel_text = "" if action_info.parallel_action else "(当选择这个动作时，请不要选择其他动作)"
|
||
|
||
# 获取动作提示模板并填充
|
||
using_action_prompt = prompt_manager.get_prompt("action")
|
||
using_action_prompt.add_context("action_name", action_name)
|
||
using_action_prompt.add_context("action_description", action_info.description)
|
||
using_action_prompt.add_context("action_parameters", param_text)
|
||
using_action_prompt.add_context("action_require", require_text)
|
||
using_action_prompt.add_context("parallel_text", parallel_text)
|
||
using_action_rendered_prompt = await prompt_manager.render_prompt(using_action_prompt)
|
||
|
||
action_options_block += using_action_rendered_prompt
|
||
|
||
return action_options_block
|
||
|
||
async def _execute_main_planner(
|
||
self,
|
||
prompt: str,
|
||
message_id_list: List[Tuple[str, "DatabaseMessages"]],
|
||
filtered_actions: Dict[str, ActionInfo],
|
||
available_actions: Dict[str, ActionInfo],
|
||
loop_start_time: float,
|
||
) -> Tuple[str, List[ActionPlannerInfo], Optional[str], Optional[str], Optional[float]]:
|
||
"""执行主规划器"""
|
||
llm_content = None
|
||
actions: List[ActionPlannerInfo] = []
|
||
llm_reasoning = None
|
||
llm_duration_ms = None
|
||
|
||
try:
|
||
# 调用LLM
|
||
llm_start = time.perf_counter()
|
||
llm_content, (reasoning_content, _, _) = await self.planner_llm.generate_response_async(prompt=prompt)
|
||
llm_duration_ms = (time.perf_counter() - llm_start) * 1000
|
||
llm_reasoning = reasoning_content
|
||
|
||
if global_config.debug.show_planner_prompt:
|
||
logger.info(f"{self.log_prefix}规划器原始提示词: {prompt}")
|
||
logger.info(f"{self.log_prefix}规划器原始响应: {llm_content}")
|
||
if reasoning_content:
|
||
logger.info(f"{self.log_prefix}规划器推理: {reasoning_content}")
|
||
else:
|
||
logger.debug(f"{self.log_prefix}规划器原始提示词: {prompt}")
|
||
logger.debug(f"{self.log_prefix}规划器原始响应: {llm_content}")
|
||
if reasoning_content:
|
||
logger.debug(f"{self.log_prefix}规划器推理: {reasoning_content}")
|
||
|
||
except Exception as req_e:
|
||
logger.error(f"{self.log_prefix}LLM 请求执行失败: {req_e}")
|
||
return (
|
||
f"LLM 请求失败，模型出现问题: {req_e}",
|
||
[
|
||
ActionPlannerInfo(
|
||
action_type="no_reply",
|
||
reasoning=f"LLM 请求失败，模型出现问题: {req_e}",
|
||
action_data={},
|
||
action_message=None,
|
||
available_actions=available_actions,
|
||
)
|
||
],
|
||
llm_content,
|
||
llm_reasoning,
|
||
llm_duration_ms,
|
||
)
|
||
|
||
# 解析LLM响应
|
||
extracted_reasoning = ""
|
||
if llm_content:
|
||
try:
|
||
json_objects, extracted_reasoning = self._extract_json_from_markdown(llm_content)
|
||
extracted_reasoning = self._replace_message_ids_with_text(extracted_reasoning, message_id_list) or ""
|
||
if json_objects:
|
||
logger.debug(f"{self.log_prefix}从响应中提取到{len(json_objects)}个JSON对象")
|
||
filtered_actions_list = list(filtered_actions.items())
|
||
for json_obj in json_objects:
|
||
actions.extend(
|
||
self._parse_single_action(
|
||
json_obj, message_id_list, filtered_actions_list, extracted_reasoning
|
||
)
|
||
)
|
||
else:
|
||
# 尝试解析为直接的JSON
|
||
logger.warning(f"{self.log_prefix}LLM没有返回可用动作: {llm_content}")
|
||
extracted_reasoning = "LLM没有返回可用动作"
|
||
actions = self._create_no_reply("LLM没有返回可用动作", available_actions)
|
||
|
||
except Exception as json_e:
|
||
logger.warning(f"{self.log_prefix}解析LLM响应JSON失败 {json_e}. LLM原始输出: '{llm_content}'")
|
||
extracted_reasoning = f"解析LLM响应JSON失败: {json_e}"
|
||
actions = self._create_no_reply(f"解析LLM响应JSON失败: {json_e}", available_actions)
|
||
traceback.print_exc()
|
||
else:
|
||
extracted_reasoning = "规划器没有获得LLM响应"
|
||
actions = self._create_no_reply("规划器没有获得LLM响应", available_actions)
|
||
|
||
# 添加循环开始时间到所有非no_reply动作
|
||
for action in actions:
|
||
action.action_data = action.action_data or {}
|
||
action.action_data["loop_start_time"] = loop_start_time
|
||
|
||
# 去重：如果同一个动作被选择了多次，随机选择其中一个
|
||
if actions:
|
||
shuffled = actions.copy()
|
||
random.shuffle(shuffled)
|
||
actions = list({a.action_type: a for a in shuffled}.values())
|
||
|
||
# 处理黑话缓存逻辑
|
||
self._process_unknown_words_cache(actions)
|
||
|
||
logger.debug(f"{self.log_prefix}规划器选择了{len(actions)}个动作: {' '.join([a.action_type for a in actions])}")
|
||
|
||
return extracted_reasoning, actions, llm_content, llm_reasoning, llm_duration_ms
|
||
|
||
def _create_no_reply(self, reasoning: str, available_actions: Dict[str, ActionInfo]) -> List[ActionPlannerInfo]:
|
||
"""创建no_reply"""
|
||
return [
|
||
ActionPlannerInfo(
|
||
action_type="no_reply",
|
||
reasoning=reasoning,
|
||
action_data={},
|
||
action_message=None,
|
||
available_actions=available_actions,
|
||
)
|
||
]
|
||
|
||
def _extract_json_from_markdown(self, content: str) -> Tuple[List[dict], str]:
|
||
# sourcery skip: for-append-to-extend
|
||
"""从Markdown格式的内容中提取JSON对象和推理内容"""
|
||
json_objects = []
|
||
reasoning_content = ""
|
||
|
||
# 使用正则表达式查找```json包裹的JSON内容
|
||
json_pattern = r"```json\s*(.*?)\s*```"
|
||
markdown_matches = re.findall(json_pattern, content, re.DOTALL)
|
||
|
||
# 提取JSON之前的内容作为推理文本
|
||
first_json_pos = len(content)
|
||
if markdown_matches:
|
||
# 找到第一个```json的位置
|
||
first_json_pos = content.find("```json")
|
||
if first_json_pos > 0:
|
||
reasoning_content = content[:first_json_pos].strip()
|
||
# 清理推理内容中的注释标记
|
||
reasoning_content = re.sub(r"^//\s*", "", reasoning_content, flags=re.MULTILINE)
|
||
reasoning_content = reasoning_content.strip()
|
||
|
||
# 处理```json包裹的JSON
|
||
for match in markdown_matches:
|
||
try:
|
||
# 清理可能的注释和格式问题
|
||
json_str = re.sub(r"//.*?\n", "\n", match) # 移除单行注释
|
||
json_str = re.sub(r"/\*.*?\*/", "", json_str, flags=re.DOTALL) # 移除多行注释
|
||
if json_str := json_str.strip():
|
||
# 尝试按行分割，每行可能是一个JSON对象
|
||
lines = [line.strip() for line in json_str.split("\n") if line.strip()]
|
||
for line in lines:
|
||
with contextlib.suppress(json.JSONDecodeError):
|
||
json_obj = json.loads(repair_json(line))
|
||
if isinstance(json_obj, dict):
|
||
if json_obj:
|
||
json_objects.append(json_obj)
|
||
elif isinstance(json_obj, list):
|
||
for item in json_obj:
|
||
if isinstance(item, dict) and item:
|
||
json_objects.append(item)
|
||
|
||
# 如果按行解析没有成功（或只得到空字典），尝试将整个块作为一个JSON对象或数组
|
||
if not json_objects:
|
||
json_obj = json.loads(repair_json(json_str))
|
||
if isinstance(json_obj, dict):
|
||
# 过滤掉空字典
|
||
if json_obj:
|
||
json_objects.append(json_obj)
|
||
elif isinstance(json_obj, list):
|
||
for item in json_obj:
|
||
if isinstance(item, dict) and item:
|
||
json_objects.append(item)
|
||
except Exception as e:
|
||
logger.warning(f"解析JSON块失败: {e}, 块内容: {match[:100]}...")
|
||
continue
|
||
|
||
# 如果没有找到完整的```json```块，尝试查找不完整的代码块（缺少结尾```）
|
||
if not json_objects:
|
||
json_start_pos = content.find("```json")
|
||
if json_start_pos != -1:
|
||
# 找到```json之后的内容
|
||
json_content_start = json_start_pos + 7 # ```json的长度
|
||
# 提取从```json之后到内容结尾的所有内容
|
||
incomplete_json_str = content[json_content_start:].strip()
|
||
|
||
# 提取JSON之前的内容作为推理文本
|
||
if json_start_pos > 0:
|
||
reasoning_content = content[:json_start_pos].strip()
|
||
reasoning_content = re.sub(r"^//\s*", "", reasoning_content, flags=re.MULTILINE)
|
||
reasoning_content = reasoning_content.strip()
|
||
|
||
if incomplete_json_str:
|
||
try:
|
||
# 清理可能的注释和格式问题
|
||
json_str = re.sub(r"//.*?\n", "\n", incomplete_json_str)
|
||
json_str = re.sub(r"/\*.*?\*/", "", json_str, flags=re.DOTALL)
|
||
json_str = json_str.strip()
|
||
|
||
if json_str:
|
||
# 尝试按行分割，每行可能是一个JSON对象
|
||
lines = [line.strip() for line in json_str.split("\n") if line.strip()]
|
||
for line in lines:
|
||
try:
|
||
json_obj = json.loads(repair_json(line))
|
||
if isinstance(json_obj, dict):
|
||
# 过滤掉空字典，避免单个 { 字符被错误修复为 {} 的情况
|
||
if json_obj:
|
||
json_objects.append(json_obj)
|
||
elif isinstance(json_obj, list):
|
||
for item in json_obj:
|
||
if isinstance(item, dict) and item:
|
||
json_objects.append(item)
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
# 如果按行解析没有成功（或只得到空字典），尝试将整个块作为一个JSON对象或数组
|
||
if not json_objects:
|
||
try:
|
||
json_obj = json.loads(repair_json(json_str))
|
||
if isinstance(json_obj, dict):
|
||
# 过滤掉空字典
|
||
if json_obj:
|
||
json_objects.append(json_obj)
|
||
elif isinstance(json_obj, list):
|
||
for item in json_obj:
|
||
if isinstance(item, dict) and item:
|
||
json_objects.append(item)
|
||
except Exception as e:
|
||
logger.debug(f"尝试解析不完整的JSON代码块失败: {e}")
|
||
except Exception as e:
|
||
logger.debug(f"处理不完整的JSON代码块时出错: {e}")
|
||
|
||
return json_objects, reasoning_content
|