Oleksandr Kozachuk
96 lines
4.4 KiB
Python
96 lines
4.4 KiB
Python
"""Structured memory manager (SPEC-002).
|
|
|
|
Observations in, facts + episodes out — via a batched, lock-guarded
|
|
consolidation pass on `memory-model`. Recall assembly is
|
|
participant-scoped (MEM-04): the model never sees facts of users who
|
|
are not part of the conversation.
|
|
"""
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import Any, Awaitable, Callable, Dict, List, Optional
|
|
|
|
from .persistence import PersistentStore
|
|
|
|
Consolidator = Callable[[List[Dict[str, Any]], List[Dict[str, Any]]], Awaitable[Optional[Dict[str, Any]]]]
|
|
|
|
DEFAULT_CONSOLIDATE_EVERY = 20
|
|
DEFAULT_EPISODES_PER_CHANNEL = 10
|
|
DEFAULT_FACT_RETENTION_DAYS = 180
|
|
OBSERVATION_EXCERPT = 500
|
|
|
|
|
|
class MemoryManager:
|
|
def __init__(
|
|
self,
|
|
store: Optional[PersistentStore],
|
|
config_getter: Callable[[], Dict[str, Any]],
|
|
consolidator: Consolidator,
|
|
channel: str,
|
|
) -> None:
|
|
self.store = store
|
|
self._config = config_getter
|
|
self._consolidator = consolidator
|
|
self.channel = channel
|
|
self._lock = asyncio.Lock()
|
|
|
|
def active(self) -> bool:
|
|
return self.store is not None and "memory-model" in self._config()
|
|
|
|
async def observe(self, user: str, kind: str, content: str) -> None:
|
|
"""Record one event; trigger consolidation when the batch is full (MEM-01/02)."""
|
|
if not self.active():
|
|
return
|
|
assert self.store is not None
|
|
await asyncio.to_thread(self.store.add_observation, self.channel, user, kind, str(content)[:OBSERVATION_EXCERPT])
|
|
every = int(self._config().get("memory-consolidate-every", DEFAULT_CONSOLIDATE_EVERY))
|
|
if await asyncio.to_thread(self.store.unconsumed_observations, self.channel) >= every:
|
|
asyncio.get_running_loop().create_task(self.consolidate_now())
|
|
|
|
async def consolidate_now(self) -> None:
|
|
"""One batched pass: observations -> self-authored facts + episode (MEM-02/03/05/06)."""
|
|
if not self.active() or self._lock.locked():
|
|
return
|
|
assert self.store is not None
|
|
async with self._lock:
|
|
observations = await asyncio.to_thread(self.store.peek_observations, self.channel)
|
|
if not observations:
|
|
return
|
|
authors = {observation["user"] for observation in observations}
|
|
known_facts = await asyncio.to_thread(self.store.facts_for, sorted(authors))
|
|
result = await self._consolidator(observations, known_facts)
|
|
if result is None:
|
|
return # model call failed — observations stay for the next trigger
|
|
for fact in result.get("facts", []):
|
|
if fact.get("user") in authors:
|
|
await asyncio.to_thread(self.store.add_user_fact, fact["user"], str(fact["fact"]), "self")
|
|
else:
|
|
logging.warning(f"memory: dropped third-party fact about {fact.get('user')!r} (MEM-03)")
|
|
episode = result.get("episode")
|
|
if episode:
|
|
await asyncio.to_thread(self.store.add_episode, self.channel, str(episode))
|
|
await asyncio.to_thread(self.store.consume_observations, self.channel, observations[-1]["id"])
|
|
config = self._config()
|
|
await asyncio.to_thread(
|
|
self.store.trim_episodes, self.channel, int(config.get("memory-episodes-per-channel", DEFAULT_EPISODES_PER_CHANNEL))
|
|
)
|
|
await asyncio.to_thread(self.store.purge_old_facts, int(config.get("memory-fact-retention-days", DEFAULT_FACT_RETENTION_DAYS)))
|
|
|
|
def memory_block(self, participants: List[str], legacy: str) -> str:
|
|
"""Assemble the {memory} block, participant-scoped (MEM-04/10)."""
|
|
if not self.active():
|
|
return legacy
|
|
assert self.store is not None
|
|
sections: List[str] = []
|
|
pinned = self.store.pinned_for(self.channel)
|
|
if pinned:
|
|
sections.append("Operator notes:\n" + "\n".join(f"- {pin['fact']}" for pin in pinned))
|
|
facts = self.store.facts_for(sorted(set(participants)))
|
|
if facts:
|
|
sections.append("What users told about themselves:\n" + "\n".join(f"- {fact['user']}: {fact['fact']}" for fact in facts))
|
|
config = self._config()
|
|
episodes = self.store.recent_episodes(self.channel, int(config.get("memory-episodes-per-channel", DEFAULT_EPISODES_PER_CHANNEL)))
|
|
if episodes:
|
|
sections.append("Recent conversation summaries:\n" + "\n".join(f"- {episode}" for episode in episodes))
|
|
return "\n\n".join(sections)
|