File size: 6,733 Bytes
dc893fb |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 |
"""Session Note Tool - Let agent record and recall important information.
This tool allows the agent to:
- Record key points and important information during sessions
- Recall previously recorded notes
- Maintain context across agent execution chains
"""
import json
from datetime import datetime
from pathlib import Path
from typing import Any
from .base import Tool, ToolResult
class SessionNoteTool(Tool):
"""Tool for recording and recalling session notes.
The agent can use this tool to:
- Record important facts, decisions, or context during sessions
- Recall information from previous sessions
- Build up knowledge over time
Example usage by agent:
- record_note("User prefers concise responses")
- record_note("Project uses Python 3.12 and async/await")
- recall_notes() -> retrieves all recorded notes
"""
def __init__(self, memory_file: str = "./workspace/.agent_memory.json"):
"""Initialize session note tool.
Args:
memory_file: Path to the note storage file
"""
self.memory_file = Path(memory_file)
# Lazy loading: file and directory are only created when first note is recorded
@property
def name(self) -> str:
return "record_note"
@property
def description(self) -> str:
return (
"Record important information as session notes for future reference. "
"Use this to record key facts, user preferences, decisions, or context "
"that should be recalled later in the agent execution chain. Each note is timestamped."
)
@property
def parameters(self) -> dict[str, Any]:
return {
"type": "object",
"properties": {
"content": {
"type": "string",
"description": "The information to record as a note. Be concise but specific.",
},
"category": {
"type": "string",
"description": "Optional category/tag for this note (e.g., 'user_preference', 'project_info', 'decision')",
},
},
"required": ["content"],
}
def _load_from_file(self) -> list:
"""Load notes from file.
Returns empty list if file doesn't exist (lazy loading).
"""
if not self.memory_file.exists():
return []
try:
return json.loads(self.memory_file.read_text())
except Exception:
return []
def _save_to_file(self, notes: list):
"""Save notes to file.
Creates parent directory and file if they don't exist (lazy initialization).
"""
# Ensure parent directory exists when actually saving
self.memory_file.parent.mkdir(parents=True, exist_ok=True)
self.memory_file.write_text(json.dumps(notes, indent=2, ensure_ascii=False))
async def execute(self, content: str, category: str = "general") -> ToolResult:
"""Record a session note.
Args:
content: The information to record
category: Category/tag for this note
Returns:
ToolResult with success status
"""
try:
# Load existing notes
notes = self._load_from_file()
# Add new note with timestamp
note = {
"timestamp": datetime.now().isoformat(),
"category": category,
"content": content,
}
notes.append(note)
# Save back to file
self._save_to_file(notes)
return ToolResult(
success=True,
content=f"Recorded note: {content} (category: {category})",
)
except Exception as e:
return ToolResult(
success=False,
content="",
error=f"Failed to record note: {str(e)}",
)
class RecallNoteTool(Tool):
"""Tool for recalling recorded session notes."""
def __init__(self, memory_file: str = "./workspace/.agent_memory.json"):
"""Initialize recall note tool.
Args:
memory_file: Path to the note storage file
"""
self.memory_file = Path(memory_file)
@property
def name(self) -> str:
return "recall_notes"
@property
def description(self) -> str:
return (
"Recall all previously recorded session notes. "
"Use this to retrieve important information, context, or decisions "
"from earlier in the session or previous agent execution chains."
)
@property
def parameters(self) -> dict[str, Any]:
return {
"type": "object",
"properties": {
"category": {
"type": "string",
"description": "Optional: filter notes by category",
},
},
}
async def execute(self, category: str = None) -> ToolResult:
"""Recall session notes.
Args:
category: Optional category filter
Returns:
ToolResult with notes content
"""
try:
if not self.memory_file.exists():
return ToolResult(
success=True,
content="No notes recorded yet.",
)
notes = json.loads(self.memory_file.read_text())
if not notes:
return ToolResult(
success=True,
content="No notes recorded yet.",
)
# Filter by category if specified
if category:
notes = [n for n in notes if n.get("category") == category]
if not notes:
return ToolResult(
success=True,
content=f"No notes found in category: {category}",
)
# Format notes for display
formatted = []
for idx, note in enumerate(notes, 1):
timestamp = note.get("timestamp", "unknown time")
cat = note.get("category", "general")
content = note.get("content", "")
formatted.append(f"{idx}. [{cat}] {content}\n (recorded at {timestamp})")
result = "Recorded Notes:\n" + "\n".join(formatted)
return ToolResult(success=True, content=result)
except Exception as e:
return ToolResult(
success=False,
content="",
error=f"Failed to recall notes: {str(e)}",
)
|