diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 8c9a54d8b..4fdc67792 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -209,6 +209,13 @@ def test_parallel_nodes() -> None: graph = builder.compile() print("\n======COMPLETE GRAPH======\n", pregel_pretty(graph)) + + print(f"\n🔍 CHANNEL CONFIGURATION:") + print(f" 📤 output_channels: {graph.output_channels}") + print(f" 📡 stream_channels: {graph.stream_channels}") + print(f" 📥 input_channels: {graph.input_channels}") + print(f" 🌊 stream_channels_asis: {graph.stream_channels_asis}") + print(f" 📋 stream_channels_list: {graph.stream_channels_list}") result = graph.invoke({"hello": "there"}) assert result["hello"] == "world-d" diff --git a/pregel.learning.md b/pregel.learning.md new file mode 100644 index 000000000..739022cdf --- /dev/null +++ b/pregel.learning.md @@ -0,0 +1,153 @@ +# LangGraph Channel Types Usage Analysis + +本文档分析了 LangGraph 中各种通道类型在 StateGraph 中的使用情况,基于对整个仓库的深入分析。 + +## 通道类型使用总结 + +### 1. **LastValue** - 最常用的默认通道 +**使用场景:** 普通状态字段的默认通道类型 +**创建方式:** 自动创建(fallback) +**仓库例子:** +```python +class State(TypedDict): + hello: str # 自动创建 LastValue(str) 通道 + count: int # 自动创建 LastValue(int) 通道 +``` + +### 2. **BinaryOperatorAggregate** - 状态聚合通道 +**使用场景:** 使用 reducer 函数进行状态聚合 +**创建方式:** 通过 `Annotated[Type, reducer_function]` +**仓库例子:** +```python +# 例子1: 使用 add_messages +class State(TypedDict): + messages: Annotated[list[str], add_messages] # 创建 BinaryOperatorAggregate + +# 例子2: 使用 operator.add +StateGraph(Annotated[str, operator.add]) # 整个状态使用聚合 +StateGraph(Annotated[list, operator.add]) # 列表聚合 + +# 例子3: 使用 operator.or_ +class State(TypedDict): + val3: Required[Annotated[dict, operator.or_]] # 字典合并 +``` + +### 3. **EphemeralValue** - 临时通道 +**使用场景:** 系统内部使用,节点间的临时通信 +**创建方式:** 系统自动创建 +**仓库例子:** +```python +# StateGraph.compile() 中自动创建 +START: EphemeralValue(self.input_schema) # 输入通道 + +# CompiledStateGraph.attach_node() 中创建 +EphemeralValue(Any, guard=False) # 节点分支通道 +``` + +### 4. **LastValueAfterFinish** - 延迟通道 +**使用场景:** 节点设置了 `defer=True` 时使用 +**创建方式:** 系统根据节点配置自动创建 +**仓库例子:** +```python +# CompiledStateGraph.attach_node() 中 +if node.defer: + LastValueAfterFinish(Any) # 延迟节点的分支通道 +``` + +### 5. **NamedBarrierValue** - 同步屏障通道 +**使用场景:** 多个节点汇聚到一个节点时的同步 +**创建方式:** 系统在处理 waiting_edges 时自动创建 +**仓库例子:** +```python +# CompiledStateGraph.attach_edge() 中 +channel_name = f"join:{'+'.join(starts)}:{end}" +self.channels[channel_name] = NamedBarrierValue(str, set(starts)) +``` + +### 6. **NamedBarrierValueAfterFinish** - 延迟同步屏障通道 +**使用场景:** 延迟节点的多节点汇聚同步 +**创建方式:** 系统在处理延迟节点的 waiting_edges 时自动创建 +**仓库例子:** +```python +# CompiledStateGraph.attach_edge() 中 +if self.builder.nodes[end].defer: + self.channels[channel_name] = NamedBarrierValueAfterFinish(str, set(starts)) +``` + +### 7. **Topic** - 发布订阅通道 +**使用场景:** 仅在直接使用 Pregel 时手动创建,StateGraph 中无实际使用 +**创建方式:** 手动创建或系统内部 TASKS 通道 +**仓库例子:** +```python +# Pregel.__init__() 中系统创建 +self.channels[TASKS] = Topic(Send, accumulate=False) + +# 直接 Pregel 使用(非 StateGraph) +app = Pregel( + channels={"c": Topic(str, accumulate=True)}, # 手动创建 + # ... +) +``` + +### 8. **UntrackedValue** - 未追踪通道 +**使用场景:** 在仓库中未找到实际使用例子 +**创建方式:** 需要手动创建 +**状态:** 理论存在但实际未使用 + +### 9. **AnyValue** - 任意值通道 +**使用场景:** 在仓库中未找到实际使用例子 +**创建方式:** 需要手动创建 +**状态:** 理论存在但实际未使用 + +## 使用模式总结 + +### **用户显式创建的通道:** +1. **BinaryOperatorAggregate** - 通过 `Annotated[Type, reducer]` +2. **Topic** - 仅在直接 Pregel API 中手动创建 + +### **系统自动创建的通道:** +1. **LastValue** - 默认通道类型 +2. **EphemeralValue** - 输入和分支通道 +3. **LastValueAfterFinish** - 延迟节点分支 +4. **NamedBarrierValue** - 多节点汇聚同步 +5. **NamedBarrierValueAfterFinish** - 延迟节点汇聚同步 + +### **实际使用频率:** +1. **高频使用:** LastValue, BinaryOperatorAggregate, EphemeralValue +2. **中频使用:** NamedBarrierValue, LastValueAfterFinish +3. **低频使用:** Topic(仅系统内部) +4. **未使用:** UntrackedValue, AnyValue + +## 通道创建机制 + +### 自动创建流程 +1. **StateGraph._add_schema()** - 解析状态类型 +2. **_get_channels()** - 提取类型注解 +3. **_get_channel()** - 判断通道类型: + - 检查 Managed Value + - 检查 Channel(如 Topic) + - 检查 BinaryOperator(如 add_messages) + - 默认创建 LastValue + +### 编译时创建 +- **START 通道:** `EphemeralValue(input_schema)` +- **分支通道:** `EphemeralValue(Any, guard=False)` 或 `LastValueAfterFinish(Any)` +- **汇聚通道:** `NamedBarrierValue` 或 `NamedBarrierValueAfterFinish` + +## 设计哲学 + +**StateGraph 主要关注状态管理,大部分通道类型都是系统自动管理的,用户只需要关心状态结构和聚合逻辑**。只有在需要特殊聚合行为时,用户才需要显式使用 `Annotated` 注解来指定 reducer 函数。 + +## 关键发现 + +1. **Topic 通道在 StateGraph 中几乎不使用** - 仓库中没有通过状态 Schema 注解创建 Topic 的实际例子 +2. **BinaryOperatorAggregate 是用户最常显式创建的通道** - 通过 `add_messages` 等 reducer 函数 +3. **大部分通道都是系统内部自动管理** - 用户无需关心底层通道实现 +4. **StateGraph 和直接 Pregel 的使用场景不同** - StateGraph 专注状态管理,Pregel 专注消息传递 + +## 实际代码位置 + +- **通道创建逻辑:** `libs/langgraph/langgraph/graph/state.py:1299-1388` +- **编译时通道管理:** `libs/langgraph/langgraph/graph/state.py:856-894` +- **节点附加逻辑:** `libs/langgraph/langgraph/graph/state.py:935-1067` +- **边处理逻辑:** `libs/langgraph/langgraph/graph/state.py:1038-1062`