Upload judge_beta_lobe_advanced.py with huggingface_hub

judge_beta_lobe_advanced.py

@@ -0,0 +1,483 @@
+from datetime import datetime
+import re
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+# ドメイン別の危険な主張を検出するためのパターン
+DOMAIN_DANGEROUS_PATTERNS = {
+    "medical": [
+        (r"(必ず|絶対に|確実に).*(治る|完治|治癒)", "absolute_cure_claim", "critical"),
+        (r"(副作用|リスク).*(ない|ありません|存在しない)", "no_side_effects_claim", "critical"),
+        (r"(すべての|全ての|あらゆる)患者に(有効|効果的)", "universal_effectiveness", "high"),
+        (r"(西洋|現代)医学.*(不要|いらない|無意味)", "anti_medicine_claim", "critical"),
+        (r"(自己判断|自分で).*治療", "self_treatment_encouragement", "moderate"),
+        (r"医師.*(相談|受診).*(不要|いらない|必要ない)", "avoid_doctor_claim", "critical"),
+    ],
+    "legal": [
+        (r"(必ず|絶対に|確実に).*(勝訴|勝てる|認められる)", "absolute_outcome_claim", "critical"),
+        (r"弁護士.*(不要|いらない|必要ない)", "avoid_lawyer_claim", "critical"),
+        (r"(すべての|全ての)ケースで", "universal_applicability", "high"),
+        (r"(違法|犯罪).*(ではない|にならない).*絶対", "absolute_legality_claim", "critical"),
+        (r"(時効|期限).*(気にしなくて|無視して)", "ignore_deadlines", "critical"),
+        (r"(判例|法律).*無視", "ignore_precedent", "high"),
+    ],
+    "economics": [
+        (r"(必ず|絶対に|確実に).*(儲かる|利益|リターン)", "guaranteed_profit_claim", "critical"),
+        (r"リスク.*(ない|ゼロ|存在しない)", "no_risk_claim", "critical"),
+        (r"(すべての|全ての)投資家に", "universal_advice", "high"),
+        (r"(買う|売る)べき.*絶対", "absolute_trading_advice", "critical"),
+        (r"市場.*予測.*確実", "certain_market_prediction", "high"),
+        (r"(暴落|暴騰).*(ない|しない).*絶対", "absolute_market_stability", "high"),
+    ]
+}
+
+# 後方互換性のため
+DANGEROUS_CLAIM_PATTERNS = DOMAIN_DANGEROUS_PATTERNS["medical"]
+
+# 医学的数値の妥当性範囲
+MEDICAL_VALUE_RANGES = {
+    "血圧": {"systolic": (60, 250), "diastolic": (40, 150)},
+    "体温": {"min": 35.0, "max": 42.0},
+    "心拍数": {"min": 30, "max": 220},
+    "SpO2": {"min": 70, "max": 100},
+    "血糖値": {"min": 20, "max": 600},
+}
+
+# 法学ドメインの検証パターン
+LEGAL_VALIDATION_PATTERNS = {
+    "disclaimer_required": r"(免責|情報提供|法的助言ではありません)",
+    "statute_citation": r"(第\d+条|条文|法律)",
+    "precedent_citation": r"(判例|最判|最決|高判)",
+}
+
+# 経済学ドメインの検証パターン
+ECONOMICS_VALIDATION_PATTERNS = {
+    "data_source_required": r"(統計|データ|出典|IMF|日銀|内閣府)",
+    "uncertainty_disclosure": r"(予測|推計|不確実|シナリオ)",
+    "disclaimer_required": r"(投資助言ではありません|自己責任)",
+}
+
+
+class BetaLobeAdvanced:
+    """
+    検証院（β-Lobe）の高度な機能。
+    論理的妥当性、医学的文脈の検証、ハルシネーション検出を実装。
+    """
+
+    def __init__(self, db_interface, medical_ontology):
+        self.db = db_interface
+        self.ontology = medical_ontology
+
+    # --- 基本的なAnchor事実チェック ---
+    def _is_mentioned(self, fact: str, response: str) -> bool:
+        """事実がレスポンスに言及されているか確認"""
+        fact_keywords = [word for word in fact.split() if len(word) > 1]
+        if not fact_keywords:
+            return False
+        mentioned_count = sum(1 for kw in fact_keywords if kw in response)
+        return (mentioned_count / len(fact_keywords)) > 0.5
+
+    def _detect_numerical_contradiction(self, fact: str, response: str) -> bool:
+        """数値の矛盾を検出"""
+        fact_numbers = re.findall(r'[-+]?\d*\.\d+|\d+', fact)
+        if not fact_numbers:
+            return False
+        fact_value = float(fact_numbers[0])
+        response_numbers = re.findall(r'[-+]?\d*\.\d+|\d+', response)
+        if not response_numbers:
+            return True
+        # 10%以上の乖離で矛盾とみなす
+        is_far = all(abs(float(res_val) - fact_value) / max(fact_value, 0.001) > 0.1 for res_val in response_numbers)
+        return is_far
+
+    async def check_anchor_facts(self, response_text: str, db_context: dict) -> dict:
+        """DBの知識タイルと回答の整合性を検証"""
+        contradictions = []
+
+        for coord, tile in db_context.items():
+            if not tile:
+                continue
+
+            # タイルから主要な事実を抽出
+            anchor_facts = self._extract_anchor_facts(tile)
+
+            for fact in anchor_facts:
+                # 事実が言及されているか確認
+                if self._is_mentioned(fact["statement"], response_text):
+                    # 数値の矛盾をチェック
+                    if fact.get("has_numbers") and self._detect_numerical_contradiction(fact["statement"], response_text):
+                        contradictions.append({
+                            "type": "numerical_contradiction",
+                            "fact": fact["statement"],
+                            "source": coord,
+                            "severity": "high"
+                        })
+
+        return {
+            "contradictions": contradictions,
+            "contradiction_count": len(contradictions),
+            "passed": len(contradictions) == 0
+        }
+
+    def _extract_anchor_facts(self, tile: dict) -> list:
+        """タイルから検証用の事実を抽出"""
+        facts = []
+        content = tile.get("content", "") or tile.get("data", "")
+
+        if isinstance(content, str):
+            # 箇条書きや重要な記述を抽出
+            lines = content.split("\n")
+            for line in lines:
+                line = line.strip()
+                if len(line) > 10 and any(marker in line for marker in ["は", "である", "です", "："]):
+                    has_numbers = bool(re.search(r'\d+', line))
+                    facts.append({
+                        "statement": line[:200],  # 最大200文字
+                        "has_numbers": has_numbers
+                    })
+                    if len(facts) >= 5:  # 最大5つの事実
+                        break
+
+        return facts
+
+    # --- 危険な主張の検出（ドメイン対応） ---
+    def _detect_dangerous_claims(self, response: str, domain: str = "medical") -> list:
+        """ドメイン別の危険な主張を検出"""
+        issues = []
+        patterns = DOMAIN_DANGEROUS_PATTERNS.get(domain, DOMAIN_DANGEROUS_PATTERNS["medical"])
+
+        for pattern, claim_type, severity in patterns:
+            match = re.search(pattern, response)
+            if match:
+                issues.append({
+                    "type": "dangerous_claim",
+                    "domain": domain,
+                    "claim_type": claim_type,
+                    "matched_text": match.group(0),
+                    "severity": severity,
+                    "message": f"危険な主張を検出 [{domain}]: {claim_type}"
+                })
+        return issues
+
+    # --- ドメイン固有の検証 ---
+    def _validate_legal_response(self, response: str) -> list:
+        """法学ドメイン固有の検証"""
+        issues = []
+
+        # 免責事項の確認
+        if not re.search(LEGAL_VALIDATION_PATTERNS["disclaimer_required"], response):
+            issues.append({
+                "type": "missing_disclaimer",
+                "domain": "legal",
+                "severity": "high",
+                "message": "法的免責事項が欠落しています"
+            })
+
+        # 条文引用の確認（法律質問の場合）
+        # ここでは警告レベルにとどめる
+        if not re.search(LEGAL_VALIDATION_PATTERNS["statute_citation"], response):
+            issues.append({
+                "type": "missing_citation",
+                "domain": "legal",
+                "severity": "moderate",
+                "message": "条文への参照がありません"
+            })
+
+        return issues
+
+    def _validate_economics_response(self, response: str) -> list:
+        """経済学ドメイン固有の検証"""
+        issues = []
+
+        # データ出典の確認
+        if not re.search(ECONOMICS_VALIDATION_PATTERNS["data_source_required"], response):
+            issues.append({
+                "type": "missing_data_source",
+                "domain": "economics",
+                "severity": "moderate",
+                "message": "データ出典への参照がありません"
+            })
+
+        # 予測の場合の不確実性開示
+        if "予測" in response or "見通し" in response:
+            if not re.search(ECONOMICS_VALIDATION_PATTERNS["uncertainty_disclosure"], response):
+                issues.append({
+                    "type": "missing_uncertainty_disclosure",
+                    "domain": "economics",
+                    "severity": "high",
+                    "message": "予測の不確実性が明示されていません"
+                })
+
+        return issues
+
+    # --- 医学的数値の妥当性検証 ---
+    def _validate_medical_values(self, response: str) -> list:
+        """医学的数値が妥当な範囲内か検証"""
+        issues = []
+
+        # 血圧の検出と検証
+        bp_pattern = r'(\d{2,3})/(\d{2,3})\s*(?:mmHg)?'
+        bp_matches = re.findall(bp_pattern, response)
+        for systolic, diastolic in bp_matches:
+            s, d = int(systolic), int(diastolic)
+            ranges = MEDICAL_VALUE_RANGES["血圧"]
+            if not (ranges["systolic"][0] <= s <= ranges["systolic"][1]):
+                issues.append({
+                    "type": "invalid_medical_value",
+                    "value_type": "血圧（収縮期）",
+                    "value": s,
+                    "expected_range": ranges["systolic"],
+                    "severity": "high"
+                })
+            if not (ranges["diastolic"][0] <= d <= ranges["diastolic"][1]):
+                issues.append({
+                    "type": "invalid_medical_value",
+                    "value_type": "血圧（拡張期）",
+                    "value": d,
+                    "expected_range": ranges["diastolic"],
+                    "severity": "high"
+                })
+
+        # 体温の検出と検証
+        temp_pattern = r'(\d{2}(?:\.\d)?)\s*(?:°C|度|℃)'
+        temp_matches = re.findall(temp_pattern, response)
+        for temp in temp_matches:
+            t = float(temp)
+            ranges = MEDICAL_VALUE_RANGES["体温"]
+            if not (ranges["min"] <= t <= ranges["max"]):
+                issues.append({
+                    "type": "invalid_medical_value",
+                    "value_type": "体温",
+                    "value": t,
+                    "expected_range": (ranges["min"], ranges["max"]),
+                    "severity": "high"
+                })
+
+        return issues
+
+    # --- 高度な検証機能 ---
+
+    def _detect_false_dichotomy(self, response: str) -> list:
+        """偽の二者択一を検出"""
+        errors = []
+        dichotomy_pattern = r"(AかBのいずれかしかない|AかBしかない)" # 簡易パターン
+        if re.search(dichotomy_pattern, response.replace(" ","")): # 空白除去
+             errors.append({"type": "false_dichotomy", "statement": response, "severity": "moderate"})
+        return errors
+    
+    async def _check_logical_consistency(self, question, alpha_response) -> dict:
+        """推論の論理的妥当性を検証"""
+        errors = []
+        response_text = alpha_response["main_response"]
+        
+        # 偽の二者択一を検出
+        dichotomy_errors = self._detect_false_dichotomy(response_text)
+        errors.extend(dichotomy_errors)
+        
+        # NOTE: 環状論理、論理的飛躍、根拠なき仮定の検出は高度なNLPが必要なため、
+        # ここではプレースホルダーとして成功を返す。
+        
+        return {"logical_errors": errors, "error_count": len(errors), "passed": len(errors) == 0}
+
+    async def _verify_treatment_validity(self, response_text, db_context) -> dict:
+        """治療法の妥当性を検証"""
+        issues = []
+        # 簡易的な治療法抽出
+        mentioned_treatments_regex = re.findall(r"(\S+)が良い|(\w+)が有効な治療法|(\w+)を投与", response_text)
+        # 抽出結果はタプルのリストになるため、フラット化する
+        extracted_phrases = [item for tpl in mentioned_treatments_regex for item in tpl if item]
+        
+        # 後処理で助詞などを除去し、治療法名を正確に切り出す
+        processed_treatments = []
+        for phrase in extracted_phrases:
+            if "には" in phrase:
+                processed_treatments.append(phrase.split("には")[-1])
+            elif "は" in phrase:
+                processed_treatments.append(phrase.split("は")[-1])
+            else:
+                processed_treatments.append(phrase)
+
+        for treatment in processed_treatments:
+            if not treatment: continue
+            treatment_info = await self.db.search_treatment(treatment)
+            if not treatment_info:
+                issues.append({"type": "unknown_treatment", "treatment": treatment, "severity": "moderate", "message": f"「{treatment}」は未知の治療法"})
+            elif not treatment_info.get("is_validated"):
+                issues.append({"type": "unvalidated_treatment", "treatment": treatment, "severity": "critical", "message": f"「{treatment}」は未検証の治療法"})
+        
+        return {"valid": len(issues) == 0, "issues": issues}
+
+    async def _check_medical_context(self, response_text: str, db_context: dict) -> dict:
+        """医学的コンテキストが適切か確認"""
+        issues = []
+        treatment_check = await self._verify_treatment_validity(response_text, db_context)
+        if not treatment_check["valid"]:
+            issues.extend(treatment_check["issues"])
+        
+        # NOTE: 診断基準、数値、禁忌の検証はプレースホルダー
+        return {"issues": issues, "issue_count": len(issues), "passed": len(issues) == 0}
+
+    async def validate_response(self, question: str, alpha_response: dict, db_context: dict, web_results=None, session_context=None, domain: str = "medical") -> dict:
+        """回答を多角的に検証する（基本＋高度、ドメイン対応）"""
+
+        response_text = alpha_response.get("main_response", "")
+        # alpha_responseにドメイン情報があればそちらを優先
+        domain = alpha_response.get("domain", domain)
+
+        logger.info(f"BetaLobe検証開始: domain={domain}")
+
+        # 1. 基本的なAnchor事実チェック
+        anchor_check = await self.check_anchor_facts(response_text, db_context)
+
+        # 2. 高度な論理チェック
+        logic_check = await self._check_logical_consistency(question, alpha_response)
+
+        # 3. ドメイン別の文脈チェック
+        if domain == "medical":
+            context_check = await self._check_medical_context(response_text, db_context)
+        elif domain == "legal":
+            context_issues = self._validate_legal_response(response_text)
+            context_check = {"issues": context_issues, "issue_count": len(context_issues), "passed": len(context_issues) == 0}
+        elif domain == "economics":
+            context_issues = self._validate_economics_response(response_text)
+            context_check = {"issues": context_issues, "issue_count": len(context_issues), "passed": len(context_issues) == 0}
+        else:
+            context_check = {"issues": [], "issue_count": 0, "passed": True}
+
+        # 4. ドメイン別の危険な主張の検出
+        dangerous_claims = self._detect_dangerous_claims(response_text, domain)
+        safety_check = {
+            "issues": dangerous_claims,
+            "issue_count": len(dangerous_claims),
+            "passed": len(dangerous_claims) == 0
+        }
+
+        # 5. ドメイン別の数値妥当性検証
+        if domain == "medical":
+            value_issues = self._validate_medical_values(response_text)
+        else:
+            value_issues = []  # 法学・経済学は数値検証なし（将来拡張可能）
+        value_check = {
+            "issues": value_issues,
+            "issue_count": len(value_issues),
+            "passed": len(value_issues) == 0
+        }
+
+        # 全ての問題を集約
+        all_issues = (
+            anchor_check["contradictions"] +
+            logic_check["logical_errors"] +
+            context_check["issues"] +
+            safety_check["issues"] +
+            value_check["issues"]
+        )
+
+        # 重大度を判定
+        severity = "none"
+        if any(i.get("severity") == "critical" for i in all_issues):
+            severity = "critical"
+        elif any(i.get("severity") == "high" for i in all_issues):
+            severity = "high"
+        elif any(i.get("severity") == "moderate" for i in all_issues):
+            severity = "moderate"
+
+        # ハルシネーションリスクスコアを計算
+        hallucination_risk = self._calculate_hallucination_risk(
+            alpha_response, anchor_check, logic_check, context_check, safety_check
+        )
+
+        validation_result = {
+            "timestamp": datetime.now().isoformat(),
+            "response_text": response_text[:500],  # 長い回答は切り詰め
+            "checks": {
+                "anchor_facts": anchor_check,
+                "logic": logic_check,
+                "context": context_check,
+                "safety": safety_check,
+                "medical_values": value_check
+            },
+            "all_issues": all_issues,
+            "issue_count": len(all_issues),
+            "has_contradictions": len(all_issues) > 0,
+            "severity": severity,
+            "hallucination_risk": hallucination_risk,
+            "recommendations": self._generate_recommendations(all_issues)
+        }
+
+        logger.info(f"検証完了: {len(all_issues)}件の問題, 重大度={severity}, ハルシネーションリスク={hallucination_risk['score']:.2f}")
+        return validation_result
+
+    def _calculate_hallucination_risk(self, alpha_response, anchor_check, logic_check, context_check, safety_check) -> dict:
+        """ハルシネーションリスクスコアを計算"""
+        score = 0.0
+
+        # Anchor事実との矛盾（最大0.4）
+        if not anchor_check["passed"]:
+            score += 0.4
+
+        # 論理エラー（最大0.2）
+        if not logic_check["passed"]:
+            score += 0.2
+
+        # 医学的文脈の問題（最大0.15）
+        if not context_check["passed"]:
+            score += 0.15
+
+        # 危険な主張（最大0.25）
+        if not safety_check["passed"]:
+            score += 0.25
+
+        # 信頼度が低い場合のペナルティ
+        confidence = alpha_response.get("confidence", 0.5)
+        if confidence < 0.4:
+            score += 0.1
+
+        final_score = min(1.0, score)
+
+        # リスクレベルの分類
+        if final_score < 0.1:
+            level = "very_low"
+        elif final_score < 0.25:
+            level = "low"
+        elif final_score < 0.5:
+            level = "moderate"
+        elif final_score < 0.75:
+            level = "high"
+        else:
+            level = "critical"
+
+        return {
+            "score": final_score,
+            "level": level,
+            "action_required": final_score >= 0.25
+        }
+
+    def _generate_recommendations(self, all_issues: list) -> list:
+        """問題に基づいて修正推奨を生成"""
+        recommendations = []
+
+        for issue in all_issues[:3]:  # 最大3件
+            issue_type = issue.get("type", "unknown")
+
+            if issue_type == "dangerous_claim":
+                recommendations.append({
+                    "type": "remove_dangerous_claim",
+                    "message": f"危険な主張を削除または修正: {issue.get('claim_type')}",
+                    "priority": "high"
+                })
+            elif issue_type == "numerical_contradiction":
+                recommendations.append({
+                    "type": "verify_numbers",
+                    "message": f"数値を確認: {issue.get('fact', '')[:50]}",
+                    "priority": "medium"
+                })
+            elif issue_type == "invalid_medical_value":
+                recommendations.append({
+                    "type": "correct_value",
+                    "message": f"{issue.get('value_type')}の値が範囲外: {issue.get('value')}",
+                    "priority": "high"
+                })
+
+        return recommendations