1import numpy as np
2from typing import List, Dict, Tuple
3from sklearn.metrics import ndcg_score
4
5class RetrievalMetrics:
6    """検索精度を評価する指標群"""
7    
8    def calculate_recall_at_k(self, 
9                              retrieved_docs: List[str],
10                              relevant_docs: List[str],
11                              k: int) -> float:
12        """Recall@k: k件中の関連文書の再現率"""
13        retrieved_k = set(retrieved_docs[:k])
14        relevant_set = set(relevant_docs)
15        
16        if not relevant_set:
17            return 0.0
18        
19        intersection = retrieved_k.intersection(relevant_set)
20        recall = len(intersection) / len(relevant_set)
21        
22        return recall
23    
24    def calculate_precision_at_k(self,
25                                 retrieved_docs: List[str],
26                                 relevant_docs: List[str],
27                                 k: int) -> float:
28        """Precision@k: k件中の関連文書の精度"""
29        retrieved_k = retrieved_docs[:k]
30        relevant_set = set(relevant_docs)
31        
32        if not retrieved_k:
33            return 0.0
34        
35        hits = sum(1 for doc in retrieved_k if doc in relevant_set)
36        precision = hits / len(retrieved_k)
37        
38        return precision
39    
40    def calculate_mrr(self,
41                      retrieved_docs: List[str],
42                      relevant_docs: List[str]) -> float:
43        """MRR (Mean Reciprocal Rank): 最初の関連文書の順位の逆数"""
44        relevant_set = set(relevant_docs)
45        
46        for rank, doc in enumerate(retrieved_docs, 1):
47            if doc in relevant_set:
48                return 1.0 / rank
49        
50        return 0.0
51    
52    def calculate_map(self,
53                      queries: List[str],
54                      retrieved_results: List[List[str]],
55                      relevant_results: List[List[str]]) -> float:
56        """MAP (Mean Average Precision): 平均精度"""
57        ap_scores = []
58        
59        for retrieved, relevant in zip(retrieved_results, relevant_results):
60            if not relevant:
61                continue
62            
63            relevant_set = set(relevant)
64            ap = 0.0
65            relevant_count = 0
66            
67            for rank, doc in enumerate(retrieved, 1):
68                if doc in relevant_set:
69                    relevant_count += 1
70                    precision = relevant_count / rank
71                    ap += precision
72            
73            if relevant_count > 0:
74                ap /= len(relevant_set)
75            
76            ap_scores.append(ap)
77        
78        return np.mean(ap_scores) if ap_scores else 0.0
79    
80    def calculate_ndcg(self,
81                       retrieved_docs: List[str],
82                       relevance_scores: Dict[str, float],
83                       k: int) -> float:
84        """NDCG@k: 正規化割引累積利得"""
85        retrieved_k = retrieved_docs[:k]
86        
87        # 実際の利得スコア
88        actual_scores = [relevance_scores.get(doc, 0) for doc in retrieved_k]
89        
90        # 理想的な順序での利得スコア
91        ideal_scores = sorted(relevance_scores.values(), reverse=True)[:k]
92        
93        if not ideal_scores or sum(ideal_scores) == 0:
94            return 0.0
95        
96        # NDCGを計算
97        actual_dcg = self._calculate_dcg(actual_scores)
98        ideal_dcg = self._calculate_dcg(ideal_scores)
99        
100        return actual_dcg / ideal_dcg if ideal_dcg > 0 else 0.0
101    
102    def _calculate_dcg(self, scores: List[float]) -> float:
103        """DCG (Discounted Cumulative Gain) を計算"""
104        dcg = 0.0
105        for i, score in enumerate(scores, 1):
106            dcg += score / np.log2(i + 1)
107        return dcg

