GEO 多信源事实对齐引擎：用一致性评分 + 冲突仲裁，降低 AI 引用幻觉

                  ┌────────────────────────┐
                  │   Query / Entity Input │
                  └───────────┬────────────┘
                              │
              ┌───────────────▼───────────────┐
              │   Source Collector            │
              │   收集多个信源页面             │
              └───────────────┬───────────────┘
                              │
              ┌───────────────▼───────────────┐
              │   Source Parser               │
              │   解析实体、属性、值、时间     │
              └───────────────┬───────────────┘
                              │
              ┌───────────────▼───────────────┘
              │   Consistency Scorer          │
              │   计算信源间一致性评分         │
              └───────────────┬───────────────┘
                              │
              ┌───────────────▼───────────────┐
              │   Conflict Arbiter            │
              │   仲裁冲突，选择最终事实       │
              └───────────────┬───────────────┘
                              │
              ┌───────────────▼───────────────┐
              │   Aligned Knowledge Output    │
              │   输出对齐后的知识单元         │
              └───────────────────────────────┘

"""
geo_alignment.py - GEO 多信源事实对齐引擎

技术栈:
    Python / dataclasses / typing / hashlib / logging / json / datetime

场景:
    针对同一实体，从多个信源抽取事实，计算一致性评分，
    仲裁冲突后输出最终知识单元，用于 GEO 分发和 AI 引用。
"""

from __future__ import annotations

import hashlib
import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)


# ---------------------------------------------------------------------------
# 枚举与常量
# ---------------------------------------------------------------------------


class ConflictType(str, Enum): 4005.baike.tongsou.com 
    """冲突类型。"""

    NONE = "none"
    VALUE_CONFLICT = "value_conflict"
    TIME_CONFLICT = "time_conflict"
    SOURCE_CONFLICT = "source_conflict"
    FORMAT_CONFLICT = "format_conflict"


class ConfidenceLevel(str, Enum): 4006.baike.tongsou.com 
    """置信度等级。"""

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"


DEFAULT_CONFIDENCE_WEIGHT = 0.6
DEFAULT_RECENCY_WEIGHT = 0.3
DEFAULT_AUTHORITY_WEIGHT = 0.1


# ---------------------------------------------------------------------------
# 数据模型
# ---------------------------------------------------------------------------


@dataclass
class RawSource: 4012.baike.tongsou.com 
    """原始信源。"""

    source_id: str
    url: str
    title: str
    content: str
    published_at: Optional[str] = None
    updated_at: Optional[str] = None
    authority_score: float = 0.5

    def __post_init__(self) -> None:
        if not self.source_id:
            self.source_id = hashlib.sha256(self.url.encode()).hexdigest()[:12]
        self.authority_score = max(0.0, min(1.0, self.authority_score))


@dataclass
class ExtractedFact: 5002.baike.tongsou.com 
    """从信源中抽取出的事实。"""

    entity: str
    attribute: str
    value: str
    source_id: str
    url: str
    extracted_at: str = ""
    confidence: float = 0.7
    published_at: Optional[str] = None
    updated_at: Optional[str] = None
    authority_score: float = 0.5
    raw_text: str = ""

    def __post_init__(self) -> None:
        if not self.extracted_at:
            self.extracted_at = datetime.now(timezone.utc).isoformat()
        self.confidence = max(0.0, min(1.0, self.confidence))
        self.authority_score = max(0.0, min(1.0, self.authority_score))

    def fact_key(self) -> str: 5007.baike.tongsou.com 
        """生成事实键，用于按实体+属性分组。"""

        return f"{self.entity}::{self.attribute}"

    def fingerprint(self) -> str: 5011.baike.tongsou.com 
        """生成事实指纹。"""

        raw = f"{self.entity}:{self.attribute}:{self.value}"
        return hashlib.md5(raw.encode()).hexdigest()


