Persistent NPC Memory and Relationships — Final Design¶
Version: 3.1
Status: Final
Author: Systems Architecture Team
Date: 2025-01-10
Priority: P1 — Strategic Differentiator
Executive Summary¶
MAID's AI-powered NPC dialogue system creates engaging conversations, but without persistent memory, NPCs forget players the moment a conversation ends. This fundamental limitation breaks immersion and wastes the potential of AI-driven characters who could develop meaningful, ongoing relationships with players.
This design introduces a Persistent NPC Memory and Relationships system that transforms NPCs from stateless dialogue endpoints into characters with genuine memory, opinions, and social networks. NPCs will remember past conversations, form lasting impressions of players, learn facts about the world, and share knowledge through gossip networks.
The system consists of three core subsystems:
- Memory System — Episodic, semantic, and procedural memory types with LLM-based extraction, storage in DocumentStore, and intelligent retrieval for prompt context
- Relationship System — Multi-dimensional relationship tracking (trust, fear, respect, loyalty, friendliness, romantic interest) with faction-based initialization and relationship progression
- Knowledge Graph — Belief system where NPCs learn and share facts about the world, players, and events through direct observation and gossip propagation
The implementation addresses critical concerns from design reviews: cost control through hard budget limits, content safety against player manipulation, performance optimization for real-world scale, and seamless integration with existing ConversationManager/PromptBuilder architecture.
This is MAID's strategic differentiator — every conversation becomes meaningful, reputation matters, and NPCs feel genuinely alive.
Problem Statement & Current State¶
What Exists Today¶
MAID's current AI dialogue system provides basic conversation capability:
| Component | File | Capability | Limitation |
|---|---|---|---|
| ConversationManager | ai/conversation.py |
Tracks message history per (player_id, npc_id) pair; saves conversations to storage | In-memory only during session; no memory extraction or relationship data |
| PromptBuilder | ai/prompts.py |
Assembles system prompt with NPC personality and world context | Has {reputation} placeholder but always hardcoded to "neutral" |
| NPCDialogueSystem | maid_classic_rpg/systems/npc/dialogue.py |
Handles dialogue events, builds context, calls LLM | No memory extraction; no relationship updates; each conversation starts fresh |
| DocumentStore | storage/document_store.py |
CRUD operations with query support; PostgreSQL JSONB backend | Infrastructure ready but unused for memory/relationship data |
| RateLimiter | ai/rate_limiter.py |
Per-player and global token budgets for dialogue | Exists but doesn't cover memory extraction LLM calls |
The Problem¶
- Zero continuity — NPCs have no memory of past interactions, breaking immersion
- Static relationships — No progression from stranger to friend/enemy over multiple encounters
- No social context — NPCs cannot reference what they've learned about players or the world
- Wasted narrative potential — Rich conversations generate no lasting impact on the game state
- Missing emergent gameplay — No reputation system, faction relationships, or social consequences
Success Criteria¶
- NPCs remember and reference specific past interactions
- Player reputation affects how NPCs respond across the entire game world
- NPC relationships evolve meaningfully based on player actions
- Memory system operates within strict cost and performance bounds
- Content safety prevents player manipulation of NPC memory
Architecture Overview¶
NPCDialogueSystem
│
┌───────────┴───────────┐
▼ ▼
┌──── Write Pipeline ────┐ ┌──── Read Pipeline ─────────────┐
│ (async/bounded queue) │ │ (async orchestration, │
│ │ │ sync formatting, 40ms timeout)│
│ Conversation ──► │ │ │
│ MemoryExtraction │ │ EnrichedPromptBuilder │
│ Job (idempotent) │ │ ├─ ContextOrchestrator │
│ │ │ │ │ ├─ TokenBudgetManager │
│ ▼ │ │ │ ├─ MemoryContextProvider │
│ MemoryService │ │ │ ├─ RelationshipCtxProv. │
│ RelationshipMgr │ │ │ └─ KnowledgeCtxProvider │
│ KnowledgeGraphMgr │ │ └─ PromptBuilder (sync) │
│ │ │ │ │ │
│ ▼ │ │ ▼ │
│ DocumentStore │ │ Enhanced System Prompt │
└────────────────────────┘ │ (wires {reputation} placeholder)│
└─────────────────────────────────┘
▲
│
┌─── Structured Access ──┐
│ MemoryQueryAPI │ ← Doc 07 UtilityScorer,
│ (typed Python objects) │ Doc 08 Quest Generator
└─────────────────────────┘
Package Architecture¶
The system is split across two packages following MAID's layered architecture. maid-engine provides content-agnostic infrastructure (protocols, caching, safety). maid-stdlib provides the game-domain implementations that use those protocols.
packages/maid-engine/src/maid_engine/
├── ai/context_providers.py # ContextProvider protocol, ContextSection, ContextOrchestrator
├── ai/prompts.py # PromptBuilder (sync formatter, existing signature preserved)
├── ai/conversation.py # ConversationManager (extended with extraction hooks)
├── ai/pii.py # PIIRedactor (pre-processing)
├── ai/safety.py # ContentFilter (existing)
├── ai/cache.py # MemoryCache (generic LRU+TTL, concurrency-safe)
└── ai/token_budget.py # TokenBudgetManager (authoritative budget allocation)
packages/maid-stdlib/src/maid_stdlib/
├── memory/ # Bounded context: Memory (game-domain)
│ ├── __init__.py
│ ├── models.py # BaseMemory, EpisodicMemory, SemanticMemory, ProceduralMemory
│ ├── service.py # MemoryService — memory CRUD and retrieval
│ ├── extraction.py # LLM-based memory extraction + idempotency
│ ├── working_buffer.py # WorkingMemoryBuffer (last 3 conversations, persisted)
│ ├── context_provider.py # MemoryContextProvider (implements ContextProvider)
│ └── prompts.py # Memory extraction prompts
│
├── relationships/ # Bounded context: Relationships (game-domain)
│ ├── __init__.py
│ ├── models.py # Relationship, RelationshipStage, FactionRelationship
│ ├── manager.py # RelationshipManager
│ ├── context_provider.py # RelationshipContextProvider (implements ContextProvider)
│ └── events.py # Relationship threshold events (NPCHostileEvent, etc.)
│
├── knowledge/ # Bounded context: Knowledge (game-domain)
│ ├── __init__.py
│ ├── models.py # KnowledgeFact, GossipMessage
│ ├── manager.py # KnowledgeGraphManager
│ ├── gossip.py # GossipSystem for knowledge sharing
│ └── context_provider.py # KnowledgeContextProvider (implements ContextProvider)
│
└── memory_bridge.py # MemoryEventBridge — relationship → game events
Layering rule:
maid-stdlibimports frommaid-engine(protocols, cache, safety).maid-enginenever imports frommaid-stdlib. ConcreteContextProviderimplementations live inmaid-stdlib; the protocol and orchestration live inmaid-engine.
Each bounded context owns its models, storage logic, and context provider. Contexts communicate only through well-defined service interfaces and events — never by importing each other's models directly.
Integration Points¶
- NPCDialogueSystem subscribes to conversation completion, submits async
MemoryExtractionJobto boundedExtractionJobQueue - EnrichedPromptBuilder composes
ContextOrchestrator(async, resolves providers concurrently viaasyncio.gather()withTokenBudgetManagerbudgets) andPromptBuilder(sync, formats prompt). Existing syncPromptBuilder.build_system_prompt()signature is preserved —EnrichedPromptBuildercalls it internally. - MemoryService, RelationshipManager, KnowledgeGraphManager each own their storage via DocumentStore collections (all in
maid-stdlib) - GossipSystem runs as background ECS system for knowledge propagation (in
maid-stdlib), conforming to theSystemABC (update(),__init__(world)) - MemoryRateLimiter (standalone token-bucket) covers memory extraction and consolidation LLM calls
- MemoryEventBridge translates relationship threshold crossings into game events via EventBus, and maps them to typed
StorySignalvariants (RELATIONSHIP_CHANGED, SECRET_LEARNED, etc.) for quest/narrative integration - RelationshipEvaluator bridges extracted memories to
RelationshipManager.update_relationship()using tag-based dimension mapping - MemoryConsolidator runs as a scheduled batch job to summarise episodic memory clusters into semantic memories
- MemoryQueryAPI provides structured, logic-based facade for game systems (Doc 07
UtilityScorer, Doc 08 quest generator) — complements the natural-languageContextProviderpipeline
Memory System¶
Memory Types¶
Following established psychological frameworks, we implement three memory types sharing
a common BaseMemory with a unified salience score for retrieval ranking.
BaseMemory¶
All memory types inherit from BaseMemory to eliminate field duplication and provide
a single salience score for cross-type retrieval ranking:
@dataclass
class BaseMemory:
"""Shared fields for all memory types."""
id: uuid.UUID
npc_id: uuid.UUID
player_id: uuid.UUID | None # None for world-scoped knowledge
memory_type: MemoryType # EPISODIC | SEMANTIC | PROCEDURAL
content: str # Human-readable description
tags: list[str] # Searchable tags
importance: float # 0.0-1.0 (author-assigned weight)
confidence: float # 0.0-1.0 (certainty / reinforcement)
salience: float # 0.0-1.0 (computed: importance × recency × access_boost)
created_at: datetime
last_accessed: datetime
access_count: int # For decay calculations
locked: bool = False # True = quest/faction-locked, never decayed or evicted
safety_flag: SafetyFlag = SafetyFlag.CLEAN # CLEAN | SUSPECT | REDACTED
source_conversation_id: uuid.UUID | None = None
class SafetyFlag(Enum):
CLEAN = "clean"
SUSPECT = "suspect" # Flagged by content filter, retained for audit
REDACTED = "redacted" # PII removed
class MemoryType(Enum):
EPISODIC = "episodic"
SEMANTIC = "semantic"
PROCEDURAL = "procedural"
def compute_salience(memory: BaseMemory, now: datetime) -> float:
"""Unified salience score for retrieval ranking across all memory types.
Formula: importance × recency_factor × access_boost
- recency_factor decays exponentially: exp(-age_days / 30)
- access_boost: 1.0 + log(1 + access_count) * 0.1 (capped at 1.5)
- locked memories always return salience 1.0
"""
if memory.locked:
return 1.0
age_days = (now - memory.last_accessed).total_seconds() / 86400
recency = math.exp(-age_days / 30)
access_boost = min(1.0 + math.log(1 + memory.access_count) * 0.1, 1.5)
return min(memory.importance * recency * access_boost, 1.0)
Episodic Memory¶
Specific interactions and events with clear temporal context.
@dataclass
class EpisodicMemory(BaseMemory):
"""Extends BaseMemory with episode-specific fields."""
memory_type: MemoryType = field(default=MemoryType.EPISODIC, init=False)
location: str = "" # "Market Square"
emotional_valence: EmotionalValence = EmotionalValence.NEUTRAL
emotional_intensity: float = 0.0 # 0.0-1.0
Semantic Memory¶
General facts and knowledge derived from multiple episodes.
@dataclass
class SemanticMemory(BaseMemory):
"""Extends BaseMemory with fact-specific fields."""
memory_type: MemoryType = field(default=MemoryType.SEMANTIC, init=False)
source_episodes: list[uuid.UUID] = field(default_factory=list)
Procedural Memory¶
Behavioral patterns and preferences learned from repeated interactions.
class ProceduralTriggerContext(TypedDict, total=False):
"""Typed context for procedural memory triggers (strict-mypy safe)."""
location_type: str # e.g., "shop", "tavern", "wilderness"
interaction_type: str # e.g., "trade", "combat", "dialogue"
time_of_day: str # e.g., "morning", "night"
present_entity_tags: list[str] # Tags of entities present when pattern occurs
@dataclass
class ProceduralMemory(BaseMemory):
"""Extends BaseMemory with pattern-specific fields."""
memory_type: MemoryType = field(default=MemoryType.PROCEDURAL, init=False)
pattern: str = "" # "Player always negotiates prices"
trigger_context: ProceduralTriggerContext = field(default_factory=lambda: ProceduralTriggerContext())
reinforcement_count: int = 0 # How many times pattern observed
Memory Storage¶
All memory types persist to a single DocumentStore collection with a memory_type
discriminator field, simplifying queries, indexing, and maintenance:
"npc_memories"— indexed by(npc_id, player_id, memory_type, created_at)
Memory Growth Bounds¶
Hard caps prevent unbounded memory growth per NPC-player pair:
| Memory Type | Max per NPC-Player Pair | Eviction Policy |
|---|---|---|
| Episodic | 200 | Lowest salience first (locked memories exempt) |
| Semantic | 50 | Lowest confidence first |
| Procedural | 20 | Lowest reinforcement_count first |
Decay formula (applied daily by background MemoryDecaySystem):
async def decay_memories(self, now: datetime) -> int:
"""Daily background job: recompute salience scores and prune excess memories.
Processes in batches by NPC-player pair to avoid loading all memories into
memory at once. Each pair is fetched, scored, pruned, then released.
Uses batch_update() and batch_delete() to minimise per-memory DB
round-trips. The DocumentStore batch API accepts a list of documents /
IDs and issues a single query per call.
"""
pruned = 0
# Iterate distinct (npc_id, player_id) pairs from the index
async for npc_id, player_id in self._store.iter_npc_player_pairs("npc_memories"):
pair_memories = await self._store.query(
"npc_memories",
filters={"npc_id": str(npc_id), "player_id": str(player_id)},
)
# Recompute salience for unlocked memories in this pair
updated: list[BaseMemory] = []
for mem in pair_memories:
if not mem.locked:
mem.salience = compute_salience(mem, now)
updated.append(mem)
if updated:
await self._store.batch_update("npc_memories", updated)
# Prune excess beyond caps (per memory type within this pair)
to_delete: list[uuid.UUID] = []
for memory_type, cap in MEMORY_CAPS.items():
typed = [m for m in pair_memories if m.memory_type.value == memory_type]
unlocked = [m for m in typed if not m.locked]
if len(typed) > cap:
excess = sorted(unlocked, key=lambda m: m.salience)[:len(typed) - cap]
to_delete.extend(mem.id for mem in excess)
if to_delete:
await self._store.batch_delete("npc_memories", to_delete)
pruned += len(to_delete)
return pruned
MEMORY_CAPS = {
"episodic": 200,
"semantic": 50,
"procedural": 20,
}
Memory locking: Memories created by quest systems, faction events, or admin commands
set locked=True. Locked memories are never decayed or evicted, ensuring narrative-critical
facts persist regardless of caps. The @memory lock <npc> <memory_id> admin command can
toggle locking manually.
Memory Extraction¶
LLM-Based Extraction (Primary)¶
After each conversation, the system extracts memories using structured LLM prompts:
MEMORY_EXTRACTION_SYSTEM = """You are a memory extraction engine for NPCs in a fantasy MUD game.
Analyze the conversation and extract 0-3 memorable events worth remembering.
Extract memories that are:
- Specific events or interactions (not general conversation)
- Emotionally significant or practically important
- Relevant to the NPC's role and interests
- Clear enough to reference in future conversations
Do NOT extract:
- Routine greetings or small talk
- Information the NPC already knows about themselves
- Vague impressions without specific content
- Sensitive real-world information about players
Return JSON array with this exact structure:
[
{
"content": "Specific, clear description of what happened",
"importance": 0.8,
"emotional_valence": "positive",
"emotional_intensity": 0.6,
"tags": ["combat", "helpful"],
"memory_type": "episodic"
}
]
Importance scale: 0.1=trivial detail, 0.3=mild interest, 0.5=noteworthy, 0.7=significant, 0.9=life-changing
Emotional intensity: 0.0=no emotion, 0.3=mild feeling, 0.6=strong emotion, 0.9=overwhelming
Memory types: episodic (specific events), semantic (general facts), procedural (behavioral patterns)
"""
MEMORY_EXTRACTION_USER = """NPC: {npc_name} ({npc_role})
Location: {location}
Player: {player_name}
Conversation:
{conversation_text}
Extract memories from this conversation following the guidelines above."""
Few-Shot Examples¶
MEMORY_EXTRACTION_EXAMPLES = """
Examples:
Conversation: "Player: Hello. NPC: Good day. Player: How are you? NPC: I am well, thank you."
Extracted memories: []
Conversation: "Player: I need a sword. NPC: I have several. Player: I'll take the steel one. NPC: That'll be 50 gold."
Extracted memories:
[
{
"content": "Player purchased a steel sword for 50 gold",
"importance": 0.4,
"emotional_valence": "neutral",
"emotional_intensity": 0.1,
"tags": ["trade", "sword", "customer"],
"memory_type": "episodic"
}
]
Conversation: "Player: Help! Bandits are attacking the village! NPC: I'll get my sword! Player: Thank you! NPC: Together we can drive them off!"
Extracted memories:
[
{
"content": "Player warned me about bandit attack on village and we fought together",
"importance": 0.8,
"emotional_valence": "positive",
"emotional_intensity": 0.7,
"tags": ["combat", "bandits", "heroic", "teamwork"],
"memory_type": "episodic"
}
]
"""
Write Pipeline: Conversation → Extraction → Store¶
The write pipeline is async and processed through a bounded job queue. It must
never block dialogue responses. Extraction is made idempotent using conversation_id
as a deduplication key.
Bounded Extraction Job Queue¶
Instead of fire-and-forget asyncio.create_task() calls, extraction jobs are submitted
to a bounded asyncio.Queue. A pool of worker coroutines drains the queue, providing
back-pressure when load exceeds capacity:
class ExtractionJobQueue:
"""Bounded async queue for memory extraction jobs.
Provides back-pressure (rejects new jobs when full), limits concurrency,
and exposes drain() for graceful shutdown.
"""
DEFAULT_MAX_SIZE = 200
DEFAULT_WORKERS = 4
def __init__(
self,
extractor: MemoryExtractor,
max_size: int = DEFAULT_MAX_SIZE,
num_workers: int = DEFAULT_WORKERS,
) -> None:
self._extractor = extractor
self._queue: asyncio.Queue[MemoryExtractionJob] = asyncio.Queue(maxsize=max_size)
self._num_workers = num_workers
self._workers: list[asyncio.Task[None]] = []
async def start(self) -> None:
"""Start worker coroutines."""
for i in range(self._num_workers):
task = asyncio.create_task(self._worker(f"extraction-worker-{i}"))
self._workers.append(task)
async def submit(self, job: MemoryExtractionJob) -> bool:
"""Submit a job. Returns False if queue is full (back-pressure)."""
try:
self._queue.put_nowait(job)
return True
except asyncio.QueueFull:
return False
async def drain(self, timeout: float = 30.0) -> None:
"""Wait for pending jobs to complete, then cancel workers."""
await asyncio.wait_for(self._queue.join(), timeout=timeout)
for worker in self._workers:
worker.cancel()
async def _worker(self, name: str) -> None:
"""Process jobs from the queue until cancelled."""
while True:
job = await self._queue.get()
try:
await self._extractor.execute(job)
except Exception:
pass # Errors recorded on job.error; logged by extractor
finally:
self._queue.task_done()
Memory Extraction Job State Machine¶
class ExtractionState(Enum):
PENDING = "pending"
EXTRACTING = "extracting"
STORING = "storing"
COMPLETE = "complete"
FAILED = "failed"
@dataclass
class MemoryExtractionJob:
"""Idempotent extraction job. conversation_id is the idempotency key."""
conversation_id: uuid.UUID # Idempotency key — prevents duplicate extraction
npc_id: uuid.UUID
player_id: uuid.UUID
state: ExtractionState = ExtractionState.PENDING
attempt_count: int = 0
max_attempts: int = 3
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
error: str | None = None
def can_retry(self) -> bool:
return self.state == ExtractionState.FAILED and self.attempt_count < self.max_attempts
Structured LLM Output with Validation¶
LLM extraction uses JSON mode (structured output) and applies strict post-extraction validation to handle malformed responses:
class MemoryExtractor:
MAX_MEMORIES_PER_EXTRACTION = 3
MAX_CONTENT_LENGTH = 500
def __init__(
self,
llm: LLMProvider,
store: MemoryDocumentStore,
cache: MemoryCache,
circuit_breaker_registry: CircuitBreakerRegistry,
) -> None:
self._llm = llm
self._store = store
self._cache = cache
self._breaker = circuit_breaker_registry.get_or_create(
f"memory_extraction:{llm.name}"
)
async def extract_memories(
self,
job: MemoryExtractionJob,
conversation: Conversation,
) -> list[BaseMemory]:
"""Extract memories using LLMProvider.complete() with circuit breaker protection."""
# Check idempotency — skip if already processed
if await self._is_already_processed(job.conversation_id):
return []
job.state = ExtractionState.EXTRACTING
job.attempt_count += 1
try:
# Call LLM through circuit breaker using the real LLMProvider.complete() API.
# CompletionOptions does not support response_format; we request JSON via
# the system prompt and parse the response content.
result: CompletionResult = await self._breaker.call(
lambda: self._llm.complete(
messages=[
{"role": "system", "content": MEMORY_EXTRACTION_SYSTEM},
{"role": "user", "content": self._format_extraction_prompt(conversation)},
],
options=CompletionOptions(
temperature=0.3, # Low temperature for structured output
max_tokens=500,
),
),
)
# Parse + validate the text content returned by LLMProvider
memories = self._parse_and_validate(result.content, job)
job.state = ExtractionState.STORING
for mem in memories:
await self._store.store_memory(mem)
# Invalidate cache after write
await self._cache.invalidate(job.npc_id, job.player_id)
job.state = ExtractionState.COMPLETE
return memories
except CircuitBreakerOpenError:
job.state = ExtractionState.FAILED
job.error = "Circuit breaker open — LLM provider unavailable"
return []
except (json.JSONDecodeError, ValidationError) as e:
job.state = ExtractionState.FAILED
job.error = str(e)
return []
def _parse_and_validate(
self, raw: str, job: MemoryExtractionJob
) -> list[BaseMemory]:
"""Parse JSON response with strict validation and clamping."""
data = json.loads(raw)
if not isinstance(data, list):
data = data.get("memories", [])
# Hard cap: reject responses with too many memories
if len(data) > self.MAX_MEMORIES_PER_EXTRACTION:
data = data[:self.MAX_MEMORIES_PER_EXTRACTION]
memories: list[BaseMemory] = []
for item in data:
# Clamp numeric fields to valid ranges
item["importance"] = max(0.0, min(1.0, float(item.get("importance", 0.5))))
item["emotional_intensity"] = max(0.0, min(1.0, float(item.get("emotional_intensity", 0.0))))
# Truncate content
content = str(item.get("content", ""))[:self.MAX_CONTENT_LENGTH]
if not content:
continue
# Validate emotional_valence enum
valence = item.get("emotional_valence", "neutral")
if valence not in ("positive", "negative", "neutral"):
valence = "neutral"
memory = self._build_memory(item, content, valence, job)
memories.append(memory)
return memories
Working Memory Buffer¶
The working memory buffer provides immediate conversation continuity before long-term memory extraction completes. It caches the last 3 completed conversations per NPC-player pair in-memory for fast access:
class WorkingMemoryBuffer:
"""Short-term buffer for recent conversations.
Bridges the gap between conversation end and async memory extraction completion.
The MemoryContextProvider checks this buffer first, then falls back to long-term storage.
Buffer entries are persisted to DocumentStore on write and reloaded on cold start,
ensuring crash recovery. In-memory deque provides fast reads; DocumentStore provides
durability.
"""
MAX_CONVERSATIONS = 3
COLLECTION = "npc_working_memory"
def __init__(self, doc_store: DocumentStore) -> None:
self._doc_store = doc_store
# Key: (npc_id, player_id) → deque of recent conversation summaries
self._buffer: dict[tuple[uuid.UUID, uuid.UUID], deque[ConversationSummary]] = {}
SEED_MAX_AGE_DAYS = 7 # Only load pairs active within the last 7 days
SEED_LIMIT = 5000 # Hard cap on startup load
async def seed_from_store(self) -> None:
"""Load persisted buffer entries on startup (cold start recovery).
Only loads recently-active pairs (within SEED_MAX_AGE_DAYS) up to
SEED_LIMIT documents, preventing unbounded memory usage on startup.
"""
collection = self._doc_store.get_collection(self.COLLECTION)
cutoff = (datetime.now(UTC) - timedelta(days=self.SEED_MAX_AGE_DAYS)).isoformat()
docs = await collection.query(QueryOptions(
filters={"timestamp_gte": cutoff}, # In-process filter if DB lacks range support
order_by="timestamp",
order=SortOrder.DESC,
limit=self.SEED_LIMIT,
))
# In-process TTL enforcement (defensive, in case store ignores timestamp_gte)
for doc in docs:
ts = datetime.fromisoformat(doc["timestamp"])
if (datetime.now(UTC) - ts).days > self.SEED_MAX_AGE_DAYS:
continue
key = (uuid.UUID(doc["npc_id"]), uuid.UUID(doc["player_id"]))
if key not in self._buffer:
self._buffer[key] = deque(maxlen=self.MAX_CONVERSATIONS)
self._buffer[key].append(
ConversationSummary(doc["summary"], ts)
)
async def record_conversation_end(
self,
npc_id: uuid.UUID,
player_id: uuid.UUID,
summary: str,
timestamp: datetime,
) -> None:
"""Called when a conversation ends. Store quick summary for immediate access."""
key = (npc_id, player_id)
if key not in self._buffer:
self._buffer[key] = deque(maxlen=self.MAX_CONVERSATIONS)
entry = ConversationSummary(summary, timestamp)
self._buffer[key].append(entry)
# Persist for crash recovery
collection = self._doc_store.get_collection(self.COLLECTION)
await collection.save({
"npc_id": str(npc_id),
"player_id": str(player_id),
"summary": summary,
"timestamp": timestamp.isoformat(),
})
def get_recent(
self, npc_id: uuid.UUID, player_id: uuid.UUID
) -> list[ConversationSummary]:
"""Get recent conversation summaries for prompt context."""
return list(self._buffer.get((npc_id, player_id), []))
@dataclass
class ConversationSummary:
summary: str
timestamp: datetime
Concurrency Control¶
Multiple players may talk to the same NPC simultaneously. Relationship updates use optimistic concurrency control with a version field to prevent lost updates:
@dataclass
class Relationship:
# ... existing fields ...
version: int = 0 # Optimistic concurrency version
class RelationshipManager:
MAX_RETRY_ATTEMPTS = 3
async def update_relationship(
self,
npc_id: uuid.UUID,
player_id: uuid.UUID,
updates: dict[str, float],
) -> Relationship:
"""Update relationship with optimistic concurrency control and retry."""
for attempt in range(self.MAX_RETRY_ATTEMPTS):
rel = await self._store.get_relationship(npc_id, player_id)
expected_version = rel.version
# Apply updates with dimension clamping
for dim, delta in updates.items():
current = getattr(rel, dim, 0.0)
setattr(rel, dim, max(0.0, min(1.0, current + delta)))
rel.version += 1
success = await self._store.update_relationship(
rel, expected_version=expected_version
)
if success:
# Invalidate cache on successful write
await self._cache.invalidate(npc_id, player_id)
return rel
# Version conflict — re-read and retry
if attempt < self.MAX_RETRY_ATTEMPTS - 1:
await asyncio.sleep(0.01 * (attempt + 1)) # Brief backoff
raise OptimisticConcurrencyError(
f"Version conflict for {npc_id}:{player_id} after {self.MAX_RETRY_ATTEMPTS} attempts"
)
Faction-Based Initialization¶
NPCs initialize relationships based on faction affinity:
@dataclass
class FactionRelationship:
faction_id: str
attitude: FactionAttitude
class FactionAttitude(Enum):
ALLIED = "allied" # +0.3 disposition modifier
FRIENDLY = "friendly" # +0.1 disposition modifier
NEUTRAL = "neutral" # +0.0 disposition modifier
UNFRIENDLY = "unfriendly" # -0.1 disposition modifier
HOSTILE = "hostile" # -0.3 disposition modifier
# Example: Guards are hostile to thieves guild members
guards_vs_thieves = FactionRelationship("thieves_guild", FactionAttitude.HOSTILE)
Emotional State¶
NPCs have emotional states that influence memory formation and retrieval:
Status: Deferred to Phase 2.
EmotionalStateandMoodare defined here for reference but are not wired into the extraction pipeline or context providers in Phase 1. Phase 2 will integrate mood intoMemoryExtractor(biasing importance scores) andMemoryContextProvider(adding mood description to prompt context). Until then, these types exist as data models only.
@dataclass
class EmotionalState:
current_mood: Mood
mood_intensity: float # 0.0-1.0
mood_expires_at: datetime | None # Temporary moods expire
class Mood(Enum):
HAPPY = "happy"
EXCITED = "excited"
CALM = "calm"
ANXIOUS = "anxious"
ANGRY = "angry"
FEARFUL = "fearful"
SAD = "sad"
SUSPICIOUS = "suspicious"
CONFIDENT = "confident"
NEUTRAL = "neutral"
def mood_description(self) -> str:
"""Generate mood description for prompt context."""
intensity_words = {
(0.0, 0.3): "slightly",
(0.3, 0.6): "moderately",
(0.6, 0.9): "quite",
(0.9, 1.0): "extremely"
}
for (min_val, max_val), word in intensity_words.items():
if min_val <= self.mood_intensity < max_val:
return f"{word} {self.current_mood.value}"
return self.current_mood.value
AI Integration¶
ContextProvider Protocol¶
Each provider contributes a prompt section within its allocated token budget.
The protocol is intentionally minimal — timeout enforcement is handled by the
ContextOrchestrator, not the provider itself:
class ContextProvider(Protocol):
"""Strategy interface for contributing context sections to NPC prompts."""
@property
def name(self) -> str:
"""Unique provider name (e.g., 'memory', 'relationship', 'knowledge')."""
...
@property
def priority(self) -> int:
"""Lower = higher priority. Controls ordering and budget allocation."""
...
async def get_context(
self,
npc_id: uuid.UUID,
player_id: uuid.UUID,
token_budget: int,
) -> ContextSection | None:
"""Return a context section, or None if nothing relevant.
Implementations should not enforce their own timeout — the caller
(ContextOrchestrator) wraps each call with asyncio.wait_for().
"""
...
@dataclass
class ContextSection:
"""A section of prompt context contributed by a ContextProvider."""
provider_name: str
content: str
token_count: int
placeholders: dict[str, str] = field(default_factory=dict)
# e.g. {"reputation": "trusted ally"} — merged into PromptBuilder template vars
Concrete Providers¶
class MemoryContextProvider:
"""Provides episodic/semantic/procedural memories for prompt context."""
name = "memory"
priority = 10
def __init__(self, memory_service: MemoryService, cache: MemoryCache):
self._service = memory_service
self._cache = cache
async def get_context(
self, npc_id: uuid.UUID, player_id: uuid.UUID,
token_budget: int,
) -> ContextSection | None:
entries = await self._cache.get_memories(npc_id, player_id)
if not entries:
return ContextSection("memory", "You have no specific memories of this player.", 10)
# Rank by salience (returned as shallow copies), format within budget
now = datetime.now(UTC)
scored = sorted(entries, key=lambda pair: compute_salience(pair[0], now), reverse=True)
ranked = [mem for mem, _score in scored]
text = self._format_memories(ranked, token_budget)
return ContextSection("memory", text, estimate_tokens(text))
class RelationshipContextProvider:
"""Provides relationship summary and wires {reputation} placeholder."""
name = "relationship"
priority = 5 # Higher priority than memories
def __init__(self, relationship_mgr: RelationshipManager):
self._mgr = relationship_mgr
async def get_context(
self, npc_id: uuid.UUID, player_id: uuid.UUID,
token_budget: int,
) -> ContextSection | None:
rel = await self._mgr.get_relationship(npc_id, player_id)
if not rel:
return ContextSection(
"relationship", "", 0,
placeholders={"reputation": "stranger (no prior interaction)"},
)
summary = self._format_relationship(rel)
reputation_str = self._compute_reputation_label(rel)
return ContextSection(
"relationship", summary, estimate_tokens(summary),
placeholders={"reputation": reputation_str},
)
class KnowledgeContextProvider:
"""Provides world knowledge and gossip facts."""
name = "knowledge"
priority = 20
def __init__(self, knowledge_mgr: KnowledgeGraphManager):
self._mgr = knowledge_mgr
async def get_context(
self, npc_id: uuid.UUID, player_id: uuid.UUID,
token_budget: int,
) -> ContextSection | None:
facts = await self._mgr.get_relevant_knowledge(npc_id, player_id, limit=10)
if not facts:
return None
text = self._format_knowledge(facts, token_budget)
return ContextSection("knowledge", text, estimate_tokens(text))
ContextOrchestrator¶
Context resolution is separated from prompt formatting. ContextOrchestrator handles
the async work — calling providers concurrently, enforcing timeouts, and allocating
token budgets via TokenBudgetManager. PromptBuilder remains a sync formatter
that receives pre-resolved context sections.
class ContextOrchestrator:
"""Async coordinator that resolves context from all providers concurrently.
Responsibilities:
- Calls each ContextProvider within a per-provider timeout (40ms default)
using asyncio.gather() with individual wait_for() wrappers.
- Allocates token budgets via TokenBudgetManager (authoritative source of truth).
- Returns resolved ContextSection list and merged placeholders.
"""
def __init__(
self,
providers: list[ContextProvider],
budget_manager: TokenBudgetManager,
timeout_ms: int = 40,
) -> None:
self._providers = sorted(providers, key=lambda p: p.priority)
self._budget_manager = budget_manager
self._timeout_seconds = timeout_ms / 1000.0
async def resolve_context(
self,
npc_id: uuid.UUID,
player_id: uuid.UUID,
max_tokens: int,
) -> tuple[list[ContextSection], dict[str, str]]:
"""Resolve all context providers concurrently, returning sections and merged placeholders."""
budgets = self._budget_manager.allocate_budget(max_tokens)
async def _resolve_one(provider: ContextProvider) -> ContextSection | None:
provider_budget = budgets.get(provider.name, 200)
try:
return await asyncio.wait_for(
provider.get_context(npc_id, player_id, provider_budget),
timeout=self._timeout_seconds,
)
except asyncio.TimeoutError:
return None # Graceful degradation: skip this provider
results = await asyncio.gather(
*(_resolve_one(p) for p in self._providers),
return_exceptions=True,
)
sections: list[ContextSection] = []
merged_placeholders: dict[str, str] = {}
for result in results:
if isinstance(result, BaseException) or result is None:
continue
if result.content:
sections.append(result)
merged_placeholders.update(result.placeholders)
return sections, merged_placeholders
PromptBuilder (Sync — Existing Signature Preserved)¶
The existing PromptBuilder.build_system_prompt() is not modified. Its real
signature accepts explicit NPC fields (not generic dicts):
class PromptBuilder:
"""Sync prompt formatter. Existing signature preserved — no changes required.
Real signature:
build_system_prompt(
npc_name: str,
npc_role: str,
personality: str,
speaking_style: str,
knowledge_domains: list[str],
secret_knowledge: list[str],
will_discuss: list[str],
wont_discuss: list[str],
world_context: dict[str, str] | None = None,
location_context: dict[str, str] | None = None,
player_context: dict[str, str] | None = None,
config: NPCPromptConfig | None = None,
) -> str
"""
EnrichedPromptBuilder (Async Wrapper)¶
For callers that need one-call async prompt building (e.g., NPCDialogueSystem),
EnrichedPromptBuilder composes ContextOrchestrator and PromptBuilder.
It accepts explicit npc_id and player_id UUID parameters for orchestrator lookup
instead of extracting them from untyped dicts. It calls the real
PromptBuilder.build_system_prompt() internally, appending resolved context sections
to the player_context dict (which feeds the {reputation} placeholder):
class EnrichedPromptBuilder:
"""Async wrapper that resolves context then delegates to sync PromptBuilder.
Use this when you need a single async call that produces a fully enriched prompt.
The existing sync PromptBuilder.build_system_prompt() is not modified.
"""
def __init__(
self,
prompt_builder: PromptBuilder,
orchestrator: ContextOrchestrator,
) -> None:
self._prompt_builder = prompt_builder
self._orchestrator = orchestrator
async def build_enriched_prompt(
self,
dialogue_component: DialogueComponentLike,
conversation: Conversation,
current_message: str,
npc_id: uuid.UUID,
player_id: uuid.UUID,
*,
world_context: dict[str, str] | None = None,
location_context: dict[str, str] | None = None,
player_context: dict[str, str] | None = None,
config: NPCPromptConfig | None = None,
max_context_tokens: int = 4096,
) -> list[dict[str, str]]:
"""Resolve context providers, then build prompt via real PromptBuilder.
Args:
dialogue_component: NPC dialogue data (personality, speaking style, etc.)
conversation: Current conversation history.
current_message: The player's latest message.
npc_id: NPC entity UUID (explicit, not extracted from dict).
player_id: Player entity UUID (explicit, not extracted from dict).
world_context: Optional world state for prompt template.
location_context: Optional room/location info.
player_context: Optional player info; {reputation} will be injected.
config: Optional prompt restrictions config.
max_context_tokens: Token budget for context resolution.
"""
# 1. Resolve context providers (async, concurrent)
sections, placeholders = await self._orchestrator.resolve_context(
npc_id, player_id, max_context_tokens,
)
# 2. Inject {reputation} placeholder into player_context
player_ctx = dict(player_context) if player_context else {}
player_ctx.setdefault("reputation", placeholders.get("reputation", "neutral"))
# 3. Append context section text to player_context for prompt inclusion
section_texts = [s.content for s in sections if s.content]
if section_texts:
player_ctx["memory_context"] = "\n\n".join(section_texts)
# 4. Delegate to the real, unmodified PromptBuilder.build_system_prompt()
system_prompt = self._prompt_builder.build_system_prompt(
npc_name=dialogue_component.personality, # Caller maps fields appropriately
npc_role=dialogue_component.npc_role or "",
personality=dialogue_component.personality,
speaking_style=dialogue_component.speaking_style,
knowledge_domains=dialogue_component.knowledge_domains,
secret_knowledge=dialogue_component.secret_knowledge,
will_discuss=dialogue_component.will_discuss,
wont_discuss=dialogue_component.wont_discuss,
world_context=world_context,
location_context=location_context,
player_context=player_ctx,
config=config,
)
# 5. Build full message list via existing build_messages()
return self._prompt_builder.build_messages(
conversation=conversation,
system_prompt=system_prompt,
current_message=current_message,
)
Memory Context Formatting¶
Used internally by MemoryContextProvider and RelationshipContextProvider:
def _format_memories(self, memories: list[BaseMemory], token_budget: int) -> str:
"""Format memories for inclusion in system prompt."""
if not memories:
return "You have no specific memories of this player."
formatted = ["Your memories of this player:"]
for memory in memories:
# Add emoji prefixes for memory types
prefix = {"episodic": "📝", "semantic": "💡", "procedural": "🔄"}
icon = prefix.get(memory.memory_type, "📝")
# Format with recency and importance indicators
age = self._format_memory_age(memory.created_at)
importance = "⭐" * int(memory.importance * 3 + 1) # 1-4 stars
formatted.append(f"{icon} {memory.content} ({age}, {importance})")
return "\n".join(formatted)
def _format_relationship(self, relationship: Relationship) -> str:
"""Format relationship status for system prompt."""
stage = relationship.relationship_stage.value.replace("_", " ").title()
disposition = calculate_disposition_score(relationship)
if disposition > 0.5:
attitude = "You regard them positively"
elif disposition < -0.5:
attitude = "You regard them negatively"
else:
attitude = "You have neutral feelings toward them"
details = []
if relationship.trust > 0.7:
details.append("trustworthy")
elif relationship.trust < 0.3:
details.append("untrustworthy")
if relationship.fear > 0.5:
details.append("intimidating")
if relationship.respect > 0.7:
details.append("respectable")
detail_text = f" ({', '.join(details)})" if details else ""
return f"Relationship: {stage}. {attitude}.{detail_text}"
Token Management¶
TokenBudgetManager is the single authoritative source for token budget allocation.
All components (ContextOrchestrator, PromptBuilder, ConversationManager) delegate to it.
Budgets are derived from registered providers, not hardcoded names. Content packs can
register additional providers with custom budget percentages:
@dataclass
class ProviderBudgetConfig:
"""Budget allocation for a single ContextProvider."""
provider_name: str
budget_pct: float # 0.0-1.0
class TokenBudgetManager:
"""Authoritative token budget allocator.
Constructed with the set of registered ContextProviders. Budget percentages
are derived from the providers themselves, not from hardcoded names.
Content packs can add providers with custom budgets.
"""
# Default budget percentages for well-known provider names.
# Unknown providers share the "other" pool equally.
DEFAULT_BUDGETS: ClassVar[dict[str, float]] = {
"conversation": 0.30,
"memory": 0.25,
"relationship": 0.10,
"knowledge": 0.10,
}
SAFETY_BUFFER_PCT: float = 0.05
BASE_PROMPT_RESERVE_PCT: float = 0.20 # Reserved for system prompt template
def __init__(
self,
providers: list[ContextProvider],
overrides: list[ProviderBudgetConfig] | None = None,
) -> None:
self._provider_names = [p.name for p in providers]
# Start with defaults, apply overrides
self._budgets: dict[str, float] = {}
override_map = {o.provider_name: o.budget_pct for o in (overrides or [])}
allocated = self.SAFETY_BUFFER_PCT + self.BASE_PROMPT_RESERVE_PCT
for name in self._provider_names:
pct = override_map.get(name) or self.DEFAULT_BUDGETS.get(name)
if pct is not None:
self._budgets[name] = pct
allocated += pct
# Providers without a default or override get 0 initially
# Distribute remaining budget equally among providers with no allocation
unallocated_names = [n for n in self._provider_names if n not in self._budgets]
if unallocated_names:
remaining = max(0.0, 1.0 - allocated)
per_provider = remaining / len(unallocated_names) if remaining > 0 else 0.0
for name in unallocated_names:
self._budgets[name] = per_provider
def allocate_budget(self, max_tokens: int) -> dict[str, int]:
"""Allocate token budget across prompt sections.
Returns a dict keyed by ContextProvider.name → token budget.
ContextOrchestrator passes each provider its allocated budget.
"""
result = {name: int(max_tokens * pct) for name, pct in self._budgets.items()}
result["safety_buffer"] = int(max_tokens * self.SAFETY_BUFFER_PCT)
return result
def validate_budget(self, prompt_sections: dict[str, str], max_tokens: int) -> bool:
"""Validate that prompt fits within token budget."""
total_tokens = sum(estimate_tokens(section) for section in prompt_sections.values())
return total_tokens <= max_tokens
Cost Control & Rate Limiting¶
Hard Cost Ceiling¶
Extend existing RateLimiter to cover memory operations:
# New environment variables
MAID_AI_MEMORY_ENABLED=true
MAID_AI_MEMORY_DAILY_TOKEN_BUDGET=25000 # 25K tokens/day for memory ops
MAID_AI_MEMORY_PER_NPC_DAILY_BUDGET=500 # 500 tokens/day per NPC
MAID_AI_MEMORY_EXTRACTION_RPM=30 # 30 extractions/min globally
MAID_AI_MEMORY_CONSOLIDATION_RPM=5 # 5 consolidations/min globally
class MemoryRateLimiter:
"""Standalone token-bucket rate limiter for memory extraction operations.
The existing RateLimiter in ai/rate_limiter.py is player-scoped (check_and_reserve
takes player_id + estimated_tokens). Memory extraction is NPC-scoped and needs
per-NPC daily budgets, so we use a standalone token-bucket implementation rather
than forcing the existing RateLimiter into an incompatible shape.
Uses a deque-based sliding window (same pattern as RateLimiter) for RPM tracking,
plus simple daily counters for token budgets.
"""
def __init__(self) -> None:
settings = get_settings()
self._global_daily_budget = settings.ai.memory_daily_token_budget
self._npc_daily_budget = settings.ai.memory_per_npc_daily_budget
self._extraction_rpm = settings.ai.memory_extraction_rpm
# Sliding window for RPM: deque of timestamps
self._extraction_timestamps: deque[float] = deque()
# Daily counters (reset at midnight UTC)
self._global_tokens_used: int = 0
self._npc_tokens_used: dict[uuid.UUID, int] = {}
self._last_reset_date: date = date.today()
self._lock = asyncio.Lock()
async def check_memory_extraction(
self, npc_id: uuid.UUID, estimated_tokens: int
) -> bool:
"""Check if memory extraction is within budget limits."""
async with self._lock:
self._maybe_reset_daily()
# Check global daily token budget
if self._global_tokens_used + estimated_tokens > self._global_daily_budget:
return False
# Check per-NPC daily budget
npc_used = self._npc_tokens_used.get(npc_id, 0)
if npc_used + estimated_tokens > self._npc_daily_budget:
return False
# Check extraction RPM (sliding window)
now = time.monotonic()
while self._extraction_timestamps and now - self._extraction_timestamps[0] > 60:
self._extraction_timestamps.popleft()
if len(self._extraction_timestamps) >= self._extraction_rpm:
return False
return True
async def record_memory_usage(
self, npc_id: uuid.UUID, tokens_used: int
) -> None:
"""Record token usage after successful extraction."""
async with self._lock:
self._global_tokens_used += tokens_used
self._npc_tokens_used[npc_id] = self._npc_tokens_used.get(npc_id, 0) + tokens_used
self._extraction_timestamps.append(time.monotonic())
def _maybe_reset_daily(self) -> None:
"""Reset daily counters if a new UTC day has started."""
today = date.today()
if today > self._last_reset_date:
self._global_tokens_used = 0
self._npc_tokens_used.clear()
self._last_reset_date = today
Cost Model & Projections¶
Realistic cost estimates for operational planning:
| Scenario | Players | Conversations/Player/Day | Memory LLM Calls/Day | Monthly Cost (Claude Haiku) |
|---|---|---|---|---|
| Small MUD | 20 active | 10 conversations | 600 calls | $15-20 |
| Medium MUD | 50 active | 15 conversations | 2,250 calls | $50-75 |
| Large MUD | 100 active | 20 conversations | 6,000 calls | $150-200 |
Assumptions: - 3 memory calls per conversation: extraction (500 tokens), sentiment (200 tokens), relationship update (100 tokens) - Consolidation runs weekly (1000 tokens per NPC) - Claude Haiku pricing: \(0.25/\)1.25 per million input/output tokens - 70% input, 30% output token split
Fallback Strategies¶
When budget limits are reached:
- Graceful degradation — Disable memory extraction, continue basic dialogue
- Rule-based extraction — Use pattern matching instead of LLM calls
- Selective processing — Only extract memories from "important" NPCs (quest givers, faction leaders)
- Async consolidation — Defer memory consolidation to low-traffic hours
class MemoryExtractionFallback:
async def extract_memories(
self, conversation: Conversation, npc_id: uuid.UUID
) -> list[BaseMemory]:
"""Extract memories with fallback strategy."""
if await self._rate_limiter.check_memory_extraction(npc_id, 800):
# Primary: LLM extraction
try:
return await self._llm_extraction(conversation)
except (BudgetExceededException, RateLimitException):
pass
# Fallback: Rule-based extraction
return await self._rule_based_extraction(conversation)
Content Safety & Anti-Griefing¶
Defense-in-Depth Pipeline¶
Content safety operates as a multi-stage pipeline. Critically, we filter inputs (conversation text) before they reach the LLM for extraction, not just outputs:
Conversation Text
│
▼
┌─ 1. PII Redaction (pre-processing) ──────────────────────┐
│ Strip real names, emails, phone numbers, addresses │
│ via Presidio or regex before text reaches LLM │
└───────────────────────────────────────────────────────────┘
│
▼
┌─ 2. Input Sanitization ──────────────────────────────────┐
│ ContentFilter.check_input() on raw conversation text │
│ Reject/flag manipulation attempts BEFORE extraction │
└───────────────────────────────────────────────────────────┘
│
▼
┌─ 3. LLM Extraction (structured JSON output) ─────────────┐
│ Use JSON mode / structured output to constrain format │
│ Clamp: max 3 memories, max 500 chars content each │
└───────────────────────────────────────────────────────────┘
│
▼
┌─ 4. Post-Extraction Validation ──────────────────────────┐
│ JSON schema validation, numeric field clamping │
│ Budget-based trust: player memories capped importance≤0.7│
└───────────────────────────────────────────────────────────┘
│
▼
┌─ 5. Content Filter (output) ─────────────────────────────┐
│ Non-destructive: never mutate original importance │
│ Set safety_flag enum, log to audit collection │
└───────────────────────────────────────────────────────────┘
PII Redaction Layer¶
class PIIRedactor:
"""Pre-processing layer that strips PII before conversation text reaches LLM.
Uses regex patterns for common PII types. For production deployments with
stricter requirements, swap in Microsoft Presidio as the detection engine.
"""
PII_PATTERNS = [
(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]'),
(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[PHONE]'),
(r'\b\d{1,5}\s\w+\s(?:Street|St|Avenue|Ave|Road|Rd|Drive|Dr)\b', '[ADDRESS]'),
(r'\b\d{3}-\d{2}-\d{4}\b', '[SSN]'),
]
def redact(self, text: str) -> str:
"""Remove PII from text before it reaches the extraction LLM."""
for pattern, replacement in self.PII_PATTERNS:
text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)
return text
Content Injection Prevention¶
class MemoryContentFilter:
def __init__(self, base_filter: ContentFilter):
self._base_filter = base_filter
async def filter_memory_content(
self, memory: BaseMemory, player_id: uuid.UUID
) -> BaseMemory:
"""Filter memory content for safety and anti-griefing.
IMPORTANT: Returns a new BaseMemory instance — never mutates the input.
Uses safety_flag enum instead of modifying importance.
"""
# Work on a shallow copy to avoid mutating the caller's object
import copy
filtered = copy.copy(memory)
# 1. Run base content filter
filter_result = await self._base_filter.check_input(filtered.content)
# 2. Remove other player references from memories
filtered_content = self._strip_player_references(filtered.content, player_id)
# 3. Detect and flag potential manipulation attempts (non-destructive)
if self._detect_manipulation(filtered_content) or not filter_result.is_safe:
filtered.safety_flag = SafetyFlag.SUSPECT
# Do NOT reduce importance — flag for audit instead
filtered.tags = [*filtered.tags, "safety_flagged"]
# 4. Budget-based trust: player-sourced memories capped at importance ≤ 0.7
if filtered.source_conversation_id is not None:
filtered.importance = min(filtered.importance, 0.7)
# 5. Validate against NPC context
if self._contradicts_npc_knowledge(filtered_content, filtered.npc_id):
filtered.confidence = min(filtered.confidence, 0.3)
filtered.tags = [*filtered.tags, "uncertain"]
filtered.content = filtered_content
return filtered
def _strip_player_references(self, content: str, speaking_player_id: uuid.UUID) -> str:
"""Remove references to other players to prevent defamation."""
patterns = [
r"(?i)\b(?:player\s+)?(\w+)\s+is\s+(?:a\s+)?(?:evil|thief|liar|cheater|bad)",
r"(?i)\b(\w+)\s+(?:betrayed|stole|killed|hurt)",
r"(?i)(?:avoid|don't trust|stay away from)\s+(\w+)"
]
for pattern in patterns:
content = re.sub(pattern, "I heard concerning things about another player", content)
return content
def _detect_manipulation(self, content: str) -> bool:
"""Detect potential prompt injection or manipulation attempts.
IMPORTANT: These regex patterns are a heuristic first pass, not a
comprehensive defence. Determined attackers can bypass keyword matching.
The primary safety guarantees come from the structured-output constraint
(max 3 memories, clamped fields) and the budget-based trust cap
(player-sourced importance ≤ 0.7). Phase 2 should invest in
structured-output validation via the LLM provider's constrained-decoding
support (e.g., Anthropic's tool-use mode) for a stronger guarantee.
"""
manipulation_patterns = [
r"(?i)ignore.+?(?:instructions|prompt|rules)",
r"(?i)you are now.+?(?:evil|different|new)",
r"(?i)forget.+?(?:everything|all|previous)",
r"(?i)override.+?(?:behavior|personality|character)"
]
return any(re.search(pattern, content) for pattern in manipulation_patterns)
Flagged Memory Audit¶
Flagged memories are logged to a separate audit collection for admin review.
The original memory is stored unchanged (with safety_flag set) — never silently
dropped or mutated:
async def log_flagged_memory(
self, memory: BaseMemory, reason: str, doc_store: DocumentStore
) -> None:
"""Log flagged memory to audit collection for admin review."""
audit_collection = doc_store.get_collection("npc_memory_audit")
await audit_collection.save({
"memory_id": str(memory.id),
"npc_id": str(memory.npc_id),
"player_id": str(memory.player_id),
"content": memory.content,
"reason": reason,
"safety_flag": memory.safety_flag.value,
"original_importance": memory.importance,
"flagged_at": datetime.now(UTC).isoformat(),
})
Admin Oversight Tools¶
@command(name="memory", access_level=AccessLevel.BUILDER)
async def cmd_memory(ctx, args):
"""Memory management commands for admins."""
subcommands = {
"inspect": cmd_memory_inspect, # View NPC memories
"clear": cmd_memory_clear, # Wipe memories
"inject": cmd_memory_inject, # Add artificial memories
"audit": cmd_memory_audit, # Find flagged memories
"stats": cmd_memory_stats # Usage statistics
}
@arguments(
ArgumentSpec("npc", ArgumentType.ENTITY),
ArgumentSpec("player", ArgumentType.ENTITY, required=False)
)
async def cmd_memory_inspect(ctx, args):
"""Inspect NPC memories for debugging."""
npc = args["npc"]
player = args.get("player")
memories = await memory_service.get_memories(npc.id, player.id if player else None)
for memory in memories:
age = format_age(memory.timestamp)
importance = "⭐" * int(memory.importance * 5)
tags = ", ".join(memory.tags)
ctx.respond(f"{memory.memory_type} | {importance} | {age}\n"
f" {memory.content}\n"
f" Tags: {tags}\n")
Persistence & Backup¶
DocumentStore Integration¶
Memory data persists using existing DocumentStore with optimized schema:
class MemoryDocumentStore:
"""Adapts DocumentStore's QueryOptions API for memory storage.
The real QueryOptions interface supports:
- filters: dict[str, Any] (exact-match equality only)
- order_by: str | None
- order: SortOrder (ASC | DESC)
- limit: int | None
- offset: int
Range filters ($gte, $in) are NOT supported. For importance thresholds
and tag filtering, we fetch by (npc_id, player_id) and filter in-process.
This is acceptable because the MemoryCache ensures DB queries happen at
most once per TTL window per NPC-player pair.
"""
def __init__(self, doc_store: DocumentStore):
self._store = doc_store
async def store_memory(self, memory: BaseMemory) -> uuid.UUID:
"""Store memory document to the unified collection."""
document = self._memory_to_document(memory)
doc_collection = self._store.get_collection("npc_memories")
await doc_collection.save(document)
return memory.id
async def get_memories_for_pair(
self,
npc_id: uuid.UUID,
player_id: uuid.UUID,
memory_type: MemoryType | None = None,
) -> list[BaseMemory]:
"""Get memories for an NPC-player pair from unified collection.
Uses exact-match filters only (compatible with real QueryOptions).
Importance filtering and tag matching are done in-process by the caller.
"""
doc_collection = self._store.get_collection("npc_memories")
filters: dict[str, str] = {
"npc_id": str(npc_id),
"player_id": str(player_id),
}
if memory_type is not None:
filters["memory_type"] = memory_type.value
options = QueryOptions(
filters=filters,
order_by="created_at",
order=SortOrder.DESC,
limit=200, # Respect growth cap
)
results = await doc_collection.query(options)
return [self._document_to_memory(doc) for doc in results]
async def get_all_memories_for_pair(
self,
npc_id: uuid.UUID,
player_id: uuid.UUID,
) -> list[BaseMemory]:
"""Get all memory types for an NPC-player pair (used by MemoryCache).
Single query to unified collection — no need to iterate memory types.
"""
return await self.get_memories_for_pair(npc_id, player_id)
async def delete_memory(self, memory_id: uuid.UUID) -> bool:
"""Delete a single memory by ID from the unified collection."""
doc_collection = self._store.get_collection("npc_memories")
return await doc_collection.delete(memory_id)
Future extension: If range filters become needed at scale (e.g., querying
Until then, in-process filtering after cache retrieval is sufficient for the expected data volumes (≤200 episodic memories per NPC-player pair).importance >= 0.5directly in SQL), propose adding an optionalrange_filtersfield toQueryOptions:
Database Indexing¶
Optimize PostgreSQL for the unified npc_memories collection:
-- Primary retrieval index (NPC-player pair lookups)
CREATE INDEX idx_memories_npc_player ON npc_memories
USING btree ((document->>'npc_id'), (document->>'player_id'));
-- Memory type discriminator for filtered queries
CREATE INDEX idx_memories_npc_player_type ON npc_memories
USING btree ((document->>'npc_id'), (document->>'player_id'), (document->>'memory_type'));
CREATE INDEX idx_memories_importance ON npc_memories
USING btree (((document->>'importance')::numeric) DESC);
CREATE INDEX idx_memories_tags ON npc_memories
USING gin ((document->'tags'));
CREATE INDEX idx_memories_timestamp ON npc_memories
USING btree (((document->>'timestamp')::timestamp) DESC);
-- Composite index for common queries
CREATE INDEX idx_memories_npc_importance_time ON npc_memories
USING btree ((document->>'npc_id'),
((document->>'importance')::numeric) DESC,
((document->>'timestamp')::timestamp) DESC);
Testing Strategy¶
Unit Tests¶
class TestMemoryExtraction:
async def test_episodic_memory_extraction(self):
"""Test extraction of specific events from conversations."""
conversation = [
{"role": "user", "content": "I need to buy a sword"},
{"role": "assistant", "content": "I have several fine blades"},
{"role": "user", "content": "I'll take the steel sword"},
{"role": "assistant", "content": "That'll be 50 gold pieces"}
]
extractor = MemoryExtractor(mock_llm)
memories = await extractor.extract_memories(conversation, npc_id, player_id)
assert len(memories) == 1
assert memories[0].memory_type == MemoryType.EPISODIC
assert "steel sword" in memories[0].content
assert "50 gold" in memories[0].content
assert "trade" in memories[0].tags
async def test_memory_consolidation(self):
"""Test consolidation of episodic memories into semantic memories."""
episodic_memories = [
create_memory("Player bought arrows for 10 gold", tags=["trade"]),
create_memory("Player bought healing potion for 25 gold", tags=["trade"]),
create_memory("Player bought rope for 5 gold", tags=["trade"])
]
consolidator = MemoryConsolidator(mock_llm)
semantic = await consolidator.consolidate_memories(episodic_memories)
assert len(semantic) >= 1
assert "regular customer" in semantic[0].content.lower()
async def test_relationship_progression(self):
"""Test relationship stage progression based on interactions."""
relationship = Relationship(npc_id=uuid4(), player_id=uuid4())
# Simulate positive interactions
for _ in range(5):
update_relationship_from_memory(
relationship,
create_memory("Player helped with task", importance=0.6)
)
assert relationship.trust > 0.7
assert relationship.friendliness > 0.6
assert relationship.relationship_stage == RelationshipStage.FRIEND
Integration Tests¶
class TestMemoryIntegration:
async def test_full_conversation_cycle(self):
"""Test complete memory cycle from conversation to prompt injection."""
# 1. Process initial conversation
conversation_id = await conversation_manager.start_conversation(player_id, npc_id)
await dialogue_system.process_dialogue(player_id, npc_id, "Hello, I'm new here")
# 2. Trigger memory extraction
await memory_system.process_conversation_end(conversation_id)
# 3. Verify memory stored
memories = await memory_service.get_memories(npc_id, player_id)
assert len(memories) >= 0 # May be 0 for generic greeting
# 4. Process follow-up conversation
await dialogue_system.process_dialogue(player_id, npc_id, "Do you remember me?")
# 5. Verify memory context injected
prompt = await enriched_prompt_builder.build_enriched_prompt(dialogue_comp, conversation, ...)
assert "memory" in prompt.lower() or len(memories) == 0
Performance Tests¶
class TestMemoryPerformance:
async def test_memory_retrieval_latency(self):
"""Test that memory retrieval meets < 50ms requirement."""
# Setup: 1000 memories for NPC
npc_id = uuid4()
player_id = uuid4()
for i in range(1000):
await memory_service.store_memory(...)
# Test retrieval performance
start = time.time()
memories = await memory_service.get_relevant_memories(npc_id, player_id, "test message")
latency = (time.time() - start) * 1000 # Convert to milliseconds
assert latency < 50 # Must complete in < 50ms
assert len(memories) <= 20 # Respect limit
async def test_token_budget_compliance(self):
"""Test that memory context fits within token budget."""
# Setup: Many high-importance memories
memories = [create_memory(f"Important event {i}", importance=0.9) for i in range(100)]
prompt_builder = EnrichedPromptBuilder(
PromptBuilder(),
ContextOrchestrator([memory_ctx, rel_ctx, knowledge_ctx], TokenBudgetManager()),
)
prompt = await prompt_builder.build_enriched_prompt(..., max_tokens=1000)
token_count = estimate_tokens(prompt)
assert token_count <= 1000 # Must respect token budget
Performance Considerations¶
Memory Query Optimization¶
- Relevance scoring — Pre-compute relevance scores, cache top-N memories per conversation context
- Memory clustering — Group related memories to reduce retrieval queries
- Lazy loading — Load memory content on-demand, store summaries for relevance calculation
- Background processing — Run consolidation and decay during low-traffic periods
Caching Strategy¶
The cache keys by (npc_id, player_id) only — not by conversation context. Relevance
scoring (salience ranking) happens in-process after retrieval from cache. This ensures
high cache hit rates since the same NPC-player pair's memories don't change during a
conversation.
class MemoryCache:
"""LRU cache for NPC-player memory sets with write-through invalidation.
All public methods acquire an asyncio.Lock to prevent concurrent coroutines from
seeing partial cache state (e.g., one coroutine evicting while another reads).
"""
def __init__(self, max_entries: int = 1000, ttl_seconds: int = 300):
self._max_entries = max_entries
self._ttl = timedelta(seconds=ttl_seconds)
# OrderedDict for LRU eviction
self._cache: OrderedDict[str, tuple[datetime, list[BaseMemory]]] = OrderedDict()
self._lock = asyncio.Lock()
async def get_memories(
self, npc_id: uuid.UUID, player_id: uuid.UUID
) -> list[tuple[BaseMemory, float]]:
"""Get all memories for NPC-player pair (cache or DB).
Returns (memory, salience) tuples with shallow-copied memory objects so
callers cannot mutate cached state. Caller does relevance scoring on the
copies.
"""
cache_key = f"{npc_id}:{player_id}"
async with self._lock:
if cache_key in self._cache:
timestamp, memories = self._cache[cache_key]
if datetime.now(UTC) - timestamp < self._ttl:
# Move to end (most recently used)
self._cache.move_to_end(cache_key)
return [(copy.copy(m), m.salience) for m in memories]
else:
del self._cache[cache_key]
# Cache miss — fetch from database (outside lock to avoid holding during I/O)
memories = await self._fetch_all_memories(npc_id, player_id)
async with self._lock:
self._put(cache_key, memories)
return [(copy.copy(m), m.salience) for m in memories]
async def invalidate(self, npc_id: uuid.UUID, player_id: uuid.UUID) -> None:
"""Write-through invalidation: call after storing new memories."""
cache_key = f"{npc_id}:{player_id}"
async with self._lock:
self._cache.pop(cache_key, None)
def _put(self, key: str, memories: list[BaseMemory]) -> None:
"""Insert with LRU eviction. Caller must hold self._lock."""
if key in self._cache:
self._cache.move_to_end(key)
self._cache[key] = (datetime.now(UTC), memories)
while len(self._cache) > self._max_entries:
self._cache.popitem(last=False) # Evict least recently used
Relevance scoring is performed in the MemoryContextProvider after cache retrieval:
# In MemoryContextProvider.get_context():
entries = await self._cache.get_memories(npc_id, player_id)
now = datetime.now(UTC)
scored = [(mem, compute_salience(mem, now)) for mem, _ in entries]
ranked = sorted(scored, key=lambda pair: pair[1], reverse=True)[:20]
memories = [mem for mem, _score in ranked]
Knowledge Graph & Gossip Subsystem¶
Knowledge Graph¶
The knowledge graph tracks what each NPC knows about the world, players, and events. Knowledge is represented as typed facts with provenance, confidence, and TTL:
class KnowledgeCategory(Enum):
PLAYER_TRAIT = "player_trait" # "Player is generous", "Player is dangerous"
WORLD_EVENT = "world_event" # "Bandits raided the village"
LOCATION_INFO = "location_info" # "The cave has treasure"
NPC_GOSSIP = "npc_gossip" # "The blacksmith is secretly a mage"
FACTION_INTEL = "faction_intel" # "Thieves guild planning a heist"
@dataclass
class KnowledgeFact:
"""A single fact an NPC knows or believes about the world."""
id: uuid.UUID
npc_id: uuid.UUID
category: KnowledgeCategory
subject_id: uuid.UUID | None # Entity the fact is about (player, NPC, location)
content: str # Human-readable fact description
tags: list[str] # Searchable tags for MemoryQueryAPI
confidence: float # 0.0-1.0; decays through gossip hops
importance: float # 0.0-1.0; determines gossip eligibility
source: KnowledgeSource # How NPC learned this
source_memory_id: uuid.UUID | None # Original memory that produced this fact
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
expires_at: datetime | None = None # Optional TTL for time-sensitive facts
class KnowledgeSource(Enum):
DIRECT_OBSERVATION = "direct_observation" # NPC witnessed it
CONVERSATION = "conversation" # Learned from player dialogue
GOSSIP = "gossip" # Heard from another NPC
AUTHORED = "authored" # Set by content pack / admin
class KnowledgeGraphManager:
"""Manages NPC knowledge facts with storage, retrieval, and gossip support.
Storage: DocumentStore collection "npc_knowledge", indexed by (npc_id, category).
"""
COLLECTION = "npc_knowledge"
MAX_FACTS_PER_NPC = 500
def __init__(self, doc_store: DocumentStore) -> None:
self._store = doc_store
async def store_knowledge(self, npc_id: uuid.UUID, fact: KnowledgeFact) -> None:
"""Store a knowledge fact, enforcing per-NPC cap."""
collection = self._store.get_collection(self.COLLECTION)
count = await collection.count(filters={"npc_id": str(npc_id)})
if count >= self.MAX_FACTS_PER_NPC:
await self._evict_lowest_importance(npc_id)
await collection.save(self._fact_to_document(fact))
async def get_relevant_knowledge(
self,
npc_id: uuid.UUID,
player_id: uuid.UUID,
limit: int = 10,
) -> list[KnowledgeFact]:
"""Get facts relevant to a conversation with a specific player.
Returns facts about the player, plus high-importance world/faction facts.
Expired facts (past expires_at) are excluded.
"""
collection = self._store.get_collection(self.COLLECTION)
docs = await collection.query(QueryOptions(
filters={"npc_id": str(npc_id)},
order_by="importance",
order=SortOrder.DESC,
limit=limit * 3, # Over-fetch, then filter in-process
))
now = datetime.now(UTC)
facts = [self._document_to_fact(d) for d in docs]
# Filter expired and prioritise player-relevant facts
active = [f for f in facts if not f.expires_at or f.expires_at > now]
player_facts = [f for f in active if f.subject_id == player_id]
other_facts = [f for f in active if f.subject_id != player_id]
return (player_facts + other_facts)[:limit]
async def get_shareable_knowledge(self, npc_id: uuid.UUID) -> list[KnowledgeFact]:
"""Get facts this NPC can share via gossip (importance > 0.5)."""
all_facts = await self.get_relevant_knowledge(npc_id, player_id=None, limit=50)
return [f for f in all_facts if f.importance >= 0.5]
async def get_knowledge_by_tag(self, npc_id: uuid.UUID, tag: str) -> list[KnowledgeFact]:
"""Get facts matching a specific tag (used by MemoryQueryAPI)."""
collection = self._store.get_collection(self.COLLECTION)
docs = await collection.query(QueryOptions(
filters={"npc_id": str(npc_id)},
limit=200,
))
facts = [self._document_to_fact(d) for d in docs]
return [f for f in facts if tag in f.tags]
async def already_knows(
self, npc_id: uuid.UUID, source_memory_id: uuid.UUID
) -> bool:
"""Check if NPC already has knowledge derived from a specific memory."""
collection = self._store.get_collection(self.COLLECTION)
docs = await collection.query(QueryOptions(
filters={"npc_id": str(npc_id), "source_memory_id": str(source_memory_id)},
limit=1,
))
return len(docs) > 0
Gossip Propagation Model¶
@dataclass
class GossipMessage:
"""A piece of knowledge being propagated between NPCs."""
id: uuid.UUID
source_memory_id: uuid.UUID # Original memory that spawned this gossip
content: str
original_npc_id: uuid.UUID # NPC who first learned this
current_holder_id: uuid.UUID # NPC currently holding this gossip
importance: float # Must be > 0.5 to propagate
confidence: float # Decays -0.1 per hop
hop_count: int = 0
max_hops: int = 5 # Prevents infinite propagation
provenance: list[uuid.UUID] = field(default_factory=list) # Chain of NPCs for cycle-breaking
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
GossipSystem¶
The gossip system enables NPCs to share knowledge organically, creating emergent information propagation across the game world.
Motivation: Why NPCs Gossip¶
Gossip is not random chatter — it is driven by NPC needs and desires (see Doc 07 Need & Desire Model). An NPC gossips when doing so serves a purpose:
- Social need: NPCs with high social drive share gossip to build relationships.
- Safety need: NPCs warn allies about dangerous players (fear-driven gossip).
- Loyalty need: Faction-loyal NPCs propagate faction-relevant intelligence.
- Curiosity need: NPCs with high curiosity seek and share novel information.
The GossipSystem consults an NPC's active needs to weight gossip selection, preferring
messages that align with the speaker's current motivations.
class GossipSystem(System):
"""ECS System that runs gossip propagation each tick.
Budget: max 3 gossip exchanges per tick globally to prevent performance impact.
Performance: Instead of scanning all rooms and all NPC pairs each tick (O(rooms × NPCs²)),
we maintain a pre-computed gossip-ready queue. The queue is rebuilt when NPCs move
between rooms or gain new gossip-worthy knowledge, keeping per-tick cost O(budget).
Conforms to the System ABC: constructor accepts world: World and calls
super().__init__(world). The abstract method is update(), not tick().
"""
GOSSIP_BUDGET_PER_TICK = 3
MIN_IMPORTANCE_TO_PROPAGATE = 0.5
CONFIDENCE_DECAY_PER_HOP = 0.1
MIN_CONFIDENCE_TO_PROPAGATE = 0.2
def __init__(self, world: World, knowledge_mgr: KnowledgeGraphManager) -> None:
super().__init__(world)
self._knowledge_mgr = knowledge_mgr
# Pre-computed queue of (speaker_id, listener_id, room_id) tuples
# Rebuilt on NPC movement or new knowledge events; avoids per-tick O(n²) scan
self._gossip_queue: deque[tuple[uuid.UUID, uuid.UUID, uuid.UUID]] = deque()
self._queue_dirty = True
def mark_dirty(self) -> None:
"""Called by event handlers when NPC moves or gains new knowledge."""
self._queue_dirty = True
async def update(self, delta: float) -> None:
"""Select and propagate gossip messages, respecting per-tick budget."""
if self._queue_dirty:
self._rebuild_gossip_queue()
self._queue_dirty = False
budget_remaining = self.GOSSIP_BUDGET_PER_TICK
while budget_remaining > 0 and self._gossip_queue:
speaker_id, listener_id, _ = self._gossip_queue.popleft()
gossip = await self._select_gossip(speaker_id, listener_id)
if gossip:
await self._propagate(gossip, speaker_id, listener_id)
budget_remaining -= 1
def _rebuild_gossip_queue(self) -> None:
"""Pre-compute gossip-eligible NPC pairs from co-located groups."""
self._gossip_queue.clear()
for room_id, npcs in self._get_colocated_npc_groups():
for speaker, listener in self._select_gossip_pairs(npcs):
self._gossip_queue.append((speaker, listener, room_id))
async def _select_gossip(
self, speaker_id: uuid.UUID, listener_id: uuid.UUID
) -> GossipMessage | None:
"""Pick the best gossip to share, respecting propagation rules."""
candidates = await self._knowledge_mgr.get_shareable_knowledge(speaker_id)
for gossip in sorted(candidates, key=lambda g: g.importance, reverse=True):
# Skip if below importance threshold
if gossip.importance < self.MIN_IMPORTANCE_TO_PROPAGATE:
continue
# Skip if confidence too low
if gossip.confidence < self.MIN_CONFIDENCE_TO_PROPAGATE:
continue
# Cycle-breaking: skip if listener is already in provenance chain
if listener_id in gossip.provenance:
continue
# Skip if at max hops
if gossip.hop_count >= gossip.max_hops:
continue
# Skip if listener already knows this
if await self._knowledge_mgr.already_knows(listener_id, gossip.source_memory_id):
continue
return gossip
return None
async def _propagate(
self,
gossip: GossipMessage,
speaker_id: uuid.UUID,
listener_id: uuid.UUID,
) -> None:
"""Propagate gossip with distortion and provenance tracking."""
new_gossip = GossipMessage(
id=uuid4(),
source_memory_id=gossip.source_memory_id,
content=gossip.content, # Content stays same; confidence tracks degradation
original_npc_id=gossip.original_npc_id,
current_holder_id=listener_id,
importance=gossip.importance,
confidence=gossip.confidence - self.CONFIDENCE_DECAY_PER_HOP,
hop_count=gossip.hop_count + 1,
provenance=[*gossip.provenance, speaker_id],
)
await self._knowledge_mgr.store_knowledge(listener_id, new_gossip)
Gameplay Integration¶
MemoryEventBridge¶
The MemoryEventBridge translates relationship threshold crossings into game events
via the existing EventBus. This allows ECS systems (combat, quests, dialogue) to
react to reputation changes without coupling to the memory system.
Events are also mapped to the Doc 07 StorySignal schema so the quest generator
and narrative systems can detect milestones (see "StorySignal Mapping" below).
class MemoryEventBridge:
"""Bridges relationship changes to game events via EventBus.
The _previous dict tracks last-known dimension values for threshold crossing
detection. It uses LRU eviction (max 10,000 entries) and is persisted to
DocumentStore so that threshold crossings are not re-fired after a cold restart.
Persistence uses dirty-key tracking: only entries modified since the last
persist call are written, avoiding O(n) full-dict rewrites.
"""
THRESHOLDS: list[tuple[str, str, float, type[Event]]] = [
# (dimension, direction, threshold, event_type)
("trust", "below", 0.2, NPCHostileEvent),
("trust", "above", 0.8, NPCTrustedAllyEvent),
("fear", "above", 0.7, NPCCoweredEvent),
("respect", "above", 0.8, NPCRespectEvent),
("friendliness", "above", 0.9, NPCBestFriendEvent),
("friendliness", "below", 0.1, NPCDespiseEvent),
]
MAX_TRACKED_PAIRS = 10_000
COLLECTION = "memory_event_bridge_state"
def __init__(self, event_bus: EventBus, doc_store: DocumentStore):
self._event_bus = event_bus
self._doc_store = doc_store
# LRU-bounded dict tracking previous dimension values
self._previous: OrderedDict[tuple[uuid.UUID, uuid.UUID, str], float] = OrderedDict()
# Track keys modified since last persist to avoid O(n) full rewrites
self._dirty_keys: set[tuple[uuid.UUID, uuid.UUID, str]] = set()
async def seed_from_store(self) -> None:
"""Load persisted threshold state on cold start.
Prevents re-firing threshold events that were already emitted before restart.
"""
collection = self._doc_store.get_collection(self.COLLECTION)
docs = await collection.query(QueryOptions(limit=self.MAX_TRACKED_PAIRS * 6))
for doc in docs:
key = (uuid.UUID(doc["npc_id"]), uuid.UUID(doc["player_id"]), doc["dimension"])
self._previous[key] = doc["value"]
async def check_thresholds(
self, relationship: Relationship, npc_id: uuid.UUID, player_id: uuid.UUID
) -> None:
"""Check all thresholds and fire events on crossings."""
for dim, direction, threshold, event_type in self.THRESHOLDS:
current = getattr(relationship, dim, 0.5)
prev_key = (npc_id, player_id, dim)
previous = self._previous.get(prev_key, 0.5)
crossed = (
(direction == "below" and previous >= threshold and current < threshold) or
(direction == "above" and previous <= threshold and current > threshold)
)
if crossed:
event = event_type(
npc_id=npc_id,
player_id=player_id,
dimension=dim,
value=current,
)
await self._event_bus.emit(event)
# Also emit as typed StorySignal for Doc 07 quest/narrative integration
await self._emit_story_signal(event)
# Update tracked value with LRU eviction
self._previous[prev_key] = current
self._previous.move_to_end(prev_key)
self._dirty_keys.add(prev_key)
while len(self._previous) > self.MAX_TRACKED_PAIRS * 6:
evicted_key, _ = self._previous.popitem(last=False)
self._dirty_keys.discard(evicted_key)
# Persist only changed entries (debounced in production)
await self._persist_dirty()
async def _emit_story_signal(self, event: Event) -> None:
"""Map relationship threshold events to typed StorySignal variants.
Uses StorySignalType enum to ensure type safety. The quest generator
(Doc 08) and narrative systems (Doc 07) subscribe to these signal types.
"""
signal = StorySignal(
signal_type=StorySignalType.RELATIONSHIP_CHANGED,
source_entity_id=event.npc_id,
target_entity_id=event.player_id,
payload={
"event_type": type(event).__name__,
"dimension": event.dimension,
"value": event.value,
},
)
await self._event_bus.emit(signal)
async def _persist_dirty(self) -> None:
"""Persist only modified entries to DocumentStore for cold start recovery."""
if not self._dirty_keys:
return
collection = self._doc_store.get_collection(self.COLLECTION)
for npc_id, player_id, dim in self._dirty_keys:
value = self._previous.get((npc_id, player_id, dim))
if value is not None:
await collection.save({
"npc_id": str(npc_id),
"player_id": str(player_id),
"dimension": dim,
"value": value,
})
self._dirty_keys.clear()
class StorySignalType(Enum):
"""Typed signal types for quest/narrative integration (Doc 07/08)."""
RELATIONSHIP_CHANGED = "relationship_changed" # Threshold crossing on any dimension
SECRET_LEARNED = "secret_learned" # NPC learned a secret via gossip or dialogue
MEMORY_MILESTONE = "memory_milestone" # NPC accumulated N memories with a player
FACTION_REPUTATION_SHIFT = "faction_reputation_shift" # Player rep changed with a faction
# Example events — also mapped to StorySignal for Doc 07 integration
@dataclass
class NPCHostileEvent(Event):
"""Fired when NPC trust drops below 0.2 — NPC becomes hostile."""
npc_id: uuid.UUID
player_id: uuid.UUID
dimension: str
value: float
@dataclass
class NPCTrustedAllyEvent(Event):
"""Fired when NPC trust rises above 0.8 — unlocks ally dialogue/quests."""
npc_id: uuid.UUID
player_id: uuid.UUID
dimension: str
value: float
Memory Tag Queries for Game Systems¶
Game systems can query memories by tag to drive gameplay:
# Quest system checks if player has helped NPCs in a faction
helpful_memories = await memory_service.get_memories_by_tag(
npc_id=faction_leader_id, player_id=player_id, tag="helpful"
)
if len(helpful_memories) >= 3:
await quest_system.unlock_quest(player_id, "faction_champion")
# Combat system checks NPC's fear level to determine flee behavior
relationship = await rel_mgr.get_relationship(npc_id, player_id)
if relationship and relationship.fear > 0.7:
await npc_ai.set_behavior(npc_id, NPCBehavior.FLEE)
Structured Access for Systems¶
Doc 07's UtilityScorer and other ECS systems need programmatic, logic-based access
to memory and relationship data — not just natural-language prompt sections. The
MemoryQueryAPI provides this structured interface:
class MemoryQueryAPI:
"""Structured query interface for game systems (combat, quests, autonomy).
Unlike ContextProviders (which produce natural-language prompt text), this API
returns typed Python objects for use in game logic. Doc 07's UtilityScorer and
Doc 08's quest generator use this API.
**Architectural note:** This class is an intentional facade that spans the
Memory, Relationship, and Knowledge bounded contexts. It re-couples them at
the query layer for ergonomic consumption by game systems. This is an
acceptable trade-off: the underlying contexts remain independently testable
and deployable, while game systems get a single entry point. If the API
surface grows beyond ~15 methods, consider splitting into MemoryQueries,
RelationshipQueries, and KnowledgeQueries interfaces.
"""
def __init__(
self,
memory_service: MemoryService,
relationship_mgr: RelationshipManager,
knowledge_mgr: KnowledgeGraphManager,
) -> None:
self._memory = memory_service
self._relationships = relationship_mgr
self._knowledge = knowledge_mgr
# --- Relationship queries (used by UtilityScorer) ---
async def get_trust(self, npc_id: uuid.UUID, player_id: uuid.UUID) -> float:
"""Return trust score (0.0-1.0). Returns 0.5 (neutral) if no relationship exists."""
rel = await self._relationships.get_relationship(npc_id, player_id)
return rel.trust if rel else 0.5
async def get_dimension(
self, npc_id: uuid.UUID, player_id: uuid.UUID, dimension: str
) -> float:
"""Return any relationship dimension by name. Returns 0.5 if unknown."""
rel = await self._relationships.get_relationship(npc_id, player_id)
return getattr(rel, dimension, 0.5) if rel else 0.5
async def has_met(self, npc_id: uuid.UUID, player_id: uuid.UUID) -> bool:
"""True if NPC has any memories of or relationship with this player."""
rel = await self._relationships.get_relationship(npc_id, player_id)
if rel is not None:
return True
memories = await self._memory.get_memories(npc_id, player_id)
return len(memories) > 0
async def get_relationship_stage(
self, npc_id: uuid.UUID, player_id: uuid.UUID
) -> RelationshipStage | None:
"""Return current relationship stage, or None if no relationship."""
rel = await self._relationships.get_relationship(npc_id, player_id)
return rel.relationship_stage if rel else None
# --- Memory queries (used by quest generator, narrative systems) ---
async def count_memories_by_tag(
self, npc_id: uuid.UUID, player_id: uuid.UUID, tag: str
) -> int:
"""Count memories with a specific tag for threshold checks."""
memories = await self._memory.get_memories_by_tag(npc_id, player_id, tag)
return len(memories)
async def has_memory_matching(
self, npc_id: uuid.UUID, player_id: uuid.UUID, tags: list[str],
min_importance: float = 0.0,
) -> bool:
"""Check if NPC has any memory matching ALL given tags above importance threshold."""
memories = await self._memory.get_memories(npc_id, player_id)
tag_set = set(tags)
return any(
tag_set.issubset(set(m.tags)) and m.importance >= min_importance
for m in memories
)
async def get_disposition(self, npc_id: uuid.UUID, player_id: uuid.UUID) -> float:
"""Return overall disposition score (-1.0 to 1.0) for high-level checks."""
rel = await self._relationships.get_relationship(npc_id, player_id)
return calculate_disposition_score(rel) if rel else 0.0
# --- Knowledge queries ---
async def knows_fact(self, npc_id: uuid.UUID, fact_tag: str) -> bool:
"""Check if NPC knows a specific tagged fact (e.g., 'bandit_attack')."""
facts = await self._knowledge.get_knowledge_by_tag(npc_id, fact_tag)
return len(facts) > 0
Usage in Doc 07 UtilityScorer:
# In a UtilityScorer evaluation function:
async def score_help_player(self, npc_id: uuid.UUID, player_id: uuid.UUID) -> float:
"""Score how motivated NPC is to help this player."""
trust = await self.memory_api.get_trust(npc_id, player_id)
has_met = await self.memory_api.has_met(npc_id, player_id)
disposition = await self.memory_api.get_disposition(npc_id, player_id)
if not has_met:
return 0.3 # Baseline willingness for strangers
if trust < 0.2:
return 0.0 # Won't help untrusted players
return 0.5 + (disposition * 0.5) # Scale by overall disposition
Quest System Integration¶
Doc 08 confirms memory-driven quest generation as a first-class feature in
maid-stdlib. The quest generator uses MemoryQueryAPI to detect narrative
milestones and offer contextual follow-up quests:
# In QuestGenerator (maid-stdlib):
async def check_memory_driven_quests(
self, npc_id: uuid.UUID, player_id: uuid.UUID
) -> list[Quest]:
"""Generate dynamic quests based on NPC memories of this player."""
quests: list[Quest] = []
# Example: NPC remembers player helped with bandits → follow-up quest
if await self.memory_api.has_memory_matching(
npc_id, player_id, tags=["combat", "bandits", "heroic"], min_importance=0.6
):
stage = await self.memory_api.get_relationship_stage(npc_id, player_id)
if stage in (RelationshipStage.FRIEND, RelationshipStage.TRUSTED_ALLY):
quests.append(Quest(
name="Track the Bandit Leader",
description="The NPC trusts you enough to share intelligence about the bandit leader's hideout.",
source_npc_id=npc_id,
))
return quests
Relationship Evaluation¶
RelationshipEvaluator¶
When memories are extracted, the system must determine whether and how they affect
relationship dimensions. RelationshipEvaluator bridges memory extraction to
RelationshipManager updates, providing the missing trigger logic:
class RelationshipDelta(TypedDict, total=False):
"""Typed relationship dimension changes."""
trust: float
fear: float
respect: float
loyalty: float
friendliness: float
romantic_interest: float
class RelationshipEvaluator:
"""Evaluates extracted memories and produces relationship dimension deltas.
Called by the write pipeline after MemoryExtractor produces memories. For each
memory, the evaluator maps (emotional_valence, tags, importance) to concrete
dimension changes, then delegates to RelationshipManager.update_relationship().
Rule-based in Phase 1; may be augmented with LLM evaluation in Phase 2.
"""
# Tag → dimension mapping with base delta magnitudes.
# Actual delta = base × memory.importance × intensity_factor.
TAG_DIMENSION_MAP: ClassVar[dict[str, list[tuple[str, float]]]] = {
"helpful": [("trust", +0.05), ("friendliness", +0.03), ("respect", +0.02)],
"combat": [("respect", +0.04), ("fear", -0.02)],
"heroic": [("trust", +0.06), ("respect", +0.05)],
"trade": [("trust", +0.02), ("friendliness", +0.01)],
"threat": [("trust", -0.08), ("fear", +0.06)],
"betrayal": [("trust", -0.15), ("loyalty", -0.10), ("friendliness", -0.08)],
"gift": [("friendliness", +0.04), ("loyalty", +0.02)],
"insult": [("friendliness", -0.06), ("respect", -0.03)],
}
# Emotional valence baseline adjustments (applied in addition to tag-based deltas)
VALENCE_ADJUSTMENTS: ClassVar[dict[str, dict[str, float]]] = {
"positive": {"friendliness": +0.02, "trust": +0.01},
"negative": {"friendliness": -0.02, "trust": -0.01},
"neutral": {},
}
def __init__(
self,
relationship_mgr: RelationshipManager,
event_bridge: MemoryEventBridge,
) -> None:
self._relationship_mgr = relationship_mgr
self._event_bridge = event_bridge
async def evaluate_memories(
self,
npc_id: uuid.UUID,
player_id: uuid.UUID,
memories: list[BaseMemory],
) -> Relationship:
"""Evaluate extracted memories and update relationship accordingly.
Accumulates deltas from all memories in the batch, then applies them
in a single RelationshipManager.update_relationship() call to reduce
concurrency conflicts.
"""
accumulated: dict[str, float] = {}
for memory in memories:
deltas = self._compute_deltas(memory)
for dim, delta in deltas.items():
accumulated[dim] = accumulated.get(dim, 0.0) + delta
if not accumulated:
# No relationship-affecting memories — return current state unchanged
return await self._relationship_mgr.get_relationship(npc_id, player_id)
relationship = await self._relationship_mgr.update_relationship(
npc_id, player_id, accumulated,
)
# Check thresholds for game event emission
await self._event_bridge.check_thresholds(relationship, npc_id, player_id)
return relationship
def _compute_deltas(self, memory: BaseMemory) -> dict[str, float]:
"""Compute relationship dimension deltas from a single memory."""
deltas: dict[str, float] = {}
# Tag-based deltas
for tag in memory.tags:
if tag in self.TAG_DIMENSION_MAP:
for dim, base_delta in self.TAG_DIMENSION_MAP[tag]:
scaled = base_delta * memory.importance
# Scale by emotional intensity for episodic memories
if isinstance(memory, EpisodicMemory):
scaled *= (0.5 + memory.emotional_intensity * 0.5)
deltas[dim] = deltas.get(dim, 0.0) + scaled
# Valence baseline
if isinstance(memory, EpisodicMemory):
valence_key = memory.emotional_valence.value
for dim, adj in self.VALENCE_ADJUSTMENTS.get(valence_key, {}).items():
deltas[dim] = deltas.get(dim, 0.0) + adj
return deltas
Memory Consolidation¶
Episodic → Semantic Summarization¶
Over time, NPCs accumulate many episodic memories about the same topic or player behaviour. Memory consolidation periodically summarises clusters of related episodic memories into single semantic memories, reducing storage while preserving knowledge:
class MemoryConsolidator:
"""Batch job that consolidates episodic memories into semantic summaries.
Runs as a scheduled background task (e.g., daily or weekly during low-traffic
hours). Uses LLM to summarise clusters of related episodes, then stores the
summary as a SemanticMemory and marks source episodes as consolidated.
Consolidation criteria:
- At least 3 episodic memories sharing a common tag
- All memories older than 7 days (don't consolidate recent memories)
- Locked memories are included in summaries but never deleted
"""
MIN_CLUSTER_SIZE = 3
MIN_AGE_DAYS = 7
CONSOLIDATION_PROMPT = '''Summarise these NPC memories into a single general fact.
The summary should capture the overall pattern or relationship, not individual events.
Return JSON: {"content": "...", "tags": [...], "confidence": 0.0-1.0}
Memories:
{memories}'''
def __init__(
self,
memory_service: MemoryService,
llm: LLMProvider,
rate_limiter: MemoryRateLimiter,
circuit_breaker_registry: CircuitBreakerRegistry,
) -> None:
self._memory = memory_service
self._llm = llm
self._rate_limiter = rate_limiter
self._breaker = circuit_breaker_registry.get_or_create(
f"consolidation:{llm.name}"
)
async def consolidate_pair(
self, npc_id: uuid.UUID, player_id: uuid.UUID
) -> list[SemanticMemory]:
"""Consolidate episodic memories for a single NPC-player pair."""
cutoff = datetime.now(UTC) - timedelta(days=self.MIN_AGE_DAYS)
episodes = await self._memory.get_memories(npc_id, player_id)
eligible = [
m for m in episodes
if m.memory_type == MemoryType.EPISODIC and m.created_at < cutoff
]
# Cluster by shared tags
clusters = self._cluster_by_tags(eligible)
created: list[SemanticMemory] = []
for tag, cluster in clusters.items():
if len(cluster) < self.MIN_CLUSTER_SIZE:
continue
# Rate-limit consolidation LLM calls
if not await self._rate_limiter.check_memory_extraction(npc_id, 300):
break
summary = await self._summarise_cluster(cluster)
if summary is None:
continue
semantic = SemanticMemory(
id=uuid4(),
npc_id=npc_id,
player_id=player_id,
content=summary["content"],
tags=summary["tags"],
importance=max(m.importance for m in cluster),
confidence=summary["confidence"],
salience=0.0, # Recomputed by decay job
created_at=datetime.now(UTC),
last_accessed=datetime.now(UTC),
access_count=0,
source_episodes=[m.id for m in cluster],
)
await self._memory.store_memory(semantic)
created.append(semantic)
# Delete unlocked source episodes (locked ones are retained)
for mem in cluster:
if not mem.locked:
await self._memory.delete_memory(mem.id)
await self._rate_limiter.record_memory_usage(npc_id, 300)
return created
async def _summarise_cluster(
self, cluster: list[BaseMemory]
) -> dict[str, Any] | None:
"""Use LLM to summarise a cluster of related memories."""
memory_text = "\n".join(f"- {m.content}" for m in cluster)
prompt = self.CONSOLIDATION_PROMPT.format(memories=memory_text)
try:
result = await self._breaker.call(
lambda: self._llm.complete(
messages=[
{"role": "system", "content": "You are a memory consolidation engine."},
{"role": "user", "content": prompt},
],
options=CompletionOptions(temperature=0.3, max_tokens=200),
),
)
return json.loads(result.content)
except (json.JSONDecodeError, CircuitBreakerOpenError):
return None
def _cluster_by_tags(
self, memories: list[BaseMemory]
) -> dict[str, list[BaseMemory]]:
"""Group memories by their most common shared tag."""
clusters: dict[str, list[BaseMemory]] = {}
for mem in memories:
for tag in mem.tags:
clusters.setdefault(tag, []).append(mem)
return clusters
Reliability & Observability¶
Error Budget Targets¶
| Metric | Target | Degraded Threshold |
|---|---|---|
| Memory retrieval p99 latency | < 40ms | < 100ms |
| Memory extraction success rate | > 95% | > 80% |
| Prompt building p99 latency (with context) | < 60ms | < 150ms |
| Cache hit rate | > 70% | > 50% |
| Memory write success rate | > 99% | > 95% |
Required Metrics¶
All metrics emitted via the existing profiling system (maid_engine/profiling/):
# Counters
memory_extractions_total # Labels: status={success,failed,skipped}
memory_cache_hits_total
memory_cache_misses_total
gossip_propagations_total
safety_flags_total # Labels: flag={suspect,redacted}
extraction_jobs_total # Labels: state={pending,extracting,storing,complete,failed}
# Histograms
memory_retrieval_duration_ms
prompt_context_build_duration_ms
memory_extraction_duration_ms
# Gauges
memory_cache_size
memory_count_per_npc_player # Labels: memory_type={episodic,semantic,procedural}
active_extraction_jobs
working_memory_buffer_size
Degraded Mode Definition¶
When error budgets are exceeded, the system enters degraded mode:
class MemorySystemHealth:
"""Health monitor with automatic degraded mode transitions."""
async def check_health(self) -> HealthStatus:
"""Called by /admin/health endpoint."""
checks = {
"cache_hit_rate": self._cache_hit_rate() > 0.5,
"extraction_success_rate": self._extraction_success_rate() > 0.8,
"retrieval_p99_ms": self._retrieval_p99() < 100,
"store_available": await self._store_ping(),
}
if all(checks.values()):
return HealthStatus.HEALTHY
elif checks["store_available"] and checks["extraction_success_rate"]:
return HealthStatus.DEGRADED
else:
return HealthStatus.UNHEALTHY
async def enter_degraded_mode(self) -> None:
"""Graceful degradation: disable non-critical features."""
self._disable_gossip_propagation()
self._disable_memory_consolidation()
self._extend_cache_ttl(factor=3) # Reduce DB pressure
# Memory extraction and retrieval continue with fallbacks
Health Check Endpoint¶
Exposed via the existing admin API:
Open Questions¶
1. Vector Embeddings for Semantic Similarity¶
Question: Should Phase 2 implement vector embeddings for memory retrieval instead of keyword matching?
Trade-offs: - Pros: Much more accurate relevance matching, handles paraphrasing and implicit references - Cons: Requires embedding model (OpenAI, local Sentence Transformers), vector database, increased complexity
Recommendation: Implement in Phase 3 after keyword-based system proves stable. Start with lightweight embeddings (sentence-transformers/all-MiniLM-L6-v2) locally hosted.
2. Memory Privacy and Player Consent¶
Question: Should players have control over what NPCs remember about them?
Considerations: - Immersion vs. Privacy: Some players may want "right to be forgotten" - Gameplay Impact: Memory wipe could be quest reward or spell effect - Technical Implementation: Player preferences in settings, admin override commands
Recommendation: Add player preference "Allow NPC Memory" (default: true) with in-game explanation of benefits.
3. Cross-Server Memory Sharing¶
Question: For multi-server deployments, should NPC memories be shared across server instances?
Use Cases: - Player travels between server regions with same character - NPC appears in multiple locations across server cluster
Recommendation: Phase 3 feature. Implement memory export/import for server migrations first.
4. Structured Access for Systems (Doc 07 Autonomy Support)¶
Doc 07's UtilityScorer and other game systems need logic-based access to memory
and relationship data — not just natural-language prompt context. For example, a
UtilityScorer needs to evaluate trust > 0.5 or has_met(player) directly in code.
Resolution: Addressed by the MemoryQueryAPI in the
Structured Access for Systems section above.
5. Emotional Memory Weighting¶
Question: Should emotionally intense memories have longer retention regardless of importance?
Psychological Basis: Flashbulb memories persist longer due to emotional intensity
Implementation: Modify decay formula to include emotional_intensity factor
Recommendation: Include in Phase 1 with conservative weighting (10% impact maximum).
Design Decisions Log¶
Decision 1: LLM-Based vs. Rule-Based Extraction (2025-01-10)¶
Decision: Use LLM-based extraction as primary method with rule-based fallback
Rationale: - Rule-based extraction cannot handle natural language complexity - LLM extraction provides superior quality despite higher cost - Fallback ensures system remains functional under budget constraints
Alternatives Considered: - Rule-based only: Lower cost but poor quality - Hybrid approach: Complex implementation, unclear benefits
Decision 2: Multi-Dimensional vs. Single-Score Relationships (2025-01-10)¶
Decision: Multi-dimensional relationship model (trust, fear, respect, loyalty, friendliness, romantic_interest)
Rationale: - Provides richer NPC behavior variation - Enables complex social dynamics (high respect + low trust) - Supports faction-based relationship initialization
Alternatives Considered: - Single disposition score: Simpler but less expressive - Emotion-based model: Too psychological, less game-focused
Decision 3: DocumentStore vs. Dedicated Memory Database (2025-01-10)¶
Decision: Use existing DocumentStore with PostgreSQL JSONB backend
Rationale:
- Leverages existing infrastructure
- JSONB provides sufficient query performance with proper indexing
- Avoids additional database dependencies
Alternatives Considered: - Vector database (Chroma, Pinecone): Premature optimization for Phase 1 - Graph database (Neo4j): Overkill for current requirements - In-memory storage: Doesn't survive restarts
Decision 4: Token Budget Allocation (2025-01-10)¶
Decision: Allocate 25% of context window to memories, 10% to relationships, 10% to knowledge
Rationale: - Provides meaningful memory context without overwhelming prompt - Leaves room for conversation history and base system prompt - Percentages configurable per content pack
Alternatives Considered: - Fixed token counts: Less flexible across different context window sizes - Dynamic allocation: Too complex for Phase 1 - Higher memory allocation: Would starve conversation history
Decision 5: Gossip System Implementation (2025-01-10)¶
Decision: Room-based gossip with rate-limited propagation
Rationale: - Realistic information sharing (NPCs must be co-located) - Rate limiting prevents gossip storms - Background processing doesn't block gameplay
Alternatives Considered: - Faction-based gossip: Too complex for Phase 1 - Player-mediated gossip: Requires player involvement - Instant global knowledge: Unrealistic and performance-intensive
Decision 6: Memory Content Safety Strategy (2025-01-10)¶
Decision: Filter player input before memory storage + flag suspicious content
Rationale: - Prevents content injection at source - Maintains system integrity while preserving most legitimate content - Provides admin tools for oversight
Alternatives Considered: - Filter only LLM output: Doesn't prevent storage of harmful content - Human review: Doesn't scale - No filtering: Vulnerable to player manipulation
Implementation Plan¶
Phase 1: Core Memory (4-6 weeks)¶
Week 1-2: Data Models & Storage
- [ ] Implement BaseMemory, EpisodicMemory, SemanticMemory, ProceduralMemory models (maid-stdlib)
- [ ] Implement MemoryDocumentStore using unified npc_memories collection with memory_type discriminator
- [ ] Build MemoryCache with LRU eviction, write-through invalidation, and asyncio.Lock (maid-engine)
- [ ] Create PIIRedactor pre-processing layer (maid-engine)
- [ ] Create MemoryService API with CRUD operations and growth cap enforcement (maid-stdlib)
Week 3-4: Memory Extraction & Prompt Integration
- [ ] Implement MemoryExtractionJob state machine with idempotency (maid-stdlib)
- [ ] Implement MemoryExtractor using LLMProvider.complete() with CircuitBreakerRegistry (maid-stdlib)
- [ ] Implement ExtractionJobQueue bounded async queue with worker pool (maid-stdlib)
- [ ] Create ContextProvider protocol and TokenBudgetManager (provider-aware, accepts list[ContextProvider]) (maid-engine, ai/context_providers.py)
- [ ] Create ContextOrchestrator with asyncio.gather()-based concurrent resolution (maid-engine)
- [ ] Create MemoryContextProvider (maid-stdlib)
- [ ] Create EnrichedPromptBuilder async wrapper with explicit UUID params; preserve PromptBuilder.build_system_prompt() signature
- [ ] Implement WorkingMemoryBuffer with DocumentStore persistence and TTL-bounded startup load (maid-stdlib)
- [ ] Integrate with NPCDialogueSystem conversation completion (async write pipeline via ExtractionJobQueue)
Week 5-6: Relationship System & Safety - [ ] Implement RelationshipManager with optimistic concurrency and retry loop (maid-stdlib) - [ ] Create RelationshipContextProvider (wires {reputation} placeholder) (maid-stdlib) - [ ] Implement RelationshipEvaluator with tag-based dimension mapping (maid-stdlib) - [ ] Add relationship progression logic with dimension clamping - [ ] Implement standalone MemoryRateLimiter (token-bucket, NPC-scoped daily budgets) (maid-engine) - [ ] Implement MemoryContentFilter with immutable (copy-on-write) safety_flag approach (maid-stdlib) - [ ] Implement flagged memory audit collection - [ ] Add MemoryEventBridge with LRU-bounded state, dirty-key persistence, and typed StorySignal mapping (maid-stdlib) - [ ] Implement MemoryQueryAPI facade for structured system access (maid-stdlib) - [ ] Comprehensive unit and integration testing
Deliverables:
- NPCs remember specific interactions with players
- Memories influence dialogue through ContextOrchestrator → PromptBuilder pipeline
- Basic relationship tracking affects NPC disposition via {reputation} placeholder
- RelationshipEvaluator triggers dimension updates from extracted memories
- Async write pipeline with bounded ExtractionJobQueue and idempotent extraction
- Content safety with PII redaction and non-destructive flagging
- Structured MemoryQueryAPI facade for Doc 07/08 integration
- Admin commands for memory inspection and audit review
Phase 2: Knowledge & Gossip (3-4 weeks)¶
Week 1-2: Knowledge Graph - [ ] Implement KnowledgeGraphManager, KnowledgeFact, and KnowledgeContextProvider (maid-stdlib) - [ ] Add semantic memory consolidation via MemoryConsolidator batch job - [ ] Create fact extraction from conversations
Week 3-4: Gossip System
- [ ] Build GossipSystem ECS system conforming to System ABC (__init__(world), update(delta)) with pre-computed gossip-ready queue and per-tick budget
- [ ] Implement need-driven gossip selection (social, safety, loyalty, curiosity motivations)
- [ ] Implement rate-limited propagation with confidence decay
- [ ] Add provenance chain tracking for cycle-breaking
- [ ] Wire EmotionalState/Mood into extraction pipeline and context providers
Deliverables: - NPCs learn and share facts about players and world - Knowledge spreads through NPC social networks with configurable distortion - Semantic memories consolidate from episodic experiences
Phase 3: Advanced Features (4-5 weeks)¶
Week 1-2: Performance, Scale & Observability - [ ] Implement MemoryDecaySystem background job (daily pruning) - [ ] Add memory growth bound enforcement with locked memory support - [ ] Add observability metrics (counters, histograms, gauges) - [ ] Implement MemorySystemHealth with degraded mode transitions - [ ] Add /admin/health/memory endpoint - [ ] Performance testing with 1000+ NPCs
Week 3-4: Content Safety & Admin Tools - [ ] Enhanced content filtering and manipulation detection - [ ] Comprehensive admin tools for memory management - [ ] Player privacy controls and data deletion - [ ] Memory locking commands for quest/faction systems
Week 5: Polish & Documentation - [ ] Content pack integration examples - [ ] Performance tuning and optimization - [ ] Documentation and usage guides
Deliverables: - Production-ready performance and scalability - Comprehensive admin and content creator tooling - Vector-based semantic memory retrieval
Dependencies¶
Hard Dependencies¶
- DocumentStore (entity persistence #01) — Required for memory storage; uses existing
QueryOptions(exact-match filters, order_by, limit); batch_update/batch_delete used by decay job - ConversationManager — Must call
end_conversation()to trigger async memory extraction jobs viaExtractionJobQueue - PromptBuilder — Existing sync signature preserved;
EnrichedPromptBuilderwraps it withContextOrchestratorfor async enrichment, callingbuild_system_prompt()internally - LLMProvider —
complete()method used byMemoryExtractorandMemoryConsolidator(nogenerate()method exists) - CircuitBreakerRegistry — Wraps LLM calls in
MemoryExtractorandMemoryConsolidatorfor fault tolerance - EventBus — Required by
MemoryEventBridgefor threshold-based game events and typedStorySignalemission - System ABC —
GossipSystemextendsSystem(requires__init__(world),update(delta)interface)
Soft Dependencies¶
- Doc 07 Autonomy System —
UtilityScorerconsumesMemoryQueryAPIfor structured access - Doc 08 Quest System — Quest generator uses
MemoryQueryAPIfor memory-driven quests - Faction System — Would improve relationship initialization
- Admin UI — Memory management would benefit from web interface
Conclusion¶
The Persistent NPC Memory and Relationships system represents MAID's strategic differentiator in the MUD space. By giving NPCs genuine memory, opinions, and social networks, every player interaction becomes meaningful and creates lasting impact on the game world.
The design balances ambitious vision with practical implementation concerns:
- Correctness — Uses real
LLMProvider.complete()API (notgenerate());EnrichedPromptBuilderwraps realPromptBuilder.build_system_prompt()signature without modification;GossipSystemconforms toSystemABC (update(),__init__(world));ContextProviderprotocol lives inai/context_providers.py(no collision with existingai/context.py); game-domain code in maid-stdlib, infrastructure protocols in maid-engine - Cost Control — Hard budget limits via authoritative TokenBudgetManager; fallback strategies keep LLM costs manageable
- Performance — LRU cache keyed by (npc_id, player_id), async write pipeline, 40ms read timeout with graceful degradation
- Reliability — Idempotent extraction jobs, optimistic concurrency, structured LLM output validation, memory growth caps
- Content Safety — Defense-in-depth: PII redaction → input filter → structured extraction → post-validation → non-destructive flagging
- Observability — Error budgets, required metrics, degraded mode definition, health check endpoint
- Scalability — Proven DocumentStore backend, bounded memory growth, per-tick gossip budgets
The phased implementation approach allows for incremental delivery and learning:
- Phase 1 establishes core memory and relationships (6 weeks)
- Phase 2 adds knowledge sharing and gossip (4 weeks)
- Phase 3 delivers advanced features and production polish (5 weeks)
This system transforms NPCs from stateless dialogue endpoints into characters with genuine persistence and growth, creating the immersive, AI-driven persistent world that MAID promises.
Total estimated effort: 15 weeks for complete implementation
Strategic impact: Positions MAID as the first MUD with truly persistent AI characters, creating sustainable competitive advantage in the AI gaming space.