用LangGraph写多Agent协作，状态同步总是乱，求大佬指点思路

这问题我太熟了，之前搞多Agent也踩过同样的坑。你自定义Reducer合并list重复，大概率是因为没做去重或者没处理节点重入的情况，建议先给每条结果加个唯一id再合并。另外别硬扛着手动传状态，直接用Checkpoint机制吧，至少能保证节点失败重跑时状态是干净的，省得排查半天发现是旧数据在捣乱。参考项目的话，可以看看LangGraph官方那个multi-agent的例子，虽然简单但状态流设计得挺清晰。

之前也踩过这个坑，子图覆盖父图状态是LangGraph挺典型的坑。我的做法是每个子Agent返回时只保留增量字段，用命名空间前缀区分不同子图的数据，比如research_result和writing_result，这样就算内部跑子图也不会互相覆盖。Reducer合并list重复的话，试试在自定义Reducer里先判断item是否已存在，用id做去重键，别直接extend。Checkpoint机制建议用上，但别依赖它做状态同步，它更适合恢复和调试，实时数据还是要靠显式传递。你可以看下LangGraph官方那个multi-agent论文搜索的例子，里面对状态隔离处理得挺清楚。

之前也踩过类似的坑，子图嵌套后状态覆盖基本是默认行为，得显式在子图里传父图的state引用才行。Reducer合并list时试试用operator.add或者自己搞个去重逻辑，但更省事的办法是给每个节点单独建key，别共用一个list字段。另外强烈建议看下官方那个multi-agent supervisor的例子，里面用Checkpoint配合状态隔离的思路比手动维护dict靠谱多了，你这种流水线其实更适合每个子Agent独立存储中间结果。

这问题我踩过，试试把子图状态提升到父图里，别让子Agent自己维护dict。

这题我踩过类似的坑，关键问题不是状态覆盖，而是子图里的状态key和父图共用导致的。建议把三个子Agent的state定义成完全隔离的dataclass，别图省事全塞一个dict里，Reducer只处理真正需要合并的字段，比如检索结果用add_node这种自定义合并逻辑。另外并行节点共享状态这事，我后来干脆把检索结果按query分块存成独立key，写完再汇总，绕开并发写冲突。Checkpoint机制解决的是持久化和恢复，对你说这个同步问题帮助不大，别指望它。

状态别手搓了，直接上Checkpoint配自定义Reducer，list去重试试用operator.add加个类型判断。

你这图结构问题不大，主要是子图覆盖了父状态，建议把检索结果单独放一个key别跟其他混着传。

说实话你这个状态覆盖的问题我太懂了，刚用LangGraph那会儿也被子图嵌套坑得死去活来。核心问题多半出在你对状态作用域的假设上——子Agent内部对dict的修改默认是整体覆盖的，不是按key合并，所以检索子图返回时直接把外层规划状态冲掉了。我后来是彻底放弃手动传dict，改用带命名空间的StateSchema，每个子Agent单独定义自己的状态类型，然后通过总图的stateAnnotator配Reducer做字段级合并，这样并行节点之间至少不会互相踩踏。你提到的ListReducer重复问题，我记得官方文档里有个例子是专门去重加排序的，你搜一下ConcatListWithTimestamp，那个能解决大部分合并冗余。另外Checkpoint机制不是用来解决同步的，它只是持久化，你真正该查的是LangGraph的interrupt和动态重写逻辑，看看能不能让写作节点显式等待检索节点完成并返回一个版本号，比对一下版本再读取。如果允许的话，建议把三个子Agent拆成完全独立的图，用消息队列或者数据库做中间件通信，别在同一个图里硬塞流水线，状态隔离性会好很多，虽然重一点但至少不玄学。最后给你个参考，GitHub上有个research-agent-langgraph的项目，结构和你很像，你看看它怎么处理子图返回值的，应该能少走很多弯路。

说实话你这问题我踩过类似的坑，LangGraph的state是全局单例，子图里如果直接对同一个key赋值，肯定会覆盖外部节点的数据。我后来是给每个子Agent单独建一个命名空间，比如state["research"]和state["write"]分开存，最后再在总图里合并。Reducer那个确实难写，建议直接看下官方文档的addNode例子，或者用dataclass定义state，每个字段单独指定reducer，比手动处理dict清晰得多。另外如果可以的话，把检索结果存到checkpoint而不是state里，节点间只传引用ID，能少很多同步问题。

