mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-24 10:52:21 +00:00
70e7132e2f
Mirror built-in memory writes to external providers only after the native memory tool succeeds and is not staged for approval. Keep OpenViking's built-in memory mirroring add-only, since Hermes native memory entries do not yet have stable OpenViking file URIs for replace/remove. Add a narrow viking_forget tool for exact user memory file deletion and document the current OpenViking write/delete behavior.
61 lines
1.8 KiB
Python
61 lines
1.8 KiB
Python
"""Helpers for mirroring built-in memory writes to external providers."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any, Dict, List
|
|
|
|
_MIRRORED_MEMORY_ACTIONS = {"add", "replace", "remove"}
|
|
|
|
|
|
def _memory_tool_result_succeeded(result: Any) -> bool:
|
|
if isinstance(result, str):
|
|
try:
|
|
result = json.loads(result)
|
|
except Exception:
|
|
return False
|
|
|
|
if isinstance(result, dict):
|
|
if result.get("success") is False:
|
|
return False
|
|
if result.get("staged") is True:
|
|
return False
|
|
if "error" in result and result.get("success") is not True:
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def collect_memory_write_notifications(
|
|
tool_result: Any,
|
|
tool_args: Dict[str, Any],
|
|
) -> List[Dict[str, str]]:
|
|
"""Return provider notifications for a successful built-in memory write."""
|
|
if not _memory_tool_result_succeeded(tool_result):
|
|
return []
|
|
|
|
target = str(tool_args.get("target") or "memory")
|
|
operations = tool_args.get("operations")
|
|
if isinstance(operations, list) and operations:
|
|
raw_operations = operations
|
|
else:
|
|
raw_operations = [{
|
|
"action": tool_args.get("action"),
|
|
"content": tool_args.get("content"),
|
|
"old_text": tool_args.get("old_text"),
|
|
}]
|
|
|
|
notifications: List[Dict[str, str]] = []
|
|
for op in raw_operations:
|
|
if not isinstance(op, dict):
|
|
continue
|
|
action = str(op.get("action") or "")
|
|
if action not in _MIRRORED_MEMORY_ACTIONS:
|
|
continue
|
|
notifications.append({
|
|
"action": action,
|
|
"target": target,
|
|
"content": str(op.get("content") or ""),
|
|
"old_text": str(op.get("old_text") or ""),
|
|
})
|
|
return notifications
|