ファクトチェックとハルシネーション防止

def check_grounding(claim: str, retrieved_docs: list[dict]) -> dict:
    doc_texts = '\n\n'.join(
        f'[DOC {i+1}] ({d["url"]})\n{d["text"][:800]}'
        for i, d in enumerate(retrieved_docs[:5])
    )
    return {
        'claim': claim,
        'docs':  doc_texts,
        'grounded': None   # to be filled by LLM verifier
    }

if __name__ == '__main__':
    docs = [{'url': 'https://example.com/geo', 'text': 'Paris is the capital of France.'}]
    result = check_grounding('Paris is the capital of France.', docs)
    print('Claim:', result['claim'])
    print('Supporting docs used:')
    print(result['docs'])

import openai, json

client = openai.OpenAI(api_key='YOUR_OPENAI_KEY')

def verify_claim(claim: str, source_texts: list[str]) -> dict:
    sources_block = '\n---\n'.join(source_texts[:3])
    prompt = (
        f'Is the following claim directly supported by the provided sources?\n'
        f'Claim: "{claim}"\n\n'
        f'Sources:\n{sources_block}\n\n'
        f'Return JSON: {{\n'
        f'  "supported": true/false,\n'
        f'  "confidence": 0.0-1.0,\n'
        f'  "reason": "one sentence explanation"\n'
        f'}}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(resp.choices[0].message.content)

INCLUDE_THRESHOLD  = 0.85
DISCLAIMER_THRESHOLD = 0.50

def gate_claim(claim: str, source_texts: list[str]) -> dict:
    result = verify_claim(claim, source_texts)
    conf = result.get('confidence', 0.0)
    supported = result.get('supported', False)

    if not supported or conf < DISCLAIMER_THRESHOLD:
        return {'action': 'exclude', 'claim': claim, 'reason': result.get('reason')}
    elif conf < INCLUDE_THRESHOLD:
        return {
            'action': 'include_with_disclaimer',
            'claim': f'[LOW CONFIDENCE] {claim}',
            'reason': result.get('reason')
        }
    else:
        return {'action': 'include', 'claim': claim}

import re

def is_high_risk_claim(claim: str) -> bool:
    patterns = [
        r'\d+\.?\d*\s*%',            # percentages: 9.1%
        r'\$\s*\d+',                  # dollar amounts
        r'\b(January|February|March|April|May|June|July|August|September|October|November|December)\s+\d{4}',  # dates
        r'\b[A-Z][a-z]+\s+[A-Z][a-z]+\s+(said|stated|argued|claimed)',  # named quotes
        r'(caused?|led to|resulted in)',                                  # causal claims
    ]
    return any(re.search(p, claim) for p in patterns)

print(is_high_risk_claim('Inflation hit 9.1% in June 2022'))  # True
print(is_high_risk_claim('Inflation was elevated'))            # False

def decompose_into_claims(section_text: str) -> list[str]:
    prompt = (
        f'Break this text into individual verifiable factual claims.\n'
        f'Each claim should be a single sentence containing exactly one fact.\n'
        f'Return JSON: {{"claims": ["..."]}}\n\n'
        f'TEXT:\n{section_text}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        response_format={'type': 'json_object'}
    )
    return json.loads(resp.choices[0].message.content).get('claims', [])

def reconstruct_section(verified_claims: list[str],
                        section_name: str) -> str:
    claims_text = '\n'.join(f'- {c}' for c in verified_claims)
    prompt = (
        f'Rewrite these verified facts as a coherent {section_name} section.\n'
        f'Do NOT add any new information not present in the claims.\n'
        f'Only use the facts provided.\n\n'
        f'VERIFIED CLAIMS:\n{claims_text}'
    )
    resp = client.chat.completions.create(
        model='gpt-4o',
        messages=[{'role': 'user', 'content': prompt}]
    )
    return resp.choices[0].message.content

import re

def extract_numbers(text: str) -> list[float]:
    matches = re.findall(r'[-+]?\d*\.?\d+', text)
    return [float(m) for m in matches]

def check_numeric_drift(claim: str, source_text: str,
                        tolerance: float = 0.01) -> bool:
    claim_nums  = extract_numbers(claim)
    source_nums = extract_numbers(source_text)

    for cn in claim_nums:
        found_match = any(abs(cn - sn) <= tolerance for sn in source_nums)
        if not found_match:
            return True  # numeric drift detected
    return False

print(check_numeric_drift(
    'Inflation reached 9.2% in June 2022',
    'The CPI rose 9.1 percent in June 2022'
))  # True — 9.2 vs 9.1

def attribution_audit(report_text: str, sources: list[dict]) -> list[str]:
    sentences = [s.strip() for s in report_text.split('.') if len(s.strip()) > 20]
    unattributed = []

    for sentence in sentences:
        found = False
        for src in sources:
            if any(word in src.get('text', '') for word in sentence.split()[:5]):
                found = True
                break
        if not found:
            unattributed.append(sentence)

    return unattributed

# Flag unattributed sentences for manual review or re-verification

if __name__ == '__main__':
    report = ("Water boils at 100 degrees Celsius at sea level. "
              "The moon is made primarily of green cheese according to this document.")
    sources = [{'text': 'Water boils at 100C (212F) at standard atmospheric pressure.'}]
    unattributed = attribution_audit(report, sources)
    print('Unattributed sentences (need manual review):')
    for s in unattributed:
        print(' -', s)

def hedge_claim(claim: str, confidence: float) -> str:
    if confidence >= 0.85:
        return claim  # state as fact
    elif confidence >= 0.65:
        hedges = ['Some sources suggest', 'According to available evidence',
                  'Analysts have noted']
        return f'{hedges[hash(claim) % len(hedges)]}, {claim.lower()}'
    else:
        return f'It has been reported (with low confidence) that {claim.lower()}'

print(hedge_claim('Inflation peaked at 9.1%', 0.90))
print(hedge_claim('Supply chain disruptions contributed to inflation', 0.72))
print(hedge_claim('Specific policy caused inflation', 0.45))

verification_log = []  # In production: a database

def log_verification(claim: str, supported: bool, confidence: float):
    verification_log.append({
        'claim':      claim[:100],
        'supported':  supported,
        'confidence': confidence
    })

def hallucination_rate() -> float:
    if not verification_log:
        return 0.0
    failed = sum(1 for v in verification_log if not v['supported'])
    return failed / len(verification_log)

def check_hallucination_alert(threshold: float = 0.15):
    rate = hallucination_rate()
    if rate > threshold:
        print(f'ALERT: Hallucination rate {rate:.1%} exceeds threshold {threshold:.1%}')
    return rate

if __name__ == '__main__':
    log_verification('The sky is blue', True, 0.95)
    log_verification('The moon is made of cheese', False, 0.2)
    log_verification('Water boils at 100C at sea level', True, 0.99)
    rate = check_hallucination_alert(threshold=0.15)
    print(f'Hallucination rate so far: {rate:.1%}')