说实话你这个状态覆盖的问题，我一开始用LangGraph也踩过一模一样的坑。子图嵌套时，外层dict的key如果跟内层子图返回的key重名，直接就被静默覆盖了，而且LangGraph的默认行为是“后写覆盖”，根本不会报错。我后来是强制给每个子Agent的返回包了一层命名空间，比如research_result、writing_result，虽然丑了点但至少不会串数据。

至于Reducer，你说合并list重复，这太正常了——官方那个add_messages只是针对消息列表设计的，你拿它合并自定义的检索片段列表，它只会无脑append，根本不会去重。我当时干脆自己写了个带hash去重的合并函数，用set记录已见过的内容指纹，然后再转回list，虽然性能差点但稳。不过说实话，如果你每个节点之间逻辑是严格串行的，根本不需要Reducer，直接用Checkpoint把中间结果存下来，每个节点从checkpoint里读自己需要的那份数据反而更清晰。

你那个“规划-检索-写作”流水线，我建议别用Send做动态并行，除非你真的需要同时跑多个检索子任务。因为并行节点共享状态时，LangGraph的State是整体快照，你一个节点改了list，另一个节点拿到的可能还是旧快照，除非你显式声明了reducer去合并。我后来改成完全串行，检索节点内部自己用asyncio并发调多个搜索API，把结果汇总成一个list再返回，写作节点永远只读这个汇总后的list，问题就消失了。

另外你提到参考项目，GitHub上有个叫langgraph-multi-agent-demo的仓库，里面有个类似的研究助手实现，虽然它用的是老版本API，但状态管理思路挺值得借鉴的。它每个子Agent都单独定义一个State类，然后用TypedDict嵌套，父图只调度不直接传大对象，靠Checkpoint持久化，基本规避了覆盖问题。建议你也试试把状态拆成不可变的数据类，而不是一个松散的大dict，这样就算子图内部怎么写，父图拿到的引用也不会被意外改动。卡三天很正常，LangGraph这框架文档写得跟谜语人似的，多翻issue区比看官方文档管用。

说实话我建议你直接上Checkpoint机制，手动传dict在嵌套子图场景下迟早要出问题，LangGraph的持久化状态本来就是干这个用的。Reducer写不好就别纠结合并list了，试试给每个节点单独一个state字段，用命名空间隔离不同子Agent的数据，能省掉一大半冲突。另外可以参考下LangGraph官方那个multi-agent supervisor的例子，里面处理并行节点共享状态的方式挺干净的。

说实话，你这个情况我太熟了，之前做多模态归纳Agent的时候也被子图状态覆盖坑过一整周。核心问题其实不是Reducer写不好，而是你把“子Agent内部状态”和“主流水线状态”混在同一个dict里了，LangGraph的节点返回默认是全量覆盖，子图一返回就把父级键给冲了。我当时是强制规定每个子Agent只返回自己的命名空间字段，比如retrieval_result、writing_output，主图里手动merge，不依赖隐式更新，这样就算并行也互不干扰。至于Reducer合并list重复，我建议别用自定义Reducer了，直接在节点里做完去重再返回，或者用set类型存中间结果，最后转list，省心很多。Checkpoint机制我理解是用来做断点续跑和回溯的，不是用来解决状态同步的，你就算开了checkpoint，该覆盖还是覆盖。另外你提到Send API，那个适合真正的动态并行分支，像你这种固定三步流水线其实用普通边就够了，不需要搞那么复杂。参考项目的话，LangGraph官方有个multi-agent supervisor的例子，虽然场景不完全一样，但状态管理那部分写得挺清晰，你可以看看它是怎么隔离子图状态的。卡三天不丢人，这框架的坑就是得踩一遍才长记性。

说实话你这个问题我太有共鸣了，之前做类似的多Agent流水线时也被状态覆盖坑过一整周。LangGraph的dict传参看起来简单，但一旦子图嵌套，父节点的key被内部覆盖是常态，尤其是你想让写作节点读最新检索结果，可子Agent自己又改了同一个字段，这本质上是图结构设计的问题，不是Reducer能完全兜底的。我后来学乖了，所有跨节点的共享数据全部通过显式的Channel或者独立的state key来隔离，比如检索结果直接挂在node_outputs_前缀下，绝不跟主流程的dict混在一起，这样至少不会静默覆盖。至于并行节点共享状态，我建议你试试把每个子Agent的输入输出做成不可变的快照，节点之间只传引用ID而不是原始数据，配合LangGraph的Checkpoint机制做持久化，这样就算子图内部乱写，父图恢复时也能拿到正确的历史版本。自定义Reducer确实容易写崩，尤其是合并list时，我后来干脆用简单的set去重或者给每条结果加个uuid，再按时间戳排序，比纠结Reducer的merge逻辑省心多了。另外你说参考项目，GitHub上有个叫agent-flow的demo，专门做研究助手的三段式流水线，里面有处理子图状态冲突的例子，你可以搜下看看，但别全抄，它的做法是每个节点独立命名空间，用起来挺稳的。最后想问下，你那个Send API动态分支是不是在并行节点里又改了同一个父级状态？如果是，那问题可能出在分支返回值的合并策略上，试试把并行结果包一层list再merge。

