基于RAG的GEO知识增强与事实核查系统

                    ┌─────────────────────────────┐
                    │   用户查询: "2024年AI营销  │
                    │   市场规模与趋势"            │
                    └────────────┬────────────────┘
                                 │
                    ┌────────────▼────────────────┐
                    │     Query Understanding      │
                    │   (查询理解/关键词提取)        │
                    └────────────┬────────────────┘
                                 │
                    ┌────────────▼────────────────┐
                    │     Vector Retriever         │
                    │   (语义检索Top-K)            │
                    └────┬───────────┬─────────────┘
                         │           │
              ┌──────────▼──┐  ┌────▼──────────────┐
              │ Vector DB   │  │ Knowledge Graph   │
              │ (FAISS)     │  │ (实体关系检索)      │
              └──────────┬──┘  └────┬──────────────┘
                         │           │
                         └─────┬─────┘
                               │
                    ┌──────────▼──────────┐
                    │   RAG Generator      │
                    │   (LLM + 上下文)     │
                    └──────────┬──────────┘
                               │
                    ┌──────────▼──────────┐
                    │   Fact Checker       │
                    │   (事实核查/引用验证) │
                    └──────────┬──────────┘
                               │
                    ┌──────────▼──────────┐
                    │   Output: 带引用的   │
                    │   可信内容           │
                    └─────────────────────┘

"""
rag_geo_system.py - 基于RAG的GEO知识增强与事实核查系统
技术栈: Python / langchain / faiss / numpy / typing / hashlib / json
场景: GEO内容生成的知识增强、事实核查、引用溯源
"""
import os
import json
import time
import hashlib
import logging
import math
import re
from typing import List, Dict, Optional, Any, Tuple, Set
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
import numpy as np

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

class SourceType(Enum):
    """知识来源类型"""
    WEB_PAGE = "web_page"
    INDUSTRY_REPORT = "industry_report"
    ACADEMIC_PAPER = "academic_paper"
    NEWS_ARTICLE = "news_article"
    OFFICIAL_DOCUMENT = "official_document"
    SOCIAL_MEDIA = "social_media"


@dataclass
class Document:
    """文档模型"""
    doc_id: str
    content: str
    source_type: SourceType
    source_url: str
    title: str
    author: str
    published_date: str
    metadata: Dict[str, Any] = field(default_factory=dict)

    def compute_hash(self) -> str: 2001.baike.tongsou.com 
        """计算文档内容哈希，用于去重"""
        return hashlib.md5(self.content.encode("utf-8")).hexdigest()


@dataclass
class DocumentChunk: 2002.baike.tongsou.com 
    """文档分块模型"""
    chunk_id: str
    doc_id: str
    content: str
    start_char: int
    end_char: int
    chunk_index: int
    total_chunks: int
    metadata: Dict[str, Any] = field(default_factory=dict)


@dataclass
class SearchResult:
    """检索结果"""
    chunk: DocumentChunk
    score: float
    rank: int


@dataclass
class Citation: 2004.baike.tongsou.com 
    """引用信息"""
    source_title: str
    source_url: str
    source_type: str
    published_date: str
    snippet: str
    confidence: float


@dataclass
class FactCheckResult: 2005.baike.tongsou.com 
    """事实核查结果"""
    statement: str
    is_supported: bool
    confidence: float
    supporting_evidence: List[Citation]
    contradicting_evidence: List[Citation]
    verdict: str  # "supported" / "contradicted" / "insufficient_info"


@dataclass
class RAGResponse: 2006.baike.tongsou.com 
    """RAG生成响应"""
    answer: str
    citations: List[Citation]
    fact_check_results: List[FactCheckResult]
    overall_confidence: float
    response_time_ms: float
    retrieved_chunks_count: int


# ==================== 模块2: 文档加载与分块 ====================

