Spaces:
Running
Running
File size: 2,570 Bytes
a31f556 | 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 | # backend/memory/procedural_memory.py
# Learned task patterns stored in MongoDB
import time
import uuid
from typing import Optional
from backend.db.mongodb import MongoDBClient
class ProceduralMemory:
def __init__(self, db_path: str = None):
# db_path is ignored now since we use MongoDB
self.collection = MongoDBClient.get_db().skills
async def record_success(self, task_name: str, steps: list[dict]):
now = time.time()
# Check if skill exists
skill = await self.collection.find_one({'name': task_name})
if skill:
count = skill.get('success_count', 0) + 1
await self.collection.update_one(
{'_id': skill['_id']},
{'$set': {
'success_count': count,
'last_used': now,
'steps': steps
}}
)
else:
skill_id = str(uuid.uuid4())
# Default trigger pattern is just the task name
triggers = [task_name.lower()]
await self.collection.insert_one({
'_id': skill_id,
'name': task_name,
'trigger_patterns': triggers,
'steps': steps,
'success_count': 1,
'last_used': now
})
async def find_matching_skill(self, user_input: str) -> Optional[dict]:
user_input_lower = user_input.lower()
# We fetch all skills and match. In a massive DB we'd use text search,
# but for procedural memory triggers, exact substring matching is fine.
cursor = self.collection.find({})
async for skill in cursor:
try:
patterns = skill.get('trigger_patterns', [])
for p in patterns:
if p in user_input_lower:
# Found a match
best_match = {
"id": skill['_id'],
"name": skill['name'],
"steps": skill['steps']
}
# Update last_used
await self.collection.update_one(
{'_id': skill['_id']},
{'$set': {'last_used': time.time()}}
)
return best_match
except Exception:
continue
return None
|