Модели мира и прогнозирующее планирование

Что такое модель мира

Представление состояния мира

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any

@dataclass
class WorldState:
    timestamp: str = ''
    entities: dict = field(default_factory=dict)  # entity_id -> attributes
    relationships: list = field(default_factory=list)
    agent_position: str = 'idle'
    pending_actions: list = field(default_factory=list)

    def __post_init__(self):
        if not self.timestamp:
            self.timestamp = datetime.utcnow().isoformat()

    def update_entity(self, entity_id: str, attributes: dict):
        current = self.entities.get(entity_id, {})
        self.entities[entity_id] = {**current, **attributes}

    def snapshot(self) -> dict:
        return {
            'timestamp': self.timestamp,
            'entities': dict(self.entities),
            'agent_position': self.agent_position
        }

# Example:
world = WorldState()
world.update_entity('server_room', {'temp_c': 22.5, 'humidity': 55})
world.update_entity('hvac_unit_1', {'status': 'off', 'setpoint': 22})
print(world.snapshot())

Модель переходов

import copy

def transition(state: WorldState, action: dict) -> WorldState:
    """
    Predict next state given current state and action.
    Returns a new WorldState (does not modify original).
    """
    next_state = copy.deepcopy(state)

    action_type = action.get('type')
    target = action.get('target')
    params = action.get('params', {})

    if action_type == 'TURN_ON_HVAC':
        next_state.update_entity(target, {
            'status': 'on',
            'setpoint': params.get('setpoint', 22)
        })
        # Simulate temperature effect (simplified linear model)
        room = action.get('room', 'server_room')
        current_temp = next_state.entities.get(room, {}).get('temp_c', 25)
        next_state.update_entity(room, {
            'temp_c': max(current_temp - 2, params.get('setpoint', 22))
        })

    elif action_type == 'SEND_ALERT':
        next_state.pending_actions.append({'alert_sent': True})

    return next_state

# Predict effect of turning on HVAC:
action = {'type': 'TURN_ON_HVAC', 'target': 'hvac_unit_1',
          'room': 'server_room', 'params': {'setpoint': 20}}
next_world = transition(world, action)
print('After action:', next_world.entities.get('server_room'))

Планирование Lookahead

def evaluate_state(state: WorldState) -> float:
    """
    Compute a scalar utility score for a world state.
    Higher = better for the agent's goals.
    """
    score = 1.0
    server_room = state.entities.get('server_room', {})
    temp = server_room.get('temp_c', 25)
    # Penalty for high temperature
    if temp > 30:
        score -= (temp - 30) * 0.2
    # Penalty for very low temperature (over-cooling)
    if temp < 18:
        score -= (18 - temp) * 0.1
    return max(0.0, score)

def lookahead(state: WorldState, possible_actions: list, depth: int = 3) -> dict:
    """Evaluate top-level actions by simulating depth steps ahead."""
    best_action = None
    best_score = -float('inf')

    for action in possible_actions:
        simulated = transition(state, action)
        # Simulate depth more steps (simplified: no branching)
        for _ in range(depth - 1):
            simulated = transition(simulated, {'type': 'NOOP'})
        score = evaluate_state(simulated)
        if score > best_score:
            best_score = score
            best_action = action

    return {'best_action': best_action, 'expected_score': round(best_score, 3)}

LLM как модель мира

import anthropic
import json

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

