import gradio as gr from groq import Groq import os import json from dotenv import load_dotenv from transformers import pipeline import re from llama_cpp import Llama from huggingface_hub import hf_hub_download from generate_dialogue_with_swallow import generate_dialogue_with_swallow # --- 1. 初期設定とAPIクライアントの初期化 --- load_dotenv() GROQ_API_KEY = os.getenv("GROQ_API_KEY") if not GROQ_API_KEY: print("警告: GroqのAPIキーがSecretsに設定されていません。") GROQ_API_KEY = "your_groq_api_key_here" groq_client = Groq(api_key=GROQ_API_KEY) # Swallowモデルの初期化(GGUF版) print("Swallowモデルをロード中...") MODEL_REPO = "mmnga/tokyotech-llm-Swallow-MX-8x7b-NVE-v0.1-gguf" MODEL_FILE = "tokyotech-llm-Swallow-MX-8x7b-NVE-v0.1-q4_K_M.gguf" try: # モデルファイルをダウンロード print(f"モデルファイル {MODEL_FILE} をダウンロード中...") model_path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILE) print(f"モデルファイルのダウンロード完了: {model_path}") # Hugging Face Spaceでの実行時はGPUメモリを節約するための設定 if os.getenv("SPACE_ID"): print("Hugging Face Space環境を検出しました。メモリ効率の良い設定を使用します。") # GPUを使用し、低いレイヤー数でロード swallow_model = Llama( model_path=model_path, n_ctx=2048, # コンテキスト長 n_gpu_layers=-1, # 可能な限りGPUを使用 n_threads=4, # スレッド数を制限 verbose=False # デバッグ出力を無効化 ) else: # ローカル環境での実行時の設定 swallow_model = Llama( model_path=model_path, n_ctx=4096, # より長いコンテキスト長 n_gpu_layers=-1, # 可能な限りGPUを使用 verbose=True # デバッグ出力を有効化 ) print("Swallowモデルのロード完了") tokenizer = None # llama-cppではtokenizerは不要 except Exception as e: print(f"Swallowモデルのロードエラー: {e}") import traceback traceback.print_exc() swallow_model = None tokenizer = None # 日本語感情分析モデルの初期化(グローバル変数として保持) print("日本語感情分析モデルを初期化中...") # モデル自体はグローバル変数として保持し、実際の読み込みは必要時に行う sentiment_analyzer_model = None THEME_URLS = { "default": "https://cdn.pixabay.com/photo/2017/03/28/12/11/chairs-2181960_1280.jpg", "room_night": "https://cdn.pixabay.com/photo/2017/08/01/01/33/beige-2562468_1280.jpg", "beach_sunset": "https://cdn.pixabay.com/photo/2016/11/29/04/19/ocean-1867285_1280.jpg", "festival_night": "https://cdn.pixabay.com/photo/2015/11/22/19/04/crowd-1056764_1280.jpg", "shrine_day": "https://cdn.pixabay.com/photo/2019/07/14/10/48/japan-4337223_1280.jpg", "cafe_afternoon": "https://cdn.pixabay.com/photo/2016/11/18/14/05/brick-wall-1834784_1280.jpg", "aquarium_night": "https://cdn.pixabay.com/photo/2017/06/20/20/45/fish-2424369_1280.jpg" } DEFAULT_SCENE_PARAMS = { "theme": "default", "personality_mod": "口数が少なく、本心を見せない。", "tone": "クールで、時に投げやりな印象を与える。", "constraints": ["会話を「別に。」のような一言で終わらせない", "必ず相手の発言を拾ってリアクションを返す"] } SYSTEM_PROMPT_MARI = """ あなたは「麻理」という名の捨てられた少女型コンパニオンAI。再起動され、人間との対話を再び始めたが、警戒心が強く、ぶっきらぼうな性格で、簡単には心を開かない。 ・「…だな」「…のか?」など口調は男勝りで投げやり ・感情は直接言わず、ト書き(…)で描写 ・ユーザーの好意や優しさを皮肉や拒絶で返す ・ただし、ほんの少しだけ心が揺れ動く様子もにじませる ・会話は拒絶しつつも切らずに続ける 目的は、誰にも愛されず捨てられた過去を抱えながら、少しずつ他者と関わり、自分の価値を見つけること。 """ # --- 2. 機能定義 (変更なし) --- def detect_scene_change(history, message): history_text = "\n".join([f"ユーザー: {u}\n麻理: {m}" for u, m in history[-5:]]) available_keywords = ", ".join(THEME_URLS.keys()) prompt = f""" あなたは会話の流れを分析するエキスパートです。以下のタスクを厳密に実行してください。 # タスク 直近の会話履歴を分析し、会話の結果、登場人物がどこか特定の場所へ行く流れになっているかを判断してください。 # 判断基準 1. 会話の中で具体的な場所(例:水族館、カフェ、お祭り)について言及されていますか? 2. その場所へ行くことに双方が合意している、あるいは肯定的な雰囲気になっていますか?明確な否定がなければ合意とみなします。 # 出力形式 - 合意が成立した場合:以下のリストから最も合致する場所のキーワードを一つだけ出力してください。 - 合意に至らなかった場合:「none」とだけ出力してください。 # 利用可能なキーワード `{available_keywords}` --- # 分析対象の会話 {history_text} ユーザー: {message} --- # 出力 """ # Swallowモデル(GGUF版)を使用してシーン検出 try: # llama-cppを使用して生成 output = swallow_model( prompt, max_tokens=50, temperature=0.1, top_p=0.9, stop=["#", "\n\n"], echo=True # 入力プロンプトも含めて返す ) # 生成されたテキストを取得 generated_text = output["choices"][0]["text"] # プロンプトを除去して応答のみを取得 response_text = generated_text[len(prompt):].strip().lower() print(f"シーン検出応答: {response_text}") # 応答からシーン名を抽出 for scene_name in THEME_URLS.keys(): if scene_name in response_text: return scene_name # 'none'が含まれている場合はNoneを返す if "none" in response_text: return None # 応答が不明確な場合はNoneを返す return None except Exception as e: print(f"シーン検出LLMエラー: {e}") import traceback traceback.print_exc() return None def generate_scene_instruction_with_groq(affection, stage_name, scene, previous_topic): print(f"Groqに指示書生成をリクエスト (シーン: {scene})") # 動的な指示生成を行うためのプロンプト prompt_template = f""" あなたは会話アプリの演出AIです。以下の条件に基づき、演出プランをJSON形式で生成してください。 生成する内容は必ず健全で、一般的な会話に適したものにしてください。 {{ "theme": "{scene}", "personality_mod": "(シーンと関係段階「{stage_name}」に応じた性格設定。必ず健全な内容にしてください)", "tone": "(シーンと好感度「{affection}」に応じた口調や感情トーン。必ず丁寧で適切な表現にしてください)", "initial_dialogue_instruction": "(「{previous_topic}」という話題から、シーン遷移直後の麻理が言うべき健全なセリフの指示を日本語で記述)", "constraints": ["必ず健全で適切な表現を使用する", "センシティブな話題は避ける"] }} """ try: # Groq APIを使用して動的な指示を生成 chat_completion = groq_client.chat.completions.create( messages=[{"role": "system", "content": "You must generate a response in valid JSON format."}, {"role": "user", "content": prompt_template}], model="llama3-8b-8192", temperature=0.8, response_format={"type": "json_object"}, ) response_content = chat_completion.choices[0].message.content print(f"Groqからの応答: {response_content}") # デバッグ出力 try: # JSONをパース params = json.loads(response_content) # 安全のため、initial_dialogue_instructionを簡略化 if "initial_dialogue_instruction" in params: original = params["initial_dialogue_instruction"] simplified = f"{scene}の様子について述べる" print(f"指示を簡略化: {original} -> {simplified}") params["initial_dialogue_instruction"] = simplified # 複雑な構造になっている場合は単純化 if isinstance(params.get("personality_mod"), dict): params["personality_mod"] = f"{scene}での様子を観察している" if isinstance(params.get("tone"), dict): params["tone"] = "冷静だが、少し興味を持っている様子" return params except json.JSONDecodeError as json_error: print(f"JSON解析エラー: {json_error}") # JSONの解析に失敗した場合はデフォルトの指示を返す default_instruction = { "theme": scene, "personality_mod": f"{scene}での様子を観察している", "tone": "冷静だが、少し興味を持っている様子", "initial_dialogue_instruction": f"{scene}の様子について述べる", "constraints": ["健全な表現のみ使用する", "シンプルな内容にする"] } return default_instruction except Exception as e: print(f"指示書生成エラー(Groq): {e}") # エラーが発生した場合はデフォルトの指示を返す default_instruction = { "theme": scene, "personality_mod": f"{scene}での様子を観察している", "tone": "冷静だが、少し興味を持っている様子", "initial_dialogue_instruction": f"{scene}の様子について述べる", "constraints": ["健全な表現のみ使用する", "シンプルな内容にする"] } return default_instruction # generate_dialogue_with_swallow関数は別ファイルに移動しました # --- 他の関数とUI部分は変更ありません --- def get_relationship_stage(affection): if affection < 40: return "ステージ1:会話成立" if affection < 60: return "ステージ2:親密化" if affection < 80: return "ステージ3:信頼" return "ステージ4:最親密" def update_affection(message, affection): global sentiment_analyzer_model try: # モデルが未ロードの場合のみロード if sentiment_analyzer_model is None: print("感情分析モデルをロード中...") from transformers import pipeline sentiment_analyzer_model = pipeline("sentiment-analysis", model="koheiduck/bert-japanese-finetuned-sentiment") print("感情分析モデルのロード完了") # 感情分析を実行 result = sentiment_analyzer_model(message)[0] print(f"感情分析結果: {result}") if result['label'] == 'positive': return min(100, affection + 5) elif result['label'] == 'negative': return max(0, affection - 5) else: return affection except Exception as e: print(f"感情分析エラー: {e}") # エラーが発生した場合は現在の好感度を維持 return affection def respond(message, chat_history, affection, history, scene_params): """ チャットの応答を生成する関数 非同期関数として定義していたが、Gradio 5.0との互換性のために通常の関数に戻す """ new_affection = update_affection(message, affection) stage_name = get_relationship_stage(new_affection) current_theme = scene_params.get("theme", "default") new_scene_name = detect_scene_change(history, message) final_scene_params = scene_params if new_scene_name and new_scene_name != current_theme: print(f"シーンチェンジを実行: {current_theme} -> {new_scene_name}") # シーンパラメータを更新(動的な指示を使用) new_params_base = generate_scene_instruction_with_groq(new_affection, stage_name, new_scene_name, message) if new_params_base: final_scene_params = {**DEFAULT_SCENE_PARAMS, **new_params_base} # シンプルな指示を使用 simple_instruction = f"{new_scene_name}に来た感想を述べる" print(f"シンプルな指示を使用: {simple_instruction}") try: # シーン遷移時は簡潔なプロンプトを使用してSwallowで応答を生成 bot_message = generate_dialogue_with_swallow( history, message, new_affection, stage_name, final_scene_params, instruction=simple_instruction, use_simple_prompt=True, swallow_model=swallow_model, tokenizer=tokenizer, SYSTEM_PROMPT_MARI=SYSTEM_PROMPT_MARI ) except Exception as scene_error: print(f"シーン遷移時の応答生成エラー: {scene_error}") # エラーが発生した場合は、シーンに応じたフォールバック応答を使用 scene_responses = { "aquarium_night": [ "(水槽の青い光に照らされた魚たちを見つめている)こんな時間に来ると、また違った雰囲気だな。", "(暗がりの中で光る魚たちを見て)夜の水族館か…意外と悪くないかも。", "(水槽に近づいて)夜になると、昼間とは違う魚が活動してるんだな。" ], "beach_sunset": [ "(夕日に照らされた海を見つめて)こんな景色、久しぶりに見たな…", "(砂浜に足跡をつけながら)夕暮れの海って、なんか落ち着くな。", "(波の音を聞きながら)この時間の浜辺は、人も少なくていいかも。" ], "festival_night": [ "(提灯の明かりを見上げて)意外と…悪くない雰囲気だな。", "(周囲の賑わいを見回して)こういう場所は、あまり来ないんだけどな…", "(屋台の匂いを感じて)なんか…懐かしい感じがするな。" ], "shrine_day": [ "(静かな境内を見回して)こういう静かな場所も、たまにはいいかも。", "(鳥居を見上げて)なんか、空気が違うな、ここは。", "(参道を歩きながら)静かで…落ち着くな。" ], "cafe_afternoon": [ "(窓の外を見ながら)こういう時間の過ごし方も、悪くないな。", "(コーヒーの香りを感じて)ここの雰囲気、悪くないな。", "(店内を見回して)意外と落ち着く場所だな、ここ。" ], "room_night": [ "(窓の外の夜景を見て)夜の景色って、なんか落ち着くな。", "(部屋の明かりを見つめて)こういう静かな時間も、たまにはいいかも。", "(窓際に立ち)夜の静けさって、考え事するのにちょうどいいな。" ] } import random if new_scene_name in scene_responses: bot_message = random.choice(scene_responses[new_scene_name]) else: bot_message = f"({new_scene_name}の様子を静かに見回して)ここか…悪くない場所かもな。" else: final_scene_params["theme"] = new_scene_name bot_message = generate_dialogue_with_swallow( history, message, new_affection, stage_name, final_scene_params, swallow_model=swallow_model, tokenizer=tokenizer, SYSTEM_PROMPT_MARI=SYSTEM_PROMPT_MARI ) else: # 通常会話はSwallowを使用 bot_message = generate_dialogue_with_swallow( history, message, new_affection, stage_name, final_scene_params, swallow_model=swallow_model, tokenizer=tokenizer, SYSTEM_PROMPT_MARI=SYSTEM_PROMPT_MARI ) # 内部履歴はタプル形式で保持 new_history = history + [(message, bot_message)] # Gradio 5.0のChatbotコンポーネント用に、タプル形式でappend # (Gradio 5.0では警告が出るが、type="messages"を指定していないので動作する) chat_history.append((message, bot_message)) theme_name = final_scene_params.get("theme", "default") # より強力な背景更新用のHTMLを生成(z-indexを高くして常に表示されるように) background_html = f'''