LangGraph结构与继承关系深度解析：类型系统与设计模式

classDiagram
    class Runnable {
        <<abstract>>
        +invoke(input, config) Any
        +ainvoke(input, config) Any
        +stream(input, config) Iterator
        +astream(input, config) AsyncIterator
        +batch(inputs, config) List
        +abatch(inputs, config) List
    }
    
    class RunnableSerializable {
        <<abstract>>
        +to_json() dict
        +from_json(data) Self
        +get_name() str
        +get_input_schema() Type
        +get_output_schema() Type
    }
    
    class CompiledGraph {
        +pregel: Pregel
        +builder: StateGraph
        +config_specs: List
        +invoke(input, config) Any
        +stream(input, config) Iterator
        +get_graph() Graph
    }
    
    class StateGraph {
        +nodes: Dict[str, StateNodeSpec]
        +edges: Set[tuple]
        +branches: Dict[str, Branch]
        +entry_point: str
        +finish_point: str
        +add_node(key, action) Self
        +add_edge(start, end) Self
        +add_conditional_edges(source, path, path_map) Self
        +compile(checkpointer, interrupt_before, interrupt_after) CompiledGraph
    }
    
    class Pregel {
        +nodes: Dict[str, PregelNode]
        +channels: Dict[str, BaseChannel]
        +input_channels: List[str]
        +output_channels: List[str]
        +stream_channels: List[str]
        +checkpointer: BaseCheckpointSaver
        +step_timeout: float
        +debug: bool
    }
    
    Runnable <|-- RunnableSerializable
    RunnableSerializable <|-- CompiledGraph
    CompiledGraph *-- StateGraph
    CompiledGraph *-- Pregel
    StateGraph ..> CompiledGraph : creates

classDiagram
    class BaseChannel {
        <<abstract>>
        +typ: Type
        +update(values) bool
        +get() Any
        +checkpoint() Any
        +from_checkpoint(checkpoint) Self
    }
    
    class LastValue {
        +typ: Type
        +default: Any
        -value: Any
        +update(values) bool
        +get() Any
    }
    
    class Topic {
        +typ: Type
        +accumulate: bool
        +unique: bool
        -values: List
        +update(values) bool
        +get() List
        +consume() List
    }
    
    class BinaryOperatorAggregate {
        +typ: Type
        +operator: Callable
        +default: Any
        -value: Any
        +update(values) bool
        +get() Any
    }
    
    class DictMergeChannel {
        +typ: Type
        +default: dict
        -value: dict
        +update(values) bool
        +get() dict
    }
    
    BaseChannel <|-- LastValue
    BaseChannel <|-- Topic
    BaseChannel <|-- BinaryOperatorAggregate
    BaseChannel <|-- DictMergeChannel
    
    note for BaseChannel "通道系统的抽象基类\n定义状态管理接口"
    note for LastValue "保存最新值的通道\n用于简单状态字段"
    note for Topic "消息队列通道\n支持累积和去重"
    note for BinaryOperatorAggregate "聚合通道\n支持自定义reducer函数"

classDiagram
    class BaseCheckpointSaver {
        <<abstract>>
        +serde: SerializerProtocol
        +get(config) CheckpointTuple
        +list(config, filter, before, limit) Iterator[CheckpointTuple]
        +put(config, checkpoint, metadata, new_versions) RunnableConfig
        +put_writes(config, writes, task_id) None
        +get_tuple(config) CheckpointTuple
    }
    
    class MemorySaver {
        -storage: Dict[str, Dict]
        -writes: Dict[str, List]
        +get(config) CheckpointTuple
        +put(config, checkpoint, metadata, new_versions) RunnableConfig
        +list(config, filter, before, limit) Iterator[CheckpointTuple]
    }
    
    class PostgresSaver {
        +conn: Connection
        +pipe: Pipeline
        +is_setup: bool
        +get(config) CheckpointTuple
        +put(config, checkpoint, metadata, new_versions) RunnableConfig
        +setup() None
        +list(config, filter, before, limit) Iterator[CheckpointTuple]
    }
    
    class SqliteSaver {
        +conn: Connection
        +is_setup: bool
        +get(config) CheckpointTuple
        +put(config, checkpoint, metadata, new_versions) RunnableConfig
        +setup() None
    }
    
    class SerializerProtocol {
        <<interface>>
        +dumps(obj) bytes
        +loads(data) Any
        +dumps_typed(obj) tuple[str, bytes]
        +loads_typed(type_name, data) Any
    }
    
    class JsonPlusSerializer {
        +protocol: int
        +fallback_to_json: bool
        +dumps(obj) bytes
        +loads(data) Any
        +dumps_typed(obj) tuple[str, bytes]
        +loads_typed(type_name, data) Any
    }
    
    BaseCheckpointSaver <|-- MemorySaver
    BaseCheckpointSaver <|-- PostgresSaver
    BaseCheckpointSaver <|-- SqliteSaver
    BaseCheckpointSaver *-- SerializerProtocol
    SerializerProtocol <|.. JsonPlusSerializer
    
    note for BaseCheckpointSaver "检查点保存器抽象基类\n定义持久化接口"
    note for MemorySaver "内存检查点保存器\n用于开发和测试"
    note for PostgresSaver "PostgreSQL检查点保存器\n生产环境推荐"

