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

Feat：添加对Action插件的支持，现在可以编写插件

Browse Source
This commit is contained in:
SengokuCola
2025-05-16 00:43:46 +08:00
parent ac6f96f805
commit cda9879bb2
28 changed files with 934 additions and 662 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
requirements.txt
View File

Binary file not shown.

221
src/chat/focus_chat/expressors/default_expressor.py
View File

@@ -10,7 +10,7 @@ from src.config.config import global_config
from src.chat.utils.utils_image import image_path_to_base64 # Local import needed after move
from src.chat.utils.timer_calculator import Timer # <--- Import Timer
from src.chat.emoji_system.emoji_manager import emoji_manager
from src.chat.focus_chat.heartflow_prompt_builder import prompt_builder
from src.chat.focus_chat.heartflow_prompt_builder import prompt_builder,Prompt
from src.chat.focus_chat.heartFC_sender import HeartFCSender
from src.chat.utils.utils import process_llm_response
from src.chat.utils.info_catcher import info_catcher_manager
@@ -18,9 +18,70 @@ from src.manager.mood_manager import mood_manager
from src.chat.heart_flow.utils_chat import get_chat_type_and_target_info
from src.chat.message_receive.chat_stream import ChatStream
from src.chat.focus_chat.hfc_utils import parse_thinking_id_to_timestamp
from src.config.config import global_config
from src.common.logger_manager import get_logger
from src.individuality.individuality import Individuality
from src.chat.utils.prompt_builder import Prompt, global_prompt_manager
from src.chat.utils.chat_message_builder import build_readable_messages, get_raw_msg_before_timestamp_with_chat
from src.chat.person_info.relationship_manager import relationship_manager
from src.chat.utils.utils import get_embedding
import time
from typing import Union, Optional
from src.common.database import db
from src.chat.utils.utils import get_recent_group_speaker
from src.manager.mood_manager import mood_manager
from src.chat.memory_system.Hippocampus import HippocampusManager
from src.chat.knowledge.knowledge_lib import qa_manager
from src.chat.focus_chat.expressors.exprssion_learner import expression_learner
import random
logger = get_logger("expressor")
def init_prompt():
Prompt(
"""
你可以参考以下的语言习惯，如果情景合适就使用，不要盲目使用,不要生硬使用，而是结合到表达中：
{style_habbits}
你现在正在群里聊天，以下是群里正在进行的聊天内容：
{chat_info}
以上是聊天内容，你需要了解聊天记录中的内容
{chat_target}
你的名字是{bot_name}，{prompt_personality}，在这聊天中，"{target_message}"引起了你的注意，对这句话，你想表达：{in_mind_reply},原因是：{reason}。你现在要思考怎么回复
你需要使用合适的语法和句法，参考聊天内容，组织一条日常且口语化的回复。
请你根据情景使用以下句法：
{grammar_habbits}
回复尽量简短一些。可以参考贴吧，知乎和微博的回复风格，你可以完全重组回复，保留最基本的表达含义就好，但注意回复要简短，但重组后保持语意通顺。
回复不要浮夸，不要用夸张修辞，平淡一些。不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )，只输出一条回复就好。
现在，你说：
""",
"default_expressor_prompt",
)
Prompt(
"""
你可以参考以下的语言习惯，如果情景合适就使用，不要盲目使用,不要生硬使用，而是结合到表达中：
{style_habbits}
你现在正在群里聊天，以下是群里正在进行的聊天内容：
{chat_info}
以上是聊天内容，你需要了解聊天记录中的内容
{chat_target}
你的名字是{bot_name}，{prompt_personality}，在这聊天中，"{target_message}"引起了你的注意，对这句话，你想表达：{in_mind_reply},原因是：{reason}。你现在要思考怎么回复
你需要使用合适的语法和句法，参考聊天内容，组织一条日常且口语化的回复。
请你根据情景使用以下句法：
{grammar_habbits}
回复尽量简短一些。可以参考贴吧，知乎和微博的回复风格，你可以完全重组回复，保留最基本的表达含义就好，但注意回复要简短，但重组后保持语意通顺。
回复不要浮夸，不要用夸张修辞，平淡一些。不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )，只输出一条回复就好。
现在，你说：
""",
"default_expressor_private_prompt", # New template for private FOCUSED chat
)
class DefaultExpressor:
def __init__(self, chat_id: str):
@@ -106,7 +167,7 @@ class DefaultExpressor:
if reply:
with Timer("发送消息", cycle_timers):
sent_msg_list = await self._send_response_messages(
sent_msg_list = await self.send_response_messages(
anchor_message=anchor_message,
thinking_id=thinking_id,
response_set=reply,
@@ -162,13 +223,10 @@ class DefaultExpressor:
# 3. 构建 Prompt
with Timer("构建Prompt", {}): # 内部计时器，可选保留
prompt = await prompt_builder.build_prompt(
build_mode="focus",
prompt = await self.build_prompt_focus(
chat_stream=self.chat_stream, # Pass the stream object
in_mind_reply=in_mind_reply,
reason=reason,
current_mind_info="",
structured_info="",
sender_name=sender_name_for_prompt, # Pass determined name
target_message=target_message,
)
@@ -223,10 +281,110 @@ class DefaultExpressor:
traceback.print_exc()
return None
async def build_prompt_focus(
self,
reason,
chat_stream,
sender_name,
in_mind_reply,
target_message,
) -> str:
individuality = Individuality.get_instance()
prompt_personality = individuality.get_prompt(x_person=0, level=2)
# Determine if it's a group chat
is_group_chat = bool(chat_stream.group_info)
# Use sender_name passed from caller for private chat, otherwise use a default for group
# Default sender_name for group chat isn't used in the group prompt template, but set for consistency
effective_sender_name = sender_name if not is_group_chat else "某人"
message_list_before_now = get_raw_msg_before_timestamp_with_chat(
chat_id=chat_stream.stream_id,
timestamp=time.time(),
limit=global_config.observation_context_size,
)
chat_talking_prompt = await build_readable_messages(
message_list_before_now,
replace_bot_name=True,
merge_messages=True,
timestamp_mode="relative",
read_mark=0.0,
truncate=True,
)
(
learnt_style_expressions,
learnt_grammar_expressions,
personality_expressions,
) = await expression_learner.get_expression_by_chat_id(chat_stream.stream_id)
style_habbits = []
grammar_habbits = []
# 1. learnt_expressions加权随机选3条
if learnt_style_expressions:
weights = [expr["count"] for expr in learnt_style_expressions]
selected_learnt = weighted_sample_no_replacement(learnt_style_expressions, weights, 3)
for expr in selected_learnt:
if isinstance(expr, dict) and "situation" in expr and "style" in expr:
style_habbits.append(f"当{expr['situation']}时，使用 {expr['style']}")
# 2. learnt_grammar_expressions加权随机选3条
if learnt_grammar_expressions:
weights = [expr["count"] for expr in learnt_grammar_expressions]
selected_learnt = weighted_sample_no_replacement(learnt_grammar_expressions, weights, 3)
for expr in selected_learnt:
if isinstance(expr, dict) and "situation" in expr and "style" in expr:
grammar_habbits.append(f"当{expr['situation']}时，使用 {expr['style']}")
# 3. personality_expressions随机选1条
if personality_expressions:
expr = random.choice(personality_expressions)
if isinstance(expr, dict) and "situation" in expr and "style" in expr:
style_habbits.append(f"当{expr['situation']}时，使用 {expr['style']}")
style_habbits_str = "\n".join(style_habbits)
grammar_habbits_str = "\n".join(grammar_habbits)
logger.debug("开始构建 focus prompt")
# --- Choose template based on chat type ---
if is_group_chat:
template_name = "default_expressor_prompt"
# Group specific formatting variables (already fetched or default)
chat_target_1 = await global_prompt_manager.get_prompt_async("chat_target_group1")
# chat_target_2 = await global_prompt_manager.get_prompt_async("chat_target_group2")
prompt = await global_prompt_manager.format_prompt(
template_name,
style_habbits=style_habbits_str,
grammar_habbits=grammar_habbits_str,
chat_target=chat_target_1,
chat_info=chat_talking_prompt,
bot_name=global_config.BOT_NICKNAME,
prompt_personality="",
reason=reason,
in_mind_reply=in_mind_reply,
target_message=target_message,
)
else: # Private chat
template_name = "default_expressor_private_prompt"
prompt = await global_prompt_manager.format_prompt(
template_name,
sender_name=effective_sender_name, # Used in private template
chat_talking_prompt=chat_talking_prompt,
bot_name=global_config.BOT_NICKNAME,
prompt_personality=prompt_personality,
reason=reason,
moderation_prompt=await global_prompt_manager.get_prompt_async("moderation_prompt"),
)
return prompt
# --- 发送器 (Sender) --- #
async def _send_response_messages(
self, anchor_message: Optional[MessageRecv], response_set: List[Tuple[str, str]], thinking_id: str
async def send_response_messages(
self, anchor_message: Optional[MessageRecv], response_set: List[Tuple[str, str]], thinking_id: str = ""
) -> Optional[MessageSending]:
"""发送回复消息 (尝试锚定到 anchor_message)，使用 HeartFCSender"""
chat = self.chat_stream
@@ -241,7 +399,11 @@ class DefaultExpressor:
stream_name = chat_manager.get_stream_name(chat_id) or chat_id # 获取流名称用于日志
# 检查思考过程是否仍在进行，并获取开始时间
thinking_start_time = await self.heart_fc_sender.get_thinking_start_time(chat_id, thinking_id)
if thinking_id:
thinking_start_time = await self.heart_fc_sender.get_thinking_start_time(chat_id, thinking_id)
else:
thinking_id = "ds"+ str(round(time.time(),2))
thinking_start_time = time.time()
if thinking_start_time is None:
logger.error(f"[{stream_name}]思考过程未找到或已结束，无法发送回复。")
@@ -274,6 +436,7 @@ class DefaultExpressor:
reply_to=reply_to,
is_emoji=is_emoji,
thinking_id=thinking_id,
thinking_start_time=thinking_start_time,
)
try:
@@ -295,6 +458,7 @@ class DefaultExpressor:
except Exception as e:
logger.error(f"{self.log_prefix}发送回复片段 {i} ({part_message_id}) 时失败: {e}")
traceback.print_exc()
# 这里可以选择是继续发送下一个片段还是中止
# 在尝试发送完所有片段后，完成原始的 thinking_id 状态
@@ -325,10 +489,10 @@ class DefaultExpressor:
reply_to: bool,
is_emoji: bool,
thinking_id: str,
thinking_start_time: float,
) -> MessageSending:
"""构建单个发送消息"""
thinking_start_time = await self.heart_fc_sender.get_thinking_start_time(self.chat_id, thinking_id)
bot_user_info = UserInfo(
user_id=global_config.BOT_QQ,
user_nickname=global_config.BOT_NICKNAME,
@@ -348,3 +512,40 @@ class DefaultExpressor:
)
return bot_message
def weighted_sample_no_replacement(items, weights, k) -> list:
"""
加权且不放回地随机抽取k个元素。
参数：
items: 待抽取的元素列表
weights: 每个元素对应的权重（与items等长，且为正数）
k: 需要抽取的元素个数
返回：
selected: 按权重加权且不重复抽取的k个元素组成的列表
如果 items 中的元素不足 k 个，就只会返回所有可用的元素
实现思路：
每次从当前池中按权重加权随机选出一个元素，选中后将其从池中移除，重复k次。
这样保证了：
1. count越大被选中概率越高
2. 不会重复选中同一个元素
"""
selected = []
pool = list(zip(items, weights))
for _ in range(min(k, len(pool))):
total = sum(w for _, w in pool)
r = random.uniform(0, total)
upto = 0
for idx, (item, weight) in enumerate(pool):
upto += weight
if upto >= r:
selected.append(item)
pool.pop(idx)
break
return selected
init_prompt()

53
src/chat/focus_chat/heartFC_chat.py
View File

@@ -14,16 +14,17 @@ from src.chat.focus_chat.heartFC_Cycleinfo import CycleDetail
from src.chat.focus_chat.info.info_base import InfoBase
from src.chat.focus_chat.info_processors.chattinginfo_processor import ChattingInfoProcessor
from src.chat.focus_chat.info_processors.mind_processor import MindProcessor
from src.chat.heart_flow.observation.memory_observation import MemoryObservation
from src.chat.focus_chat.info_processors.working_memory_processor import WorkingMemoryProcessor
from src.chat.heart_flow.observation.hfcloop_observation import HFCloopObservation
from src.chat.heart_flow.observation.working_observation import WorkingObservation
from src.chat.heart_flow.observation.working_observation import WorkingMemoryObservation
from src.chat.heart_flow.observation.chatting_observation import ChattingObservation
from src.chat.focus_chat.info_processors.tool_processor import ToolProcessor
from src.chat.focus_chat.expressors.default_expressor import DefaultExpressor
from src.chat.focus_chat.memory_activator import MemoryActivator
from src.chat.focus_chat.info_processors.base_processor import BaseProcessor
from src.chat.focus_chat.planners.planner import ActionPlanner
from src.chat.focus_chat.planners.action_factory import ActionManager
from src.chat.focus_chat.planners.action_manager import ActionManager
from src.chat.focus_chat.working_memory.working_memory import WorkingMemory
install(extra_lines=3)
@@ -57,7 +58,7 @@ async def _handle_cycle_delay(action_taken_this_cycle: bool, cycle_start_time: f
class HeartFChatting:
"""
管理一个连续的Plan-Replier-Sender循环
管理一个连续的Focus Chat循环
用于在特定聊天流中生成回复。
其生命周期现在由其关联的 SubHeartflow 的 FOCUSED 状态控制。
"""
@@ -79,19 +80,22 @@ class HeartFChatting:
# 基础属性
self.stream_id: str = chat_id # 聊天流ID
self.chat_stream: Optional[ChatStream] = None # 关联的聊天流
self.observations: List[Observation] = observations # 关联的观察列表，用于监控聊天流状态
self.on_consecutive_no_reply_callback = on_consecutive_no_reply_callback
self.log_prefix: str = str(chat_id) # Initial default, will be updated
self.memory_observation = MemoryObservation(observe_id=self.stream_id)
self.hfcloop_observation = HFCloopObservation(observe_id=self.stream_id)
self.working_observation = WorkingObservation(observe_id=self.stream_id)
self.chatting_observation = observations[0]
self.memory_activator = MemoryActivator()
self.working_memory = WorkingMemory(chat_id=self.stream_id)
self.working_observation = WorkingMemoryObservation(observe_id=self.stream_id, working_memory=self.working_memory)
self.expressor = DefaultExpressor(chat_id=self.stream_id)
self.action_manager = ActionManager()
self.action_planner = ActionPlanner(log_prefix=self.log_prefix, action_manager=self.action_manager)
self.hfcloop_observation.set_action_manager(self.action_manager)
self.all_observations = observations
# --- 处理器列表 ---
self.processors: List[BaseProcessor] = []
self._register_default_processors()
@@ -108,9 +112,7 @@ class HeartFChatting:
self._cycle_counter = 0
self._cycle_history: Deque[CycleDetail] = deque(maxlen=10) # 保留最近10个循环的信息
self._current_cycle: Optional[CycleDetail] = None
self.total_no_reply_count: int = 0 # 连续不回复计数器
self._shutting_down: bool = False # 关闭标志位
self.total_waiting_time: float = 0.0 # 累计等待时间
async def _initialize(self) -> bool:
"""
@@ -151,6 +153,7 @@ class HeartFChatting:
self.processors.append(ChattingInfoProcessor())
self.processors.append(MindProcessor(subheartflow_id=self.stream_id))
self.processors.append(ToolProcessor(subheartflow_id=self.stream_id))
self.processors.append(WorkingMemoryProcessor(subheartflow_id=self.stream_id))
logger.info(f"{self.log_prefix} 已注册默认处理器: {[p.__class__.__name__ for p in self.processors]}")
async def start(self):
@@ -349,13 +352,12 @@ class HeartFChatting:
async def _observe_process_plan_action_loop(self, cycle_timers: dict, thinking_id: str) -> tuple[bool, str]:
try:
with Timer("观察", cycle_timers):
await self.observations[0].observe()
await self.memory_observation.observe()
# await self.observations[0].observe()
await self.chatting_observation.observe()
await self.working_observation.observe()
await self.hfcloop_observation.observe()
observations: List[Observation] = []
observations.append(self.observations[0])
observations.append(self.memory_observation)
observations.append(self.chatting_observation)
observations.append(self.working_observation)
observations.append(self.hfcloop_observation)
@@ -363,6 +365,8 @@ class HeartFChatting:
"observations": observations,
}
self.all_observations = observations
with Timer("回忆", cycle_timers):
running_memorys = await self.memory_activator.activate_memory(observations)
@@ -395,8 +399,7 @@ class HeartFChatting:
elif action_type == "no_reply":
action_str = "不回复"
else:
action_type = "unknown"
action_str = "未知动作"
action_str = action_type
logger.info(f"{self.log_prefix} 麦麦决定'{action_str}', 原因'{reasoning}'")
@@ -452,14 +455,14 @@ class HeartFChatting:
reasoning=reasoning,
cycle_timers=cycle_timers,
thinking_id=thinking_id,
observations=self.observations,
observations=self.all_observations,
expressor=self.expressor,
chat_stream=self.chat_stream,
current_cycle=self._current_cycle,
log_prefix=self.log_prefix,
on_consecutive_no_reply_callback=self.on_consecutive_no_reply_callback,
total_no_reply_count=self.total_no_reply_count,
total_waiting_time=self.total_waiting_time,
# total_no_reply_count=self.total_no_reply_count,
# total_waiting_time=self.total_waiting_time,
shutting_down=self._shutting_down,
)
@@ -470,14 +473,6 @@ class HeartFChatting:
# 处理动作并获取结果
success, reply_text = await action_handler.handle_action()
# 更新状态计数器
if action == "no_reply":
self.total_no_reply_count = getattr(action_handler, "total_no_reply_count", self.total_no_reply_count)
self.total_waiting_time = getattr(action_handler, "total_waiting_time", self.total_waiting_time)
elif action == "reply":
self.total_no_reply_count = 0
self.total_waiting_time = 0.0
return success, reply_text
except Exception as e:
@@ -526,5 +521,3 @@ class HeartFChatting:
if last_n is not None:
history = history[-last_n:]
return [cycle.to_dict() for cycle in history]

1
src/chat/focus_chat/heartFC_sender.py
View File

@@ -106,6 +106,7 @@ class HeartFCSender:
and not message.is_private_message()
and message.reply.processed_plain_text != "[System Trigger Context]"
):
message.set_reply(message.reply)
logger.debug(f"[{chat_id}] 应用 set_reply 逻辑: {message.processed_plain_text[:20]}...")
await message.process()

205
src/chat/focus_chat/heartflow_prompt_builder.py
View File

@@ -6,14 +6,13 @@ from src.chat.utils.chat_message_builder import build_readable_messages, get_raw
from src.chat.person_info.relationship_manager import relationship_manager
from src.chat.utils.utils import get_embedding
import time
from typing import Union, Optional, Dict, Any
from typing import Union, Optional
from src.common.database import db
from src.chat.utils.utils import get_recent_group_speaker
from src.manager.mood_manager import mood_manager
from src.chat.memory_system.Hippocampus import HippocampusManager
from src.chat.knowledge.knowledge_lib import qa_manager
from src.chat.focus_chat.expressors.exprssion_learner import expression_learner
import traceback
import random
@@ -21,27 +20,6 @@ logger = get_logger("prompt")
def init_prompt():
Prompt(
"""
你可以参考以下的语言习惯，如果情景合适就使用，不要盲目使用,不要生硬使用，而是结合到表达中：
{style_habbits}
你现在正在群里聊天，以下是群里正在进行的聊天内容：
{chat_info}
以上是聊天内容，你需要了解聊天记录中的内容
{chat_target}
你的名字是{bot_name}，{prompt_personality}，在这聊天中，"{target_message}"引起了你的注意，对这句话，你想表达：{in_mind_reply},原因是：{reason}。你现在要思考怎么回复
你需要使用合适的语法和句法，参考聊天内容，组织一条日常且口语化的回复。
请你根据情景使用以下句法：
{grammar_habbits}
回复尽量简短一些。可以参考贴吧，知乎和微博的回复风格，你可以完全重组回复，保留最基本的表达含义就好，但注意回复要简短，但重组后保持语意通顺。
回复不要浮夸，不要用夸张修辞，平淡一些。不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )，只输出一条回复就好。
现在，你说：
""",
"heart_flow_prompt",
)
Prompt(
"""
@@ -82,29 +60,6 @@ def init_prompt():
Prompt("\n你有以下这些**知识**：\n{prompt_info}\n请你**记住上面的知识**，之后可能会用到。\n", "knowledge_prompt")
# --- Template for HeartFChatting (FOCUSED mode) ---
Prompt(
"""
{info_from_tools}
你正在和 {sender_name} 私聊。
聊天记录如下：
{chat_talking_prompt}
现在你想要回复。
你需要扮演一位网名叫{bot_name}的人进行回复，这个人的特点是："{prompt_personality}"。
你正在和 {sender_name} 私聊, 现在请你读读你们之前的聊天记录，然后给出日常且口语化的回复，平淡一些。
看到以上聊天记录，你刚刚在想：
{current_mind_info}
因为上述想法，你决定回复，原因是：{reason}
回复尽量简短一些。请注意把握聊天内容，{reply_style2}。{prompt_ger}，不要复读自己说的话
{reply_style1}，说中文，不要刻意突出自身学科背景，注意只输出回复内容。
{moderation_prompt}。注意：回复不要输出多余内容(包括前后缀，冒号和引号，括号，表情包，at或 @等 )。""",
"heart_flow_private_prompt", # New template for private FOCUSED chat
)
# --- Template for NormalChat (CHAT mode) ---
Prompt(
"""
{memory_prompt}
@@ -126,118 +81,6 @@ def init_prompt():
)
async def _build_prompt_focus(
reason, current_mind_info, structured_info, chat_stream, sender_name, in_mind_reply, target_message
) -> str:
individuality = Individuality.get_instance()
prompt_personality = individuality.get_prompt(x_person=0, level=2)
# Determine if it's a group chat
is_group_chat = bool(chat_stream.group_info)
# Use sender_name passed from caller for private chat, otherwise use a default for group
# Default sender_name for group chat isn't used in the group prompt template, but set for consistency
effective_sender_name = sender_name if not is_group_chat else "某人"
message_list_before_now = get_raw_msg_before_timestamp_with_chat(
chat_id=chat_stream.stream_id,
timestamp=time.time(),
limit=global_config.observation_context_size,
)
chat_talking_prompt = await build_readable_messages(
message_list_before_now,
replace_bot_name=True,
merge_messages=True,
timestamp_mode="relative",
read_mark=0.0,
truncate=True,
)
if structured_info:
structured_info_prompt = await global_prompt_manager.format_prompt(
"info_from_tools", structured_info=structured_info
)
else:
structured_info_prompt = ""
# 从/data/expression/对应chat_id/expressions.json中读取表达方式
(
learnt_style_expressions,
learnt_grammar_expressions,
personality_expressions,
) = await expression_learner.get_expression_by_chat_id(chat_stream.stream_id)
style_habbits = []
grammar_habbits = []
# 1. learnt_expressions加权随机选3条
if learnt_style_expressions:
weights = [expr["count"] for expr in learnt_style_expressions]
selected_learnt = weighted_sample_no_replacement(learnt_style_expressions, weights, 3)
for expr in selected_learnt:
if isinstance(expr, dict) and "situation" in expr and "style" in expr:
style_habbits.append(f"当{expr['situation']}时，使用 {expr['style']}")
# 2. learnt_grammar_expressions加权随机选3条
if learnt_grammar_expressions:
weights = [expr["count"] for expr in learnt_grammar_expressions]
selected_learnt = weighted_sample_no_replacement(learnt_grammar_expressions, weights, 3)
for expr in selected_learnt:
if isinstance(expr, dict) and "situation" in expr and "style" in expr:
grammar_habbits.append(f"当{expr['situation']}时，使用 {expr['style']}")
# 3. personality_expressions随机选1条
if personality_expressions:
expr = random.choice(personality_expressions)
if isinstance(expr, dict) and "situation" in expr and "style" in expr:
style_habbits.append(f"当{expr['situation']}时，使用 {expr['style']}")
style_habbits_str = "\n".join(style_habbits)
grammar_habbits_str = "\n".join(grammar_habbits)
logger.debug("开始构建 focus prompt")
# --- Choose template based on chat type ---
if is_group_chat:
template_name = "heart_flow_prompt"
# Group specific formatting variables (already fetched or default)
chat_target_1 = await global_prompt_manager.get_prompt_async("chat_target_group1")
# chat_target_2 = await global_prompt_manager.get_prompt_async("chat_target_group2")
prompt = await global_prompt_manager.format_prompt(
template_name,
# info_from_tools=structured_info_prompt,
style_habbits=style_habbits_str,
grammar_habbits=grammar_habbits_str,
chat_target=chat_target_1, # Used in group template
# chat_talking_prompt=chat_talking_prompt,
chat_info=chat_talking_prompt,
bot_name=global_config.BOT_NICKNAME,
# prompt_personality=prompt_personality,
prompt_personality="",
reason=reason,
in_mind_reply=in_mind_reply,
target_message=target_message,
# moderation_prompt=await global_prompt_manager.get_prompt_async("moderation_prompt"),
# sender_name is not used in the group template
)
else: # Private chat
template_name = "heart_flow_private_prompt"
prompt = await global_prompt_manager.format_prompt(
template_name,
info_from_tools=structured_info_prompt,
sender_name=effective_sender_name, # Used in private template
chat_talking_prompt=chat_talking_prompt,
bot_name=global_config.BOT_NICKNAME,
prompt_personality=prompt_personality,
# chat_target and chat_target_2 are not used in private template
current_mind_info=current_mind_info,
reason=reason,
moderation_prompt=await global_prompt_manager.get_prompt_async("moderation_prompt"),
)
# --- End choosing template ---
# logger.debug(f"focus_chat_prompt (is_group={is_group_chat}): \n{prompt}")
return prompt
class PromptBuilder:
def __init__(self):
self.prompt_built = ""
@@ -257,17 +100,6 @@ class PromptBuilder:
) -> Optional[str]:
if build_mode == "normal":
return await self._build_prompt_normal(chat_stream, message_txt or "", sender_name)
elif build_mode == "focus":
return await _build_prompt_focus(
reason,
current_mind_info,
structured_info,
chat_stream,
sender_name,
in_mind_reply,
target_message,
)
return None
async def _build_prompt_normal(self, chat_stream, message_txt: str, sender_name: str = "某人") -> str:
@@ -689,40 +521,5 @@ class PromptBuilder:
# 返回所有找到的内容，用换行分隔
return "\n".join(str(result["content"]) for result in results)
def weighted_sample_no_replacement(items, weights, k) -> list:
"""
加权且不放回地随机抽取k个元素。
参数：
items: 待抽取的元素列表
weights: 每个元素对应的权重（与items等长，且为正数）
k: 需要抽取的元素个数
返回：
selected: 按权重加权且不重复抽取的k个元素组成的列表
如果 items 中的元素不足 k 个，就只会返回所有可用的元素
实现思路：
每次从当前池中按权重加权随机选出一个元素，选中后将其从池中移除，重复k次。
这样保证了：
1. count越大被选中概率越高
2. 不会重复选中同一个元素
"""
selected = []
pool = list(zip(items, weights))
for _ in range(min(k, len(pool))):
total = sum(w for _, w in pool)
r = random.uniform(0, total)
upto = 0
for idx, (item, weight) in enumerate(pool):
upto += weight
if upto >= r:
selected.append(item)
pool.pop(idx)
break
return selected
init_prompt()
prompt_builder = PromptBuilder()

9
src/chat/focus_chat/info/info_base.py
View File

@@ -17,6 +17,7 @@ class InfoBase:
type: str = "base"
data: Dict[str, Any] = field(default_factory=dict)
processed_info:str = ""
def get_type(self) -> str:
"""获取信息类型
@@ -58,3 +59,11 @@ class InfoBase:
if isinstance(value, list):
return value
return []
def get_processed_info(self) -> str:
"""获取处理后的信息
Returns:
str: 处理后的信息字符串
"""
return self.processed_info

2
src/chat/focus_chat/info_processors/chattinginfo_processor.py
View File

@@ -54,6 +54,8 @@ class ChattingInfoProcessor(BaseProcessor):
for obs in observations:
# print(f"obs: {obs}")
if isinstance(obs, ChattingObservation):
# print("1111111111111111111111读取111111111111111")
obs_info = ObsInfo()
await self.chat_compress(obs)

204
src/chat/focus_chat/info_processors/mind_processor.py
View File

@@ -16,11 +16,6 @@ from .base_processor import BaseProcessor
from src.chat.focus_chat.info.mind_info import MindInfo
from typing import List, Optional
from src.chat.heart_flow.observation.hfcloop_observation import HFCloopObservation
from src.chat.focus_chat.info_processors.processor_utils import (
calculate_similarity,
calculate_replacement_probability,
get_spark,
)
from typing import Dict
from src.chat.focus_chat.info.info_base import InfoBase
@@ -28,7 +23,6 @@ logger = get_logger("processor")
def init_prompt():
# --- Group Chat Prompt ---
group_prompt = """
你的名字是{bot_name}
{memory_str}
@@ -44,31 +38,29 @@ def init_prompt():
现在请你继续输出观察和规划，输出要求：
1. 先关注未读新消息的内容和近期回复历史
2. 根据新信息，修改和删除之前的观察和规划
3. 根据聊天内容继续输出观察和规划，{hf_do_next}
3. 根据聊天内容继续输出观察和规划
4. 注意群聊的时间线索，话题由谁发起，进展状况如何，思考聊天的时间线。
6. 语言简洁自然，不要分点，不要浮夸，不要修辞，仅输出思考内容就好"""
Prompt(group_prompt, "sub_heartflow_prompt_before")
# --- Private Chat Prompt ---
private_prompt = """
你的名字是{bot_name}
{memory_str}
{extra_info}
{relation_prompt}
你的名字是{bot_name},{prompt_personality},你现在{mood_info}
{cycle_info_block}
现在是{time_now}，你正在上网，和 {chat_target_name} 私聊，以下是你们的聊天内容：
现在是{time_now}，你正在上网，和qq群里的网友们聊天，以下是正在进行的聊天内容：
{chat_observe_info}
以下是你之前对聊天的观察和规划：
以下是你之前对聊天的观察和规划，你的名字是{bot_name}：
{last_mind}
请仔细阅读聊天内容，想想你和 {chat_target_name} 的关系，回顾你们刚刚的交流,你刚刚发言和对方的反应，思考聊天的主题。
请思考你要不要回复以及如何回复对方。
思考并输出你的内心想法
输出要求：
1. 根据聊天内容生成你的想法，{hf_do_next}
2. 不要分点、不要使用表情符号
3. 避免多余符号(冒号、引号、括号等)
4. 语言简洁自然，不要浮夸
5. 如果你刚发言，对方没有回复你，请谨慎回复"""
现在请你继续输出观察和规划，输出要求：
1. 先关注未读新消息的内容和近期回复历史
2. 根据新信息，修改和删除之前的观察和规划
3. 根据聊天内容继续输出观察和规划
4. 注意群聊的时间线索，话题由谁发起，进展状况如何，思考聊天的时间线。
6. 语言简洁自然，不要分点，不要浮夸，不要修辞，仅输出思考内容就好"""
Prompt(private_prompt, "sub_heartflow_prompt_private_before")
@@ -210,45 +202,28 @@ class MindProcessor(BaseProcessor):
for person in person_list:
relation_prompt += await relationship_manager.build_relationship_info(person, is_id=True)
# 构建个性部分
# prompt_personality = individuality.get_prompt(x_person=2, level=2)
# 获取当前时间
time_now = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
spark_prompt = get_spark()
# ---------- 5. 构建最终提示词 ----------
template_name = "sub_heartflow_prompt_before" if is_group_chat else "sub_heartflow_prompt_private_before"
logger.debug(f"{self.log_prefix} 使用{'群聊' if is_group_chat else '私聊'}思考模板")
prompt = (await global_prompt_manager.get_prompt_async(template_name)).format(
bot_name=individuality.name,
memory_str=memory_str,
extra_info=self.structured_info_str,
# prompt_personality=prompt_personality,
relation_prompt=relation_prompt,
bot_name=individuality.name,
time_now=time_now,
time_now=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()),
chat_observe_info=chat_observe_info,
# mood_info="mood_info",
hf_do_next=spark_prompt,
last_mind=previous_mind,
cycle_info_block=hfcloop_observe_info,
chat_target_name=chat_target_name,
)
# 在构建完提示词后，生成最终的prompt字符串
final_prompt = prompt
content = "" # 初始化内容变量
content = "(不知道该想些什么...)"
try:
# 调用LLM生成响应
response, _ = await self.llm_model.generate_response_async(prompt=final_prompt)
# 直接使用LLM返回的文本响应作为 content
content = response if response else ""
content, _ = await self.llm_model.generate_response_async(prompt=prompt)
if not content:
logger.warning(f"{self.log_prefix} LLM返回空结果，思考失败。")
except Exception as e:
# 处理总体异常
logger.error(f"{self.log_prefix} 执行LLM请求或处理响应时出错: {e}")
@@ -256,16 +231,8 @@ class MindProcessor(BaseProcessor):
content = "思考过程中出现错误"
# 记录初步思考结果
logger.debug(f"{self.log_prefix} 思考prompt: \n{final_prompt}\n")
# 处理空响应情况
if not content:
content = "(不知道该想些什么...)"
logger.warning(f"{self.log_prefix} LLM返回空结果，思考失败。")
# ---------- 8. 更新思考状态并返回结果 ----------
logger.debug(f"{self.log_prefix} 思考prompt: \n{prompt}\n")
logger.info(f"{self.log_prefix} 思考结果: {content}")
# 更新当前思考内容
self.update_current_mind(content)
return content
@@ -275,138 +242,5 @@ class MindProcessor(BaseProcessor):
self.past_mind.append(self.current_mind)
self.current_mind = response
def de_similar(self, previous_mind, new_content):
try:
similarity = calculate_similarity(previous_mind, new_content)
replacement_prob = calculate_replacement_probability(similarity)
logger.debug(f"{self.log_prefix} 新旧想法相似度: {similarity:.2f}, 替换概率: {replacement_prob:.2f}")
# 定义词语列表 (移到判断之前)
yu_qi_ci_liebiao = ["嗯", "哦", "啊", "唉", "哈", "唔"]
zhuan_zhe_liebiao = ["但是", "不过", "然而", "可是", "只是"]
cheng_jie_liebiao = ["然后", "接着", "此外", "而且", "另外"]
zhuan_jie_ci_liebiao = zhuan_zhe_liebiao + cheng_jie_liebiao
if random.random() < replacement_prob:
# 相似度非常高时，尝试去重或特殊处理
if similarity == 1.0:
logger.debug(f"{self.log_prefix} 想法完全重复 (相似度 1.0)，执行特殊处理...")
# 随机截取大约一半内容
if len(new_content) > 1: # 避免内容过短无法截取
split_point = max(
1, len(new_content) // 2 + random.randint(-len(new_content) // 4, len(new_content) // 4)
)
truncated_content = new_content[:split_point]
else:
truncated_content = new_content # 如果只有一个字符或者为空，就不截取了
# 添加语气词和转折/承接词
yu_qi_ci = random.choice(yu_qi_ci_liebiao)
zhuan_jie_ci = random.choice(zhuan_jie_ci_liebiao)
content = f"{yu_qi_ci}{zhuan_jie_ci}，{truncated_content}"
logger.debug(f"{self.log_prefix} 想法重复，特殊处理后: {content}")
else:
# 相似度较高但非100%，执行标准去重逻辑
logger.debug(f"{self.log_prefix} 执行概率性去重 (概率: {replacement_prob:.2f})...")
logger.debug(
f"{self.log_prefix} previous_mind类型: {type(previous_mind)}, new_content类型: {type(new_content)}"
)
matcher = difflib.SequenceMatcher(None, previous_mind, new_content)
logger.debug(f"{self.log_prefix} matcher类型: {type(matcher)}")
deduplicated_parts = []
last_match_end_in_b = 0
# 获取并记录所有匹配块
matching_blocks = matcher.get_matching_blocks()
logger.debug(f"{self.log_prefix} 匹配块数量: {len(matching_blocks)}")
logger.debug(
f"{self.log_prefix} 匹配块示例(前3个): {matching_blocks[:3] if len(matching_blocks) > 3 else matching_blocks}"
)
# get_matching_blocks()返回形如[(i, j, n), ...]的列表，其中i是a中的索引，j是b中的索引，n是匹配的长度
for idx, match in enumerate(matching_blocks):
if not isinstance(match, tuple):
logger.error(f"{self.log_prefix} 匹配块 {idx} 不是元组类型，而是 {type(match)}: {match}")
continue
try:
_i, j, n = match # 解包元组为三个变量
logger.debug(f"{self.log_prefix} 匹配块 {idx}: i={_i}, j={j}, n={n}")
if last_match_end_in_b < j:
# 确保添加的是字符串，而不是元组
try:
non_matching_part = new_content[last_match_end_in_b:j]
logger.debug(
f"{self.log_prefix} 添加非匹配部分: '{non_matching_part}', 类型: {type(non_matching_part)}"
)
if not isinstance(non_matching_part, str):
logger.warning(
f"{self.log_prefix} 非匹配部分不是字符串类型: {type(non_matching_part)}"
)
non_matching_part = str(non_matching_part)
deduplicated_parts.append(non_matching_part)
except Exception as e:
logger.error(f"{self.log_prefix} 处理非匹配部分时出错: {e}")
logger.error(traceback.format_exc())
last_match_end_in_b = j + n
except Exception as e:
logger.error(f"{self.log_prefix} 处理匹配块时出错: {e}")
logger.error(traceback.format_exc())
logger.debug(f"{self.log_prefix} 去重前部分列表: {deduplicated_parts}")
logger.debug(f"{self.log_prefix} 列表元素类型: {[type(part) for part in deduplicated_parts]}")
# 确保所有元素都是字符串
deduplicated_parts = [str(part) for part in deduplicated_parts]
# 防止列表为空
if not deduplicated_parts:
logger.warning(f"{self.log_prefix} 去重后列表为空，添加空字符串")
deduplicated_parts = [""]
logger.debug(f"{self.log_prefix} 处理后的部分列表: {deduplicated_parts}")
try:
deduplicated_content = "".join(deduplicated_parts).strip()
logger.debug(f"{self.log_prefix} 拼接后的去重内容: '{deduplicated_content}'")
except Exception as e:
logger.error(f"{self.log_prefix} 拼接去重内容时出错: {e}")
logger.error(traceback.format_exc())
deduplicated_content = ""
if deduplicated_content:
# 根据概率决定是否添加词语
prefix_str = ""
if random.random() < 0.3: # 30% 概率添加语气词
prefix_str += random.choice(yu_qi_ci_liebiao)
if random.random() < 0.7: # 70% 概率添加转折/承接词
prefix_str += random.choice(zhuan_jie_ci_liebiao)
# 组合最终结果
if prefix_str:
content = f"{prefix_str}，{deduplicated_content}" # 更新 content
logger.debug(f"{self.log_prefix} 去重并添加引导词后: {content}")
else:
content = deduplicated_content # 更新 content
logger.debug(f"{self.log_prefix} 去重后 (未添加引导词): {content}")
else:
logger.warning(f"{self.log_prefix} 去重后内容为空，保留原始LLM输出: {new_content}")
content = new_content # 保留原始 content
else:
logger.debug(f"{self.log_prefix} 未执行概率性去重 (概率: {replacement_prob:.2f})")
# content 保持 new_content 不变
except Exception as e:
logger.error(f"{self.log_prefix} 应用概率性去重或特殊处理时出错: {e}")
logger.error(traceback.format_exc())
# 出错时保留原始 content
content = new_content
return content
init_prompt()

56
src/chat/focus_chat/info_processors/processor_utils.py
View File

@@ -1,56 +0,0 @@
import difflib
import random
import time
def calculate_similarity(text_a: str, text_b: str) -> float:
"""
计算两个文本字符串的相似度。
"""
if not text_a or not text_b:
return 0.0
matcher = difflib.SequenceMatcher(None, text_a, text_b)
return matcher.ratio()
def calculate_replacement_probability(similarity: float) -> float:
"""
根据相似度计算替换的概率。
规则：
- 相似度 <= 0.4: 概率 = 0
- 相似度 >= 0.9: 概率 = 1
- 相似度 == 0.6: 概率 = 0.7
- 0.4 < 相似度 <= 0.6: 线性插值 (0.4, 0) 到 (0.6, 0.7)
- 0.6 < 相似度 < 0.9: 线性插值 (0.6, 0.7) 到 (0.9, 1.0)
"""
if similarity <= 0.4:
return 0.0
elif similarity >= 0.9:
return 1.0
elif 0.4 < similarity <= 0.6:
# p = 3.5 * s - 1.4
probability = 3.5 * similarity - 1.4
return max(0.0, probability)
else: # 0.6 < similarity < 0.9
# p = s + 0.1
probability = similarity + 0.1
return min(1.0, max(0.0, probability))
def get_spark():
local_random = random.Random()
current_minute = int(time.strftime("%M"))
local_random.seed(current_minute)
hf_options = [
("可以参考之前的想法，在原来想法的基础上继续思考", 0.2),
("可以参考之前的想法，在原来的想法上尝试新的话题", 0.4),
("不要太深入", 0.2),
("进行深入思考", 0.2),
]
# 加权随机选择思考指导
hf_do_next = local_random.choices(
[option[0] for option in hf_options], weights=[option[1] for option in hf_options], k=1
)[0]
return hf_do_next

2
src/chat/focus_chat/info_processors/tool_processor.py
View File

@@ -155,7 +155,7 @@ class ToolProcessor(BaseProcessor):
)
# 调用LLM，专注于工具使用
logger.debug(f"开始执行工具调用{prompt}")
# logger.debug(f"开始执行工具调用{prompt}")
response, _, tool_calls = await self.llm_model.generate_response_tool_async(prompt=prompt, tools=tools)
logger.debug(f"获取到工具原始输出:\n{tool_calls}")

112
src/chat/focus_chat/planners/action_factory.py → src/chat/focus_chat/planners/action_manager.py
View File

@@ -1,18 +1,18 @@
from typing import Dict, List, Optional, Callable, Coroutine, Type, Any, Union
import os
import importlib
from src.chat.focus_chat.planners.actions.base_action import BaseAction, _ACTION_REGISTRY, _DEFAULT_ACTIONS
from typing import Dict, List, Optional, Callable, Coroutine, Type, Any
from src.chat.focus_chat.planners.actions.base_action import BaseAction, _ACTION_REGISTRY
from src.chat.heart_flow.observation.observation import Observation
from src.chat.focus_chat.expressors.default_expressor import DefaultExpressor
from src.chat.message_receive.chat_stream import ChatStream
from src.chat.focus_chat.heartFC_Cycleinfo import CycleDetail
from src.common.logger_manager import get_logger
import importlib
import pkgutil
import os
# 导入动作类，确保装饰器被执行
from src.chat.focus_chat.planners.actions.reply_action import ReplyAction
from src.chat.focus_chat.planners.actions.no_reply_action import NoReplyAction
import src.chat.focus_chat.planners.actions # noqa
logger = get_logger("action_factory")
logger = get_logger("action_manager")
# 定义动作信息类型
ActionInfo = Dict[str, Any]
@@ -38,14 +38,12 @@ class ActionManager:
# 加载所有已注册动作
self._load_registered_actions()
# 加载插件动作
self._load_plugin_actions()
# 初始化时将默认动作加载到使用中的动作
self._using_actions = self._default_actions.copy()
# logger.info(f"当前可用动作: {list(self._using_actions.keys())}")
# for action_name, action_info in self._using_actions.items():
# logger.info(f"动作名称: {action_name}, 动作信息: {action_info}")
def _load_registered_actions(self) -> None:
"""
加载所有通过装饰器注册的动作
@@ -54,23 +52,24 @@ class ActionManager:
# 从_ACTION_REGISTRY获取所有已注册动作
for action_name, action_class in _ACTION_REGISTRY.items():
# 获取动作相关信息
action_description:str = getattr(action_class, "action_description", "")
action_parameters:dict[str:str] = getattr(action_class, "action_parameters", {})
action_require:list[str] = getattr(action_class, "action_require", [])
is_default:bool = getattr(action_class, "default", False)
# 不读取插件动作和基类
if action_name == "base_action" or action_name == "plugin_action":
continue
action_description: str = getattr(action_class, "action_description", "")
action_parameters: dict[str:str] = getattr(action_class, "action_parameters", {})
action_require: list[str] = getattr(action_class, "action_require", [])
is_default: bool = getattr(action_class, "default", False)
if action_name and action_description:
# 创建动作信息字典
action_info = {
"description": action_description,
"parameters": action_parameters,
"require": action_require
"require": action_require,
}
# 注册2
print("注册2")
print(action_info)
# 添加到所有已注册的动作
self._registered_actions[action_name] = action_info
@@ -80,12 +79,52 @@ class ActionManager:
logger.info(f"所有注册动作: {list(self._registered_actions.keys())}")
logger.info(f"默认动作: {list(self._default_actions.keys())}")
# for action_name, action_info in self._default_actions.items():
# logger.info(f"动作名称: {action_name}, 动作信息: {action_info}")
for action_name, action_info in self._default_actions.items():
logger.info(f"动作名称: {action_name}, 动作信息: {action_info}")
except Exception as e:
logger.error(f"加载已注册动作失败: {e}")
def _load_plugin_actions(self) -> None:
"""
加载所有插件目录中的动作
"""
try:
# 检查插件目录是否存在
plugin_path = "src.plugins"
plugin_dir = plugin_path.replace('.', os.path.sep)
if not os.path.exists(plugin_dir):
logger.info(f"插件目录 {plugin_dir} 不存在，跳过插件动作加载")
return
# 导入插件包
try:
plugins_package = importlib.import_module(plugin_path)
except ImportError as e:
logger.error(f"导入插件包失败: {e}")
return
# 遍历插件包中的所有子包
for _, plugin_name, is_pkg in pkgutil.iter_modules(plugins_package.__path__, plugins_package.__name__ + '.'):
if not is_pkg:
continue
# 检查插件是否有actions子包
plugin_actions_path = f"{plugin_name}.actions"
try:
# 尝试导入插件的actions包
importlib.import_module(plugin_actions_path)
logger.info(f"成功加载插件动作模块: {plugin_actions_path}")
except ImportError as e:
logger.debug(f"插件 {plugin_name} 没有actions子包或导入失败: {e}")
continue
# 再次从_ACTION_REGISTRY获取所有动作（包括刚刚从插件加载的）
self._load_registered_actions()
except Exception as e:
logger.error(f"加载插件动作失败: {e}")
def create_action(
self,
action_name: str,
@@ -99,8 +138,8 @@ class ActionManager:
current_cycle: CycleDetail,
log_prefix: str,
on_consecutive_no_reply_callback: Callable[[], Coroutine[None, None, None]],
total_no_reply_count: int = 0,
total_waiting_time: float = 0.0,
# total_no_reply_count: int = 0,
# total_waiting_time: float = 0.0,
shutting_down: bool = False,
) -> Optional[BaseAction]:
"""
@@ -136,7 +175,7 @@ class ActionManager:
return None
try:
# 创建动作实例并传递所有必要参数
# 创建动作实例
instance = handler_class(
action_name=action_name,
action_data=action_data,
@@ -144,14 +183,14 @@ class ActionManager:
cycle_timers=cycle_timers,
thinking_id=thinking_id,
observations=observations,
on_consecutive_no_reply_callback=on_consecutive_no_reply_callback,
current_cycle=current_cycle,
log_prefix=log_prefix,
total_no_reply_count=total_no_reply_count,
total_waiting_time=total_waiting_time,
shutting_down=shutting_down,
expressor=expressor,
chat_stream=chat_stream,
current_cycle=current_cycle,
log_prefix=log_prefix,
on_consecutive_no_reply_callback=on_consecutive_no_reply_callback,
# total_no_reply_count=total_no_reply_count,
# total_waiting_time=total_waiting_time,
shutting_down=shutting_down,
)
return instance
@@ -233,11 +272,7 @@ class ActionManager:
if require is None:
require = []
action_info = {
"description": description,
"parameters": parameters,
"require": require
}
action_info = {"description": description, "parameters": parameters, "require": require}
self._registered_actions[action_name] = action_info
return True
@@ -282,6 +317,3 @@ class ActionManager:
"""
return _ACTION_REGISTRY.get(action_name)
# 创建全局实例
ActionFactory = ActionManager()

5
src/chat/focus_chat/planners/actions/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
# 导入所有动作模块以确保装饰器被执行
from . import reply_action # noqa
from . import no_reply_action # noqa
# 在此处添加更多动作模块导入

17
src/chat/focus_chat/planners/actions/base_action.py
View File

@@ -25,8 +25,8 @@ def register_action(cls):
logger.error(f"动作类 {cls.__name__} 缺少必要的属性: action_name 或 action_description")
return cls
action_name = getattr(cls, "action_name")
action_description = getattr(cls, "action_description")
action_name = cls.action_name
action_description = cls.action_description
is_default = getattr(cls, "default", False)
if not action_name or not action_description:
@@ -60,14 +60,13 @@ class BaseAction(ABC):
cycle_timers: 计时器字典
thinking_id: 思考ID
"""
#每个动作必须实现
self.action_name:str = "base_action"
self.action_description:str = "基础动作"
self.action_parameters:dict = {}
self.action_require:list[str] = []
self.default:bool = False
# 每个动作必须实现
self.action_name: str = "base_action"
self.action_description: str = "基础动作"
self.action_parameters: dict = {}
self.action_require: list[str] = []
self.default: bool = False
self.action_data = action_data
self.reasoning = reasoning

40
src/chat/focus_chat/planners/actions/no_reply_action.py
View File

@@ -29,7 +29,7 @@ class NoReplyAction(BaseAction):
action_require = [
"话题无关/无聊/不感兴趣/不懂",
"最后一条消息是你自己发的且无人回应你",
"你发送了太多消息，且无人回复"
"你发送了太多消息，且无人回复",
]
default = True
@@ -43,10 +43,10 @@ class NoReplyAction(BaseAction):
on_consecutive_no_reply_callback: Callable[[], Coroutine[None, None, None]],
current_cycle: CycleDetail,
log_prefix: str,
total_no_reply_count: int = 0,
total_waiting_time: float = 0.0,
# total_no_reply_count: int = 0,
# total_waiting_time: float = 0.0,
shutting_down: bool = False,
**kwargs
**kwargs,
):
"""初始化不回复动作处理器
@@ -69,8 +69,8 @@ class NoReplyAction(BaseAction):
self.on_consecutive_no_reply_callback = on_consecutive_no_reply_callback
self._current_cycle = current_cycle
self.log_prefix = log_prefix
self.total_no_reply_count = total_no_reply_count
self.total_waiting_time = total_waiting_time
# self.total_no_reply_count = total_no_reply_count
# self.total_waiting_time = total_waiting_time
self._shutting_down = shutting_down
async def handle_action(self) -> Tuple[bool, str]:
@@ -96,34 +96,6 @@ class NoReplyAction(BaseAction):
# 从计时器获取实际等待时间
current_waiting = self.cycle_timers.get("等待新消息", 0.0)
if not self._shutting_down:
self.total_no_reply_count += 1
self.total_waiting_time += current_waiting # 累加等待时间
logger.debug(
f"{self.log_prefix} 连续不回复计数增加: {self.total_no_reply_count}/{CONSECUTIVE_NO_REPLY_THRESHOLD}, "
f"本次等待: {current_waiting:.2f}秒, 累计等待: {self.total_waiting_time:.2f}秒"
)
# 检查是否同时达到次数和时间阈值
time_threshold = 0.66 * WAITING_TIME_THRESHOLD * CONSECUTIVE_NO_REPLY_THRESHOLD
if (
self.total_no_reply_count >= CONSECUTIVE_NO_REPLY_THRESHOLD
and self.total_waiting_time >= time_threshold
):
logger.info(
f"{self.log_prefix} 连续不回复达到阈值 ({self.total_no_reply_count}次) "
f"且累计等待时间达到 {self.total_waiting_time:.2f}秒 (阈值 {time_threshold}秒)，"
f"调用回调请求状态转换"
)
# 调用回调。注意：这里不重置计数器和时间，依赖回调函数成功改变状态来隐式重置上下文。
await self.on_consecutive_no_reply_callback()
elif self.total_no_reply_count >= CONSECUTIVE_NO_REPLY_THRESHOLD:
# 仅次数达到阈值，但时间未达到
logger.debug(
f"{self.log_prefix} 连续不回复次数达到阈值 ({self.total_no_reply_count}次) "
f"但累计等待时间 {self.total_waiting_time:.2f}秒 未达到时间阈值 ({time_threshold}秒)，暂不调用回调"
)
# else: 次数和时间都未达到阈值，不做处理
return True, "" # 不回复动作没有回复文本

215
src/chat/focus_chat/planners/actions/plugin_action.py Normal file
View File

@@ -0,0 +1,215 @@
import traceback
from typing import Tuple, Dict, List, Any, Optional
from src.chat.focus_chat.planners.actions.base_action import BaseAction, register_action
from src.chat.heart_flow.observation.chatting_observation import ChattingObservation
from src.chat.focus_chat.hfc_utils import create_empty_anchor_message
from src.common.logger_manager import get_logger
from src.chat.person_info.person_info import person_info_manager
from abc import abstractmethod
logger = get_logger("plugin_action")
class PluginAction(BaseAction):
"""插件动作基类
封装了主程序内部依赖，提供简化的API接口给插件开发者
"""
def __init__(self, action_data: dict, reasoning: str, cycle_timers: dict, thinking_id: str, **kwargs):
"""初始化插件动作基类"""
super().__init__(action_data, reasoning, cycle_timers, thinking_id)
# 存储内部服务和对象引用
self._services = {}
# 从kwargs提取必要的内部服务
if "observations" in kwargs:
self._services["observations"] = kwargs["observations"]
if "expressor" in kwargs:
self._services["expressor"] = kwargs["expressor"]
if "chat_stream" in kwargs:
self._services["chat_stream"] = kwargs["chat_stream"]
if "current_cycle" in kwargs:
self._services["current_cycle"] = kwargs["current_cycle"]
self.log_prefix = kwargs.get("log_prefix", "")
async def get_user_id_by_person_name(self, person_name: str) -> Tuple[str, str]:
"""根据用户名获取用户ID"""
person_id = person_info_manager.get_person_id_by_person_name(person_name)
user_id = await person_info_manager.get_value(person_id, "user_id")
platform = await person_info_manager.get_value(person_id, "platform")
return platform, user_id
# 提供简化的API方法
async def send_message(self, text: str, target: Optional[str] = None) -> bool:
"""发送消息的简化方法
Args:
text: 要发送的消息文本
target: 目标消息（可选）
Returns:
bool: 是否发送成功
"""
try:
expressor = self._services.get("expressor")
chat_stream = self._services.get("chat_stream")
if not expressor or not chat_stream:
logger.error(f"{self.log_prefix} 无法发送消息：缺少必要的内部服务")
return False
# 构造简化的动作数据
reply_data = {
"text": text,
"target": target or "",
"emojis": []
}
# 获取锚定消息（如果有）
observations = self._services.get("observations", [])
chatting_observation: ChattingObservation = next(
obs for obs in observations
if isinstance(obs, ChattingObservation)
)
anchor_message = chatting_observation.search_message_by_text(reply_data["target"])
# 如果没有找到锚点消息，创建一个占位符
if not anchor_message:
logger.info(f"{self.log_prefix} 未找到锚点消息，创建占位符")
anchor_message = await create_empty_anchor_message(
chat_stream.platform, chat_stream.group_info, chat_stream
)
else:
anchor_message.update_chat_stream(chat_stream)
response_set = [
("text", text),
]
# 调用内部方法发送消息
success = await expressor.send_response_messages(
anchor_message=anchor_message,
response_set=response_set,
)
return success
except Exception as e:
logger.error(f"{self.log_prefix} 发送消息时出错: {e}")
traceback.print_exc()
return False
async def send_message_by_expressor(self, text: str, target: Optional[str] = None) -> bool:
"""发送消息的简化方法
Args:
text: 要发送的消息文本
target: 目标消息（可选）
Returns:
bool: 是否发送成功
"""
try:
expressor = self._services.get("expressor")
chat_stream = self._services.get("chat_stream")
if not expressor or not chat_stream:
logger.error(f"{self.log_prefix} 无法发送消息：缺少必要的内部服务")
return False
# 构造简化的动作数据
reply_data = {
"text": text,
"target": target or "",
"emojis": []
}
# 获取锚定消息（如果有）
observations = self._services.get("observations", [])
chatting_observation: ChattingObservation = next(
obs for obs in observations
if isinstance(obs, ChattingObservation)
)
anchor_message = chatting_observation.search_message_by_text(reply_data["target"])
# 如果没有找到锚点消息，创建一个占位符
if not anchor_message:
logger.info(f"{self.log_prefix} 未找到锚点消息，创建占位符")
anchor_message = await create_empty_anchor_message(
chat_stream.platform, chat_stream.group_info, chat_stream
)
else:
anchor_message.update_chat_stream(chat_stream)
# 调用内部方法发送消息
success, _ = await expressor.deal_reply(
cycle_timers=self.cycle_timers,
action_data=reply_data,
anchor_message=anchor_message,
reasoning=self.reasoning,
thinking_id=self.thinking_id
)
return success
except Exception as e:
logger.error(f"{self.log_prefix} 发送消息时出错: {e}")
return False
def get_chat_type(self) -> str:
"""获取当前聊天类型
Returns:
str: 聊天类型 ("group" 或 "private")
"""
chat_stream = self._services.get("chat_stream")
if chat_stream and hasattr(chat_stream, "group_info"):
return "group" if chat_stream.group_info else "private"
return "unknown"
def get_recent_messages(self, count: int = 5) -> List[Dict[str, Any]]:
"""获取最近的消息
Args:
count: 要获取的消息数量
Returns:
List[Dict]: 消息列表，每个消息包含发送者、内容等信息
"""
messages = []
observations = self._services.get("observations", [])
if observations and len(observations) > 0:
obs = observations[0]
if hasattr(obs, "get_talking_message"):
raw_messages = obs.get_talking_message()
# 转换为简化格式
for msg in raw_messages[-count:]:
simple_msg = {
"sender": msg.get("sender", "未知"),
"content": msg.get("content", ""),
"timestamp": msg.get("timestamp", 0)
}
messages.append(simple_msg)
return messages
@abstractmethod
async def process(self) -> Tuple[bool, str]:
"""插件处理逻辑，子类必须实现此方法
Returns:
Tuple[bool, str]: (是否执行成功, 回复文本)
"""
pass
async def handle_action(self) -> Tuple[bool, str]:
"""实现BaseAction的抽象方法，调用子类的process方法
Returns:
Tuple[bool, str]: (是否执行成功, 回复文本)
"""
return await self.process()

34
src/chat/focus_chat/planners/actions/reply_action.py
View File

@@ -1,10 +1,8 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from src.common.logger_manager import get_logger
from src.chat.utils.timer_calculator import Timer
from src.chat.focus_chat.planners.actions.base_action import BaseAction, register_action
from typing import Tuple, List, Optional
from typing import Tuple, List
from src.chat.heart_flow.observation.observation import Observation
from src.chat.focus_chat.expressors.default_expressor import DefaultExpressor
from src.chat.message_receive.chat_stream import ChatStream
@@ -22,23 +20,22 @@ class ReplyAction(BaseAction):
处理构建和发送消息回复的动作。
"""
action_name:str = "reply"
action_description:str = "表达想法，可以只包含文本、表情或两者都有"
action_parameters:dict[str:str] = {
action_name: str = "reply"
action_description: str = "表达想法，可以只包含文本、表情或两者都有"
action_parameters: dict[str:str] = {
"text": "你想要表达的内容（可选）",
"emojis": "描述当前使用表情包的场景（可选）",
"target": "你想要回复的原始文本内容（非必须，仅文本，不包含发送者)（可选）",
}
action_require:list[str] = [
action_require: list[str] = [
"有实质性内容需要表达",
"有人提到你，但你还没有回应他",
"在合适的时候添加表情（不要总是添加）",
"如果你要回复特定某人的某句话，或者你想回复较早的消息，请在target中指定那句话的原始文本",
"除非有明确的回复目标，如果选择了target，不用特别提到某个人的人名",
"如果你有明确的,要回复特定某人的某句话，或者你想回复较早的消息，请在target中指定那句话的原始文本",
"一次只回复一个人，一次只回复一个话题,突出重点",
"如果是自己发的消息想继续，需自然衔接",
"避免重复或评价自己的发言,不要和自己聊天",
"注意：回复尽量简短一些。可以参考贴吧，知乎和微博的回复风格，回复不要浮夸，不要用夸张修辞，平淡一些。"
"注意：回复尽量简短一些。可以参考贴吧，知乎和微博的回复风格，回复不要浮夸，不要用夸张修辞，平淡一些。不要有额外的符号，尽量简单简短",
]
default = True
@@ -54,7 +51,7 @@ class ReplyAction(BaseAction):
chat_stream: ChatStream,
current_cycle: CycleDetail,
log_prefix: str,
**kwargs
**kwargs,
):
"""初始化回复动作处理器
@@ -89,7 +86,7 @@ class ReplyAction(BaseAction):
reasoning=self.reasoning,
reply_data=self.action_data,
cycle_timers=self.cycle_timers,
thinking_id=self.thinking_id
thinking_id=self.thinking_id,
)
async def _handle_reply(
@@ -105,13 +102,16 @@ class ReplyAction(BaseAction):
"emojis": "微笑" # 表情关键词列表（可选）
}
"""
# 重置连续不回复计数器
self.total_no_reply_count = 0
self.total_waiting_time = 0.0
# 从聊天观察获取锚定消息
observations: ChattingObservation = self.observations[0]
anchor_message = observations.serch_message_by_text(reply_data["target"])
chatting_observation: ChattingObservation = next(
obs for obs in self.observations
if isinstance(obs, ChattingObservation)
)
if reply_data.get("target"):
anchor_message = chatting_observation.search_message_by_text(reply_data["target"])
else:
anchor_message = None
# 如果没有找到锚点消息，创建一个占位符
if not anchor_message:

53
src/chat/focus_chat/planners/planner.py
View File

@@ -4,7 +4,6 @@ from typing import List, Dict, Any, Optional
from rich.traceback import install
from src.chat.models.utils_model import LLMRequest
from src.config.config import global_config
from src.chat.focus_chat.heartflow_prompt_builder import prompt_builder
from src.chat.focus_chat.info.info_base import InfoBase
from src.chat.focus_chat.info.obs_info import ObsInfo
from src.chat.focus_chat.info.cycle_info import CycleInfo
@@ -13,16 +12,21 @@ from src.chat.focus_chat.info.structured_info import StructuredInfo
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_factory import ActionManager
from src.chat.focus_chat.planners.action_factory import ActionInfo
from src.chat.focus_chat.planners.action_manager import ActionManager
from src.chat.focus_chat.planners.action_manager import ActionInfo
logger = get_logger("planner")
install(extra_lines=3)
def init_prompt():
Prompt(
"""你的名字是{bot_name},{prompt_personality}，{chat_context_description}。需要基于以下信息决定如何参与对话：
"""{extra_info_block}
你的名字是{bot_name},{prompt_personality}，{chat_context_description}。需要基于以下信息决定如何参与对话：
{chat_content_block}
{mind_info_block}
{cycle_info_block}
@@ -44,17 +48,17 @@ def init_prompt():
}}
请输出你的决策 JSON：""",
"planner_prompt",)
"planner_prompt",
)
Prompt(
"""
action_name: {action_name}
描述：{action_description}
参数：
{action_parameters}
{action_parameters}
动作要求：
{action_require}
""",
{action_require}""",
"action_prompt",
)
@@ -85,6 +89,7 @@ class ActionPlanner:
try:
# 获取观察信息
extra_info: list[str] = []
for info in all_plan_info:
if isinstance(info, ObsInfo):
logger.debug(f"{self.log_prefix} 观察信息: {info}")
@@ -104,6 +109,8 @@ class ActionPlanner:
elif isinstance(info, StructuredInfo):
logger.debug(f"{self.log_prefix} 结构化信息: {info}")
structured_info = info.get_data()
else:
extra_info.append(info.get_processed_info())
current_available_actions = self.action_manager.get_using_actions()
@@ -116,6 +123,7 @@ class ActionPlanner:
# structured_info=structured_info, # <-- Pass SubMind info
current_available_actions=current_available_actions, # <-- Pass determined actions
cycle_info=cycle_info, # <-- Pass cycle info
extra_info=extra_info,
)
# --- 调用 LLM (普通文本生成) ---
@@ -142,15 +150,13 @@ class ActionPlanner:
extracted_action = parsed_json.get("action", "no_reply")
extracted_reasoning = parsed_json.get("reasoning", "LLM未提供理由")
# 新的reply格式
if extracted_action == "reply":
action_data = {
"text": parsed_json.get("text", []),
"emojis": parsed_json.get("emojis", []),
"target": parsed_json.get("target", ""),
}
else:
action_data = {} # 其他动作可能不需要额外数据
# 将所有其他属性添加到action_data
action_data = {}
for key, value in parsed_json.items():
if key not in ["action", "reasoning"]:
action_data[key] = value
# 对于reply动作不需要额外处理，因为相关字段已经在上面的循环中添加到action_data
if extracted_action not in current_available_actions:
logger.warning(
@@ -197,7 +203,6 @@ class ActionPlanner:
# 返回结果字典
return plan_result
async def build_planner_prompt(
self,
is_group_chat: bool, # Now passed as argument
@@ -206,6 +211,7 @@ class ActionPlanner:
current_mind: Optional[str],
current_available_actions: Dict[str, ActionInfo],
cycle_info: Optional[str],
extra_info: list[str],
) -> str:
"""构建 Planner LLM 的提示词 (获取模板并填充数据)"""
try:
@@ -218,7 +224,6 @@ class ActionPlanner:
)
chat_context_description = f"你正在和 {chat_target_name} 私聊"
chat_content_block = ""
if observed_messages_str:
chat_content_block = f"聊天记录：\n{observed_messages_str}"
@@ -234,7 +239,6 @@ class ActionPlanner:
individuality = Individuality.get_instance()
personality_block = individuality.get_prompt(x_person=2, level=2)
action_options_block = ""
for using_actions_name, using_actions_info in current_available_actions.items():
# print(using_actions_name)
@@ -247,11 +251,11 @@ class ActionPlanner:
param_text = ""
for param_name, param_description in using_actions_info["parameters"].items():
param_text += f"{param_name}: {param_description}\n"
param_text += f" {param_name}: {param_description}\n"
require_text = ""
for require_item in using_actions_info["require"]:
require_text += f"- {require_item}\n"
require_text += f" - {require_item}\n"
using_action_prompt = using_action_prompt.format(
action_name=using_actions_name,
@@ -262,8 +266,8 @@ class ActionPlanner:
action_options_block += using_action_prompt
extra_info_block = "\n".join(extra_info)
extra_info_block = f"以下是一些额外的信息，现在请你阅读以下内容，进行决策\n{extra_info_block}\n以上是一些额外的信息，现在请你阅读以下内容，进行决策"
planner_prompt_template = await global_prompt_manager.get_prompt_async("planner_prompt")
prompt = planner_prompt_template.format(
@@ -274,6 +278,7 @@ class ActionPlanner:
mind_info_block=mind_info_block,
cycle_info_block=cycle_info,
action_options_text=action_options_block,
extra_info_block=extra_info_block,
)
return prompt

6
src/chat/heart_flow/observation/chatting_observation.py
View File

@@ -14,6 +14,7 @@ from typing import Optional
import difflib
from src.chat.message_receive.message import MessageRecv # 添加 MessageRecv 导入
from src.chat.heart_flow.observation.observation import Observation
from src.common.logger_manager import get_logger
from src.chat.heart_flow.utils_chat import get_chat_type_and_target_info
from src.chat.utils.prompt_builder import Prompt
@@ -43,6 +44,7 @@ class ChattingObservation(Observation):
def __init__(self, chat_id):
super().__init__(chat_id)
self.chat_id = chat_id
self.platform = "qq"
# --- Initialize attributes (defaults) ---
self.is_group_chat: bool = False
@@ -105,7 +107,7 @@ class ChattingObservation(Observation):
mid_memory_str += f"{mid_memory['theme']}\n"
return mid_memory_str + "现在群里正在聊：\n" + self.talking_message_str
def serch_message_by_text(self, text: str) -> Optional[MessageRecv]:
def search_message_by_text(self, text: str) -> Optional[MessageRecv]:
"""
根据回复的纯文本
1. 在talking_message中查找最新的，最匹配的消息
@@ -150,7 +152,7 @@ class ChattingObservation(Observation):
}
message_info = {
"platform": find_msg.get("platform"),
"platform": self.platform,
"message_id": find_msg.get("message_id"),
"time": find_msg.get("time"),
"group_info": group_info,

21
src/chat/heart_flow/observation/hfcloop_observation.py
View File

@@ -3,6 +3,7 @@
from datetime import datetime
from src.common.logger_manager import get_logger
from src.chat.focus_chat.heartFC_Cycleinfo import CycleDetail
from src.chat.focus_chat.planners.action_manager import ActionManager
from typing import List
# Import the new utility function
@@ -16,15 +17,17 @@ class HFCloopObservation:
self.observe_id = observe_id
self.last_observe_time = datetime.now().timestamp() # 初始化为当前时间
self.history_loop: List[CycleDetail] = []
self.action_manager = ActionManager()
def get_observe_info(self):
return self.observe_info
def add_loop_info(self, loop_info: CycleDetail):
# logger.debug(f"添加循环信息111111111111111111111111111111111111: {loop_info}")
# print(f"添加循环信息111111111111111111111111111111111111: {loop_info}")
self.history_loop.append(loop_info)
def set_action_manager(self, action_manager: ActionManager):
self.action_manager = action_manager
async def observe(self):
recent_active_cycles: List[CycleDetail] = []
for cycle in reversed(self.history_loop):
@@ -62,7 +65,6 @@ class HFCloopObservation:
if cycle_info_block:
cycle_info_block = f"\n你最近的回复\n{cycle_info_block}\n"
else:
# 如果最近的活动循环不是文本回复，或者没有活动循环
cycle_info_block = "\n"
# 获取history_loop中最新添加的
@@ -72,8 +74,17 @@ class HFCloopObservation:
end_time = last_loop.end_time
if start_time is not None and end_time is not None:
time_diff = int(end_time - start_time)
cycle_info_block += f"\n距离你上一次阅读消息已经过去了{time_diff}分钟\n"
if time_diff > 60:
cycle_info_block += f"\n距离你上一次阅读消息已经过去了{time_diff/60}分钟\n"
else:
cycle_info_block += f"\n距离你上一次阅读消息已经过去了{time_diff}秒\n"
else:
cycle_info_block += "\n无法获取上一次阅读消息的时间\n"
cycle_info_block += "\n你还没看过消息\n"
using_actions = self.action_manager.get_using_actions()
for action_name, action_info in using_actions.items():
action_description = action_info["description"]
cycle_info_block += f"\n你在聊天中可以使用{action_name}，这个动作的描述是{action_description}\n"
self.observe_info = cycle_info_block

1
src/chat/heart_flow/observation/observation.py
View File

@@ -5,7 +5,6 @@ from src.common.logger_manager import get_logger
logger = get_logger("observation")
# 所有观察的基类
class Observation:
def __init__(self, observe_id):

9
src/chat/person_info/person_info.py
View File

@@ -95,6 +95,15 @@ class PersonInfoManager:
else:
return False
def get_person_id_by_person_name(self, person_name: str):
"""根据用户名获取用户ID"""
document = db.person_info.find_one({"person_name": person_name})
if document:
return document["person_id"]
else:
return ""
@staticmethod
async def create_person_info(person_id: str, data: dict = None):
"""创建一个项"""

101
src/plugins.md Normal file
View File

@@ -0,0 +1,101 @@
# 如何编写MaiBot插件
## 基本步骤
1. 在`src/plugins/你的插件名/actions/`目录下创建插件文件
2. 继承`PluginAction`基类
3. 实现`process`方法
## 插件结构示例
```python
from src.common.logger_manager import get_logger
from src.chat.focus_chat.planners.actions.plugin_action import PluginAction, register_action
from typing import Tuple
logger = get_logger("your_action_name")
@register_action
class YourAction(PluginAction):
"""你的动作描述"""
action_name = "your_action_name" # 动作名称，必须唯一
action_description = "这个动作的详细描述，会展示给用户"
action_parameters = {
"param1": "参数1的说明（可选）",
"param2": "参数2的说明（可选）"
}
action_require = [
"使用场景1",
"使用场景2"
]
default = False # 是否默认启用
async def process(self) -> Tuple[bool, str]:
"""插件核心逻辑"""
# 你的代码逻辑...
return True, "执行结果"
```
## 可用的API方法
插件可以使用`PluginAction`基类提供的以下API：
### 1. 发送消息
```python
await self.send_message("要发送的文本", target="可选的回复目标")
```
### 2. 获取聊天类型
```python
chat_type = self.get_chat_type() # 返回 "group" 或 "private" 或 "unknown"
```
### 3. 获取最近消息
```python
messages = self.get_recent_messages(count=5) # 获取最近5条消息
# 返回格式: [{"sender": "发送者", "content": "内容", "timestamp": 时间戳}, ...]
```
### 4. 获取动作参数
```python
param_value = self.action_data.get("param_name", "默认值")
```
### 5. 日志记录
```python
logger.info(f"{self.log_prefix} 你的日志信息")
logger.warning("警告信息")
logger.error("错误信息")
```
## 返回值说明
`process`方法必须返回一个元组，包含两个元素：
- 第一个元素(bool): 表示动作是否执行成功
- 第二个元素(str): 执行结果的文本描述
```python
return True, "执行成功的消息"
# 或
return False, "执行失败的原因"
```
## 最佳实践
1. 使用`action_parameters`清晰定义你的动作需要的参数
2. 使用`action_require`描述何时应该使用你的动作
3. 使用`action_description`准确描述你的动作功能
4. 使用`logger`记录重要信息，方便调试
5. 避免操作底层系统，尽量使用`PluginAction`提供的API
## 注册与加载
插件会在系统启动时自动加载，只要放在正确的目录并添加了`@register_action`装饰器。
若设置`default = True`，插件会自动添加到默认动作集；否则需要在系统中手动启用。

1
src/plugins/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""插件系统包"""

4
src/plugins/test_plugin/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
"""测试插件包"""
"""
这是一个测试插件
"""

6
src/plugins/test_plugin/actions/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
"""测试插件动作模块"""
# 导入所有动作模块以确保装饰器被执行
from . import test_action # noqa
# from . import online_action # noqa
from . import mute_action # noqa

48
src/plugins/test_plugin/actions/mute_action.py Normal file
View File

@@ -0,0 +1,48 @@
from src.common.logger_manager import get_logger
from src.chat.focus_chat.planners.actions.plugin_action import PluginAction, register_action
from typing import Tuple
logger = get_logger("mute_action")
@register_action
class MuteAction(PluginAction):
"""测试动作处理类"""
action_name = "mute_action"
action_description = "如果某人违反了公序良俗，或者别人戳你太多，，或者某人刷屏，一定要禁言某人，如果你很生气，可以禁言某人"
action_parameters = {
"target": "禁言对象，输入你要禁言的对象的名字，必填，",
"duration": "禁言时长，输入你要禁言的时长，单位为秒，必填",
}
action_require = [
"当有人违反了公序良俗时使用",
"当有人刷屏时使用",
"当有人要求禁言自己时使用",
"当有人戳你两次以上时，防止刷屏，禁言他，必须牢记",
"当千石可乐或可乐酱要求你禁言时使用",
"当你想回避某个话题时使用",
]
default = True # 不是默认动作，需要手动添加到使用集
async def process(self) -> Tuple[bool, str]:
"""处理测试动作"""
logger.info(f"{self.log_prefix} 执行online动作: {self.reasoning}")
# 发送测试消息
target = self.action_data.get("target")
duration = self.action_data.get("duration")
reason = self.action_data.get("reason")
platform, user_id = await self.get_user_id_by_person_name(target)
await self.send_message_by_expressor(f"我要禁言{target}，{platform},时长{duration}秒，理由{reason}，表达情绪")
try:
await self.send_message(f"[command]mute,{user_id},{duration}")
except Exception as e:
logger.error(f"{self.log_prefix} 执行mute动作时出错: {e}")
await self.send_message_by_expressor(f"执行mute动作时出错: {e}")
return False, "执行mute动作时出错"
return True, "测试动作执行成功"

44
src/plugins/test_plugin/actions/online_action.py Normal file
View File

@@ -0,0 +1,44 @@
from src.common.logger_manager import get_logger
from src.chat.focus_chat.planners.actions.plugin_action import PluginAction, register_action
from typing import Tuple
logger = get_logger("check_online_action")
@register_action
class CheckOnlineAction(PluginAction):
"""测试动作处理类"""
action_name = "check_online_action"
action_description = "这是一个检查在线状态的动作，当有人要求你检查Maibot（麦麦 机器人）在线状态时使用"
action_parameters = {
"mode": "查看模式"
}
action_require = [
"当有人要求你检查Maibot（麦麦 机器人）在线状态时使用",
"mode参数为version时查看在线版本状态，默认用这种",
"mode参数为type时查看在线系统类型分布",
]
default = True # 不是默认动作，需要手动添加到使用集
async def process(self) -> Tuple[bool, str]:
"""处理测试动作"""
logger.info(f"{self.log_prefix} 执行online动作: {self.reasoning}")
# 发送测试消息
mode = self.action_data.get("mode", "type")
await self.send_message_by_expressor("我看看")
try:
if mode == "type":
await self.send_message(f"#online detail")
elif mode == "version":
await self.send_message(f"#online")
except Exception as e:
logger.error(f"{self.log_prefix} 执行online动作时出错: {e}")
await self.send_message_by_expressor("执行online动作时出错: {e}")
return False, "执行online动作时出错"
return True, "测试动作执行成功"

38
src/plugins/test_plugin/actions/test_action.py Normal file
View File

@@ -0,0 +1,38 @@
from src.common.logger_manager import get_logger
from src.chat.focus_chat.planners.actions.plugin_action import PluginAction, register_action
from typing import Tuple
logger = get_logger("test_action")
@register_action
class TestAction(PluginAction):
"""测试动作处理类"""
action_name = "test_action"
action_description = "这是一个测试动作，当有人要求你测试插件系统时使用"
action_parameters = {
"test_param": "测试参数（可选）"
}
action_require = [
"测试情况下使用",
"想测试插件动作加载时使用",
]
default = False # 不是默认动作，需要手动添加到使用集
async def process(self) -> Tuple[bool, str]:
"""处理测试动作"""
logger.info(f"{self.log_prefix} 执行测试动作: {self.reasoning}")
# 获取聊天类型
chat_type = self.get_chat_type()
logger.info(f"{self.log_prefix} 当前聊天类型: {chat_type}")
# 获取最近消息
recent_messages = self.get_recent_messages(3)
logger.info(f"{self.log_prefix} 最近3条消息: {recent_messages}")
# 发送测试消息
test_param = self.action_data.get("test_param", "默认参数")
await self.send_message_by_expressor(f"测试动作执行成功，参数: {test_param}")
return True, "测试动作执行成功"