""" Natural language to WorldSmithAI DSL generation. This module converts user prompts into ``WorldSpec`` JSON. It is designed for root-level Hugging Face Spaces ``app.py`` usage and does not assume any ``app/`` package or folder structure. The SLM's responsibility is strictly: Natural language -> JSON DSL The SLM must not generate Python code. The deterministic Python engine consumes only validated ``WorldSpec`` objects. Example root-level app.py usage: from llm.world_generator import generate_world_spec from factory.world_factory import WorldFactory def run(prompt: str): spec = generate_world_spec(prompt) world = WorldFactory().create_world(spec) return spec.to_json_string(), world Using a custom SLM client: generator = WorldGenerator(client=my_client) result = generator.generate("A research ecosystem with competing labs") print(result.json_text) Future extensibility: - Add first-class Hugging Face transformers adapters. - Add constrained decoding when available. - Add schema-version migrations. - Add model-specific prompt profiles. - Add automatic JSON repair loops. - Add world-quality self-critique before factory construction. """ from __future__ import annotations import copy import inspect import json import logging import re from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, field from enum import Enum from typing import Any, Protocol, runtime_checkable from dsl.parser import DSLParseError, ParseResult, WorldDSLParser from dsl.schema import ( AgentSpec, BehaviorSpec, EventSpec, PolicySpec, ResourceSpec, SimulationSpec, SpaceSpec, WorldSpec, ) from dsl.validator import ValidationConfig, ValidationReport, ValidationSeverity, validate_world_spec logger = logging.getLogger(__name__) DEFAULT_SYSTEM_PROMPT = """ You are WorldSmithAI's world DSL generator. Your only job is to convert a user's natural language description into a WorldSmithAI WorldSpec JSON object. Critical rules: - Output JSON only. - Do not include Markdown fences. - Do not include explanations. - Do not generate Python code. - Do not invent classes such as Sheep, Wolf, Scientist, Dragon, City, or Truck. - You must only use behavior names that are present in the known behavior list. - If the user asks for hunting, predation, chasing, fighting, or killing, use attack, seek, flee, move, or raid. - Do not invent behavior names such as hunt, graze, sleep, mate, patrol, forage, guard, chase, or rest unless they are explicitly listed in the known behavior registry. - Use generic DSL entities: agents, resources, events, behaviors, policies. - Agents may have any imaginative type string, but the runtime object is still Agent. - Behaviors must be referenced by registry name. - Policies must be referenced by registry type. - Keep the world small enough for a lightweight Gradio demo. - Prefer 4 to 8 agents unless the user asks for more. - Prefer 2D positions when possible. - Include state, memory, goals, behaviors, and policy for each agent. - Use deterministic values and avoid randomness in the DSL. """.strip() WORLD_DSL_FORMAT_GUIDE = """ Return one JSON object with this structure: { "schema_version": "1.0", "id": "short_world_id", "name": "Human readable world name", "description": "Short description", "simulation": { "steps": 50, "seed": 0, "scheduler": "sequential", "activation": "sequential", "collect_history": true }, "space": { "dimensions": 2, "bounds": [[0, 10], [0, 10]], "toroidal": false, "enforce_bounds": true }, "agents": [ { "id": "agent_1", "type": "generic_type", "position": [1, 1], "state": {"energy": 10, "credits": 5}, "memory": {"known_options": {"cooperate": 1.0}}, "goals": [{"id": "survive", "importance": 1.0, "score": 1.0}], "behaviors": [ {"name": "remember", "params": {"category": "observation", "content": {"note": "initial memory"}}}, {"name": "prioritize", "params": {}}, {"name": "choose_goal", "params": {}}, {"name": "communicate", "params": {"message": "status_update", "max_recipients": 1}}, {"name": "cooperate", "params": {"effort": 1.0}} ], "policy": { "type": "rule_policy", "params": { "rules": [ {"behavior_name": "choose_goal", "score_delta": 3.0}, {"behavior_name": "communicate", "score_delta": 1.0} ] } }, "alive": true, "metadata": {} } ], "resources": [ { "id": "resource_1", "type": "generic_resource", "amount": 10, "position": [5, 5], "regeneration_rate": 0, "max_amount": 20, "metadata": {} } ], "events": [], "metrics": [ {"name": "diversity", "params": {"collection": "agents", "group_by_path": "type"}} ], "metadata": {} } Known behavior names available in this project include: move, wander, seek, flee, eat, harvest, mine, collect, buy, sell, exchange, attack, defend, raid, study, publish, learn, cooperate, communicate, share, recommend, negotiate, route, deliver, ship, queue, charge, build, repair, expand, demolish, regulate, tax, subsidize, litigate, approve, enforce, bid, ask, price, discount, forecast, rebalance, adopt, switch, imitate, abandon, evaluate, prioritize, schedule, allocate_time, choose_goal, remember, forget, reinforce, update_belief. Behavior mapping guidance: - "hunt" should usually be represented as "attack" plus optionally "seek". - "run away" should be represented as "flee". - "look for" or "track" should be represented as "seek". - "random movement" should be represented as "wander" or "move". Known policy types include: rule_policy, contextual_bandit. """.strip() DEFAULT_MINIMAL_EXAMPLE = { "schema_version": "1.0", "id": "tiny_research_world", "name": "Tiny Research World", "description": "A small generic research ecosystem.", "simulation": { "steps": 40, "seed": 0, "scheduler": "sequential", "activation": "sequential", "collect_history": True, }, "space": { "dimensions": 2, "bounds": [[0, 10], [0, 10]], "toroidal": False, "enforce_bounds": True, }, "agents": [ { "id": "scientist_1", "type": "scientist", "position": [1, 1], "state": {"energy": 10, "credits": 5, "knowledge": 1}, "memory": {"known_options": {"study": 1.0, "communicate": 0.5}}, "goals": [{"id": "discover", "importance": 2.0, "score": 2.0}], "behaviors": [ { "name": "remember", "params": { "category": "observation", "content": {"note": "initial research context"}, }, }, {"name": "prioritize", "params": {}}, {"name": "choose_goal", "params": {}}, {"name": "communicate", "params": {"message": "research_update"}}, {"name": "cooperate", "params": {"goal": "discover", "effort": 1.0}}, ], "policy": { "type": "rule_policy", "params": { "rules": [ {"behavior_name": "choose_goal", "score_delta": 3.0}, {"behavior_name": "communicate", "score_delta": 2.0}, {"behavior_name": "cooperate", "score_delta": 1.0}, ] }, }, "alive": True, "metadata": {}, } ], "resources": [ { "id": "knowledge_pool", "type": "knowledge", "amount": 12, "position": [5, 5], "regeneration_rate": 0.2, "max_amount": 20, "metadata": {}, } ], "events": [], "metrics": [ {"name": "diversity", "params": {"collection": "agents", "group_by_path": "type"}}, {"name": "entropy", "params": {"collection": "agents", "value_path": "type"}}, {"name": "interestingness", "params": {"collection": "agents", "group_by_path": "type"}}, ], "metadata": {"generated_by": "worldsmithai_example"}, } class GenerationMode(str, Enum): """Supported generation modes.""" MODEL = "model" FALLBACK = "fallback" AUTO = "auto" class ResponseFormat(str, Enum): """Expected model response format.""" JSON_OBJECT = "json_object" TEXT_WITH_JSON = "text_with_json" class WorldGenerationError(RuntimeError): """Raised when natural-language-to-DSL generation fails.""" class ModelClientError(WorldGenerationError): """Raised when an SLM client cannot be called successfully.""" class WorldGenerationValidationError(WorldGenerationError): """Raised when generated DSL fails strict validation.""" def __init__(self, report: ValidationReport) -> None: """Initialize the exception with a semantic validation report.""" self.report = report super().__init__(report.summary()) @runtime_checkable class SupportsGenerate(Protocol): """Protocol for simple text generation clients.""" def generate(self, *args: Any, **kwargs: Any) -> Any: """Generate text from prompt-like arguments.""" @dataclass(frozen=True) class PromptBundle: """Prompt payload sent to an SLM client. The bundle includes both chat-style messages and flattened prompt text so many different client APIs can be supported from root-level ``app.py``. """ system_prompt: str user_prompt: str messages: tuple[Mapping[str, str], ...] metadata: Mapping[str, Any] = field(default_factory=dict) @property def text(self) -> str: """Return a flattened prompt for completion-style models.""" return f"{self.system_prompt}\n\nUSER REQUEST:\n{self.user_prompt}" def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly prompt bundle.""" return { "system_prompt": self.system_prompt, "user_prompt": self.user_prompt, "messages": [dict(message) for message in self.messages], "metadata": dict(self.metadata), } @dataclass(frozen=True) class WorldGenerationResult: """Result returned by ``WorldGenerator.generate``. Attributes: spec: Validated world specification. raw_response: Raw text returned by the model, or fallback JSON text. parse_result: Parser result when available. validation_report: Optional semantic validation report. mode: Whether model or deterministic fallback produced the final spec. prompt_bundle: Prompt used for model generation when available. metadata: Additional diagnostics useful for Gradio display. """ spec: WorldSpec raw_response: str parse_result: ParseResult | None = None validation_report: ValidationReport | None = None mode: GenerationMode = GenerationMode.MODEL prompt_bundle: PromptBundle | None = None metadata: Mapping[str, Any] = field(default_factory=dict) @property def json_text(self) -> str: """Return the generated world spec as formatted JSON.""" return self.spec.to_json_string(indent=2, exclude_none=True) @property def is_semantically_valid(self) -> bool: """Return whether semantic validation passed or was not run.""" return self.validation_report is None or self.validation_report.is_valid def to_dict(self) -> dict[str, Any]: """Return a JSON-friendly result summary.""" return { "world_id": self.spec.id, "world_name": self.spec.name, "mode": self.mode.value, "agent_count": len(self.spec.agents), "resource_count": len(self.spec.resources), "event_count": len(self.spec.events), "behavior_count": len(self.spec.behavior_names), "is_semantically_valid": self.is_semantically_valid, "validation_report": None if self.validation_report is None else self.validation_report.to_dict(), "parse_result": None if self.parse_result is None else self.parse_result.to_dict(), "metadata": copy.deepcopy(dict(self.metadata)), } @dataclass class WorldGenerationConfig: """Configuration for ``WorldGenerator``. Defaults are designed for a Gradio demo: - model output is parsed strictly, - semantic validation runs but does not raise by default, - deterministic fallback is enabled, - generated worlds stay small enough for interactive simulation. """ mode: GenerationMode | str = GenerationMode.AUTO response_format: ResponseFormat | str = ResponseFormat.JSON_OBJECT system_prompt: str = DEFAULT_SYSTEM_PROMPT include_format_guide: bool = True include_example: bool = True include_json_schema_excerpt: bool = False default_steps: int = 60 default_seed: int = 0 max_agents_hint: int = 8 max_resources_hint: int = 6 parser: WorldDSLParser = field(default_factory=WorldDSLParser) semantic_validation: bool = True strict_semantic_validation: bool = False semantic_validation_config: ValidationConfig | None = None fallback_on_model_error: bool = True fallback_on_parse_error: bool = True fallback_on_semantic_error: bool = False repair_attempts: int = 1 model_kwargs: Mapping[str, Any] = field(default_factory=dict) metadata: Mapping[str, Any] = field(default_factory=dict) def resolved_mode(self) -> GenerationMode: """Return normalized generation mode.""" if isinstance(self.mode, GenerationMode): return self.mode return GenerationMode(str(self.mode)) def resolved_response_format(self) -> ResponseFormat: """Return normalized response format.""" if isinstance(self.response_format, ResponseFormat): return self.response_format return ResponseFormat(str(self.response_format)) @dataclass class DeterministicWorldSpecBuilder: """Build a valid generic fallback ``WorldSpec`` without an SLM. This fallback exists so the Gradio demo can still run when no SLM client is configured or when the model returns invalid JSON. Unlike the earliest fallback version, this builder preserves important prompt entities when possible. For example, a prompt mentioning sheep, wolves, a hunter, and grass will produce sheep agents, wolf agents, a hunter agent, and grass resources instead of generic farm-support roles. """ default_steps: int = 60 default_seed: int = 0 max_agents: int = 6 def build( self, prompt: str, *, constraints: Mapping[str, Any] | str | None = None, ) -> WorldSpec: """Build a deterministic world spec from prompt keywords.""" domain = self._infer_domain(prompt) max_agents = self._max_agents_from_constraints(constraints) or self.max_agents max_agents = max(1, int(max_agents)) roles = self._roles_for_prompt(prompt, domain, max_agents=max_agents) resources = self._resources_for_prompt(prompt, domain) world_id = _slugify(f"{domain}_world") agents = tuple( self._agent_spec( index=index, role=role, domain=domain, roles=roles, ) for index, role in enumerate(roles) ) resource_specs = tuple( ResourceSpec( id=self._resource_id(resource_type, index), type=resource_type, amount=float(10 + index * 4), position=self._resource_position_for_index(index), regeneration_rate=self._resource_regeneration_rate(resource_type), max_amount=float(30 + index * 5), metadata={ "domain": domain, "generated_by": "deterministic_fallback", "prompt_entity": resource_type in self._prompt_entity_resource_names(prompt), }, ) for index, resource_type in enumerate(resources) ) events = self._events_for_prompt( prompt=prompt, domain=domain, roles=roles, resources=resources, ) return WorldSpec( schema_version="1.0", id=world_id, name=_title_from_domain(domain), description=self._description(prompt, domain, constraints), simulation=SimulationSpec( steps=max(1, int(self.default_steps)), seed=self.default_seed, scheduler="sequential", activation="sequential", collect_history=True, ), space=SpaceSpec( dimensions=2, bounds=((0.0, 10.0), (0.0, 10.0)), toroidal=False, enforce_bounds=True, ), agents=agents, resources=resource_specs, events=events, metrics=( {"name": "diversity", "params": {"collection": "agents", "group_by_path": "type"}}, {"name": "entropy", "params": {"collection": "agents", "value_path": "type"}}, {"name": "stability", "params": {"collection": "agents", "group_by_path": "type"}}, {"name": "interestingness", "params": {"collection": "agents", "group_by_path": "type"}}, ), metadata={ "generated_by": "deterministic_fallback", "domain": domain, "source_prompt": prompt, "entity_aware_fallback": True, "constraints": constraints if isinstance(constraints, Mapping) else str(constraints or ""), }, ) def _agent_spec( self, *, index: int, role: str, domain: str, roles: Sequence[str], ) -> AgentSpec: """Create one generic fallback agent spec.""" agent_id = self._agent_id_for_roles(roles, index) target_agent_id = self._next_agent_id_for_roles(roles, index) goals = self._goals_for_role(role, domain) goal_records = [ { "id": goal, "importance": float(len(goals) - goal_index), "score": float(len(goals) - goal_index), "description": f"{role} works toward {goal}", } for goal_index, goal in enumerate(goals) ] x, y = self._position_for_index(index) role_state = self._state_for_role(role, domain, index) behaviors = [ BehaviorSpec( name="remember", params={ "category": "initial_context", "content": { "role": role, "domain": domain, "note": f"Seed memory for {role} created by entity-aware fallback generator.", }, "tags": ["initial", domain, role], }, ), BehaviorSpec( name="evaluate", params={ "category": f"{role}_strategy", "candidate_options": self._options_for_role(role), "known_options_path": "memory.known_options", "adoption_threshold": 0.7, "write_decision_path": "memory.latest_strategy_decision", }, ), BehaviorSpec( name="prioritize", params={ "source_path": "memory.goals", "item_attribute_weights": { "importance": 1.0, "score": 1.0, }, }, ), BehaviorSpec( name="choose_goal", params={ "source_path": "memory.latest_priorities.items", "current_goal_path": "state.current_goal", }, ), BehaviorSpec( name="communicate", params={ "target_agent_id": target_agent_id, "message": f"{role}_status_update", "max_recipients": 1, } if target_agent_id is not None else { "message": f"{role}_status_update", "max_recipients": 1, }, ), BehaviorSpec( name="cooperate", params={ "target_agent_id": target_agent_id, "goal": goals[0], "effort": self._effort_for_role(role), } if target_agent_id is not None else { "goal": goals[0], "effort": self._effort_for_role(role), }, ), BehaviorSpec( name="update_belief", params={ "belief_id": f"{role}_situation_belief", "proposition": self._belief_for_role(role), "evidence_path": self._evidence_path_for_role(role), "learning_rate": 0.25, "write_belief_path": "memory.latest_situation_belief", }, ), ] if role in {"hunter", "farmer"}: behaviors.append( BehaviorSpec( name="recommend", params={ "target_agent_id": target_agent_id, "recommendation": "protect_flock_and_monitor_wolves", "confidence": 0.75, } if target_agent_id is not None else { "recommendation": "protect_flock_and_monitor_wolves", "confidence": 0.75, }, ) ) if role == "wolf": behaviors.append( BehaviorSpec( name="share", params={ "target_agent_id": target_agent_id, "content_path": "state.pack_signal", "share_key": "pack_signal_share", "relationship_delta": 0.05, } if target_agent_id is not None else { "content_path": "state.pack_signal", "share_key": "pack_signal_share", "relationship_delta": 0.05, }, ) ) policy_rules = [ {"behavior_name": "choose_goal", "score_delta": 4.0, "priority_delta": 1.0}, {"behavior_name": "prioritize", "score_delta": 3.0}, {"behavior_name": "communicate", "score_delta": 2.0}, {"behavior_name": "cooperate", "score_delta": 1.5}, {"behavior_name": "evaluate", "score_delta": 1.0}, {"behavior_name": "update_belief", "score_delta": 0.8}, {"behavior_name": "remember", "score_delta": 0.5}, ] if role in {"hunter", "farmer"}: policy_rules.append({"behavior_name": "recommend", "score_delta": 1.2}) if role == "wolf": policy_rules.append({"behavior_name": "share", "score_delta": 1.2}) return AgentSpec( id=agent_id, type=role, position=(x, y), state=role_state, memory={ "goals": goal_records, "known_options": self._known_options_for_role(role), "relationships": {}, "source_role": role, }, goals=goal_records, behaviors=tuple(behaviors), policy=PolicySpec( type="rule_policy", params={ "selection_strategy": "highest_score", "rules": policy_rules, }, ), alive=True, metadata={ "domain": domain, "generated_by": "deterministic_fallback", "prompt_role": role, }, ) def _roles_for_prompt( self, prompt: str, domain: str, *, max_agents: int, ) -> tuple[str, ...]: """Return role list that preserves important prompt entities.""" text = prompt.lower() roles: list[str] = [] if domain == "farm": roles.append("farmer") has_sheep = self._contains_any(text, ("sheep", "lamb", "flock")) has_wolf = self._contains_any(text, ("wolf", "wolves", "pack")) has_hunter = self._contains_any(text, ("hunter", "ranger", "shepherd")) if has_sheep: roles.append("sheep") if has_wolf: roles.append("wolf") if has_hunter: roles.append("hunter") if has_sheep and len(roles) < max_agents: roles.append("sheep") if has_wolf and len(roles) < max_agents: roles.append("wolf") if not has_sheep and not has_wolf and not has_hunter: roles.extend(self._roles_for_domain(domain)) return tuple(roles[:max_agents]) roles.extend(self._roles_for_domain(domain)) return tuple(roles[:max_agents]) def _resources_for_prompt(self, prompt: str, domain: str) -> tuple[str, ...]: """Return resource types that preserve important prompt entities.""" text = prompt.lower() if domain == "farm": resources: list[str] = [] if self._contains_any(text, ("grass", "pasture")): resources.append("grass") else: resources.append("food") if self._contains_any(text, ("sheep", "flock", "lamb")): resources.append("flock_safety") resources.append("shelter") if self._contains_any(text, ("wolf", "wolves", "pack", "hunter")): resources.append("predator_pressure") resources.append("water") if "soil_health" not in resources: resources.append("soil_health") return tuple(dict.fromkeys(resources)) return self._resources_for_domain(domain) def _events_for_prompt( self, *, prompt: str, domain: str, roles: Sequence[str], resources: Sequence[str], ) -> tuple[EventSpec, ...]: """Return domain events that preserve important prompt entities.""" text = prompt.lower() events: list[EventSpec] = [] wolf_ids = [ self._agent_id_for_roles(roles, index) for index, role in enumerate(roles) if role == "wolf" ] sheep_ids = [ self._agent_id_for_roles(roles, index) for index, role in enumerate(roles) if role == "sheep" ] hunter_ids = [ self._agent_id_for_roles(roles, index) for index, role in enumerate(roles) if role == "hunter" ] farmer_ids = [ self._agent_id_for_roles(roles, index) for index, role in enumerate(roles) if role == "farmer" ] resource_ids = { resource_type: self._resource_id(resource_type, index) for index, resource_type in enumerate(resources) } if domain == "farm" and self._contains_any(text, ("wolf", "wolves", "pack")): events.append( EventSpec( id="wolf_pack_enters", name="wolf_pack_enters", trigger_step=max(1, self.default_steps // 4), payload={ "description": "A pack of wolves enters the farm and increases pressure on the flock.", "signals": { "predator_pressure": 0.35, "flock_fear": 0.25, }, }, enabled=True, target_agent_ids=tuple(wolf_ids + sheep_ids + farmer_ids), target_resource_ids=tuple( resource_id for key, resource_id in resource_ids.items() if key in {"predator_pressure", "flock_safety", "grass"} ), metadata={ "generated_by": "entity_aware_fallback", "narrative_hint": "Predator pressure tests flock safety and farm coordination.", }, ) ) if domain == "farm" and self._contains_any(text, ("hunter", "ranger")): events.append( EventSpec( id="hunter_enters", name="hunter_enters", trigger_step=max(2, self.default_steps // 3), payload={ "description": "A hunter enters the farm area and changes the balance between wolves and sheep.", "signals": { "protection_pressure": 0.3, "wolf_caution": 0.2, }, }, enabled=True, target_agent_ids=tuple(hunter_ids + wolf_ids + farmer_ids + sheep_ids), target_resource_ids=tuple( resource_id for key, resource_id in resource_ids.items() if key in {"predator_pressure", "flock_safety"} ), metadata={ "generated_by": "entity_aware_fallback", "narrative_hint": "Hunter arrival adds a stabilizing or disruptive force.", }, ) ) if not events: events.append( EventSpec( id="midpoint_pressure", name="midpoint_pressure", trigger_step=max(1, self.default_steps // 2), payload={ "description": "A generic pressure event that can be narrated later.", "domain": domain, }, enabled=True, metadata={"generated_by": "fallback_builder"}, ) ) return tuple(events) def _state_for_role(self, role: str, domain: str, index: int) -> dict[str, Any]: """Return role-specific initial state.""" base_state: dict[str, Any] = { "energy": float(max(3, 10 - index)), "credits": float(5 + index), "influence": float(index + 1), "domain_pressure": 0.2, "inventory": {}, } if role == "sheep": base_state.update( { "hunger": 0.4, "fear": 0.35, "herd_signal": 0.75, "grazing_need": 0.8, "inventory": {"wool": 1}, } ) return base_state if role == "wolf": base_state.update( { "hunger": 0.85, "pack_signal": 0.8, "caution": 0.3, "predator_intent": 0.9, "inventory": {}, } ) return base_state if role == "hunter": base_state.update( { "tracking_skill": 0.8, "protection_signal": 0.7, "caution": 0.55, "inventory": {"tools": 2, "supplies": 2}, } ) return base_state if role == "farmer": base_state.update( { "flock_care": 0.8, "grass_management": 0.65, "protection_signal": 0.5, "inventory": {"hay": 3, "tools": 2}, } ) return base_state base_state.update( { "role_signal": 0.6, "inventory": {}, } ) return base_state def _goals_for_role(self, role: str, domain: str) -> tuple[str, ...]: """Return role-specific fallback goal ids.""" if role == "sheep": return ("graze_on_grass", "stay_with_flock", "avoid_wolves") if role == "wolf": return ("coordinate_pack", "seek_food_pressure", "avoid_hunter") if role == "hunter": return ("protect_flock", "track_wolves", "avoid_unnecessary_harm") if role == "farmer": return ("protect_sheep", "maintain_grass", "coordinate_hunter") return self._goals_for_domain(domain) def _known_options_for_role(self, role: str) -> dict[str, Any]: """Return role-specific known options.""" if role == "sheep": return { "graze": {"score": 1.0, "energy": 0.8}, "flock_together": {"score": 1.2, "safety": 0.9}, "avoid_predator": {"score": 1.4, "safety": 1.0}, } if role == "wolf": return { "pack_coordination": {"score": 1.3, "success": 0.8}, "cautious_entry": {"score": 1.0, "safety": 0.7}, "pressure_flock": {"score": 1.2, "food": 0.8}, } if role == "hunter": return { "track_wolves": {"score": 1.2, "protection": 0.9}, "coordinate_farmer": {"score": 1.1, "trust": 0.8}, "guard_pasture": {"score": 1.0, "safety": 0.9}, } if role == "farmer": return { "protect_flock": {"score": 1.2, "safety": 0.9}, "maintain_pasture": {"score": 1.0, "grass": 0.8}, "coordinate_hunter": {"score": 1.1, "protection": 0.8}, } return { "cooperate": 1.0, "communicate": 0.8, "evaluate": 0.6, } def _options_for_role(self, role: str) -> list[str]: """Return candidate option ids for evaluation behavior.""" return list(self._known_options_for_role(role).keys()) def _belief_for_role(self, role: str) -> str: """Return a role-specific belief proposition.""" if role == "sheep": return "The flock can stay safe while grazing." if role == "wolf": return "The pack can coordinate under farm and hunter pressure." if role == "hunter": return "The hunter can reduce wolf pressure without destabilizing the farm." if role == "farmer": return "The farm can protect sheep while maintaining grass resources." return "The current situation is manageable." def _evidence_path_for_role(self, role: str) -> str: """Return a role-specific evidence path.""" if role == "sheep": return "state.herd_signal" if role == "wolf": return "state.pack_signal" if role == "hunter": return "state.protection_signal" if role == "farmer": return "state.flock_care" return "state.domain_pressure" def _effort_for_role(self, role: str) -> float: """Return role-specific cooperation effort.""" if role in {"hunter", "farmer", "wolf"}: return 1.0 if role == "sheep": return 0.7 return 0.8 @staticmethod def _resource_position_for_index(index: int) -> tuple[float, float]: """Return deterministic in-bounds 2D position for a resource index.""" positions = ( (2.0, 7.0), (4.0, 6.0), (6.0, 5.0), (8.0, 4.0), (2.0, 3.0), (4.0, 2.0), (6.0, 7.0), (8.0, 6.0), (5.0, 5.0), (7.0, 3.0), ) return positions[index % len(positions)] def _position_for_index(self, index: int) -> tuple[float, float]: """Return deterministic 2D position for agent index.""" positions = ( (1.0, 1.0), (3.0, 6.5), (7.5, 6.5), (8.5, 2.0), (4.0, 7.5), (6.5, 2.5), (2.0, 4.0), (8.0, 8.0), ) return positions[index % len(positions)] def _agent_id_for_roles(self, roles: Sequence[str], index: int) -> str: """Return stable id for an agent in a possibly duplicated role list.""" role = roles[index] ordinal = sum(1 for candidate in roles[: index + 1] if candidate == role) return f"{role}_{ordinal}" def _next_agent_id_for_roles(self, roles: Sequence[str], index: int) -> str | None: """Return next cyclic target agent id.""" if len(roles) <= 1: return None next_index = (index + 1) % len(roles) return self._agent_id_for_roles(roles, next_index) def _resource_id(self, resource_type: str, index: int) -> str: """Return stable resource id.""" return f"{_slugify(resource_type)}_{index + 1}" def _resource_regeneration_rate(self, resource_type: str) -> float: """Return resource-specific regeneration rate.""" if resource_type in {"grass", "food", "water", "soil_health"}: return 0.2 if resource_type in {"flock_safety", "predator_pressure"}: return 0.05 return 0.0 def _max_agents_from_constraints( self, constraints: Mapping[str, Any] | str | None, ) -> int | None: """Extract max_agents from parsed constraints when available.""" if not isinstance(constraints, Mapping): return None value = constraints.get("max_agents") if value is None: return None try: return max(1, int(value)) except (TypeError, ValueError): return None def _prompt_entity_resource_names(self, prompt: str) -> set[str]: """Return resource names explicitly suggested by the prompt.""" text = prompt.lower() names: set[str] = set() if self._contains_any(text, ("grass", "pasture")): names.add("grass") return names @staticmethod def _contains_any(text: str, words: Sequence[str]) -> bool: """Return whether any word-like string appears in text.""" lowered = text.lower() return any(word in lowered for word in words) @staticmethod def _infer_domain(prompt: str) -> str: """Infer a broad fallback domain from prompt keywords.""" text = prompt.lower() keyword_domains = ( ("farm", ("farm", "crop", "soil", "harvest", "animal", "pasture", "sheep", "wolf", "wolves", "grass")), ("medieval", ("medieval", "kingdom", "castle", "guild", "village", "empire")), ("research", ("research", "scientist", "paper", "lab", "study", "knowledge")), ("startup", ("startup", "founder", "investor", "customer", "market", "product")), ("social", ("social", "network", "influencer", "community", "friend", "viral")), ("transport", ("transport", "route", "traffic", "delivery", "ship", "vehicle")), ("power", ("power", "grid", "energy", "battery", "generator", "electric")), ("space", ("space", "colony", "planet", "terraform", "asteroid", "orbital")), ("fantasy", ("fantasy", "dragon", "mana", "mage", "kingdom", "quest")), ) for domain, keywords in keyword_domains: if any(keyword in text for keyword in keywords): return domain return "generic" @staticmethod def _roles_for_domain(domain: str) -> tuple[str, ...]: """Return fallback agent roles for a domain.""" roles_by_domain = { "farm": ("farmer", "pollinator", "soil_guardian", "market_vendor"), "medieval": ("ruler", "merchant", "artisan", "scholar", "guard"), "research": ("scientist", "reviewer", "engineer", "curator", "funding_agent"), "startup": ("founder", "customer", "investor", "operator", "competitor"), "social": ("creator", "moderator", "member", "recommender", "critic"), "transport": ("hub", "carrier", "dispatcher", "charger", "passenger"), "power": ("generator", "storage_node", "consumer", "regulator", "maintainer"), "space": ("colonist", "engineer", "botanist", "navigator", "miner"), "fantasy": ("mage", "merchant", "guardian", "dragon", "healer"), "generic": ("explorer", "builder", "trader", "learner", "coordinator"), } return roles_by_domain.get(domain, roles_by_domain["generic"]) @staticmethod def _resources_for_domain(domain: str) -> tuple[str, ...]: """Return fallback resource types for a domain.""" resources_by_domain = { "farm": ("food", "water", "soil_health"), "medieval": ("grain", "gold", "craft_materials"), "research": ("knowledge", "funding", "attention"), "startup": ("capital", "users", "compute"), "social": ("attention", "trust", "content"), "transport": ("capacity", "fuel", "demand"), "power": ("energy", "storage", "load"), "space": ("oxygen", "minerals", "habitat_capacity"), "fantasy": ("mana", "relics", "influence"), "generic": ("energy", "materials", "knowledge"), } return resources_by_domain.get(domain, resources_by_domain["generic"]) @staticmethod def _goals_for_domain(domain: str) -> tuple[str, ...]: """Return fallback goal ids for a domain.""" goals_by_domain = { "farm": ("sustain_harvest", "share_resources", "adapt_to_weather"), "medieval": ("maintain_order", "grow_trade", "protect_settlement"), "research": ("discover", "publish", "collaborate"), "startup": ("find_users", "improve_product", "raise_capital"), "social": ("grow_trust", "spread_useful_content", "moderate_conflict"), "transport": ("deliver_reliably", "reduce_congestion", "balance_capacity"), "power": ("meet_demand", "prevent_outage", "rebalance_energy"), "space": ("survive_colony", "expand_habitat", "extract_resources"), "fantasy": ("protect_realm", "gather_mana", "negotiate_alliances"), "generic": ("survive", "learn", "cooperate"), } return goals_by_domain.get(domain, goals_by_domain["generic"]) @staticmethod def _description( prompt: str, domain: str, constraints: Mapping[str, Any] | str | None, ) -> str: """Build fallback world description.""" constraint_text = "" if constraints: constraint_text = f" Constraints: {constraints}" return ( f"A deterministic entity-aware fallback {domain} world generated from the prompt: " f"{prompt.strip()[:240]}{constraint_text}" ) @dataclass class WorldGenerator: """Generate ``WorldSpec`` objects from natural language prompts. The generator is intentionally adapter-neutral. A client can be any object exposing ``generate``, ``complete``, ``chat``, or ``__call__``. The module does not import a specific model SDK. """ client: Any | None = None config: WorldGenerationConfig = field(default_factory=WorldGenerationConfig) def generate( self, prompt: str, *, constraints: Mapping[str, Any] | str | None = None, examples: Sequence[WorldSpec | Mapping[str, Any] | str] | None = None, model_kwargs: Mapping[str, Any] | None = None, ) -> WorldGenerationResult: """Generate a validated ``WorldSpec`` from natural language. Args: prompt: Natural language world description. constraints: Optional extra constraints from the UI. examples: Optional few-shot examples as specs, mappings, or strings. model_kwargs: Optional generation kwargs for the model client. Returns: ``WorldGenerationResult``. Raises: WorldGenerationError: If generation fails and fallback is disabled. """ normalized_prompt = _require_prompt(prompt) mode = self.config.resolved_mode() if mode is GenerationMode.FALLBACK or self.client is None: return self._fallback_result( normalized_prompt, constraints=constraints, reason="fallback_mode" if mode is GenerationMode.FALLBACK else "client_not_configured", ) prompt_bundle = self.build_prompt( normalized_prompt, constraints=constraints, examples=examples, ) try: raw_response = self._call_client(prompt_bundle, model_kwargs=model_kwargs) except Exception as exc: if self.config.fallback_on_model_error and mode is GenerationMode.AUTO: logger.warning("Model client failed; using deterministic fallback: %s", exc) return self._fallback_result( normalized_prompt, constraints=constraints, reason="model_client_error", error=exc, prompt_bundle=prompt_bundle, ) raise ModelClientError(f"Model client failed: {exc}") from exc parse_result: ParseResult | None = None parse_error: BaseException | None = None current_response = raw_response for attempt in range(max(1, int(self.config.repair_attempts) + 1)): try: parse_result = self.config.parser.parse_result(current_response) break except Exception as exc: parse_error = exc if attempt >= int(self.config.repair_attempts): break current_response = self._repair_response( prompt_bundle=prompt_bundle, bad_response=current_response, error=exc, model_kwargs=model_kwargs, ) if parse_result is None: if self.config.fallback_on_parse_error and mode is GenerationMode.AUTO: logger.warning("Generated DSL could not be parsed; using fallback: %s", parse_error) return self._fallback_result( normalized_prompt, constraints=constraints, reason="parse_error", error=parse_error, raw_response=raw_response, prompt_bundle=prompt_bundle, ) raise WorldGenerationError(f"Generated DSL could not be parsed: {parse_error}") from parse_error validation_report = self._semantic_validation(parse_result.spec) if validation_report is not None and not validation_report.is_valid: if self.config.strict_semantic_validation: if self.config.fallback_on_semantic_error and mode is GenerationMode.AUTO: return self._fallback_result( normalized_prompt, constraints=constraints, reason="semantic_validation_error", raw_response=raw_response, prompt_bundle=prompt_bundle, ) raise WorldGenerationValidationError(validation_report) return WorldGenerationResult( spec=parse_result.spec, raw_response=raw_response, parse_result=parse_result, validation_report=validation_report, mode=GenerationMode.MODEL, prompt_bundle=prompt_bundle, metadata={ **copy.deepcopy(dict(self.config.metadata)), "repair_attempts_used": 0 if current_response == raw_response else 1, }, ) def generate_spec( self, prompt: str, *, constraints: Mapping[str, Any] | str | None = None, examples: Sequence[WorldSpec | Mapping[str, Any] | str] | None = None, model_kwargs: Mapping[str, Any] | None = None, ) -> WorldSpec: """Generate and return only the ``WorldSpec``.""" return self.generate( prompt, constraints=constraints, examples=examples, model_kwargs=model_kwargs, ).spec def generate_json( self, prompt: str, *, constraints: Mapping[str, Any] | str | None = None, examples: Sequence[WorldSpec | Mapping[str, Any] | str] | None = None, model_kwargs: Mapping[str, Any] | None = None, ) -> str: """Generate and return formatted WorldSpec JSON.""" return self.generate( prompt, constraints=constraints, examples=examples, model_kwargs=model_kwargs, ).json_text def build_prompt( self, prompt: str, *, constraints: Mapping[str, Any] | str | None = None, examples: Sequence[WorldSpec | Mapping[str, Any] | str] | None = None, ) -> PromptBundle: """Build the prompt bundle sent to the model client.""" sections: list[str] = [] sections.append(f"World request:\n{prompt.strip()}") if constraints: sections.append(f"Additional constraints:\n{_serialize_constraints(constraints)}") sections.append( "Return a compact but complete WorldSmithAI WorldSpec JSON object. " "The JSON must be parseable by Python json.loads." ) if self.config.include_format_guide: sections.append(f"WorldSpec format guide:\n{WORLD_DSL_FORMAT_GUIDE}") if self.config.include_json_schema_excerpt: sections.append(f"Schema excerpt:\n{world_spec_schema_excerpt()}") example_payloads = self._example_payloads(examples) if self.config.include_example and example_payloads: sections.append("Example valid WorldSpec JSON:\n" + example_payloads[0]) sections.append( "Final answer requirements:\n" "- JSON object only.\n" "- No markdown.\n" "- No comments.\n" "- No Python code.\n" "- No prose before or after the JSON." ) user_prompt = "\n\n".join(sections) system_prompt = self.config.system_prompt return PromptBundle( system_prompt=system_prompt, user_prompt=user_prompt, messages=( {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ), metadata={ "response_format": self.config.resolved_response_format().value, "default_steps": self.config.default_steps, "max_agents_hint": self.config.max_agents_hint, "max_resources_hint": self.config.max_resources_hint, }, ) def _example_payloads( self, examples: Sequence[WorldSpec | Mapping[str, Any] | str] | None, ) -> tuple[str, ...]: """Return serialized few-shot examples.""" if examples is None: return (json.dumps(DEFAULT_MINIMAL_EXAMPLE, indent=2),) payloads: list[str] = [] for example in examples: if isinstance(example, WorldSpec): payloads.append(example.to_json_string(indent=2, exclude_none=True)) elif isinstance(example, Mapping): payloads.append(json.dumps(example, indent=2)) elif isinstance(example, str): payloads.append(example.strip()) return tuple(payload for payload in payloads if payload) def _call_client( self, prompt_bundle: PromptBundle, *, model_kwargs: Mapping[str, Any] | None = None, ) -> str: """Call the configured model client and normalize its response to text.""" if self.client is None: raise ModelClientError("No model client is configured") kwargs = { **copy.deepcopy(dict(self.config.model_kwargs)), **copy.deepcopy(dict(model_kwargs or {})), } callable_obj, call_style = self._resolve_client_callable(self.client) call_kwargs = { "prompt": prompt_bundle.text, "system_prompt": prompt_bundle.system_prompt, "user_prompt": prompt_bundle.user_prompt, "messages": [dict(message) for message in prompt_bundle.messages], "response_format": {"type": "json_object"}, **kwargs, } if call_style == "chat": call_kwargs.setdefault("prompt", prompt_bundle.user_prompt) response = _call_with_supported_kwargs(callable_obj, call_kwargs) return normalize_model_response(response) def _repair_response( self, *, prompt_bundle: PromptBundle, bad_response: str, error: BaseException, model_kwargs: Mapping[str, Any] | None, ) -> str: """Ask the model client to repair malformed JSON.""" repair_user_prompt = ( "The previous response was not valid WorldSmithAI JSON.\n\n" f"Parser error:\n{error}\n\n" "Previous response:\n" f"{bad_response[:8000]}\n\n" "Return only a corrected WorldSpec JSON object. No markdown. No prose." ) repair_bundle = PromptBundle( system_prompt=prompt_bundle.system_prompt, user_prompt=repair_user_prompt, messages=( {"role": "system", "content": prompt_bundle.system_prompt}, {"role": "user", "content": repair_user_prompt}, ), metadata={"repair": True}, ) return self._call_client(repair_bundle, model_kwargs=model_kwargs) @staticmethod def _resolve_client_callable(client: Any) -> tuple[Callable[..., Any], str]: """Resolve a callable generation method from a model client.""" for method_name, style in ( ("generate", "generate"), ("complete", "complete"), ("chat", "chat"), ): method = getattr(client, method_name, None) if callable(method): return method, style if callable(client): return client, "callable" raise ModelClientError( "Model client must expose generate, complete, chat, or be callable" ) def _semantic_validation(self, spec: WorldSpec) -> ValidationReport | None: """Run optional semantic validation.""" if not self.config.semantic_validation: return None validation_config = self.config.semantic_validation_config if validation_config is None: validation_config = ValidationConfig( require_known_behaviors=True, require_known_policies=True, unknown_registry_item_severity=ValidationSeverity.WARNING, constructor_param_severity=ValidationSeverity.WARNING, unresolved_reference_severity=ValidationSeverity.WARNING, ) return validate_world_spec(spec, config=validation_config) def _fallback_result( self, prompt: str, *, constraints: Mapping[str, Any] | str | None, reason: str, error: BaseException | None = None, raw_response: str | None = None, prompt_bundle: PromptBundle | None = None, ) -> WorldGenerationResult: """Build a deterministic fallback result.""" builder = DeterministicWorldSpecBuilder( default_steps=self.config.default_steps, default_seed=self.config.default_seed, max_agents=min(max(1, int(self.config.max_agents_hint)), 6), ) spec = builder.build(prompt, constraints=constraints) validation_report = self._semantic_validation(spec) raw = raw_response or spec.to_json_string(indent=2, exclude_none=True) return WorldGenerationResult( spec=spec, raw_response=raw, parse_result=None, validation_report=validation_report, mode=GenerationMode.FALLBACK, prompt_bundle=prompt_bundle, metadata={ **copy.deepcopy(dict(self.config.metadata)), "fallback_reason": reason, "error": None if error is None else f"{error.__class__.__name__}: {error}", }, ) def generate_world_spec( prompt: str, *, client: Any | None = None, constraints: Mapping[str, Any] | str | None = None, config: WorldGenerationConfig | None = None, examples: Sequence[WorldSpec | Mapping[str, Any] | str] | None = None, model_kwargs: Mapping[str, Any] | None = None, ) -> WorldSpec: """Convenience function that returns a generated ``WorldSpec``. This is the simplest function to call from root-level ``app.py``. """ generator = WorldGenerator(client=client, config=config or WorldGenerationConfig()) return generator.generate_spec( prompt, constraints=constraints, examples=examples, model_kwargs=model_kwargs, ) def generate_world_json( prompt: str, *, client: Any | None = None, constraints: Mapping[str, Any] | str | None = None, config: WorldGenerationConfig | None = None, examples: Sequence[WorldSpec | Mapping[str, Any] | str] | None = None, model_kwargs: Mapping[str, Any] | None = None, ) -> str: """Convenience function that returns generated WorldSpec JSON.""" generator = WorldGenerator(client=client, config=config or WorldGenerationConfig()) return generator.generate_json( prompt, constraints=constraints, examples=examples, model_kwargs=model_kwargs, ) def generate_world_result( prompt: str, *, client: Any | None = None, constraints: Mapping[str, Any] | str | None = None, config: WorldGenerationConfig | None = None, examples: Sequence[WorldSpec | Mapping[str, Any] | str] | None = None, model_kwargs: Mapping[str, Any] | None = None, ) -> WorldGenerationResult: """Convenience function that returns the full generation result.""" generator = WorldGenerator(client=client, config=config or WorldGenerationConfig()) return generator.generate( prompt, constraints=constraints, examples=examples, model_kwargs=model_kwargs, ) def build_world_generation_prompt( prompt: str, *, constraints: Mapping[str, Any] | str | None = None, config: WorldGenerationConfig | None = None, ) -> PromptBundle: """Build the prompt bundle without calling a model.""" generator = WorldGenerator(client=None, config=config or WorldGenerationConfig()) return generator.build_prompt(prompt, constraints=constraints) def normalize_model_response(response: Any) -> str: """Normalize common model-client response shapes into text. Supported shapes: - plain string - bytes - mapping with text/content/generated_text/output - OpenAI-like mapping with choices[0].message.content - object with text/content/generated_text attributes """ if response is None: raise ModelClientError("Model client returned None") if isinstance(response, str): return response.strip() if isinstance(response, bytes): return response.decode("utf-8").strip() if isinstance(response, Mapping): direct = _first_present( response, ("text", "content", "generated_text", "output", "response"), ) if direct is not None: return normalize_model_response(direct) choices = response.get("choices") if isinstance(choices, Sequence) and choices: first_choice = choices[0] if isinstance(first_choice, Mapping): message = first_choice.get("message") if isinstance(message, Mapping) and message.get("content") is not None: return normalize_model_response(message["content"]) if first_choice.get("text") is not None: return normalize_model_response(first_choice["text"]) if response.get("data") is not None: return normalize_model_response(response["data"]) for attribute in ("text", "content", "generated_text", "output", "response"): value = getattr(response, attribute, None) if value is not None: return normalize_model_response(value) raise ModelClientError( f"Could not extract text from model response of type {response.__class__.__name__}" ) def world_spec_schema_excerpt() -> str: """Return a compact schema guide suitable for small-model prompts.""" excerpt = { "WorldSpec": { "required": ["id", "name", "agents", "resources", "events"], "fields": { "schema_version": "string", "id": "string", "name": "string", "description": "string|null", "simulation": "SimulationSpec", "space": "SpaceSpec|null", "agents": "AgentSpec[]", "resources": "ResourceSpec[]", "events": "EventSpec[]", "metrics": "MetricSpec[]", "metadata": "object", }, }, "AgentSpec": { "fields": { "id": "string", "type": "string", "position": "number[]|null", "state": "object", "memory": "object", "goals": "JSON", "behaviors": "BehaviorSpec[]", "policy": "PolicySpec|null", "alive": "boolean", "metadata": "object", } }, "BehaviorSpec": {"fields": {"name": "string", "params": "object"}}, "PolicySpec": {"fields": {"type": "string", "params": "object"}}, "ResourceSpec": { "fields": { "id": "string", "type": "string", "amount": "number", "position": "number[]|null", "regeneration_rate": "number", "max_amount": "number|null", "metadata": "object", } }, "EventSpec": { "fields": { "id": "string|null", "name": "string", "trigger_step": "integer", "payload": "object", "enabled": "boolean", } }, } return json.dumps(excerpt, indent=2) def extract_json_object_text(text: str) -> str: """Extract the first balanced JSON object from text. This is exposed as a utility for app-level debugging. The parser already uses equivalent extraction internally. """ parser = WorldDSLParser() return parser.extract_json_text(text) def _call_with_supported_kwargs(callable_obj: Callable[..., Any], kwargs: Mapping[str, Any]) -> Any: """Call a function using only kwargs supported by its signature. If the callable accepts ``**kwargs``, all supplied kwargs are passed. Otherwise unsupported kwargs are filtered out. If no matching keyword is accepted but a positional prompt appears possible, the flattened prompt is passed positionally. """ 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_params = [ parameter for parameter in parameters.values() if parameter.kind in { inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD, } ] if positional_params and "prompt" in kwargs: return callable_obj(kwargs["prompt"]) return callable_obj() def _first_present(mapping: Mapping[str, Any], keys: Sequence[str]) -> Any: """Return first present non-null mapping value.""" for key in keys: if key in mapping and mapping[key] is not None: return mapping[key] return None def _serialize_constraints(constraints: Mapping[str, Any] | str) -> str: """Serialize UI constraints for prompt inclusion.""" if isinstance(constraints, str): return constraints.strip() return json.dumps(constraints, indent=2, sort_keys=True) def _require_prompt(prompt: str) -> str: """Validate and normalize a user prompt.""" normalized = str(prompt).strip() if not normalized: raise WorldGenerationError("World generation prompt must not be empty") return normalized def _slugify(value: str) -> str: """Return a stable lowercase identifier.""" lowered = value.lower().strip() slug = re.sub(r"[^a-z0-9]+", "_", lowered) slug = re.sub(r"_+", "_", slug).strip("_") return slug or "world" def _title_from_domain(domain: str) -> str: """Return a human-readable fallback world title.""" words = domain.replace("_", " ").split() return " ".join(word.capitalize() for word in words) + " World" __all__ = [ "DEFAULT_MINIMAL_EXAMPLE", "DEFAULT_SYSTEM_PROMPT", "WORLD_DSL_FORMAT_GUIDE", "DeterministicWorldSpecBuilder", "GenerationMode", "ModelClientError", "PromptBundle", "ResponseFormat", "SupportsGenerate", "WorldGenerationConfig", "WorldGenerationError", "WorldGenerationResult", "WorldGenerationValidationError", "WorldGenerator", "build_world_generation_prompt", "extract_json_object_text", "generate_world_json", "generate_world_result", "generate_world_spec", "normalize_model_response", "world_spec_schema_excerpt", ]