classDiagram
    class PregelNode {
        +channels: List[str]
        +triggers: List[str]
        +mapper: ChannelInvoke
        +writers: List[ChannelWrite]
        +bound: Runnable
        +metadata: dict
        +retry_policy: RetryPolicy
        +interrupt_before: bool
        +interrupt_after: bool
    }
    
    class StateNodeSpec {
        +action: Runnable
        +ends: Set[str]
        +metadata: dict
        +retry_policy: RetryPolicy
        +input: Type
        +output: Type
    }
    
    class ToolNode {
        +tools: Sequence[Union[BaseTool, Callable]]
        +name: str
        +tags: List[str]
        +handle_tool_errors: bool
        +__call__(state) dict
    }
    
    class ValidationNode {
        +schemas: Dict[str, Type]
        +strict: bool
        +__call__(state) dict
    }
    
    class FunctionNode {
        +func: Callable
        +name: str
        +__call__(state) Any
    }
    
    class LambdaNode {
        +func: Callable
        +__call__(state) Any
    }
    
    PregelNode *-- StateNodeSpec : compiles_from
    StateNodeSpec <|-- ToolNode
    StateNodeSpec <|-- ValidationNode
    StateNodeSpec <|-- FunctionNode
    StateNodeSpec <|-- LambdaNode
    
    note for PregelNode "编译后的执行节点\n包含完整的执行信息"
    note for StateNodeSpec "节点规格定义\n编译前的节点描述"
    note for ToolNode "工具执行节点\n支持并行工具调用"

# 文件：langgraph/pregel/types.py
from typing import TypedDict, Optional, Any, Dict, List, Union, Callable
from dataclasses import dataclass
from enum import Enum

class TaskStatus(Enum):
    """任务状态枚举"""
    PENDING = "pending"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

@dataclass
class PregelTask:
    """Pregel任务定义
    
    这是执行引擎中任务的核心数据结构，包含：
    1. 任务标识信息
    2. 执行配置
    3. 状态管理
    4. 错误处理
    5. 性能监控
    
    设计原则：
    - 不可变性：任务创建后核心属性不可变
    - 类型安全：使用强类型确保数据一致性
    - 可序列化：支持检查点保存和恢复
    - 可监控：包含丰富的元数据用于监控
    """
    
    # === 核心标识 ===
    id: str                          # 任务唯一标识符
    name: str                        # 任务名称（通常是节点名）
    path: tuple[str, ...]           # 任务路径（支持子图）
    
    # === 执行配置 ===
    node: "PregelNode"              # 关联的Pregel节点
    config: "RunnableConfig"        # 运行时配置
    triggers: List[str]             # 触发通道列表
    
    # === 状态管理 ===
    status: TaskStatus = TaskStatus.PENDING    # 任务状态
    retry_count: int = 0                       # 重试次数
    priority: int = 0                          # 任务优先级
    
    # === 时间信息 ===
    created_at: float = 0.0         # 创建时间戳
    started_at: Optional[float] = None        # 开始执行时间
    completed_at: Optional[float] = None      # 完成时间
    
    # === 错误处理 ===
    error: Optional[Exception] = None         # 执行错误
    error_message: Optional[str] = None       # 错误消息
    
    # === 性能监控 ===
    execution_time: float = 0.0     # 执行耗时
    memory_usage: int = 0           # 内存使用量（字节）
    
    # === 元数据 ===
    metadata: Dict[str, Any] = None # 自定义元数据
    
    def __post_init__(self):
        """初始化后处理"""
        if self.metadata is None:
            self.metadata = {}
        if self.created_at == 0.0:
            import time
            self.created_at = time.time()
    
    @property
    def is_completed(self) -> bool:
        """任务是否已完成"""
        return self.status in (TaskStatus.COMPLETED, TaskStatus.FAILED, TaskStatus.CANCELLED)
    
    @property
    def duration(self) -> Optional[float]:
        """任务执行时长"""
        if self.started_at and self.completed_at:
            return self.completed_at - self.started_at
        return None
    
    def mark_started(self) -> None:
        """标记任务开始执行"""
        import time
        self.started_at = time.time()
        self.status = TaskStatus.RUNNING
    
    def mark_completed(self, execution_time: float = 0.0) -> None:
        """标记任务完成"""
        import time
        self.completed_at = time.time()
        self.status = TaskStatus.COMPLETED
        self.execution_time = execution_time
    
    def mark_failed(self, error: Exception) -> None:
        """标记任务失败"""
        import time
        self.completed_at = time.time()
        self.status = TaskStatus.FAILED
        self.error = error
        self.error_message = str(error)

class PregelTaskError(Exception):
    """Pregel任务执行错误
    
    包含丰富的错误上下文信息，用于：
    1. 错误诊断和调试
    2. 错误恢复和重试
    3. 错误统计和监控
    4. 用户友好的错误报告
    """
    
    def __init__(
        self,
        task_name: str,
        error: Exception,
        retry_count: int = 0,
        task_id: Optional[str] = None,
        context: Optional[Dict[str, Any]] = None
    ):
        self.task_name = task_name
        self.original_error = error
        self.retry_count = retry_count
        self.task_id = task_id
        self.context = context or {}
        
        # 构建错误消息
        message = f"Task '{task_name}' failed"
        if retry_count > 0:
            message += f" after {retry_count} retries"
        message += f": {error}"
        
        super().__init__(message)
    
    @property
    def is_retryable(self) -> bool:
        """错误是否可重试"""
        # 某些错误类型不应该重试
        non_retryable_types = (
            ValueError,      # 参数错误
            TypeError,       # 类型错误
            AttributeError,  # 属性错误
            KeyError,        # 键错误（通常是配置问题）
        )
        
        return not isinstance(self.original_error, non_retryable_types)
    
    def to_dict(self) -> Dict[str, Any]:
        """转换为字典格式"""
        return {
            "task_name": self.task_name,
            "task_id": self.task_id,
            "error_type": type(self.original_error).__name__,
            "error_message": str(self.original_error),
            "retry_count": self.retry_count,
            "is_retryable": self.is_retryable,
            "context": self.context,
        }

