| import logging |
| import json |
|
|
| from ..core import BaseService |
| from ..memory import MemoryEngine |
|
|
| logger = logging.getLogger("qkernel.executors.memory") |
|
|
|
|
| class MemoryExecutor(BaseService): |
| name: str = "memory" |
|
|
| def __init__(self, engine: MemoryEngine | None = None) -> None: |
| self.engine = engine or MemoryEngine() |
|
|
| async def process(self, event: dict) -> dict: |
| action = event.get("action", "read") |
| user_id = event.get("user_id", "anonymous") |
| session_id = event.get("session_id", "default") |
|
|
| if action == "store": |
| content = event.get("content", "") |
| role = event.get("role", "user") |
| if not content.strip(): |
| return {"error": "empty_content", "action": action} |
| self.engine.write(user_id, session_id, role, content) |
| return {"action": action, "stored": True, "length": len(content)} |
|
|
| if action == "search": |
| query = event.get("query", "") |
| limit = int(event.get("limit", 10)) |
| records = self.engine.read(user_id, session_id, limit=limit) |
| if query.strip(): |
| records = [ |
| r for r in records if query.lower() in r.get("content", "").lower() |
| ] |
| return {"action": action, "results": records} |
|
|
| if action == "stats": |
| return {"action": action, "stats": self.engine.get_stats()} |
|
|
| records = self.engine.read(user_id, session_id) |
| return { |
| "action": "read", |
| "records": records, |
| "note": json.dumps({"user_id": user_id, "session_id": session_id}), |
| } |
