""" Generic memory behaviors for WorldSmithAI. This module implements domain-agnostic memory behaviors for arbitrary agent-based worlds. It deliberately avoids domain-specific classes such as ScientificMemory, MarketMemory, CombatMemory, SocialBelief, SpellKnowledge, or InstitutionalMemory. Memory and beliefs are represented as generic dictionaries stored on an agent's memory mapping. This lets DSL-generated worlds model observations, beliefs, experiences, preferences, relationships, learned strategies, prices, laws, research findings, routes, rituals, or social information without changing the simulation engine. Implemented behaviors: - remember: store a memory record from explicit content or agent paths. - forget: mark records forgotten, remove records, or delete a state/memory path. - reinforce: strengthen or weaken selected memory records or numeric paths. - update_belief: update a numeric belief from deterministic evidence. Example: behavior = RememberBehavior( category="observation", content={"resource": "food", "amount": 4.0}, importance=0.8, tags=("food", "local"), ) outcome = behavior.execute(agent, world) behavior = UpdateBeliefBehavior( belief_id="food_is_scarce", proposition="Food is scarce nearby", evidence_path="state.food_scarcity_signal", learning_rate=0.25, ) outcome = behavior.execute(agent, world) Future extensibility: - Add memory decay in the scheduler. - Add vector embeddings while keeping records serializable. - Add episodic, semantic, and procedural namespaces. - Add retrieval behaviors for planning and contextual-bandit policies. - Add event emission for memory writes, forgetting, reinforcement, and belief shifts. - Add SLM-generated memory summaries without allowing the SLM to mutate world state directly. """ from __future__ import annotations import copy import logging from collections.abc import Iterable, Mapping, MutableMapping, MutableSequence, Sequence from dataclasses import dataclass, field from enum import Enum from numbers import Real from types import MappingProxyType from typing import TYPE_CHECKING, Any, ClassVar import numpy as np from core.behavior import Behavior if TYPE_CHECKING: from core.agent import Agent from core.world import World logger = logging.getLogger(__name__) _MISSING = object() _EPSILON = 1.0e-12 class StorageLocation(str, Enum): """Supported mutable storage locations on an agent.""" STATE = "state" MEMORY = "memory" class MemoryStatus(str, Enum): """Generic lifecycle statuses for memory records.""" ACTIVE = "active" REINFORCED = "reinforced" WEAKENED = "weakened" FORGOTTEN = "forgotten" class BeliefUpdateRule(str, Enum): """Supported deterministic belief update rules.""" WEIGHTED_AVERAGE = "weighted_average" BAYESIAN = "bayesian" @dataclass(frozen=True) class MemoryOutcome: """Serializable result returned by memory behavior execution. The core engine may ignore this object, but downstream systems such as metrics, visualizers, event streams, dashboards, debuggers, policies, and narrators can consume the structured payload. """ behavior: str actor_id: str success: bool memory_ids: tuple[str, ...] = () belief_ids: tuple[str, ...] = () record_ids: tuple[str, ...] = () step: int | None = None details: Mapping[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly representation of the outcome.""" return { "behavior": self.behavior, "actor_id": self.actor_id, "success": self.success, "memory_ids": list(self.memory_ids), "belief_ids": list(self.belief_ids), "record_ids": list(self.record_ids), "step": self.step, "details": copy.deepcopy(dict(self.details)), } @dataclass(frozen=True) class MemoryHandle: """Mutable reference to a stored memory record.""" key: str record: MutableMapping[str, Any] store: MutableMapping[str, Any] def _agent_id(agent: Agent) -> str: """Return a stable string identifier for an agent.""" return str(getattr(agent, "id")) def _is_alive(agent: Agent) -> bool: """Return whether an agent can participate in behavior execution.""" return bool(getattr(agent, "alive", True)) def _world_step(world: World) -> int | None: """Return the current world step if available.""" value = getattr(world, "step_count", None) if isinstance(value, Real) and not isinstance(value, bool): return int(value) return None def _is_number(value: Any) -> bool: """Return whether a value is a real numeric scalar, excluding booleans.""" return isinstance(value, (Real, np.integer, np.floating)) and not isinstance(value, bool) def _as_float(value: Any, default: float = 0.0) -> float: """Safely convert a numeric-like value to float.""" if _is_number(value): return float(value) return default def _as_int(value: Any, default: int = 0) -> int: """Safely convert a numeric-like value to int.""" if _is_number(value): return int(value) return default def _clamp(value: float, minimum: float, maximum: float) -> float: """Clamp a numeric value to inclusive bounds.""" return min(max(float(value), float(minimum)), float(maximum)) def _clamp_unit(value: float) -> float: """Clamp a numeric value to the interval [0, 1].""" return _clamp(value, 0.0, 1.0) def _normalize_storage(location: StorageLocation | str) -> StorageLocation: """Normalize a storage location value.""" if isinstance(location, StorageLocation): return location return StorageLocation(str(location)) def _normalize_belief_rule(rule: BeliefUpdateRule | str) -> BeliefUpdateRule: """Normalize a belief update rule value.""" if isinstance(rule, BeliefUpdateRule): return rule return BeliefUpdateRule(str(rule)) def _agent_state(agent: Agent) -> MutableMapping[str, Any]: """Return an agent's mutable state mapping, creating one if needed.""" state = getattr(agent, "state", None) if isinstance(state, MutableMapping): return state replacement: dict[str, Any] = {} setattr(agent, "state", replacement) return replacement def _agent_memory(agent: Agent) -> MutableMapping[str, Any]: """Return an agent's mutable memory mapping, creating one if needed.""" memory = getattr(agent, "memory", None) if isinstance(memory, MutableMapping): return memory replacement: dict[str, Any] = {} setattr(agent, "memory", replacement) return replacement def _container_for(agent: Agent, location: StorageLocation | str) -> MutableMapping[str, Any]: """Return an agent container for a state-or-memory storage location.""" normalized = _normalize_storage(location) if normalized is StorageLocation.STATE: return _agent_state(agent) if normalized is StorageLocation.MEMORY: return _agent_memory(agent) raise ValueError(f"Unsupported storage location: {location!r}") def _ensure_mapping(parent: MutableMapping[str, Any], key: str) -> MutableMapping[str, Any]: """Return a nested mutable mapping under ``key``, creating one if absent.""" value = parent.get(key) if isinstance(value, MutableMapping): return value if isinstance(value, Mapping): replacement = dict(value) parent[key] = replacement return replacement replacement: dict[str, Any] = {} parent[key] = replacement return replacement def _ensure_list(parent: MutableMapping[str, Any], key: str) -> MutableSequence[Any]: """Return a nested mutable sequence under ``key``, creating one if absent.""" value = parent.get(key) if isinstance(value, MutableSequence): return value replacement: list[Any] = [] parent[key] = replacement return replacement def _append_bounded(items: MutableSequence[Any], value: Any, max_items: int) -> None: """Append an item while enforcing an optional maximum history length.""" items.append(value) if max_items > 0 and len(items) > max_items: del items[: len(items) - max_items] def _split_path(path: str) -> tuple[str, ...]: """Split a dot-separated path into components.""" return tuple(part for part in str(path).split(".") if part) def _get_path(container: Mapping[str, Any], path: str, default: Any = None) -> Any: """Read a possibly nested value from a mapping using dot notation.""" parts = _split_path(path) if not parts: return default current: Any = container for part in parts: if not isinstance(current, Mapping) or part not in current: return default current = current[part] return current def _set_path(container: MutableMapping[str, Any], path: str, value: Any) -> None: """Write a possibly nested value to a mapping using dot notation.""" parts = _split_path(path) if not parts: return current: MutableMapping[str, Any] = container for part in parts[:-1]: nested = current.get(part) if not isinstance(nested, MutableMapping): nested = {} current[part] = nested current = nested current[parts[-1]] = value def _delete_path(container: MutableMapping[str, Any], path: str) -> bool: """Delete a possibly nested value from a mapping using dot notation.""" parts = _split_path(path) if not parts: return False current: MutableMapping[str, Any] = container for part in parts[:-1]: nested = current.get(part) if not isinstance(nested, MutableMapping): return False current = nested if parts[-1] not in current: return False current.pop(parts[-1], None) return True def _increment_path(container: MutableMapping[str, Any], path: str, delta: float) -> float: """Increment a numeric value at a path and return the updated value.""" current_value = _get_path(container, path, 0.0) updated_value = _as_float(current_value, 0.0) + float(delta) _set_path(container, path, updated_value) return updated_value def _read_agent_value(agent: Agent, path: str, default: Any = _MISSING) -> Any: """Read an agent value using optional state or memory prefixes. Supported prefixes: - ``state.foo.bar`` - ``state:foo.bar`` - ``memory.foo.bar`` - ``memory:foo.bar`` If no prefix is supplied, state is checked first, then memory. """ normalized_path = str(path) if normalized_path.startswith("state."): return _get_path(_agent_state(agent), normalized_path.removeprefix("state."), default) if normalized_path.startswith("state:"): return _get_path(_agent_state(agent), normalized_path.removeprefix("state:"), default) if normalized_path.startswith("memory."): return _get_path(_agent_memory(agent), normalized_path.removeprefix("memory."), default) if normalized_path.startswith("memory:"): return _get_path(_agent_memory(agent), normalized_path.removeprefix("memory:"), default) state_value = _get_path(_agent_state(agent), normalized_path, _MISSING) if state_value is not _MISSING: return state_value return _get_path(_agent_memory(agent), normalized_path, default) def _write_agent_value( agent: Agent, path: str, value: Any, *, default_location: StorageLocation | str = StorageLocation.MEMORY, ) -> None: """Write an agent value using optional state or memory prefixes.""" normalized_path = str(path) if normalized_path.startswith("state."): _set_path(_agent_state(agent), normalized_path.removeprefix("state."), value) return if normalized_path.startswith("state:"): _set_path(_agent_state(agent), normalized_path.removeprefix("state:"), value) return if normalized_path.startswith("memory."): _set_path(_agent_memory(agent), normalized_path.removeprefix("memory."), value) return if normalized_path.startswith("memory:"): _set_path(_agent_memory(agent), normalized_path.removeprefix("memory:"), value) return _set_path(_container_for(agent, default_location), normalized_path, value) def _delete_agent_value( agent: Agent, path: str, *, default_location: StorageLocation | str = StorageLocation.MEMORY, ) -> bool: """Delete an agent value using optional state or memory prefixes.""" normalized_path = str(path) if normalized_path.startswith("state."): return _delete_path(_agent_state(agent), normalized_path.removeprefix("state.")) if normalized_path.startswith("state:"): return _delete_path(_agent_state(agent), normalized_path.removeprefix("state:")) if normalized_path.startswith("memory."): return _delete_path(_agent_memory(agent), normalized_path.removeprefix("memory.")) if normalized_path.startswith("memory:"): return _delete_path(_agent_memory(agent), normalized_path.removeprefix("memory:")) return _delete_path(_container_for(agent, default_location), normalized_path) def _increment_agent_value( agent: Agent, path: str, delta: float, *, default_location: StorageLocation | str = StorageLocation.MEMORY, ) -> float: """Increment an agent value using optional state or memory prefixes.""" current_value = _read_agent_value(agent, path, 0.0) updated_value = _as_float(current_value, 0.0) + float(delta) _write_agent_value(agent, path, updated_value, default_location=default_location) return updated_value def _safe_equal(left: Any, right: Any) -> bool: """Return safe equality for arbitrary DSL values.""" try: result = left == right except (TypeError, ValueError): return False if isinstance(result, np.ndarray): return bool(result.all()) return bool(result) def _constraint_matches(actual: Any, expected: Any) -> bool: """Return whether a value satisfies a generic DSL constraint.""" if isinstance(expected, Mapping): exists = expected.get("exists") if exists is not None: has_value = actual is not _MISSING and actual is not None if bool(exists) != has_value: return False if actual is _MISSING: return False if "equals" in expected and not _safe_equal(actual, expected["equals"]): return False if "not_equals" in expected and _safe_equal(actual, expected["not_equals"]): return False if "min" in expected: if not _is_number(actual) or float(actual) < float(expected["min"]): return False if "max" in expected: if not _is_number(actual) or float(actual) > float(expected["max"]): return False if "in" in expected: valid_values = expected["in"] if not isinstance(valid_values, Iterable) or isinstance(valid_values, (str, bytes)): return False if actual not in valid_values: return False if "not_in" in expected: invalid_values = expected["not_in"] if isinstance(invalid_values, Iterable) and not isinstance(invalid_values, (str, bytes)): if actual in invalid_values: return False if "contains" in expected: contained_value = expected["contains"] if isinstance(actual, Mapping): if contained_value not in actual: return False elif isinstance(actual, Iterable) and not isinstance(actual, (str, bytes)): if contained_value not in actual: return False else: return False return True if actual is _MISSING: return False return _safe_equal(actual, expected) def _matches_constraints(agent: Agent, constraints: Mapping[str, Any]) -> bool: """Return whether an agent satisfies all configured constraints.""" for path, expected in constraints.items(): actual = _read_agent_value(agent, str(path), _MISSING) if not _constraint_matches(actual, expected): return False return True def _record_store( agent: Agent, *, location: StorageLocation | str, store_key: str, ) -> MutableMapping[str, Any]: """Return a mutable record store from state or memory.""" container = _container_for(agent, location) existing = _get_path(container, store_key) if isinstance(existing, MutableMapping): return existing if isinstance(existing, Mapping): replacement = dict(existing) _set_path(container, store_key, replacement) return replacement replacement: dict[str, Any] = {} _set_path(container, store_key, replacement) return replacement def _record_history( agent: Agent, *, history_key: str, record: Mapping[str, Any], max_history: int, ) -> None: """Append a bounded memory history record to agent memory.""" history = _ensure_list(_agent_memory(agent), history_key) _append_bounded(history, copy.deepcopy(dict(record)), max_history) def _generated_record_id( agent: Agent, *, behavior: str, category: str, sequence_count: int, step: int | None, ) -> str: """Generate a deterministic record id.""" step_label = "unknown_step" if step is None else str(step) return f"{step_label}:{_agent_id(agent)}:{behavior}:{category}:{sequence_count}" def _record_created_step(record: Mapping[str, Any]) -> int: """Return a stable created-step value for sorting.""" return _as_int(record.get("created_step"), 0) def _record_importance(record: Mapping[str, Any]) -> float: """Return a record's numeric importance.""" return _as_float(record.get("importance"), 0.0) def _record_strength(record: Mapping[str, Any]) -> float: """Return a record's numeric strength.""" return _as_float(record.get("strength"), 0.0) def _record_tags(record: Mapping[str, Any]) -> set[str]: """Return a record's tags as strings.""" tags = record.get("tags", ()) if isinstance(tags, Iterable) and not isinstance(tags, (str, bytes, Mapping)): return {str(tag) for tag in tags} if tags is None: return set() return {str(tags)} def _tags_match(record: Mapping[str, Any], required_tags: Sequence[str]) -> bool: """Return whether a record contains all required tags.""" if not required_tags: return True existing_tags = _record_tags(record) return all(str(tag) in existing_tags for tag in required_tags) def _memory_handle( store: MutableMapping[str, Any], record_id: str, ) -> MemoryHandle | None: """Return a mutable handle for a memory record by id.""" existing = store.get(record_id) if isinstance(existing, MutableMapping): return MemoryHandle(key=record_id, record=existing, store=store) if isinstance(existing, Mapping): replacement = dict(existing) store[record_id] = replacement return MemoryHandle(key=record_id, record=replacement, store=store) return None def _matching_memory_handles( store: MutableMapping[str, Any], *, record_id: str | None = None, category: str | None = None, tags: Sequence[str] = (), importance_below: float | None = None, max_age_steps: int | None = None, step: int | None = None, include_forgotten: bool = False, ) -> tuple[MemoryHandle, ...]: """Return deterministic memory handles matching configured filters.""" handles: list[MemoryHandle] = [] if record_id is not None: handle = _memory_handle(store, str(record_id)) if handle is None: return () if not include_forgotten and handle.record.get("status") == MemoryStatus.FORGOTTEN.value: return () return (handle,) for key in sorted(store.keys(), key=str): raw_record = store.get(key) if isinstance(raw_record, MutableMapping): record = raw_record elif isinstance(raw_record, Mapping): record = dict(raw_record) store[key] = record else: continue if not include_forgotten and record.get("status") == MemoryStatus.FORGOTTEN.value: continue if category is not None and str(record.get("category", "")) != str(category): continue if not _tags_match(record, tags): continue if importance_below is not None and _record_importance(record) >= float(importance_below): continue if max_age_steps is not None and step is not None: age = step - _record_created_step(record) if age < int(max_age_steps): continue handles.append(MemoryHandle(key=str(key), record=record, store=store)) handles.sort( key=lambda handle: ( _record_importance(handle.record), _record_strength(handle.record), _record_created_step(handle.record), handle.key, ) ) return tuple(handles) def _prune_store( store: MutableMapping[str, Any], *, max_records: int, protected_keys: Sequence[str] = (), ) -> tuple[str, ...]: """Prune low-value records if the store exceeds a configured limit.""" if max_records <= 0 or len(store) <= max_records: return () protected = {str(key) for key in protected_keys} candidates: list[tuple[float, float, int, str]] = [] for key, raw_record in store.items(): if str(key) in protected: continue if not isinstance(raw_record, Mapping): candidates.append((0.0, 0.0, 0, str(key))) continue candidates.append( ( _record_importance(raw_record), _record_strength(raw_record), _record_created_step(raw_record), str(key), ) ) removed: list[str] = [] for _, _, _, key in sorted(candidates): if len(store) <= max_records: break store.pop(key, None) removed.append(key) return tuple(removed) def _has_costs(agent: Agent, costs: Mapping[str, Any]) -> bool: """Return whether an agent can pay all configured numeric costs.""" for path, raw_amount in costs.items(): if not _is_number(raw_amount): continue amount = max(0.0, float(raw_amount)) if amount <= 0: continue available = _as_float(_read_agent_value(agent, str(path), 0.0), 0.0) if available + _EPSILON < amount: return False return True def _consume_costs(agent: Agent, costs: Mapping[str, Any]) -> dict[str, float]: """Consume configured numeric costs and return consumed amounts.""" consumed: dict[str, float] = {} for path, raw_amount in costs.items(): if not _is_number(raw_amount): continue amount = max(0.0, float(raw_amount)) if amount <= 0: continue _increment_agent_value(agent, str(path), -amount) consumed[str(path)] = amount return consumed def _success( behavior: str, agent: Agent, *, memory_ids: Sequence[str] = (), belief_ids: Sequence[str] = (), record_ids: Sequence[str] = (), details: Mapping[str, Any] | None = None, world: World | None = None, ) -> dict[str, Any]: """Build a successful behavior outcome dictionary.""" return MemoryOutcome( behavior=behavior, actor_id=_agent_id(agent), success=True, memory_ids=tuple(str(memory_id) for memory_id in memory_ids), belief_ids=tuple(str(belief_id) for belief_id in belief_ids), record_ids=tuple(str(record_id) for record_id in record_ids), step=_world_step(world) if world is not None else None, details=details or {}, ).to_dict() def _failure( behavior: str, agent: Agent, reason: str, *, details: Mapping[str, Any] | None = None, world: World | None = None, ) -> dict[str, Any]: """Build a failed behavior outcome dictionary.""" payload: dict[str, Any] = {"reason": reason} if details: payload.update(details) logger.debug("Behavior %s failed for agent %s: %s", behavior, _agent_id(agent), reason) return MemoryOutcome( behavior=behavior, actor_id=_agent_id(agent), success=False, step=_world_step(world) if world is not None else None, details=payload, ).to_dict() @dataclass class RememberBehavior(Behavior): """Store an explicit or observed memory record on an agent. The memory payload can come from explicit content, selected agent paths, state snapshots, memory snapshots, or any combination of those sources. """ name: ClassVar[str] = "remember" record_id: str | None = None category: str = "general" content: Any = None source_paths: tuple[str, ...] = () snapshot_state_keys: tuple[str, ...] = () snapshot_memory_keys: tuple[str, ...] = () memory_store_key: str = "memories" direct_write_path: str | None = None importance: float = 1.0 confidence: float = 1.0 strength: float = 1.0 tags: tuple[str, ...] = () metadata: Mapping[str, Any] = field(default_factory=dict) merge_existing: bool = True overwrite_content: bool = True max_records: int = 1000 requirements: Mapping[str, Any] = field(default_factory=dict) remember_costs: Mapping[str, Any] = field(default_factory=dict) history_memory_key: str = "memory_history" max_history: int = 500 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether the agent can remember this information.""" if not _is_alive(agent): return False if self.requirements and not _matches_constraints(agent, self.requirements): return False if self.remember_costs and not _has_costs(agent, self.remember_costs): return False return ( self.content is not None or bool(self.source_paths) or bool(self.snapshot_state_keys) or bool(self.snapshot_memory_keys) or self.direct_write_path is not None ) def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Create or update a memory record.""" if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) step = _world_step(world) store = _record_store( agent, location=StorageLocation.MEMORY, store_key=self.memory_store_key, ) record_id = self.record_id or _generated_record_id( agent, behavior=self.name, category=self.category, sequence_count=len(store), step=step, ) payload = self._payload(agent) consumed_costs = _consume_costs(agent, self.remember_costs) previous_record = copy.deepcopy(store.get(record_id)) if isinstance(store.get(record_id), Mapping) else None existing = store.get(record_id) if isinstance(existing, MutableMapping) and self.merge_existing: record = existing version = _as_int(record.get("version"), 0) + 1 if self.overwrite_content or "content" not in record: record["content"] = payload else: version = 1 record = { "id": record_id, "category": self.category, "content": payload, "created_step": step, } store[record_id] = record record.update( { "id": record_id, "category": self.category, "importance": float(self.importance), "confidence": _clamp_unit(float(self.confidence)), "strength": max(0.0, float(self.strength)), "tags": tuple(str(tag) for tag in self.tags), "metadata": copy.deepcopy(dict(self.metadata)), "status": MemoryStatus.ACTIVE.value, "active": True, "version": version, "updated_step": step, } ) if self.direct_write_path is not None: _write_agent_value( agent, self.direct_write_path, copy.deepcopy(payload), default_location=StorageLocation.MEMORY, ) pruned_ids = _prune_store( store, max_records=self.max_records, protected_keys=(record_id,), ) history_record = { "behavior": self.name, "record_id": record_id, "category": self.category, "version": version, "costs": consumed_costs, "pruned_ids": pruned_ids, "step": step, } _record_history( agent, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) logger.debug( "Agent %s remembered record %s in category %s", _agent_id(agent), record_id, self.category, ) return _success( self.name, agent, memory_ids=(record_id,), details={ "record_id": record_id, "category": self.category, "record": copy.deepcopy(record), "previous_record": previous_record, "direct_write_path": self.direct_write_path, "costs": consumed_costs, "pruned_ids": pruned_ids, }, world=world, ) def _payload(self, agent: Agent) -> Any: """Build the memory payload from configured sources.""" observations = { path: copy.deepcopy(_read_agent_value(agent, path, None)) for path in self.source_paths } state_snapshot = { key: copy.deepcopy(_get_path(_agent_state(agent), key, None)) for key in self.snapshot_state_keys } memory_snapshot = { key: copy.deepcopy(_get_path(_agent_memory(agent), key, None)) for key in self.snapshot_memory_keys } if not observations and not state_snapshot and not memory_snapshot: return copy.deepcopy(self.content) payload: dict[str, Any] = {} if self.content is not None: payload["content"] = copy.deepcopy(self.content) if observations: payload["observations"] = observations if state_snapshot: payload["state_snapshot"] = state_snapshot if memory_snapshot: payload["memory_snapshot"] = memory_snapshot return payload @dataclass class ForgetBehavior(Behavior): """Forget memory records or delete a generic state or memory path. Forgetting can either mark memory records as forgotten for auditability or physically remove them from the memory store. """ name: ClassVar[str] = "forget" record_id: str | None = None category: str | None = None tags: tuple[str, ...] = () target_path: str | None = None target_path_default_location: StorageLocation | str = StorageLocation.MEMORY memory_store_key: str = "memories" remove_record: bool = False reason: str = "unspecified" importance_below: float | None = None max_age_steps: int | None = None include_already_forgotten: bool = False max_records_to_forget: int = 1 requirements: Mapping[str, Any] = field(default_factory=dict) forget_costs: Mapping[str, Any] = field(default_factory=dict) history_memory_key: str = "memory_history" max_history: int = 500 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether the agent can forget the configured target.""" if not _is_alive(agent): return False if self.requirements and not _matches_constraints(agent, self.requirements): return False if self.forget_costs and not _has_costs(agent, self.forget_costs): return False if self.target_path is not None: if _read_agent_value(agent, self.target_path, _MISSING) is not _MISSING: return True if not self._has_record_criteria(): return False store = _record_store( agent, location=StorageLocation.MEMORY, store_key=self.memory_store_key, ) return bool(self._matching_handles(store, world)) def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Forget selected records and optionally delete a direct path.""" if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) step = _world_step(world) consumed_costs = _consume_costs(agent, self.forget_costs) deleted_path = False if self.target_path is not None: deleted_path = _delete_agent_value( agent, self.target_path, default_location=self.target_path_default_location, ) store = _record_store( agent, location=StorageLocation.MEMORY, store_key=self.memory_store_key, ) handles = self._matching_handles(store, world) if self.max_records_to_forget > 0: handles = handles[: int(self.max_records_to_forget)] forgotten_ids: list[str] = [] previous_records: dict[str, Any] = {} for handle in handles: previous_records[handle.key] = copy.deepcopy(dict(handle.record)) forgotten_ids.append(handle.key) if self.remove_record: handle.store.pop(handle.key, None) continue handle.record["status"] = MemoryStatus.FORGOTTEN.value handle.record["active"] = False handle.record["forgotten_step"] = step handle.record["forget_reason"] = self.reason handle.record["updated_step"] = step if not forgotten_ids and not deleted_path: return _failure(self.name, agent, "nothing_forgotten", world=world) history_record = { "behavior": self.name, "forgotten_ids": forgotten_ids, "removed": self.remove_record, "deleted_path": self.target_path if deleted_path else None, "reason": self.reason, "costs": consumed_costs, "step": step, } _record_history( agent, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) logger.debug( "Agent %s forgot %s record(s)", _agent_id(agent), len(forgotten_ids), ) return _success( self.name, agent, memory_ids=tuple(forgotten_ids), details={ "forgotten_ids": forgotten_ids, "removed": self.remove_record, "deleted_path": self.target_path if deleted_path else None, "previous_records": previous_records, "reason": self.reason, "costs": consumed_costs, }, world=world, ) def _has_record_criteria(self) -> bool: """Return whether at least one memory-record criterion is configured.""" return any( ( self.record_id is not None, self.category is not None, bool(self.tags), self.importance_below is not None, self.max_age_steps is not None, ) ) def _matching_handles( self, store: MutableMapping[str, Any], world: World, ) -> tuple[MemoryHandle, ...]: """Return memory handles selected for forgetting.""" return _matching_memory_handles( store, record_id=self.record_id, category=self.category, tags=self.tags, importance_below=self.importance_below, max_age_steps=self.max_age_steps, step=_world_step(world), include_forgotten=self.include_already_forgotten, ) @dataclass class ReinforceBehavior(Behavior): """Reinforce or weaken selected memories and optional numeric paths. Positive signals increase strength, importance, and confidence. Negative signals decrease them. The behavior is deterministic and does not decide what is true; it only updates configured numeric traces. """ name: ClassVar[str] = "reinforce" record_id: str | None = None category: str | None = None tags: tuple[str, ...] = () memory_store_key: str = "memories" signal: float = 1.0 signal_path: str | None = None learning_rate: float = 1.0 strength_key: str = "strength" importance_key: str = "importance" confidence_key: str = "confidence" strength_delta_scale: float = 1.0 importance_delta_scale: float = 0.25 confidence_delta_scale: float = 0.1 min_strength: float = 0.0 max_strength: float = 1000.0 min_importance: float = 0.0 max_importance: float = 1000.0 target_path: str | None = None target_path_default_location: StorageLocation | str = StorageLocation.MEMORY target_delta_scale: float = 1.0 create_if_missing: bool = False created_category: str = "general" max_records_to_reinforce: int = 1 requirements: Mapping[str, Any] = field(default_factory=dict) reinforce_costs: Mapping[str, Any] = field(default_factory=dict) history_memory_key: str = "memory_history" max_history: int = 500 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether reinforcement can be applied.""" if not _is_alive(agent): return False if self.requirements and not _matches_constraints(agent, self.requirements): return False if self.reinforce_costs and not _has_costs(agent, self.reinforce_costs): return False if not _is_number(self._effective_signal(agent)): return False if self.target_path is not None: return True store = _record_store( agent, location=StorageLocation.MEMORY, store_key=self.memory_store_key, ) if self._matching_handles(store): return True return self.create_if_missing def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Apply reinforcement to selected records and optional target path.""" if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) step = _world_step(world) signal = float(self._effective_signal(agent)) scaled_signal = signal * float(self.learning_rate) consumed_costs = _consume_costs(agent, self.reinforce_costs) store = _record_store( agent, location=StorageLocation.MEMORY, store_key=self.memory_store_key, ) handles = self._matching_handles(store) if not handles and self.create_if_missing: created_id = self.record_id or _generated_record_id( agent, behavior=self.name, category=self.created_category, sequence_count=len(store), step=step, ) store[created_id] = { "id": created_id, "category": self.category or self.created_category, "content": None, "importance": 0.0, "confidence": 0.0, "strength": 0.0, "tags": tuple(str(tag) for tag in self.tags), "status": MemoryStatus.ACTIVE.value, "active": True, "created_step": step, } handles = tuple(handle for handle in (_memory_handle(store, created_id),) if handle is not None) if self.max_records_to_reinforce > 0: handles = handles[: int(self.max_records_to_reinforce)] updates: list[dict[str, Any]] = [] for handle in handles: previous_strength = _as_float(handle.record.get(self.strength_key), 0.0) previous_importance = _as_float(handle.record.get(self.importance_key), 0.0) previous_confidence = _as_float(handle.record.get(self.confidence_key), 0.0) new_strength = _clamp( previous_strength + scaled_signal * float(self.strength_delta_scale), self.min_strength, self.max_strength, ) new_importance = _clamp( previous_importance + scaled_signal * float(self.importance_delta_scale), self.min_importance, self.max_importance, ) new_confidence = _clamp_unit( previous_confidence + scaled_signal * float(self.confidence_delta_scale) ) handle.record[self.strength_key] = new_strength handle.record[self.importance_key] = new_importance handle.record[self.confidence_key] = new_confidence handle.record["status"] = ( MemoryStatus.REINFORCED.value if scaled_signal >= 0 else MemoryStatus.WEAKENED.value ) handle.record["active"] = True handle.record["reinforcement_count"] = _as_int(handle.record.get("reinforcement_count"), 0) + 1 handle.record["last_reinforcement_signal"] = signal handle.record["updated_step"] = step updates.append( { "record_id": handle.key, "strength_before": previous_strength, "strength_after": new_strength, "importance_before": previous_importance, "importance_after": new_importance, "confidence_before": previous_confidence, "confidence_after": new_confidence, } ) target_path_update: dict[str, Any] | None = None if self.target_path is not None: previous_value = _as_float(_read_agent_value(agent, self.target_path, 0.0), 0.0) new_value = _increment_agent_value( agent, self.target_path, scaled_signal * float(self.target_delta_scale), default_location=self.target_path_default_location, ) target_path_update = { "path": self.target_path, "value_before": previous_value, "value_after": new_value, } if not updates and target_path_update is None: return _failure(self.name, agent, "nothing_reinforced", world=world) history_record = { "behavior": self.name, "signal": signal, "scaled_signal": scaled_signal, "updates": updates, "target_path_update": target_path_update, "costs": consumed_costs, "step": step, } _record_history( agent, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) logger.debug( "Agent %s reinforced %s memory record(s)", _agent_id(agent), len(updates), ) return _success( self.name, agent, memory_ids=tuple(update["record_id"] for update in updates), details={ "signal": signal, "scaled_signal": scaled_signal, "updates": updates, "target_path_update": target_path_update, "costs": consumed_costs, }, world=world, ) def _effective_signal(self, agent: Agent) -> float: """Return the configured or path-derived reinforcement signal.""" if self.signal_path is None: return float(self.signal) return _as_float(_read_agent_value(agent, self.signal_path, self.signal), float(self.signal)) def _matching_handles(self, store: MutableMapping[str, Any]) -> tuple[MemoryHandle, ...]: """Return memory handles selected for reinforcement.""" return _matching_memory_handles( store, record_id=self.record_id, category=self.category, tags=self.tags, include_forgotten=False, ) @dataclass class UpdateBeliefBehavior(Behavior): """Update a generic numeric belief from deterministic evidence. Belief values are clamped to [0, 1]. The behavior supports a simple weighted-average update and a Bayesian-style update. Both are deterministic and operate only on scalar values. """ name: ClassVar[str] = "update_belief" belief_id: str = "" proposition: str | None = None evidence: float | bool | None = None evidence_path: str | None = None evidence_weight: float = 1.0 update_rule: BeliefUpdateRule | str = BeliefUpdateRule.WEIGHTED_AVERAGE learning_rate: float = 0.5 prior: float = 0.5 initial_confidence: float = 0.5 confidence_delta: float = 0.05 confidence_decay: float = 0.0 likelihood_if_true: float | None = None likelihood_if_false: float | None = None beliefs_store_key: str = "beliefs" evidence_history_key: str = "evidence_history" write_belief_path: str | None = None metadata: Mapping[str, Any] = field(default_factory=dict) requirements: Mapping[str, Any] = field(default_factory=dict) update_costs: Mapping[str, Any] = field(default_factory=dict) max_evidence_history: int = 200 history_memory_key: str = "memory_history" max_history: int = 500 def check_preconditions(self, agent: Agent, world: World) -> bool: """Return whether the belief can be updated.""" if not _is_alive(agent) or not self.belief_id: return False if self.requirements and not _matches_constraints(agent, self.requirements): return False if self.update_costs and not _has_costs(agent, self.update_costs): return False return self._evidence_value(agent) is not None def execute(self, agent: Agent, world: World) -> dict[str, Any]: """Update the configured belief from evidence.""" if not self.check_preconditions(agent, world): return _failure(self.name, agent, "preconditions_not_met", world=world) evidence_value = self._evidence_value(agent) if evidence_value is None: return _failure(self.name, agent, "evidence_not_available", world=world) step = _world_step(world) consumed_costs = _consume_costs(agent, self.update_costs) store = _record_store( agent, location=StorageLocation.MEMORY, store_key=self.beliefs_store_key, ) existing = store.get(self.belief_id) if isinstance(existing, MutableMapping): belief = existing elif isinstance(existing, Mapping): belief = dict(existing) store[self.belief_id] = belief else: belief = { "id": self.belief_id, "proposition": self.proposition or self.belief_id, "value": _clamp_unit(float(self.prior)), "confidence": _clamp_unit(float(self.initial_confidence)), "created_step": step, } store[self.belief_id] = belief previous_value = _clamp_unit(_as_float(belief.get("value"), self.prior)) previous_confidence = _clamp_unit(_as_float(belief.get("confidence"), self.initial_confidence)) new_value = self._updated_belief_value(previous_value, evidence_value) confidence_after_decay = previous_confidence * (1.0 - _clamp_unit(float(self.confidence_decay))) new_confidence = _clamp_unit(confidence_after_decay + float(self.confidence_delta) * abs(evidence_value - previous_value)) evidence_record = { "value": evidence_value, "weight": max(0.0, float(self.evidence_weight)), "source_path": self.evidence_path, "previous_belief": previous_value, "updated_belief": new_value, "step": step, } evidence_history = _ensure_list(belief, self.evidence_history_key) _append_bounded(evidence_history, copy.deepcopy(evidence_record), self.max_evidence_history) belief.update( { "id": self.belief_id, "proposition": self.proposition or belief.get("proposition") or self.belief_id, "value": new_value, "confidence": new_confidence, "evidence_count": _as_int(belief.get("evidence_count"), 0) + 1, "last_evidence": evidence_value, "update_rule": _normalize_belief_rule(self.update_rule).value, "metadata": copy.deepcopy(dict(self.metadata)), "updated_step": step, } ) if self.write_belief_path is not None: _write_agent_value( agent, self.write_belief_path, copy.deepcopy(belief), default_location=StorageLocation.MEMORY, ) history = _ensure_list(_agent_memory(agent), self.history_memory_key) record_id = _generated_record_id( agent, behavior=self.name, category="belief", sequence_count=len(history), step=step, ) history_record = { "id": record_id, "behavior": self.name, "belief_id": self.belief_id, "previous_value": previous_value, "new_value": new_value, "previous_confidence": previous_confidence, "new_confidence": new_confidence, "evidence": evidence_value, "costs": consumed_costs, "step": step, } _record_history( agent, history_key=self.history_memory_key, record=history_record, max_history=self.max_history, ) logger.debug( "Agent %s updated belief %s from %.3f to %.3f", _agent_id(agent), self.belief_id, previous_value, new_value, ) return _success( self.name, agent, belief_ids=(self.belief_id,), record_ids=(record_id,), details={ "belief_id": self.belief_id, "belief": copy.deepcopy(belief), "previous_value": previous_value, "new_value": new_value, "previous_confidence": previous_confidence, "new_confidence": new_confidence, "evidence": evidence_value, "costs": consumed_costs, "write_belief_path": self.write_belief_path, }, world=world, ) def _evidence_value(self, agent: Agent) -> float | None: """Return the evidence value as a clamped numeric scalar.""" raw_value: Any if self.evidence_path is not None: raw_value = _read_agent_value(agent, self.evidence_path, _MISSING) if raw_value is _MISSING: return None else: raw_value = self.evidence if isinstance(raw_value, bool): return 1.0 if raw_value else 0.0 if isinstance(raw_value, Mapping): nested_value = raw_value.get("value", raw_value.get("evidence", _MISSING)) if nested_value is _MISSING: return None raw_value = nested_value if not _is_number(raw_value): return None return _clamp_unit(float(raw_value)) def _updated_belief_value(self, previous_value: float, evidence_value: float) -> float: """Return the updated belief value using the configured rule.""" rule = _normalize_belief_rule(self.update_rule) learning_rate = _clamp_unit(float(self.learning_rate)) evidence_weight = max(0.0, float(self.evidence_weight)) effective_rate = _clamp_unit(learning_rate * evidence_weight) if rule is BeliefUpdateRule.BAYESIAN: likelihood_true = ( _clamp_unit(float(self.likelihood_if_true)) if self.likelihood_if_true is not None else evidence_value ) likelihood_false = ( _clamp_unit(float(self.likelihood_if_false)) if self.likelihood_if_false is not None else 1.0 - evidence_value ) numerator = previous_value * likelihood_true denominator = numerator + (1.0 - previous_value) * likelihood_false if denominator <= _EPSILON: posterior = previous_value else: posterior = numerator / denominator return _clamp_unit(previous_value + effective_rate * (posterior - previous_value)) return _clamp_unit(previous_value + effective_rate * (evidence_value - previous_value)) Remember = RememberBehavior Forget = ForgetBehavior Reinforce = ReinforceBehavior UpdateBelief = UpdateBeliefBehavior BEHAVIOR_REGISTRY: Mapping[str, type[Behavior]] = MappingProxyType( { RememberBehavior.name: RememberBehavior, ForgetBehavior.name: ForgetBehavior, ReinforceBehavior.name: ReinforceBehavior, UpdateBeliefBehavior.name: UpdateBeliefBehavior, } ) __all__ = [ "BEHAVIOR_REGISTRY", "BeliefUpdateRule", "Forget", "ForgetBehavior", "MemoryHandle", "MemoryOutcome", "MemoryStatus", "Reinforce", "ReinforceBehavior", "Remember", "RememberBehavior", "StorageLocation", "UpdateBelief", "UpdateBeliefBehavior", ]