readme

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"

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`