forked from Rockachopa/Timmy-time-dashboard
This commit is contained in:
253
src/timmy/memory/tools.py
Normal file
253
src/timmy/memory/tools.py
Normal file
@@ -0,0 +1,253 @@
|
||||
"""Tool functions for Timmy's memory system.
|
||||
|
||||
memory_search, memory_read, memory_store, memory_forget — runtime tool wrappers.
|
||||
jot_note, log_decision — artifact production tools.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import re
|
||||
from datetime import UTC, datetime
|
||||
from pathlib import Path
|
||||
|
||||
from timmy.memory.crud import delete_memory, search_memories, store_memory
|
||||
from timmy.memory.semantic import semantic_memory
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def memory_search(query: str, limit: int = 10) -> str:
|
||||
"""Search past conversations, notes, and stored facts for relevant context.
|
||||
|
||||
Searches across both the vault (indexed markdown files) and the
|
||||
runtime memory store (facts and conversation fragments stored via
|
||||
memory_write).
|
||||
|
||||
Args:
|
||||
query: What to search for (e.g. "Bitcoin strategy", "server setup").
|
||||
limit: Number of results to return (default 10).
|
||||
|
||||
Returns:
|
||||
Formatted string of relevant memory results.
|
||||
"""
|
||||
# Guard: model sometimes passes None for limit
|
||||
if limit is None:
|
||||
limit = 10
|
||||
|
||||
parts: list[str] = []
|
||||
|
||||
# 1. Search semantic vault (indexed markdown files)
|
||||
vault_results = semantic_memory.search(query, limit)
|
||||
for content, score in vault_results:
|
||||
if score < 0.2:
|
||||
continue
|
||||
parts.append(f"[vault score {score:.2f}] {content[:300]}")
|
||||
|
||||
# 2. Search runtime vector store (stored facts/conversations)
|
||||
try:
|
||||
runtime_results = search_memories(query, limit=limit, min_relevance=0.2)
|
||||
for entry in runtime_results:
|
||||
label = entry.context_type or "memory"
|
||||
parts.append(f"[{label}] {entry.content[:300]}")
|
||||
except Exception as exc:
|
||||
logger.debug("Vector store search unavailable: %s", exc)
|
||||
|
||||
if not parts:
|
||||
return "No relevant memories found."
|
||||
return "\n\n".join(parts)
|
||||
|
||||
|
||||
def memory_read(query: str = "", top_k: int = 5) -> str:
|
||||
"""Read from persistent memory — search facts, notes, and past conversations.
|
||||
|
||||
This is the primary tool for recalling stored information. If no query
|
||||
is given, returns the most recent personal facts. With a query, it
|
||||
searches semantically across all stored memories.
|
||||
|
||||
Args:
|
||||
query: Optional search term. Leave empty to list recent facts.
|
||||
top_k: Maximum results to return (default 5).
|
||||
|
||||
Returns:
|
||||
Formatted string of memory contents.
|
||||
"""
|
||||
if top_k is None:
|
||||
top_k = 5
|
||||
|
||||
parts: list[str] = []
|
||||
|
||||
# Always include personal facts first
|
||||
try:
|
||||
facts = search_memories(query or "", limit=top_k, min_relevance=0.0)
|
||||
fact_entries = [e for e in facts if (e.context_type or "") == "fact"]
|
||||
if fact_entries:
|
||||
parts.append("## Personal Facts")
|
||||
for entry in fact_entries[:top_k]:
|
||||
parts.append(f"- {entry.content[:300]}")
|
||||
except Exception as exc:
|
||||
logger.debug("Vector store unavailable for memory_read: %s", exc)
|
||||
|
||||
# If a query was provided, also do semantic search
|
||||
if query:
|
||||
search_result = memory_search(query, top_k)
|
||||
if search_result and search_result != "No relevant memories found.":
|
||||
parts.append("\n## Search Results")
|
||||
parts.append(search_result)
|
||||
|
||||
if not parts:
|
||||
return "No memories stored yet. Use memory_write to store information."
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
def memory_store(topic: str, report: str, type: str = "research") -> str:
|
||||
"""Store a piece of information in persistent memory, particularly for research outputs.
|
||||
|
||||
Use this tool to store structured research findings or other important documents.
|
||||
Stored memories are searchable via memory_search across all channels.
|
||||
|
||||
Args:
|
||||
topic: A concise title or topic for the research output.
|
||||
report: The detailed content of the research output or document.
|
||||
type: Type of memory — "research" for research outputs (default),
|
||||
"fact" for permanent facts, "conversation" for conversation context,
|
||||
"document" for other document fragments.
|
||||
|
||||
Returns:
|
||||
Confirmation that the memory was stored.
|
||||
"""
|
||||
if not report or not report.strip():
|
||||
return "Nothing to store — report is empty."
|
||||
|
||||
# Combine topic and report for embedding and storage content
|
||||
full_content = f"Topic: {topic.strip()}\n\nReport: {report.strip()}"
|
||||
|
||||
valid_types = ("fact", "conversation", "document", "research")
|
||||
if type not in valid_types:
|
||||
type = "research"
|
||||
|
||||
try:
|
||||
# Dedup check for facts and research — skip if similar exists
|
||||
if type in ("fact", "research"):
|
||||
existing = search_memories(full_content, limit=3, context_type=type, min_relevance=0.75)
|
||||
if existing:
|
||||
return (
|
||||
f"Similar {type} already stored (id={existing[0].id[:8]}). Skipping duplicate."
|
||||
)
|
||||
|
||||
entry = store_memory(
|
||||
content=full_content,
|
||||
source="agent",
|
||||
context_type=type,
|
||||
metadata={"topic": topic},
|
||||
)
|
||||
return f"Stored in memory (type={type}, id={entry.id[:8]}). This is now searchable across all channels."
|
||||
except Exception as exc:
|
||||
logger.error("Failed to write memory: %s", exc)
|
||||
return f"Failed to store memory: {exc}"
|
||||
|
||||
|
||||
def memory_forget(query: str) -> str:
|
||||
"""Remove a stored memory that is outdated, incorrect, or no longer relevant.
|
||||
|
||||
Searches for memories matching the query and deletes the closest match.
|
||||
Use this when the user says to forget something or when stored information
|
||||
has changed.
|
||||
|
||||
Args:
|
||||
query: Description of the memory to forget (e.g. "my phone number",
|
||||
"the old server address").
|
||||
|
||||
Returns:
|
||||
Confirmation of what was forgotten, or a message if nothing matched.
|
||||
"""
|
||||
if not query or not query.strip():
|
||||
return "Nothing to forget — query is empty."
|
||||
|
||||
try:
|
||||
results = search_memories(query.strip(), limit=3, min_relevance=0.3)
|
||||
if not results:
|
||||
return "No matching memories found to forget."
|
||||
|
||||
# Delete the closest match
|
||||
best = results[0]
|
||||
deleted = delete_memory(best.id)
|
||||
if deleted:
|
||||
return f'Forgotten: "{best.content[:80]}" (type={best.context_type})'
|
||||
return "Memory not found (may have already been deleted)."
|
||||
except Exception as exc:
|
||||
logger.error("Failed to forget memory: %s", exc)
|
||||
return f"Failed to forget: {exc}"
|
||||
|
||||
|
||||
# ── Artifact tools ───────────────────────────────────────────────────────────
|
||||
|
||||
NOTES_DIR = Path.home() / ".timmy" / "notes"
|
||||
DECISION_LOG = Path.home() / ".timmy" / "decisions.md"
|
||||
|
||||
|
||||
def jot_note(title: str, body: str) -> str:
|
||||
"""Write a markdown note to Timmy's workspace (~/.timmy/notes/).
|
||||
|
||||
Use this tool to capture ideas, drafts, summaries, or any artifact that
|
||||
should persist beyond the conversation. Each note is saved as a
|
||||
timestamped markdown file.
|
||||
|
||||
Args:
|
||||
title: Short descriptive title (used as filename slug).
|
||||
body: Markdown content of the note.
|
||||
|
||||
Returns:
|
||||
Confirmation with the file path of the saved note.
|
||||
"""
|
||||
if not title or not title.strip():
|
||||
return "Cannot jot — title is empty."
|
||||
if not body or not body.strip():
|
||||
return "Cannot jot — body is empty."
|
||||
|
||||
NOTES_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
slug = re.sub(r"[^a-z0-9]+", "-", title.strip().lower()).strip("-")[:60]
|
||||
timestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
|
||||
filename = f"{timestamp}_{slug}.md"
|
||||
filepath = NOTES_DIR / filename
|
||||
|
||||
content = f"# {title.strip()}\n\n> Created: {datetime.now(UTC).isoformat()}\n\n{body.strip()}\n"
|
||||
filepath.write_text(content)
|
||||
logger.info("jot_note: wrote %s", filepath)
|
||||
return f"Note saved: {filepath}"
|
||||
|
||||
|
||||
def log_decision(decision: str, rationale: str = "") -> str:
|
||||
"""Append an architectural or design decision to the running decision log.
|
||||
|
||||
Use this tool when a significant decision is made during conversation —
|
||||
technology choices, design trade-offs, scope changes, etc.
|
||||
|
||||
Args:
|
||||
decision: One-line summary of the decision.
|
||||
rationale: Why this decision was made (optional but encouraged).
|
||||
|
||||
Returns:
|
||||
Confirmation that the decision was logged.
|
||||
"""
|
||||
if not decision or not decision.strip():
|
||||
return "Cannot log — decision is empty."
|
||||
|
||||
DECISION_LOG.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Create file with header if it doesn't exist
|
||||
if not DECISION_LOG.exists():
|
||||
DECISION_LOG.write_text(
|
||||
"# Decision Log\n\nRunning log of architectural and design decisions.\n\n"
|
||||
)
|
||||
|
||||
stamp = datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC")
|
||||
entry = f"## {stamp} — {decision.strip()}\n\n"
|
||||
if rationale and rationale.strip():
|
||||
entry += f"{rationale.strip()}\n\n"
|
||||
entry += "---\n\n"
|
||||
|
||||
with open(DECISION_LOG, "a") as f:
|
||||
f.write(entry)
|
||||
|
||||
logger.info("log_decision: %s", decision.strip()[:80])
|
||||
return f"Decision logged: {decision.strip()}"
|
||||
Reference in New Issue
Block a user