classDiagram
    class Graph {
        <<abstract>>
        +nodes: Dict[str, Node]
        +edges: List[Edge]
        +add_node(key, node) Self
        +add_edge(start, end) Self
        +compile() CompiledGraph
    }
    
    class StateGraph {
        +state_schema: Type
        +nodes: Dict[str, StateNodeSpec]
        +edges: Set[tuple]
        +branches: Dict[str, Branch]
        +_schema_config: SchemaConfig
        +_channel_specs: Dict[str, ChannelSpec]
        +add_node(key, action) Self
        +add_edge(start, end) Self
        +add_conditional_edges(source, path, path_map) Self
    }
    
    class MessageGraph {
        +nodes: Dict[str, Node]
        +edges: Set[tuple]
        +entry_point: str
        +finish_point: str
        +add_node(key, node) Self
        +add_edge(start, end) Self
    }
    
    class CompiledStateGraph {
        +builder: StateGraph
        +pregel: Pregel
        +attach_node(key, node) PregelNode
        +get_input_schema() Type
        +get_output_schema() Type
    }
    
    class CompiledMessageGraph {
        +builder: MessageGraph
        +pregel: Pregel
    }
    
    Graph <|-- StateGraph
    Graph <|-- MessageGraph
    StateGraph ..> CompiledStateGraph : compiles_to
    MessageGraph ..> CompiledMessageGraph : compiles_to
    
    note for StateGraph "状态驱动的图构建器\n支持复杂状态管理"
    note for MessageGraph "消息驱动的图构建器\n简化的消息传递模型"

# 文件：langgraph/graph/state.py
from typing import TypedDict, Callable, Dict, Any, Union, Optional, Mapping
from dataclasses import dataclass

@dataclass
class Branch:
    """条件分支定义
    
    这是LangGraph条件路由的核心数据结构，支持：
    1. 动态路径选择
    2. 复杂条件逻辑
    3. 默认路径处理
    4. 路径映射管理
    
    设计特点：
    - 函数式设计：条件函数是纯函数
    - 类型安全：强类型的路径映射
    - 可扩展性：支持自定义条件逻辑
    - 性能优化：条件函数结果可缓存
    """
    
    # === 核心组件 ===
    condition: Callable[..., Union[str, List[str]]]  # 条件函数
    path_map: Dict[str, str]                         # 路径映射
    then: Optional[str] = None                       # 默认后续节点
    
    # === 配置选项 ===
    cache_condition_result: bool = False             # 是否缓存条件结果
    condition_timeout: Optional[float] = None        # 条件函数超时
    
    def __post_init__(self):
        """初始化后验证"""
        if not callable(self.condition):
            raise TypeError("condition must be callable")
        
        if not isinstance(self.path_map, dict):
            raise TypeError("path_map must be a dictionary")
        
        # 验证路径映射的值
        for key, value in self.path_map.items():
            if not isinstance(key, str) or not isinstance(value, str):
                raise TypeError("path_map keys and values must be strings")
    
    def evaluate_condition(
        self, 
        state: Any, 
        config: Optional[Dict[str, Any]] = None
    ) -> Union[str, List[str]]:
        """评估条件函数
        
        Args:
            state: 当前状态
            config: 运行时配置
            
        Returns:
            Union[str, List[str]]: 条件结果
            
        Raises:
            ConditionEvaluationError: 条件评估失败时
        """
        try:
            # 应用超时控制
            if self.condition_timeout:
                import signal
                
                def timeout_handler(signum, frame):
                    raise TimeoutError("Condition evaluation timeout")
                
                signal.signal(signal.SIGALRM, timeout_handler)
                signal.alarm(int(self.condition_timeout))
            
            try:
                # 执行条件函数
                if config:
                    # 尝试传递config参数
                    import inspect
                    sig = inspect.signature(self.condition)
                    if 'config' in sig.parameters:
                        result = self.condition(state, config=config)
                    else:
                        result = self.condition(state)
                else:
                    result = self.condition(state)
                
                return result
            
            finally:
                if self.condition_timeout:
                    signal.alarm(0)  # 取消超时
        
        except Exception as e:
            raise ConditionEvaluationError(
                f"Failed to evaluate condition: {e}"
            ) from e
    
    def resolve_next_nodes(
        self, 
        condition_result: Union[str, List[str]]
    ) -> List[str]:
        """解析下一个节点
        
        Args:
            condition_result: 条件函数结果
            
        Returns:
            List[str]: 下一个节点列表
        """
        if isinstance(condition_result, str):
            # 单个结果
            if condition_result in self.path_map:
                return [self.path_map[condition_result]]
            elif self.then:
                return [self.then]
            else:
                raise ValueError(f"No path found for condition result: {condition_result}")
        
        elif isinstance(condition_result, list):
            # 多个结果（并行执行）
            next_nodes = []
            for result in condition_result:
                if result in self.path_map:
                    next_nodes.append(self.path_map[result])
                elif self.then:
                    next_nodes.append(self.then)
                else:
                    raise ValueError(f"No path found for condition result: {result}")
            
            return list(set(next_nodes))  # 去重
        
        else:
            raise TypeError(f"Invalid condition result type: {type(condition_result)}")

