并行分支、Reducer 与子图边界
两个并行节点同一轮返回同一个普通字段，默认覆盖规则无法决定谁赢。LangGraph 会拒绝这种冲突，而不是保证最后写入者胜出。为可合并数据定义 reducer，或让各节点写独占字段。
真实 fan-out / fan-in
复用第一章环境，完整实验：
from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
evidence: Annotated[list[str],add]
summary: str
b = StateGraph(State)
b.add_node("a",lambda s:{"evidence":["A"]})
b.add_node("b",lambda s:{"evidence":["B"]})
b.add_node("join",lambda s:{"summary":"+".join(sorted(s["evidence"]))})
b.add_edge(START,"a")
b.add_edge(START,"b")
b.add_edge(["a","b"],"join")
b.add_edge("join",END)
g = b.compile()
r = g.invoke({"evidence":[],"summary":""})
assert r["summary"] == "A+B"
print(r)列表 add 把两份更新追加，join 的列表前置边等待 a、b 完成。不要把 add_edge("a","join") 和 add_edge("b","join") 在所有图形里都视为相同的联合等待语义。输出排序由我们显式指定，不依赖并行完成顺序。
Reducer 不等于业务去重
如果一个节点返回完整历史列表，而 reducer 又追加，旧项会翻倍。节点应返回本轮新增项；重试可能再次生成同一证据，使用带 ID 的去重 reducer，并在相同 ID 不同内容时报告冲突。消息用 add_messages 能按消息 ID 更新，其语义不同于普通列表 add；截断上下文时仍保留调用与结果配对。
子图可以复用共同状态字段；字段不同则用包装节点在父状态与子状态之间转换。子图只接收所需内容，秘密字段留在运行上下文/工具服务；TypedDict 中写了 secret 不会自动变成私密字段。子图失败责任由父图明确处理，不让“子图结束”冒充整个任务完成。
动态 fan-out 的 Send 适用于输入数量运行时确定的任务，数量必须受入口上限和并发配额约束。部分失败场景可以让每个分支返回带 status 的结果，再由 join 判断是否继续；未知程序异常应仍失败，不能统一伪装成空证据。
动态Send与子图字段映射
固定两分支适合任务数已知的流程。Send在运行时为每个子任务创建独立输入；worker返回的局部更新经父状态reducer汇总。子图状态类型不同，必须显式映射父body→子text，再将length映射回来，不能直接把子图所有私有字段合并进父状态。下面使用真实LangGraph，无网络模型。
import operator
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
class Child(TypedDict):
text: str
length: int
c=StateGraph(Child)
c.add_node('measure',lambda s:{'length':len(s['text'])})
c.add_edge(START,'measure');c.add_edge('measure',END)
child=c.compile()
class Parent(TypedDict):
jobs: list[dict]
results: Annotated[list[dict],operator.add]
class Job(TypedDict):
id: str
body: str
def dispatch(s):
return [Send('worker',job) for job in s['jobs']] or END
def worker(job):
try:
if not isinstance(job['body'],str):raise ValueError('body must be text')
# 父任务body映射到子图text；只发布length，不泄露子图整个状态。
result=child.invoke({'text':job['body']})
row={'id':job['id'],'ok':True,'length':result['length']}
except ValueError as e:
row={'id':job['id'],'ok':False,'error':str(e)}
return {'results':[row]}
g=StateGraph(Parent);g.add_node('worker',worker)
g.add_conditional_edges(START,dispatch);g.add_edge('worker',END)
graph=g.compile()
result=graph.invoke({'jobs':[{'id':'a','body':'abc'},{'id':'b','body':None}], 'results':[]})
rows={r['id']:r for r in result['results']}
assert rows['a']['length']==3 and not rows['b']['ok']
assert graph.invoke({'jobs':[],'results':[]})['results']==[]
print(rows)输出含a成功、b失败。异常被转成带任务ID的结果后，其他分支仍可汇总；若让异常向外抛出，整次invoke会失败，这与“部分失败结果”是不同接口。列表合并不能保证业务排序，消费者按稳定ID归并。空任务直接END，避免没有worker却等待一个不存在的汇合。
这里worker内的子图没有interrupt；如果子图也会暂停，需要把检查点/命名空间传递纳入恢复设计，不应把普通invoke示例宣传为持久子图恢复。
练习
动态分派变体：增加c任务body为空字符串，应成功返回length=0；把b改为合法字符串后，三个任务都成功且父状态没有text字段。
让 a 返回[A,A]，本例 add 会保留重复。改成去重 reducer，结果应仍只有A、B；说明它不适合需要重复次数的事件账本。