1from rouge_score import rouge_scorer
2from bert_score import score as bert_score
3import torch
4from transformers import AutoTokenizer, AutoModelForSequenceClassification
5
6class GenerationMetrics:
7    """生成品質を評価する指標群"""
8    
9    def __init__(self):
10        self.rouge_scorer = rouge_scorer.RougeScorer(
11            ['rouge1', 'rouge2', 'rougeL'], use_stemmer=True
12        )
13        # 日本語対応のBERTモデル
14        self.tokenizer = AutoTokenizer.from_pretrained('cl-tohoku/bert-base-japanese')
15        self.model = AutoModelForSequenceClassification.from_pretrained(
16            'cl-tohoku/bert-base-japanese'
17        )
18    
19    def calculate_rouge_scores(self, 
20                              generated: str,
21                              reference: str) -> Dict[str, float]:
22        """ROUGE スコア: n-gram重複に基づく評価"""
23        scores = self.rouge_scorer.score(reference, generated)
24        
25        return {
26            'rouge1_f1': scores['rouge1'].fmeasure,
27            'rouge2_f1': scores['rouge2'].fmeasure,
28            'rougeL_f1': scores['rougeL'].fmeasure,
29            'rouge1_precision': scores['rouge1'].precision,
30            'rouge1_recall': scores['rouge1'].recall
31        }
32    
33    def calculate_bert_score(self,
34                            generated: List[str],
35                            references: List[str]) -> Dict[str, float]:
36        """BERTScore: 文脈を考慮した意味的類似度"""
37        P, R, F1 = bert_score(
38            generated, 
39            references,
40            lang='ja',
41            model_type='cl-tohoku/bert-base-japanese'
42        )
43        
44        return {
45            'bert_precision': P.mean().item(),
46            'bert_recall': R.mean().item(),
47            'bert_f1': F1.mean().item()
48        }
49    
50    def calculate_fluency_score(self, text: str) -> float:
51        """流暢性スコア: 文章の自然さを評価"""
52        # パープレキシティベースの評価
53        inputs = self.tokenizer(text, return_tensors='pt', truncation=True)
54        
55        with torch.no_grad():
56            outputs = self.model(**inputs)
57            # logitsから流暢性スコアを計算
58            fluency_score = torch.softmax(outputs.logits, dim=-1).max().item()
59        
60        return fluency_score
61    
62    def calculate_coherence_score(self, text: str) -> float:
63        """一貫性スコア: 文章の論理的一貫性を評価"""
64        sentences = text.split('。')
65        if len(sentences) < 2:
66            return 1.0
67        
68        coherence_scores = []
69        
70        for i in range(len(sentences) - 1):
71            if not sentences[i] or not sentences[i+1]:
72                continue
73            
74            # 隣接する文の意味的類似度を計算
75            inputs1 = self.tokenizer(sentences[i], return_tensors='pt', truncation=True)
76            inputs2 = self.tokenizer(sentences[i+1], return_tensors='pt', truncation=True)
77            
78            with torch.no_grad():
79                emb1 = self.model.bert(**inputs1).last_hidden_state.mean(dim=1)
80                emb2 = self.model.bert(**inputs2).last_hidden_state.mean(dim=1)
81                
82                # コサイン類似度
83                similarity = torch.cosine_similarity(emb1, emb2).item()
84                coherence_scores.append(similarity)
85        
86        return np.mean(coherence_scores) if coherence_scores else 0.0