class ConditionEvaluationError(Exception):
    """条件评估错误"""
    pass

@dataclass
class ChannelSpec:
    """通道规格定义
    
    定义状态字段如何映射到通道的规格，包括：
    1. 类型信息
    2. 默认值
    3. Reducer函数
    4. 验证规则
    """
    
    type: type                                       # 字段类型
    default: Any = None                             # 默认值
    reducer: Optional[Callable[[Any, Any], Any]] = None  # Reducer函数
    validator: Optional[Callable[[Any], bool]] = None    # 验证函数
    
    def __post_init__(self):
        """初始化后验证"""
        if self.reducer and not callable(self.reducer):
            raise TypeError("reducer must be callable")
        
        if self.validator and not callable(self.validator):
            raise TypeError("validator must be callable")

@dataclass 
class SchemaConfig:
    """模式配置
    
    控制状态模式的解析和处理行为：
    1. 类型推断策略
    2. 默认值处理
    3. Reducer自动选择
    4. 验证规则应用
    """
    
    # === 类型推断 ===
    auto_infer_reducers: bool = True            # 自动推断reducer
    strict_typing: bool = False                 # 严格类型检查
    
    # === 默认值处理 ===
    use_field_defaults: bool = True             # 使用字段默认值
    allow_none_values: bool = True              # 允许None值
    
    # === Reducer选择 ===
    default_list_reducer: str = "extend"        # 列表默认reducer
    default_dict_reducer: str = "update"        # 字典默认reducer
    default_set_reducer: str = "union"          # 集合默认reducer
    
    # === 验证配置 ===
    validate_on_update: bool = False            # 更新时验证
    validate_on_access: bool = False            # 访问时验证

classDiagram
    class Pregel {
        +nodes: Dict[str, PregelNode]
        +channels: Dict[str, BaseChannel]
        +input_channels: List[str]
        +output_channels: List[str]
        +stream_channels: List[str]
        +checkpointer: BaseCheckpointSaver
        +step_timeout: float
        +debug: bool
        +invoke(input, config) Any
        +stream(input, config) Iterator
        +transform(input, config) Iterator
    }
    
    class PregelLoop {
        +pregel: Pregel
        +checkpointer: BaseCheckpointSaver
        +step: int
        +stop_reason: StopReason
        +tick(tasks, config) Dict[str, Any]
        +prepare_next_tasks(checkpoint, channels, managed, config) List[PregelTask]
    }
    
    class ExecutionContext {
        +config: RunnableConfig
        +checkpoint: Checkpoint
        +channels: Dict[str, BaseChannel]
        +step: int
        +stop_reason: StopReason
        +exception: Exception
        +debug: bool
    }
    
    class TaskScheduler {
        +priority_queue: PriorityQueue
        +resource_manager: ResourceManager
        +plan_execution_step(context) List[PregelTask]
        +schedule_task(task) None
        +get_ready_tasks() List[PregelTask]
    }
    
    class StopReason {
        <<enumeration>>
        COMPLETED
        INTERRUPT_BEFORE
        INTERRUPT_AFTER
        ERROR
        LIMIT_REACHED
        USER_CANCELLED
    }
    
    Pregel *-- PregelLoop
    PregelLoop *-- ExecutionContext
    PregelLoop *-- TaskScheduler
    ExecutionContext *-- StopReason
    
    note for Pregel "BSP执行引擎核心\n管理节点和通道"
    note for PregelLoop "执行循环控制器\n实现BSP算法"
    note for ExecutionContext "执行上下文\n维护执行状态"

# 文件：langgraph/pregel/executor.py
from typing import List, Dict, Any, Optional, Set, Callable
from dataclasses import dataclass, field
from enum import Enum
import heapq
import threading
import time
from concurrent.futures import ThreadPoolExecutor, Future

class TaskPriority(Enum):
    """任务优先级枚举"""
    CRITICAL = 0    # 关键任务（系统任务）
    HIGH = 1        # 高优先级
    NORMAL = 2      # 普通优先级
    LOW = 3         # 低优先级
    BACKGROUND = 4  # 后台任务

@dataclass
class ResourceRequirement:
    """资源需求定义
    
    定义任务执行所需的资源，用于：
    1. 资源分配和调度
    2. 并发控制
    3. 性能优化
    4. 资源监控
    """
    
    # === CPU资源 ===
    cpu_cores: float = 1.0              # 需要的CPU核心数
    cpu_intensive: bool = False         # 是否CPU密集型
    
    # === 内存资源 ===
    memory_mb: int = 100               # 需要的内存（MB）
    memory_intensive: bool = False      # 是否内存密集型
    
    # === IO资源 ===
    io_intensive: bool = False         # 是否IO密集型
    network_access: bool = False       # 是否需要网络访问
    
    # === 特殊资源 ===
    gpu_required: bool = False         # 是否需要GPU
    exclusive_access: bool = False     # 是否需要独占访问
    
    # === 时间限制 ===
    max_execution_time: Optional[float] = None  # 最大执行时间
    
    def is_compatible_with(self, available: "AvailableResources") -> bool:
        """检查是否与可用资源兼容"""
        return (
            self.cpu_cores <= available.cpu_cores and
            self.memory_mb <= available.memory_mb and
            (not self.gpu_required or available.gpu_available) and
            (not self.network_access or available.network_available)
        )

