tiphys.tools.memory

 1"""
 2Memory tools for the agent.
 3
 4Allows the agent to store and retrieve facts from long-term memory.
 5"""
 6
 7from pydantic import BaseModel, Field
 8
 9from tiphys.memory.store import get_memory_store
10from tiphys.tools.base import tool
11
12
13class StoreMemoryArgs(BaseModel):
14    """Arguments for storing a memory."""
15
16    content: str = Field(..., description="The information or fact to remember")
17    scope: str = Field(
18        "global", description="Scope: 'global', 'main' (agent-specific), or a group/channel ID"
19    )
20    importance: int = Field(1, description="Importance of the memory (1-5)")
21    tags: list[str] = Field(default_factory=list, description="Optional tags for categorization")
22
23
24@tool(name="store_memory", description="Save a fact or information to long-term memory")
25async def store_memory(
26    content: str,
27    scope: str = "global",
28    importance: int = 1,
29    tags: list[str] | None = None,
30) -> str:
31    """
32    Store information that might be useful in future conversations.
33    """
34    if tags is None:
35        tags = []
36    store = get_memory_store()
37
38    metadata = {
39        "importance": importance,
40        "tags": ",".join(tags) if tags else "",
41    }
42
43    entry = store.add(scope=scope, content=content, metadata=metadata)
44    return f"Information saved to {scope} memory (ID: {entry.id})"
45
46
47class SearchMemoryArgs(BaseModel):
48    """Arguments for searching memories."""
49
50    query: str = Field(..., description="The topic or question to search for")
51    scopes: list[str] = Field(["global"], description="Scopes to search in")
52    limit: int = Field(5, description="Maximum number of results to return")
53
54
55@tool(name="search_memory", description="Search long-term memory for relevant facts")
56async def search_memory(
57    query: str,
58    scopes: list[str] | None = None,
59    limit: int = 5,
60) -> str:
61    """
62    Search past interactions and stored facts for information relevant to the query.
63    """
64    if scopes is None:
65        scopes = ["global"]
66    store = get_memory_store()
67
68    # In a real agent loop, we'd inject current session scope automatically
69    # For now, we rely on the agent to pick the right scopes
70
71    results = store.search(scopes=scopes, query=query, limit=limit)
72
73    if not results:
74        return "No relevant memories found."
75
76    lines = [f"Found {len(results)} relevant memories:"]
77    for i, res in enumerate(results, 1):
78        lines.append(f"{i}. [{res.match_type}] {res.entry.content}")
79        if res.entry.metadata.get("tags"):
80            lines.append(f"   Tags: {res.entry.metadata['tags']}")
81
82    return "\n".join(lines)
83
84
85@tool(name="forget_memory", description="Delete a specific memory entry by ID")
86async def forget_memory(scope: str, entry_id: str) -> str:
87    """Permanently remove a memory from the store."""
88    store = get_memory_store()
89
90    if store.delete(scope, entry_id):
91        return f"Memory {entry_id} deleted from {scope}."
92    return f"Memory {entry_id} not found in {scope}."