""" 月次レポート生成サービス - LLM分析モジュール 各セクションのデータをLLMに渡して分析コメントを生成 """ import os import sys from typing import Optional sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config import ( LLM_ANALYSIS_ENABLED, OPENAI_API_KEY, OPENAI_MODEL, OPENAI_TEMPERATURE, OPENAI_MAX_TOKENS, ) class LLMAnalyzer: """LLMを使った分析コメント生成クラス""" # セクション別プロンプト定義 SECTION_PROMPTS = { 'summary': """あなたは治験チャットボットの運用アナリストです。 以下の月次サマリーデータを分析し、全体傾向と前月比の変化について 簡潔に評価してください。良い点と注意が必要な点があれば両方指摘してください。 データ: {data} 分析コメントを2-3文で日本語で記載してください。""", 'usage_overview': """あなたは治験チャットボットの運用アナリストです。 以下の利用状況データ（日別推移、前月比較）を分析し、 利用傾向や前月からの変化について簡潔にコメントしてください。 データ: {data} 分析コメントを2-3文で日本語で記載してください。""", 'usage_pattern': """あなたは治験チャットボットの運用アナリストです。 以下の利用パターンデータ（時間帯別・曜日別）を分析し、 運用改善への示唆（例：ピーク時間帯の対応強化など）を簡潔にコメントしてください。 データ: {data} 分析コメントを2-3文で日本語で記載してください。""", 'quality': """あなたは治験チャットボットの運用アナリストです。 以下の回答品質データ（解決率、フィードバック率、満足度、低評価傾向）を分析し、 品質改善のポイントを簡潔にコメントしてください。 重要な注意事項: - フィードバック率が20%未満の場合、満足度の統計的信頼性が低いことを指摘してください - サンプル数が少ない場合は「フィードバック率が低く、満足度は参考値」と明記してください - 改善提案は「Q&Aデータの追加・修正」など顧客がコントロールできる範囲に限定してください - システム機能の改善提案（フィードバック促進の仕組み導入など）は絶対に含めないでください データ: {data} 分析コメントを3-4文で日本語で記載してください。""", 'keywords': """あなたは治験チャットボットの運用アナリストです。 以下のキーワード分析データ（頻出キーワード、未解決に多いキーワード）を分析し、 ユーザーの関心事や未対応領域について簡潔にコメントしてください。 データ: {data} 分析コメントを2-3文で日本語で記載してください。""", 'improvement': """あなたは治験チャットボットの運用アナリストです。 以下の回答できなかった質問一覧を分析し、 改善優先度が高いトピックや具体的なQ&A追加提案を簡潔にコメントしてください。 データ: {data} 分析コメントを2-3文で日本語で記載してください。具体的なQ&A追加候補があれば1-2件提案してください。""", } def __init__(self, enabled: bool = None): """ 初期化 Args: enabled: LLM分析を有効にするか（Noneの場合は設定ファイルに従う） """ self._enabled = enabled if enabled is not None else LLM_ANALYSIS_ENABLED self._client = None self._client_init_failed = False # クライアント初期化失敗フラグ self._analyses = {} @property def enabled(self) -> bool: """LLM分析が有効かどうか""" return self._enabled and bool(OPENAI_API_KEY) def _get_client(self): """OpenAI クライアントを取得（遅延初期化）""" # 既に初期化に失敗している場合はNoneを返す if self._client_init_failed: return None if self._client is None: try: from openai import OpenAI self._client = OpenAI(api_key=OPENAI_API_KEY) except ImportError: print("警告: openaiパッケージがインストールされていません。pip install openai を実行してください。") self._enabled = False self._client_init_failed = True except Exception as e: print(f"警告: OpenAIクライアントの初期化に失敗しました: {e}") self._enabled = False self._client_init_failed = True return self._client def analyze_section(self, section: str, data: dict) -> Optional[str]: """ セクションのデータを分析してコメントを生成 Args: section: セクション名（summary, usage_overview, usage_pattern, quality, keywords, improvement） data: 分析対象のデータ Returns: 分析コメント（文字列）、失敗時はNone """ if not self.enabled: return None if section not in self.SECTION_PROMPTS: print(f"警告: 未知のセクション '{section}'") return None # キャッシュチェック if section in self._analyses: return self._analyses[section] client = self._get_client() if client is None: return None # データを文字列に変換 data_str = self._format_data(section, data) # プロンプトを構築 prompt = self.SECTION_PROMPTS[section].format(data=data_str) try: response = client.chat.completions.create( model=OPENAI_MODEL, messages=[ {"role": "system", "content": "あなたは治験チャットボットの運用データを分析する専門家です。簡潔で実用的な分析コメントを日本語で提供してください。"}, {"role": "user", "content": prompt} ], temperature=OPENAI_TEMPERATURE, max_completion_tokens=OPENAI_MAX_TOKENS, ) result = response.choices[0].message.content.strip() self._analyses[section] = result return result except Exception as e: print(f"警告: セクション '{section}' のLLM分析に失敗しました: {e}") return None def _format_data(self, section: str, data: dict) -> str: """セクションに応じてデータを整形""" if section == 'summary': return self._format_summary_data(data) elif section == 'usage_overview': return self._format_usage_overview_data(data) elif section == 'usage_pattern': return self._format_usage_pattern_data(data) elif section == 'quality': return self._format_quality_data(data) elif section == 'keywords': return self._format_keywords_data(data) elif section == 'improvement': return self._format_improvement_data(data) else: return str(data) def _format_summary_data(self, data: dict) -> str: """サマリーセクション用のデータ整形""" summary = data.get('summary', {}) comparison = data.get('comparison', {}) lines = [ "【サマリー】", f"・総会話数: {summary.get('total_conversations', 0):,}件", f"・ユニークIP数（≒ユーザー数）: {summary.get('unique_ips', 0):,}", f"・ユーザーあたり平均会話数: {summary.get('conversations_per_user', 0)}回", f"・自動解決率: {summary.get('resolution_rate', 0)}%", f"・未解決率: {summary.get('unresolved_rate', 0)}%", f"・満足度: {summary.get('satisfaction_rate', 0)}%", "", "【前月比較】", f"・総会話数の変化: {comparison.get('changes', {}).get('total_conversations', 0):+}%", f"・回答できた数の変化: {comparison.get('changes', {}).get('answered_count', 0):+}%", f"・回答できなかった数の変化: {comparison.get('changes', {}).get('unanswered_count', 0):+}%", ] return "\n".join(lines) def _format_usage_overview_data(self, data: dict) -> str: """利用状況概要セクション用のデータ整形""" daily_stats = data.get('daily_stats', []) comparison = data.get('comparison', {}) # 日別統計のサマリー if daily_stats: total_days = len(daily_stats) max_day = max(daily_stats, key=lambda x: x.get('total_conversations', 0)) if daily_stats else {} min_day = min(daily_stats, key=lambda x: x.get('total_conversations', 0)) if daily_stats else {} avg_daily = sum(d.get('total_conversations', 0) for d in daily_stats) / total_days if total_days > 0 else 0 else: total_days = 0 max_day = {} min_day = {} avg_daily = 0 # 前月比較データ current = comparison.get('current', {}) previous = comparison.get('previous', {}) changes = comparison.get('changes', {}) lines = [ "【日別統計サマリー】", f"・データ日数: {total_days}日", f"・1日あたり平均会話数: {avg_daily:.1f}件", f"・最多日: {max_day.get('date', 'N/A')} ({max_day.get('total_conversations', 0)}件)", f"・最少日: {min_day.get('date', 'N/A')} ({min_day.get('total_conversations', 0)}件)", "", "【前月比較】", f"・総会話数: 前月{previous.get('total_conversations', 0)}件 → 当月{current.get('total_conversations', 0)}件（{changes.get('total_conversations', 0):+}%）", f"・回答できた: 前月{previous.get('answered_count', 0)}件 → 当月{current.get('answered_count', 0)}件", f"・回答できなかった: 前月{previous.get('unanswered_count', 0)}件 → 当月{current.get('unanswered_count', 0)}件", ] return "\n".join(lines) def _format_usage_pattern_data(self, data: dict) -> str: """利用パターンセクション用のデータ整形""" hourly_stats = data.get('hourly_stats', []) dow_stats = data.get('day_of_week_stats', []) # 時間帯別のピーク if hourly_stats: peak_hour = max(hourly_stats, key=lambda x: x.get('count', 0)) if hourly_stats else {} else: peak_hour = {} # 曜日別のピーク if dow_stats: peak_dow = max(dow_stats, key=lambda x: x.get('count', 0)) if dow_stats else {} else: peak_dow = {} lines = [ "【時間帯別傾向】", f"・ピーク時間帯: {peak_hour.get('hour', 'N/A')}時台 ({peak_hour.get('count', 0)}件)", "", "【曜日別傾向】", f"・最も利用が多い曜日: {peak_dow.get('day_name', 'N/A')} ({peak_dow.get('count', 0)}件)", ] # 時間帯別の分布（上位5件） if hourly_stats: sorted_hourly = sorted(hourly_stats, key=lambda x: x.get('count', 0), reverse=True)[:5] lines.append("") lines.append("【時間帯別上位5】") for h in sorted_hourly: lines.append(f"・{h.get('hour', 0)}時台: {h.get('count', 0)}件") return "\n".join(lines) def _format_quality_data(self, data: dict) -> str: """回答品質セクション用のデータ整形""" summary = data.get('summary', {}) recent_disliked = data.get('recent_disliked', []) total_conversations = summary.get('total_conversations', 0) total_feedbacks = summary.get('total_feedbacks', 0) feedback_rate = summary.get('feedback_rate', 0) lines = [ "【回答品質指標】", f"・総会話数: {total_conversations:,}件", f"・回答できた: {summary.get('answered_count', 0):,}件", f"・回答できなかった: {summary.get('unanswered_count', 0):,}件", f"・自動解決率: {summary.get('resolution_rate', 0)}%", "", "【フィードバック指標】", f"・フィードバック数: {total_feedbacks}件 / {total_conversations}会話", f"・フィードバック率: {feedback_rate}%", f"・いいね: {summary.get('likes_count', 0):,}件", f"・よくない: {summary.get('dislikes_count', 0):,}件", f"・満足度: {summary.get('satisfaction_rate', 0)}%（いいね÷フィードバック数）", ] # 低評価の質問サンプル if recent_disliked: lines.append("") lines.append("【低評価を受けた質問サンプル】") for d in recent_disliked[:5]: q = d.get('question', '') if q: lines.append(f"・{q[:50]}{'...' if len(q) > 50 else ''}") return "\n".join(lines) def _format_keywords_data(self, data: dict) -> str: """キーワード分析セクション用のデータ整形""" keywords = data.get('keywords', []) unresolved_keywords = data.get('unresolved_keywords', []) lines = ["【頻出キーワード上位10】"] for kw in keywords[:10]: # タプル形式 (keyword, count) または辞書形式に対応 if isinstance(kw, tuple): lines.append(f"・{kw[0]}: {kw[1]}回") else: lines.append(f"・{kw.get('keyword', '')}: {kw.get('count', 0)}回") lines.append("") lines.append("【未解決に多いキーワード上位10】") for kw in unresolved_keywords[:10]: if isinstance(kw, tuple): lines.append(f"・{kw[0]}: {kw[1]}回") else: lines.append(f"・{kw.get('keyword', '')}: {kw.get('count', 0)}回") return "\n".join(lines) def _format_improvement_data(self, data: dict) -> str: """改善提案セクション用のデータ整形""" unanswered = data.get('unanswered_questions', []) # 重複除去してカウント unique_questions = {} for q in unanswered: question = q.get('user_message', '') if question: if question not in unique_questions: unique_questions[question] = 1 else: unique_questions[question] += 1 sorted_questions = sorted(unique_questions.items(), key=lambda x: x[1], reverse=True) lines = [ f"【回答できなかった質問】", f"・総数: {len(unanswered)}件", f"・ユニーク数: {len(unique_questions)}件", "", "【頻出する未回答質問上位10】", ] for q, count in sorted_questions[:10]: lines.append(f"・({count}回) {q[:60]}{'...' if len(q) > 60 else ''}") return "\n".join(lines) def analyze_all(self, data: dict) -> dict: """ 全セクションの分析を実行 Args: data: レポート全体のデータ Returns: セクション名をキー、分析コメントを値とする辞書 """ if not self.enabled: return {} results = {} sections = [ ('summary', data), ('usage_overview', data), ('usage_pattern', data), ('quality', data), ('keywords', data), ('improvement', data), ] for section, section_data in sections: # 途中で無効化された場合は終了 if not self.enabled: break print(f" LLM分析中: {section}...") result = self.analyze_section(section, section_data) if result: results[section] = result return results def get_analysis(self, section: str) -> Optional[str]: """ キャッシュされた分析結果を取得 Args: section: セクション名 Returns: 分析コメント（キャッシュがない場合はNone） """ return self._analyses.get(section)