| from collections import defaultdict | |
| from typing import Dict, List | |
| from app.config import config | |
| # { uid: { "text": [...], "stt": [...] } } | |
| _store: Dict[str, Dict[str, List[dict]]] = defaultdict( | |
| lambda: {"text": [], "stt": []} | |
| ) | |
| def get_history(uid: str, channel: str) -> List[dict]: | |
| return list(_store[uid][channel]) | |
| def append_turn(uid: str, channel: str, role: str, content: str) -> None: | |
| history = _store[uid][channel] | |
| history.append({"role": role, "content": content}) | |
| max_messages = config.MAX_HISTORY_TURNS * 2 | |
| if len(history) > max_messages: | |
| _store[uid][channel] = history[-max_messages:] | |
| def clear_session(uid: str) -> None: | |
| _store.pop(uid, None) |