Maas-Agent/apps/application/views/application_memory.py

# coding=utf-8
"""
智能体记忆视图
提供记忆的 CRUD API
"""
from django.utils.translation import gettext as _
from rest_framework.request import Request
from rest_framework.views import APIView

from common.auth import TokenAuth
from common.exception.app_exception import AppApiException
from common.result import result
from application.services.application_memory_service import ApplicationMemoryService


class ApplicationMemoryView(APIView):
    """智能体记忆管理"""
    authentication_classes = [TokenAuth]

    class List(APIView):
        """获取记忆列表"""
        authentication_classes = [TokenAuth]

        def get(self, request: Request, application_id: str):
            memory_type = request.query_params.get('memory_type')
            is_enabled = request.query_params.get('is_enabled', 'true') == 'true'
            limit = int(request.query_params.get('limit', 100))

            memories = ApplicationMemoryService.get_memories(
                application_id=application_id,
                memory_type=memory_type,
                is_enabled=is_enabled,
                limit=limit
            )

            data = [{
                'id': str(m.id),
                'memory_type': m.memory_type,
                'content': m.content,
                'metadata': m.metadata,
                'is_enabled': m.is_enabled,
                'relevance_score': m.relevance_score,
                'create_time': m.create_time.isoformat() if m.create_time else None,
                'update_time': m.update_time.isoformat() if m.update_time else None,
            } for m in memories]

            return result.success(data)

    class Create(APIView):
        """创建记忆"""
        authentication_classes = [TokenAuth]

        def post(self, request: Request, application_id: str):
            content = request.data.get('content', '')
            memory_type = request.data.get('memory_type', 'DIALOGUE')
            metadata = request.data.get('metadata', {})

            if not content:
                raise AppApiException(400, _('记忆内容不能为空'))

            memory = ApplicationMemoryService.create_memory(
                application_id=application_id,
                content=content,
                memory_type=memory_type,
                metadata=metadata
            )

            return result.success({
                'id': str(memory.id),
                'memory_type': memory.memory_type,
                'content': memory.content,
                'metadata': memory.metadata,
                'is_enabled': memory.is_enabled,
                'relevance_score': memory.relevance_score,
                'create_time': memory.create_time.isoformat() if memory.create_time else None,
            })

    class Operate(APIView):
        """记忆操作（更新/删除）"""
        authentication_classes = [TokenAuth]

        def put(self, request: Request, application_id: str, memory_id: str):
            content = request.data.get('content')
            memory_type = request.data.get('memory_type')
            is_enabled = request.data.get('is_enabled')
            relevance_score = request.data.get('relevance_score')
            metadata = request.data.get('metadata')

            memory = ApplicationMemoryService.update_memory(
                memory_id=memory_id,
                content=content,
                memory_type=memory_type,
                is_enabled=is_enabled,
                relevance_score=relevance_score,
                metadata=metadata
            )

            if not memory:
                raise AppApiException(404, _('记忆不存在'))

            return result.success({
                'id': str(memory.id),
                'memory_type': memory.memory_type,
                'content': memory.content,
                'metadata': memory.metadata,
                'is_enabled': memory.is_enabled,
                'relevance_score': memory.relevance_score,
                'update_time': memory.update_time.isoformat() if memory.update_time else None,
            })

        def delete(self, request: Request, application_id: str, memory_id: str):
            success = ApplicationMemoryService.delete_memory(memory_id)
            if not success:
                raise AppApiException(404, _('记忆不存在'))
            return result.success(True)

    class BatchDelete(APIView):
        """批量删除记忆"""
        authentication_classes = [TokenAuth]

        def post(self, request: Request, application_id: str):
            memory_ids = request.data.get('memory_ids', [])
            if not memory_ids:
                raise AppApiException(400, _('请选择要删除的记忆'))

            count = ApplicationMemoryService.batch_delete_memories(memory_ids)
            return result.success({'deleted_count': count})

    class Search(APIView):
        """搜索记忆"""
        authentication_classes = [TokenAuth]

        def get(self, request: Request, application_id: str):
            query = request.query_params.get('query', '')
            limit = int(request.query_params.get('limit', 10))

            if not query:
                raise AppApiException(400, _('搜索关键词不能为空'))

            memories = ApplicationMemoryService.search_memories(
                application_id=application_id,
                query=query,
                limit=limit
            )

            data = [{
                'id': str(m.id),
                'memory_type': m.memory_type,
                'content': m.content,
                'metadata': m.metadata,
                'relevance_score': m.relevance_score,
                'create_time': m.create_time.isoformat() if m.create_time else None,
            } for m in memories]

            return result.success(data)