杜宇轩
42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
import re
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass, field
|
|
from typing import List
|
|
|
|
SAFE_ID = re.compile(r"^[a-zA-Z0-9._-]")
|
|
|
|
|
|
def sanitize_id(value: str, fallback:str) -> str:
|
|
"""Sanitize a string to be used as an ID by removing unsafe characters."""
|
|
return SAFE_ID.sub("_", value) or fallback
|
|
|
|
|
|
@dataclass
|
|
class Session:
|
|
"""Conversasion history for one (user_id, session_id) pair.
|
|
``message`` is in Strands / Bedrock-converse format (provider-neutral::
|
|
{"role": "user"|"assistant"|"system",
|
|
"content": [{"text": ...} | {"toolUse": ...} | {"toolResult": ...} ]})
|
|
"""
|
|
session_id: str
|
|
user_id: str
|
|
messages: List[dict] = field(default_factory=list)
|
|
|
|
|
|
class SessionStore(ABC):
|
|
"""Abstract base class for session storage backends."""
|
|
|
|
@abstractmethod
|
|
def load(self, user_id: str, session_id: str) -> Session:
|
|
"""Return prior history for the key (empty Session if none exists yet)."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def save(self, session: Session) -> None:
|
|
"""Save a session to the store."""
|
|
pass
|
|
|
|
@abstractmethod
|
|
def clear(self, user_id: str, session_id: str) -> None:
|
|
"""Clear a session from the store."""
|
|
pass |