ai-agent/MemoryAgents/backend/app/adapters/base.py

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Any

from ..config import settings
from ..db import repository, utcnow
from ..llm import llm
from ..schemas import AuditEvent, ChatResponse, MemoryItem, SystemDescriptor


class AdapterUnavailable(RuntimeError):
    pass


class MemoryAgent(ABC):
    """所有记忆系统必须遵守的统一接口。"""
    id: str
    descriptor: SystemDescriptor

    def __init__(self) -> None:
        self.repo = repository
        self.llm = llm

    @property
    def status(self) -> SystemDescriptor:
        return self.descriptor

    async def memories(self, query: str | None = None) -> list[MemoryItem]:
        return await self.repo.list_memories(self.id, query)

    async def audit(self) -> list[AuditEvent]:
        return await self.repo.list_audit(self.id)

    async def reset(self) -> None:
        await self.repo.reset(self.id)

    async def delete_memory(self, memory_id: str) -> bool:
        """删除记忆时同步清理向量和记录审计。"""
        deleted = await self.repo.delete_memory(self.id, memory_id)
        if deleted:
            await self.repo.delete_embedding(self.id, memory_id)
            await self._audit("DELETE/Memory", target=memory_id)
        return deleted

    async def _audit(
        self,
        operation: str,
        target: str | None = None,
        status: str = "ok",
        details: dict[str, Any] | None = None,
    ) -> AuditEvent:
        """写入一条适配器操作记录。"""
        event = AuditEvent(
            id=self.repo.memory_id("audit"),
            system=self.id,
            workspace_id=settings.workspace_id,
            operation=operation,
            target=target,
            status=status,
            details=details or {},
            created_at=utcnow(),
        )
        return await self.repo.add_audit(event)

    @abstractmethod
    async def chat(self, message: str) -> ChatResponse:
        raise NotImplementedError