冲突处理与结果合并

结果合并是把多个子任务的输出整理成一个可以继续使用的结论。冲突处理先判断这些输出是否在回答同一件事，再判断哪些依据适用。最后才决定合并、选择、继续核查或转人工。

两位Agent说法不同，未必有人算错。一位读的是国内退款规则，另一位读的是海外渠道；一位读当前规则，另一位读旧版FAQ。先把范围和版本对齐，才能知道它们是否真的矛盾。

反过来，三位Agent说法相同，也未必有三份证据。三个人都转发同一篇FAQ，证据仍只有一份。复读机开会，不会把旧规则投票变成新规则。

三票对一票，为什么最后选择5天

固定资料中，FAQ v1写3天，当前正式规则policy v2写5天。三个读取者都引用同一份FAQ，一个读取者引用当前规则；另有一条旧版正式规则，以及一条海外范围结果。

应用登记哪些来源可用、各来源当前版本，以及来源等级。合并时先排除旧版本和错误范围，再按登记的等级选择依据，最后检查同级依据是否一致。

这套业务规则会选出5天。若两位读取者都声称引用policy v2，却分别写5天与7天，程序保留conflict，表示需要重新读取原资料或人工核对。它没有凭Agent名字决定谁赢。

保存完整程序并运行

在自己的学习目录打开PowerShell，创建一个新的子目录：

mkdir l6-conflicts
cd l6-conflicts

把下面完整代码保存为 conflicts_demo.py。固定数据和运行入口都已包含，文件名保留 .py 后缀。

"""合并关于退款到账天数的候选结论。"""
from dataclasses import dataclass
from collections import Counter

@dataclass(frozen=True)
class Claim:
    agent: str
    days: int
    source: str
    version: int
    scope: str = "domestic"

# 来源登记由应用提供，不接受Agent自己给自己评权威等级。
SOURCES = {"policy": {"rank": 2, "version": 2}, "faq": {"rank": 1, "version": 1}}

def merge(claims, scope="domestic"):
    valid = [c for c in claims if c.scope == scope and c.source in SOURCES
             and c.version == SOURCES[c.source]["version"]]
    if not valid:
        return {"status": "unknown", "days": None, "sources": []}
    best_rank = max(SOURCES[c.source]["rank"] for c in valid)
    finalists = [c for c in valid if SOURCES[c.source]["rank"] == best_rank]
    values = {c.days for c in finalists}
    # 同一资料被三位Agent复制，仍然只算一个来源。
    sources = sorted({c.source for c in finalists})
    if len(values) != 1:
        return {"status": "conflict", "days": None, "sources": sources}
    return {"status": "resolved", "days": values.pop(), "sources": sources}

def main():
    candidates = [
        Claim("reader-A", 3, "faq", 1),
        Claim("reader-B", 3, "faq", 1),
        Claim("reader-C", 3, "faq", 1),
        Claim("policy-reader", 5, "policy", 2),
        Claim("old-reader", 2, "policy", 1),
        Claim("overseas-reader", 7, "policy", 2, "overseas"),
    ]
    result = merge(candidates)
    majority = Counter(claim.days for claim in candidates).most_common(1)[0][0]
    support = sorted({claim.source for claim in candidates if claim.days == majority})
    print(f"人数多数：{majority}天；去重来源=" + ",".join(support))
    print(f"规则合并：{result['status']}；天数={result['days']}；来源={','.join(result['sources'])}")
    conflict = merge([Claim("A", 5, "policy", 2), Claim("B", 7, "policy", 2)])
    print(f"同级矛盾：{conflict['status']}；天数={conflict['days']}")
    print("无有效依据：" + merge([Claim("C", 1, "unverified", 1)])["status"])

if __name__ == "__main__":
    main()

保存后执行 dir，确认能看到 conflicts_demo.py，再运行：

ls
python -X utf8 conflicts_demo.py

程序输出：

人数多数：3天；去重来源=faq
规则合并：resolved；天数=5；来源=policy
同级矛盾：conflict；天数=None
无有效依据：unknown

合并规则按什么顺序工作

每条Claim包含agent、days、source、version与scope。agent是产出者，days是候选天数，source是依据编号，version是依据版本，scope是适用范围。这个候选对象把“谁说的”与“依据是什么”分开保存。

merge首先筛scope。默认只处理domestic，即国内范围；海外结果不会参与国内问题的判断。随后要求source存在于SOURCES，并且版本等于登记中的当前版本。

这两项成立后，才比较rank，即来源等级。rank由应用维护的登记表给出，候选作者不能给自己填写“权威999级”。例子登记正式规则为2、FAQ为1，这是当前应用的合并政策；其他业务需要根据资料治理规则配置，而不是把数字当成普遍真理。

finalists是最高等级的候选。它们天数一致，就产生resolved及对应天数；同级候选出现不同天数，产生conflict且days为None；一个有效候选都没有，产生unknown。None表示没有可确定的值，不是0天。

合并前核对来源与适用范围

sources用集合去重。同一份FAQ被三个读取者引用，输出只列一个faq来源。去重防止重复传播被误认为独立支持，却不能证明资料内容正确。

来源编号相同、引用片段不同，也可能仍有真冲突。有人漏读条件，有人混用了支付渠道。保留候选原文、片段定位和读取版本，比只保存“5天”更方便追查。程序先用可核对的小字段理解流程，真实资料还需校验引用确实指向该内容。

“较高等级获胜”也有适用前提。当前正式规则如果并不适用于该订单，不能因为它等级高就压过正确的特殊渠道规则。范围字段应表达影响判断的业务条件，必要时包括渠道、地区、生效时间和客户类型。

合并方式要匹配数据含义

多份资料描述互补事实时，可以按字段合并，例如一位提供付款金额，另一位提供发货状态。多份资料对同一事实给不同值时，需要冲突规则。投票用于某些分类任务时，也要检查样本是否独立、是否用了同一模型和同一证据。

字符串拼接能保留两段文字，却可能把“3天”和“5天”同时交给客户。简单平均得到4天也没有业务依据。账务金额、政策条件和权限判断通常需要能指出来源的决策。

无法确定时，保留冲突对象和待查事项比静默覆盖更有用。后续协调者应识别conflict与unknown，停止依赖它们的付款或承诺动作；状态为resolved也仍需遵守原有执行权限。

改适用范围，观察候选集合变化

把main中的merge(candidates)改为merge(candidates, scope="overseas")，运行python -X utf8 conflicts_demo.py。规则合并那一行变为resolved、天数7、来源policy。国内FAQ和国内规则因范围不同被排除。

再把同级矛盾示例中第二条Claim的7改为5，运行同一条命令。同级矛盾那一行会变为resolved和天数5；它仍只有policy这一个来源，没有因为两个读取者得出相同结论而多一份证据。

随后把主结果的merge调用恢复为merge(candidates)，把第二条同级Claim恢复为7天。再运行时，主结果是国内5天，同级示例是conflict；确认这两行后继续配套检查。

看到unknown时，先检查范围、来源登记和版本。看到conflict时，保留双方引用并核对原资料；修改投票人数不会消除同级证据的内容矛盾。

参考阅读：LangGraph状态合并与reducers，重点看同一字段的多个更新如何按显式规则处理。退款来源等级与范围过滤是上面应用自行定义的政策。

运行配套检查

把下面测试完整保存为 test_conflicts_demo.py，与主程序放在同一目录。6项检查覆盖来源选择、复制去重、旧版排除、范围过滤、同级冲突及无依据结果。

"""检查来源去重、范围、版本和未决冲突。"""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import conflicts_demo as demo

class ConflictTests(unittest.TestCase):
    def test_authoritative_current_source_wins(self):
        claims = [demo.Claim("A", 3, "faq", 1), demo.Claim("B", 5, "policy", 2)]
        self.assertEqual(demo.merge(claims)["days"], 5)

    def test_copies_do_not_add_sources(self):
        claims = [demo.Claim(str(i), 3, "faq", 1) for i in range(3)]
        self.assertEqual(demo.merge(claims)["sources"], ["faq"])

    def test_stale_version_is_removed(self):
        self.assertEqual(demo.merge([demo.Claim("A", 2, "policy", 1)])["status"], "unknown")

    def test_other_scope_is_removed(self):
        claim = demo.Claim("A", 7, "policy", 2, "overseas")
        self.assertEqual(demo.merge([claim])["status"], "unknown")
        self.assertEqual(demo.merge([claim], "overseas")["days"], 7)

    def test_same_rank_conflict_is_not_silently_resolved(self):
        claims = [demo.Claim("A", 5, "policy", 2), demo.Claim("B", 7, "policy", 2)]
        self.assertEqual(demo.merge(claims), {"status": "conflict", "days": None, "sources": ["policy"]})

    def test_unknown_source_and_empty_list(self):
        self.assertEqual(demo.merge([])["status"], "unknown")
        self.assertEqual(demo.merge([demo.Claim("A", 1, "self_claim", 1)])["status"], "unknown")

if __name__ == "__main__":
    unittest.main()

python -X utf8 -m unittest -v test_conflicts_demo.py

测试名称后应显示 ok，末尾显示 Ran 6 tests 和 OK。若找不到模块，先用 dir 确认两个文件保存在同一目录。