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

refac：tool去处理器化

Browse Source
This commit is contained in:
SengokuCola
2025-07-01 17:47:56 +08:00
parent 6b6f99659d
commit 3544daeadb
12 changed files with 522 additions and 702 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -19,7 +19,7 @@ from src.chat.heart_flow.observation.working_observation import WorkingMemoryObs
from src.chat.heart_flow.observation.chatting_observation import ChattingObservation
from src.chat.heart_flow.observation.structure_observation import StructureObservation
from src.chat.heart_flow.observation.actions_observation import ActionObservation
from src.chat.focus_chat.info_processors.tool_processor import ToolProcessor
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_factory import PlannerFactory
@@ -34,8 +34,7 @@ from src.person_info.relationship_builder_manager import relationship_builder_ma
install(extra_lines=3)
# 超时常量配置
ACTION_MODIFICATION_TIMEOUT = 15.0 # 动作修改任务超时时限（秒）
# 注释：原来的动作修改超时常量已移除，因为改为顺序执行
# 定义观察器映射：键是观察器名称，值是 (观察器类, 初始化参数)
OBSERVATION_CLASSES = {
@@ -51,11 +50,6 @@ PROCESSOR_CLASSES = {
"WorkingMemoryProcessor": (WorkingMemoryProcessor, "working_memory_processor"),
}
# 定义后期处理器映射：在规划后、动作执行前运行的处理器
POST_PLANNING_PROCESSOR_CLASSES = {
"ToolProcessor": (ToolProcessor, "tool_use_processor"),
}
logger = get_logger("hfc") # Logger Name Changed
@@ -128,23 +122,11 @@ class HeartFChatting:
if not config_key or getattr(config_processor_settings, config_key, True):
self.enabled_processor_names.append(proc_name)
# 初始化后期处理器（规划后执行的处理器）
self.enabled_post_planning_processor_names = []
for proc_name, (_proc_class, config_key) in POST_PLANNING_PROCESSOR_CLASSES.items():
# 对于关系相关处理器，需要同时检查关系配置项
if not config_key or getattr(config_processor_settings, config_key, True):
self.enabled_post_planning_processor_names.append(proc_name)
# logger.info(f"{self.log_prefix} 将启用的处理器: {self.enabled_processor_names}")
# logger.info(f"{self.log_prefix} 将启用的后期处理器: {self.enabled_post_planning_processor_names}")
self.processors: List[BaseProcessor] = []
self._register_default_processors()
# 初始化后期处理器
self.post_planning_processors: List[BaseProcessor] = []
self._register_post_planning_processors()
self.action_manager = ActionManager()
self.action_planner = PlannerFactory.create_planner(
log_prefix=self.log_prefix, action_manager=self.action_manager
@@ -186,7 +168,7 @@ class HeartFChatting:
# 检查是否需要跳过WorkingMemoryObservation
if name == "WorkingMemoryObservation":
# 如果工作记忆处理器被禁用，则跳过WorkingMemoryObservation
if not global_config.focus_chat_processor.working_memory_processor:
if not global_config.focus_chat.working_memory_processor:
logger.debug(f"{self.log_prefix} 工作记忆处理器已禁用，跳过注册观察器 {name}")
continue
@@ -211,16 +193,13 @@ class HeartFChatting:
processor_info = PROCESSOR_CLASSES.get(name) # processor_info is (ProcessorClass, config_key)
if processor_info:
processor_actual_class = processor_info[0] # 获取实际的类定义
# 根据处理器类名判断是否需要 subheartflow_id
if name in [
"WorkingMemoryProcessor",
]:
self.processors.append(processor_actual_class(subheartflow_id=self.stream_id))
elif name == "ChattingInfoProcessor":
# 根据处理器类名判断构造参数
if name == "ChattingInfoProcessor":
self.processors.append(processor_actual_class())
elif name == "WorkingMemoryProcessor":
self.processors.append(processor_actual_class(subheartflow_id=self.stream_id))
else:
# 对于PROCESSOR_CLASSES中定义但此处未明确处理构造的处理器
# (例如, 新增了一个处理器到PROCESSOR_CLASSES, 它不需要id, 也不叫ChattingInfoProcessor)
try:
self.processors.append(processor_actual_class()) # 尝试无参构造
logger.debug(f"{self.log_prefix} 注册处理器 {name} (尝试无参构造).")
@@ -239,46 +218,7 @@ class HeartFChatting:
else:
logger.warning(f"{self.log_prefix} 没有注册任何处理器。这可能是由于配置错误或所有处理器都被禁用了。")
def _register_post_planning_processors(self):
"""根据 self.enabled_post_planning_processor_names 注册后期处理器"""
self.post_planning_processors = [] # 清空已有的
for name in self.enabled_post_planning_processor_names: # 'name' is "PersonImpressionpProcessor", etc.
processor_info = POST_PLANNING_PROCESSOR_CLASSES.get(name) # processor_info is (ProcessorClass, config_key)
if processor_info:
processor_actual_class = processor_info[0] # 获取实际的类定义
# 根据处理器类名判断是否需要 subheartflow_id
if name in [
"ToolProcessor",
"RelationshipBuildProcessor",
"RealTimeInfoProcessor",
"ExpressionSelectorProcessor",
]:
self.post_planning_processors.append(processor_actual_class(subheartflow_id=self.stream_id))
else:
# 对于POST_PLANNING_PROCESSOR_CLASSES中定义但此处未明确处理构造的处理器
# (例如, 新增了一个处理器到POST_PLANNING_PROCESSOR_CLASSES, 它不需要id, 也不叫PersonImpressionpProcessor)
try:
self.post_planning_processors.append(processor_actual_class()) # 尝试无参构造
logger.debug(f"{self.log_prefix} 注册后期处理器 {name} (尝试无参构造).")
except TypeError:
logger.error(
f"{self.log_prefix} 后期处理器 {name} 构造失败。它可能需要参数（如 subheartflow_id）但未在注册逻辑中明确处理。"
)
else:
# 这理论上不应该发生，因为 enabled_post_planning_processor_names 是从 POST_PLANNING_PROCESSOR_CLASSES 的键生成的
logger.warning(
f"{self.log_prefix} 在 POST_PLANNING_PROCESSOR_CLASSES 中未找到名为 '{name}' 的处理器定义，将跳过注册。"
)
if self.post_planning_processors:
logger.info(
f"{self.log_prefix} 已注册后期处理器: {[p.__class__.__name__ for p in self.post_planning_processors]}"
)
else:
logger.warning(
f"{self.log_prefix} 没有注册任何后期处理器。这可能是由于配置错误或所有后期处理器都被禁用了。"
)
async def start(self):
"""检查是否需要启动主循环，如果未激活则启动。"""
@@ -460,19 +400,7 @@ class HeartFChatting:
("\n前处理器耗时: " + "; ".join(processor_time_strings)) if processor_time_strings else ""
)
# 新增：输出每个后处理器的耗时
post_processor_time_costs = self._current_cycle_detail.loop_post_processor_info.get(
"post_processor_time_costs", {}
)
post_processor_time_strings = []
for pname, ptime in post_processor_time_costs.items():
formatted_ptime = f"{ptime * 1000:.2f}毫秒" if ptime < 1 else f"{ptime:.2f}秒"
post_processor_time_strings.append(f"{pname}: {formatted_ptime}")
post_processor_time_log = (
("\n后处理器耗时: " + "; ".join(post_processor_time_strings))
if post_processor_time_strings
else ""
)
logger.info(
f"{self.log_prefix} 第{self._current_cycle_detail.cycle_id}次思考,"
@@ -480,7 +408,6 @@ class HeartFChatting:
f"动作: {self._current_cycle_detail.loop_plan_info.get('action_result', {}).get('action_type', '未知动作')}"
+ (f"\n详情: {'; '.join(timer_strings)}" if timer_strings else "")
+ processor_time_log
+ post_processor_time_log
)
# 记录性能数据
@@ -491,8 +418,7 @@ class HeartFChatting:
"action_type": action_result.get("action_type", "unknown"),
"total_time": self._current_cycle_detail.end_time - self._current_cycle_detail.start_time,
"step_times": cycle_timers.copy(),
"processor_time_costs": processor_time_costs, # 前处理器时间
"post_processor_time_costs": post_processor_time_costs, # 后处理器时间
"processor_time_costs": processor_time_costs, # 处理器时间
"reasoning": action_result.get("reasoning", ""),
"success": self._current_cycle_detail.loop_action_info.get("action_taken", False),
}
@@ -634,122 +560,7 @@ class HeartFChatting:
return all_plan_info, processor_time_costs
async def _process_post_planning_processors_with_timing(
self, observations: List[Observation], action_type: str, action_data: dict
) -> tuple[dict, dict]:
"""
处理后期处理器（规划后执行的处理器）并收集详细时间统计
包括：关系处理器、表达选择器、记忆激活器
参数:
observations: 观察器列表
action_type: 动作类型
action_data: 原始动作数据
返回:
tuple[dict, dict]: (更新后的动作数据, 后处理器时间统计)
"""
logger.info(f"{self.log_prefix} 开始执行后期处理器（带详细统计）")
# 创建所有后期任务
task_list = []
task_to_name_map = {}
task_start_times = {}
post_processor_time_costs = {}
# 添加后期处理器任务
for processor in self.post_planning_processors:
processor_name = processor.__class__.__name__
async def run_processor_with_timeout_and_timing(proc=processor, name=processor_name):
start_time = time.time()
try:
result = await asyncio.wait_for(
proc.process_info(observations=observations, action_type=action_type, action_data=action_data),
30,
)
end_time = time.time()
post_processor_time_costs[name] = end_time - start_time
logger.debug(f"{self.log_prefix} 后期处理器 {name} 耗时: {end_time - start_time:.3f}秒")
return result
except Exception as e:
end_time = time.time()
post_processor_time_costs[name] = end_time - start_time
logger.warning(f"{self.log_prefix} 后期处理器 {name} 执行异常，耗时: {end_time - start_time:.3f}秒")
raise e
task = asyncio.create_task(run_processor_with_timeout_and_timing())
task_list.append(task)
task_to_name_map[task] = ("processor", processor_name)
task_start_times[task] = time.time()
logger.info(f"{self.log_prefix} 启动后期处理器任务: {processor_name}")
# 如果没有任何后期任务，直接返回
if not task_list:
logger.info(f"{self.log_prefix} 没有启用的后期处理器或记忆激活器")
return action_data, {}
# 等待所有任务完成
pending_tasks = set(task_list)
all_post_plan_info = []
while pending_tasks:
done, pending_tasks = await asyncio.wait(pending_tasks, return_when=asyncio.FIRST_COMPLETED)
for task in done:
task_type, task_name = task_to_name_map[task]
try:
result = await task
if task_type == "processor":
logger.info(f"{self.log_prefix} 后期处理器 {task_name} 已完成!")
if result is not None:
all_post_plan_info.extend(result)
else:
logger.warning(f"{self.log_prefix} 后期处理器 {task_name} 返回了 None")
except asyncio.TimeoutError:
# 对于超时任务，记录已用时间
elapsed_time = time.time() - task_start_times[task]
if task_type == "processor":
post_processor_time_costs[task_name] = elapsed_time
logger.warning(
f"{self.log_prefix} 后期处理器 {task_name} 超时（>30s），已跳过，耗时: {elapsed_time:.3f}秒"
)
except Exception as e:
# 对于异常任务，记录已用时间
elapsed_time = time.time() - task_start_times[task]
if task_type == "processor":
post_processor_time_costs[task_name] = elapsed_time
logger.error(
f"{self.log_prefix} 后期处理器 {task_name} 执行失败，耗时: {elapsed_time:.3f}秒. 错误: {e}",
exc_info=True,
)
# 将后期处理器的结果整合到 action_data 中
updated_action_data = action_data.copy()
structured_info = ""
for info in all_post_plan_info:
if isinstance(info, StructuredInfo):
structured_info = info.get_processed_info()
if structured_info:
updated_action_data["structured_info"] = structured_info
if all_post_plan_info:
logger.info(f"{self.log_prefix} 后期处理完成，产生了 {len(all_post_plan_info)} 个信息项")
# 输出详细统计信息
if post_processor_time_costs:
stats_str = ", ".join(
[f"{name}: {time_cost:.3f}s" for name, time_cost in post_processor_time_costs.items()]
)
logger.info(f"{self.log_prefix} 后期处理器详细耗时统计: {stats_str}")
return updated_action_data, post_processor_time_costs
async def _observe_process_plan_action_loop(self, cycle_timers: dict, thinking_id: str) -> dict:
try:
@@ -765,10 +576,10 @@ class HeartFChatting:
await self.relationship_builder.build_relation()
# 并行执行调整动作、回忆和处理器阶段
with Timer("调整动作、处理", cycle_timers):
# 创建并行任务
async def modify_actions_task():
# 顺序执行调整动作和处理器阶段
# 第一步：动作修改
with Timer("动作修改", cycle_timers):
try:
# 调用完整的动作修改流程
await self.action_modifier.modify_actions(
observations=self.observations,
@@ -776,44 +587,17 @@ class HeartFChatting:
await self.action_observation.observe()
self.observations.append(self.action_observation)
return True
# 创建两个并行任务，为LLM调用添加超时保护
action_modify_task = asyncio.create_task(
asyncio.wait_for(modify_actions_task(), timeout=ACTION_MODIFICATION_TIMEOUT)
)
processor_task = asyncio.create_task(self._process_processors(self.observations))
# 等待两个任务完成，使用超时保护和详细错误处理
action_modify_result = None
all_plan_info = []
processor_time_costs = {}
try:
action_modify_result, (all_plan_info, processor_time_costs) = await asyncio.gather(
action_modify_task, processor_task, return_exceptions=True
)
# 检查各个任务的结果
if isinstance(action_modify_result, Exception):
if isinstance(action_modify_result, asyncio.TimeoutError):
logger.error(f"{self.log_prefix} 动作修改任务超时")
else:
logger.error(f"{self.log_prefix} 动作修改任务失败: {action_modify_result}")
processor_result = (all_plan_info, processor_time_costs)
if isinstance(processor_result, Exception):
if isinstance(processor_result, asyncio.TimeoutError):
logger.error(f"{self.log_prefix} 处理器任务超时")
else:
logger.error(f"{self.log_prefix} 处理器任务失败: {processor_result}")
all_plan_info = []
processor_time_costs = {}
else:
all_plan_info, processor_time_costs = processor_result
logger.debug(f"{self.log_prefix} 动作修改完成")
except Exception as e:
logger.error(f"{self.log_prefix} 并行任务gather失败: {e}")
logger.error(f"{self.log_prefix} 动作修改失败: {e}")
# 继续执行，不中断流程
# 第二步：信息处理器
with Timer("信息处理器", cycle_timers):
try:
all_plan_info, processor_time_costs = await self._process_processors(self.observations)
except Exception as e:
logger.error(f"{self.log_prefix} 信息处理器失败: {e}")
# 设置默认值以继续执行
all_plan_info = []
processor_time_costs = {}
@@ -833,7 +617,6 @@ class HeartFChatting:
"observed_messages": plan_result.get("observed_messages", ""),
}
# 修正：将后期处理器从执行动作Timer中分离出来
action_type, action_data, reasoning = (
plan_result.get("action_result", {}).get("action_type", "error"),
plan_result.get("action_result", {}).get("action_data", {}),
@@ -849,22 +632,7 @@ class HeartFChatting:
logger.debug(f"{self.log_prefix} 麦麦想要：'{action_str}'")
# 添加：单独计时后期处理器，并收集详细统计
post_processor_time_costs = {}
if action_type != "no_reply":
with Timer("后期处理器", cycle_timers):
logger.debug(f"{self.log_prefix} 执行后期处理器（动作类型: {action_type}）")
# 记录详细的后处理器时间
post_start_time = time.time()
action_data, post_processor_time_costs = await self._process_post_planning_processors_with_timing(
self.observations, action_type, action_data
)
post_end_time = time.time()
logger.info(f"{self.log_prefix} 后期处理器总耗时: {post_end_time - post_start_time:.3f}秒")
else:
logger.debug(f"{self.log_prefix} 跳过后期处理器（动作类型: {action_type}）")
# 修正：纯动作执行计时
# 动作执行计时
with Timer("动作执行", cycle_timers):
success, reply_text, command = await self._handle_action(
action_type, reasoning, action_data, cycle_timers, thinking_id
@@ -877,17 +645,11 @@ class HeartFChatting:
"taken_time": time.time(),
}
# 添加后处理器统计到loop_info
loop_post_processor_info = {
"post_processor_time_costs": post_processor_time_costs,
}
loop_info = {
"loop_observation_info": loop_observation_info,
"loop_processor_info": loop_processor_info,
"loop_plan_info": loop_plan_info,
"loop_action_info": loop_action_info,
"loop_post_processor_info": loop_post_processor_info, # 新增
}
return loop_info