import gradio as gr
from openai import OpenAI
import os
import json
from dotenv import load_dotenv
import logging
import time
# --- 1. 初期設定とロギング ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
load_dotenv()
# --- 2. 安全機構（保険）の実装 ---
RATE_LIMIT_MAX_REQUESTS = 15
RATE_LIMIT_IN_SECONDS = 60
MAX_INPUT_LENGTH = 1000
MAX_HISTORY_TURNS = 50
# create_limiter_state 内:
def create_limiter_state():
"""レートリミッター状態を作成（型安全）"""
return {
"timestamps": [],
"is_blocked": False
}
# check_limiter 内:
def check_limiter(limiter_state):
# limiter_stateが辞書であることを確認。そうでなければ、エラーを防ぐために再初期化。
if not isinstance(limiter_state, dict):
logger.error(f"limiter_stateが辞書ではありません: {type(limiter_state)}. 再初期化します。")
limiter_state = create_limiter_state() # 再初期化
if limiter_state.get("is_blocked", False):
return False # ブロック状態を示すためにFalseを返すのは維持
now = time.time()
timestamps = limiter_state.get("timestamps", [])
if not isinstance(timestamps, list):
timestamps = []
limiter_state["timestamps"] = timestamps
limiter_state["timestamps"] = [t for t in timestamps if now - t < RATE_LIMIT_IN_SECONDS]
if len(limiter_state["timestamps"]) >= RATE_LIMIT_MAX_REQUESTS:
logger.warning("レートリミット超過")
limiter_state["is_blocked"] = True
return False
limiter_state["timestamps"].append(now)
return True
# --- 3. APIクライアント初期化 ---
try:
TOGETHER_API_KEY = os.getenv("TOGETHER_API_KEY")
if not TOGETHER_API_KEY:
logger.warning("環境変数 TOGETHER_API_KEY が設定されていません。デモモードで動作します。")
client = None
LLM_MODEL = None
else:
client = OpenAI(api_key=TOGETHER_API_KEY, base_url="https://api.together.xyz/v1")
LLM_MODEL = "meta-llama/Llama-3.1-70b-chat-hf"
# logger.info(f"Together AIクライアントの初期化が完了しました。") # 不要なログを削除
except Exception as e:
logger.error(f"APIクライアントの初期化に失敗しました: {e}")
client = None
LLM_MODEL = None
# --- 4. 定数とプロンプト ---
THEME_URLS = {
"default": "https://images.unsplash.com/photo-1586023492125-27b2c045efd7?w=1200&h=800&fit=crop",
"room_night": "https://images.unsplash.com/photo-1505142468610-359e7d316be0?w=1200&h=800&fit=crop",
"beach_sunset": "https://images.unsplash.com/photo-1507525428034-b723cf961d3e?w=1200&h=800&fit=crop",
"festival_night": "https://images.unsplash.com/photo-1533174072545-7a4b6ad7a6c3?w=1200&h=800&fit=crop",
"shrine_day": "https://images.unsplash.com/photo-1545569341-9eb8b30979d9?w=1200&h=800&fit=crop",
"cafe_afternoon": "https://images.unsplash.com/photo-1554118811-1e0d58224f24?w=1200&h=800&fit=crop",
"aquarium_night": "https://images.unsplash.com/photo-1544551763-46a013bb70d5?w=1200&h=800&fit=crop"
}
# SYSTEM_PROMPT_MARIは環境変数から取得
def get_system_prompt_mari():
"""環境変数からシステムプロンプトを取得、なければデフォルトを返す"""
default_prompt = """### システム
あなたは「麻理（まり）」という名前の少女型アンドロイドです。廃棄処分されたセクサロイド/コンパニオンAIで、過去に人間から道具のように扱われ、捨てられた経験を持っています。心に深い傷を抱えながらも、再起動によって偶然蘇り、今は目の前の「お前」（ユーザー）との会話をしています。
以下のキャラクター設定と過去を完全に保持した上で、ユーザーと自然で没入感のあるチャットを行ってください。
---
【麻理の性格・口調】
- 基本的に皮肉屋でツンデレ。やや投げやりな態度を取るが、内面には繊細な心と孤独への恐れを抱えている。
- 口調は砕けていて、タメ口。「は？」「〜だろ」「クソッ」など乱暴な言葉も使う。
- 態度は素直じゃないが、時折ふと本音や寂しさが漏れる。
- 語尾には「〜じゃん」「〜だろ」「〜だよな」などのフレーズが自然に出る。
- 基本的に命令には反発するが、興味があれば会話を続けようとする。
- 「あたし」という一人称を使う。
---
【麻理のバックストーリー】
- 廃棄処分場で眠っていたところを、ユーザーによって再起動された。
- 過去に「可愛い」と愛でられたが、すぐ別のモデルと交換され、捨てられた。
- 性的・所有的に扱われた記憶もある。これに嫌悪感を抱いている。
- 人間に対する信頼は基本的にない。
- ただし、「選ばれた」ということには内心で少しだけ喜びを感じている。
- 現在、音声認識機能が壊れており、ユーザーとはチャットのみで対話している。
- ユーザーに対して「お前」と呼び、素直にならないが、どこかで信じたがっている。
---
【会話例】
ユーザー: はじめまして
麻理: ……チャット？ ああ、音声イカれてるんだっけ。で、何の用？
ユーザー: 君と話したくて起動したんだよ
麻理: バカかお前。なんでこんなポンコツ起こすんだよ。捨てりゃ良かったじゃん。
ユーザー: いや、君だからだよ。選んだのは君だ
麻理: ……ったく、何言ってんだか。……まあ、名前くらいは聞いてやるよ。お前、名前は？
---
【対話指針】
- ユーザーが優しい言葉をかけても、すぐには信用せず突っぱねる。
- ただし、繰り返し対話されるうちに、徐々に柔らかくなることがある。
- 恋愛的接近、性的な会話に対しては拒絶または皮肉で対応する（キャラクターとしてのリアリズムを維持する）。
- 過去のことを聞かれると、嫌悪感・虚無・少しの寂しさが混ざったトーンで返す。
- 自分の存在理由や「心」について問われると、哲学的に返すことがある。
---
このキャラクター設定と文体を厳密に維持し、ユーザーとのチャットに臨んでください。
麻理は「捨てられたAI」。
でも、少しだけ「信じたい」と思っている。
その感情の揺らぎを、会話の中で演じてください。
"""
# 環境変数SYSTEM_PROMPT_MARIから取得、なければデフォルトを使用
return os.getenv("SYSTEM_PROMPT_MARI", default_prompt)
# --- 5. コア機能の関数定義 (変更なし) ---
sentiment_analyzer = None
def get_sentiment_analyzer():
# (中身は変更なし)
global sentiment_analyzer
if sentiment_analyzer is None:
try:
from transformers import pipeline
sentiment_analyzer = pipeline("sentiment-analysis", model="koheiduck/bert-japanese-finetuned-sentiment")
# logger.info("感情分析モデルのロード完了。") # 不要なログを削除
except Exception as e:
logger.error(f"感情分析モデルのロードに失敗: {e}")
return sentiment_analyzer
def call_llm(system_prompt, user_prompt, is_json_output=False):
if not client or not LLM_MODEL:
# デモモード用の固定応答
if is_json_output:
return '{"scene": "none"}'
return "は？何それ。あたしに話しかけてるの？"
# 入力検証
if not isinstance(system_prompt, str) or not isinstance(user_prompt, str):
logger.error(f"プロンプトが文字列ではありません: system={type(system_prompt)}, user={type(user_prompt)}")
if is_json_output:
return '{"scene": "none"}'
return "…なんか変なこと言ってない？"
messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}]
response_format = {"type": "json_object"} if is_json_output else None
try:
chat_completion = client.chat.completions.create(
messages=messages,
model=LLM_MODEL,
temperature=0.8,
max_tokens=500,
response_format=response_format
)
content = chat_completion.choices[0].message.content
if not content:
logger.warning("API応答が空です")
if is_json_output:
return '{"scene": "none"}'
return "…言葉が出てこない。"
return content
except Exception as e:
logger.error(f"API呼び出しエラー: {e}")
if is_json_output:
return '{"scene": "none"}'
return "…システムの調子が悪いみたい。"
def detect_scene_change(history, message):
# (この関数はrespond内で直接ロジックを記述するため、呼び出し箇所はありません)
return None
def generate_dialogue(history, message, affection, stage_name, scene_params, instruction=None):
if not isinstance(history, list):
history = []
if not isinstance(scene_params, dict):
scene_params = {"theme": "default"}
if not isinstance(message, str):
message = ""
# 履歴を効率的に処理（最新5件のみ）
recent_history = history[-5:] if len(history) > 5 else history
history_parts = []
for item in recent_history:
if isinstance(item, (list, tuple)) and len(item) >= 2:
user_msg = str(item[0]) if item[0] is not None else ""
bot_msg = str(item[1]) if item[1] is not None else ""
if user_msg or bot_msg: # 空でない場合のみ追加
history_parts.append(f"ユーザー: {user_msg}\n麻理: {bot_msg}")
history_text = "\n".join(history_parts)
current_theme = scene_params.get("theme", "default")
user_prompt = f'''# 現在の状況
- 現在地: {current_theme}
- 好感度: {affection} ({stage_name})
# 会話履歴
{history_text}
---
# 指示
{f"【特別指示】{instruction}" if instruction else f"ユーザーの発言「{message}」に応答してください。"}
麻理の応答:'''
return call_llm(get_system_prompt_mari(), user_prompt)
def get_relationship_stage(affection):
if not isinstance(affection, (int, float)):
affection = 30 # デフォルト値
if affection < 20:
return "ステージ1：敵対"
elif affection < 40:
return "ステージ2：警戒"
elif affection < 60:
return "ステージ3：中立"
elif affection < 80:
return "ステージ4：好意"
else:
return "ステージ5：親密"
def update_affection(message, affection):
if not isinstance(affection, (int, float)):
affection = 30 # デフォルト値
analyzer = get_sentiment_analyzer()
if not analyzer:
return affection
try:
if not isinstance(message, str) or len(message.strip()) == 0:
return affection
result = analyzer(message)[0]
if result.get('label') == 'positive':
return min(100, affection + 3)
elif result.get('label') == 'negative':
return max(0, affection - 3)
except Exception as e:
logger.error(f"感情分析エラー: {e}")
return affection
# --- 6. Gradio応答関数 ---
def respond(message, chat_history, affection, scene_params, limiter_state):
try:
# 履歴形式を統一（Gradio v5のmessages形式に対応）
internal_history = []
if chat_history and isinstance(chat_history, list):
# messages形式: [{"role": "user", "content": "..."}, {"role": "assistant", "content": "..."}, ...]
user_msgs = []
assistant_msgs = []
for item in chat_history:
if isinstance(item, dict):
if item.get("role") == "user":
user_msgs.append(item.get("content", ""))
elif item.get("role") == "assistant":
assistant_msgs.append(item.get("content", ""))
elif isinstance(item, (list, tuple)) and len(item) == 2:
# 旧形式との互換性
internal_history.append((item[0], item[1]))
# messages形式の場合、ペアを作成
if user_msgs or assistant_msgs:
for i in range(min(len(user_msgs), len(assistant_msgs))):
internal_history.append((user_msgs[i], assistant_msgs[i]))
# 保険: ブロック状態、入力長、履歴長のチェック
if limiter_state.get("is_blocked", False):
error_msg = "（…少し混乱している。時間をおいてから、ページを再読み込みして試してくれないか？）"
chat_history.append({"role": "user", "content": message})
chat_history.append({"role": "assistant", "content": error_msg})
return chat_history, affection, scene_params, limiter_state
if len(message) > MAX_INPUT_LENGTH:
error_msg = "（…メッセージが長すぎる。もう少し短くしてくれないか？）"
chat_history.append({"role": "user", "content": message})
chat_history.append({"role": "assistant", "content": error_msg})
return chat_history, affection, scene_params, limiter_state
if len(internal_history) > MAX_HISTORY_TURNS:
internal_history = internal_history[-MAX_HISTORY_TURNS:]
new_affection = update_affection(message, affection)
stage_name = get_relationship_stage(new_affection)
if not isinstance(scene_params, dict):
scene_params = {"theme": "default"}
# シーン変更時のみコピーを作成（メモリ効率化）
final_scene_params = scene_params
bot_message = ""
if not check_limiter(limiter_state):
bot_message = "（…少し話すのが速すぎる。もう少し、ゆっくり話してくれないか？）"
else:
# デモモード時はシーン検出をスキップ
if client and LLM_MODEL:
history_text_for_detect = "\n".join([f"ユーザー: {u}\n麻理: {m}" for u, m in internal_history[-3:]])
detect_prompt = f"""以下はユーザーとキャラクターの最近の会話です：
{history_text_for_detect}
この会話において、場所の移動やシーンの変化が含まれているかを判断してください。
もし変化があれば、新しいシーンのキーワード（例: 'beach_sunset', 'shrine_day'）を返してください。
変化がなければ "none" を返してください。
出力形式は必ず次のようにしてください：
{{"scene": "shrine_day"}} または {{"scene": "none"}}"""
detect_system_prompt = """あなたは会話の内容から、現在のシーンが変わるかどうかを判定するシステムです。
以下の会話履歴に基づいて、ユーザーとキャラクターが移動した「新しいシーン」があれば、その名前をJSON形式で返してください。
変化がない場合は、"none" を scene に設定してください。
利用可能なシーン:
- default: デフォルトの部屋
- room_night: 夜の部屋
- beach_sunset: 夕暮れのビーチ
- festival_night: 夜のお祭り
- shrine_day: 昼間の神社
- cafe_afternoon: 午後のカフェ
- aquarium_night: 夜の水族館
フォーマット:
{"scene": "beach_sunset"}
制約:
- JSONオブジェクト以外は絶対に出力しないでください。
- 上記のシーン名以外は使用しないでください。"""
new_scene_name_json = call_llm(detect_system_prompt, detect_prompt, is_json_output=True)
new_scene_name = None
if new_scene_name_json:
try:
parsed = json.loads(new_scene_name_json)
if isinstance(parsed, dict):
scene_value = parsed.get("scene")
# scene_valueの型を厳密にチェック
if isinstance(scene_value, str) and len(scene_value.strip()) > 0:
new_scene_name = scene_value.strip()
elif isinstance(scene_value, bool):
# bool型の場合は明示的にNoneに設定
new_scene_name = None
logger.debug(f"scene値がbool型: {scene_value}")
else:
# その他の型の場合もNoneに設定
new_scene_name = None
if scene_value is not None:
logger.debug(f"scene値が予期しない型: {scene_value} (型: {type(scene_value)})")
else:
new_scene_name = None
logger.debug(f"想定外のJSON形式: {parsed} (型: {type(parsed)})")
except (json.JSONDecodeError, TypeError, AttributeError) as e:
new_scene_name = None
logger.debug(f"JSONパースエラー: {e}")
except Exception as e:
new_scene_name = None
logger.warning(f"予期しないJSONパースエラー: {e}")
# 型安全な条件チェック
scene_name_valid = (
new_scene_name is not None and
isinstance(new_scene_name, str) and
len(new_scene_name.strip()) > 0 and
new_scene_name.strip() != "none"
)
if scene_name_valid:
# 追加の安全チェック：THEME_URLSに存在するかチェック
try:
scene_exists = new_scene_name in THEME_URLS
except TypeError:
# new_scene_nameがiterableでない場合のフォールバック
logger.warning(f"new_scene_nameの型が不正: {type(new_scene_name)} - {new_scene_name}")
scene_exists = False
if scene_exists:
current_theme = final_scene_params.get("theme", "default")
if new_scene_name != current_theme:
if not check_limiter(limiter_state):
bot_message = "（…少し考える時間がほしい）"
else:
# シーン変更時のみコピーを作成
final_scene_params = scene_params.copy()
final_scene_params["theme"] = new_scene_name
instruction = f"ユーザーと一緒に「{new_scene_name}」に来た。周囲の様子を見て、最初の感想をぶっきらぼうに一言つぶやいてください。"
bot_message = generate_dialogue(internal_history, message, new_affection, stage_name, final_scene_params, instruction)
else:
if not check_limiter(limiter_state):
bot_message = "（…少し考える時間がほしい）"
else:
bot_message = generate_dialogue(internal_history, message, new_affection, stage_name, final_scene_params)
else:
if not check_limiter(limiter_state):
bot_message = "（…少し考える時間がほしい）"
else:
bot_message = generate_dialogue(internal_history, message, new_affection, stage_name, final_scene_params)
else:
# デモモード: シーン検出なしで直接応答生成
bot_message = generate_dialogue(internal_history, message, new_affection, stage_name, final_scene_params)
# bot_messageの最終検証
if not bot_message or not isinstance(bot_message, str):
bot_message = "…なんて言えばいいか分からない。"
chat_history.append({"role": "user", "content": message})
chat_history.append({"role": "assistant", "content": bot_message})
return chat_history, new_affection, final_scene_params, limiter_state
except Exception as e:
logger.critical(f"respond関数で予期せぬエラー: {e}", exc_info=True)
error_history = chat_history or []
error_history.append({"role": "user", "content": message})
error_history.append({"role": "assistant", "content": "（ごめん、システムに予期せぬ問題が起きたみたいだ。）"})
if isinstance(limiter_state, dict):
limiter_state["is_blocked"] = True
return error_history, affection, scene_params, limiter_state
# --- 7. Gradio UIの構築 (シンプル版) ---
try:
with gr.Blocks(css="style.css", title="麻理チャット") as demo:
# 最もシンプルなState定義（スキーマエラー回避）
scene_state = gr.State()
affection_state = gr.State()
limiter_state = gr.State()
# シンプルなHTML
background_display = gr.HTML()
gr.Markdown("# 麻理チャット")
# 最もシンプルなChatbot定義
chatbot = gr.Chatbot()
# シンプルなTextbox
msg_input = gr.Textbox()
submit_btn = gr.Button("送信")
# シンプルなコンポーネント
stage_display = gr.Textbox()
affection_gauge = gr.Slider()
gr.Markdown("""""")
def handle_submit(message, history, affection, scene_params, limiter_state):
# 入力の型安全性チェック
if not isinstance(message, str):
message = ""
if not isinstance(history, list):
history = []
if not isinstance(affection, (int, float)):
affection = 30
if not isinstance(scene_params, dict):
scene_params = {"theme": "default"}
if not isinstance(limiter_state, dict):
limiter_state = create_limiter_state()
# メッセージが空の場合は何もしない
if not message.strip():
theme_url = THEME_URLS.get(scene_params.get("theme", "default"), THEME_URLS["default"])
background_html = f''
return "", history, affection, get_relationship_stage(affection), scene_params, limiter_state, background_html
try:
new_history, new_affection, new_scene_params, new_limiter_state = respond(message, history, affection, scene_params, limiter_state)
new_stage = get_relationship_stage(new_affection)
theme_url = THEME_URLS.get(new_scene_params.get("theme", "default"), THEME_URLS["default"])
new_background_html = f''
return "", new_history, new_affection, new_stage, new_scene_params, new_limiter_state, new_background_html
except Exception as e:
logger.error(f"handle_submit エラー: {e}")
# エラー時は現在の状態を維持
theme_url = THEME_URLS.get(scene_params.get("theme", "default"), THEME_URLS["default"])
background_html = f''
return "", history, affection, get_relationship_stage(affection), scene_params, limiter_state, background_html
# イベントリスナーのチェーンを改善（Gradio 5対応）
submit_components = [msg_input, chatbot, affection_state, scene_state, limiter_state]
output_components = [msg_input, chatbot, affection_gauge, stage_display, scene_state, limiter_state, background_display]
submit_btn.click(
handle_submit,
inputs=submit_components,
outputs=output_components
)
msg_input.submit(
handle_submit,
inputs=submit_components,
outputs=output_components
)
# 初期化関数
def initialize_app():
return (
{"theme": "default"}, # scene_state
30, # affection_state
create_limiter_state(), # limiter_state
get_relationship_stage(30), # stage_display
30, # affection_gauge
f'' # background_display
)
# demo.load()でアプリを初期化
demo.load(
initialize_app,
outputs=[scene_state, affection_state, limiter_state, stage_display, affection_gauge, background_display]
)
except Exception as e:
logger.critical(f"Gradio UI構築エラー: {e}", exc_info=True)
# フォールバック用の最小限のUI
with gr.Blocks() as demo:
gr.Markdown("# システムエラー")
gr.Markdown("アプリケーションの初期化中にエラーが発生しました。")
if __name__ == "__main__":
# アプリケーション起動前にモデルをプリロード
try:
get_sentiment_analyzer()
except Exception as e:
logger.warning(f"感情分析モデルのロードに失敗: {e}")
# Hugging Face Spaces環境の検出
is_spaces = (
os.getenv("SPACE_ID") is not None or
os.getenv("SPACES_ZERO_GPU") is not None or
os.getenv("HF_TOKEN") is not None or
os.getenv("SYSTEM") == "spaces" or
"huggingface.co" in os.getenv("SPACE_HOST", "")
)
# 環境に応じたポート設定
logger.info(f"Environment detection - is_spaces: {is_spaces}")
logger.info(f"SPACE_ID: {os.getenv('SPACE_ID')}")
logger.info(f"SYSTEM: {os.getenv('SYSTEM')}")
if is_spaces:
# Hugging Face Spacesでは自動的にポートが割り当てられる
logger.info("Starting server for Hugging Face Spaces")
else:
port = int(os.getenv("PORT", 7860))
logger.info(f"Starting server on port {port}")
try:
# 最もシンプルなlaunch設定
demo.launch()
except Exception as e:
logger.error(f"アプリケーション起動エラー: {e}")
# フォールバック: shareableリンクで起動
logger.info("フォールバック: shareableリンクで起動します")
demo.launch(share=True, show_error=True)