""" Narrative summarization for WorldSmithAI. This module converts runtime world state, simulation history, and metric outputs into human-readable narrative summaries. It is designed for root-level Hugging Face Spaces ``app.py`` usage and does not assume an ``app/`` package. The narrator is domain-agnostic. It does not know about sheep, wolves, scientists, dragons, farms, cities, transport hubs, power grids, startups, or fantasy worlds. It reads generic fields such as ``type``, ``state``, ``memory``, ``amount``, ``alive``, ``agents``, ``resources``, ``events``, and metric result objects. Two modes are supported: - deterministic: pure Python narrative generation. - model / auto: optional model-assisted polishing from a structured context. The model, when used, receives summaries only and never mutates world state. Example: from llm.narrator import narrate_world summary = narrate_world(world) Root-level Gradio app.py example: def run_simulation(prompt: str): world, history, metrics = run_world(prompt) return narrate_simulation(history, metric_results=metrics) Future extensibility: - Add event-log-aware storytelling. - Add character-level agent narratives. - Add causality and counterfactual explanations. - Add God-Agent critique using metric context. - Add multilingual summaries. - Add model-specific prompt profiles. """ from __future__ import annotations import copy import inspect import json import logging import math from collections.abc import Callable, Iterable, Mapping, MutableSequence, Sequence from dataclasses import dataclass, field from enum import Enum from numbers import Real from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable if TYPE_CHECKING: from core.world import World logger = logging.getLogger(__name__) _MISSING = object() _EPSILON = 1.0e-12 DEFAULT_NARRATOR_SYSTEM_PROMPT = """ You are WorldSmithAI's narrative summarizer. Your job is to explain an agent-based world simulation using the structured context provided by the Python engine. Rules: - Do not invent facts not present in the context. - Do not claim that the world contains specific domain entities unless the context names them. - Do not describe Python code. - Do not mutate, propose mutations, or output DSL. - Write clear, useful narrative text for a demo viewer. - Keep the tone consistent with the requested style and audience. """.strip() class NarrationMode(str, Enum): """Narration execution modes.""" DETERMINISTIC = "deterministic" MODEL = "model" AUTO = "auto" class NarrationStyle(str, Enum): """Narrative styles.""" CONCISE = "concise" ANALYTICAL = "analytical" STORY = "story" EXECUTIVE = "executive" DEBUG = "debug" class NarrationAudience(str, Enum): """Target audience for the narrative.""" GENERAL = "general" TECHNICAL = "technical" HACKATHON_JUDGE = "hackathon_judge" class NarrativeSection(str, Enum): """Standard narrative section names.""" OVERVIEW = "overview" AGENTS = "agents" RESOURCES = "resources" METRICS = "metrics" DYNAMICS = "dynamics" ACTIVITY = "activity" TAKEAWAY = "takeaway" @dataclass(frozen=True) class NarrationPrompt: """Prompt payload for optional model-assisted narration.""" system_prompt: str user_prompt: str messages: tuple[Mapping[str, str], ...] context: Mapping[str, Any] = field(default_factory=dict) @property def text(self) -> str: """Return a flattened prompt for completion-style clients.""" return f"{self.system_prompt}\n\n{self.user_prompt}" def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly prompt representation.""" return { "system_prompt": self.system_prompt, "user_prompt": self.user_prompt, "messages": [dict(message) for message in self.messages], "context": _json_safe(self.context), } @dataclass(frozen=True) class NarrativeContext: """Structured world context used for deterministic or model narration.""" world_id: str | None world_name: str | None world_description: str | None step: int | None agent_count: int alive_agent_count: int inactive_agent_count: int agent_type_counts: Mapping[str, int] = field(default_factory=dict) dominant_agent_type: str | None = None resource_count: int = 0 resource_type_counts: Mapping[str, int] = field(default_factory=dict) resource_type_amounts: Mapping[str, float] = field(default_factory=dict) dominant_resource_type: str | None = None event_count: int = 0 pending_event_count: int = 0 metric_summaries: Mapping[str, Any] = field(default_factory=dict) activity_summary: Mapping[str, Any] = field(default_factory=dict) timeline_summary: Mapping[str, Any] = field(default_factory=dict) notable_agents: tuple[Mapping[str, Any], ...] = () notable_resources: tuple[Mapping[str, Any], ...] = () metadata: Mapping[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly context dictionary.""" return { "world_id": self.world_id, "world_name": self.world_name, "world_description": self.world_description, "step": self.step, "agent_count": self.agent_count, "alive_agent_count": self.alive_agent_count, "inactive_agent_count": self.inactive_agent_count, "agent_type_counts": copy.deepcopy(dict(self.agent_type_counts)), "dominant_agent_type": self.dominant_agent_type, "resource_count": self.resource_count, "resource_type_counts": copy.deepcopy(dict(self.resource_type_counts)), "resource_type_amounts": copy.deepcopy(dict(self.resource_type_amounts)), "dominant_resource_type": self.dominant_resource_type, "event_count": self.event_count, "pending_event_count": self.pending_event_count, "metric_summaries": _json_safe(copy.deepcopy(dict(self.metric_summaries))), "activity_summary": _json_safe(copy.deepcopy(dict(self.activity_summary))), "timeline_summary": _json_safe(copy.deepcopy(dict(self.timeline_summary))), "notable_agents": [_json_safe(dict(agent)) for agent in self.notable_agents], "notable_resources": [ _json_safe(dict(resource)) for resource in self.notable_resources ], "metadata": _json_safe(copy.deepcopy(dict(self.metadata))), } @dataclass(frozen=True) class NarrativeResult: """Narrative output produced by ``WorldNarrator``.""" text: str context: NarrativeContext sections: Mapping[str, str] = field(default_factory=dict) bullets: tuple[str, ...] = () mode: NarrationMode = NarrationMode.DETERMINISTIC raw_model_response: str | None = None prompt: NarrationPrompt | None = None metadata: Mapping[str, Any] = field(default_factory=dict) @property def summary(self) -> str: """Alias for narrative text.""" return self.text def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly result representation.""" return { "text": self.text, "sections": copy.deepcopy(dict(self.sections)), "bullets": list(self.bullets), "mode": self.mode.value, "raw_model_response": self.raw_model_response, "context": self.context.to_dict(), "prompt": None if self.prompt is None else self.prompt.to_dict(), "metadata": _json_safe(copy.deepcopy(dict(self.metadata))), } @dataclass class NarratorConfig: """Configuration for ``WorldNarrator``. Defaults are safe for root-level Gradio apps: - pure Python deterministic narration works without a model, - model errors fall back to deterministic text, - summaries stay compact enough for UI display. """ mode: NarrationMode | str = NarrationMode.DETERMINISTIC style: NarrationStyle | str = NarrationStyle.ANALYTICAL audience: NarrationAudience | str = NarrationAudience.HACKATHON_JUDGE system_prompt: str = DEFAULT_NARRATOR_SYSTEM_PROMPT model_kwargs: Mapping[str, Any] = field(default_factory=dict) fallback_on_model_error: bool = True max_notable_agents: int = 6 max_notable_resources: int = 6 max_metric_entries: int = 12 max_activity_paths: int = 12 include_metrics: bool = True compute_default_metrics: bool = True include_activity: bool = True include_timeline: bool = True include_debug_context: bool = False max_text_chars: int | None = None activity_paths: tuple[str, ...] = ( "memory.policy_decisions", "memory.bandit_decisions", "memory.market_trades", "memory.construction_history", "memory.adoption_history", "memory.planning_history", "memory.memory_history", "memory.inbox", "memory.outbox", "memory.goal_history", "memory.tax_receipts", "memory.governance_history", ) metadata: Mapping[str, Any] = field(default_factory=dict) def resolved_mode(self) -> NarrationMode: """Return normalized narration mode.""" if isinstance(self.mode, NarrationMode): return self.mode return NarrationMode(str(self.mode)) def resolved_style(self) -> NarrationStyle: """Return normalized narration style.""" if isinstance(self.style, NarrationStyle): return self.style return NarrationStyle(str(self.style)) def resolved_audience(self) -> NarrationAudience: """Return normalized narration audience.""" if isinstance(self.audience, NarrationAudience): return self.audience return NarrationAudience(str(self.audience)) @runtime_checkable class SupportsNarrationClient(Protocol): """Protocol for optional model clients.""" def generate(self, *args: Any, **kwargs: Any) -> Any: """Generate text from a prompt-like payload.""" class NarrationError(RuntimeError): """Base error raised by narration utilities.""" class NarrationClientError(NarrationError): """Raised when an optional model client fails.""" @dataclass class WorldNarrator: """Narrate WorldSmithAI worlds and simulation histories. The class can operate with no model client. When a client is supplied and mode is ``model`` or ``auto``, it asks the model to polish a structured context into a narrative. If the model fails and fallback is enabled, the deterministic narrative is returned. """ client: Any | None = None config: NarratorConfig = field(default_factory=NarratorConfig) def narrate( self, world: World, *, history: Sequence[Any] | None = None, metric_results: Mapping[str, Any] | Sequence[Any] | None = None, extra_context: Mapping[str, Any] | None = None, ) -> NarrativeResult: """Narrate a current world state. Args: world: Runtime world object. history: Optional sequence of world snapshots, chart snapshots, or mappings used to summarize dynamics. metric_results: Optional metric result objects or mappings. extra_context: Optional UI or app-level context. Returns: ``NarrativeResult``. """ context = self.build_context( world, history=history, metric_results=metric_results, extra_context=extra_context, ) mode = self.config.resolved_mode() if mode is NarrationMode.DETERMINISTIC or self.client is None: return self._deterministic_result(context, mode=NarrationMode.DETERMINISTIC) prompt = self.build_prompt(context) try: raw_response = self._call_client(prompt) text = self._postprocess_text(raw_response) if not text: raise NarrationClientError("Narration client returned empty text") return NarrativeResult( text=text, context=context, sections={}, bullets=(), mode=NarrationMode.MODEL, raw_model_response=raw_response, prompt=prompt, metadata=copy.deepcopy(dict(self.config.metadata)), ) except Exception as exc: if mode is NarrationMode.AUTO and self.config.fallback_on_model_error: logger.warning("Narration client failed; using deterministic fallback: %s", exc) result = self._deterministic_result(context, mode=NarrationMode.DETERMINISTIC) return NarrativeResult( text=result.text, context=result.context, sections=result.sections, bullets=result.bullets, mode=NarrationMode.DETERMINISTIC, raw_model_response=None, prompt=prompt, metadata={ **copy.deepcopy(dict(result.metadata)), "fallback_reason": "model_error", "error": f"{exc.__class__.__name__}: {exc}", }, ) raise NarrationClientError(f"Narration client failed: {exc}") from exc def narrate_text( self, world: World, *, history: Sequence[Any] | None = None, metric_results: Mapping[str, Any] | Sequence[Any] | None = None, extra_context: Mapping[str, Any] | None = None, ) -> str: """Narrate a world and return only text.""" return self.narrate( world, history=history, metric_results=metric_results, extra_context=extra_context, ).text def build_context( self, world: World, *, history: Sequence[Any] | None = None, metric_results: Mapping[str, Any] | Sequence[Any] | None = None, extra_context: Mapping[str, Any] | None = None, ) -> NarrativeContext: """Build structured narrative context from world state and metrics.""" agents = _iter_collection(getattr(world, "agents", ())) resources = _iter_collection(getattr(world, "resources", ())) events = _iter_collection(getattr(world, "events", ())) alive_agents = tuple(agent for agent in agents if _is_alive(agent)) inactive_agents = tuple(agent for agent in agents if not _is_alive(agent)) agent_type_counts = _count_by_label(alive_agents, "type", include_missing=True) resource_type_counts = _count_by_label(resources, "type", include_missing=True) resource_type_amounts = _resource_amounts_by_type(resources) metric_summaries = {} if self.config.include_metrics: metric_summaries = self._metric_summaries(world, metric_results) activity_summary = {} if self.config.include_activity: activity_summary = self._activity_summary(world) timeline_summary = {} if self.config.include_timeline and history: timeline_summary = summarize_timeline(history) world_metadata = _mapping_or_empty(getattr(world, "metadata", {})) return NarrativeContext( world_id=_optional_str(getattr(world, "id", None)), world_name=_optional_str(getattr(world, "name", None)), world_description=_optional_str(getattr(world, "description", None)), step=_world_step(world), agent_count=len(agents), alive_agent_count=len(alive_agents), inactive_agent_count=len(inactive_agents), agent_type_counts=agent_type_counts, dominant_agent_type=_dominant_key(agent_type_counts), resource_count=len(resources), resource_type_counts=resource_type_counts, resource_type_amounts=resource_type_amounts, dominant_resource_type=_dominant_key(resource_type_amounts), event_count=len(events), pending_event_count=_pending_event_count(events, _world_step(world)), metric_summaries=metric_summaries, activity_summary=activity_summary, timeline_summary=timeline_summary, notable_agents=self._notable_agents(agents), notable_resources=self._notable_resources(resources), metadata={ **copy.deepcopy(world_metadata), **copy.deepcopy(dict(extra_context or {})), "narrator_style": self.config.resolved_style().value, "narrator_audience": self.config.resolved_audience().value, }, ) def build_prompt(self, context: NarrativeContext) -> NarrationPrompt: """Build optional model prompt from structured context.""" style = self.config.resolved_style().value audience = self.config.resolved_audience().value context_json = json.dumps(context.to_dict(), indent=2, sort_keys=True) user_prompt = ( f"Write a {style} WorldSmithAI simulation narrative for a " f"{audience} audience.\n\n" "Use only the structured context below. Do not invent events, " "agents, resources, or outcomes.\n\n" f"STRUCTURED CONTEXT:\n{context_json}\n\n" "Return only the narrative text." ) return NarrationPrompt( system_prompt=self.config.system_prompt, user_prompt=user_prompt, messages=( {"role": "system", "content": self.config.system_prompt}, {"role": "user", "content": user_prompt}, ), context=context.to_dict(), ) def _deterministic_result( self, context: NarrativeContext, *, mode: NarrationMode, ) -> NarrativeResult: """Build a deterministic narrative result.""" sections = self._deterministic_sections(context) bullets = self._bullet_points(context, sections) text = self._format_sections(sections) if self.config.max_text_chars is not None: text = _truncate_text(text, self.config.max_text_chars) return NarrativeResult( text=text, context=context, sections=sections, bullets=bullets, mode=mode, raw_model_response=None, prompt=None, metadata=copy.deepcopy(dict(self.config.metadata)), ) def _deterministic_sections(self, context: NarrativeContext) -> dict[str, str]: """Return deterministic narrative sections based on configured style.""" style = self.config.resolved_style() if style is NarrationStyle.CONCISE: return self._concise_sections(context) if style is NarrationStyle.STORY: return self._story_sections(context) if style is NarrationStyle.EXECUTIVE: return self._executive_sections(context) if style is NarrationStyle.DEBUG: return self._debug_sections(context) return self._analytical_sections(context) def _concise_sections(self, context: NarrativeContext) -> dict[str, str]: """Return concise deterministic narrative sections.""" overview = ( f"{_world_label(context)} is at step {_step_label(context.step)} with " f"{context.alive_agent_count} active agent(s) across " f"{len(context.agent_type_counts)} type(s)." ) if context.dominant_agent_type: overview += f" The largest active group is {context.dominant_agent_type!r}." metrics = self._metric_sentence(context) dynamics = self._timeline_sentence(context) takeaway = self._takeaway_sentence(context) return _drop_empty_sections( { NarrativeSection.OVERVIEW.value: overview, NarrativeSection.METRICS.value: metrics, NarrativeSection.DYNAMICS.value: dynamics, NarrativeSection.TAKEAWAY.value: takeaway, } ) def _analytical_sections(self, context: NarrativeContext) -> dict[str, str]: """Return analytical deterministic narrative sections.""" sections = { NarrativeSection.OVERVIEW.value: ( f"{_world_label(context)} currently contains {context.agent_count} total " f"agent(s), {context.alive_agent_count} active agent(s), " f"{context.resource_count} resource object(s), and " f"{context.event_count} scheduled or recorded event object(s). " f"The simulation is at step {_step_label(context.step)}." ), NarrativeSection.AGENTS.value: self._agent_section(context), NarrativeSection.RESOURCES.value: self._resource_section(context), NarrativeSection.METRICS.value: self._metrics_section(context), NarrativeSection.DYNAMICS.value: self._dynamics_section(context), NarrativeSection.ACTIVITY.value: self._activity_section(context), NarrativeSection.TAKEAWAY.value: self._takeaway_sentence(context), } return _drop_empty_sections(sections) def _story_sections(self, context: NarrativeContext) -> dict[str, str]: """Return story-style deterministic narrative sections.""" world_name = _world_label(context) dominant_agent = context.dominant_agent_type or "its agents" dominant_resource = context.dominant_resource_type or "shared resources" overview = ( f"In {world_name}, the world has reached step {_step_label(context.step)}. " f"{context.alive_agent_count} active agent(s) are shaping the simulation, " f"with {dominant_agent!r} currently standing out in the population." ) resources = "" if context.resource_type_amounts: resources = ( f"The material backdrop is led by {dominant_resource!r}, while " f"{len(context.resource_type_amounts)} resource type(s) remain visible in the world." ) dynamics = self._dynamics_section(context) activity = self._activity_section(context) takeaway = self._takeaway_sentence(context) return _drop_empty_sections( { NarrativeSection.OVERVIEW.value: overview, NarrativeSection.RESOURCES.value: resources, NarrativeSection.DYNAMICS.value: dynamics, NarrativeSection.ACTIVITY.value: activity, NarrativeSection.TAKEAWAY.value: takeaway, } ) def _executive_sections(self, context: NarrativeContext) -> dict[str, str]: """Return executive-summary style sections.""" return _drop_empty_sections( { NarrativeSection.OVERVIEW.value: ( f"{_world_label(context)} is running with {context.alive_agent_count} " f"active agent(s), {len(context.agent_type_counts)} agent type(s), " f"and {context.resource_count} resource object(s)." ), NarrativeSection.METRICS.value: self._metrics_section(context), NarrativeSection.DYNAMICS.value: self._dynamics_section(context), NarrativeSection.TAKEAWAY.value: self._takeaway_sentence(context), } ) def _debug_sections(self, context: NarrativeContext) -> dict[str, str]: """Return debug-style deterministic narrative sections.""" sections = self._analytical_sections(context) if self.config.include_debug_context: sections["debug_context"] = json.dumps(context.to_dict(), indent=2, sort_keys=True) return sections def _agent_section(self, context: NarrativeContext) -> str: """Return agent-focused narrative section.""" if context.agent_count == 0: return "No agents are present, so no agent-driven behavior can emerge yet." type_text = _top_items_text(context.agent_type_counts, value_name="agent") inactive_text = "" if context.inactive_agent_count: inactive_text = f" {context.inactive_agent_count} agent(s) are inactive." notable = _notable_objects_text(context.notable_agents, object_label="agent") return ( f"Agent composition: {type_text}.{inactive_text}" + (f" Notable agents include {notable}." if notable else "") ) def _resource_section(self, context: NarrativeContext) -> str: """Return resource-focused narrative section.""" if context.resource_count == 0: return "No resources are currently represented as world resource objects." count_text = _top_items_text(context.resource_type_counts, value_name="resource") amount_text = _top_items_text( context.resource_type_amounts, value_name="amount", numeric=True, ) notable = _notable_objects_text(context.notable_resources, object_label="resource") sentence = f"Resource composition: {count_text}." if context.resource_type_amounts: sentence += f" By amount, the leading resources are {amount_text}." if notable: sentence += f" Notable resources include {notable}." return sentence def _metrics_section(self, context: NarrativeContext) -> str: """Return metric-focused narrative section.""" if not context.metric_summaries: return "" parts: list[str] = [] interestingness = _metric_value(context.metric_summaries, "interestingness", "score") interestingness_level = _metric_value(context.metric_summaries, "interestingness", "level") if interestingness is not None: level_text = f" ({interestingness_level})" if interestingness_level else "" parts.append(f"interestingness is {_format_number(interestingness)}{level_text}") diversity = _metric_value(context.metric_summaries, "diversity", "gini_simpson_index") if diversity is not None: parts.append(f"diversity is {_format_number(diversity)}") entropy = _metric_value(context.metric_summaries, "entropy", "normalized_entropy") if entropy is not None: parts.append(f"normalized entropy is {_format_number(entropy)}") stability = _metric_value(context.metric_summaries, "stability", "stability_score") if stability is not None: parts.append(f"stability is {_format_number(stability)}") if not parts: names = ", ".join(sorted(context.metric_summaries.keys())) return f"Metric outputs are available for: {names}." return "Current metric signals indicate that " + ", ".join(parts) + "." def _dynamics_section(self, context: NarrativeContext) -> str: """Return timeline/dynamics narrative section.""" if not context.timeline_summary: return "" snapshots = _as_int(context.timeline_summary.get("snapshot_count"), 0) if snapshots <= 1: return "" agent_delta = _as_float(context.timeline_summary.get("agent_count_delta"), 0.0) resource_delta = _as_float(context.timeline_summary.get("resource_amount_delta"), 0.0) first_step = context.timeline_summary.get("first_step") last_step = context.timeline_summary.get("last_step") trend_parts = [ f"Across {snapshots} captured snapshot(s)" ] if first_step is not None or last_step is not None: trend_parts.append(f"from step {first_step} to {last_step}") trend = " ".join(trend_parts) return ( f"{trend}, active population changed by {_format_signed(agent_delta)} " f"and total resource amount changed by {_format_signed(resource_delta)}. " f"This suggests {_dynamic_interpretation(agent_delta, resource_delta)}." ) def _activity_section(self, context: NarrativeContext) -> str: """Return activity narrative section.""" if not context.activity_summary: return "" activity_count = _as_int(context.activity_summary.get("activity_count"), 0) if activity_count <= 0: return "No major memory, policy, market, planning, or communication traces are visible yet." by_path = context.activity_summary.get("activity_by_path", {}) if not isinstance(by_path, Mapping) or not by_path: return f"The world exposes {activity_count} activity trace(s)." top_paths = _top_items_text( { str(key).replace("memory.", ""): _as_float(value) for key, value in by_path.items() }, value_name="trace", numeric=True, limit=4, ) return ( f"The world exposes {activity_count} activity trace(s), with the strongest " f"signals coming from {top_paths}." ) def _metric_sentence(self, context: NarrativeContext) -> str: """Return one compact metric sentence.""" section = self._metrics_section(context) return section def _timeline_sentence(self, context: NarrativeContext) -> str: """Return one compact timeline sentence.""" return self._dynamics_section(context) def _takeaway_sentence(self, context: NarrativeContext) -> str: """Return a high-level deterministic takeaway.""" interestingness = _metric_value(context.metric_summaries, "interestingness", "score") stability = _metric_value(context.metric_summaries, "stability", "stability_score") entropy = _metric_value(context.metric_summaries, "entropy", "normalized_entropy") if interestingness is not None: score = _as_float(interestingness) if score >= 0.75: return "Takeaway: the world is highly active and varied enough to be a strong demo candidate." if score >= 0.5: return "Takeaway: the world shows meaningful structure and should produce a useful demo narrative." if score >= 0.25: return "Takeaway: the world is coherent but may need more interactions or competing pressures." return "Takeaway: the world is currently simple; adding more agents, resources, events, or adaptive behavior would make it richer." if context.agent_count == 0: return "Takeaway: add agents to create behavior and emergent dynamics." if entropy is not None and _as_float(entropy) <= 0.1: return "Takeaway: the world is coherent but concentrated around one dominant group." if stability is not None and _as_float(stability) < 0.3: return "Takeaway: the world appears volatile, which may be useful if the demo highlights instability." return "Takeaway: the world has enough structure for simulation, visualization, and narrative inspection." def _bullet_points(self, context: NarrativeContext, sections: Mapping[str, str]) -> tuple[str, ...]: """Return concise bullets extracted from context.""" bullets: list[str] = [] bullets.append( f"{context.alive_agent_count} active agent(s) across {len(context.agent_type_counts)} type(s)" ) if context.dominant_agent_type: bullets.append(f"Dominant agent type: {context.dominant_agent_type}") if context.resource_type_amounts and context.dominant_resource_type: amount = context.resource_type_amounts.get(context.dominant_resource_type, 0.0) bullets.append( f"Dominant resource by amount: {context.dominant_resource_type} ({_format_number(amount)})" ) interestingness = _metric_value(context.metric_summaries, "interestingness", "score") if interestingness is not None: bullets.append(f"Interestingness score: {_format_number(interestingness)}") if context.timeline_summary: bullets.append("Timeline data available for trend narration") if sections.get(NarrativeSection.TAKEAWAY.value): bullets.append(sections[NarrativeSection.TAKEAWAY.value].replace("Takeaway: ", "")) return tuple(bullets) def _format_sections(self, sections: Mapping[str, str]) -> str: """Format deterministic sections according to style.""" style = self.config.resolved_style() if style is NarrationStyle.CONCISE: return " ".join(text for text in sections.values() if text).strip() if style is NarrationStyle.STORY: return "\n\n".join(text for text in sections.values() if text).strip() lines: list[str] = [] for name, text in sections.items(): if not text: continue heading = name.replace("_", " ").title() lines.append(f"{heading}\n{text}") return "\n\n".join(lines).strip() def _metric_summaries( self, world: World, metric_results: Mapping[str, Any] | Sequence[Any] | None, ) -> dict[str, Any]: """Return metric summaries from supplied or default metric computations.""" summaries: dict[str, Any] = {} if metric_results is not None: summaries.update(_normalize_metric_results(metric_results)) if self.config.compute_default_metrics: summaries.update( { key: value for key, value in compute_default_metric_summaries(world).items() if key not in summaries } ) if self.config.max_metric_entries > 0 and len(summaries) > self.config.max_metric_entries: trimmed_keys = sorted(summaries.keys())[: self.config.max_metric_entries] summaries = {key: summaries[key] for key in trimmed_keys} return summaries def _activity_summary(self, world: World) -> dict[str, Any]: """Return generic activity summary from agent memories and world events.""" items = _iter_collection(getattr(world, "agents", ())) activity_by_path: dict[str, int] = {} activity_count = 0 for item in items: for path in self.config.activity_paths: contribution = _activity_contribution(_read_path(item, path, _MISSING)) if contribution <= 0: continue activity_by_path[path] = activity_by_path.get(path, 0) + contribution activity_count += contribution world_event_count = _activity_contribution(getattr(world, "events", None)) if world_event_count: activity_by_path["world.events"] = world_event_count activity_count += world_event_count sorted_activity = dict( sorted(activity_by_path.items(), key=lambda pair: (-pair[1], pair[0])) [: max(0, int(self.config.max_activity_paths))] ) return { "activity_count": activity_count, "activity_by_path": sorted_activity, } def _notable_agents(self, agents: Sequence[Any]) -> tuple[Mapping[str, Any], ...]: """Return compact summaries of notable agents.""" summaries = [_agent_summary(agent) for agent in agents] summaries.sort( key=lambda item: ( not bool(item.get("alive", True)), -_as_float(item.get("state_size"), 0.0), str(item.get("id", "")), ) ) return tuple(summaries[: max(0, int(self.config.max_notable_agents))]) def _notable_resources(self, resources: Sequence[Any]) -> tuple[Mapping[str, Any], ...]: """Return compact summaries of notable resources.""" summaries = [_resource_summary(resource) for resource in resources] summaries.sort( key=lambda item: ( -_as_float(item.get("amount"), 0.0), str(item.get("id", "")), ) ) return tuple(summaries[: max(0, int(self.config.max_notable_resources))]) def _call_client(self, prompt: NarrationPrompt) -> str: """Call the optional model client.""" if self.client is None: raise NarrationClientError("No narration client is configured") callable_obj = _resolve_client_callable(self.client) kwargs = { "prompt": prompt.text, "system_prompt": prompt.system_prompt, "user_prompt": prompt.user_prompt, "messages": [dict(message) for message in prompt.messages], **copy.deepcopy(dict(self.config.model_kwargs)), } response = _call_with_supported_kwargs(callable_obj, kwargs) return normalize_model_response(response) @staticmethod def _postprocess_text(text: str) -> str: """Clean model narrative text.""" cleaned = str(text).strip() cleaned = cleaned.removeprefix("```").removesuffix("```").strip() return cleaned def narrate_world( world: World, *, history: Sequence[Any] | None = None, metric_results: Mapping[str, Any] | Sequence[Any] | None = None, client: Any | None = None, config: NarratorConfig | None = None, extra_context: Mapping[str, Any] | None = None, ) -> str: """Narrate a world and return text. This is the simplest function to use from root-level Gradio ``app.py``. """ narrator = WorldNarrator(client=client, config=config or NarratorConfig()) return narrator.narrate_text( world, history=history, metric_results=metric_results, extra_context=extra_context, ) def narrate_world_result( world: World, *, history: Sequence[Any] | None = None, metric_results: Mapping[str, Any] | Sequence[Any] | None = None, client: Any | None = None, config: NarratorConfig | None = None, extra_context: Mapping[str, Any] | None = None, ) -> NarrativeResult: """Narrate a world and return the full ``NarrativeResult``.""" narrator = WorldNarrator(client=client, config=config or NarratorConfig()) return narrator.narrate( world, history=history, metric_results=metric_results, extra_context=extra_context, ) def narrate_simulation( history: Sequence[Any], *, metric_results: Mapping[str, Any] | Sequence[Any] | None = None, client: Any | None = None, config: NarratorConfig | None = None, extra_context: Mapping[str, Any] | None = None, ) -> str: """Narrate a simulation history and return text. The latest world-like item in ``history`` is treated as the current state. """ if not history: return "No simulation history was provided." latest_world = history[-1] narrator = WorldNarrator(client=client, config=config or NarratorConfig()) return narrator.narrate_text( latest_world, history=history, metric_results=metric_results, extra_context=extra_context, ) def build_narrative_context( world: World, *, history: Sequence[Any] | None = None, metric_results: Mapping[str, Any] | Sequence[Any] | None = None, config: NarratorConfig | None = None, extra_context: Mapping[str, Any] | None = None, ) -> NarrativeContext: """Build structured narrative context without producing prose.""" narrator = WorldNarrator(config=config or NarratorConfig()) return narrator.build_context( world, history=history, metric_results=metric_results, extra_context=extra_context, ) def compute_default_metric_summaries(world: World) -> dict[str, Any]: """Compute default metric summaries when metric modules are available. Failures are logged and ignored so narration remains robust in partial installations or during incremental hackathon development. """ summaries: dict[str, Any] = {} try: from metrics.diversity import compute_agent_type_diversity summaries["diversity"] = compute_agent_type_diversity(world).to_dict() except Exception: logger.debug("Could not compute default diversity summary", exc_info=True) try: from metrics.entropy import compute_agent_type_entropy summaries["entropy"] = compute_agent_type_entropy(world).to_dict() except Exception: logger.debug("Could not compute default entropy summary", exc_info=True) try: from metrics.stability import compute_agent_population_stability summaries["stability"] = compute_agent_population_stability(world).to_dict() except Exception: logger.debug("Could not compute default stability summary", exc_info=True) try: from metrics.interestingness import compute_interestingness summaries["interestingness"] = compute_interestingness(world).to_dict() except Exception: logger.debug("Could not compute default interestingness summary", exc_info=True) return summaries def summarize_timeline(history: Sequence[Any]) -> dict[str, Any]: """Summarize simulation history into trend-friendly numbers.""" if not history: return {} snapshots = [_timeline_snapshot(item, index=index) for index, item in enumerate(history)] snapshots = [snapshot for snapshot in snapshots if snapshot] if not snapshots: return {} first = snapshots[0] last = snapshots[-1] return { "snapshot_count": len(snapshots), "first_step": first.get("step"), "last_step": last.get("step"), "first_agent_count": first.get("alive_agent_count", 0), "last_agent_count": last.get("alive_agent_count", 0), "agent_count_delta": _as_float(last.get("alive_agent_count"), 0.0) - _as_float(first.get("alive_agent_count"), 0.0), "first_resource_amount": first.get("total_resource_amount", 0.0), "last_resource_amount": last.get("total_resource_amount", 0.0), "resource_amount_delta": _as_float(last.get("total_resource_amount"), 0.0) - _as_float(first.get("total_resource_amount"), 0.0), "snapshots": snapshots, } def normalize_model_response(response: Any) -> str: """Normalize common model-client response shapes into text.""" if response is None: raise NarrationClientError("Narration client returned None") if isinstance(response, str): return response.strip() if isinstance(response, bytes): return response.decode("utf-8").strip() if isinstance(response, Mapping): for key in ("text", "content", "generated_text", "output", "response"): if key in response and response[key] is not None: return normalize_model_response(response[key]) choices = response.get("choices") if isinstance(choices, Sequence) and choices: first = choices[0] if isinstance(first, Mapping): message = first.get("message") if isinstance(message, Mapping) and message.get("content") is not None: return normalize_model_response(message["content"]) if first.get("text") is not None: return normalize_model_response(first["text"]) for attr in ("text", "content", "generated_text", "output", "response"): value = getattr(response, attr, None) if value is not None: return normalize_model_response(value) raise NarrationClientError( f"Could not extract text from response type {response.__class__.__name__}" ) def _resolve_client_callable(client: Any) -> Callable[..., Any]: """Resolve a callable generation method from a client.""" for name in ("generate", "complete", "chat"): method = getattr(client, name, None) if callable(method): return method if callable(client): return client raise NarrationClientError( "Narration client must expose generate, complete, chat, or be callable" ) def _call_with_supported_kwargs(callable_obj: Callable[..., Any], kwargs: Mapping[str, Any]) -> Any: """Call a function using only supported keyword arguments.""" try: signature = inspect.signature(callable_obj) except (TypeError, ValueError): return callable_obj(**dict(kwargs)) parameters = signature.parameters accepts_var_keyword = any( parameter.kind is inspect.Parameter.VAR_KEYWORD for parameter in parameters.values() ) if accepts_var_keyword: return callable_obj(**dict(kwargs)) accepted = { key: value for key, value in kwargs.items() if key in parameters } if accepted: return callable_obj(**accepted) positional = [ parameter for parameter in parameters.values() if parameter.kind in { inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD, } ] if positional: return callable_obj(kwargs.get("prompt", "")) return callable_obj() def _normalize_metric_results(metric_results: Mapping[str, Any] | Sequence[Any]) -> dict[str, Any]: """Normalize metric results into a mapping keyed by metric name.""" output: dict[str, Any] = {} if isinstance(metric_results, Mapping): for key, value in metric_results.items(): output[str(key)] = _metric_to_dict(value) return output if isinstance(metric_results, Sequence) and not isinstance(metric_results, (str, bytes)): for index, result in enumerate(metric_results): result_dict = _metric_to_dict(result) name = ( result_dict.get("metric_name") or result_dict.get("name") or getattr(result, "metric_name", None) or getattr(result, "name", None) or f"metric_{index}" ) output[str(name)] = result_dict return output return output def _metric_to_dict(value: Any) -> dict[str, Any]: """Convert a metric result-like object to a dictionary.""" if hasattr(value, "to_dict") and callable(value.to_dict): raw = value.to_dict() return dict(raw) if isinstance(raw, Mapping) else {"value": raw} if isinstance(value, Mapping): return copy.deepcopy(dict(value)) if _is_number(value): return {"value": float(value)} return {"value": str(value)} def _metric_value(metrics: Mapping[str, Any], metric_name: str, key: str) -> Any: """Read a value from normalized metric summaries.""" metric = metrics.get(metric_name) if not isinstance(metric, Mapping): return None return metric.get(key) def _timeline_snapshot(item: Any, *, index: int) -> dict[str, Any]: """Convert a world-like or mapping-like history item into a compact snapshot.""" if isinstance(item, Mapping): if "values" in item and isinstance(item["values"], Mapping): values = item["values"] return { "index": index, "step": item.get("step"), "alive_agent_count": _as_float(values.get("alive_agent_count", values.get("agent_count", 0.0))), "total_resource_amount": _as_float(values.get("total_resource_amount", 0.0)), } return { "index": index, "step": item.get("step", item.get("world_step", index)), "alive_agent_count": _as_float(item.get("alive_agent_count", item.get("agent_count", 0.0))), "total_resource_amount": _as_float(item.get("total_resource_amount", item.get("resource_amount", 0.0))), } agents = _iter_collection(getattr(item, "agents", ())) resources = _iter_collection(getattr(item, "resources", ())) return { "index": index, "step": _world_step(item), "alive_agent_count": sum(1 for agent in agents if _is_alive(agent)), "total_resource_amount": sum(_as_float(getattr(resource, "amount", 0.0)) for resource in resources), } def _agent_summary(agent: Any) -> dict[str, Any]: """Return compact agent summary.""" state = _mapping_or_empty(getattr(agent, "state", {})) memory = _mapping_or_empty(getattr(agent, "memory", {})) numeric_state = { key: float(value) for key, value in state.items() if _is_number(value) } return { "id": _optional_str(getattr(agent, "id", None)), "type": _optional_str(getattr(agent, "type", None)), "alive": _is_alive(agent), "position": _json_safe(getattr(agent, "position", None)), "state_size": len(state), "memory_size": len(memory), "numeric_state": dict(sorted(numeric_state.items(), key=lambda pair: pair[0])[:6]), "current_goal": _read_path(agent, "state.current_goal", None), } def _resource_summary(resource: Any) -> dict[str, Any]: """Return compact resource summary.""" return { "id": _optional_str(getattr(resource, "id", None)), "type": _optional_str(getattr(resource, "type", None)), "amount": _as_float(getattr(resource, "amount", 0.0)), "position": _json_safe(getattr(resource, "position", None)), "metadata": _json_safe(_mapping_or_empty(getattr(resource, "metadata", {}))), } def _count_by_label( items: Sequence[Any], path: str, *, include_missing: bool, missing_label: str = "unknown", ) -> dict[str, int]: """Count items by a generic path.""" counts: dict[str, int] = {} for item in items: value = _read_path(item, path, _MISSING) if value is _MISSING or value is None or str(value) == "": if not include_missing: continue value = missing_label label = _stable_label(value) counts[label] = counts.get(label, 0) + 1 return dict(sorted(counts.items(), key=lambda pair: pair[0])) def _resource_amounts_by_type(resources: Sequence[Any]) -> dict[str, float]: """Return resource amount totals by resource type.""" amounts: dict[str, float] = {} for resource in resources: resource_type = _stable_label(getattr(resource, "type", "unknown")) amount = _as_float(getattr(resource, "amount", 0.0), 0.0) amounts[resource_type] = amounts.get(resource_type, 0.0) + amount return dict(sorted(amounts.items(), key=lambda pair: pair[0])) def _pending_event_count(events: Sequence[Any], step: int | None) -> int: """Return count of events whose trigger step has not passed.""" if step is None: return len(events) count = 0 for event in events: trigger_step = getattr(event, "trigger_step", None) if _is_number(trigger_step) and int(trigger_step) >= step: count += 1 return count def _activity_contribution(value: Any) -> int: """Return generic activity contribution count.""" if value is _MISSING or value is None: return 0 if isinstance(value, Mapping): return len(value) if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): return len(value) if isinstance(value, bool): return 1 if value else 0 if _is_number(value): return 1 if float(value) != 0.0 else 0 return 1 def _world_label(context: NarrativeContext) -> str: """Return a human-readable world label.""" return context.world_name or context.world_id or "The world" def _step_label(step: int | None) -> str: """Return a display label for a simulation step.""" return "unknown" if step is None else str(step) def _dominant_key(values: Mapping[str, float | int]) -> str | None: """Return key with largest numeric value.""" if not values: return None return max(values.items(), key=lambda pair: (_as_float(pair[1]), str(pair[0])))[0] def _top_items_text( values: Mapping[str, float | int], *, value_name: str, numeric: bool = False, limit: int = 5, ) -> str: """Return compact human-readable top items text.""" if not values: return "none" items = sorted( values.items(), key=lambda pair: (-_as_float(pair[1]), str(pair[0])), )[: max(1, int(limit))] if numeric: return ", ".join(f"{key}={_format_number(value)}" for key, value in items) return ", ".join(f"{key} ({int(_as_float(value))} {value_name}(s))" for key, value in items) def _notable_objects_text(objects: Sequence[Mapping[str, Any]], *, object_label: str) -> str: """Return compact notable object text.""" if not objects: return "" parts: list[str] = [] for item in objects[:5]: item_id = item.get("id") or "unknown" item_type = item.get("type") or object_label parts.append(f"{item_id} [{item_type}]") return ", ".join(parts) def _dynamic_interpretation(agent_delta: float, resource_delta: float) -> str: """Return simple deterministic interpretation of dynamics.""" if abs(agent_delta) <= _EPSILON and abs(resource_delta) <= _EPSILON: return "a relatively steady simulation state" if agent_delta > 0 and resource_delta > 0: return "growth in both population and resource availability" if agent_delta > 0 and resource_delta < 0: return "population growth under increasing resource pressure" if agent_delta < 0 and resource_delta > 0: return "population contraction while resources recover or accumulate" if agent_delta < 0 and resource_delta < 0: return "system-wide contraction or stress" if agent_delta > 0: return "population expansion" if agent_delta < 0: return "population decline" if resource_delta > 0: return "resource accumulation" return "resource depletion" def _drop_empty_sections(sections: Mapping[str, str]) -> dict[str, str]: """Remove empty narrative sections.""" return { str(key): str(value).strip() for key, value in sections.items() if str(value).strip() } def _format_number(value: Any) -> str: """Format numeric values compactly.""" if not _is_number(value): return str(value) numeric = float(value) if numeric.is_integer(): return str(int(numeric)) return f"{numeric:.3f}".rstrip("0").rstrip(".") def _format_signed(value: float) -> str: """Format a signed numeric change.""" numeric = float(value) sign = "+" if numeric >= 0 else "" return f"{sign}{_format_number(numeric)}" def _truncate_text(text: str, max_chars: int | None) -> str: """Truncate narrative text to a maximum character count.""" if max_chars is None or max_chars <= 0 or len(text) <= max_chars: return text if max_chars <= 3: return text[:max_chars] return text[: max_chars - 3].rstrip() + "..." def _optional_str(value: Any) -> str | None: """Return optional string value.""" if value is None: return None text = str(value) return text if text else None def _world_step(world: Any) -> int | None: """Return 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) and not isinstance(value, bool) def _as_float(value: Any, default: float = 0.0) -> float: """Convert numeric-like value to float.""" if _is_number(value): return float(value) return default def _as_int(value: Any, default: int = 0) -> int: """Convert numeric-like value to int.""" if _is_number(value): return int(value) return default def _is_alive(item: Any) -> bool: """Return whether an item is alive when it exposes an alive field.""" return bool(getattr(item, "alive", True)) def _iter_collection(raw_collection: Any) -> tuple[Any, ...]: """Return items from mapping-backed or sequence-backed collections.""" if raw_collection is None: return () if isinstance(raw_collection, Mapping): values = raw_collection.values() elif isinstance(raw_collection, Iterable) and not isinstance(raw_collection, (str, bytes)): values = raw_collection else: values = (raw_collection,) return tuple(item for item in values if item is not None) def _mapping_or_empty(value: Any) -> Mapping[str, Any]: """Return value if it is a mapping, otherwise an empty mapping.""" return value if isinstance(value, Mapping) else {} def _split_path(path: str) -> tuple[str, ...]: """Split a dot-separated path.""" return tuple(part for part in str(path).split(".") if part) def _get_mapping_path(container: Mapping[str, Any], path: str, default: Any = _MISSING) -> Any: """Read nested mapping path.""" current: Any = container for part in _split_path(path): if not isinstance(current, Mapping) or part not in current: return default current = current[part] return current def _get_object_path(root: Any, path: str, default: Any = _MISSING) -> Any: """Read nested object or mapping path.""" current: Any = root for part in _split_path(path): if isinstance(current, Mapping): if part not in current: return default current = current[part] continue if isinstance(current, Sequence) and not isinstance(current, (str, bytes)) and part.isdigit(): index = int(part) if index >= len(current): return default current = current[index] continue if not hasattr(current, part): return default current = getattr(current, part) return current def _read_path(item: Any, path: str, default: Any = _MISSING) -> Any: """Read a generic object path. Supported prefixes: - ``state.foo`` - ``state:foo`` - ``memory.foo`` - ``memory:foo`` - ``metadata.foo`` - ``metadata:foo`` """ normalized = str(path) if normalized.startswith("state."): return _get_mapping_path( _mapping_or_empty(getattr(item, "state", {})), normalized.removeprefix("state."), default, ) if normalized.startswith("state:"): return _get_mapping_path( _mapping_or_empty(getattr(item, "state", {})), normalized.removeprefix("state:"), default, ) if normalized.startswith("memory."): return _get_mapping_path( _mapping_or_empty(getattr(item, "memory", {})), normalized.removeprefix("memory."), default, ) if normalized.startswith("memory:"): return _get_mapping_path( _mapping_or_empty(getattr(item, "memory", {})), normalized.removeprefix("memory:"), default, ) if normalized.startswith("metadata."): return _get_mapping_path( _mapping_or_empty(getattr(item, "metadata", {})), normalized.removeprefix("metadata."), default, ) if normalized.startswith("metadata:"): return _get_mapping_path( _mapping_or_empty(getattr(item, "metadata", {})), normalized.removeprefix("metadata:"), default, ) return _get_object_path(item, normalized, default) def _stable_label(value: Any) -> str: """Return stable string label for arbitrary values.""" if value is None: return "none" if isinstance(value, bool): return "true" if value else "false" if _is_number(value): numeric = float(value) if numeric.is_integer(): return str(int(numeric)) return str(numeric) if isinstance(value, Mapping): parts = [f"{key}={_stable_label(value[key])}" for key in sorted(value.keys(), key=str)] return "{" + ",".join(parts) + "}" if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): return "[" + ",".join(_stable_label(item) for item in value) + "]" return str(value) def _json_safe(value: Any) -> Any: """Return JSON-friendly representation of arbitrary data.""" if value is None or isinstance(value, (str, bool)): return value if _is_number(value): numeric = float(value) if not math.isfinite(numeric): return None if numeric.is_integer(): return int(numeric) return numeric if isinstance(value, Mapping): return {str(key): _json_safe(nested) for key, nested in value.items()} if isinstance(value, Sequence) and not isinstance(value, (str, bytes)): return [_json_safe(item) for item in value] if hasattr(value, "to_dict") and callable(value.to_dict): return _json_safe(value.to_dict()) return str(value) def _append_bounded(items: MutableSequence[Any], value: Any, max_items: int) -> None: """Append while enforcing optional max history length.""" items.append(value) if max_items > 0 and len(items) > max_items: del items[: len(items) - max_items] __all__ = [ "DEFAULT_NARRATOR_SYSTEM_PROMPT", "NarrationAudience", "NarrationClientError", "NarrationError", "NarrationMode", "NarrationPrompt", "NarrationStyle", "NarrativeContext", "NarrativeResult", "NarrativeSection", "NarratorConfig", "SupportsNarrationClient", "WorldNarrator", "build_narrative_context", "compute_default_metric_summaries", "narrate_simulation", "narrate_world", "narrate_world_result", "normalize_model_response", "summarize_timeline", ]