@dataclass
class AvailableResources:
    """可用资源定义"""
    
    cpu_cores: float = 4.0             # 可用CPU核心数
    memory_mb: int = 4096              # 可用内存（MB）
    gpu_available: bool = False        # GPU是否可用
    network_available: bool = True     # 网络是否可用
    
    # === 动态资源 ===
    current_cpu_usage: float = 0.0     # 当前CPU使用率
    current_memory_usage: int = 0      # 当前内存使用量
    
    def allocate(self, requirement: ResourceRequirement) -> bool:
        """分配资源"""
        if not requirement.is_compatible_with(self):
            return False
        
        self.cpu_cores -= requirement.cpu_cores
        self.memory_mb -= requirement.memory_mb
        self.current_cpu_usage += requirement.cpu_cores
        self.current_memory_usage += requirement.memory_mb
        
        return True
    
    def release(self, requirement: ResourceRequirement) -> None:
        """释放资源"""
        self.cpu_cores += requirement.cpu_cores
        self.memory_mb += requirement.memory_mb
        self.current_cpu_usage -= requirement.cpu_cores
        self.current_memory_usage -= requirement.memory_mb

class TaskScheduler:
    """任务调度器
    
    负责任务的调度和执行管理，实现：
    1. 优先级调度
    2. 资源感知调度
    3. 负载均衡
    4. 死锁避免
    5. 性能优化
    
    调度算法：
    - 基于优先级的抢占式调度
    - 资源感知的智能分配
    - 工作窃取的负载均衡
    - 反压机制的流控
    """
    
    def __init__(
        self,
        max_workers: int = 4,
        resource_manager: Optional["ResourceManager"] = None,
        enable_work_stealing: bool = True
    ):
        self.max_workers = max_workers
        self.resource_manager = resource_manager or ResourceManager()
        self.enable_work_stealing = enable_work_stealing
        
        # === 调度队列 ===
        self._priority_queue: List[tuple] = []  # (priority, timestamp, task)
        self._ready_queue: List[PregelTask] = []
        self._running_tasks: Dict[str, PregelTask] = {}
        self._completed_tasks: Dict[str, Any] = {}
        
        # === 线程同步 ===
        self._queue_lock = threading.RLock()
        self._stats_lock = threading.Lock()
        
        # === 执行器 ===
        self._executor = ThreadPoolExecutor(
            max_workers=max_workers,
            thread_name_prefix="langgraph-task"
        )
        
        # === 统计信息 ===
        self._stats = SchedulerStats()
        
        # === 控制标志 ===
        self._shutdown = False
        self._paused = False
    
    def plan_execution_step(
        self, 
        context: "ExecutionContext"
    ) -> List[PregelTask]:
        """规划执行步骤
        
        这是调度器的核心方法，负责：
        1. 分析当前状态
        2. 确定可执行任务
        3. 应用调度策略
        4. 返回任务列表
        
        Args:
            context: 执行上下文
            
        Returns:
            List[PregelTask]: 计划执行的任务列表
        """
        planning_start = time.time()
        
        try:
            # === 第一阶段：任务发现 ===
            candidate_tasks = self._discover_candidate_tasks(context)
            
            if not candidate_tasks:
                return []
            
            # === 第二阶段：依赖分析 ===
            ready_tasks = self._analyze_task_dependencies(candidate_tasks, context)
            
            # === 第三阶段：资源检查 ===
            schedulable_tasks = self._check_resource_availability(ready_tasks)
            
            # === 第四阶段：优先级排序 ===
            prioritized_tasks = self._apply_priority_scheduling(schedulable_tasks)
            
            # === 第五阶段：并发控制 ===
            final_tasks = self._apply_concurrency_limits(prioritized_tasks, context)
            
            # === 统计记录 ===
            planning_duration = time.time() - planning_start
            
            with self._stats_lock:
                self._stats.record_planning_operation(
                    candidates_count=len(candidate_tasks),
                    ready_count=len(ready_tasks),
                    scheduled_count=len(final_tasks),
                    duration=planning_duration
                )
            
            return final_tasks
        
        except Exception as e:
            logger.error(f"Task planning failed: {e}")
            return []
    
    def _discover_candidate_tasks(
        self, 
        context: "ExecutionContext"
    ) -> List[PregelTask]:
        """发现候选任务
        
        Args:
            context: 执行上下文
            
        Returns:
            List[PregelTask]: 候选任务列表
        """
        candidate_tasks = []
        
        # 遍历所有节点，检查触发条件
        for node_name, node in context.pregel.nodes.items():
            if self._is_node_triggered(node, context):
                task = self._create_task_for_node(node_name, node, context)
                candidate_tasks.append(task)
        
        return candidate_tasks
    
    def _is_node_triggered(
        self, 
        node: "PregelNode", 
        context: "ExecutionContext"
    ) -> bool:
        """检查节点是否被触发
        
        Args:
            node: Pregel节点
            context: 执行上下文
            
        Returns:
            bool: 是否被触发
        """
        # 检查所有触发通道
        for trigger_channel in node.triggers:
            channel = context.channels.get(trigger_channel)
            if channel and channel.has_updates():
                return True
        
        return False
    
    def _create_task_for_node(
        self,
        node_name: str,
        node: "PregelNode",
        context: "ExecutionContext"
    ) -> PregelTask:
        """为节点创建任务
        
        Args:
            node_name: 节点名称
            node: Pregel节点
            context: 执行上下文
            
        Returns:
            PregelTask: 创建的任务
        """
        # 分析资源需求
        resource_req = self._analyze_resource_requirement(node)
        
        # 确定任务优先级
        priority = self._determine_task_priority(node, context)
        
        # 创建任务
        task = PregelTask(
            id=f"{node_name}_{context.step}_{int(time.time() * 1000)}",
            name=node_name,
            path=(node_name,),
            node=node,
            config=context.config,
            triggers=node.triggers,
            priority=priority.value,
            metadata={
                "resource_requirement": resource_req,
                "step": context.step,
                "created_by": "scheduler"
            }
        )
        
        return task
    
    def _analyze_resource_requirement(
        self, 
        node: "PregelNode"
    ) -> ResourceRequirement:
        """分析节点的资源需求
        
        Args:
            node: Pregel节点
            
        Returns:
            ResourceRequirement: 资源需求
        """
        # 从节点元数据获取资源需求
        metadata = node.metadata or {}
        
        return ResourceRequirement(
            cpu_cores=metadata.get("cpu_cores", 1.0),
            memory_mb=metadata.get("memory_mb", 100),
            cpu_intensive=metadata.get("cpu_intensive", False),
            memory_intensive=metadata.get("memory_intensive", False),
            io_intensive=metadata.get("io_intensive", False),
            network_access=metadata.get("network_access", False),
            gpu_required=metadata.get("gpu_required", False),
            max_execution_time=metadata.get("max_execution_time")
        )
    
    def _determine_task_priority(
        self,
        node: "PregelNode",
        context: "ExecutionContext"
    ) -> TaskPriority:
        """确定任务优先级
        
        Args:
            node: Pregel节点
            context: 执行上下文
            
        Returns:
            TaskPriority: 任务优先级
        """
        # 检查节点元数据中的优先级设置
        metadata = node.metadata or {}
        
        if "priority" in metadata:
            priority_str = metadata["priority"].upper()
            try:
                return TaskPriority[priority_str]
            except KeyError:
                pass
        
        # 根据节点类型推断优先级
        if node.interrupt_before or node.interrupt_after:
            return TaskPriority.HIGH
        
        # 默认普通优先级
        return TaskPriority.NORMAL

