aaPanel/mod/project/agent/chat_client/memory.py

import json
import os
import time
import uuid
import public
from typing import List, Dict, Any, Union


class MemoryManager:
    def __init__(self, session_id: str, sessions_dir: str = "sessions", sliding_window_size: int = 10,
                 model_name: str = None):
        self.session_id = session_id
        self.sliding_window_size = sliding_window_size
        self.model_name = model_name
        self.session_dir = os.path.join(sessions_dir, session_id)
        self.file_path = os.path.join(self.session_dir, "sessions.json")
        self.meta_file_path = os.path.join(self.session_dir, "meta.json")
        self.history: List[Dict[str, Any]] = []
        self._ensure_sessions_dir()
        self.load_session()
        self._load_or_create_meta()

    def _ensure_sessions_dir(self):
        if not os.path.exists(self.session_dir):
            try:
                os.makedirs(self.session_dir, exist_ok=True)
            except OSError as e:
                public.print_log(f"[ERROR] Failed to create session directory {self.session_dir}: {str(e)}")
                raise

    def _load_or_create_meta(self):
        from mod.project.agent.chat_client.tools.base import atomic_update_json

        def _m(d):
            # 不存在/空: 初始化
            if not d:
                return {
                    "session_id": self.session_id,
                    "model_name": self.model_name,
                    "created_at": time.time(),
                    "total_tokens": 0,
                    "input_tokens": 0,
                    "output_tokens": 0,
                }
            # 存在: 迁移, 无变化则跳过(返回 None 不写)
            changed = False
            if self.model_name and d.get("model_name") != self.model_name:
                d["model_name"] = self.model_name
                changed = True
            if "total_tokens" not in d:
                d["total_tokens"] = 0
                d["input_tokens"] = 0
                d["output_tokens"] = 0
                changed = True
            return d if changed else None

        ok, msg = atomic_update_json(self.meta_file_path, _m)
        if not ok:
            public.print_log(f"[ERROR] Failed to init/update meta file {self.meta_file_path}: {msg}")
            raise RuntimeError(f"meta init/update failed: {msg}")

    def update_meta_tokens(self, total_tokens: int, input_tokens: int, output_tokens: int):
        from mod.project.agent.chat_client.tools.base import atomic_update_json

        def _m(d):
            if not d:  # 文件不存在/空, 兜底初始化
                d = {"session_id": self.session_id, "created_at": time.time()}
            d["total_tokens"] = total_tokens
            d["input_tokens"] = input_tokens
            d["output_tokens"] = output_tokens
            d["updated_at"] = time.time()
            return d

        try:
            atomic_update_json(self.meta_file_path, _m)
        except Exception as e:
            public.print_log(f"[ERROR] update_meta_tokens failed: {str(e)}")

    def load_session(self):
        if os.path.exists(self.file_path):
            try:
                with open(self.file_path, 'r', encoding='utf-8') as f:
                    self.history = json.load(f)
            except Exception as e:
                self.history = []
        else:
            self.history = []

    def save_session(self):
        try:
            with open(self.file_path, 'w', encoding='utf-8') as f:
                json.dump(self.history, f, ensure_ascii=False, indent=2)
        except Exception as e:
            pass

    def add_message(self, role: str, content: Union[str, List[Dict[str, Any]]], id: str = None, **kwargs):
        msg = {
            "id": id if id else str(uuid.uuid4()),
            "role": role,
            "content": content,
            "timestamp": time.time(),
            **kwargs
        }
        self.history.append(msg)
        self.save_session()
        return msg

    def _split_into_rounds(self) -> List[List[Dict[str, Any]]]:
        """
        将历史消息分割为对话轮次。
        一轮对话定义为：从一个 'user' 消息开始，包含随后的所有 'assistant'/'tool' 消息，
        直到遇到下一个 'user' 消息或历史结束。
        """
        rounds = []
        current_round = []

        for msg in self.history:
            # 过滤掉 reasoning_content
            clean_msg = msg.copy()
            # if "reasoning_content" in clean_msg:
            # del clean_msg["reasoning_content"]

            if clean_msg['role'] == 'user':
                if current_round:
                    rounds.append(current_round)
                current_round = [clean_msg]
            else:
                # 兼容性：如果历史记录不是以 user 开头（罕见），也归入当前轮次（或创建新轮次）
                if not current_round and not rounds:
                    # 孤立的非 user 消息，作为第一轮
                    current_round = [clean_msg]
                else:
                    current_round.append(clean_msg)

        if current_round:
            rounds.append(current_round)

        return rounds

    def get_sliding_window(self) -> List[Dict[str, Any]]:
        """
        返回最后 N 轮对话中的所有消息。
        配置项 SLIDING_WINDOW_SIZE 现在表示轮次数，而非单条消息数。
        """
        rounds = self._split_into_rounds()

        # 获取最后 N 轮
        last_n_rounds = rounds[-self.sliding_window_size:]

        # 展平为消息列表
        window_messages = []
        for r in last_n_rounds:
            window_messages.extend(r)

        return window_messages

    def get_dynamic_window(self, token_budget: int, estimator) -> List[Dict[str, Any]]:
        """按 token 预算从最新轮向前选择历史窗口。"""
        rounds = self._split_into_rounds()
        if not rounds or token_budget <= 0:
            return self.get_sliding_window()

        selected_rounds = []
        used_tokens = 0

        for round_msgs in reversed(rounds):
            round_tokens = estimator(round_msgs)
            if selected_rounds and used_tokens + round_tokens > token_budget:
                break
            selected_rounds.append(round_msgs)
            used_tokens += round_tokens
            if used_tokens >= token_budget:
                break

        if not selected_rounds:
            selected_rounds.append(rounds[-1])

        window_messages = []
        for r in reversed(selected_rounds):
            window_messages.extend(r)
        return window_messages

    def get_full_history(self) -> List[Dict[str, Any]]:
        return self.history

    def get_total_rounds(self) -> int:
        return len(self._split_into_rounds())

    def get_message_rounds(self) -> int:
        """
        获取当前对话轮数
        Returns:
            int: 对话轮次数量
        """
        return len(self._split_into_rounds())

    def compact_messages(self, summary_msg: Dict, messages_to_keep: List) -> None:
        """
        执行消息压缩，将历史替换为摘要消息+保留的最近消息
        压缩后结构：[summary_msg, ...messages_to_keep]
        Args:
            summary_msg: 压缩摘要消息字典，role为user，包含历史摘要
            messages_to_keep: 需要保留的最近消息列表（最后N轮对话）
        """
        self.history = [summary_msg] + messages_to_keep
        self.save_session()

    def check_auto_compact(self, max_context_tokens: int, threshold_ratio: float = 0.75) -> bool:
        """
        检查是否需要自动压缩（预留方法，当前不启用）
        Args:
            max_context_tokens: 最大上下文token数
            threshold_ratio: 触发压缩的阈值比例（已使用token / 最大token）
        Returns:
            bool: 是否需要压缩
        Note: 此方法当前仅作为预留入口，自动压缩功能未启用
        """
        # 自动压缩功能当前未启用，此方法仅作为预留代码位置
        # 如需实现，可使用 Agent._estimate_messages_tokens 进行 token 估算
        return False