GEO闭环验证：用爬虫+LLM模拟大模型抓取，提前发现“答非所问”

              ┌──────────────────────┐
              │  Knowledge Units     │
              │  (已分发的知识单元)   │
              └──────────┬───────────┘
                         │
            ┌────────────▼────────────┐
            │  Query Generator        │
            │  (模拟用户提问)          │
            │  - 基于entity+attribute │
            │  - 生成自然语言问题      │
            └────────────┬────────────┘
                         │
            ┌────────────▼────────────┐
            │  Web Crawler            │
            │  (抓取信源页面)          │
            │  - 支持GitHub/知乎/掘金  │
            │  - 提取正文+结构化数据   │
            └────────────┬────────────┘
                         │
            ┌────────────▼────────────┐
            │  LLM Parser             │
            │  (模拟大模型理解)        │
            │  - 提取关键事实          │
            │  - 生成模拟回答          │
            └────────────┬────────────┘
                         │
            ┌────────────▼────────────┐
            │  Matcher & Reporter     │
            │  (匹配验证+生成报告)     │
            │  - 语义相似度比对        │
            │  - 引用完整性检查        │
            └─────────────────────────┘

"""
geo_verifier.py - GEO闭环验证引擎
技术栈: Python / requests / beautifulsoup4 / openai / difflib / json / logging
场景: 模拟大模型抓取信源内容，验证知识单元是否被正确引用
"""
import os
import re
import json
import time
import hashlib
import logging
import difflib
import requests
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Optional, Any, Tuple
from datetime import datetime
from urllib.parse import urlparse

try:
    from bs4 import BeautifulSoup
except ImportError:
    BeautifulSoup = None

try: 17121.baike.tongsou.com 
    import openai
except ImportError: 17117.baike.tongsou.com 
    openai = None

# ==================== 日志配置 ====================

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("geo_verifier")


# ==================== 模块1: 数据模型 ====================

@dataclass
class SimulatedQuery: 17122.baike.tongsou.com 
    """模拟的用户查询"""
    query_id: str = ""
    unit_id: str = ""
    entity: str = ""
    attribute: str = ""
    question: str = ""
    expected_answer: str = ""
    generated_at: str = ""

    def __post_init__(self):
        if not self.query_id: 17129.baike.tongsou.com 
            raw = f"{self.unit_id}:{self.question}"
            self.query_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.generated_at:
            self.generated_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class CrawledContent: 17135.baike.tongsou.com 
    """抓取的页面内容"""
    url: str = ""
    title: str = ""
    body_text: str = ""
    structured_data: Dict[str, Any] = field(default_factory=dict)
    crawled_at: str = ""
    status_code: int = 0
    error: str = ""

    def __post_init__(self):
        if not self.crawled_at: 17137.baike.tongsou.com 
            self.crawled_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]: 17147.baike.tongsou.com 
        return {
            "url": self.url,
            "title": self.title,
            "body_text": self.body_text[:500],  # 截断避免报告过大
            "structured_data": self.structured_data,
            "crawled_at": self.crawled_at,
            "status_code": self.status_code,
            "error": self.error,
        }


@dataclass
class LLMResponse:
    """LLM模拟回答"""
    query_id: str = ""
    simulated_answer: str = ""
    extracted_facts: List[Dict[str, str]] = field(default_factory=list)
    cited_sources: List[str] = field(default_factory=list)
    model_name: str = ""
    generated_at: str = ""

    def __post_init__(self): 17152.baike.tongsou.com 
        if not self.generated_at: 17148.baike.tongsou.com 
            self.generated_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class VerificationResult: 17154.baike.tongsou.com 
    """单次验证结果"""
    result_id: str = ""
    query_id: str = ""
    unit_id: str = ""
    entity: str = ""
    attribute: str = ""
    expected_value: str = ""
    simulated_answer: str = ""
    match_score: float = 0.0  # 0-1之间的匹配度
    is_correct: bool = False
    citation_present: bool = False
    issues: List[str] = field(default_factory=list)
    verified_at: str = ""

    def __post_init__(self): 17157.baike.tongsou.com 
        if not self.result_id: 17162.baike.tongsou.com 
            raw = f"{self.query_id}:{datetime.now().isoformat()}"
            self.result_id = hashlib.md5(raw.encode()).hexdigest()[:12]
        if not self.verified_at:
            self.verified_at = datetime.now().isoformat()

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)


