Adolfo Reyna
98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
"""Tools that let the agent improve itself between conversations.
|
|
|
|
Three kinds of thing get learned in a voice conversation, and they belong in
|
|
different places:
|
|
|
|
- **How a word was misheard** is about this microphone and this recogniser. It
|
|
goes to `corrections.txt` in the workspace — local, because it would be
|
|
meaningless on another machine.
|
|
- **A lasting preference** — "stop explaining so much" — goes to the personal
|
|
brain's `preferences.md`, because it is true of Adolfo regardless of which
|
|
assistant is listening.
|
|
- **A technique or lesson** goes to the brain's `notes.md`, same reasoning.
|
|
|
|
This module is only the mechanism. *When* to record something, and which file
|
|
it belongs in, is personal setup rather than a property of the agent, so that
|
|
lives in a skill at `~/Workspace/.claude/skills/memory/`. Editing the discipline
|
|
means editing that file, not this one.
|
|
|
|
Everything written lands in a git-tracked file or the brain, so it can be
|
|
reviewed and undone.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
from claude_agent_sdk import create_sdk_mcp_server, tool
|
|
from loguru import logger
|
|
|
|
SERVER_NAME = "memory"
|
|
|
|
# The tool names Claude sees, and must be allowed to call.
|
|
TOOL_NAMES = [
|
|
f"mcp__{SERVER_NAME}__remember_correction",
|
|
f"mcp__{SERVER_NAME}__remember_preference",
|
|
f"mcp__{SERVER_NAME}__remember_note",
|
|
]
|
|
|
|
def build_server(*, workspace: Path, brain=None):
|
|
"""Create the in-process MCP server exposing the memory tools."""
|
|
corrections_file = workspace / "corrections.txt"
|
|
|
|
@tool(
|
|
"remember_correction",
|
|
"Record that speech recognition misheard a word, so it is fixed from now on.",
|
|
{"heard": str, "intended": str},
|
|
)
|
|
async def remember_correction(args):
|
|
heard = (args.get("heard") or "").strip()
|
|
intended = (args.get("intended") or "").strip()
|
|
if not heard or not intended or heard.lower() == intended.lower():
|
|
return {"content": [{"type": "text", "text": "Nothing to record."}]}
|
|
|
|
# A one-word rule risks firing on ordinary speech; a phrase is safer.
|
|
rule = f"{heard} => {intended}"
|
|
existing = corrections_file.read_text() if corrections_file.exists() else ""
|
|
if rule.lower() in existing.lower():
|
|
return {"content": [{"type": "text", "text": "Already known."}]}
|
|
|
|
with corrections_file.open("a") as f:
|
|
if not existing.endswith("\n"):
|
|
f.write("\n")
|
|
f.write(f"{rule}\n")
|
|
logger.info(f"Learned correction: {rule}")
|
|
return {"content": [{"type": "text", "text": f"Recorded: {rule}"}]}
|
|
|
|
@tool(
|
|
"remember_preference",
|
|
"Record a lasting preference about how Adolfo wants to be worked with.",
|
|
{"preference": str},
|
|
)
|
|
async def remember_preference(args):
|
|
return _to_brain(brain, "preference", args.get("preference"))
|
|
|
|
@tool(
|
|
"remember_note",
|
|
"Record a technique, lesson or fact worth having in future sessions.",
|
|
{"note": str},
|
|
)
|
|
async def remember_note(args):
|
|
return _to_brain(brain, "note", args.get("note"))
|
|
|
|
return create_sdk_mcp_server(
|
|
name=SERVER_NAME,
|
|
tools=[remember_correction, remember_preference, remember_note],
|
|
)
|
|
|
|
|
|
def _to_brain(brain, kind: str, text: str | None) -> dict:
|
|
text = (text or "").strip()
|
|
if not text:
|
|
return {"content": [{"type": "text", "text": "Nothing to record."}]}
|
|
if brain is None or not brain.reachable:
|
|
logger.warning(f"Brain unreachable; dropped {kind}: {text[:60]}")
|
|
return {"content": [{"type": "text", "text": "Memory unavailable right now."}]}
|
|
|
|
ok, detail = brain.append(kind, f"- {text}")
|
|
message = f"Recorded to {detail}." if ok else f"Could not record: {detail}"
|
|
return {"content": [{"type": "text", "text": message}]}
|