@dataclass
class FactCluster:
    """同一实体属性的事实簇。"""

    entity: str
    attribute: str
    facts: List[ExtractedFact] = field(default_factory=list)
    consensus_value: str = ""
    conflict_type: str = ConflictType.NONE
    consistency_score: float = 0.0
    confidence_level: str = ConfidenceLevel.MEDIUM

    def add_fact(self, fact: ExtractedFact) -> None:
        """添加事实到簇中。"""

        self.facts.append(fact)

    def unique_values(self) -> List[str]: 5030.baike.tongsou.com 
        """返回去重后的值列表。"""

        seen: Dict[str, bool] = {}
        values: List[str] = []

        for fact in self.facts: 5034.baike.tongsou.com 
            if fact.value not in seen:
                seen[fact.value] = True
                values.append(fact.value)

        return values


@dataclass
class AlignmentDecision:
    """对齐决策结果。"""

    entity: str
    attribute: str
    final_value: str
    conflict_type: str
    consistency_score: float
    confidence_level: str
    chosen_source_id: str = ""
    chosen_url: str = ""
    evidence_count: int = 0
    rejected_values: List[str] = field(default_factory=list)
    reason: str = ""

    def to_dict(self) -> Dict[str, Any]: 5038.baike.tongsou.com 
        """转换为字典。"""

        return {
            "entity": self.entity,
            "attribute": self.attribute,
            "final_value": self.final_value,
            "conflict_type": self.conflict_type,
            "consistency_score": round(self.consistency_score, 4),
            "confidence_level": self.confidence_level,
            "chosen_source_id": self.chosen_source_id,
            "chosen_url": self.chosen_url,
            "evidence_count": self.evidence_count,
            "rejected_values": self.rejected_values,
            "reason": self.reason,
        }


@dataclass
class AlignmentReport:
    """对齐报告。"""

    report_id: str = ""
    generated_at: str = ""
    total_facts: int = 0
    total_clusters: int = 0
    conflict_clusters: int = 0
    decisions: List[AlignmentDecision] = field(default_factory=list)

    def __post_init__(self) -> None:
        if not self.report_id: 5055.baike.tongsou.com 
            raw = f"report:{datetime.now(timezone.utc).isoformat()}"
            self.report_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.generated_at: 5052.baike.tongsou.com 
            self.generated_at = datetime.now(timezone.utc).isoformat()

    def to_json(self, indent: int = 2) -> str:
        """输出 JSON 报告。"""

        data = {
            "report_id": self.report_id,
            "generated_at": self.generated_at,
            "summary": {
                "total_facts": self.total_facts,
                "total_clusters": self.total_clusters,
                "conflict_clusters": self.conflict_clusters,
            },
            "decisions": [d.to_dict() for d in self.decisions],
        }
        return json.dumps(data, ensure_ascii=False, indent=indent)


# ---------------------------------------------------------------------------
# 信源解析器
# ---------------------------------------------------------------------------