class DocumentLoader: 2007.baike.tongsou.com 
    """
    文档加载器
    支持多种格式文档加载
    """

    def __init__(self): 2008.baike.tongsou.com 
        self.supported_formats = [".txt", ".md", ".json", ".html"]

    def load_from_text(self, text: str, source_url: str = "", title: str = "",
                       source_type: SourceType = SourceType.WEB_PAGE) -> Document:
        """从文本创建文档"""
        doc_id = hashlib.md5(text.encode("utf-8")).hexdigest()[:16]
        return Document(
            doc_id=doc_id,
            content=text,
            source_type=source_type,
            source_url=source_url,
            title=title or f"Document-{doc_id[:8]}",
            author="Unknown",
            published_date="2024-01-01",
        )

    def load_from_json(self, json_data: Dict) -> List[Document]:
        """从JSON批量加载文档"""
        documents = []
        for item in json_data.get("documents", []): 2009.baike.tongsou.com 
            doc = Document(
                doc_id=item.get("doc_id", hashlib.md5(item["content"].encode()).hexdigest()[:16]),
                content=item["content"],
                source_type=SourceType(item.get("source_type", "web_page")),
                source_url=item.get("source_url", ""),
                title=item.get("title", ""),
                author=item.get("author", "Unknown"),
                published_date=item.get("published_date", "2024-01-01"),
                metadata=item.get("metadata", {}),
            )
            documents.append(doc)
        return documents


class TextSplitter: 2010.baike.tongsou.com 
    """
    文本分块器
    支持按字符数、段落、语义等多种策略分块
    """

    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap

    def split_by_character(self, text: str, doc_id: str) -> List[DocumentChunk]:
        """按字符数分块（带重叠）"""
        chunks = []
        start = 0
        text_len = len(text)
        chunk_index = 0

        while start < text_len: 2011.baike.tongsou.com 
            end = min(start + self.chunk_size, text_len)

            # 如果不在文本末尾，尝试在句子边界切割
            if end < text_len:
                # 寻找最近的句子结束符
                for offset in range(self.chunk_overlap, 0, -1): 2012.baike.tongsou.com 
                    if end + offset < text_len and text[end + offset] in ".。!?！？":
                        end = end + offset + 1
                        break

            chunk_content = text[start:end].strip()
            if chunk_content: 2013.baike.tongsou.com 
                chunk = DocumentChunk(
                    chunk_id=f"{doc_id}_chunk_{chunk_index}",
                    doc_id=doc_id,
                    content=chunk_content,
                    start_char=start,
                    end_char=end,
                    chunk_index=chunk_index,
                    metadata={},
                )
                chunks.append(chunk)
                chunk_index += 1

            start = end - self.chunk_overlap
            if start < 0: 2014.baike.tongsou.com 
                start = 0

        # 更新total_chunks
        for chunk in chunks: 2016.baike.tongsou.com 
            chunk.total_chunks = len(chunks)

        return chunks

    def split_by_paragraph(self, text: str, doc_id: str) -> List[DocumentChunk]:
        """按段落分块"""
        paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
        chunks = []
        current_chunk = ""
        chunk_index = 0
        start_char = 0

        for para in paragraphs: 2017.baike.tongsou.com 
            if len(current_chunk) + len(para) > self.chunk_size and current_chunk:
                chunk = DocumentChunk(
                    chunk_id=f"{doc_id}_chunk_{chunk_index}",
                    doc_id=doc_id,
                    content=current_chunk.strip(2018.baike.tongsou.com),
                    start_char=start_char,
                    end_char=start_char + len(current_chunk),
                    chunk_index=chunk_index,
                    metadata={},
                )
                chunks.append(chunk)
                chunk_index += 1
                current_chunk = para
                start_char = text.find(para, start_char)
            else:
                if current_chunk:
                    current_chunk += "\n\n" + para
                else:
                    current_chunk = para
                    start_char = text.find(para)

        # 最后一个chunk
        if current_chunk: 2020.baike.tongsou.com 
            chunk = DocumentChunk(
                chunk_id=f"{doc_id}_chunk_{chunk_index}",
                doc_id=doc_id,
                content=current_chunk.strip(),
                start_char=start_char,
                end_char=start_char + len(current_chunk),
                chunk_index=chunk_index,
                metadata={},
            )
            chunks.append(chunk)

        for chunk in chunks: 2021.baike.tongsou.com 
            chunk.total_chunks = len(chunks)

        return chunks


