diff options
| -rw-r--r-- | agent.py | 26 | ||||
| -rw-r--r-- | client.py | 25 | ||||
| -rw-r--r-- | memory.py | 72 |
3 files changed, 117 insertions, 6 deletions
@@ -23,16 +23,14 @@ class Agent: user_profile, ) - self.working_memory = WorkingMemory( - objective="Testing working memory", - facts=["The user is testing the agent's working memory system."], - notes=["This is temporary test data."], - ) + self.working_memory = WorkingMemory() self.messages: list[dict[str, object]] = [] def chat(self, user_input: str) -> str: - + if self.working_memory.objective is None: + self.working_memory.start_task(user_input) + print(f"\n[memory: {self.working_memory}]") self.messages.append( { "role": "user", @@ -90,6 +88,7 @@ class Agent: else: try: result = tool.execute(call.arguments).content + self.working_memory.add_note(f"{call.name} returned: {result}") except Exception as error: result = f"Tool error: {error}" @@ -101,3 +100,18 @@ class Agent: } ) raise RuntimeError("Tool call limit exceeded") + + def end_task(self) -> None: + self.working_memory.end_task() + + def start_task(self, objective: str) -> None: + self.working_memory.start_task(objective) + + def pause_task(self) -> None: + self.working_memory.pause_task() + + def resume_task(self) -> None: + self.working_memory.resume_task() + + def complete_task(self) -> None: + self.working_memory.complete_task() @@ -37,5 +37,30 @@ while True: if user_input.lower() in {"exit", "quit"}: break + match user_input: + case "/end": + agent.end_task() + print("[task ended]") + continue + + case "/task": + print(f"\n[memory: {agent.working_memory}]\n") + continue + + case "/pause": + agent.pause_task() + print("[task paused]") + continue + + case "/resume": + agent.resume_task() + print("[task resumed]") + continue + + case "/complete": + agent.complete_task() + print("[task completed]") + continue + answer = agent.chat(user_input) print(f"Agent: {answer}") @@ -1,4 +1,11 @@ from dataclasses import dataclass, field +from enum import Enum + + +class TaskStatus(Enum): + ACTIVE = "active" + PAUSED = "paused" + COMPLETED = "completed" @dataclass @@ -6,3 +13,68 @@ 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() |