执行策略与 run()

run() 是执行入口，支持四种策略：

report = px.run(
    graph,
    strategy="async",          # sequential | thread | async | dependency
    max_workers=8,             # thread 策略的线程池大小
    concurrency_limits={"db": 2},  # 按 concurrency_key 限流
    dry_run=False,             # True = 仅打印计划
    verbose=True,              # True = 打印执行过程
    on_event=callback,         # 状态转换回调
    state=px.JSONBackend("state.json"),  # 断点续跑后端
    continue_on_error=False,   # True = 单任务失败不中断整体
)

策略对比

策略	并发模型	适用场景	同步任务	异步任务
`sequential`	串行	调试、CPU 密集	直接调用	事件循环
`thread`	线程池	I/O 密集同步	线程池	不支持
`async`	事件循环	I/O 密集异步	卸载到线程池	事件循环
`dependency`	依赖驱动	最大化并行度	卸载到线程池	事件循环

所有策略都遵循 RetryPolicy、timeout、上下文注入、状态后端、concurrency_limits，并发出 TaskEvent``（RUNNING/SUCCESS/FAILED/SKIPPED）。``dependency 策略无层屏障：任务在其所有硬依赖完成后立即启动。

上下文注入规则

按顺序求值：

标注为 ``Context`` 的参数 → 接收完整上游结果映射
名称匹配依赖 的参数 → 接收该依赖的结果（含软依赖，缺失时注入默认值）
``**kwargs`` 参数 → 接收所有依赖结果（dict）
``TaskSpec.args`` / ``kwargs`` → 为非依赖参数提供静态值

from typing import Any, Dict

def aggregate(ctx: px.Context) -> Dict[str, Any]:
    """ctx 包含所有 depends_on 任务的返回值。"""
    return dict(ctx)

def merge(fetch_a: str, fetch_b: str) -> str:
    """fetch_a / fetch_b 自动注入。"""
    return fetch_a + fetch_b

断点续跑

from pyflowx import JSONBackend

backend = JSONBackend("state.json", ttl=3600)
report = px.run(graph, strategy="sequential", state=backend)

run() 内部以 backend.batch() 包裹整个执行：所有 save 延迟到运行结束时统一落盘一次。

缓存键：默认存储键为任务名。配置 cache_key 函数后，键为 "name:cache_key_value"。

完整 API 说明详见 API 参考。