92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
import time
|
|
from typing import Dict, Any, Tuple
|
|
|
|
from src.common.logger import get_logger
|
|
from src.chat.chat_loop.hfc_utils import CycleDetail
|
|
from .hfc_context import HfcContext
|
|
|
|
logger = get_logger("hfc.cycle")
|
|
|
|
class CycleTracker:
|
|
def __init__(self, context: HfcContext):
|
|
"""
|
|
初始化循环跟踪器
|
|
|
|
Args:
|
|
context: HFC聊天上下文对象
|
|
|
|
功能说明:
|
|
- 负责跟踪和记录每次思考循环的详细信息
|
|
- 管理循环的开始、结束和信息存储
|
|
"""
|
|
self.context = context
|
|
|
|
def start_cycle(self) -> Tuple[Dict[str, float], str]:
|
|
"""
|
|
开始新的思考循环
|
|
|
|
Returns:
|
|
tuple: (循环计时器字典, 思考ID字符串)
|
|
|
|
功能说明:
|
|
- 增加循环计数器
|
|
- 创建新的循环详情对象
|
|
- 生成唯一的思考ID
|
|
- 初始化循环计时器
|
|
"""
|
|
self.context.cycle_counter += 1
|
|
self.context.current_cycle_detail = CycleDetail(self.context.cycle_counter)
|
|
self.context.current_cycle_detail.thinking_id = f"tid{str(round(time.time(), 2))}"
|
|
cycle_timers = {}
|
|
return cycle_timers, self.context.current_cycle_detail.thinking_id
|
|
|
|
def end_cycle(self, loop_info: Dict[str, Any], cycle_timers: Dict[str, float]):
|
|
"""
|
|
结束当前思考循环
|
|
|
|
Args:
|
|
loop_info: 循环信息，包含规划和动作信息
|
|
cycle_timers: 循环计时器，记录各阶段耗时
|
|
|
|
功能说明:
|
|
- 设置循环详情的完整信息
|
|
- 将当前循环加入历史记录
|
|
- 记录计时器和结束时间
|
|
- 打印循环统计信息
|
|
"""
|
|
if self.context.current_cycle_detail:
|
|
self.context.current_cycle_detail.set_loop_info(loop_info)
|
|
self.context.history_loop.append(self.context.current_cycle_detail)
|
|
self.context.current_cycle_detail.timers = cycle_timers
|
|
self.context.current_cycle_detail.end_time = time.time()
|
|
self.print_cycle_info(cycle_timers)
|
|
|
|
def print_cycle_info(self, cycle_timers: Dict[str, float]):
|
|
"""
|
|
打印循环统计信息
|
|
|
|
Args:
|
|
cycle_timers: 循环计时器字典
|
|
|
|
功能说明:
|
|
- 格式化各阶段的耗时信息
|
|
- 计算总体循环持续时间
|
|
- 输出详细的性能统计日志
|
|
- 显示选择的动作类型
|
|
"""
|
|
if not self.context.current_cycle_detail:
|
|
return
|
|
|
|
timer_strings = []
|
|
for name, elapsed in cycle_timers.items():
|
|
formatted_time = f"{elapsed * 1000:.2f}毫秒" if elapsed < 1 else f"{elapsed:.2f}秒"
|
|
timer_strings.append(f"{name}: {formatted_time}")
|
|
|
|
if self.context.current_cycle_detail.end_time and self.context.current_cycle_detail.start_time:
|
|
duration = self.context.current_cycle_detail.end_time - self.context.current_cycle_detail.start_time
|
|
logger.info(
|
|
f"{self.context.log_prefix} 第{self.context.current_cycle_detail.cycle_id}次思考,"
|
|
f"耗时: {duration:.1f}秒, "
|
|
f"选择动作: {self.context.current_cycle_detail.loop_plan_info.get('action_result', {}).get('action_type', '未知动作')}"
|
|
+ (f"\n详情: {'; '.join(timer_strings)}" if timer_strings else "")
|
|
) |