From 1a67c5edb6fa6165e99cc25ed380303ed817af09 Mon Sep 17 00:00:00 2001 From: batsumaru <> Date: Tue, 15 Sep 2026 15:13:34 +0900 Subject: Cleaned up project tree --- personal_agent/memory/models.py | 101 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 101 insertions(+) create mode 100644 personal_agent/memory/models.py (limited to 'personal_agent/memory/models.py') diff --git a/personal_agent/memory/models.py b/personal_agent/memory/models.py new file mode 100644 index 0000000..aab2064 --- /dev/null +++ b/personal_agent/memory/models.py @@ -0,0 +1,101 @@ +from dataclasses import dataclass, field +from datetime import UTC, datetime +from enum import Enum +from uuid import UUID, uuid7 + + +class MemoryCategory(Enum): + SEMANTIC = "semantic" + EPISODIC = "episodic" + + +@dataclass +class LongTermMemory: + id: UUID + content: str + category: MemoryCategory + created_at: datetime + + @classmethod + def create(cls, content: str, category: MemoryCategory) -> LongTermMemory: + return cls( + id=uuid7(), content=content, category=category, created_at=datetime.now(UTC) + ) + + +class TaskStatus(Enum): + ACTIVE = "active" + PAUSED = "paused" + COMPLETED = "completed" + + +@dataclass +class WorkingMemory: + objective: str | None = None + facts: list[str] = field(default_factory=list) + notes: list[str] = field(default_factory=list) + status: TaskStatus | None = None + + def __str__(self) -> str: + parts: list[str] = [] + + if self.objective: + parts.append(f"Objective: {self.objective}") + + if self.status: + parts.append(f"Status: {self.status.value}") + + if self.facts: + parts.append("Facts:") + parts.extend(f"- {fact}" for fact in self.facts) + + if self.notes: + parts.append("Notes:") + parts.extend(f"- {note}" for note in self.notes) + + return "\n".join(parts) if parts else "(empty)" + + def set_objective(self, objective: str) -> None: + self.objective = objective + + def add_fact(self, fact: str) -> None: + self.facts.append(fact) + + def remove_fact(self, fact: str) -> None: + if fact in self.facts: + self.facts.remove(fact) + + def add_note(self, note: str) -> None: + self.notes.append(note) + + def remove_note(self, note: str) -> None: + if note in self.notes: + self.notes.remove(note) + + def set_status(self, status: TaskStatus): + self.status = status + + def clear(self) -> None: + self.objective = None + self.status = None + self.facts.clear() + self.notes.clear() + + def start_task(self, objective: str) -> None: + self.clear() + self.objective = objective + self.set_status(TaskStatus.ACTIVE) + + def pause_task(self) -> None: + if self.status == TaskStatus.ACTIVE: + self.set_status(TaskStatus.PAUSED) + + def resume_task(self) -> None: + if self.status == TaskStatus.PAUSED: + self.set_status(TaskStatus.ACTIVE) + + def complete_task(self) -> None: + self.set_status(TaskStatus.COMPLETED) + + def end_task(self) -> None: + self.clear() -- cgit v1.3