1b8876c4bb
本次提交对亲和流聊天器（AFC）的计划与决策核心进行了重大重构和功能增强，旨在提升其响应的灵活性、鲁棒性和可观测性。
主要变更包括：
1. **多动作支持与解析重构**:
- `PlanFilter` 现在能够正确解析并处理 LLM 返回的动作列表（`"actions": [...]`），而不仅限于单个动作，这使得机器人能够执行更复杂的组合行为。
- 增强了动作解析的鲁棒性，当找不到 `target_message_id` 时会优雅降级（如 `reply` 变为 `no_action`），并会根据当前实际可用的动作列表对 LLM 的选择进行验证。
2. **提示词工程与思考模式优化**:
- 重新设计了核心 Planner 提示词，将 `thinking` 字段定义为“思绪流”，引导 LLM 生成更自然、更符合角色的内心独白，而非简单的决策理由，从而提升决策质量和角色扮演的沉浸感。
- 强制要求 LLM 为需要目标消息的动作提供 `target_message_id`，提高了动作执行的准确性。
3. **上下文构建与鲁棒性增强**:
- 在 `PlanFilter` 中增加了上下文回退机制，当内存中缺少历史消息时（如冷启动），会自动从数据库加载最近的消息记录，确保决策所需上下文的完整性。
- 简化了提供给 LLM 的未读消息格式，移除了兴趣度分数等内部信息，并加入了用户昵称，使其更易于理解和处理。
4. **可观测性与日志改进**:
- 在 AFC 的多个关键节点（消息接收、决策、动作执行）增加了彩色的详细日志，使其决策流程像 HFC 一样清晰可见，极大地方便了调试。
- 将系统中多个模块（视频分析、兴趣度匹配、情绪管理）的常规日志级别从 `INFO` 调整为 `DEBUG`，以减少在生产环境中的日志噪音。
5. **动作描述优化**:
- 优化了 `set_emoji_like` 和 `emoji` 等动作的描述，使其意图更清晰，帮助 LLM 做出更准确的动作选择。
198 lines
7.9 KiB
Python
198 lines
7.9 KiB
Python
import re
|
||
from typing import List, Tuple, Type
|
||
|
||
from src.plugin_system import (
|
||
BasePlugin,
|
||
register_plugin,
|
||
BaseAction,
|
||
ComponentInfo,
|
||
ActionActivationType,
|
||
ConfigField,
|
||
)
|
||
from src.common.logger import get_logger
|
||
from src.plugin_system.apis import send_api
|
||
from .qq_emoji_list import qq_face
|
||
from src.plugin_system.base.component_types import ChatType
|
||
|
||
logger = get_logger("set_emoji_like_plugin")
|
||
|
||
|
||
def get_emoji_id(emoji_input: str) -> str | None:
|
||
"""根据输入获取表情ID"""
|
||
# 如果输入本身就是数字ID，直接返回
|
||
if emoji_input.isdigit() or (isinstance(emoji_input, str) and emoji_input.startswith("😊")):
|
||
if emoji_input in qq_face:
|
||
return emoji_input
|
||
|
||
# 尝试从 "[表情：xxx]" 格式中提取
|
||
match = re.search(r"\[表情：(.+?)\]", emoji_input)
|
||
if match:
|
||
emoji_name = match.group(1).strip()
|
||
else:
|
||
emoji_name = emoji_input.strip()
|
||
|
||
# 遍历查找
|
||
for key, value in qq_face.items():
|
||
# value 的格式是 "[表情：xxx]"
|
||
if f"[表情：{emoji_name}]" == value:
|
||
return key
|
||
|
||
return None
|
||
|
||
|
||
# ===== Action组件 =====
|
||
class SetEmojiLikeAction(BaseAction):
|
||
"""设置消息表情回应"""
|
||
|
||
# === 基本信息（必须填写）===
|
||
action_name = "set_emoji_like"
|
||
action_description = "为某条已经存在的消息添加‘贴表情’回应（类似点赞），而不是发送新消息。可以在觉得某条消息非常有趣、值得赞同或者需要特殊情感回应时主动使用。"
|
||
activation_type = ActionActivationType.ALWAYS # 消息接收时激活(?)
|
||
chat_type_allow = ChatType.GROUP
|
||
parallel_action = True
|
||
|
||
# === 功能描述（必须填写）===
|
||
# 从 qq_face 字典中提取所有表情名称用于提示
|
||
emoji_options = []
|
||
for name in qq_face.values():
|
||
match = re.search(r"\[表情：(.+?)\]", name)
|
||
if match:
|
||
emoji_options.append(match.group(1))
|
||
|
||
action_parameters = {
|
||
"emoji": f"要回应的表情,必须从以下表情中选择: {', '.join(emoji_options)}",
|
||
"set": "是否设置回应 (True/False)",
|
||
}
|
||
action_require = [
|
||
"当需要对一个已存在消息进行‘贴表情’回应时使用",
|
||
"这是一个对旧消息的操作，而不是发送新消息",
|
||
"如果你想发送一个新的表情包消息，请使用 'emoji' 动作",
|
||
]
|
||
llm_judge_prompt = """
|
||
判定是否需要使用贴表情动作的条件：
|
||
1. 用户明确要求使用贴表情包
|
||
2. 这是一个适合表达强烈情绪的场合
|
||
3. 不要发送太多表情包，如果你已经发送过多个表情包则回答"否"
|
||
|
||
请回答"是"或"否"。
|
||
"""
|
||
associated_types = ["text"]
|
||
|
||
async def execute(self) -> Tuple[bool, str]:
|
||
"""执行设置表情回应的动作"""
|
||
message_id = None
|
||
if self.has_action_message:
|
||
logger.debug(str(self.action_message))
|
||
if isinstance(self.action_message, dict):
|
||
message_id = self.action_message.get("message_id")
|
||
logger.info(f"获取到的消息ID: {message_id}")
|
||
else:
|
||
logger.error("未提供消息ID")
|
||
await self.store_action_info(
|
||
action_build_into_prompt=True,
|
||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 未提供消息ID",
|
||
action_done=False,
|
||
)
|
||
return False, "未提供消息ID"
|
||
|
||
emoji_input = self.action_data.get("emoji")
|
||
set_like = self.action_data.get("set", True)
|
||
|
||
if not emoji_input:
|
||
logger.error("未提供表情")
|
||
return False, "未提供表情"
|
||
logger.info(f"设置表情回应: {emoji_input}, 是否设置: {set_like}")
|
||
|
||
emoji_id = get_emoji_id(emoji_input)
|
||
if not emoji_id:
|
||
logger.error(f"找不到表情: '{emoji_input}'。请从可用列表中选择。")
|
||
await self.store_action_info(
|
||
action_build_into_prompt=True,
|
||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 找不到表情: '{emoji_input}'",
|
||
action_done=False,
|
||
)
|
||
return False, f"找不到表情: '{emoji_input}'。请从可用列表中选择。"
|
||
|
||
# 4. 使用适配器API发送命令
|
||
if not message_id:
|
||
logger.error("未提供消息ID")
|
||
await self.store_action_info(
|
||
action_build_into_prompt=True,
|
||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: 未提供消息ID",
|
||
action_done=False,
|
||
)
|
||
return False, "未提供消息ID"
|
||
|
||
try:
|
||
# 使用适配器API发送贴表情命令
|
||
response = await send_api.adapter_command_to_stream(
|
||
action="set_msg_emoji_like",
|
||
params={"message_id": message_id, "emoji_id": emoji_id, "set": set_like},
|
||
stream_id=self.chat_stream.stream_id if self.chat_stream else None,
|
||
timeout=30.0,
|
||
storage_message=False,
|
||
)
|
||
|
||
if response["status"] == "ok":
|
||
logger.info(f"设置表情回应成功: {response}")
|
||
await self.store_action_info(
|
||
action_build_into_prompt=True,
|
||
action_prompt_display=f"执行了set_emoji_like动作,{emoji_input},设置表情回应: {emoji_id}, 是否设置: {set_like}",
|
||
action_done=True,
|
||
)
|
||
return True, f"成功设置表情回应: {response.get('message', '成功')}"
|
||
else:
|
||
error_msg = response.get("message", "未知错误")
|
||
logger.error(f"设置表情回应失败: {error_msg}")
|
||
await self.store_action_info(
|
||
action_build_into_prompt=True,
|
||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: {error_msg}",
|
||
action_done=False,
|
||
)
|
||
return False, f"设置表情回应失败: {error_msg}"
|
||
|
||
except Exception as e:
|
||
logger.error(f"设置表情回应失败: {e}")
|
||
await self.store_action_info(
|
||
action_build_into_prompt=True,
|
||
action_prompt_display=f"执行了set_emoji_like动作：{self.action_name},失败: {e}",
|
||
action_done=False,
|
||
)
|
||
return False, f"设置表情回应失败: {e}"
|
||
|
||
|
||
# ===== 插件注册 =====
|
||
@register_plugin
|
||
class SetEmojiLikePlugin(BasePlugin):
|
||
"""设置消息表情回应插件"""
|
||
|
||
# 插件基本信息
|
||
plugin_name: str = "set_emoji_like" # 内部标识符
|
||
enable_plugin: bool = True
|
||
dependencies: List[str] = [] # 插件依赖列表
|
||
python_dependencies: List[str] = [] # Python包依赖列表，现在使用内置API
|
||
config_file_name: str = "config.toml" # 配置文件名
|
||
|
||
# 配置节描述
|
||
config_section_descriptions = {"plugin": "插件基本信息", "components": "插件组件"}
|
||
|
||
# 配置Schema定义
|
||
config_schema: dict = {
|
||
"plugin": {
|
||
"name": ConfigField(type=str, default="set_emoji_like", description="插件名称"),
|
||
"version": ConfigField(type=str, default="1.0.0", description="插件版本"),
|
||
"enabled": ConfigField(type=bool, default=True, description="是否启用插件"),
|
||
"config_version": ConfigField(type=str, default="1.1", description="配置版本"),
|
||
},
|
||
"components": {
|
||
"action_set_emoji_like": ConfigField(type=bool, default=True, description="是否启用设置表情回应功能"),
|
||
},
|
||
}
|
||
|
||
def get_plugin_components(self) -> List[Tuple[ComponentInfo, Type]]:
|
||
if self.get_config("components.action_set_emoji_like"):
|
||
return [
|
||
(SetEmojiLikeAction.get_action_info(), SetEmojiLikeAction),
|
||
]
|
||
return []
|