@dataclass
class SchedulerStats:
    """调度器统计信息"""
    
    # === 计数统计 ===
    total_tasks_scheduled: int = 0
    total_tasks_completed: int = 0
    total_tasks_failed: int = 0
    
    # === 时间统计 ===
    total_planning_time: float = 0.0
    total_execution_time: float = 0.0
    average_task_duration: float = 0.0
    
    # === 资源统计 ===
    peak_memory_usage: int = 0
    peak_cpu_usage: float = 0.0
    
    def record_planning_operation(
        self,
        candidates_count: int,
        ready_count: int,
        scheduled_count: int,
        duration: float
    ) -> None:
        """记录规划操作统计"""
        self.total_planning_time += duration
        self.total_tasks_scheduled += scheduled_count
    
    def record_task_completion(
        self,
        task_duration: float,
        success: bool
    ) -> None:
        """记录任务完成统计"""
        if success:
            self.total_tasks_completed += 1
        else:
            self.total_tasks_failed += 1
        
        self.total_execution_time += task_duration
        
        # 更新平均任务时长
        total_completed = self.total_tasks_completed + self.total_tasks_failed
        if total_completed > 0:
            self.average_task_duration = self.total_execution_time / total_completed

classDiagram
    class ReactAgent {
        +model: BaseChatModel
        +tools: List[BaseTool]
        +system_message: str
        +checkpointer: BaseCheckpointSaver
        +interrupt_before: List[str]
        +interrupt_after: List[str]
        +max_steps: int
    }
    
    class AgentState {
        +messages: List[BaseMessage]
        +remaining_steps: int
        +is_last_step: bool
    }
    
    class ToolNode {
        +tools: Sequence[Union[BaseTool, Callable]]
        +name: str
        +tags: List[str]
        +handle_tool_errors: bool
        +__call__(state: AgentState) dict
    }
    
    class ValidationNode {
        +schemas: Dict[str, Type]
        +strict: bool
        +__call__(state: AgentState) dict
    }
    
    class AgentExecutor {
        +agent: ReactAgent
        +tools: List[BaseTool]
        +max_iterations: int
        +early_stopping_method: str
        +handle_parsing_errors: bool
    }
    
    ReactAgent *-- AgentState
    ReactAgent *-- ToolNode
    ReactAgent *-- ValidationNode
    ReactAgent ..> AgentExecutor : creates
    
    note for ReactAgent "ReAct模式智能体\n推理和行动循环"
    note for AgentState "智能体状态\n包含消息和步骤信息"
    note for ToolNode "工具执行节点\n支持并行工具调用"