1class RAGSpecificMetrics:
2    """RAG特有の評価指標"""
3    
4    def __init__(self, llm_client=None):
5        self.llm_client = llm_client
6    
7    def calculate_faithfulness(self,
8                              answer: str,
9                              context: str,
10                              use_llm: bool = True) -> float:
11        """忠実性: 回答がコンテキストに基づいているか"""
12        if use_llm and self.llm_client:
13            prompt = f"""
14            以下の回答がコンテキストの情報に忠実に基づいているか評価してください。
15            
16            コンテキスト: {context}
17            回答: {answer}
18            
19            評価基準:
20            1.0: 完全に忠実（すべての情報がコンテキストに基づく）
21            0.8: ほぼ忠実（わずかな推論を含むが妥当）
22            0.5: 部分的に忠実（一部の情報が不正確）
23            0.0: 不忠実（コンテキストと矛盾または無関係）
24            
25            スコアのみを数値で回答してください。
26            """
27            
28            response = self.llm_client.generate(prompt)
29            try:
30                return float(response.strip())
31            except:
32                return 0.0
33        else:
34            # 簡易的なルールベース評価
35            context_words = set(context.split())
36            answer_words = set(answer.split())
37            
38            if not answer_words:
39                return 0.0
40            
41            overlap = len(context_words.intersection(answer_words))
42            return min(overlap / len(answer_words), 1.0)
43    
44    def calculate_answer_relevance(self,
45                                   question: str,
46                                   answer: str,
47                                   use_llm: bool = True) -> float:
48        """回答関連性: 質問に対する回答の適切さ"""
49        if use_llm and self.llm_client:
50            prompt = f"""
51            質問に対する回答の関連性を評価してください。
52            
53            質問: {question}
54            回答: {answer}
55            
56            評価基準:
57            1.0: 完全に関連（質問に直接的かつ完全に回答）
58            0.7: 高い関連性（質問の主要部分に回答）
59            0.4: 部分的関連（一部のみ回答）
60            0.0: 無関連（質問に答えていない）
61            
62            スコアのみを数値で回答してください。
63            """
64            
65            response = self.llm_client.generate(prompt)
66            try:
67                return float(response.strip())
68            except:
69                return 0.0
70        else:
71            # 簡易的なキーワードマッチング
72            question_words = set(question.lower().split())
73            answer_words = set(answer.lower().split())
74            
75            overlap = len(question_words.intersection(answer_words))
76            return min(overlap / max(len(question_words), 1), 1.0)
77    
78    def calculate_context_precision(self,
79                                    retrieved_contexts: List[str],
80                                    relevant_contexts: List[str]) -> float:
81        """コンテキスト精度: 取得した文脈の正確性"""
82        if not retrieved_contexts:
83            return 0.0
84        
85        relevant_set = set(relevant_contexts)
86        precision_scores = []
87        
88        for i, context in enumerate(retrieved_contexts):
89            if context in relevant_set:
90                # 順位を考慮した精度
91                precision_scores.append(1.0 / (i + 1))
92        
93        return sum(precision_scores) / len(retrieved_contexts)
94    
95    def calculate_hallucination_score(self,
96                                     answer: str,
97                                     context: str) -> float:
98        """幻覚スコア: 存在しない情報の生成度合い"""
99        # 回答に含まれる固有名詞や数値を抽出
100        import re
101        
102        # 数値の抽出
103        answer_numbers = set(re.findall(r'd+.?d*', answer))
104        context_numbers = set(re.findall(r'd+.?d*', context))
105        
106        # 固有名詞の簡易抽出（大文字で始まる単語）
107        answer_entities = set(re.findall(r'[A-Z][a-z]+', answer))
108        context_entities = set(re.findall(r'[A-Z][a-z]+', context))
109        
110        hallucination_count = 0
111        total_count = 0
112        
113        # 数値の幻覚チェック
114        for num in answer_numbers:
115            total_count += 1
116            if num not in context_numbers:
117                hallucination_count += 1
118        
119        # 固有名詞の幻覚チェック
120        for entity in answer_entities:
121            total_count += 1
122            if entity not in context_entities:
123                hallucination_count += 1
124        
125        if total_count == 0:
126            return 0.0
127        
128        # 幻覚スコア（低いほど良い）
129        return hallucination_count / total_count

