快速上手

核心思想：参数名即依赖。写一个普通函数，参数名匹配上游任务名，框架自动注入结果。

最小示例

import pyflowx as px

def extract() -> list[int]:
    return [1, 2, 3]

# 参数名 extract 自动匹配上游任务名 → 自动注入
def double(extract: list[int]) -> list[int]:
    return [x * 2 for x in extract]

graph = px.Graph.from_specs([
    px.TaskSpec("extract", extract),
    px.TaskSpec("double", double, ("extract",)),
])

report = px.run(graph, strategy="sequential")
print(report["double"])  # [2, 4, 6]

三种任务形态

  1. **函数任务**（fn）：普通 Python 函数，参数名驱动自动注入

  2. **命令任务**（cmd）：执行外部命令，支持 list[str] / str``（shell）/ ``Callable

  3. YAML 声明式：从 YAML 文件加载任务图

graph = px.Graph.from_specs([
    px.TaskSpec("list", cmd=["ls", "-la"]),
    px.TaskSpec("greet", fn=lambda: "hello"),
])

执行策略

PyFlowX 提供四种执行策略：

策略

并发模型

适用场景

sequential

串行

调试、CPU 密集

thread

线程池

I/O 密集同步

async

事件循环

I/O 密集异步

dependency

依赖驱动

最大化并行度（默认推荐）

 
report = px.run(graph, strategy="dependency")

结果访问

report["task_name"]              # 任务返回值
report.result_of("task_name")     # 完整 TaskResult
report.success                   # 整体是否成功
report.summary()                 # 统计字典
report.failed_tasks()            # 失败任务名列表

下一步