core/conversation_cache.py

from collections import OrderedDict
import time
from typing import Optional, Tuple

class ConversationCache:
    """TTL + LRU 会话缓存，用于会话粘性（conversation_id -> token_id, session_id）"""
    def __init__(self, max_size: int = 2000, ttl: int = 1800):
        self.max_size = max_size
        self.ttl = ttl
        self._cache: OrderedDict[str, Tuple[float, str, str]] = OrderedDict()  # conv_id -> (expire_ts, token_id, session_id)

    def _clean(self):
        now = time.time()
        # 删除过期条目
        expired = [k for k, (exp, _, _) in self._cache.items() if now >= exp]
        for k in expired:
            self._cache.pop(k, None)

    def get(self, conv_id: str) -> Optional[Tuple[str, str]]:
        self._clean()
        if conv_id not in self._cache:
            return None
        exp, token_id, session_id = self._cache[conv_id]
        if time.time() >= exp:
            self._cache.pop(conv_id, None)
            return None
        # LRU 更新：移到末尾
        self._cache.move_to_end(conv_id)
        return token_id, session_id

    def bind(self, conv_id: str, token_id: str, session_id: str):
        self._clean()
        if conv_id in self._cache:
            self._cache.move_to_end(conv_id)
        else:
            if len(self._cache) >= self.max_size:
                # LRU 淘汰最旧
                self._cache.popitem(last=False)
        self._cache[conv_id] = (time.time() + self.ttl, token_id, session_id)

    def drop(self, conv_id: str):
        self._cache.pop(conv_id, None)

    def size(self) -> int:
        self._clean()
        return len(self._cache)