forked from Rockachopa/Timmy-time-dashboard
254 lines
9.0 KiB
Python
254 lines
9.0 KiB
Python
"""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()}"
|