统一处理机器人识别逻辑,支持多平台(包括 WebUI)

pull/1485/head
墨梓柒 2025-12-29 21:42:07 +08:00
parent b348a81b19
commit 3dbb53b73c
No known key found for this signature in database
GPG Key ID: 4A65B9DBA35F7635
8 changed files with 124 additions and 33 deletions

View File

@ -18,7 +18,7 @@ from src.chat.utils.chat_message_builder import (
get_raw_msg_before_timestamp_with_chat,
replace_user_references,
)
from src.chat.utils.utils import get_chat_type_and_target_info
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_stream import get_chat_manager
from src.plugin_system.base.component_types import ActionInfo, ComponentType, ActionActivationType
@ -311,11 +311,9 @@ class ActionPlanner:
return action_planner_infos
def _is_message_from_self(self, message: "DatabaseMessages") -> bool:
"""判断消息是否由机器人自身发送"""
"""判断消息是否由机器人自身发送(支持多平台,包括 WebUI"""
try:
return str(message.user_info.user_id) == str(global_config.bot.qq_account) and (
message.user_info.platform or ""
) == (global_config.bot.platform or "")
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

View File

@ -16,7 +16,7 @@ from src.chat.message_receive.message import UserInfo, Seg, MessageRecv, Message
from src.chat.message_receive.chat_stream import ChatStream
from src.chat.message_receive.uni_message_sender import UniversalMessageSender
from src.chat.utils.timer_calculator import Timer # <--- Import Timer
from src.chat.utils.utils import get_chat_type_and_target_info
from src.chat.utils.utils import get_chat_type_and_target_info, is_bot_self
from src.chat.utils.prompt_builder import global_prompt_manager
from src.chat.utils.chat_message_builder import (
build_readable_messages,
@ -815,10 +815,8 @@ class DefaultReplyer:
person_list_short: List[Person] = []
for msg in message_list_before_short:
if (
global_config.bot.qq_account == msg.user_info.user_id
and global_config.bot.platform == msg.user_info.platform
):
# 使用统一的 is_bot_self 函数判断是否是机器人自己(支持多平台,包括 WebUI
if is_bot_self(msg.user_info.platform, msg.user_info.user_id):
continue
if (
reply_message

View File

@ -16,7 +16,7 @@ from src.chat.message_receive.message import UserInfo, Seg, MessageRecv, Message
from src.chat.message_receive.chat_stream import ChatStream
from src.chat.message_receive.uni_message_sender import UniversalMessageSender
from src.chat.utils.timer_calculator import Timer # <--- Import Timer
from src.chat.utils.utils import get_chat_type_and_target_info
from src.chat.utils.utils import get_chat_type_and_target_info, is_bot_self
from src.chat.utils.prompt_builder import global_prompt_manager
from src.chat.utils.chat_message_builder import (
build_readable_messages,
@ -679,10 +679,8 @@ class PrivateReplyer:
person_list_short: List[Person] = []
for msg in message_list_before_short:
if (
global_config.bot.qq_account == msg.user_info.user_id
and global_config.bot.platform == msg.user_info.platform
):
# 使用统一的 is_bot_self 函数判断是否是机器人自己(支持多平台,包括 WebUI
if is_bot_self(msg.user_info.platform, msg.user_info.user_id):
continue
if (
reply_message
@ -823,7 +821,8 @@ class PrivateReplyer:
# 兜底:即使 multiple_reply_style 配置异常也不影响正常回复
reply_style = global_config.personality.reply_style
if global_config.bot.qq_account == user_id and platform == global_config.bot.platform:
# 使用统一的 is_bot_self 函数判断是否是机器人自己(支持多平台,包括 WebUI
if is_bot_self(platform, user_id):
return await global_prompt_manager.format_prompt(
"private_replyer_self_prompt",
expression_habits_block=expression_habits_block,

View File

@ -13,7 +13,7 @@ from src.common.data_models.message_data_model import MessageAndActionModel
from src.common.database.database_model import ActionRecords
from src.common.database.database_model import Images
from src.person_info.person_info import Person, get_person_id
from src.chat.utils.utils import translate_timestamp_to_human_readable, assign_message_ids
from src.chat.utils.utils import translate_timestamp_to_human_readable, assign_message_ids, is_bot_self
install(extra_lines=3)
logger = get_logger("chat_message_builder")
@ -43,12 +43,9 @@ def replace_user_references(
if name_resolver is None:
def default_resolver(platform: str, user_id: str) -> str:
# 检查是否是机器人自己(支持多平台)
if replace_bot_name:
if platform == "qq" and user_id == global_config.bot.qq_account:
return f"{global_config.bot.nickname}(你)"
if platform == "telegram" and user_id == getattr(global_config.bot, "telegram_account", ""):
return f"{global_config.bot.nickname}(你)"
# 检查是否是机器人自己(支持多平台,包括 WebUI
if replace_bot_name and is_bot_self(platform, user_id):
return f"{global_config.bot.nickname}(你)"
person = Person(platform=platform, user_id=user_id)
return person.person_name or user_id # type: ignore
@ -61,8 +58,8 @@ def replace_user_references(
aaa = match[1]
bbb = match[2]
try:
# 检查是否是机器人自己
if replace_bot_name and bbb == global_config.bot.qq_account:
# 检查是否是机器人自己(支持多平台,包括 WebUI
if replace_bot_name and is_bot_self(platform, bbb):
reply_person_name = f"{global_config.bot.nickname}(你)"
else:
reply_person_name = name_resolver(platform, bbb) or aaa
@ -468,10 +465,8 @@ def _build_readable_messages_internal(
person_name = (
person.person_name or f"{user_nickname}" or (f"昵称:{user_cardname}" if user_cardname else "某人")
)
if replace_bot_name and (
(platform == global_config.bot.platform and user_id == global_config.bot.qq_account)
or (platform == "telegram" and user_id == getattr(global_config.bot, "telegram_account", ""))
):
# 使用统一的 is_bot_self 函数判断是否是机器人自己(支持多平台,包括 WebUI
if replace_bot_name and is_bot_self(platform, user_id):
person_name = f"{global_config.bot.nickname}(你)"
# 使用独立函数处理用户引用格式

View File

@ -67,6 +67,53 @@ def get_current_platform_account(platform: str, platform_accounts: dict[str, str
return platform_accounts.get(platform, "")
def is_bot_self(platform: str, user_id: str) -> bool:
"""判断给定的平台和用户ID是否是机器人自己
这个函数统一处理所有平台包括 QQTelegramWebUI 的机器人识别逻辑
Args:
platform: 消息平台 "qq", "telegram", "webui"
user_id: 用户ID
Returns:
bool: 如果是机器人自己则返回 True否则返回 False
"""
if not platform or not user_id:
return False
# 将 user_id 转为字符串进行比较
user_id_str = str(user_id)
# 获取机器人的 QQ 账号(主账号)
qq_account = str(global_config.bot.qq_account or "")
# QQ 平台:直接比较 QQ 账号
if platform == "qq":
return user_id_str == qq_account
# WebUI 平台:机器人回复时使用的是 QQ 账号,所以也比较 QQ 账号
if platform == "webui":
return user_id_str == qq_account
# 获取各平台账号映射
platforms_list = getattr(global_config.bot, "platforms", []) or []
platform_accounts = parse_platform_accounts(platforms_list)
# Telegram 平台
if platform == "telegram":
tg_account = platform_accounts.get("tg", "") or platform_accounts.get("telegram", "")
return user_id_str == tg_account if tg_account else False
# 其他平台:尝试从 platforms 配置中查找
platform_account = platform_accounts.get(platform, "")
if platform_account:
return user_id_str == platform_account
# 默认情况:与主 QQ 账号比较(兼容性)
return user_id_str == qq_account
def is_mentioned_bot_in_message(message: MessageRecv) -> tuple[bool, bool, float]:
"""检查消息是否提到了机器人(统一多平台实现)"""
text = message.processed_plain_text or ""

View File

@ -19,6 +19,7 @@ from src.config.config import global_config, model_config
from src.llm_models.utils_model import LLMRequest
from src.plugin_system.apis import message_api
from src.chat.utils.chat_message_builder import build_readable_messages
from src.chat.utils.utils import is_bot_self
from src.person_info.person_info import Person
from src.chat.message_receive.chat_stream import get_chat_manager
from src.chat.utils.prompt_builder import Prompt, global_prompt_manager
@ -415,11 +416,11 @@ class ChatHistorySummarizer:
# 1. 检查当前批次内是否有 bot 发言(只检查当前批次,不往前推)
# 原因:我们要记录的是 bot 参与过的对话片段,如果当前批次内 bot 没有发言,
# 说明 bot 没有参与这段对话,不应该记录
bot_user_id = str(global_config.bot.qq_account)
has_bot_message = False
for msg in messages:
if msg.user_info.user_id == bot_user_id:
# 使用统一的 is_bot_self 函数判断是否是机器人自己(支持多平台,包括 WebUI
if is_bot_self(msg.user_info.platform, msg.user_info.user_id):
has_bot_message = True
break

View File

@ -228,8 +228,59 @@ class Person:
return person
def _is_bot_self(self, platform: str, user_id: str) -> bool:
"""判断给定的平台和用户ID是否是机器人自己
这个函数统一处理所有平台包括 QQTelegramWebUI 的机器人识别逻辑
Args:
platform: 消息平台 "qq", "telegram", "webui"
user_id: 用户ID
Returns:
bool: 如果是机器人自己则返回 True否则返回 False
"""
if not platform or not user_id:
return False
# 将 user_id 转为字符串进行比较
user_id_str = str(user_id)
# 获取机器人的 QQ 账号(主账号)
qq_account = str(global_config.bot.qq_account or "")
# QQ 平台:直接比较 QQ 账号
if platform == "qq":
return user_id_str == qq_account
# WebUI 平台:机器人回复时使用的是 QQ 账号,所以也比较 QQ 账号
if platform == "webui":
return user_id_str == qq_account
# 获取各平台账号映射
platforms_list = getattr(global_config.bot, "platforms", []) or []
platform_accounts = {}
for platform_entry in platforms_list:
if ":" in platform_entry:
platform_name, account = platform_entry.split(":", 1)
platform_accounts[platform_name.strip()] = account.strip()
# Telegram 平台
if platform == "telegram":
tg_account = platform_accounts.get("tg", "") or platform_accounts.get("telegram", "")
return user_id_str == tg_account if tg_account else False
# 其他平台:尝试从 platforms 配置中查找
platform_account = platform_accounts.get(platform, "")
if platform_account:
return user_id_str == platform_account
# 默认情况:与主 QQ 账号比较(兼容性)
return user_id_str == qq_account
def __init__(self, platform: str = "", user_id: str = "", person_id: str = "", person_name: str = ""):
if platform == global_config.bot.platform and user_id == global_config.bot.qq_account:
# 使用统一的机器人识别函数(支持多平台,包括 WebUI
if self._is_bot_self(platform, user_id):
self.is_known = True
self.person_id = get_person_id(platform, user_id)
self.user_id = user_id

View File

@ -13,6 +13,7 @@ from typing import List, Dict, Any, Tuple, Optional
from src.common.data_models.database_data_model import DatabaseMessages
from src.common.database.database_model import Images
from src.config.config import global_config
from src.chat.utils.utils import is_bot_self
from src.chat.utils.chat_message_builder import (
get_raw_msg_by_timestamp,
get_raw_msg_by_timestamp_with_chat,
@ -511,7 +512,8 @@ def filter_mai_messages(messages: List[DatabaseMessages]) -> List[DatabaseMessag
Returns:
过滤后的消息列表
"""
return [msg for msg in messages if msg.user_info.user_id != str(global_config.bot.qq_account)]
# 使用统一的 is_bot_self 函数判断是否是机器人自己(支持多平台,包括 WebUI
return [msg for msg in messages if not is_bot_self(msg.user_info.platform, msg.user_info.user_id)]
def translate_pid_to_description(pid: str) -> str: