引用溯源：RAG答案可信度的护城河

# 简化版 RAG 引用溯源演示: 检索 -> 定位 -> 冲突处理 -> 可信度分级class SourceDoc:    """来源文档片段"""    def __init__(self, doc_id, text, authority=0.5):        self.doc_id = doc_id          # 文档 ID        self.text = text              # 片段原文        self.authority = authority    # 来源权威度        self.offset = 0               # 在原文中的字符偏移量        self.version = 1              # 版本号(快照指纹)  class CitationSystem:    """引用溯源系统"""    def __init__(self):        self.doc_map = {}             # 文档 ID -> 片段对象        self.check_count = 0          # 冲突检测计数     def add_source(self, doc):        """登记来源片段"""        self.doc_map[doc.doc_id] = doc        return doc.doc_id     def locate(self, keyword, doc):        """精确定位关键词在原文中的偏移量"""        idx = doc.text.find(keyword)        if idx >= 0:            doc.offset = idx            return True        return False     def lcs_length(self, a, b):        """计算最长公共子序列长度"""        m, n = len(a), len(b)        dp = [[0] * (n + 1) for _ in range(m + 1)]        for i in range(1, m + 1):            for j in range(1, n + 1):                if a[i - 1] == b[j - 1]:                    dp[i][j] = dp[i - 1][j - 1] + 1                else:                    dp[i][j] = max(dp[i - 1][j], dp[i][j - 1])        return dp[m][n]     def match_score(self, sentence, doc):        """计算句子与原文片段的 LCS 匹配度"""        lcs = self.lcs_length(sentence, doc.text)        return 2.0 * lcs / (len(sentence) + len(doc.text) + 1e-9)     def detect_conflict(self, evidence_map):        """检测多源冲突: 同一主张出现相反立场"""        claims = {}        for doc_id, items in evidence_map.items():            for item in items:                key = item["claim"]                claims.setdefault(key, []).append((doc_id, item["support"]))        return {k: v for k, v in claims.items() if len(v) > 1}     def weighted_vote(self, conflict_claims):        """加权投票解决冲突"""        result = {}        for claim, votes in conflict_claims.items():            score = 0.0            for doc_id, support in votes:                doc = self.doc_map[doc_id]                score += doc.authority * (1.0 if support else -1.0)            result[claim] = score        return result     def credibility(self, authority, relevance, recency, consensus):        """可信度分级: 权威度+相关度+时效性+一致性"""        return 0.3 * authority + 0.3 * relevance + 0.2 * recency + 0.2 * consensus  def main():    system = CitationSystem()    doc_a = SourceDoc("A001", "Vaccine efficacy requires long-term follow-up.", 0.9)    doc_b = SourceDoc("B002", "Vaccine efficacy looks fine in short term.", 0.6)    system.add_source(doc_a)    system.add_source(doc_b)     print("locate A:", system.locate("Vaccine efficacy", doc_a), "offset:", doc_a.offset)    print("locate B:", system.locate("Vaccine efficacy", doc_b), "offset:", doc_b.offset)     sent = "Vaccine efficacy needs long-term data"    print("match score A:", round(system.match_score(sent, doc_a), 3))    print("match score B:", round(system.match_score(sent, doc_b), 3))     evidence = {        "A001": [{"claim": "long-term safety", "support": True}],        "B002": [{"claim": "long-term safety", "support": False}],    }    conflicts = system.detect_conflict(evidence)    print("conflicts:", conflicts)     votes = system.weighted_vote(conflicts)    print("weighted votes:", votes)     print("credibility:", round(system.credibility(0.9, 0.85, 0.7, 0.8), 2))  if __name__ == "__main__":    main()