ワールドモデルと予測的プランニング
行動前に将来の状態をシミュレーションする、AI のメンタルシミュレーションを扱います。
「ワールドモデルと予測的プランニング」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
世界モデルとは
世界モデルとは、環境に関するエージェント内部の表現です。どのようなオブジェクトが存在するか、それらの状態はどうなっているか、どのようなルールに従って変化するか、エージェントがそれらに影響を与えるために何ができるかを表します。エージェントによる世界のメンタルシミュレーションです。
世界モデルがなければ、エージェントは観測に反応することしかできません。世界モデルがあれば、予測と計画が可能になります。
世界状態の表現
世界状態とは、ある時点における環境に関するすべての関連事実を切り取ったスナップショットです。IoTエージェントの場合、現在のセンサー読み取り値やデバイスの状態などが該当します。ソフトウェアエージェントの場合は、管理対象となるファイル、API、データベースの状態などが該当します。
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())状態遷移モデル
状態遷移モデルは、アクションに応じて世界状態がどのように変化するかを定義します。next_state = transition(current_state, action)実際にアクションを実行せずに、アクションを実行した後の世界がどのようになるかをエージェントが予測できるようにします。
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'))先読み計画
先読み計画では、アクションを実行に移す前に、将来のNステップをシミュレーションします。エージェントは複数のアクション系列を探索し、それぞれの結果として得られる状態を評価して、予測される結果が最良となる系列を選択します。
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
形式化が難しい複雑な環境では、LLM自体を世界モデルとして利用できます。エージェントが特定のアクションを実行した場合に何が起きるかを予測するようLLMに指示します。これは精度と一般性のトレードオフです。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"]}')世界モデルにおける不確実性
世界モデルは近似にすぎません。単一の点推定ではなく、可能性のある状態に対する確率分布を保持することで、不確実性に対処します。不確実性が高い場合は、慎重なアクションを取るか、行動する前により多くの情報を収集します。
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()}')世界モデルの更新
各アクションの後、エージェントは実際の結果を観測し、世界モデルを更新します。予測が間違っていた場合は、より正確になるようにモデルを更新します。これは、モデルベースのエージェント学習を促進するobserve-update-planサイクルです。
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ステップのシミュレーション
自然言語の領域では、提案した計画を実行する前に、LLMに3ステップをシミュレーションさせます。LLMは各ステップで予測される結果を評価し、潜在的な問題を指摘します。これにより、予測結果が悪いアクションをエージェントが実行するのを防ぎます。
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世界モデルを信頼できるタイミング
世界モデルの品質は、遷移関数と初期状態の正確さに左右されます。急速に変化する環境では、モデルはすぐに古くなります。陳腐化のしきい値を実装してください。モデルがN秒間更新されておらず、かつ行動のリスクが高い場合は、まず環境を再観測します。
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)反実仮想推論
世界モデルを使うと、反実仮想推論が可能になります。「行動Bではなく行動Aを取っていたら、何が起きていたでしょうか？」これは、同じ失敗を繰り返さずにそこから学ぶための強力な手段です。エージェントは反実仮想をシミュレートし、自分の選択がなぜ最適ではない結果につながったのかを理解できます。
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)理解度チェック
モデルベースエージェントがモデルフリーエージェントより優れている主な点は何でしょうか？
復習：世界モデルと予測的プランニング
よくできました！重要なポイント：
- 世界の状態：ある時点における、関連するすべてのエンティティ属性のスナップショット
- 遷移モデル：
next_state = transition(state, action)— 影響を予測する - 先読み：Nステップ先までシミュレートし、結果を評価して、最適な行動を選ぶ
- 世界モデルとしてのLLM：形式化が難しい領域で有効
- 不確実性：一点の推定値ではなく分布を維持し、不確実性が高い場合の行動を避ける
- 陳腐化：モデルが古い場合は、リスクの高い行動の前に再観測する
次は、アラインメントの課題です。本当に有益な自律エージェントの構築が難しい理由を見ていきます。
よくある質問
「ワールドモデルと予測的プランニング」レッスンは無料ですか？
はい。「ワールドモデルと予測的プランニング」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「ワールドモデルと予測的プランニング」で何を学びますか？
行動前に将来の状態をシミュレーションする、AI のメンタルシミュレーションを扱います。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。
「ワールドモデルと予測的プランニング」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- アシスタントから自律エージェントへ
- ワールドモデルと予測的プランニング
- 自律エージェントにおけるアライメントの課題
- 研究の最前線：AGI とその先