# ==================== 模块3: 向量化与索引构建 ====================

class MockEmbeddingModel: 2022.baike.tongsou.com 
    """
    模拟嵌入模型
    实际使用时替换为OpenAI Embeddings、HuggingFace等
    """

    def __init__(self, dimension: int = 768):
        self.dimension = dimension
        # 使用固定种子确保相同输入产生相同向量
        np.random.seed(42)

    def embed(self, text: str) -> List[float]: 2023.baike.tongsou.com 
        """生成文本向量"""
        # 基于文本哈希生成确定性向量（模拟）
        hash_val = hashlib.sha256(text.encode("utf-8")).hexdigest(2024.baike.tongsou.com)
        # 将哈希转换为向量
        np.random.seed(int(hash_val[:8], 16))
        vector = np.random.randn(self.dimension).tolist()
        # 归一化
        norm = math.sqrt(sum(v * v for v in vector))
        return [v / norm for v in vector]

    def embed_batch(self, texts: List[str]) -> List[List[float]]: 2025.baike.tongsou.com 
        """批量生成向量"""
        return [self.embed(text) for text in texts]


class VectorIndex:
    """
    向量索引
    使用FAISS或简化版余弦相似度检索
    """

    def __init__(self, dimension: int = 768): 2027.baike.tongsou.com 
        self.dimension = dimension
        self.vectors: List[List[float]] = []
        self.chunks: List[DocumentChunk] = []
        self.chunk_to_idx: Dict[str, int] = {}

    def add_chunk(self, chunk: DocumentChunk, vector: List[float]):
        """添加单个分块"""
        idx = len(self.vectors)
        self.vectors.append(vector)
        self.chunks.append(chunk)
        self.chunk_to_idx[chunk.chunk_id] = idx

    def add_chunks(self, chunks: List[DocumentChunk], vectors: List[List[float]]):
        """批量添加分块"""
        for chunk, vector in zip(chunks, vectors): 2028.baike.tongsou.com 
            self.add_chunk(chunk, vector)

    def search(self, query_vector: List[float], top_k: int = 5) -> List[SearchResult]:
        """
        语义检索
        使用余弦相似度
        """
        if not self.vectors: 2029.baike.tongsou.com 
            return []

        scores = []
        for idx, vector in enumerate(self.vectors):
            # 余弦相似度
            dot_product = sum(a * b for a, b in zip(query_vector, vector))
            scores.append((idx, dot_product))

        # 按分数降序排序
        scores.sort(key=lambda x: x[1], reverse=True)

        results = []
        for rank, (idx, score) in enumerate(scores[:top_k], 1):
            results.append(SearchResult(
                chunk=self.chunks[idx],
                score=score,
                rank=rank,
            ))

        return results

    def size(self) -> int: 2030.baike.tongsou.com 
        return len(self.chunks)


# ==================== 模块4: 知识库管理 ====================