# 文件：langgraph/prebuilt/tool_node.py
from typing import Any, Callable, Dict, List, Optional, Sequence, Union, Type
from abc import ABC, abstractmethod
from dataclasses import dataclass
import asyncio
import inspect
from concurrent.futures import ThreadPoolExecutor, as_completed

class ToolInterface(ABC):
    """工具接口抽象基类
    
    定义所有工具必须实现的基本接口，确保：
    1. 统一的调用方式
    2. 一致的错误处理
    3. 标准的元数据格式
    4. 可扩展的功能
    """
    
    @property
    @abstractmethod
    def name(self) -> str:
        """工具名称"""
        pass
    
    @property
    @abstractmethod
    def description(self) -> str:
        """工具描述"""
        pass
    
    @abstractmethod
    def invoke(self, input: Any, config: Optional[Dict[str, Any]] = None) -> Any:
        """同步调用工具"""
        pass
    
    @abstractmethod
    async def ainvoke(self, input: Any, config: Optional[Dict[str, Any]] = None) -> Any:
        """异步调用工具"""
        pass
    
    @property
    def input_schema(self) -> Optional[Type]:
        """输入模式"""
        return None
    
    @property
    def output_schema(self) -> Optional[Type]:
        """输出模式"""
        return None

@dataclass
class ToolCall:
    """工具调用定义
    
    表示一个具体的工具调用请求，包含：
    1. 工具标识信息
    2. 调用参数
    3. 执行配置
    4. 结果处理
    """
    
    # === 核心信息 ===
    id: str                              # 调用ID
    name: str                           # 工具名称
    args: Dict[str, Any]                # 调用参数
    
    # === 执行配置 ===
    timeout: Optional[float] = None     # 超时时间
    retry_count: int = 0               # 重试次数
    async_execution: bool = False       # 是否异步执行
    
    # === 结果信息 ===
    result: Any = None                 # 执行结果
    error: Optional[Exception] = None   # 执行错误
    execution_time: float = 0.0        # 执行耗时
    
    # === 元数据 ===
    metadata: Dict[str, Any] = None    # 自定义元数据
    
    def __post_init__(self):
        if self.metadata is None:
            self.metadata = {}
    
    @property
    def is_completed(self) -> bool:
        """是否已完成"""
        return self.result is not None or self.error is not None
    
    @property
    def is_successful(self) -> bool:
        """是否成功"""
        return self.result is not None and self.error is None

class ToolRegistry:
    """工具注册表
    
    管理所有可用工具的注册表，提供：
    1. 工具注册和发现
    2. 工具验证和检查
    3. 工具元数据管理
    4. 工具生命周期管理
    """
    
    def __init__(self):
        self._tools: Dict[str, ToolInterface] = {}
        self._tool_metadata: Dict[str, Dict[str, Any]] = {}
        self._tool_groups: Dict[str, List[str]] = {}
    
    def register_tool(
        self,
        tool: ToolInterface,
        group: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None
    ) -> None:
        """注册工具
        
        Args:
            tool: 工具实例
            group: 工具组名
            metadata: 工具元数据
        """
        if not isinstance(tool, ToolInterface):
            raise TypeError("Tool must implement ToolInterface")
        
        tool_name = tool.name
        
        if tool_name in self._tools:
            raise ValueError(f"Tool '{tool_name}' already registered")
        
        # 注册工具
        self._tools[tool_name] = tool
        self._tool_metadata[tool_name] = metadata or {}
        
        # 添加到组
        if group:
            if group not in self._tool_groups:
                self._tool_groups[group] = []
            self._tool_groups[group].append(tool_name)
    
    def get_tool(self, name: str) -> Optional[ToolInterface]:
        """获取工具"""
        return self._tools.get(name)
    
    def list_tools(self, group: Optional[str] = None) -> List[str]:
        """列出工具"""
        if group:
            return self._tool_groups.get(group, [])
        return list(self._tools.keys())
    
    def get_tools_by_group(self, group: str) -> List[ToolInterface]:
        """按组获取工具"""
        tool_names = self._tool_groups.get(group, [])
        return [self._tools[name] for name in tool_names]

