LangGraph 并行节点同时更新状态，怎样避免覆盖、重复和串任务？

实现与取舍

代码示例

按稳定证据 ID 去重并拒绝冲突

纯函数示例，演示幂等集合合并；业务允许多版本时应单独定义版本选择规则。

def merge_evidence(*batches):
    merged = {}
    for batch in batches:
        for item in batch:
            key = item["id"]
            if key in merged and merged[key] != item:
                raise ValueError("conflicting evidence: " + key)
            merged[key] = dict(item)
    return [merged[key] for key in sorted(merged)]

a = [{"id": "e1", "version": 1, "text": "approved"}]
b = [{"id": "e2", "version": 1, "text": "checked"}]
assert merge_evidence(a, b, a) == merge_evidence(b, a)
print([x["id"] for x in merge_evidence(a, b, a)])
try:
    merge_evidence(a, [{"id": "e1", "version": 2, "text": "changed"}])
except ValueError:
    print("conflict rejected")

预期输出

['e1', 'e2']
conflict rejected

工程推演

场景

面试假设：两个检索节点都更新 results，线上偶发丢失一半引用。

设计决策

按分支存储结果并用证据 ID 合并，汇总时校验分支状态。

验证目标

不同完成顺序得到同一组有效证据，重试不增加重复条目。

适用边界

余额、排序消息等数据不能照搬集合合并，需要定义独立规则。

易错点

• 对所有字段统一追加
• 依赖任务完成顺序
• 在同一状态字段混合用户偏好与临时结果

参考资料

依据公开技术资料设计；参考资料支持技术机制，场景与评分标准为本站设计，不代表某公司面试原题。 新增问答与迁移案例用于原理讲解，来源核查与案例运行验证分别记录。

• LangGraph：Thinking in LangGraph
• LangGraph：检查点与跨线程存储
• LangGraph：状态与 reducer

检查自己理解到哪一步

读完后可以对照这些标准解释原理、边界和取舍。掌握程度由你自评；需要进一步验证时，再完成下方小任务。

基础达标

能识别并行覆盖，定义字段归属。

中高级信号

给出去重、版本和汇总完成条件。

资深信号

能用乱序、冲突与重放反例证明合并语义。