import unicodedata from enum import Enum from typing import Any, Dict, List, Optional from pydantic import BaseModel, field_validator class MemoryType(str, Enum): EPISODIC = "episodic" SEMANTIC = "semantic" STATE = "state" PROCEDURAL = "procedural" _MEMORY_TYPE_ALIASES = { "episodico": MemoryType.EPISODIC, "episodic": MemoryType.EPISODIC, "semantico": MemoryType.SEMANTIC, "semantic": MemoryType.SEMANTIC, "estado": MemoryType.STATE, "state": MemoryType.STATE, "procedural": MemoryType.PROCEDURAL, "procedimental": MemoryType.PROCEDURAL, } def normalize_memory_type(value: Any) -> MemoryType: if isinstance(value, MemoryType): return value if value is None or (isinstance(value, str) and not value.strip()): return MemoryType.SEMANTIC key = unicodedata.normalize("NFKD", str(value).strip().lower()) key = "".join(c for c in key if not unicodedata.combining(c)) return _MEMORY_TYPE_ALIASES.get(key, MemoryType.SEMANTIC) class Memory(BaseModel): id: str content: str type: MemoryType created_at: str source: str # "seed", "agent", "user" context_tags: List[str] access_count: int = 0 last_accessed: Optional[str] = None relevance_score: float = 1.0 decay_rate: float active: bool = True summary: str class NewMemoryItem(BaseModel): content: str type: MemoryType = MemoryType.SEMANTIC context_tags: List[str] = [] summary: str = "" @field_validator("type", mode="before") @classmethod def coerce_memory_type(cls, value: Any) -> MemoryType: return normalize_memory_type(value) class AgentLLMOutput(BaseModel): response: str memories_used: List[str] = [] new_memories: List[NewMemoryItem] = [] class ChatRequest(BaseModel): message: str conversation_history: List[Dict[str, str]] = [] class ChatResponse(BaseModel): response: str memories_used: List[str] new_memories: List[Memory] all_memories: List[Memory]