def llm_predict(
    current_state: dict,
    proposed_action: dict,
    domain_description: str = ''
) -> dict:
    prompt = (
        f'Domain: {domain_description}\n\n'
        f'Current state: {json.dumps(current_state, indent=2)}\n\n'
        f'Proposed action: {json.dumps(proposed_action, indent=2)}\n\n'
        'Predict the next world state after this action. '
        'Consider direct effects and second-order effects.\n'
        'Return JSON: {"predicted_state": dict, '
        '"confidence": float, "side_effects": [str]}'
    )
    response = client.messages.create(
        model='claude-opus-4-5', max_tokens=512,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return json.loads(response.content[0].text)

Планирование на основе модели и без модели

# Comparison table
COMPARISON = {
    'model_based': {
        'pros': [
            'Can plan before acting (lookahead)',
            'Works well with limited data',
            'Interpretable — you can inspect the world model',
            'Can simulate counterfactuals'
        ],
        'cons': [
            'Model can be wrong (model mismatch)',
            'Complex to build for arbitrary domains',
            'Planning is computationally expensive'
        ],
        'examples': 'AlphaGo, MCTS, agent with explicit WorldState'
    },
    'model_free': {
        'pros': [
            'No need to model the world explicitly',
            'Learns from raw experience',
            'Simpler architecture'
        ],
        'cons': [
            'Requires many interactions to learn',
            'Poor generalisation to new scenarios',
            'Cannot reason about unseen situations'
        ],
        'examples': 'Q-learning, policy gradient, basic RLHF'
    }
}

for approach, info in COMPARISON.items():
    print(f'{approach}: {info["examples"]}')

Uncertainty в моделях мира

from statistics import mean, stdev

@dataclass
class UncertainState:
    entity_id: str
    attribute: str
    samples: list  # multiple estimates

    def mean(self) -> float:
        return mean(self.samples) if self.samples else 0.0

    def uncertainty(self) -> float:
        if len(self.samples) < 2:
            return float('inf')
        return stdev(self.samples)

    def should_gather_more_data(self, threshold: float = 2.0) -> bool:
        return self.uncertainty() > threshold

# Example: uncertain temperature estimate from multiple sensors
temp_state = UncertainState(
    entity_id='server_room',
    attribute='temp_c',
    samples=[22.1, 22.8, 21.9, 34.5, 22.3]  # one outlier
)
print(f'Mean temp: {temp_state.mean():.1f}C')
print(f'Uncertainty (stdev): {temp_state.uncertainty():.2f}')
print(f'Gather more data: {temp_state.should_gather_more_data()}')

Обновление модели мира

class WorldModelAgent:
    def __init__(self):
        self.world = WorldState()
        self.prediction_errors = []

    def act(self, action: dict) -> dict:
        # 1. Predict next state
        predicted = transition(self.world, action)

        # 2. Take the action in the real world
        actual_observation = self._execute_action(action)

        # 3. Measure prediction error
        for entity_id, attrs in actual_observation.items():
            for attr, actual_val in attrs.items():
                predicted_val = predicted.entities.get(entity_id, {}).get(attr)
                if predicted_val is not None:
                    error = abs(actual_val - predicted_val)
                    self.prediction_errors.append(error)

        # 4. Update world model with actual observation
        for entity_id, attrs in actual_observation.items():
            self.world.update_entity(entity_id, attrs)

        return actual_observation

    def _execute_action(self, action: dict) -> dict:
        # Placeholder — in production this calls real APIs/sensors
        return {'server_room': {'temp_c': 23.0}}

    def model_accuracy(self) -> float:
        if not self.prediction_errors:
            return 1.0
        return max(0.0, 1.0 - mean(self.prediction_errors) / 10)

Планирование с LLM: моделирование 3 шагов

import anthropic
import json

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

def simulate_plan_steps(goal: str, proposed_steps: list) -> dict:
    steps_str = json.dumps(proposed_steps, indent=2)
    prompt = (
        f'Goal: {goal}\n\n'
        f'Proposed plan steps:\n{steps_str}\n\n'
        'Simulate executing each step in order. For each step predict:\n'
        '- What changes in the world?\n'
        '- What could go wrong?\n'
        '- Is this step reversible?\n\n'
        'Return JSON: {"step_simulations": [{"step": int, '
        '"predicted_outcome": str, "risks": [str], "reversible": bool}], '
        '"plan_safe": bool, "recommendation": str}'
    )
    response = client.messages.create(
        model='claude-opus-4-5', max_tokens=1024,
        messages=[{'role': 'user', 'content': prompt}]
    )
    result = json.loads(response.content[0].text)
    if not result['plan_safe']:
        print(f'Plan safety concern: {result["recommendation"]}')
    return result

Когда доверять модели мира

from datetime import datetime, timedelta

class StaleAwareWorldModel:
    def __init__(self, max_age_seconds: int = 60):
        self.state = WorldState()
        self.max_age = timedelta(seconds=max_age_seconds)
        self.last_observation_time = datetime.utcnow()

    def is_stale(self) -> bool:
        age = datetime.utcnow() - self.last_observation_time
        return age > self.max_age

    def refresh(self, observation_fn):
        new_observations = observation_fn()
        for entity_id, attrs in new_observations.items():
            self.state.update_entity(entity_id, attrs)
        self.last_observation_time = datetime.utcnow()
        print(f'World model refreshed')

    def plan_action(self, action: dict, observation_fn) -> dict:
        if self.is_stale():
            print('World model stale — refreshing before planning')
            self.refresh(observation_fn)
        return lookahead(self.state, [action], depth=3)

Контрфактическое рассуждение

import anthropic
import json

client = anthropic.Anthropic(api_key='YOUR_API_KEY')

def counterfactual_analysis(
    actual_state_before: dict,
    action_taken: dict,
    actual_outcome: dict,
    alternative_action: dict
) -> dict:
    prompt = (
        'Analyse a counterfactual scenario:\n\n'
        f'Initial state: {json.dumps(actual_state_before, indent=2)}\n'
        f'Action taken: {json.dumps(action_taken, indent=2)}\n'
        f'Actual outcome: {json.dumps(actual_outcome, indent=2)}\n\n'
        f'Alternative action considered: {json.dumps(alternative_action, indent=2)}\n\n'
        'What would likely have happened with the alternative action?\n'
        'Return JSON: {"counterfactual_outcome": str, '
        '"better_choice": str, "lesson": str}'
    )
    response = client.messages.create(
        model='claude-opus-4-5', max_tokens=512,
        messages=[{'role': 'user', 'content': prompt}]
    )
    return json.loads(response.content[0].text)

Проверка знаний

Итоги: модели мира и прогнозирующее планирование