5.6 Information Provenance and Multi-Source Synthesis

from dataclasses import dataclass
from datetime import datetime
from enum import Enum

class SourceType(Enum):
    DOCUMENT = "document"
    API = "api"
    DATABASE = "database"
    WEB = "web"
    HUMAN_INPUT = "human_input"
    MODEL_GENERATION = "model_generation"  # Claude generated — no external source

@dataclass
class SourceCitation:
    source_type: SourceType
    location: str          # URL, file path, or API endpoint
    retrieved_at: datetime
    reliability: float     # 0-1 confidence in source
    excerpt: str           # The actual text cited

@dataclass
class Claim:
    statement: str
    sources: list[SourceCitation]
    confidence: float
    
    @property
    def is_grounded(self):
        """A claim is grounded if it has at least one non-model source"""
        return any(s.source_type != SourceType.MODEL_GENERATION for s in self.sources)

def synthesize_agreement(claims_from_sources):
    """Multiple sources agree — high confidence"""
    common_claims = find_common(claims_from_sources)
    
    for claim in common_claims:
        claim.confidence = min(1.0, 0.5 + 0.2 * len(claim.sources))
        # More sources agree = higher confidence
    
    return common_claims

def resolve_conflict(claim_a, claim_b):
    """Two sources disagree — need resolution strategy"""
    
    strategies = {
        "recency": lambda a, b: a if a.retrieved_at > b.retrieved_at else b,
        "authority": lambda a, b: a if a.source_reliability > b.source_reliability else b,
        "specificity": lambda a, b: a if a.is_more_specific else b,
    }
    
    # Apply hierarchy: official docs > API response > web > model generation
    authority_rank = {
        SourceType.DOCUMENT: 4,    # Official docs
        SourceType.API: 3,         # Live API data
        SourceType.DATABASE: 3,    # Internal data
        SourceType.WEB: 2,         # Web search
        SourceType.HUMAN_INPUT: 2, # User said
        SourceType.MODEL_GENERATION: 1,  # Claude generated
    }
    
    a_rank = max(authority_rank[s.source_type] for s in claim_a.sources)
    b_rank = max(authority_rank[s.source_type] for s in claim_b.sources)
    
    if a_rank != b_rank:
        winner = claim_a if a_rank > b_rank else claim_b
        return {
            "resolved": winner,
            "conflict_note": f"Conflict between sources. Chose {winner.sources[0].location} (higher authority)",
            "discarded": claim_b if a_rank > b_rank else claim_a
        }
    
    # Same authority — flag for human review
    return {
        "resolved": None,
        "conflict_note": "Equal authority sources disagree — human review needed",
        "options": [claim_a, claim_b]
    }

def generate_report_with_provenance(topic, sources):
    """Generate a report where every claim has a citation"""
    
    report = client.messages.create(
        system="""Generate a report. For EVERY factual claim, add a citation 
        in [Source N] format. At the end, list all sources with their details.
        If you are not sure about a claim, mark it as [UNVERIFIED].""",
        messages=[{"role": "user", "content": f"""
        Topic: {topic}
        
        Available sources:
        {format_sources(sources)}
        
        Write a report with citations for every claim.
        """}]
    )
    
    # Verify citations actually exist in sources
    claims = extract_claims_with_citations(report)
    for claim in claims:
        if not verify_citation_exists(claim.citation, sources):
            claim.flag = "CITATION_NOT_FOUND"  # Possible hallucination
    
    return report, claims

def check_freshness(source: SourceCitation, max_age_days=30):
    age = (datetime.now() - source.retrieved_at).days
    
    if age > max_age_days:
        return {
            "fresh": False,
            "age_days": age,
            "recommendation": "Re-fetch this source — data may be outdated"
        }
    return {"fresh": True, "age_days": age}