backend-python/src/services/memory_service.py

"""
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)}