Add phd_research_os/conflict_detector.py

phd_research_os/conflict_detector.py

@@ -0,0 +1,184 @@
+"""
+PhD Research OS — Conflict Detection (Phase 5)
+=================================================
+Pairwise contradiction detection across claim database.
+Uses embedding similarity + AI brain for conflict assessment.
+"""
+
+import json
+from typing import Optional
+
+from .db import get_db, search_claims, create_conflict, from_fixed
+from .agents import ResearchOSBrain
+
+
+class ConflictDetector:
+    """
+    Detect contradictions between claims in the database.
+    
+    Strategy:
+    1. Find claim pairs with high semantic similarity but different conclusions
+    2. Use AI brain to assess if they truly conflict
+    3. Generate conflict resolution hypotheses (always LOW confidence)
+    
+    Per Research OS spec:
+    - hypothesis_confidence is ALWAYS "low"
+    - Agent can NEVER auto-promote conflict resolution above Level 5
+    - Human review required for all resolutions
+    """
+    
+    def __init__(self, db_path: str = None, brain: ResearchOSBrain = None):
+        self.db_path = db_path
+        self.brain = brain
+    
+    def find_candidate_pairs(self, topic_filter: str = None,
+                             max_pairs: int = 100) -> list:
+        """
+        Find claim pairs that might conflict.
+        
+        Heuristic: Claims with overlapping keywords but from different sources.
+        (Upgradeable to embedding similarity > 0.85 when ChromaDB is set up)
+        """
+        conn = get_db(self.db_path)
+        
+        # Get all claims, optionally filtered
+        claims = search_claims(conn, query=topic_filter, limit=500)
+        conn.close()
+        
+        if len(claims) < 2:
+            return []
+        
+        pairs = []
+        seen = set()
+        
+        for i, claim_a in enumerate(claims):
+            for j, claim_b in enumerate(claims[i+1:], i+1):
+                # Skip same source
+                if claim_a.get('source_doi') == claim_b.get('source_doi'):
+                    continue
+                
+                # Simple keyword overlap heuristic
+                words_a = set(claim_a['text'].lower().split())
+                words_b = set(claim_b['text'].lower().split())
+                
+                # Remove common stopwords
+                stopwords = {'the', 'a', 'an', 'is', 'was', 'were', 'are', 'been',
+                           'be', 'have', 'has', 'had', 'do', 'does', 'did', 'will',
+                           'would', 'could', 'should', 'may', 'might', 'shall',
+                           'in', 'on', 'at', 'to', 'for', 'of', 'with', 'by',
+                           'from', 'and', 'or', 'but', 'not', 'no', 'this', 'that',
+                           'these', 'those', 'it', 'its', 'we', 'our', 'they'}
+                words_a -= stopwords
+                words_b -= stopwords
+                
+                if not words_a or not words_b:
+                    continue
+                
+                overlap = len(words_a & words_b) / min(len(words_a), len(words_b))
+                
+                if overlap >= 0.3:  # At least 30% keyword overlap
+                    pair_key = tuple(sorted([claim_a['claim_id'], claim_b['claim_id']]))
+                    if pair_key not in seen:
+                        seen.add(pair_key)
+                        pairs.append({
+                            'claim_a': claim_a,
+                            'claim_b': claim_b,
+                            'keyword_overlap': overlap
+                        })
+                
+                if len(pairs) >= max_pairs:
+                    break
+            if len(pairs) >= max_pairs:
+                break
+        
+        # Sort by overlap (highest first — most likely conflicts)
+        pairs.sort(key=lambda x: x['keyword_overlap'], reverse=True)
+        return pairs
+    
+    def detect_conflicts(self, topic_filter: str = None, 
+                         max_pairs: int = 50) -> list:
+        """
+        Run full conflict detection pipeline.
+        
+        Returns list of detected conflicts with hypotheses.
+        """
+        if self.brain is None:
+            print("Warning: No brain configured. Using keyword-only heuristic.")
+            return self._keyword_only_detection(topic_filter, max_pairs)
+        
+        pairs = self.find_candidate_pairs(topic_filter, max_pairs)
+        print(f"Found {len(pairs)} candidate pairs for conflict analysis")
+        
+        conflicts = []
+        conn = get_db(self.db_path)
+        
+        for i, pair in enumerate(pairs):
+            print(f"  Analyzing pair {i+1}/{len(pairs)}...")
+            
+            response = self.brain.detect_conflicts(
+                pair['claim_a']['text'],
+                pair['claim_b']['text']
+            )
+            
+            if response.success and response.data.get('conflict_detected', False):
+                conflict_id = create_conflict(
+                    conn,
+                    pair['claim_a']['claim_id'],
+                    pair['claim_b']['claim_id'],
+                    response.data.get('conflict_type', 'value_mismatch'),
+                    response.data.get('generated_hypothesis', ''),
+                    response.data.get('key_differences', [])
+                )
+                
+                conflicts.append({
+                    'conflict_id': conflict_id,
+                    'claim_a': pair['claim_a']['claim_id'],
+                    'claim_b': pair['claim_b']['claim_id'],
+                    'type': response.data.get('conflict_type'),
+                    'hypothesis': response.data.get('generated_hypothesis'),
+                    'confidence': 'low',  # ALWAYS low per spec
+                })
+        
+        conn.close()
+        
+        print(f"\nDetected {len(conflicts)} conflicts from {len(pairs)} candidates")
+        print(f"False positive rate estimate: check manually")
+        
+        return conflicts
+    
+    def _keyword_only_detection(self, topic_filter: str = None,
+                                max_pairs: int = 50) -> list:
+        """Fallback: keyword-only conflict detection without AI brain."""
+        pairs = self.find_candidate_pairs(topic_filter, max_pairs)
+        
+        # Without AI brain, we flag high-overlap pairs from different sources
+        # that have different epistemic tags or very different confidence
+        flagged = []
+        conn = get_db(self.db_path)
+        
+        for pair in pairs:
+            a = pair['claim_a']
+            b = pair['claim_b']
+            
+            # Flag if: high overlap + different conclusions indicators
+            confidence_diff = abs(a['confidence'] - b['confidence'])
+            different_tags = a['epistemic_tag'] != b['epistemic_tag']
+            
+            if pair['keyword_overlap'] > 0.5 and (confidence_diff > 0.3 or different_tags):
+                conflict_id = create_conflict(
+                    conn,
+                    a['claim_id'],
+                    b['claim_id'],
+                    'value_mismatch',
+                    'Flagged by keyword overlap heuristic — requires AI or human review',
+                    ['keyword_overlap > 0.5', f'confidence_diff = {confidence_diff:.2f}']
+                )
+                flagged.append({
+                    'conflict_id': conflict_id,
+                    'claim_a': a['claim_id'],
+                    'claim_b': b['claim_id'],
+                    'overlap': pair['keyword_overlap'],
+                })
+        
+        conn.close()
+        return flagged