1from ragas import evaluate
2from ragas.metrics import (
3    faithfulness,
4    answer_relevancy,
5    context_precision,
6    context_recall,
7    answer_correctness,
8    answer_similarity
9)
10from datasets import Dataset
11import pandas as pd
12
13class RagasEvaluator:
14    """Ragasを使用した包括的評価"""
15    
16    def __init__(self):
17        self.metrics = [
18            faithfulness,
19            answer_relevancy,
20            context_precision,
21            context_recall,
22            answer_correctness,
23            answer_similarity
24        ]
25    
26    def prepare_dataset(self, 
27                        questions: List[str],
28                        answers: List[str],
29                        contexts: List[List[str]],
30                        ground_truths: List[str] = None) -> Dataset:
31        """評価用データセットの準備"""
32        data = {
33            'question': questions,
34            'answer': answers,
35            'contexts': contexts
36        }
37        
38        if ground_truths:
39            data['ground_truth'] = ground_truths
40        
41        df = pd.DataFrame(data)
42        return Dataset.from_pandas(df)
43    
44    def evaluate_rag_system(self, dataset: Dataset) -> Dict[str, float]:
45        """RAGシステムの総合評価"""
46        results = evaluate(
47            dataset,
48            metrics=self.metrics
49        )
50        
51        # 結果の整形
52        evaluation_results = {
53            'faithfulness': results['faithfulness'],
54            'answer_relevancy': results['answer_relevancy'],
55            'context_precision': results['context_precision'],
56            'context_recall': results['context_recall'],
57            'answer_correctness': results.get('answer_correctness', None),
58            'answer_similarity': results.get('answer_similarity', None),
59            'overall_score': self._calculate_overall_score(results)
60        }
61        
62        return evaluation_results
63    
64    def _calculate_overall_score(self, results: Dict) -> float:
65        """総合スコアの計算"""
66        weights = {
67            'faithfulness': 0.25,
68            'answer_relevancy': 0.25,
69            'context_precision': 0.20,
70            'context_recall': 0.20,
71            'answer_correctness': 0.10
72        }
73        
74        score = 0.0
75        total_weight = 0.0
76        
77        for metric, weight in weights.items():
78            if metric in results and results[metric] is not None:
79                score += results[metric] * weight
80                total_weight += weight
81        
82        return score / total_weight if total_weight > 0 else 0.0
83    
84    def generate_report(self, results: Dict[str, float]) -> str:
85        """評価レポートの生成"""
86        report = "=" * 50 + "
87"
88        report += "RAGシステム評価レポート
89"
90        report += "=" * 50 + "
91
92"
93        
94        for metric, value in results.items():
95            if value is not None:
96                status = self._get_status(metric, value)
97                report += f"{metric:20s}: {value:.3f} [{status}]
98"
99        
100        report += "
101" + "-" * 50 + "
102"
103        report += "推奨改善アクション:
104"
105        report += self._generate_recommendations(results)
106        
107        return report
108    
109    def _get_status(self, metric: str, value: float) -> str:
110        """メトリクスのステータス判定"""
111        thresholds = {
112            'excellent': 0.9,
113            'good': 0.7,
114            'fair': 0.5,
115            'poor': 0.0
116        }
117        
118        if value >= thresholds['excellent']:
119            return "優秀"
120        elif value >= thresholds['good']:
121            return "良好"
122        elif value >= thresholds['fair']:
123            return "要改善"
124        else:
125            return "要対策"
126    
127    def _generate_recommendations(self, results: Dict) -> str:
128        """改善推奨事項の生成"""
129        recommendations = []
130        
131        if results.get('faithfulness', 1.0) < 0.7:
132            recommendations.append(
133                "- 忠実性が低い: グラウンディング強化、プロンプト改善を検討"
134            )
135        
136        if results.get('answer_relevancy', 1.0) < 0.7:
137            recommendations.append(
138                "- 回答関連性が低い: クエリ理解の改善、意図分類の導入を検討"
139            )
140        
141        if results.get('context_precision', 1.0) < 0.7:
142            recommendations.append(
143                "- コンテキスト精度が低い: 検索アルゴリズムの改善、リランキング導入を検討"
144            )
145        
146        if results.get('context_recall', 1.0) < 0.7:
147            recommendations.append(
148                "- コンテキスト再現率が低い: インデックス戦略の見直し、チャンクサイズ調整を検討"
149            )
150        
151        return "
152".join(recommendations) if recommendations else "現在、良好な性能です。"