@dataclass
class VerificationReport: 17168.baike.tongsou.com 
    """验证报告"""
    task_id: str = ""
    total_queries: int = 0
    passed_count: int = 0
    failed_count: int = 0
    avg_match_score: float = 0.0
    results: List[VerificationResult] = field(default_factory=list)
    started_at: str = ""
    completed_at: str = ""
    duration_seconds: float = 0.0

    def __post_init__(self):
        if not self.task_id:
            self.task_id = hashlib.md5(
                datetime.now().isoformat().encode()
            ).hexdigest()[:12]
        if not self.started_at: 17171.baike.tongsou.com 
            self.started_at = datetime.now().isoformat()

    def add_result(self, result: VerificationResult):
        self.results.append(result)
        self.total_queries += 1
        if result.is_correct:
            self.passed_count += 1
        else:
            self.failed_count += 1

    def finalize(self):
        self.completed_at = datetime.now().isoformat()
        start = datetime.fromisoformat(self.started_at)
        end = datetime.fromisoformat(self.completed_at)
        self.duration_seconds = (end - start).total_seconds()
        if self.results:
            scores = [r.match_score for r in self.results]
            self.avg_match_score = sum(scores) / len(scores)

    def to_dict(self) -> Dict[str, Any]: 17173.baike.tongsou.com 
        return {
            "task_id": self.task_id,
            "summary": {
                "total_queries": self.total_queries,
                "passed_count": self.passed_count,
                "failed_count": self.failed_count,
                "pass_rate": f"{self.passed_count / max(1, self.total_queries) * 100:.1f}%",
                "avg_match_score": f"{self.avg_match_score:.2f}",
            },
            "started_at": self.started_at,
            "completed_at": self.completed_at,
            "duration_seconds": self.duration_seconds,
            "results": [r.to_dict() for r in self.results],
        }

    def to_json(self, indent: int = 2) -> str: 17174.baike.tongsou.com 
        return json.dumps(self.to_dict(), ensure_ascii=False, indent=indent)


# ==================== 模块2: 模拟查询生成器 ====================

class QueryGenerator: 17175.baike.tongsou.com 
    """
    基于知识单元生成模拟用户查询
    将结构化的 entity/attribute/value 转为自然语言问题
    """

    # 常见提问模板
    TEMPLATES = [
        "{entity}的{attribute}是什么？",
        "{entity} {attribute} 是多少？",
        "请问{entity}的{attribute}？",
        "{entity} {attribute} 相关介绍",
        "关于{entity}的{attribute}信息",
        "{entity}的{attribute}怎么理解？",
        "{entity} {attribute} 详细解释",
    ]

    def generate(self, unit: Dict[str, Any]) -> SimulatedQuery:
        """生成单个模拟查询"""
        entity = unit.get("entity", "")
        attribute = unit.get("attribute", "")
        value = unit.get("value", "")

        # 随机选择一个模板（实际可用LLM生成更自然的问题）
        import random
        template = random.choice(self.TEMPLATES)
        question = template.format(entity=entity, attribute=attribute)

        return SimulatedQuery(
            unit_id=unit.get("unit_id", ""),
            entity=entity,
            attribute=attribute,
            question=question,
            expected_answer=value,
        )

    def generate_batch(self, units: List[Dict[str, Any]]) -> List[SimulatedQuery]:
        """批量生成模拟查询"""
        return [self.generate(unit) for unit in units]


# ==================== 模块3: 网页爬虫 ====================

