1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
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
importance: int
@classmethod
def create(
cls, content: str, category: MemoryCategory, importance: int = 2
) -> LongTermMemory:
if importance not in (1, 2, 3):
raise ValueError("importance must be between 1 and 3")
return cls(
id=uuid7(),
content=content,
category=category,
created_at=datetime.now(UTC),
importance=importance,
)
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()
|