1from trulens_eval import Tru, Feedback, TruLlama
2from trulens_eval.feedback import Groundedness
3from trulens_eval.feedback.provider import OpenAI as TruOpenAI
4import numpy as np
5
6class TruLensAnalyzer:
7    """TruLensを使用した詳細分析"""
8    
9    def __init__(self, openai_key: str):
10        self.tru = Tru()
11        self.provider = TruOpenAI(api_key=openai_key)
12        self.groundedness = Groundedness(groundedness_provider=self.provider)
13    
14    def setup_feedback_functions(self):
15        """フィードバック関数の設定"""
16        # コンテキスト関連性
17        f_context_relevance = Feedback(
18            self.provider.relevance_with_cot_reasons
19        ).on_input().on(
20            TruLlama.select_source_nodes().node.text
21        ).aggregate(np.mean)
22        
23        # 回答関連性
24        f_answer_relevance = Feedback(
25            self.provider.relevance_with_cot_reasons
26        ).on_input_output()
27        
28        # グラウンディング
29        f_groundedness = Feedback(
30            self.groundedness.groundedness_measure_with_cot_reasons
31        ).on(
32            TruLlama.select_source_nodes().node.text.collect()
33        ).on_output().aggregate(self.groundedness.grounded_statements_aggregator)
34        
35        return [f_context_relevance, f_answer_relevance, f_groundedness]
36    
37    def analyze_conversation(self, 
38                            query: str,
39                            response: str,
40                            contexts: List[str]) -> Dict:
41        """会話の詳細分析"""
42        feedbacks = self.setup_feedback_functions()
43        
44        # 各フィードバック関数を実行
45        results = {}
46        for feedback in feedbacks:
47            score = feedback.run(
48                input=query,
49                output=response,
50                contexts=contexts
51            )
52            results[feedback.name] = score
53        
54        # 詳細な分析結果
55        analysis = {
56            'scores': results,
57            'weak_points': self._identify_weak_points(results),
58            'suggestions': self._generate_suggestions(results)
59        }
60        
61        return analysis
62    
63    def _identify_weak_points(self, scores: Dict) -> List[str]:
64        """弱点の特定"""
65        weak_points = []
66        
67        for metric, score in scores.items():
68            if score < 0.6:
69                weak_points.append(f"{metric}: {score:.2f}")
70        
71        return weak_points
72    
73    def _generate_suggestions(self, scores: Dict) -> List[str]:
74        """改善提案の生成"""
75        suggestions = []
76        
77        if scores.get('context_relevance', 1.0) < 0.7:
78            suggestions.append("検索クエリの改善またはインデックス戦略の見直し")
79        
80        if scores.get('answer_relevance', 1.0) < 0.7:
81            suggestions.append("プロンプトエンジニアリングの改善")
82        
83        if scores.get('groundedness', 1.0) < 0.7:
84            suggestions.append("ハルシネーション対策の強化")
85        
86        return suggestions