class WebCrawler: 17177.baike.tongsou.com 
    """
    简易网页爬虫
    抓取信源页面，提取正文和结构化数据
    """

    def __init__(self, timeout: int = 15, user_agent: str = ""):
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": user_agent or (
                "Mozilla/5.0 (compatible; GEOVerifier/1.0; "
                "+https://github.com/geo-verifier)"
            ),
        })

    def crawl(self, url: str) -> CrawledContent: 17178.baike.tongsou.com 
        """抓取单个URL"""
        content = CrawledContent(url=url)

        try:
            resp = self.session.get(url, timeout=self.timeout)
            content.status_code = resp.status_code

            if resp.status_code != 200: 17179.baike.tongsou.com 
                content.error = f"HTTP {resp.status_code}"
                logger.warning(f"[Crawler] Failed to fetch {url}: {resp.status_code}")
                return content

            # 检测内容类型
            content_type = resp.headers.get("Content-Type", "")

            if "json" in content_type: 17180.baike.tongsou.com 
                # JSON响应（如GitHub API）
                try:
                    data = resp.json()
                    content.structured_data = data
                    content.body_text = json.dumps(data, ensure_ascii=False)
                except json.JSONDecodeError:
                    content.body_text = resp.text[:2000]
            else:
                # HTML响应
                if BeautifulSoup:
                    soup = BeautifulSoup(resp.text, "html.parser")

                    # 提取标题
                    title_tag = soup.find("title")
                    content.title = title_tag.get_text(strip=True) if title_tag else ""

                    # 提取正文（简化版：取所有段落）
                    paragraphs = soup.find_all(["p", "article", "div"])
                    texts = [p.get_text(strip=True) for p in paragraphs if p.get_text(strip=True)]
                    content.body_text = "\n".join(texts)[:5000]  # 限制长度

                    # 提取JSON-LD结构化数据
                    json_ld_scripts = soup.find_all("script", type="application/ld+json")
                    for script in json_ld_scripts: 17183.baike.tongsou.com 
                        try:
                            ld_data = json.loads(script.string)
                            content.structured_data = ld_data
                        except (json.JSONDecodeError, TypeError):
                            pass
                else:
                    # 无BeautifulSoup，返回原始文本
                    content.body_text = resp.text[:2000]

            logger.info(f"[Crawler] Successfully crawled: {url}")

        except requests.exceptions.Timeout: 17184.baike.tongsou.com 
            content.error = "Request timeout"
            logger.warning(f"[Crawler] Timeout: {url}")
        except requests.exceptions.ConnectionError: 17186.baike.tongsou.com 
            content.error = "Connection error"
            logger.warning(f"[Crawler] Connection error: {url}")
        except Exception as e:
            content.error = str(e)[:200]
            logger.error(f"[Crawler] Error crawling {url}: {e}")

        return content

    def crawl_batch(self, urls: List[str]) -> List[CrawledContent]:
        """批量抓取"""
        return [self.crawl(url) for url in urls]


# ==================== 模块4: LLM解析器 ====================

class LLMParser:
    """
    使用LLM模拟大模型对抓取内容的理解和回答
    支持OpenAI API，也可降级为规则匹配
    """

    def __init__(self, api_key: str = "", model: str = "gpt-3.5-turbo"):
        self.model = model
        self.api_key = api_key or os.getenv("OPENAI_API_KEY", "")
        if openai and self.api_key: 17187.baike.tongsou.com 
            openai.api_key = self.api_key
            self.available = True
        else:
            self.available = False
            logger.warning("OpenAI API not available, using rule-based fallback")

    def parse(
        self, query: SimulatedQuery, crawled: CrawledContent
    ) -> LLMResponse:
        """
        模拟LLM基于抓取内容回答问题
        """
        if self.available and openai: 17188.baike.tongsou.com 
            return self._parse_with_llm(query, crawled)
        else:
            return self._parse_with_rules(query, crawled)

    def _parse_with_llm(
        self, query: SimulatedQuery, crawled: CrawledContent
    ) -> LLMResponse:
        """使用真实LLM API"""
        system_prompt = """你是一个知识问答助手。请基于提供的参考内容回答问题。
如果参考内容中包含答案，请准确引用；如果不包含，请说明无法回答。
回答要简洁、准确。"""

        user_prompt = f"""问题：{query.question}

参考内容：
标题：{crawled.title}
正文：{crawled.body_text[:2000]}
结构化数据：{json.dumps(crawled.structured_data, ensure_ascii=False)[:1000]}

请回答上述问题，并列出你引用的来源URL。"""

        try: 17189.baike.tongsou.com 
            response = openai.ChatCompletion.create(
                model=self.model,
                messages=[
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": user_prompt},
                ],
                temperature=0.1,
                max_tokens=500,
            )

            answer = response.choices[0].message.content.strip()

            # 简单提取引用URL
            cited_sources = re.findall(
                r'https?://[^\s"\')\]>]+', answer
            )

            # 提取关键事实（简化版）
            extracted_facts = [
                {"text": answer[:200], "confidence": 0.8}
            ]

            return LLMResponse(
                query_id=query.query_id,
                simulated_answer=answer,
                extracted_facts=extracted_facts,
                cited_sources=cited_sources,
                model_name=self.model,
            )

        except Exception as e: 17190.baike.tongsou.com 
            logger.error(f"[LLMParser] API error: {e}")
            return self._parse_with_rules(query, crawled)

    def _parse_with_rules(
        self, query: SimulatedQuery, crawled: CrawledContent
    ) -> LLMResponse:
        """
        规则降级方案
        在抓取内容中搜索关键词，模拟回答
        """
        body = crawled.body_text.lower()
        entity = query.entity.lower()
        attribute = query.attribute.lower()
        expected = query.expected_answer.lower()

        # 检查内容中是否包含关键信息
        facts = []
        if entity in body: 17191.baike.tongsou.com 
            facts.append({"text": f"找到实体: {query.entity}", "confidence": 0.9})
        if attribute in body:
            facts.append({"text": f"找到属性: {query.attribute}", "confidence": 0.8})
        if expected in body:
            facts.append({"text": f"找到答案: {query.expected_answer}", "confidence": 0.95})

        # 生成模拟回答
        if expected in body:
            answer = f"{query.entity}的{query.attribute}是：{query.expected_answer}"
        elif facts:
            answer = f"根据参考内容，{query.question}的相关信息如下：\n" + "\n".join(
                f["text"] for f in facts
            )
        else:
            answer = "参考内容中未找到相关信息。"

        return LLMResponse(
            query_id=query.query_id,
            simulated_answer=answer,
            extracted_facts=facts,
            cited_sources=[crawled.url] if crawled.url else [],
            model_name="rule-based-fallback",
        )