class ToolExecutor:
    """工具执行器
    
    负责工具的实际执行，提供：
    1. 同步和异步执行
    2. 并发控制
    3. 错误处理和重试
    4. 性能监控
    5. 资源管理
    """
    
    def __init__(
        self,
        registry: ToolRegistry,
        max_workers: int = 4,
        default_timeout: float = 30.0,
        enable_parallel_execution: bool = True
    ):
        self.registry = registry
        self.max_workers = max_workers
        self.default_timeout = default_timeout
        self.enable_parallel_execution = enable_parallel_execution
        
        # 执行器
        self._thread_executor = ThreadPoolExecutor(max_workers=max_workers)
        
        # 统计信息
        self._execution_stats = ExecutionStats()
    
    def execute_tool_calls(
        self,
        tool_calls: List[ToolCall],
        config: Optional[Dict[str, Any]] = None
    ) -> List[ToolCall]:
        """执行工具调用列表
        
        Args:
            tool_calls: 工具调用列表
            config: 执行配置
            
        Returns:
            List[ToolCall]: 执行结果
        """
        if not tool_calls:
            return []
        
        if len(tool_calls) == 1:
            # 单个工具调用
            return [self._execute_single_tool_call(tool_calls[0], config)]
        
        elif self.enable_parallel_execution:
            # 并行执行
            return self._execute_parallel_tool_calls(tool_calls, config)
        
        else:
            # 串行执行
            return [self._execute_single_tool_call(call, config) for call in tool_calls]
    
    def _execute_single_tool_call(
        self,
        tool_call: ToolCall,
        config: Optional[Dict[str, Any]] = None
    ) -> ToolCall:
        """执行单个工具调用
        
        Args:
            tool_call: 工具调用
            config: 执行配置
            
        Returns:
            ToolCall: 更新后的工具调用
        """
        start_time = time.time()
        
        try:
            # 获取工具
            tool = self.registry.get_tool(tool_call.name)
            if not tool:
                raise ValueError(f"Tool '{tool_call.name}' not found")
            
            # 准备参数
            invoke_args = self._prepare_invoke_args(tool, tool_call.args, config)
            
            # 执行工具
            timeout = tool_call.timeout or self.default_timeout
            
            if timeout:
                # 带超时执行
                future = self._thread_executor.submit(tool.invoke, invoke_args, config)
                result = future.result(timeout=timeout)
            else:
                # 普通执行
                result = tool.invoke(invoke_args, config)
            
            # 更新结果
            tool_call.result = result
            tool_call.execution_time = time.time() - start_time
            
            # 记录统计
            self._execution_stats.record_success(
                tool_call.name, tool_call.execution_time
            )
            
        except Exception as e:
            # 错误处理
            tool_call.error = e
            tool_call.execution_time = time.time() - start_time
            
            # 记录统计
            self._execution_stats.record_error(
                tool_call.name, tool_call.execution_time, str(e)
            )
        
        return tool_call
    
    def _execute_parallel_tool_calls(
        self,
        tool_calls: List[ToolCall],
        config: Optional[Dict[str, Any]] = None
    ) -> List[ToolCall]:
        """并行执行工具调用
        
        Args:
            tool_calls: 工具调用列表
            config: 执行配置
            
        Returns:
            List[ToolCall]: 执行结果列表
        """
        # 提交所有任务
        future_to_call = {
            self._thread_executor.submit(
                self._execute_single_tool_call, call, config
            ): call
            for call in tool_calls
        }
        
        # 收集结果
        results = []
        for future in as_completed(future_to_call):
            try:
                result_call = future.result()
                results.append(result_call)
            except Exception as e:
                # 这种情况不应该发生，因为异常已在单个执行中处理
                original_call = future_to_call[future]
                original_call.error = e
                results.append(original_call)
        
        # 按原始顺序排序
        call_id_to_result = {call.id: call for call in results}
        ordered_results = [call_id_to_result[call.id] for call in tool_calls]
        
        return ordered_results
    
    def _prepare_invoke_args(
        self,
        tool: ToolInterface,
        args: Dict[str, Any],
        config: Optional[Dict[str, Any]] = None
    ) -> Any:
        """准备调用参数
        
        Args:
            tool: 工具实例
            args: 原始参数
            config: 执行配置
            
        Returns:
            Any: 准备好的参数
        """
        # 检查工具的输入模式
        if tool.input_schema:
            try:
                # 使用Pydantic验证参数
                validated_args = tool.input_schema(**args)
                return validated_args
            except Exception as e:
                raise ValueError(f"Invalid arguments for tool '{tool.name}': {e}")
        
        # 检查函数签名
        if hasattr(tool, 'invoke'):
            sig = inspect.signature(tool.invoke)
            params = list(sig.parameters.keys())
            
            if len(params) == 1:
                # 单参数，直接传递args
                return args
            elif len(params) == 2 and 'config' in params:
                # 两参数，包含config
                return args
            else:
                # 多参数，展开args
                return args
        
        return args

@dataclass
class ExecutionStats:
    """执行统计信息"""
    
    # === 计数统计 ===
    total_executions: int = 0
    successful_executions: int = 0
    failed_executions: int = 0
    
    # === 时间统计 ===
    total_execution_time: float = 0.0
    average_execution_time: float = 0.0
    min_execution_time: float = float('inf')
    max_execution_time: float = 0.0
    
    # === 工具统计 ===
    tool_usage_count: Dict[str, int] = None
    tool_error_count: Dict[str, int] = None
    
    def __post_init__(self):
        if self.tool_usage_count is None:
            self.tool_usage_count = {}
        if self.tool_error_count is None:
            self.tool_error_count = {}
    
    def record_success(self, tool_name: str, execution_time: float) -> None:
        """记录成功执行"""
        self.total_executions += 1
        self.successful_executions += 1
        self.total_execution_time += execution_time
        
        # 更新时间统计
        self.average_execution_time = self.total_execution_time / self.total_executions
        self.min_execution_time = min(self.min_execution_time, execution_time)
        self.max_execution_time = max(self.max_execution_time, execution_time)
        
        # 更新工具统计
        self.tool_usage_count[tool_name] = self.tool_usage_count.get(tool_name, 0) + 1
    
    def record_error(self, tool_name: str, execution_time: float, error: str) -> None:
        """记录执行错误"""
        self.total_executions += 1
        self.failed_executions += 1
        self.total_execution_time += execution_time
        
        # 更新时间统计
        self.average_execution_time = self.total_execution_time / self.total_executions
        
        # 更新错误统计
        self.tool_error_count[tool_name] = self.tool_error_count.get(tool_name, 0) + 1