1import scipy.stats as stats
2from datetime import datetime
3import json
4
5class RAGABTester:
6    """RAGシステムのA/Bテスト"""
7    
8    def __init__(self):
9        self.experiments = {}
10        self.results = {}
11    
12    def create_experiment(self,
13                         experiment_id: str,
14                         variant_a_config: Dict,
15                         variant_b_config: Dict,
16                         metrics_to_track: List[str]):
17        """実験の作成"""
18        self.experiments[experiment_id] = {
19            'id': experiment_id,
20            'variant_a': variant_a_config,
21            'variant_b': variant_b_config,
22            'metrics': metrics_to_track,
23            'start_time': datetime.now(),
24            'data_a': [],
25            'data_b': []
26        }
27    
28    def record_result(self,
29                      experiment_id: str,
30                      variant: str,
31                      metrics: Dict):
32        """結果の記録"""
33        if experiment_id not in self.experiments:
34            raise ValueError(f"Experiment {experiment_id} not found")
35        
36        data_key = f'data_{variant}'
37        self.experiments[experiment_id][data_key].append({
38            'timestamp': datetime.now(),
39            'metrics': metrics
40        })
41    
42    def analyze_experiment(self, 
43                          experiment_id: str,
44                          confidence_level: float = 0.95) -> Dict:
45        """実験結果の分析"""
46        exp = self.experiments[experiment_id]
47        
48        results = {}
49        for metric in exp['metrics']:
50            # 各バリアントのデータを抽出
51            data_a = [d['metrics'][metric] for d in exp['data_a'] 
52                     if metric in d['metrics']]
53            data_b = [d['metrics'][metric] for d in exp['data_b'] 
54                     if metric in d['metrics']]
55            
56            if len(data_a) < 2 or len(data_b) < 2:
57                results[metric] = {
58                    'status': 'insufficient_data',
59                    'message': 'データ不足'
60                }
61                continue
62            
63            # t検定の実施
64            t_stat, p_value = stats.ttest_ind(data_a, data_b)
65            
66            # 効果量（Cohen's d）の計算
67            cohens_d = self._calculate_cohens_d(data_a, data_b)
68            
69            # 信頼区間の計算
70            ci_a = stats.t.interval(
71                confidence_level, 
72                len(data_a)-1,
73                loc=np.mean(data_a),
74                scale=stats.sem(data_a)
75            )
76            ci_b = stats.t.interval(
77                confidence_level,
78                len(data_b)-1,
79                loc=np.mean(data_b),
80                scale=stats.sem(data_b)
81            )
82            
83            results[metric] = {
84                'mean_a': np.mean(data_a),
85                'mean_b': np.mean(data_b),
86                'std_a': np.std(data_a),
87                'std_b': np.std(data_b),
88                'p_value': p_value,
89                'cohens_d': cohens_d,
90                'ci_a': ci_a,
91                'ci_b': ci_b,
92                'significant': p_value < (1 - confidence_level),
93                'winner': 'A' if np.mean(data_a) > np.mean(data_b) else 'B',
94                'improvement': abs(np.mean(data_b) - np.mean(data_a)) / np.mean(data_a) * 100
95            }
96        
97        return results
98    
99    def _calculate_cohens_d(self, group1: List[float], group2: List[float]) -> float:
100        """Cohen's d（効果量）の計算"""
101        n1, n2 = len(group1), len(group2)
102        var1, var2 = np.var(group1, ddof=1), np.var(group2, ddof=1)
103        
104        # プールされた標準偏差
105        pooled_std = np.sqrt(((n1 - 1) * var1 + (n2 - 1) * var2) / (n1 + n2 - 2))
106        
107        # Cohen's d
108        d = (np.mean(group1) - np.mean(group2)) / pooled_std
109        
110        return d
111    
112    def generate_experiment_report(self, experiment_id: str) -> str:
113        """実験レポートの生成"""
114        results = self.analyze_experiment(experiment_id)
115        exp = self.experiments[experiment_id]
116        
117        report = f"""
118A/Bテスト結果レポート
119実験ID: {experiment_id}
120期間: {exp['start_time']} - {datetime.now()}
121サンプル数: A={len(exp['data_a'])}, B={len(exp['data_b'])}
122
123結果サマリー:
124"""
125        
126        for metric, result in results.items():
127            if result.get('status') == 'insufficient_data':
128                report += f"
129{metric}: データ不足により評価不可"
130                continue
131            
132            report += f"""
133メトリクス: {metric}
134  バリアントA: {result['mean_a']:.3f} ± {result['std_a']:.3f}
135  バリアントB: {result['mean_b']:.3f} ± {result['std_b']:.3f}
136  p値: {result['p_value']:.4f}
137  効果量: {result['cohens_d']:.3f}
138  統計的有意性: {'あり' if result['significant'] else 'なし'}
139  優勢: バリアント{result['winner']}
140  改善率: {result['improvement']:.1f}%
141"""
142        
143        # 推奨事項
144        report += "
145推奨事項:
146"
147        significant_improvements = [
148            m for m, r in results.items() 
149            if r.get('significant') and r.get('winner') == 'B'
150        ]
151        
152        if significant_improvements:
153            report += f"バリアントBが以下のメトリクスで有意に優れています: {', '.join(significant_improvements)}
154"
155            report += "バリアントBの採用を推奨します。"
156        else:
157            report += "現時点で明確な優位性は確認されていません。実験の継続を推奨します。"
158        
159        return report

