はじめに
店舗の棚の並びを変えると、売上は変わるのか。どの通路が混雑しているのか。
実際に改装してみるまでわからない——そんな問題を、AIエージェントに仮想の顧客として動いてもらうことで事前に検証できないか、と思い作成してみました。
今回はlocalLLMとclaude APIを使って実装しました。
完成したもの
- フロアエディタ: 棚・レジ・入口などを自由に配置
- リアルタイムシミュレーション: 顧客エージェントが自律的に動くアニメーション
- 混雑ヒートマップ: ガウシアンスムージングによるエリア別混雑度の可視化
- LLMレポート生成: シミュレーション後にClaude APIが日本語で改善提案を自動作成
技術スタック
| レイヤー | 使用技術 |
|---|---|
| フロントエンド | Vue.js 3 + Vite + HTML Canvas + Socket.IO + Chart.js |
| バックエンド | Python 3.11 + Flask 3 + Flask-SocketIO |
| シミュレーション | NumPy + scipy（ヒートマップ） |
| AI | Claude API（claude-haiku-4-5）— ペルソナ生成・レポート生成 |
| インフラ | Docker Compose |
環境
今回の実装・検証を行った環境は以下の通りです。MacBook Proのローカル環境でLLMを動かしつつ、シミュレーションを実行しています。
| 項目 | バージョン / 内容 |
|---|---|
| OS | Tahoe 26.4 |
| マシン | MacBook Pro (Apple M5 Pro / 48GB RAM) |
| Python | 3.14.4 |
| Node.js | v22.22.2 |
| Vue.js | 3.4.0 |
| Vite | 5.2.0 |
| Local LLM | Ollama / Llama 3.3 (70B) |
| Cloud LLM | Claude 4.5 (Anthropic API) |
システム設計
ブラウザ (Vue.js)
├── FloorEditor ─── POST /api/floor ──────→ レイアウト保存・読込
├── SimulationCanvas ←── WebSocket ─────── フレームデータ受信（30fps）
├── TimeControls ─── POST /api/simulation/{id}/control → 速度変更・スキップ
└── ReportView ───── POST /api/report/{id} ──────────→ LLMレポート取得
Flask バックエンド
├── REST API (/api/floor, /api/simulation, /api/report)
├── Flask-SocketIO (/socket.io) ─── フレームpush
└── Simulation Engine（バックグラウンドスレッド）
├── FloorMap グリッドマップ管理
├── CustomerAgent × 300 A* + 混雑速度モデル
├── CongestionGrid リアルタイム密度管理
├── CongestionAnalyzer（NumPy + scipy）ヒートマップ
└── Claude API ペルソナ生成・レポート作成
実装のポイント
1. 顧客エージェントの状態機械
各顧客エージェントは以下の状態機械で動作します。状態の列挙型（Enum）で行動フェーズを明確化しました。
class AgentState(str, Enum):
ENTERING = "entering"
THINKING = "thinking" # LLM思考中
SHOPPING = "shopping" # 次の目標を探す
MOVING_TO_ITEM = "moving_to_item" # 棚へ移動中
PICKING_ITEM = "picking_item" # 商品取得（滞留）
MOVING_TO_REGISTER = "moving_to_register"
WAITING_REGISTER = "waiting_register"
PAYING = "paying"
EXITING = "exiting"
DONE = "done"
step() メソッド内で現在の状態に応じた処理を呼び分けるシンプルな構造です。
def step(self, floor_map, register_queues, congestion_grid, all_agents):
self.total_ticks += 1
if self.state == AgentState.SHOPPING:
self._choose_next_item(floor_map)
elif self.state == AgentState.MOVING_TO_ITEM:
done = self._follow_path(congestion_grid, all_agents)
if done:
self.state = AgentState.PICKING_ITEM
self.dwell_ticks = random.randint(*self.dwell_time_range)
elif self.state == AgentState.PICKING_ITEM:
self.dwell_ticks -= 1
if self.dwell_ticks <= 0:
if self.remaining_items:
self.remaining_items.pop(0)
self.state = AgentState.SHOPPING
# ...（以下、PAYING, EXITINGなど）
2. A*経路探索の実装
棚を避けながら目的のコーナーまで最短経路で歩くという機能は、グリッドマップ上のA*アルゴリズムで実装した。Pythonの heapq を使った実装で、マンハッタン距離を使用しました。
def astar(floor_map, start, goal, max_iterations=5000):
"""棚・壁を避けてstartからgoalまでの最短経路を返す"""
open_heap = []
heapq.heappush(open_heap, (0.0, start))
came_from = {}
g_score = {start: 0.0}
f_score = {start: heuristic(start, goal)}
iterations = 0
while open_heap and iterations < max_iterations:
iterations += 1
_, current = heapq.heappop(open_heap)
if current == goal:
# パスを再構築して返す
path = []
while current in came_from:
path.append(current)
current = came_from[current]
path.reverse()
return path
cx, cy = current
for dx, dy in ((0, 1), (0, -1), (1, 0), (-1, 0)):
nx, ny = cx + dx, cy + dy
neighbor = (nx, ny)
if not floor_map.is_walkable(nx, ny) and neighbor != goal:
continue # 壁・棚はスキップ
tentative_g = g_score[current] + 1.0
if tentative_g < g_score.get(neighbor, float("inf")):
came_from[neighbor] = current
g_score[neighbor] = tentative_g
f_score[neighbor] = tentative_g + heuristic(neighbor, goal)
heapq.heappush(open_heap, (f_score[neighbor], neighbor))
return None # 到達不能
棚は移動できないため is_walkable() で除外しますが、棚の隣のセルをアプローチ可能点として検索する補助関数も合わせて実装した。
def find_adjacent_walkable(floor_map, target):
"""棚・レジに隣接する通路セルを返す（到達不能防止）"""
tx, ty = target
for dx, dy in ((0, 1), (0, -1), (1, 0), (-1, 0)):
nx, ny = tx + dx, ty + dy
if floor_map.is_walkable(nx, ny):
return (nx, ny)
return None
3. 混雑による速度低下モデル
混雑の再現を少し悩みました。
今回はグリッドセルの人口密度に基づいて速度に倍率をかけることで混雑時の移動速度を調整しました。
class CongestionGrid:
def speed_factor(self, x: int, y: int) -> float:
"""密度に応じた速度倍率を返す（1.0 = 通常、0.3 = 最低速度）"""
density = self._density[y][x]
# 密度0 → 速度100%、密度5人以上 → 速度30%（最低）
return max(0.3, 1.0 - density * 0.14)
移動処理でこの倍率を乗算するだけで、レジ前での通路の詰まりが再現される。
def _follow_path(self, congestion_grid, all_agents):
tx, ty = self.path[self.path_index]
dx = tx - self.x
dy = ty - self.y
dist = (dx**2 + dy**2) ** 0.5
# 混雑度に応じて速度を動的に調整
speed = self.speed * congestion_grid.speed_factor(int(self.x), int(self.y))
if dist <= speed:
self.x, self.y = float(tx), float(ty)
self.path_index += 1
else:
self.x += (dx / dist) * speed
self.y += (dy / dist) * speed
4. 混雑ヒートマップ（NumPy + scipy）
エリアごとの混雑度を可視化するために、ガウシアンフィルタを使っています。単純な密度カウントをぼかして、どこが通りにくいかがわかるヒートマップになります。
import numpy as np
from scipy.ndimage import gaussian_filter
def compute_heatmap(density_flat, width, height, sigma=1.5):
"""密度グリッドをガウシアンぼかしして0〜1に正規化したヒートマップを返す"""
arr = np.array(density_flat, dtype=float).reshape(height, width)
smoothed = gaussian_filter(arr, sigma=sigma)
max_val = smoothed.max()
if max_val > 0:
smoothed /= max_val # 最大値で正規化
return smoothed.flatten().tolist()
このフラットリストをそのまま WebSocket でフロントエンドに送り、Canvas 上でセルごとに rgba(255, 0, 0, 値) で塗り重ねることでヒートマップのオーバーレイを実現しました。
5. ペルソナ生成
LLMにエージェントの個性を持たせる仕組みです。
各エージェントは6種類のペルソナタイプのいずれかに属します。
| ペルソナ | 特徴 | 移動速度 |
|---|---|---|
| 家族連れ (family) | まとめ買い・広範囲を回遊 | 0.7（ゆっくり） |
| 単身 (single) | 目的買い・惣菜・飲料中心 | 1.2 |
| シニア (senior) | じっくり選ぶ | 0.5（最も遅い） |
| 会社員 (office) | 惣菜・飲料を素早く | 1.3（最も速い） |
| 節約型 (budget) | 特売・割引コーナーを巡る | 1.0 |
| 主婦 (housewife) | 食材中心・効率的に買い出し | 0.9 |
ペルソナに応じた買い物リストは、キーワードマッチングで動的に生成します。
6. 自律的なルート計画
ローカルLLMがどのルートで買い物するか決めています。
プロンプトエンジニアリング
エージェントが入店した瞬間、バックグラウンドで以下のようなプロンプトを生成し、AIに問いかけます。
あなたはスーパーマーケットに入店したばかりの{persona}です。
顧客タイプ: {常連客 / 新規客}
店内の配置とコーナー一覧: {topology}
【あなたの思考プロセス】
1. あなたのペルソナに合わせて、今日何を買うか（数品程度）を想像してください。
2. その商品を買うために行くべきコーナーを選んでください。
3. それらのコーナーを回る順番を計画してください。基本的には入口付近（青果など）から回るのが自然です。
【出力ルール】
- コーナー名をカンマ区切りで出力すること（例: 青果, 精肉, レジ）
回答のパース処理
AIからの「青果, 精肉, レジ」といったカンマ区切りのテキスト回答を、プログラムで利用可能な内部IDのリストへと変換します。
# AIの回答をパースして、存在するゾーン名とマッチング
raw_zones = [z.strip() for z in resp.split(",")]
final_route = []
for rz in raw_zones:
for az in available_zones:
if az in rz: # 部分一致で柔軟にマッチング
final_route.append(az)
break
# 計画したルートをエージェントにセット
agent.remaining_items = dedup_route
7. LLMレポート生成（Claude API）
シミュレーション完了後は、収集したデータを元に Claude API がコンサルレポートを自動生成します。
プロンプトにはシミュレーションのログ（ペルソナ別来客数・平均滞在時間・エリア別混雑スコア・レジ待ち行列長）を詰め込みます。
prompt = f"""あなたはスーパーマーケットのレイアウトコンサルタントです。
以下のシミュレーション結果を分析し、日本語でレポートを作成してください。
## シミュレーション概要
- 総顧客数: {total_agents}人
- シミュレーション時間: {total_ticks // 60}分
- グリッドサイズ: {width}×{height}（約{width * height}㎡）
## エリア別混雑スコア（0〜1、ピーク値）
{zone_scores_str}
## レジ情報
- レジ台数: {register_count}台
- シミュレーション中の最大待ち行列長: {max_queue}人
以下の構成でレポートを作成してください:
1. 総評（2〜3文）
2. 混雑エリアTOP3と原因分析
3. レジ待ち評価
4. 推奨レイアウト改善施策（具体的に3つ）
5. 改善後の期待効果"""
text = chat([{"role": "user", "content": prompt}], max_tokens=3000)
まとめ
今回はclaude APIとlacal LLMを使った店舗レイアウトシミュレーターを作ってみました。
短時間の作成だったので、まだまだ改善の余地はありますが、local LLMでここまで面白いものが作れるのかと感動しました。
local LLMの限界値として同時に10人ぐらいが最大値であった。
ペルソナを設計してそこからAIにルートを考えさせたことでよりリアルな動きになったのではないかと思いました。