Chapter 29: Common Pitfalls, Bias, and Failure Modes¶

import anthropic
from typing import List, Dict

def confirmation_bias_audit(hypothesis: str, evidence_list: List[Dict]) -> str:
    """
    Use AI to identify potential confirmation bias in evidence interpretation
    """
    client = anthropic.Anthropic()

    evidence_text = "\n".join([
        f"Evidence {i+1} [{e.get('label', 'Unlabeled')}]: {e.get('description', '')}"
        for i, e in enumerate(evidence_list)
    ])

    prompt = f"""You are a cognitive bias auditor reviewing an OSINT investigation.

HYPOTHESIS UNDER INVESTIGATION:
{hypothesis}

EVIDENCE COLLECTED:
{evidence_text}

Audit this investigation for confirmation bias:

1. EVIDENCE BALANCE
   - Is there a roughly equal effort to find disconfirming evidence?
   - Are all evidence items directly related to the hypothesis, or are some tangential?

2. INTERPRETATION SCRUTINY
   - For each evidence item, what is the MOST CHARITABLE alternative interpretation?
   - Which evidence items are ambiguous but being treated as confirmation?

3. MISSING EVIDENCE
   - What evidence should exist if the hypothesis were TRUE that hasn't been found?
   - What evidence would you expect to find if the hypothesis were FALSE?
   - Has the investigator looked for this disconfirming evidence?

4. BASE RATE NEGLECT
   - How common are the indicators cited compared to the population?
   - Would these same indicators appear in innocent explanations?

5. VERDICT
   - Rate the confirmation bias risk: LOW / MEDIUM / HIGH
   - What specific steps should be taken to stress-test this hypothesis?

Be direct. If the investigation appears biased, say so clearly."""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1500,
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content[0].text


def devils_advocate_analysis(hypothesis: str, supporting_evidence: List[str]) -> str:
    """
    Force explicit consideration of alternative hypotheses
    """
    client = anthropic.Anthropic()

    evidence_text = "\n".join([f"- {e}" for e in supporting_evidence])

    prompt = f"""You are playing devil's advocate on an OSINT investigation.

CURRENT HYPOTHESIS:
{hypothesis}

EVIDENCE CITED IN SUPPORT:
{evidence_text}

Your role is to CHALLENGE this hypothesis as vigorously as possible:

1. ALTERNATIVE HYPOTHESES
   - List 3-5 alternative explanations that would produce the same evidence
   - For each alternative, how does it fit the evidence?
   - Which alternative is most likely if the primary hypothesis is wrong?

2. EVIDENCE WEAKNESSES
   - What are the limitations of each piece of evidence?
   - What could explain each evidence item innocently?
   - How could the evidence have been fabricated or manipulated?

3. MISSING CONTEXT
   - What context would change the interpretation of this evidence?
   - What is the base rate of these indicators in the target population?

4. STRONGEST COUNTERARGUMENT
   - What is the single strongest argument AGAINST the hypothesis?

5. REQUIRED ADDITIONAL EVIDENCE
   - What evidence, if found, would definitively confirm OR rule out the hypothesis?

Do not be gentle. The goal is to identify weaknesses before they become publication errors."""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1500,
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content[0].text

def base_rate_analysis(indicator: str, target_population: str,
                       indicator_prevalence: str) -> str:
    """
    Structured analysis to counter base rate neglect
    """
    client = anthropic.Anthropic()

    prompt = f"""Analyze the evidential value of an indicator, accounting for base rates.

INDICATOR: {indicator}
TARGET POPULATION: {target_population}
INDICATOR PREVALENCE ESTIMATE: {indicator_prevalence}

Apply Bayesian reasoning:

1. BASE RATE ANALYSIS
   - In the general population of {target_population}, how common is this indicator?
   - Is the investigator treating a common indicator as if it were rare?

2. LIKELIHOOD RATIO
   - How much more common is this indicator in [target subjects] vs. [innocent subjects]?
   - What is the actual evidential weight of this indicator?

3. PRIOR PROBABILITY
   - Before this indicator, what was the base rate probability of the hypothesis being true?
   - Has the investigator established a meaningful prior?

4. CONCLUSION
   - What is the appropriate weight to assign this indicator?
   - Is it being over-weighted or under-weighted relative to its evidential value?

Provide concrete numbers and reasoning, not just qualitative assessment."""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1000,
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content[0].text