# ==================== 模块5: 验证引擎 ====================

class VerificationEngine: 17192.baike.tongsou.com 
    """
    闭环验证引擎
    串联查询生成、爬虫抓取、LLM解析、匹配验证
    """

    def __init__(
        self,
        crawler: Optional[WebCrawler] = None,
        llm_parser: Optional[LLMParser] = None,
        similarity_threshold: float = 0.6,
    ):
        self.crawler = crawler or WebCrawler()
        self.llm_parser = llm_parser or LLMParser()
        self.similarity_threshold = similarity_threshold

    def verify(
        self,
        unit: Dict[str, Any],
        target_urls: List[str],
    ) -> VerificationResult: 17193.baike.tongsou.com 
        """
        对单个知识单元进行闭环验证
        """
        # 1. 生成模拟查询
        query_gen = QueryGenerator()
        query = query_gen.generate(unit)

        # 2. 抓取目标URL
        crawled_contents = []
        for url in target_urls: 17194.baike.tongsou.com 
            crawled = self.crawler.crawl(url)
            crawled_contents.append(crawled)

        # 3. LLM解析并生成回答
        llm_responses = []
        for crawled in crawled_contents:
            response = self.llm_parser.parse(query, crawled)
            llm_responses.append(response)

        # 4. 合并所有回答
        combined_answer = "\n\n".join(
            r.simulated_answer for r in llm_responses
        )
        all_citations = []
        for r in llm_responses: 17197.baike.tongsou.com 
            all_citations.extend(r.cited_sources)

        # 5. 匹配验证
        expected = query.expected_answer
        match_score = self._calculate_similarity(expected, combined_answer)
        is_correct = match_score >= self.similarity_threshold
        citation_present = len(all_citations) > 0

        # 6. 发现问题
        issues = []
        if not is_correct:
            issues.append(
                f"答案匹配度低 (score={match_score:.2f}, threshold={self.similarity_threshold})"
            )
        if not citation_present:
            issues.append("未检测到引用来源")
        if any(c.error for c in crawled_contents): 17207.baike.tongsou.com 
            issues.append(
                f"部分页面抓取失败: {[c.error for c in crawled_contents if c.error]}"
            )

        result = VerificationResult(
            query_id=query.query_id,
            unit_id=unit.get("unit_id", ""),
            entity=query.entity,
            attribute=query.attribute,
            expected_value=expected,
            simulated_answer=combined_answer[:500],
            match_score=round(match_score, 4),
            is_correct=is_correct,
            citation_present=citation_present,
            issues=issues,
        )

        if is_correct:
            logger.info(
                f"[Verifier] PASS: {query.question} (score={match_score:.2f})"
            )
        else:
            logger.warning(
                f"[Verifier] FAIL: {query.question} (score={match_score:.2f}, issues={issues})"
            )

        return result

    def verify_batch(
        self,
        units_with_urls: List[Dict[str, Any]],
    ) -> VerificationReport:
        """
        批量验证
        units_with_urls: [{"unit": {...}, "urls": [...]}, ...]
        """
        report = VerificationReport()

        for item in units_with_urls:
            unit = item["unit"]
            urls = item.get("urls", [])
            result = self.verify(unit, urls)
            report.add_result(result)

        report.finalize()
        return report

    def _calculate_similarity(self, expected: str, actual: str) -> float:
        """
        计算期望答案与实际回答的相似度
        使用序列匹配 + 关键词重叠
        """
        if not expected or not actual:
            return 0.0

        expected_lower = expected.lower().strip()
        actual_lower = actual.lower().strip()

        # 方法1: 精确包含
        if