MemOS/src/memos/api/handlers/feedback_handler.py at 129caaaa31fe11984a280e5cb76d7fb62cdabe3f · MemTensor/MemOS · GitHub

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
"""
Feeback handler for memory add/update functionality.
"""

from memos.api.handlers.base_handler import BaseHandler, HandlerDependencies
from memos.api.handlers.cube_scope import resolve_cube_ids
from memos.api.product_models import APIFeedbackRequest, MemoryResponse
from memos.log import get_logger
from memos.multi_mem_cube.composite_cube import CompositeCubeView
from memos.multi_mem_cube.single_cube import SingleCubeView
from memos.multi_mem_cube.views import MemCubeView


logger = get_logger(__name__)


class FeedbackHandler(BaseHandler):
    """
    Handler for memory feedback operations.

    Provides fast, fine-grained, and mixture-based feedback modes.
    """

    def __init__(self, dependencies: HandlerDependencies):
        """
        Initialize feedback handler.

        Args:
            dependencies: HandlerDependencies instance
        """
        super().__init__(dependencies)
        self._validate_dependencies("mem_reader", "mem_scheduler", "searcher", "reranker")

    def handle_feedback_memories(self, feedback_req: APIFeedbackRequest) -> MemoryResponse:
        """
        Main handler for feedback memories endpoint.

        Args:
            feedback_req: feedback request containing content and parameters

        Returns:
            MemoryResponse with formatted results
        """
        cube_view = self._build_cube_view(feedback_req)

        process_record = cube_view.feedback_memories(feedback_req)

        self.logger.info(f"[FeedbackHandler] Final feedback results count={len(process_record)}")

        return MemoryResponse(
            message="Memory feedback successfully",
            data=[process_record],
        )

    def _resolve_cube_ids(self, feedback_req: APIFeedbackRequest) -> list[str]:
        """
        Normalize target cube ids from feedback_req.
        """
        return resolve_cube_ids(feedback_req.writable_cube_ids, feedback_req.user_id)

    def _build_cube_view(self, feedback_req: APIFeedbackRequest) -> MemCubeView:
        cube_ids = self._resolve_cube_ids(feedback_req)

        if len(cube_ids) == 1:
            cube_id = cube_ids[0]
            return SingleCubeView(
                cube_id=cube_id,
                naive_mem_cube=None,
                mem_reader=None,
                mem_scheduler=self.mem_scheduler,
                logger=self.logger,
                searcher=None,
                feedback_server=self.feedback_server,
            )
        else:
            single_views = [
                SingleCubeView(
                    cube_id=cube_id,
                    naive_mem_cube=None,
                    mem_reader=None,
                    mem_scheduler=self.mem_scheduler,
                    logger=self.logger,
                    searcher=None,
                    feedback_server=self.feedback_server,
                )
                for cube_id in cube_ids
            ]
            return CompositeCubeView(
                cube_views=single_views,
                logger=self.logger,
            )