并行分支、Reducer 与子图边界：练习答案

参考实现

完整实验：

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
def unique(left,right): return sorted(set(left) | set(right))
class State(TypedDict):
    evidence: Annotated[list[str],unique]
b = StateGraph(State)
b.add_node("a",lambda s:{"evidence":["A","A"]})
b.add_node("b",lambda s:{"evidence":["B"]})
b.add_edge(START,"a"); b.add_edge(START,"b")
b.add_edge("a",END); b.add_edge("b",END)
assert b.compile().invoke({"evidence":[]})["evidence"] == ["A","B"]
print("unique evidence")

同文本去重只适用于本题；生产来源应按证据 ID 和版本合并，否则两份独立文档可能被误合并。

返回正文

动态 Send 答案

将动态分派正文块保存为 send_lab.py，再运行下列独立变体。

from send_lab import graph
r=graph.invoke({'jobs':[{'id':'a','body':'abc'},{'id':'b','body':'xy'},{'id':'c','body':''}],'results':[]})
rows={x['id']:x for x in r['results']}
assert len(rows)==3 and all(x['ok'] for x in rows.values())
assert rows['c']['length']==0 and rows['b']['length']==2
assert 'text' not in r
print('three successful children; empty text is valid; child-only state not published')