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

from __future__ import annotations

from typing import Any

from ..config import settings
from ..schemas import ChatResponse, SystemDescriptor
from .base import AdapterUnavailable, MemoryAgent


class LettaAgent(MemoryAgent):
    id = "letta"

    def __init__(self) -> None:
        super().__init__()
        self._client: Any | None = None
        self._agent_id: str | None = None
        self._import_error: str | None = None
        try:
            from letta_client import Letta  # type: ignore

            self._client_class = Letta
        except Exception as exc:
            self._client_class = None
            self._import_error = str(exc)
        ready = bool(self._client_class and settings.letta_enabled)
        self.descriptor = SystemDescriptor(
            id="letta",
            name="Letta",
            paradigm="有状态 Agent 与虚拟内存",
            description="用 Core、Archival、Recall 三层状态，让 Agent 在有限上下文和外部记忆间协作。",
            available=ready,
            mode="real-sdk" if ready else "unavailable",
            status="ready" if ready else "not-configured",
            package="letta-client",
            setup_hint=None if ready else (
                "当前仅保留 Letta 适配骨架，请保持 LETTA_ENABLED=false。"
                "完成 Agent ID 恢复、分层记忆投影与删除闭环后再启用。"
            ),
        )

    def _ensure_client(self) -> Any:
        if not self._client_class or not settings.letta_enabled:
            raise AdapterUnavailable(self.descriptor.setup_hint or "Letta SDK 不可用")
        if self._client is None:
            kwargs = {"base_url": settings.letta_base_url}
            if settings.letta_api_key:
                kwargs["token"] = settings.letta_api_key
            self._client = self._client_class(**kwargs)
        return self._client

    async def chat(self, message: str) -> ChatResponse:
        client = self._ensure_client()
        try:
            if self._agent_id is None:
                agent = client.agents.create(
                    name="memory-agents-solutions",
                    memory_blocks=[
                        {"label": "persona", "value": "你是一个严谨的 AI 编程助手。"},
                        {"label": "workspace", "value": f"工作区：{settings.workspace_id}"},
                    ],
                )
                self._agent_id = agent.id
            response = client.agents.messages.create(
                agent_id=self._agent_id,
                messages=[{"role": "user", "content": message}],
            )
            if hasattr(response, "__await__"):
                response = await response
            answer = str(response)
        except Exception as exc:
            await self._audit("LETTA/message", status="error", details={"error": str(exc)})
            raise AdapterUnavailable(f"Letta 调用失败：{exc}") from exc
        await self._audit("LETTA/message", details={"agent_id": self._agent_id})
        return ChatResponse(
            system="letta", answer=answer, mode="real-sdk",
            memory_context=await self.memories(), memory_events=[{"event": "agent_message", "agent_id": self._agent_id}],
            audit_events=await self.audit(),
        )