Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 为normal增加action_plan能力

Browse Source
This commit is contained in:
tcmofashi
2025-06-02 21:06:00 +08:00
parent 13c3a40ca5
commit d932b48444
11 changed files with 867 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

258
src/chat/normal_chat/normal_chat_planner.py Normal file
View File

@@ -0,0 +1,258 @@
import json
from typing import Dict, Any
from rich.traceback import install
from src.llm_models.utils_model import LLMRequest
from src.config.config import global_config
from src.common.logger_manager import get_logger
from src.chat.utils.prompt_builder import Prompt, global_prompt_manager
from src.individuality.individuality import individuality
from src.chat.focus_chat.planners.action_manager import ActionManager
from src.chat.normal_chat.normal_prompt import prompt_builder
from src.chat.message_receive.message import MessageThinking
from json_repair import repair_json
logger = get_logger("normal_chat_planner")
install(extra_lines=3)
def init_prompt():
Prompt(
"""
你的自我认知是：
{self_info_block}
注意，除了下面动作选项之外，你在聊天中不能做其他任何事情，这是你能力的边界，现在请你选择合适的action:
{action_options_text}
重要说明：
- "no_action" 表示只进行普通聊天回复，不执行任何额外动作
- 其他action表示在普通回复的基础上，执行相应的额外动作
你必须从上面列出的可用action中选择一个，并说明原因。
你的决策必须以严格的 JSON 格式输出，且仅包含 JSON 内容，不要有任何其他文字或解释。
{moderation_prompt}
当前聊天上下文：
{chat_context}
基于以上聊天上下文和用户的最新消息，选择最合适的action。
请你以下面格式输出你选择的action：
{{
"action": "action_name",
"reasoning": "说明你做出该action的原因",
"参数1": "参数1的值",
"参数2": "参数2的值",
"参数3": "参数3的值",
...
}}
请输出你的决策 JSON：""",
"normal_chat_planner_prompt",
)
Prompt(
"""
action_name: {action_name}
描述：{action_description}
参数：
{action_parameters}
动作要求：
{action_require}""",
"normal_chat_action_prompt",
)
class NormalChatPlanner:
def __init__(self, log_prefix: str, action_manager: ActionManager):
self.log_prefix = log_prefix
# LLM规划器配置
self.planner_llm = LLMRequest(
model=global_config.model.normal_chat_2,
max_tokens=1000,
request_type="normal_chat.planner", # 用于normal_chat动作规划
)
self.action_manager = action_manager
async def plan(self, message: MessageThinking, sender_name: str = "某人") -> Dict[str, Any]:
"""
Normal Chat 规划器: 使用LLM根据上下文决定做出什么动作。
参数:
message: 思考消息对象
sender_name: 发送者名称
"""
action = "no_action" # 默认动作改为no_action
reasoning = "规划器初始化默认"
action_data = {}
try:
# 设置默认值
nickname_str = ""
for nicknames in global_config.bot.alias_names:
nickname_str += f"{nicknames},"
name_block = f"你的名字是{global_config.bot.nickname},你的昵称有{nickname_str}，有人也会用这些昵称称呼你。"
personality_block = individuality.get_personality_prompt(x_person=2, level=2)
identity_block = individuality.get_identity_prompt(x_person=2, level=2)
self_info = name_block + personality_block + identity_block
# 获取当前可用的动作
current_available_actions = self.action_manager.get_using_actions()
# 如果没有可用动作或只有no_action动作，直接返回no_action
if not current_available_actions or (
len(current_available_actions) == 1 and "no_action" in current_available_actions
):
logger.debug(f"{self.log_prefix}规划器: 没有可用动作或只有no_action动作，返回no_action")
return {
"action_result": {"action_type": action, "action_data": action_data, "reasoning": reasoning},
"chat_context": "",
"action_prompt": "",
}
# 构建normal_chat的上下文 (使用与normal_chat相同的prompt构建方法)
chat_context = await prompt_builder.build_prompt(
message_txt=message.processed_plain_text,
sender_name=sender_name,
chat_stream=message.chat_stream,
)
# 构建planner的prompt
prompt = await self.build_planner_prompt(
self_info_block=self_info,
chat_context=chat_context,
current_available_actions=current_available_actions,
)
if not prompt:
logger.warning(f"{self.log_prefix}规划器: 构建提示词失败")
return {
"action_result": {"action_type": action, "action_data": action_data, "reasoning": reasoning},
"chat_context": chat_context,
"action_prompt": "",
}
# 使用LLM生成动作决策
try:
content, reasoning_content, model_name = await self.planner_llm.generate_response(prompt)
logger.debug(f"{self.log_prefix}规划器原始响应: {content}")
# 解析JSON响应
try:
# 尝试修复JSON
fixed_json = repair_json(content)
action_result = json.loads(fixed_json)
action = action_result.get("action", "no_action")
reasoning = action_result.get("reasoning", "未提供原因")
# 提取其他参数作为action_data
action_data = {k: v for k, v in action_result.items() if k not in ["action", "reasoning"]}
# 验证动作是否在可用动作列表中
if action not in current_available_actions:
logger.warning(f"{self.log_prefix}规划器选择了不可用的动作: {action}, 回退到no_action")
action = "no_action"
reasoning = f"选择的动作{action}不在可用列表中，回退到no_action"
action_data = {}
except json.JSONDecodeError as e:
logger.warning(f"{self.log_prefix}规划器JSON解析失败: {e}, 内容: {content}")
action = "no_action"
reasoning = "JSON解析失败，使用默认动作"
action_data = {}
except Exception as e:
logger.error(f"{self.log_prefix}规划器LLM调用失败: {e}")
action = "no_action"
reasoning = "LLM调用失败，使用默认动作"
action_data = {}
except Exception as outer_e:
logger.error(f"{self.log_prefix}规划器异常: {outer_e}")
chat_context = "无法获取聊天上下文" # 设置默认值
prompt = "" # 设置默认值
action = "no_action"
reasoning = "规划器出现异常，使用默认动作"
action_data = {}
logger.debug(f"{self.log_prefix}规划器决策动作:{action}, 动作信息: '{action_data}', 理由: {reasoning}")
# 恢复到默认动作集
self.action_manager.restore_actions()
logger.debug(
f"{self.log_prefix}规划后恢复到默认动作集, 当前可用: {list(self.action_manager.get_using_actions().keys())}"
)
action_result = {"action_type": action, "action_data": action_data, "reasoning": reasoning}
plan_result = {
"action_result": action_result,
"chat_context": chat_context,
"action_prompt": prompt,
}
return plan_result
async def build_planner_prompt(
self,
self_info_block: str,
chat_context: str,
current_available_actions: Dict[str, Any],
) -> str:
"""构建 Normal Chat Planner LLM 的提示词"""
try:
# 构建动作选项文本
action_options_text = ""
for action_name, action_info in current_available_actions.items():
action_description = action_info.get("description", "")
action_parameters = action_info.get("parameters", {})
action_require = action_info.get("require", [])
# 格式化参数
parameters_text = ""
for param_name, param_desc in action_parameters.items():
parameters_text += f" - {param_name}: {param_desc}\n"
# 格式化要求
require_text = ""
for req in action_require:
require_text += f" - {req}\n"
# 构建单个动作的提示
action_prompt = await global_prompt_manager.format_prompt(
"normal_chat_action_prompt",
action_name=action_name,
action_description=action_description,
action_parameters=parameters_text,
action_require=require_text,
)
action_options_text += action_prompt + "\n\n"
# 审核提示
moderation_prompt = "请确保你的回复符合平台规则，避免不当内容。"
# 使用模板构建最终提示词
prompt = await global_prompt_manager.format_prompt(
"normal_chat_planner_prompt",
self_info_block=self_info_block,
action_options_text=action_options_text,
moderation_prompt=moderation_prompt,
chat_context=chat_context,
)
return prompt
except Exception as e:
logger.error(f"{self.log_prefix}构建Planner提示词失败: {e}")
return ""
init_prompt()