Spaces:
Running
Running
File size: 2,632 Bytes
4ef118d | 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 | """
Memory helper utilities for Qurio backend.
Provides helpers for running AGNO memory optimizations and managing the
default lite agent used when memory calls must stay lightweight.
"""
from __future__ import annotations
from typing import Any
from agno.memory.strategies.types import MemoryOptimizationStrategyType
from agno.utils.log import logger
from .agent_registry import build_memory_agent
def _resolve_strategy(value: str | None) -> MemoryOptimizationStrategyType:
if not value:
return MemoryOptimizationStrategyType.SUMMARIZE
normalized = value.upper()
if normalized in MemoryOptimizationStrategyType.__members__:
return MemoryOptimizationStrategyType[normalized]
for member in MemoryOptimizationStrategyType:
if member.name.lower() == normalized.lower() or member.value.lower() == normalized.lower():
return member
return MemoryOptimizationStrategyType.SUMMARIZE
def optimize_user_memories(
user_id: str,
strategy: str | None = None,
apply: bool = True,
provider: str | None = None,
model: str | None = None,
base_url: str | None = None,
api_key: str | None = None,
) -> dict[str, Any]:
if not user_id:
logger.warning("Memory optimization skipped: missing user_id.")
return {"success": False, "message": "user_id is required"}
logger.info(
"Optimize memories request: user=%s strategy=%s provider=%s model=%s",
user_id,
strategy or "SUMMARIZE",
provider,
model,
)
agent = build_memory_agent(
user_id=user_id,
provider=provider,
model=model,
base_url=base_url,
api_key=api_key,
)
memory_manager = getattr(agent, "memory_manager", None)
if not memory_manager:
logger.warning("Memory optimization skipped: memory manager unavailable.")
return {"success": False, "message": "memory manager unavailable"}
resolved_strategy = _resolve_strategy(strategy)
try:
optimized = memory_manager.optimize_memories(
user_id=user_id,
strategy=resolved_strategy,
apply=apply,
)
logger.info(
"Memory optimization completed for user=%s strategy=%s apply=%s (%s entries)",
user_id,
resolved_strategy.name,
apply,
len(optimized or []),
)
return {"success": True, "strategy": resolved_strategy.name, "result": optimized}
except Exception as exc:
logger.error("Memory optimization failed: %s", exc)
return {"success": False, "message": str(exc)}
|