-
Notifications
You must be signed in to change notification settings - Fork 17
Expand file tree
/
Copy pathremember.py
More file actions
47 lines (37 loc) · 1.64 KB
/
remember.py
File metadata and controls
47 lines (37 loc) · 1.64 KB
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
"""remember — persist a user fact via the memory service (★②).
V1 is the manual ``/remember`` path: the model (or a slash command) records a
fact verbatim. Recall/extraction strategies evolve behind MemoryService without
this tool changing. The service is injected at assembly time.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from ..core.ports.audit import AuditEvent
from ..core.types import ToolResult, ToolSpec
from ..memory.service import MemoryService
if TYPE_CHECKING:
from ..harness.context import HarnessContext
class Remember:
def __init__(self, memory: MemoryService) -> None:
self._memory = memory
@property
def spec(self) -> ToolSpec:
return ToolSpec(
name="remember",
description="Persist a fact about the user/data so it informs future turns.",
parameters={
"type": "object",
"properties": {"text": {"type": "string"}},
"required": ["text"],
},
)
async def run(self, args: dict[str, Any], ctx: "HarnessContext") -> ToolResult:
text = (args.get("text") or "").strip()
if not text:
return ToolResult(call_id="", content="nothing to remember", is_error=True)
fact = await self._memory.remember(ctx.identity.user_id, text)
if ctx.audit is not None:
await ctx.audit.record(
AuditEvent(actor=ctx.identity.user_id, action="remember",
scope=ctx.identity.session_key(), detail={"fact_id": fact.id})
)
return ToolResult(call_id="", content=f"🧠 Remembered: {text}")