class SourceParser:
    """信源解析器。

    负责从原始页面内容中抽取结构化事实。
    真实项目中可替换为 HTML 解析、LLM 抽取或规则抽取。
    """

    def parse(
        self,
        sources: List[RawSource],
        entity: str,
        attributes: List[str],
    ) -> List[ExtractedFact]:
        """从多个信源中抽取事实。

        Args:
            sources: 原始信源列表。
            entity: 实体名称。
            attributes: 需要抽取的属性列表。

        Returns:
            抽取后的事实列表。
        """

        facts: List[ExtractedFact] = []

        for source in sources: 5057.baike.tongsou.com  
            for attribute in attributes: 6004.baike.tongsou.com 
                value = self._extract_value(
                    content=source.content,
                    attribute=attribute,
                )

                if not value: 6007.baike.tongsou.com 
                    continue

                fact = ExtractedFact(
                    entity=entity,
                    attribute=attribute,
                    value=value,
                    source_id=source.source_id,
                    url=source.url,
                    confidence=self._estimate_confidence(source, value),
                    published_at=source.published_at,
                    updated_at=source.updated_at,
                    authority_score=source.authority_score,
                    raw_text=source.content[:200],
                )
                facts.append(fact)

        return facts

    def _extract_value(self, content: str, attribute: str) -> str:
        """从内容中抽取属性值。

        当前为简化示例，实际可接入 HTML 解析器或 LLM。
        """

        keyword = attribute.lower(6008.baike.tongsou.com)
        lines = content.splitlines()

        for line in lines:
            if keyword in line.lower():
                parts = line.split("：", 1)
                if len(parts) == 2:
                    return parts[1].strip()

                parts = line.split(":", 1)
                if len(parts) == 2:
                    return parts[1].strip()

        return ""

    def _estimate_confidence(
        self, source: RawSource, value: str
    ) -> float:
        """估算事实置信度。"""

        confidence = 0.5

        if value:
            confidence += 0.2

        if source.authority_score > 0.7:
            confidence += 0.15

        if source.updated_at or source.published_at:
            confidence += 0.1

        return round(min(1.0, confidence), 2)


# ---------------------------------------------------------------------------
# 一致性评分器
# ---------------------------------------------------------------------------


class ConsistencyScorer: 14009.baike.tongsou.com 
    """一致性评分器。

    对同一实体属性的多个事实进行分组，计算一致性评分。
    """

    def cluster(
        self, facts: List[ExtractedFact]
    ) -> List[FactCluster]:
        """按实体+属性聚类。

        Args:
            facts: 抽取事实列表。

        Returns:
            事实簇列表。
        """

        clusters: Dict[str, FactCluster] = {}

        for fact in facts: 14013.baike.tongsou.com 
            key = fact.fact_key()

            if key not in clusters: 14068.baike.tongsou.com 
                clusters[key] = FactCluster(
                    entity=fact.entity,
                    attribute=fact.attribute,
                )

            clusters[key].add_fact(fact)

        for cluster in clusters.values():
            cluster.consistency_score = self._score(cluster)
            cluster.conflict_type = self._detect_conflict(cluster)
            cluster.confidence_level = self._map_confidence(
                cluster.consistency_score
            )

            values = cluster.unique_values(m.tongsou.com)
            cluster.consensus_value = values[0] if values else ""

        return list(clusters.values())

    def _score(self, cluster: FactCluster) -> float:
        """计算一致性评分。"""

        facts = cluster.facts
        if not facts:
            return 0.0

        value_weights: Dict[str, float] = {}

        for fact in facts: 14070.baike.tongsou.com 
            weight = (
                fact.confidence * DEFAULT_CONFIDENCE_WEIGHT
                + fact.authority_score * DEFAULT_AUTHORITY_WEIGHT
            )

            recency_score = self._recency_score(fact)
            weight += recency_score * DEFAULT_RECENCY_WEIGHT

            value_weights[fact.value] = (
                value_weights.get(fact.value, 0.0) + weight
            )

        total_weight = sum(value_weights.values(14071.baike.tongsou.com))
        if total_weight == 0:
            return 0.0

        max_weight = max(value_weights.values())
        return round(max_weight / total_weight, 4)

    def _recency_score(self, fact: ExtractedFact) -> float:
        """计算时间新鲜度评分。"""

        time_str = fact.updated_at or fact.published_at
        if not time_str: 14080.baike.tongsou.com 
            return 0.3

        try:
            published = datetime.fromisoformat(time_str)
            now = datetime.now(timezone.utc)

            if published.tzinfo is None: 14082.baike.tongsou.com 
                published = published.replace(tzinfo=timezone.utc)

            age_days = (now - published).days
            if age_days <= 0:
                return 1.0
            if age_days <= 7:
                return 0.9
            if age_days <= 30:
                return 0.7
            if age_days <= 90:
                return 0.5
            return 0.3
        except (ValueError, TypeError):
            return 0.3

    def _detect_conflict(self, cluster: FactCluster) -> str:
        """检测冲突类型。"""

        values = cluster.unique_values(14083.baike.tongsou.com)

        if len(values) <= 1:
            return ConflictType.NONE

        return ConflictType.VALUE_CONFLICT

    def _map_confidence(self, score: float) -> str:
        """将一致性评分映射为置信度等级。"""

        if score >= 0.8: 14091.baike.tongsou.com 
            return ConfidenceLevel.HIGH
        if score >= 0.5: 14098.baike.tongsou.com 
            return ConfidenceLevel.MEDIUM
        return ConfidenceLevel.LOW