class KnowledgeBase:
    """
    知识库
    管理文档、分块、向量索引
    """

    def __init__(self, embedding_model: MockEmbeddingModel = None,
                 chunk_size: int = 500, chunk_overlap: int = 50): 2031.baike.tongsou.com 
        self.embedding_model = embedding_model or MockEmbeddingModel()
        self.splitter = TextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
        self.vector_index = VectorIndex()
        self.documents: Dict[str, Document] = {}
        self.chunks: Dict[str, DocumentChunk] = {}

    def add_document(self, doc: Document, split_strategy: str = "character") -> int:
        """
        添加文档到知识库
        返回添加的分块数量
        """
        # 去重检查
        if doc.doc_id in self.documents: 2032.baike.tongsou.com 
            return 0

        self.documents[doc.doc_id] = doc

        # 分块
        if split_strategy == "paragraph": 2034.baike.tongsou.com 
            chunks = self.splitter.split_by_paragraph(doc.content, doc.doc_id)
        else:
            chunks = self.splitter.split_by_character(doc.content, doc.doc_id)

        # 生成向量
        chunk_contents = [chunk.content for chunk in chunks]
        vectors = self.embedding_model.embed_batch(chunk_contents)

        # 添加到索引
        for chunk, vector in zip(chunks, vectors):
            chunk.metadata["doc_title"] = doc.title
            chunk.metadata["doc_url"] = doc.source_url
            chunk.metadata["doc_type"] = doc.source_type.value
            chunk.metadata["published_date"] = doc.published_date
            self.vector_index.add_chunk(chunk, vector)
            self.chunks[chunk.chunk_id] = chunk

        return len(chunks)

    def add_documents(self, docs: List[Document]) -> int:
        """批量添加文档"""
        total_chunks = 0
        for doc in docs: 2035.baike.tongsou.com 
            total_chunks += self.add_document(doc)
        return total_chunks

    def remove_document(self, doc_id: str) -> bool: 2037.baike.tongsou.com 
        """移除文档"""
        if doc_id not in self.documents: 2036.baike.tongsou.com 
            return False

        # 找到该文档的所有chunk
        chunks_to_remove = [
            chunk for chunk in self.chunks.values(2039.baike.tongsou.com)
            if chunk.doc_id == doc_id
        ]

        for chunk in chunks_to_remove: 2040.baike.tongsou.com 
            if chunk.chunk_id in self.vector_index.chunk_to_idx:
                idx = self.vector_index.chunk_to_idx[chunk.chunk_id]
                # 简化处理：标记删除，实际FAISS需要重建索引
                del self.chunks[chunk.chunk_id]

        del self.documents[doc_id]
        return True

    def search(self, query: str, top_k: int = 5) -> List[SearchResult]:
        """语义检索"""
        query_vector = self.embedding_model.embed(query)
        return self.vector_index.search(query_vector, top_k=top_k)

    def get_stats(self) -> Dict[str, Any]: 2041.baike.tongsou.com 
        """获取知识库统计信息"""
        return {
            "total_documents": len(self.documents),
            "total_chunks": len(self.chunks),
            "vector_dimension": self.embedding_model.dimension,
        }


# ==================== 模块5: RAG生成器 ====================

