并行结果、部分失败与共享状态一致性

两个研究任务谁先完成不应改变结果属于谁。并行计算的输出需按任务 ID 收集，某一路失败仍返回明确状态；汇总不能只拼接成功文本。共享状态若由最后写入者覆盖，会把先完成任务悄悄抹掉。

用可控事件打乱完成顺序

Python 3.12 完整实验，asyncio.Event 控制完成先后，不靠网络速度。

import asyncio
async def main():
    release = asyncio.Event()
    async def slow():
        await release.wait()
        return {"id":"a", "status":"success", "evidence":["A"]}
    async def fast():
        release.set()
        return {"id":"b", "status":"failed", "evidence":[]}
    results = await asyncio.gather(slow(),fast())
    indexed = {r["id"]:r for r in results}
    assert indexed["a"]["evidence"] == ["A"]
    assert indexed["b"]["status"] == "failed"
    overall = "success" if all(r["status"] == "success" for r in results) else "partial"
    assert overall == "partial"
    print(overall, sorted(indexed))
asyncio.run(main())

gather 返回输入顺序，但实际消息系统可能按完成顺序送达，因此业务依赖 ID。这个实验把失败编码为数据；若协程抛异常，默认 gather 会传播异常，不能把它误当上面的部分结果协议。包装时区分预期业务失败与程序错误，不要把所有异常吞成空答案。

合并、重复与冲突

追加式 reducer 适合独立证据，但重复投递会追加两次。用证据 ID 去重，并核对相同 ID 的内容与版本；同 ID 不同内容应冲突，不应默默覆盖。任务产物更新采用 expected_version 比较，只有持有当前版本者提交。多个作者修改同一文件时，分支隔离与最后审阅比共享聊天摘要可靠。

长期记忆写入走单独审核流程，保存来源、租户、有效期和纠错版本。一个子 Agent 的猜测不能被其他角色转发后变成多份独立证据。取消时停止创建新任务，并等待或记录已运行任务的最终状态；外部服务不支持取消时，仍需检查迟到结果的提交权。

练习

实现 merge(rows)，按 evidence id 去重；同 id 同 text 合并为一条，同 id 不同 text 抛 ValueError。测试顺序变化不改变按 id 排序后的结果。

核对本章答案

下一章