Survey Insight MCP Server

"""LLM APIを使った課題分析・改善提案モジュール.""" import logging import os from pathlib import Path from typing import Optional from anthropic import AsyncAnthropic from dotenv import load_dotenv logger = logging.getLogger(__name__) class AIAnalyzer: """LLM APIを使った課題分析・改善提案を行うクラス.""" def __init__( self, provider: Optional[str] = None, api_key: Optional[str] = None, model: Optional[str] = None ): """初期化. Args: provider: LLMプロバイダー名（省略時は環境変数LLM_PROVIDERを使用） api_key: APIキー（省略時は環境変数LLM_API_KEYを使用） model: モデル名（省略時は環境変数LLM_MODELまたはデフォルト） """ # .envファイルから環境変数を読み込み env_path = Path.cwd() / ".env" if env_path.exists(): load_dotenv(env_path) logger.info(f".envファイルを読み込みました: {env_path}") else: logger.warning(f".envファイルが見つかりません: {env_path}") # プロバイダーの取得（デフォルト: anthropic） self.provider = provider or os.getenv("LLM_PROVIDER", "anthropic") # APIキーの取得（後方互換性のためANTHROPIC_API_KEYもチェック） self.api_key = api_key or os.getenv("LLM_API_KEY") or os.getenv("ANTHROPIC_API_KEY") # モデル名の取得（後方互換性のためCLAUDE_MODELもチェック） default_model = "claude-3-5-sonnet-20241022" if self.provider == "anthropic" else "gemini-2.0-flash-exp" self.model = model or os.getenv("LLM_MODEL") or os.getenv("CLAUDE_MODEL", default_model) # LLM APIクライアント初期化 self.client = None if self.api_key: try: if self.provider == "anthropic": self.client = AsyncAnthropic(api_key=self.api_key) logger.info(f"Anthropic Claude APIクライアントを初期化しました (model: {self.model})") elif self.provider == "google": try: import google.generativeai as genai genai.configure(api_key=self.api_key) self.client = genai.GenerativeModel(self.model) logger.info(f"Google Gemini APIクライアントを初期化しました (model: {self.model})") except ImportError: logger.error("Google Gemini APIを使用するには、google-generativeaiパッケージが必要です") logger.error("インストール: uv pip install google-generativeai") self.client = None else: logger.error(f"未対応のLLMプロバイダー: {self.provider}") logger.error("利用可能なプロバイダー: anthropic, google") except Exception as e: logger.error(f"LLM APIクライアント初期化エラー: {e}") self.client = None else: logger.warning("LLM_API_KEYが設定されていません。AI分析機能は無効化されます。") logger.warning("AI分析を有効にするには、.envファイルにLLM_PROVIDER=anthropic, LLM_API_KEY=your_api_keyを設定してください。") async def analyze_issues( self, keywords: list[dict], axis_analysis: Optional[dict] = None, comments_sample: list[str] = None ) -> dict: """課題を分析. Args: keywords: キーワードリスト axis_analysis: 分析軸データ（オプション） comments_sample: コメントサンプル（オプション） Returns: dict: 課題分析結果 - issues: 課題リスト - priority_issues: 優先度高の課題 - summary: サマリー - warning: 警告メッセージ（オプション） """ if not self.client: logger.warning("AI分析がスキップされました（APIキー未設定）") return { "issues": [], "priority_issues": [], "summary": "AI分析は無効化されています", "warning": "LLM_API_KEYが設定されていません。AI課題分析を有効にするには、プロジェクトルートに.envファイルを作成し、LLM_PROVIDER=anthropic, LLM_API_KEY=your_api_keyを設定してください。", } # プロンプト構築 prompt = self._build_analysis_prompt(keywords, axis_analysis, comments_sample) try: logger.info(f"{self.provider.upper()} API ({self.model}) で課題分析を実行中...") # プロバイダーに応じてAPI呼び出し if self.provider == "anthropic": response = await self.client.messages.create( model=self.model, max_tokens=2000, messages=[{"role": "user", "content": prompt}] ) content = response.content[0].text elif self.provider == "google": # Geminiは同期APIなので、asyncio.to_threadでブロッキング呼び出しを非同期化 import asyncio response = await asyncio.to_thread(self.client.generate_content, prompt) # レスポンスのチェック if not response.candidates: logger.error("Gemini API: レスポンスにcandidatesがありません") logger.error(f"レスポンス: {response}") raise ValueError("Gemini APIからの応答が空です") # テキストを取得 try: content = response.text except Exception as e: logger.error(f"Gemini API: response.textの取得に失敗: {e}") # 代替方法でテキストを取得 if response.candidates and response.candidates[0].content.parts: content = response.candidates[0].content.parts[0].text else: raise ValueError(f"Gemini APIからテキストを取得できません: {e}") else: raise ValueError(f"未対応のLLMプロバイダー: {self.provider}") # 結果をパース（簡易的な実装） issues = self._parse_issues(content) logger.info(f"課題分析完了: {len(issues)}件の課題を検出") return { "issues": issues, "priority_issues": [issue for issue in issues if issue.get("priority") == "高"], "summary": content, } except Exception as e: logger.error(f"課題分析に失敗: {e}") error_msg = str(e) # エラーの種類に応じて適切な警告メッセージを生成 if "authentication" in error_msg.lower() or "invalid" in error_msg.lower() or "api_key" in error_msg.lower(): warning_msg = f"{self.provider.upper()}_API_KEYが無効です。正しいAPIキーを.envファイルに設定してください。" elif "rate_limit" in error_msg.lower(): warning_msg = "APIリクエストのレート制限に達しました。しばらく時間をおいてから再実行してください。" elif "quota" in error_msg.lower() or "billing" in error_msg.lower(): warning_msg = "APIクォータまたは請求の問題が発生しています。アカウントの請求設定を確認してください。" else: warning_msg = f"AI課題分析でエラーが発生しました: {error_msg}" return { "issues": [], "priority_issues": [], "summary": f"エラー: {error_msg}", "warning": warning_msg, } async def generate_solutions(self, issues: list[dict]) -> list[dict]: """改善提案を生成. Args: issues: 課題リスト Returns: list[dict]: 改善提案リスト - issue: 課題 - solution: 改善策 - priority: 優先度 - difficulty: 実施難易度 - expected_effect: 期待効果 """ if not self.client or not issues: logger.warning("改善提案生成がスキップされました") return [] # プロンプト構築 prompt = self._build_solution_prompt(issues) try: logger.info(f"{self.provider.upper()} API ({self.model}) で改善提案を生成中...") # プロバイダーに応じてAPI呼び出し if self.provider == "anthropic": response = await self.client.messages.create( model=self.model, max_tokens=3000, messages=[{"role": "user", "content": prompt}] ) content = response.content[0].text elif self.provider == "google": # Geminiは同期APIなので、asyncio.to_threadでブロッキング呼び出しを非同期化 import asyncio response = await asyncio.to_thread(self.client.generate_content, prompt) # レスポンスのチェック if not response.candidates: logger.error("Gemini API: レスポンスにcandidatesがありません") logger.error(f"レスポンス: {response}") raise ValueError("Gemini APIからの応答が空です") # テキストを取得 try: content = response.text except Exception as e: logger.error(f"Gemini API: response.textの取得に失敗: {e}") # 代替方法でテキストを取得 if response.candidates and response.candidates[0].content.parts: content = response.candidates[0].content.parts[0].text else: raise ValueError(f"Gemini APIからテキストを取得できません: {e}") else: raise ValueError(f"未対応のLLMプロバイダー: {self.provider}") # 結果をパース solutions = self._parse_solutions(content) logger.info(f"改善提案生成完了: {len(solutions)}件") return solutions except Exception as e: logger.error(f"改善提案生成に失敗: {e}") logger.warning("改善提案の生成をスキップしました。課題分析は表示されますが、改善提案は表示されません。") return [] def _build_analysis_prompt( self, keywords: list[dict], axis_analysis: Optional[dict], comments_sample: Optional[list[str]] ) -> str: """分析用プロンプト構築. Args: keywords: キーワードリスト axis_analysis: 分析軸データ comments_sample: コメントサンプル Returns: str: プロンプト """ prompt = """あなたは組織診断の専門家です。以下の社内アンケートデータから、組織の課題を分析してください。 # キーワード分析結果 """ # キーワードトップ20を追加 for i, kw in enumerate(keywords[:20], 1): prompt += f"{i}. {kw['keyword']}: {kw['count']}回 (スコア: {kw['score']:.2f})\n" # 分析軸データがあれば追加 if axis_analysis: prompt += "\n# 分析軸別データ\n" for axis_name, result in axis_analysis.get("axes_results", {}).items(): prompt += f"\n## {axis_name}\n" categories = result.get("categories", {}) for category, data in list(categories.items())[:5]: prompt += f"- {category}: {data['count']}件\n" # コメントサンプルがあれば追加 if comments_sample: prompt += "\n# コメントサンプル（抜粋）\n" for i, comment in enumerate(comments_sample[:10], 1): prompt += f"{i}. {comment[:100]}...\n" prompt += """ # 分析タスク 上記データから、以下の観点で課題を抽出してください： 1. **ネガティブキーワード**: 不満や問題を示すキーワードから課題を特定 2. **軸間格差**: 分析軸間の差異から潜在的な問題を発見 3. **パターン認識**: コメントの傾向から組織文化の課題を抽出 # 出力形式 各課題について、以下の形式で出力してください： 課題1: [課題のタイトル] - 詳細: [具体的な説明] - 根拠: [データに基づく根拠] - 優先度: [高/中/低] - 影響範囲: [全社/特定部署/個人レベル] 課題2: ... """ return prompt def _build_solution_prompt(self, issues: list[dict]) -> str: """改善提案用プロンプト構築. Args: issues: 課題リスト Returns: str: プロンプト """ prompt = """あなたは組織開発のコンサルタントです。以下の課題に対して、具体的かつ実行可能な改善策を提案してください。 # 検出された課題 """ for i, issue in enumerate(issues[:10], 1): prompt += f"\n課題{i}: {issue.get('title', '不明')}\n" prompt += f"- 詳細: {issue.get('detail', '不明')}\n" prompt += f"- 優先度: {issue.get('priority', '中')}\n" prompt += """ # 改善提案の要件 各課題に対して、以下の要素を含む改善策を提案してください： 1. **具体的アクションプラン**: 実施可能な施策（3-5個） 2. **優先順位**: どの施策から着手すべきか 3. **実施難易度**: [高/中/低] 4. **期待効果**: 定量的・定性的な効果 5. **実施期間**: 短期（1-3ヶ月）/ 中期（3-6ヶ月）/ 長期（6ヶ月以上） # 出力形式 課題1への改善提案: 【アクションプラン】 1. [施策1] 2. [施策2] 3. [施策3] 【優先順位】 [優先度の理由] 【実施難易度】 [難易度の説明] 【期待効果】 [具体的な効果] 【実施期間】 [期間の説明] --- 課題2への改善提案: ... """ return prompt def _parse_issues(self, content: str) -> list[dict]: """課題をパース（簡易実装）. Args: content: LLM APIのレスポンス Returns: list[dict]: 課題リスト """ issues = [] lines = content.split("\n") current_issue = {} for line in lines: line = line.strip() if line.startswith("課題"): if current_issue: issues.append(current_issue) # 課題タイトル抽出 title = line.split(":", 1)[1].strip() if ":" in line else line current_issue = {"title": title} elif line.startswith("- 詳細:"): current_issue["detail"] = line.split(":", 1)[1].strip() elif line.startswith("- 優先度:"): current_issue["priority"] = line.split(":", 1)[1].strip() elif line.startswith("- 影響範囲:"): current_issue["impact"] = line.split(":", 1)[1].strip() if current_issue: issues.append(current_issue) return issues def _parse_solutions(self, content: str) -> list[dict]: """改善提案をパース（簡易実装）. Args: content: LLM APIのレスポンス Returns: list[dict]: 改善提案リスト """ solutions = [] sections = content.split("---") for section in sections: if "への改善提案" in section: solution = {"actions": [], "priority": "", "difficulty": "", "expected_effect": "", "period": ""} # アクションプラン抽出 if "【アクションプラン】" in section: action_section = section.split("【アクションプラン】")[1].split("【")[0] for line in action_section.split("\n"): if line.strip() and (line.strip()[0].isdigit() or line.strip().startswith("-")): solution["actions"].append(line.strip()) # 他の要素も同様に抽出 if "【優先順位】" in section: solution["priority"] = section.split("【優先順位】")[1].split("【")[0].strip() if "【実施難易度】" in section: solution["difficulty"] = section.split("【実施難易度】")[1].split("【")[0].strip() if "【期待効果】" in section: solution["expected_effect"] = section.split("【期待効果】")[1].split("【")[0].strip() if "【実施期間】" in section: solution["period"] = section.split("【実施期間】")[1].strip() solutions.append(solution) return solutions