Spaces:
Sleeping
Sleeping
File size: 1,894 Bytes
06ce7ac | 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 | """Memory Agent: ?????? ?????? ???????."""
import logging
from typing import Any, Dict
from memory.short_term import ShortTermMemory
from memory.episodic_db import EpisodicDB
from memory.knowledge_vector import KnowledgeVectorStore
from api.deps import get_logger, is_remote_brain_only
from agents.supervisor_agent import AgentOS
logger = get_logger("kapo.agent.memory")
class MemoryAgent:
def __init__(self):
self.remote_only = is_remote_brain_only()
self.short = ShortTermMemory()
self.episodic = None if self.remote_only else EpisodicDB()
self.knowledge = None if self.remote_only else KnowledgeVectorStore()
self.agent_os = None if self.remote_only else AgentOS()
def write_short_term(self, key: str, value: Dict[str, Any]) -> None:
if self.remote_only:
return
self.short.set(key, value)
def read_short_term(self, key: str) -> Dict[str, Any] | None:
if self.remote_only:
return None
return self.short.get(key)
def store_experience(self, payload: Dict[str, Any]) -> None:
if self.remote_only or self.episodic is None:
return
self.episodic.insert_experience(
task=payload.get("task", ""),
plan=payload.get("plan", {}),
tools_used=payload.get("tools_used", {}),
result=payload.get("result", {}),
success=1 if payload.get("success") else 0,
)
def query_knowledge(self, text: str, top_k: int = 3):
if self.remote_only or self.knowledge is None:
return []
return self.knowledge.query(text, top_k=top_k)
def run_agent_os(self):
"""????? ???? ??????? ?????? ??????."""
if self.remote_only or self.agent_os is None:
return {"skipped": True, "reason": "REMOTE_BRAIN_ONLY"}
return self.agent_os.run()
|