babylon.ai.director

Narrative Director - AI Game Master observing the simulation.

The NarrativeDirector implements the SimulationObserver protocol to generate narrative from state changes. It sits in the Ideological Superstructure layer and cannot modify simulation state.

Design Philosophy: - Observer, not controller: watches state transitions - Narrative from material: derives story from state changes - Fail-safe: errors don’t propagate to simulation (ADR003)

Sprint 3.2: Added RAG integration for historical/theoretical context. The Materialist Retrieval bridges Engine with the Archive (ChromaDB).

Sprint 4.1: Updated to consume typed SimulationEvent objects from state.events instead of string-based event_log.

Sprint 4.2: Added Persona support for customizable narrative voices.

Classes

NarrativeDirector([use_llm, rag_pipeline, ...])

AI Game Master that observes simulation and generates narrative.
class babylon.ai.director.NarrativeDirector(use_llm=False, rag_pipeline=None, prompt_builder=None, llm=None, persona=None)[source]

Bases: object

AI Game Master that observes simulation and generates narrative.

The Director watches state transitions and produces human-readable narrative describing the class struggle dynamics.

Sprint 3.2: Added RAG integration for “The Materialist Retrieval”. The Director can now query the Archive (ChromaDB) for historical and theoretical context to inform narrative generation.

Sprint 3.4: Added Semantic Bridge to translate simulation event keywords into theoretical query strings for better RAG retrieval.

Parameters:
name

Observer identifier (“NarrativeDirector”).

use_llm

Whether to use LLM for narrative (False = template-based).

rag_pipeline

Optional RAG pipeline for context retrieval.

SEMANTIC_MAP

Class constant mapping event keywords to theory queries.

Example

>>> from babylon.ai import NarrativeDirector
>>> from babylon.engine import Simulation
>>> from babylon.rag import RagPipeline
>>>
>>> # With RAG integration
>>> rag = RagPipeline()
>>> director = NarrativeDirector(rag_pipeline=rag)
>>> sim = Simulation(initial_state, config, observers=[director])
>>> sim.run(10)  # Director queries RAG for context
>>> sim.end()
SEMANTIC_MAP: dict[EventType, str] = {EventType.CONSCIOUSNESS_TRANSMISSION: 'development of class consciousness and proletariat solidarity', EventType.ECONOMIC_CRISIS: 'tendency of the rate of profit to fall and capitalist crisis', EventType.EXCESSIVE_FORCE: 'state violence police brutality and repression', EventType.IMPERIAL_SUBSIDY: 'role of repression in maintaining imperialist client states', EventType.MASS_AWAKENING: 'leninist theory of revolutionary situation and mass strike', EventType.PHASE_TRANSITION: 'phase transition revolutionary organization vanguard party', EventType.RUPTURE: 'dialectical contradiction rupture revolutionary crisis', EventType.SOLIDARITY_SPIKE: 'solidarity networks mutual aid class organization', EventType.SURPLUS_EXTRACTION: 'marxist theory of surplus value extraction and exploitation', EventType.UPRISING: 'mass uprising revolutionary insurrection george floyd protests'}
FALLBACK_QUERY: str = 'dialectical materialism class struggle'
SIGNIFICANT_EVENT_TYPES: frozenset[EventType] = frozenset({EventType.ECONOMIC_CRISIS, EventType.EXCESSIVE_FORCE, EventType.MASS_AWAKENING, EventType.PHASE_TRANSITION, EventType.RUPTURE, EventType.SURPLUS_EXTRACTION, EventType.UPRISING})
__init__(use_llm=False, rag_pipeline=None, prompt_builder=None, llm=None, persona=None)[source]

Initialize the NarrativeDirector.

Parameters:

  • use_llm (bool) – If True, use LLM for narrative generation. If False, use template-based generation (default).

  • rag_pipeline (RagPipeline | None) – Optional RagPipeline for context retrieval. If None, RAG features are disabled (backward compat).

  • prompt_builder (DialecticalPromptBuilder | None) – Optional custom DialecticalPromptBuilder. If None, creates default builder.

  • llm (LLMProvider | None) – Optional LLMProvider for text generation. If None, no LLM generation occurs (backward compat).

  • persona (Persona | None) – Optional Persona for customizing narrative voice. If provided (and no custom prompt_builder), creates a DialecticalPromptBuilder with this persona.

Return type:

None

property name: str

Return observer identifier.

Returns:

The string “NarrativeDirector”.

property use_llm: bool

Return whether LLM is enabled.

Returns:

True if LLM-based narrative is enabled, False otherwise.

property rag_pipeline: RagPipeline | None

Return the RAG pipeline if configured.

Returns:

RagPipeline instance or None if not configured.

property narrative_log: list[str]

Return generated narrative entries.

Returns a copy of the internal list to prevent external modification.

Returns:

List of generated narrative strings.

on_simulation_start(initial_state, config)[source]

Initialize narrative context at simulation start.

Logs the simulation start event with initial state info.

Parameters:
Return type:

None

on_tick(previous_state, new_state)[source]

Analyze state change and log narrative.

Detects new typed events added during this tick, retrieves RAG context, and builds the full context hierarchy for narrative generation.

Sprint 4.1: Now processes typed SimulationEvent objects from state.events instead of string-based event_log.

Parameters:

  • previous_state (WorldState) – WorldState before the tick.

  • new_state (WorldState) – WorldState after the tick.

Return type:

None

on_simulation_end(final_state)[source]

Generate summary at simulation end.

Logs the simulation end event with final state info.

Parameters:

final_state (WorldState) – The final WorldState when simulation ends.

Return type:

None