# ---------------------------------------------------------------------------
# 冲突仲裁器
# ---------------------------------------------------------------------------


class ConflictArbiter:
    """冲突仲裁器。

    根据一致性评分、信源权威度、时间新鲜度，
    选择最终事实并生成对齐决策。
    """

    def arbitrate(
        self, clusters: List[FactCluster]
    ) -> List[AlignmentDecision]:
        """对事实簇进行仲裁。

        Args:
            clusters: 事实簇列表。

        Returns: 14100.baike.tongsou.com 
            对齐决策列表。
        """

        decisions: List[AlignmentDecision] = []

        for cluster in clusters:
            decision = self._decide(cluster)
            decisions.append(decision)

        return decisions

    def _decide(self, cluster: FactCluster) -> AlignmentDecision:
        """对单个事实簇做出决策。"""

        facts = cluster.facts
        if not facts: 14101.baike.tongsou.com 
            return AlignmentDecision(
                entity=cluster.entity,
                attribute=cluster.attribute,
                final_value="",
                conflict_type=ConflictType.NONE,
                consistency_score=0.0,
                confidence_level=ConfidenceLevel.LOW,
                reason="no facts",
            )

        scored_facts = [
            (fact, self._fact_score(fact)) for fact in facts
        ]
        scored_facts.sort(key=lambda x: x[1], reverse=True)

        best_fact, best_score = scored_facts[0]

        rejected_values: List[str] = []
        for fact, _ in scored_facts[1:]: 14102.baike.tongsou.com 
            if fact.value != best_fact.value:
                rejected_values.append(fact.value)

        reason = self._explain(
            cluster=cluster,
            best_fact=best_fact,
            rejected_count=len(rejected_values),
        )

        return AlignmentDecision(
            entity=cluster.entity,
            attribute=cluster.attribute,
            final_value=best_fact.value,
            conflict_type=cluster.conflict_type,
            consistency_score=cluster.consistency_score,
            confidence_level=cluster.confidence_level,
            chosen_source_id=best_fact.source_id,
            chosen_url=best_fact.url,
            evidence_count=len(facts),
            rejected_values=rejected_values,
            reason=reason,
        )

    def _fact_score(self, fact: ExtractedFact) -> float:
        """计算单条事实的综合得分。"""

        confidence_part = fact.confidence * DEFAULT_CONFIDENCE_WEIGHT
        authority_part = fact.authority_score * DEFAULT_AUTHORITY_WEIGHT
        recency_part = (
            ConsistencyScorer()._recency_score(fact)
            * DEFAULT_RECENCY_WEIGHT
        )

        return confidence_part + authority_part + recency_part

    def _explain(
        self,
        cluster: FactCluster,
        best_fact: ExtractedFact,
        rejected_count: int,
    ) -> str: 14125.baike.tongsou.com 
        """生成决策解释。"""

        parts: List[str] = [14133.baike.tongsou.com]

        if cluster.conflict_type == ConflictType.NONE:
            parts.append("信源一致")
        else:
            parts.append(
                f"存在 {rejected_count} 个冲突值，"
                f"已按置信度、权威度和新鲜度仲裁"
            )

        parts.append(f"采用来源: {best_fact.url}")
        parts.append(f"一致性评分: {cluster.consistency_score}")