# topic.py:23
class Topic(
    Generic[Value],
    BaseChannel[Sequence[Value], Value | list[Value], list[Value]],
):
    """A configurable PubSub Topic.

    Args:
        typ: The type of the value stored in the channel.
        accumulate: Whether to accumulate values across steps.
            If `False`, the channel will be emptied after each step.
    """
    __slots__ = ("values", "accumulate")

构造函数简单到极点（channels/topic.py:36-41）：

# topic.py:36
def __init__(self, typ: type[Value], accumulate: bool = False) -> None:
    super().__init__(typ)
    # attrs
    self.accumulate = accumulate
    # state
    self.values = list[Value]()          # 起点永远是空列表 []

# topic.py:15
def _flatten(values: Sequence[Value | list[Value]]) -> Iterator[Value]:
    for value in values:
        if isinstance(value, list):
            yield from value          # 是列表 → 逐个吐出（摊平一层）
        else:
            yield value               # 是单值 → 原样吐出

Topic 最有特色的能力藏在 update 的头几行（channels/topic.py:77-85）：

# topic.py:77
def update(self, values: Sequence[Value | list[Value]]) -> bool:
    updated = False
    if not self.accumulate:                 # ← 不累积模式：先清空旧的
        updated = bool(self.values)         #   本来有东西 → 算"变化了"
        self.values = list[Value]()         #   开一个全新空列表
    if flat_values := tuple(_flatten(values)):
        updated = True
        self.values.extend(flat_values)     # 把这一步的新值排进去
    return updated

把 update 和 get 连起来看完整取用链路（channels/topic.py:77-91）：

# topic.py:77
def update(self, values):
    updated = False
    if not self.accumulate:
        updated = bool(self.values)
        self.values = list[Value]()
    if flat_values := tuple(_flatten(values)):   # 摊平后物化成 tuple
        updated = True
        self.values.extend(flat_values)
    return updated                                # 告诉引擎"这步这个通道变没变"

# topic.py:87
def get(self) -> Sequence[Value]:
    if self.values:
        return list(self.values)                 # 返回一份拷贝，防外部改内部
    else:
        raise EmptyChannelError                   # 空列表 = 通道"没值"

Topic 是要持久化的（对比明天的 EphemeralValue 就不存）。它的存/取很直白（channels/topic.py:63-75）：

# topic.py:63
def checkpoint(self) -> list[Value]:
    return self.values                        # 直接把整个列表交出去存档

# topic.py:66
def from_checkpoint(self, checkpoint: list[Value]) -> Self:
    empty = self.__class__(self.typ, self.accumulate)
    empty.key = self.key
    if checkpoint is not MISSING:
        if isinstance(checkpoint, tuple):
            # backwards compatibility
            empty.values = checkpoint[1]      # ← 旧版本存的是 (guard, values) 二元组
        else:
            empty.values = checkpoint         # 新版本直接就是 list
    return empty

✋ 10 分钟动手

# 1. 通读 Topic 全文（95 行）
sed -n '1,95p' libs/langgraph/langgraph/channels/topic.py

# 2. 重点对照 _flatten 与 update 的清空分支
sed -n '15,20p;77,85p' libs/langgraph/langgraph/channels/topic.py

# 3. 亲手验证两种模式的差别
python - <<'PY'
from langgraph.channels.topic import Topic
acc = Topic(str, accumulate=True)
acc.update(["a"]); acc.update([["b","c"]])   # 单值/列表混投
print("累积:", acc.get())                     # ['a','b','c']
tmp = Topic(str, accumulate=False)
tmp.update(["x"]); print("步1:", tmp.get())   # ['x']
changed = tmp.update([])                       # 空投递
print("步2 changed?", changed)                # True（因为清空了旧的）
PY