def source_coverage_audit(investigation_topic: str, sources_consulted: List[str]) -> Dict:
    """
    Audit the completeness of source coverage for an investigation
    """
    # Master source checklist by category
    COMPLETE_SOURCE_CHECKLIST = {
        "corporate_entity": [
            "OpenCorporates",
            "Secretary of State (each relevant state)",
            "SEC EDGAR (if US public company)",
            "UK Companies House (if UK)",
            "EU corporate registries (if EU)",
            "ICIJ Offshore Leaks Database",
            "OpenSanctions",
            "OFAC SDN List",
            "FINRA BrokerCheck (if financial services)",
        ],
        "individual": [
            "LinkedIn",
            "Twitter/X",
            "Facebook (public content)",
            "Google Search (full name variants)",
            "Court records (PACER, CourtListener)",
            "Property records (county assessor)",
            "Voter registration (where public)",
            "Professional license boards",
            "FINRA BrokerCheck",
            "SEC enforcement actions",
            "OpenCorporates (officer search)",
            "ICIJ Offshore Leaks",
            "OpenSanctions",
        ],
        "domain_network": [
            "WHOIS (current and historical)",
            "Passive DNS",
            "Certificate Transparency logs",
            "Shodan",
            "VirusTotal",
            "URLhaus",
            "Censys",
        ],
        "financial_crime": [
            "OFAC SDN list",
            "OpenSanctions",
            "ICIJ Offshore Leaks",
            "SEC enforcement actions",
            "FinCEN advisories",
            "FATF country assessments",
            "Blockchain explorers (if crypto)",
        ]
    }

    # Determine relevant checklist
    relevant_sources = set()
    for category, sources in COMPLETE_SOURCE_CHECKLIST.items():
        # Simple keyword matching — in production, more sophisticated
        if any(keyword in investigation_topic.lower() for keyword in category.split('_')):
            relevant_sources.update(sources)

    if not relevant_sources:
        relevant_sources = set(COMPLETE_SOURCE_CHECKLIST['individual'])  # Default

    consulted_normalized = set(s.lower() for s in sources_consulted)
    relevant_normalized = {s.lower(): s for s in relevant_sources}

    unchecked = {original for norm, original in relevant_normalized.items()
                if norm not in consulted_normalized}

    coverage_pct = (len(relevant_sources) - len(unchecked)) / len(relevant_sources) * 100

    return {
        'total_relevant_sources': len(relevant_sources),
        'sources_consulted': len(sources_consulted),
        'coverage_percentage': coverage_pct,
        'unchecked_sources': list(unchecked),
        'assessment': 'ADEQUATE' if coverage_pct > 75 else ('PARTIAL' if coverage_pct > 50 else 'INSUFFICIENT'),
        'recommendation': f"Check {len(unchecked)} additional sources before concluding" if unchecked else "Coverage appears adequate"
    }

def entity_disambiguation_check(entity_name: str, entity_type: str,
                                 known_attributes: Dict) -> str:
    """
    Generate disambiguation questions to prevent entity confusion
    """
    client = anthropic.Anthropic()

    attributes_text = "\n".join([f"- {k}: {v}" for k, v in known_attributes.items()])

    prompt = f"""You are helping an OSINT investigator avoid entity confusion.

ENTITY BEING INVESTIGATED:
Name: {entity_name}
Type: {entity_type}
Known Attributes:
{attributes_text}

Generate a disambiguation checklist:

1. COMMON CONFUSION RISKS
   - How common is this name? (very common names create high confusion risk)
   - What other entities might share this name or address?
   - Are there corporate name changes or aliases to consider?

2. REQUIRED VERIFICATION FIELDS
   - What specific identifiers are needed to confirm each data item belongs to THIS entity?
   - For a company: jurisdiction + registration number + incorporation date minimum
   - For a person: full name + DOB + geographic anchor minimum

3. MISATTRIBUTION RED FLAGS
   - What data items are most at risk of being misattributed?
   - Which sources are most likely to mix entities?

4. VERIFICATION METHODOLOGY
   - How should the investigator confirm each data item is correctly attributed?
   - What is the minimum corroboration needed before attributing sensitive findings?

Be specific about the risks for this particular entity name and type."""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1000,
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content[0].text

def ai_output_verification_checklist(ai_generated_claims: List[str]) -> List[Dict]:
    """
    Generate verification requirements for AI-generated claims
    """
    verification_requirements = []

    for claim in ai_generated_claims:
        requirement = {
            'claim': claim,
            'verification_needed': True,
            'verification_approach': '',
            'risk_level': 'MEDIUM'
        }

        # Categorize by claim type
        import re

        # Dates
        if re.search(r'\b(19|20)\d{2}\b', claim) or any(word in claim.lower() for word in ['year', 'date', 'founded', 'established']):
            requirement['verification_approach'] = "Verify against primary source (registry, filing, news)"
            requirement['risk_level'] = 'HIGH'

        # URLs or email addresses
        elif re.search(r'https?://|@', claim):
            requirement['verification_approach'] = "Verify URL exists and content matches claim; verify email via MX record"
            requirement['risk_level'] = 'HIGH'

        # Names
        elif re.search(r'\b[A-Z][a-z]+ [A-Z][a-z]+\b', claim):
            requirement['verification_approach'] = "Verify named individual in primary sources (LinkedIn, company website, filings)"
            requirement['risk_level'] = 'HIGH'

        # Financial amounts
        elif re.search(r'\$[\d,]+|\b\d+\s*(million|billion)\b', claim):
            requirement['verification_approach'] = "Verify financial figures in SEC filings, news, or official statements"
            requirement['risk_level'] = 'HIGH'

        # General factual claims
        else:
            requirement['verification_approach'] = "Find primary source citation; do not rely on AI assertion"
            requirement['risk_level'] = 'MEDIUM'

        verification_requirements.append(requirement)

    return verification_requirements