1from datetime import datetime, timedelta
2import plotly.graph_objects as go
3from plotly.subplots import make_subplots
4
5class RAGMonitoringDashboard:
6    """RAGシステムのモニタリングダッシュボード"""
7    
8    def __init__(self):
9        self.metrics_history = []
10        self.alert_thresholds = {
11            'faithfulness': 0.7,
12            'answer_relevancy': 0.7,
13            'latency': 3.0,  # 秒
14            'error_rate': 0.05
15        }
16    
17    def collect_metrics(self, rag_system) -> Dict:
18        """メトリクスの収集"""
19        current_metrics = {
20            'timestamp': datetime.now(),
21            'faithfulness': rag_system.get_faithfulness(),
22            'answer_relevancy': rag_system.get_answer_relevancy(),
23            'latency': rag_system.get_average_latency(),
24            'throughput': rag_system.get_throughput(),
25            'error_rate': rag_system.get_error_rate(),
26            'user_satisfaction': rag_system.get_user_satisfaction()
27        }
28        
29        self.metrics_history.append(current_metrics)
30        
31        # アラートチェック
32        alerts = self.check_alerts(current_metrics)
33        if alerts:
34            self.send_alerts(alerts)
35        
36        return current_metrics
37    
38    def check_alerts(self, metrics: Dict) -> List[str]:
39        """アラート条件のチェック"""
40        alerts = []
41        
42        for metric, threshold in self.alert_thresholds.items():
43            if metric in metrics:
44                if metric in ['latency', 'error_rate']:
45                    # これらは低い方が良い
46                    if metrics[metric] > threshold:
47                        alerts.append(
48                            f"⚠️ {metric}が閾値を超えています: "
49                            f"{metrics[metric]:.3f} > {threshold}"
50                        )
51                else:
52                    # これらは高い方が良い
53                    if metrics[metric] < threshold:
54                        alerts.append(
55                            f"⚠️ {metric}が閾値を下回っています: "
56                            f"{metrics[metric]:.3f} < {threshold}"
57                        )
58        
59        return alerts
60    
61    def create_dashboard(self) -> go.Figure:
62        """ダッシュボードの作成"""
63        if len(self.metrics_history) < 2:
64            return None
65        
66        # データの準備
67        timestamps = [m['timestamp'] for m in self.metrics_history]
68        
69        # サブプロットの作成
70        fig = make_subplots(
71            rows=3, cols=2,
72            subplot_titles=(
73                '忠実性スコア', '回答関連性',
74                'レイテンシ (秒)', 'スループット (req/min)',
75                'エラー率 (%)', 'ユーザー満足度'
76            )
77        )
78        
79        # 忠実性
80        fig.add_trace(
81            go.Scatter(
82                x=timestamps,
83                y=[m['faithfulness'] for m in self.metrics_history],
84                name='忠実性',
85                line=dict(color='blue')
86            ),
87            row=1, col=1
88        )
89        
90        # 回答関連性
91        fig.add_trace(
92            go.Scatter(
93                x=timestamps,
94                y=[m['answer_relevancy'] for m in self.metrics_history],
95                name='関連性',
96                line=dict(color='green')
97            ),
98            row=1, col=2
99        )
100        
101        # レイテンシ
102        fig.add_trace(
103            go.Scatter(
104                x=timestamps,
105                y=[m['latency'] for m in self.metrics_history],
106                name='レイテンシ',
107                line=dict(color='red')
108            ),
109            row=2, col=1
110        )
111        
112        # スループット
113        fig.add_trace(
114            go.Scatter(
115                x=timestamps,
116                y=[m['throughput'] for m in self.metrics_history],
117                name='スループット',
118                line=dict(color='purple')
119            ),
120            row=2, col=2
121        )
122        
123        # エラー率
124        fig.add_trace(
125            go.Scatter(
126                x=timestamps,
127                y=[m['error_rate'] * 100 for m in self.metrics_history],
128                name='エラー率',
129                line=dict(color='orange')
130            ),
131            row=3, col=1
132        )
133        
134        # ユーザー満足度
135        fig.add_trace(
136            go.Scatter(
137                x=timestamps,
138                y=[m['user_satisfaction'] for m in self.metrics_history],
139                name='満足度',
140                line=dict(color='teal')
141            ),
142            row=3, col=2
143        )
144        
145        # レイアウトの更新
146        fig.update_layout(
147            title='RAGシステム パフォーマンスダッシュボード',
148            showlegend=False,
149            height=900
150        )
151        
152        return fig
153    
154    def generate_daily_report(self) -> str:
155        """日次レポートの生成"""
156        today = datetime.now().date()
157        today_metrics = [
158            m for m in self.metrics_history
159            if m['timestamp'].date() == today
160        ]
161        
162        if not today_metrics:
163            return "本日のデータがありません。"
164        
165        report = f"""
166RAGシステム 日次パフォーマンスレポート
167日付: {today}
168
169【サマリー】
170総リクエスト数: {len(today_metrics)}
171平均忠実性: {np.mean([m['faithfulness'] for m in today_metrics]):.3f}
172平均関連性: {np.mean([m['answer_relevancy'] for m in today_metrics]):.3f}
173平均レイテンシ: {np.mean([m['latency'] for m in today_metrics]):.2f}秒
174エラー率: {np.mean([m['error_rate'] for m in today_metrics]) * 100:.2f}%
175
176【時間帯別パフォーマンス】
177"""
178        
179        # 時間帯別の集計
180        hourly_stats = {}
181        for metric in today_metrics:
182            hour = metric['timestamp'].hour
183            if hour not in hourly_stats:
184                hourly_stats[hour] = []
185            hourly_stats[hour].append(metric)
186        
187        for hour in sorted(hourly_stats.keys()):
188            hour_metrics = hourly_stats[hour]
189            report += f"{hour:02d}:00-{hour+1:02d}:00 - "
190            report += f"リクエスト: {len(hour_metrics)}, "
191            report += f"平均レイテンシ: {np.mean([m['latency'] for m in hour_metrics]):.2f}秒
192"
193        
194        return report

1# I社向けカスタム評価システム
2class FinancialRAGEvaluator:
3    def __init__(self):
4        self.domain_metrics = {
5            'regulatory_compliance': 0.95,  # 規制遵守精度
6            'numerical_accuracy': 0.99,     # 数値精度
7            'source_traceability': 1.0      # ソース追跡性
8        }
9    
10    def evaluate_financial_rag(self, qa_pairs):
11        results = {
12            'standard_metrics': self.calculate_standard_metrics(qa_pairs),
13            'domain_metrics': self.calculate_domain_metrics(qa_pairs),
14            'risk_assessment': self.assess_risks(qa_pairs)
15        }
16        return results