食に対する個人の嗜好は、「珍しいものが食べたい」「調理が簡単なものが良い」など、非常に多様かつ動的である。従来の推薦システムは、栄養バランスやコストといった静的な指標に基づいて献立を提案するものが主であった。本研究では、ユーザーからのフィードバック(満足度評価)を元に、強化学習(多腕バンディットアルゴリズム)を用いて「ユーザーがどのようなタイプの献立を好むか」という提案戦略を学習し、その戦略に基づいて多目的最適化(遺伝的アルゴリズム)が献立を生成する、パーソナライズされた動的な献立推薦システムのコアエンジンを構築・検証することを目的とする。
本システムは、強化学習エージェント(バンディット)と多目的最適化エンジン(遺伝的アルゴリズム)を連携させた、クローズドループのシミュレーション環境である。全体の処理フローは以下の通り。
1. 戦略決定: `bandit_logic.py` が過去の学習記録 `mab_feedback.csv` を参照し、今回最適化すべきUX戦略(腕)を決定する。
2. 献立生成: `2献立作成(GraphicalRecipes).py` が、決定された戦略に基づき多目的最適化を実行し、献立候補群(パレート解)を生成する。
3. 評価: `run_experiment.py` が生成された献立の一つをランダムに選択し、`virtual_user.py` に渡して評価させる。
4. フィードバック: `virtual_user.py` は自身の隠れた好みに基づいて満足度(報酬)を計算し、返す。
5. 学習記録: `run_experiment.py` は、(1)で選択された戦略と、(4)で得られた報酬のペアを `mab_feedback.csv` に追記する。
6. 上記1~5を規定回数繰り返し、学習を行う。
現在のシステムは、`virtual_user.py` を用いた自動シミュレーション環境だが、これを実際のサービスとして人間が利用する場合、以下のようなフローが想定される。 1. ユーザーのログイン: ユーザーがシステムにログインする。ユーザーごとに過去の評価履歴が管理される。 2. 戦略決定: `bandit_logic.py` が、そのユーザーの過去の評価履歴(`mab_feedback.csv`に相当)を読み込み、「今日のあなたへのおすすめ方針」として最適な腕(例:腕3「調理しやすさ重視」)を選択する。 3. 献立生成: `2献立作成(GraphicalRecipes).py` が、選択された戦略に基づいて、複数の優れた献立候補(パレート解)を生成する。 4. 献立の提示: `server1(GraphicalRecipes).py`が起動したWebアプリケーションが、生成された複数の献立候補をユーザーに提示する。ユーザーは気分や状況に合わせて、その中から一つを選ぶ。 5. 調理と食事: ユーザーは選んだ献立を実際に調理し、食事をする。 6. 満足度の評価: 後日、ユーザーはWebアプリケーション上で、前回の献立に対する総合的な満足度を1~5の星などで評価する。 7. フィードバックの記録: `server1(GraphicalRecipes).py`は、そのユーザーの評価(報酬)を、「どの戦略で提案したか」という情報と紐づけて、そのユーザーの学習データとして記録する。
| 扱うデータ | 用途 | ファイル名 | ファイルの場所 |
| システム制御 | シミュレーション全体の制御、各モジュールの呼び出し、結果の記録 | run_experiment.py | (ルート) |
| システム制御 | 多目的最適化による献立生成、GUIによる手動設定 | 2献立作成(GraphicalRecipes).py | (ルート) |
| Webサーバー | 生成された献立をブラウザで表示するためのWebサーバー機能 | server1(GraphicalRecipes).py | (ルート) |
| システム制御 | UCB1バンディットアルゴリズムによる戦略決定 | bandit_logic.py | (ルート) |
| システム制御 | 仮想ユーザーによる献立評価と報酬計算 | virtual_user.py | (ルート) |
| 入力データ | 各レシピの栄養素・コスト・UXスコアの格納 | recipe_noX.csv | (./data/hyouka/) |
| 設定データ | 手動実行時のユーザー情報やアレルギー設定を保存 | menu_creation_settings.json | (ルート) |
| 出力データ(学習ログ) | 強化学習の試行ごとの結果(腕、報酬)を記録 | mab_feedback.csv | (ルート) |
| 出力データ(献立) | 生成された献立候補群の詳細情報をJSON形式で保存 | all_details.json | (./static/) |
| 出力データ(グラフ) | 遺伝的アルゴリズムのパレート解の分布を可視化 | palate.png | (ルート) |
1.必要なライブラリをインストールする.
pip install pandas numpy pymoo PySimpleGUI japanize-matplotlib flask
2.(任意・初回のみ)ユーザー設定ファイルを作成する.
python 2献立作成(GraphicalRecipes).py
本システムには、大きく分けて2つの実行モードがある。
【A】シミュレーション実験を実行する場合(開発者・研究者向け)
python run_experiment.py
【B】献立推薦システムを実際に利用する場合(一般ユーザー向け)
python 2献立作成(GraphicalRecipes).py --auto2. Webサーバーの起動: 次に、以下のコマンドを実行してWebサーバーを起動する。
python server1(GraphicalRecipes).py3. ブラウザで確認: ターミナルに表示されるURL(例: `http://127.0.0.1:5000`)にウェブブラウザでアクセスし、献立の確認や評価を行う。
import subprocess import os import json import random import csv from datetime import datetime from virtual_user import get_satisfaction import pandas as pd # --- 実験設定 --- NUM_TRIALS = 200 # 実験の繰り返し回数 MAB_FEEDBACK_FILE = 'mab_feedback.csv'
def run_single_trial():
"""1回分の実験(献立作成→評価→学習データ追記)を実行する"""
# 1. 献立作成スクリプトを実行
print("\n--- 献立作成エンジンを実行中... ---")
try:
subprocess.run(['python', '2献立作成(GraphicalRecipes).py', '--auto'], check=True)
except subprocess.CalledProcessError as e:
print(f"エラー: 2献立作成(GraphicalRecipes).py の実行に失敗しました。 - {e}")
return False # 2. 生成された献立候補から1つをランダムに選択
details_path = os.path.join('static', 'all_details.json')
try:
with open(details_path, 'r', encoding='utf-8') as f:
all_candidates = json.load(f)
if not all_candidates:
print("警告: 献立候補が生成されませんでした。この試行をスキップします。")
return True
selected_candidate = random.choice(all_candidates)
all_recipes_in_menu = []
for day in selected_candidate.get('day_recipes', []):
for meal in day.get('meals', []):
all_recipes_in_menu.extend(meal.get('recipes', []))
except (FileNotFoundError, json.JSONDecodeError, IndexError) as e:
print(f"エラー: 生成された詳細ファイルの読み込みに失敗しました。 - {e}")
return False
# 3. 仮想ユーザーが評価し、報酬を計算
try:
with open('last_chosen_arm.txt', 'r', encoding='utf-8') as f:
chosen_arm = int(f.read().strip())
except (FileNotFoundError, ValueError) as e:
print(f"エラー: last_chosen_arm.txt の読み込みに失敗しました。 - {e}")
return False
reward = get_satisfaction(chosen_arm, all_recipes_in_menu)
# 4. 報酬データをmab_feedback.csvに追記
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
file_exists = os.path.exists(MAB_FEEDBACK_FILE)
try:
with open(MAB_FEEDBACK_FILE, 'a', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
if not file_exists:
writer.writerow(['timestamp', 'chosen_arm', 'reward'])
writer.writerow([timestamp, chosen_arm, reward])
print(f"フィードバックを記録しました: arm={chosen_arm}, reward={reward}")
except IOError as e:
print(f"エラー: {MAB_FEEDBACK_FILE} への書き込みに失敗しました。 - {e}")
return False
return True
if __name__ == '__main__':
if os.path.exists(MAB_FEEDBACK_FILE):
os.remove(MAB_FEEDBACK_FILE)
print(f"古い {MAB_FEEDBACK_FILE} を削除し、実験を初期化しました。")
print(f"\n===== {NUM_TRIALS}回のシミュレーション実験を開始します =====")
for i in range(NUM_TRIALS):
print(f"\n---【 試行 {i + 1}/{NUM_TRIALS} 】---")
success = run_single_trial()
if not success:
print("\n重大なエラーが発生したため、実験を中止します。")
break
print(f"\n===== 実験終了 =====")
# (最終結果の集計処理) ...import pandas as pd import numpy as np import os MAB_FEEDBACK_FILE = 'mab_feedback.csv' N_ARMS = 4
def choose_arm():
if not os.path.exists(MAB_FEEDBACK_FILE):
return np.random.randint(N_ARMS)
df = pd.read_csv(MAB_FEEDBACK_FILE)
if df.empty:
return np.random.randint(N_ARMS) total_plays = len(df)
arm_stats = []
for i in range(N_ARMS):
arm_df = df[df['chosen_arm'] == i]
plays = len(arm_df)
if plays == 0:
return i
normalized_rewards = (arm_df['reward'] - 1) / 4.0
avg_reward = normalized_rewards.mean()
arm_stats.append({'plays': plays, 'avg_reward': avg_reward})
ucb_scores = []
for i in range(N_ARMS):
exploration_bonus = np.sqrt((2 * np.log(total_plays)) / arm_stats[i]['plays'])
score = arm_stats[i]['avg_reward'] + exploration_bonus
ucb_scores.append(score)
return np.argmax(ucb_scores)
def get_satisfaction(chosen_arm, selected_menu_details):
# この仮想ユーザーの「真の好み」の重み
true_weights = {
0: 0.15, # q1: 入手のしやすさ
1: 0.05, # q2: 意外性
2: 0.10, # q3: 時間帯
3: 0.70 # q4: 調理しやすさ
} total_score = 0
num_recipes = len(selected_menu_details)
if num_recipes == 0:
return 1 # レシピがなければ最低評価
# 献立に含まれる全レシピのUXスコアを、真の好みで重み付けして合計
for recipe in selected_menu_details:
total_score += true_weights[0] * recipe.get('q1_score', 0)
total_score += true_weights[1] * recipe.get('q2_score', 0)
total_score += true_weights[2] * recipe.get('q3_score', 0)
total_score += true_weights[3] * recipe.get('q4_score', 0)
avg_score = total_score / num_recipes
reward = round(avg_score * 4) + 1
reward = max(1, min(5, int(reward)))
print(f"仮想ユーザー評価: 平均スコア={avg_score:.3f} -> 報酬={reward}")
return reward
# --- 設定項目 --- MAB_FEEDBACK_FILE = 'mab_feedback.csv' N_ARMS = 4 # 腕の数(UX指標の数 q1, q2, q3, q4)
def choose_arm():
"""
多腕バンディット問題(UCB1アルゴリズム)に基づき、次に選択すべき腕を決定する関数。
:return: 選択された腕のインデックス (0, 1, 2, or 3)
"""
# --- 1. 学習データの読み込み ---
if not os.path.exists(MAB_FEEDBACK_FILE):
# まだフィードバックが一件もない場合、ランダムに腕を選択
print("MABフィードバックファイルが存在しないため、ランダムに腕を選択します。")
return np.random.randint(N_ARMS)
df = pd.read_csv(MAB_FEEDBACK_FILE)
if df.empty:
# ファイルは存在するが中身が空の場合も、ランダムに腕を選択
print("MABフィードバックが空のため、ランダムに腕を選択します。")
return np.random.randint(N_ARMS)
# --- 2. 各腕の成績を計算 --- total_plays = len(df) arm_stats = []
for i in range(N_ARMS):
arm_df = df[df['chosen_arm'] == i]
plays = len(arm_df)
if plays == 0:
# まだ一度も選ばれたことがない腕があれば、それを最優先で選択(探索)
print(f"腕{i}が未選択のため、優先的に選択します。")
return i
# 報酬を0-1の範囲に正規化 (元の報酬は1-5なので、(x-1)/4で計算)
normalized_rewards = (arm_df['reward'] - 1) / 4.0
avg_reward = normalized_rewards.mean()
arm_stats.append({'plays': plays, 'avg_reward': avg_reward})
# --- 3. UCB1スコアの計算と腕の選択 ---
ucb_scores = []
for i in range(N_ARMS):
# UCB1スコア = 平均報酬 + √(2 * log(全試行回数) / この腕の試行回数)
exploration_bonus = np.sqrt((2 * np.log(total_plays)) / arm_stats[i]['plays'])
score = arm_stats[i]['avg_reward'] + exploration_bonus
ucb_scores.append(score)
print(f"腕{i}: 平均報酬={arm_stats[i]['avg_reward']:.3f}, UCBスコア={score:.3f}")
# 最もスコアが高い腕を選択
chosen_arm = np.argmax(ucb_scores)
print(f"UCBスコアが最も高い腕 {chosen_arm} を選択しました。")
return chosen_arm
# このファイルが直接実行された場合のテスト用コード if __name__ == '__main__':
# mab_feedback.csv が存在すれば、それに基づいて腕を選択
# 存在しなければ、ランダムに選択される
print("\n--- バンディットアルゴリズムのテスト実行 ---")
selected_arm = choose_arm()
print(f"\nテスト結果: 次に選択すべき腕は {selected_arm} です。")
def get_satisfaction(chosen_arm, selected_menu_details):
"""
仮想ユーザーが献立を評価し、総合満足度(報酬)を計算する関数。
このユーザーは「q4: 調理できそうか」を最も重視する。
:param chosen_arm: バンディットアルゴリズムが選択した腕 (0, 1, 2, or 3)
:param selected_menu_details: ユーザーに提示された献立の詳細情報(辞書のリスト)
:return: 総合満足度 (1-5の整数)
"""
# この仮想ユーザーの「真の好み」の重み
# q4(調理しやすさ)の重みが0.7と最も高く、q1(入手しやすさ)も少し重視する設定
true_weights = {
0: 0.15, # q1: 入手しやすさ
1: 0.05, # q2: 意外性
2: 0.10, # q3: 時間帯
3: 0.70 # q4: 調理しやすさ
}
total_score = 0
num_recipes = len(selected_menu_details)
if num_recipes == 0:
return 1 # レシピがなければ最低評価
# 献立に含まれる全レシピのUXスコアを、真の好みで重み付けして合計
for recipe in selected_menu_details:
total_score += true_weights[0] * recipe.get('q1_score', 0)
total_score += true_weights[1] * recipe.get('q2_score', 0)
total_score += true_weights[2] * recipe.get('q3_score', 0)
total_score += true_weights[3] * recipe.get('q4_score', 0)
# 平均スコアを計算 (0.0 〜 1.0 の範囲になる)
avg_score = total_score / num_recipes
# 平均スコアを1〜5の整数に変換して「報酬」とする
# 完全に好みに合致(スコア1.0)すれば5、全く合わなければ(スコア0.0)1になる
reward = round(avg_score * 4) + 1
# 報酬が1未満や5より大きくなることを防ぐ
reward = max(1, min(5, int(reward)))
print(f"仮想ユーザー評価: 平均スコア={avg_score:.3f} -> 報酬={reward}")
return reward
200回のシミュレーションを実行した結果、各腕の選択回数は以下のようになった。
#ref(): File not found: "final_arm_selection_chart.png" at page "辻さん卒論"
(ここに、最終的な腕の選択回数の集計結果のグラフ画像を挿入)
#ref(): File not found: "final_arm_selection_table.png" at page "辻さん卒論"
(ここに、最終的な腕の選択回数の集計結果の表画像を挿入)
この結果から、本システムは仮想ユーザーの最も重要な好みである「調理しやすさ(腕3)」を最適戦略として正しく学習し、最も多く選択(42.0%)したことが確認できる。同時に、他の戦略も継続的に探索しており、活用と探索のバランスが機能していることが示された。