def investigation_premortem(hypothesis: str, key_findings: List[str],
                            planned_action: str) -> str:
    """
    Pre-mortem analysis: assume the conclusion is wrong and work backward
    """
    client = anthropic.Anthropic()

    findings_text = "\n".join([f"- {f}" for f in key_findings])

    prompt = f"""Conduct a pre-mortem analysis on this OSINT investigation.

CONCLUSION ABOUT TO BE ACTED UPON:
{hypothesis}

KEY FINDINGS SUPPORTING THIS CONCLUSION:
{findings_text}

PLANNED ACTION:
{planned_action}

ASSUME THE CONCLUSION IS WRONG. Work backward:

1. HOW DID WE GET IT WRONG?
   - What are the 3 most plausible ways this conclusion could be incorrect?
   - For each: what evidence would look identical whether the hypothesis is true or false?

2. WHAT DID WE MISS?
   - What evidence should we have collected that might have changed the conclusion?
   - What alternative explanations were dismissed too quickly?

3. WHAT HAPPENS IF WE'RE WRONG?
   - What are the consequences of acting on a false conclusion?
   - Who is harmed if the named subject is innocent?
   - What is the legal exposure of a false finding?

4. WHAT ADDITIONAL VERIFICATION IS NEEDED?
   - Given the consequences of being wrong, what minimum additional verification is required?
   - Is there an irreversible action being taken? If so, has verification been sufficient?

5. GO/NO-GO RECOMMENDATION
   - Should the planned action proceed, or should additional verification be completed first?
   - What specific conditions must be met before proceeding?

Be thorough. The purpose is to catch errors before they cause harm."""

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content[0].text

class ConfidenceCalibrationTracker:
    """
    Track investigator confidence vs. accuracy for calibration
    """

    def __init__(self, investigator_id: str):
        self.investigator_id = investigator_id
        self.predictions = []

    def record_prediction(self, claim: str, confidence: float,
                         investigation_id: str) -> str:
        """
        Record a prediction with confidence level for later verification
        confidence: 0.0 to 1.0
        """
        prediction_id = f"{investigation_id}_{len(self.predictions)}"

        self.predictions.append({
            'prediction_id': prediction_id,
            'claim': claim,
            'confidence': confidence,
            'investigation_id': investigation_id,
            'outcome': None,
            'verified': False,
            'recorded_at': __import__('datetime').datetime.now().isoformat()
        })

        return prediction_id

    def record_outcome(self, prediction_id: str, was_correct: bool) -> None:
        """Record whether a prediction was correct"""
        for pred in self.predictions:
            if pred['prediction_id'] == prediction_id:
                pred['outcome'] = was_correct
                pred['verified'] = True
                break

    def calculate_calibration(self) -> Dict:
        """
        Calculate calibration score across verified predictions
        """
        verified = [p for p in self.predictions if p['verified']]

        if len(verified) < 10:
            return {
                'message': f"Need at least 10 verified predictions; have {len(verified)}",
                'predictions_needed': 10 - len(verified)
            }

        # Group by confidence bucket
        buckets = {
            '50-60%': [], '60-70%': [], '70-80%': [],
            '80-90%': [], '90-95%': [], '95-100%': []
        }

        for pred in verified:
            conf = pred['confidence']
            if 0.5 <= conf < 0.6:
                buckets['50-60%'].append(pred['outcome'])
            elif 0.6 <= conf < 0.7:
                buckets['60-70%'].append(pred['outcome'])
            elif 0.7 <= conf < 0.8:
                buckets['70-80%'].append(pred['outcome'])
            elif 0.8 <= conf < 0.9:
                buckets['80-90%'].append(pred['outcome'])
            elif 0.9 <= conf < 0.95:
                buckets['90-95%'].append(pred['outcome'])
            elif conf >= 0.95:
                buckets['95-100%'].append(pred['outcome'])

        calibration = {}
        for bucket, outcomes in buckets.items():
            if outcomes:
                actual_rate = sum(outcomes) / len(outcomes)
                calibration[bucket] = {
                    'count': len(outcomes),
                    'actual_accuracy': actual_rate,
                    'claimed_confidence': float(bucket.split('-')[0].rstrip('%')) / 100
                }

        overall_accuracy = sum(p['outcome'] for p in verified) / len(verified)

        return {
            'total_verified': len(verified),
            'overall_accuracy': overall_accuracy,
            'calibration_by_confidence': calibration,
            'assessment': 'Well calibrated' if abs(overall_accuracy - 0.75) < 0.1 else 'Recalibration needed'
        }