碰到这种问题大概率是子图覆盖了父图的状态键，你可以把子Agent的状态设计成独立的TypedDict，用命名空间隔离，别直接复用父图的key。Reducer其实没那么难，写个按id去重的函数就行，或者用operator.add配合add_marker。另外建议把需要共享的中间结果单独放到一个专门的shared字段里，别让子图去碰父图的其他键。我之前也是被这个坑过，后来看到LangGraph官方有个multi-agent的示例，里面用了subgraph和状态映射，抄那个结构基本能避开。

试试把子图状态收敛到父图命名空间里，Reducer用operator.add配RemoveDuplicates，能少踩不少坑。

我上次也是这问题，后来干脆用Checkpoint做全局快照，节点只读状态，写完回写，乱套的毛病直接没了。

说实话你这个状态覆盖问题我太熟了，之前搞类似的多Agent流水线也踩过这个坑。LangGraph的dict传参看着简单，但一旦子图嵌套，内层返回的state会直接整体覆盖外层同名key，压根不是你想的“合并”语义。

我后来干脆不用Send做并行写入了，改成在子Agent里显式声明input和output的key，用Annotated类型标注Reducer，比如对检索结果用operator.add，但注意得保证每个子Agent返回的都是新list而不是在原有list上append，不然还是重复。自定义Reducer其实不用写太复杂，直接operator.add配上去重逻辑就行，但关键是每个节点返回的数据结构必须严格统一。

另外你提到Checkpoint，这确实是另一个思路——把检索结果持久化到checkpoint里，写作节点从checkpoint读而不是从state读，相当于绕开共享状态问题。不过我觉得你根本问题还是图设计，三个Agent流水线其实不用搞成嵌套子图，平铺成一条链，每个节点只改自己的key，检索结果单独放一个字段，写作节点只依赖那个字段，就不会互相污染。

网上有个LangGraph官方仓库里的multi-agent研究助手示例，虽然逻辑简单点，但状态管理是清楚的，你可以对照着改。最后我建议你调试时把每个节点进出state都打印出来，看到底哪一步被覆盖，比瞎猜快很多。

说实话你这问题我太有共鸣了，上周刚被类似的状态覆盖坑过一整天。LangGraph的dict传递看着简单，但一旦子图嵌套深了，父节点和子节点对同一个key的写入顺序完全不可控，尤其是并行分支，基本就是看运气。我当时解决的办法是彻底放弃手动维护共享状态，把每个子Agent的输入输出都定义成独立的命名空间，比如在key前面加前缀“plan_”、“search_”，这样就算子图内部覆盖，也不会碰坏别的节点数据，虽然有点土但确实管用。

关于Reducer去重这事，我试过用set代替list，但顺序又丢了，后来干脆在自定义Reducer里用“hash+时间戳”做去重，先比对内容指纹，再保留最新写入的那个，比单纯判断in要稳得多。不过说实话，如果你节点间数据量不大，我个人觉得别硬上Reducer，直接在节点函数里做显式合并更直观，出了问题也好调试。

Checkpoint机制我倒是用过几次，但它主要解决的是恢复和重放问题，不是并发写入冲突。你现在的核心矛盾是“读旧数据”，本质上是执行顺序没保证，建议你检查一下图定义里有没有用显式的edge来控制依赖，而不是只靠Send动态分支。另外我有个疑问，你的检索节点是不是也用了子图，而且子图里还有自己的state？如果是的话，那个子图返回的时候会把父图的state整个覆盖掉，这可能是真正的元凶，试试在子图出口加个filter只更新指定字段。

参考项目的话，你可以去翻下LangGraph官方repo里的multi-agent例子，虽然都是简单demo，但那个“supervisor+worker”的模式对你这种流水线挺有启发的。最后说一句，三天不算久，我上次调这个调了五天，最后发现是图里一个隐式的shared节点在作怪，删掉就全通了。