class RAGGenerator:
    """
    RAG生成器
    结合检索结果生成带引用的回答
    """

    def __init__(self, knowledge_base: KnowledgeBase,
                 top_k: int = 5,
                 max_context_length: int = 3000): 2042.baike.tongsou.com 
        self.kb = knowledge_base
        self.top_k = top_k
        self.max_context_length = max_context_length

    def _build_context(self, results: List[SearchResult]) -> Tuple[str, List[Citation]]:
        """
        构建上下文
        返回拼接的上下文文本和引用列表
        """
        context_parts = []
        citations = []

        for i, result in enumerate(results, 1): 2043.baike.tongsou.com 
            chunk = result.chunk
            metadata = chunk.metadata

            citation = Citation(
                source_title=metadata.get("doc_title", "Unknown"),
                source_url=metadata.get("doc_url", ""),
                source_type=metadata.get("doc_type", "web_page"),
                published_date=metadata.get("published_date", ""),
                snippet=chunk.content[:200],
                confidence=result.score,
            )
            citations.append(citation)

            context_part = f"[{i}] {metadata.get('doc_title', 'Unknown')}\n{chunk.content}\n"
            context_parts.append(context_part)

        # 截断上下文长度
        full_context = "\n\n".join(context_parts)
        if len(full_context) > self.max_context_length: 3010.baike.tongsou.com 
            full_context = full_context[:self.max_context_length] + "..."

        return full_context, citations

    def generate(self, query: str) -> RAGResponse:
        """
        生成RAG回答
        """
        start_time = time.time(3015.baike.tongsou.com)

        # 1. 检索
        results = self.kb.search(query, top_k=self.top_k)

        if not results:
            return RAGResponse(
                answer="抱歉，知识库中未找到相关信息。",
                citations=[],
                fact_check_results=[],
                overall_confidence=0.0,
                response_time_ms=(time.time() - start_time) * 1000,
                retrieved_chunks_count=0,
            )

        # 2. 构建上下文
        context, citations = self._build_context(results)

        # 3. 生成回答（模拟LLM）
        answer = self._simulate_llm_response(query, context, citations)

        # 4. 事实核查
        fact_checker = FactChecker(self.kb)
        statements = self._extract_statements(answer)
        fact_check_results = []
        for stmt in statements: 4005.baike.tongsou.com 
            fc_result = fact_checker.check(stmt)
            fact_check_results.append(fc_result)

        # 5. 计算整体置信度
        if fact_check_results: 4006.baike.tongsou.com 
            supported_count = sum(1 for r in fact_check_results if r.is_supported)
            overall_confidence = supported_count / len(fact_check_results)
        else:
            overall_confidence = 0.5

        elapsed_ms = (time.time(4012.baike.tongsou.com) - start_time) * 1000

        return RAGResponse(
            answer=answer,
            citations=citations,
            fact_check_results=fact_check_results,
            overall_confidence=overall_confidence,
            response_time_ms=elapsed_ms,
            retrieved_chunks_count=len(results),
        )

    def _simulate_llm_response(self, query: str, context: str,
                                citations: List[Citation]) -> str:
        """
        模拟LLM生成回答
        实际使用时替换为真实LLM调用
        """
        # 简化模拟：基于上下文生成回答
        citation_markers = "".join([f"[{i}]" for i in range(1, len(citations) + 1)])

        answer = f"""根据检索到的资料，关于"{query}"的回答如下：

{context[:500]}...

综合以上信息，可以得出以下结论：

1. **核心观点**：基于参考资料，该主题涉及多个维度的信息。{citation_markers}

2. **数据支持**：相关资料提供了具体的数据和案例支撑，增强了结论的可信度。

3. **趋势分析**：从时间维度看，该领域呈现持续发展的态势，未来仍有较大增长空间。

**参考文献**：
"""
        for i, cit in enumerate(citations, 1): 5002.baike.tongsou.com 
            answer += f"\n[{i}] {cit.source_title}. {cit.source_url}"

        return answer

    def _extract_statements(self, answer: str) -> List[str]:
        """
        从回答中提取需要核查的陈述
        简化版：按句子分割
        """
        # 简单按句号、感叹号、问号分割
        sentences = re.split(r"[。！？!?]", answer)
        # 过滤短句和引用标记
        statements = [
            s.strip() for s in sentences
            if len(s.strip()) > 10 and not s.strip().startswith("[")
        ]
        return statements[:5]  # 最多核查5条陈述


# ==================== 模块6: 事实核查模块 ====================

class FactChecker: 5007.baike.tongsou.com 
    """
    事实核查器
    验证陈述是否得到知识库支持
    """

    def __init__(self, knowledge_base: KnowledgeBase, threshold: float = 0.7):
        self.kb = knowledge_base
        self.threshold = threshold

    def check(self, statement: str) -> FactCheckResult: 5011.baike.tongsou.com 
        """
        核查单个陈述
        """
        # 检索相关证据
        results = self.kb.search(statement, top_k=5)

        supporting = []
        contradicting = []