feat: 新增子图执行 — Graph.subgraph_with_deps 计算传递闭包, run(only=/tags=) 按名称或标签运行图子集, yamlrun CLI 支持 --only/--tags; 修复 test_executors.py 中 TaskStatus 缺少 px. 前缀的预存问题

.trae/docs/iter-10-subgraph-execution.md

@@ -0,0 +1,55 @@
+# 迭代 10：子图执行
+
+## 本轮目标
+
+1. 支持按名称/标签运行图的子集，自动包含传递依赖
+2. `px.run(graph, only=[...], tags=[...])` API
+3. CLI `pf yamlrun pipeline.yaml --only task1,task2 --tags ingest`
+
+## 改动文件清单
+
+- `src/pyflowx/graph.py` — 新增 `subgraph_with_deps()` 方法（传递闭包）
+- `src/pyflowx/executors.py` — `run()` 新增 `only`/`tags` 参数
+- `src/pyflowx/cli/pf.py` — `yamlrun` 命令新增 `--only`/`--tags` 选项
+- `tests/test_graph.py` — 测试 `subgraph_with_deps`
+- `tests/test_executors.py` — 测试 `run(only=...)` / `run(tags=...)`
+- `README.md` — 更新文档
+
+## 关键设计
+
+### 1. Graph.subgraph_with_deps(names)
+
+计算传递闭包：从 `names` 出发，沿 `depends_on` + `soft_depends_on` 向上遍历，
+收集所有必须先完成的任务。
+
+### 2. run(only, tags)
+
+- `only`: 任务名列表，运行这些任务及其传递依赖
+- `tags`: 标签列表，运行匹配任意标签的任务及其传递依赖
+- 两者可组合：并集后计算传递闭包
+- 都为 None 时运行全图（现有行为不变）
+
+### 3. CLI
+
+`pf yamlrun pipeline.yaml --only task1,task2 --tags ingest,report`
+
+## 验收标准
+
+- subgraph_with_deps 正确计算传递闭包（含软依赖）
+- run(only=...) 只运行指定任务及其依赖
+- run(tags=...) 只运行匹配标签的任务及其依赖
+- CLI --only/--tags 选项可用
+- 覆盖率 ≥ 95%
+- ruff / pyrefly / pytest 全部通过
+
+## 验证结果
+
+- **ruff**：All checks passed
+- **pyrefly**：0 errors
+- **pytest**：1230 passed（含 14 个新增子图测试）
+- **覆盖率**：97.28%
+- **修复**：test_executors.py 中 TaskStatus 缺少 px. 前缀的预存问题
+
+## 遗留事项
+
+- 无

README.md

@@ -36,6 +36,7 @@ PyFlowX 把"任务依赖"这件事做到极致简单：**参数名就是依赖
 - **任务通知** —— `Notifier` 协议支持飞书/钉钉/企业微信 webhook 与 Python 回调，全生命周期事件可配置
 - **流式获取** —— `run_iter()` 生成器逐个 yield 任务结果，适用于大量任务逐个处理
 - **任务取消** —— `CancelToken` 线程安全取消令牌，支持外部取消与 `KeyboardInterrupt` 优雅停止
+- **子图执行** —— `run(only=[...], tags=[...])` 按名称或标签运行图的子集，自动包含传递依赖
 - **状态后端** —— `MemoryBackend` / `JSONBackend` / `SQLiteBackend`（WAL 模式，适合大规模任务）
 - **YAML 任务编排** —— GitHub Actions 风格 `jobs`/`needs`/`strategy.matrix`/`if` 条件，`pf yamlrun pipeline.yaml` 一键执行
 - **最小依赖** —— `rich` + `typer` + `typing-extensions`（3.13 以下）+ `pyyaml`
@@ -514,6 +515,28 @@ event = threading.Event()
 px.run(graph, cancel_event=event)
 ```
 
+### 子图执行
+
+`run(only=...)` / `run(tags=...)` 只运行指定任务及其传递依赖，适用于调试单个任务或增量执行：
+
+```python
+# 只运行 "report" 任务及其所有上游依赖
+report = px.run(graph, only=["report"])
+
+# 按标签过滤：只运行带 "ingest" 标签的任务及其依赖
+report = px.run(graph, tags=["ingest"])
+
+# 两者可组合（取并集）
+report = px.run(graph, only=["report"], tags=["ingest"])
+```
+
+CLI 也支持 `--only` / `--tags`：
+
+```bash
+pf yamlrun pipeline.yaml --only report,deploy
+pf yamlrun pipeline.yaml --tags ingest,report
+```
+
 ## 开发
 
 ```bash

src/pyflowx/cli/pf.py

@@ -276,6 +276,8 @@ class PfApp:
         _ = parser.add_argument("--dry-run", action="store_true", help="仅打印执行计划")
         _ = parser.add_argument("--list", action="store_true", help="列出所有任务名")
         _ = parser.add_argument("--quiet", action="store_true", help="静默模式")
+        _ = parser.add_argument("--only", help="只运行指定任务（逗号分隔）及其依赖")
+        _ = parser.add_argument("--tags", help="只运行匹配标签的任务（逗号分隔）及其依赖")
 
         parsed = parser.parse_args(argv)
 
@@ -294,8 +296,11 @@ class PfApp:
             self._console.print(graph.describe())
             return CliExitCode.SUCCESS.value
 
+        only = parsed.only.split(",") if parsed.only else None
+        tags = parsed.tags.split(",") if parsed.tags else None
+
         try:
-            report = run(graph, strategy=parsed.strategy, verbose=not parsed.quiet)
+            report = run(graph, strategy=parsed.strategy, verbose=not parsed.quiet, only=only, tags=tags)
             return CliExitCode.SUCCESS.value if report.success else CliExitCode.FAILURE.value
         except KeyboardInterrupt:
             print("\n操作已取消", file=sys.stderr)

src/pyflowx/executors.py

@@ -49,7 +49,7 @@ import logging
 import queue
 import threading
 import time
-from collections.abc import Awaitable, Callable, Iterator, Mapping
+from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping
 from datetime import datetime
 from typing import Any, Literal, cast
 
@@ -810,6 +810,29 @@ def _resolve_progress(
     return monitor, notifier_list, effective_callback
 
 
+def _apply_subgraph_filter(
+    graph: Graph,
+    only: Iterable[str] | None,
+    tags: Iterable[str] | None,
+) -> Graph:
+    """根据 ``only``/``tags`` 过滤图，返回包含传递依赖的子图。
+
+    ``only`` 与 ``tags`` 取并集：匹配任一条件的任务及其所有传递依赖
+    （硬依赖 + 软依赖）都会被包含在子图中，使子图可独立执行。
+    """
+    names: set[str] = set()
+    if only is not None:
+        names.update(only)
+    if tags is not None:
+        tag_set = set(tags)
+        for name, spec in graph.all_specs().items():
+            if tag_set & set(spec.tags):
+                names.add(name)
+    if not names:
+        return Graph(defaults=graph.defaults)
+    return graph.subgraph_with_deps(names)
+
+
 def _dispatch_strategy(
     strategy: Strategy,
     graph: Graph,
@@ -873,6 +896,8 @@ def run(
     progress: bool | ProgressCallback = False,
     notifiers: Notifier | list[Notifier] | None = None,
     cancel_event: threading.Event | CancelToken | None = None,
+    only: Iterable[str] | None = None,
+    tags: Iterable[str] | None = None,
     _report: RunReport | None = None,
 ) -> RunReport:
     """执行图并返回 :class:`RunReport`。
@@ -911,6 +936,10 @@ def run(
         :class:`threading.Event`，设置后待运行任务被标记为 SKIPPED，
         运行中任务等待完成后返回部分结果。``KeyboardInterrupt`` 也会
         触发相同的优雅取消流程。
+    only:
+        只运行指定任务名及其传递依赖。与 ``tags`` 取并集。
+    tags:
+        只运行匹配任意标签的任务及其传递依赖。与 ``only`` 取并集。
 
     抛出
     ----
@@ -924,6 +953,10 @@ def run(
         _print_dry_run(graph, layers)
         return RunReport(success=True)
 
+    # 子图过滤：only/tags 选择任务子集及其传递依赖
+    if only is not None or tags is not None:
+        graph = _apply_subgraph_filter(graph, only, tags)
+
     # verbose 模式下, 把所有 spec 的 verbose 标记设为 True,
     # 使 execute_command 打印执行命令与返回码 (任务生命周期由 callback 打印)
     if verbose:

src/pyflowx/graph.py

@@ -437,6 +437,38 @@ class Graph:
         ]
         return Graph.from_specs(kept, defaults=self.defaults)
 
+    def subgraph_with_deps(self, names: Iterable[str]) -> Graph:
+        """返回包含 ``names`` 及其所有传递依赖的新图。
+
+        与 :meth:`subgraph_by_names` 不同，本方法会沿 ``depends_on`` 和
+        ``soft_depends_on`` 向上遍历，确保被选中的任务所需的上游全部包含在内，
+        使子图可独立执行。
+
+        Parameters
+        ----------
+        names:
+            需要执行的任务名。每个名称必须存在于当前图中。
+        """
+        seeds: set[str] = set(names)
+        for n in seeds:
+            if n not in self.specs:
+                raise KeyError(f"Unknown task name: {n!r}")
+        # BFS 收集传递依赖（硬依赖 + 软依赖）
+        closure: set[str] = set()
+        queue: list[str] = list(seeds)
+        while queue:
+            name = queue.pop()
+            if name in closure:
+                continue
+            closure.add(name)
+            for dep in self.all_deps(name):
+                if dep not in closure:
+                    queue.append(dep)
+        kept: list[TaskSpec[Any]] = [
+            _prune_deps(spec, lambda d: d in closure) for spec in self.specs.values() if spec.name in closure
+        ]
+        return Graph.from_specs(kept, defaults=self.defaults)
+
     # ------------------------------------------------------------------ #
     # Fan-out / map-reduce
     # ------------------------------------------------------------------ #

tests/test_executors.py

@@ -650,3 +650,147 @@ def test_downstream_executes_when_upstream_succeeds() -> None:
     assert report.result_of("upstream").status == px.TaskStatus.SUCCESS
     assert report.result_of("downstream").status == px.TaskStatus.SUCCESS
     assert report["downstream"] == "hello_processed"
+
+
+# ---------------------------------------------------------------------- #
+# 子图执行：run(only=...) / run(tags=...)
+# ---------------------------------------------------------------------- #
+def test_run_only_executes_subset_with_deps() -> None:
+    """run(only=...) 只运行指定任务及其传递依赖。"""
+    executed: list[str] = []
+
+    def make(name: str) -> Any:
+        def fn() -> str:
+            executed.append(name)
+            return name
+
+        return fn
+
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", make("a")),
+            px.TaskSpec("b", make("b"), depends_on=("a",)),
+            px.TaskSpec("c", make("c"), depends_on=("b",)),
+            px.TaskSpec("d", make("d")),  # 不在 c 的依赖链上
+        ]
+    )
+    report = px.run(graph, strategy="sequential", only=["c"])
+    assert report.success
+    assert set(report.results.keys()) == {"a", "b", "c"}
+    assert "d" not in report.results
+    assert "d" not in executed
+
+
+def test_run_tags_executes_matching_with_deps() -> None:
+    """run(tags=...) 只运行匹配标签的任务及其传递依赖。"""
+    executed: list[str] = []
+
+    def make(name: str) -> Any:
+        def fn() -> str:
+            executed.append(name)
+            return name
+
+        return fn
+
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", make("a"), tags=("ingest",)),
+            px.TaskSpec("b", make("b"), depends_on=("a",), tags=("ingest",)),
+            px.TaskSpec("c", make("c"), depends_on=("b",), tags=("report",)),
+            px.TaskSpec("d", make("d"), tags=("other",)),
+        ]
+    )
+    report = px.run(graph, strategy="sequential", tags=["ingest"])
+    assert report.success
+    assert set(report.results.keys()) == {"a", "b"}
+    assert "c" not in report.results
+    assert "d" not in report.results
+
+
+def test_run_only_and_tags_union() -> None:
+    """run(only=..., tags=...) 取并集后包含传递依赖。"""
+    executed: list[str] = []
+
+    def make(name: str) -> Any:
+        def fn() -> str:
+            executed.append(name)
+            return name
+
+        return fn
+
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", make("a"), tags=("ingest",)),
+            px.TaskSpec("b", make("b"), depends_on=("a",), tags=("ingest",)),
+            px.TaskSpec("c", make("c"), depends_on=("b",), tags=("report",)),
+            px.TaskSpec("d", make("d"), tags=("other",)),
+        ]
+    )
+    # only=["c"] → a/b/c; tags=["other"] → d; 并集 → a/b/c/d
+    report = px.run(graph, strategy="sequential", only=["c"], tags=["other"])
+    assert report.success
+    assert set(report.results.keys()) == {"a", "b", "c", "d"}
+
+
+def test_run_only_unknown_raises() -> None:
+    """run(only=...) 传入未知任务名应抛出 KeyError。"""
+    graph = px.Graph.from_specs([px.TaskSpec("a", lambda: 1)])
+    with pytest.raises(KeyError, match="Unknown task name"):
+        _ = px.run(graph, strategy="sequential", only=["missing"])
+
+
+def test_run_only_empty_returns_empty_report() -> None:
+    """run(only=[]) 无匹配任务时应返回空报告。"""
+    graph = px.Graph.from_specs([px.TaskSpec("a", lambda: 1)])
+    report = px.run(graph, strategy="sequential", only=[])
+    assert report.success
+    assert len(report.results) == 0
+
+
+def test_run_only_with_dependency_strategy() -> None:
+    """run(only=...) 在 dependency 策略下也正确过滤。"""
+    executed: list[str] = []
+
+    def make(name: str) -> Any:
+        def fn() -> str:
+            executed.append(name)
+            return name
+
+        return fn
+
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", make("a")),
+            px.TaskSpec("b", make("b"), depends_on=("a",)),
+            px.TaskSpec("c", make("c"), depends_on=("b",)),
+        ]
+    )
+    report = px.run(graph, strategy="dependency", only=["b"])
+    assert report.success
+    assert set(report.results.keys()) == {"a", "b"}
+    assert "c" not in report.results
+
+
+def test_run_only_preserves_context_injection() -> None:
+    """子图执行时上下文注入仍应正常工作。"""
+
+    def a() -> int:
+        return 42
+
+    def b(a: int) -> int:
+        return a * 2
+
+    def c(b: int) -> int:
+        return b + 1
+
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", a),
+            px.TaskSpec("b", b, depends_on=("a",)),
+            px.TaskSpec("c", c, depends_on=("b",)),
+        ]
+    )
+    report = px.run(graph, strategy="sequential", only=["c"])
+    assert report["c"] == 85  # 42 * 2 + 1
+    assert report["b"] == 84
+    assert report["a"] == 42

tests/test_graph.py

@@ -158,6 +158,89 @@ def test_subgraph_by_names() -> None:
     assert sub.dependencies("b") == ("a",)
 
 
+def test_subgraph_with_deps_basic() -> None:
+    """subgraph_with_deps 应包含传递依赖：选 c 时 a/b 也应包含。"""
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", _fn),
+            px.TaskSpec("b", _fn, depends_on=("a",)),
+            px.TaskSpec("c", _fn, depends_on=("b",)),
+        ]
+    )
+    sub = graph.subgraph_with_deps(["c"])
+    assert set(sub.names) == {"a", "b", "c"}
+    assert sub.dependencies("b") == ("a",)
+    assert sub.dependencies("c") == ("b",)
+
+
+def test_subgraph_with_deps_soft_deps() -> None:
+    """subgraph_with_deps 应包含软依赖。"""
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", _fn),
+            px.TaskSpec("b", _fn),
+            px.TaskSpec("c", _fn, depends_on=("a",), soft_depends_on=("b",)),
+        ]
+    )
+    sub = graph.subgraph_with_deps(["c"])
+    assert set(sub.names) == {"a", "b", "c"}
+
+
+def test_subgraph_with_deps_diamond() -> None:
+    """菱形依赖：选 d 时 a/b/c 都应包含。"""
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", _fn),
+            px.TaskSpec("b", _fn, depends_on=("a",)),
+            px.TaskSpec("c", _fn, depends_on=("a",)),
+            px.TaskSpec("d", _fn, depends_on=("b", "c")),
+        ]
+    )
+    sub = graph.subgraph_with_deps(["d"])
+    assert set(sub.names) == {"a", "b", "c", "d"}
+
+
+def test_subgraph_with_deps_multiple_seeds() -> None:
+    """多个种子任务的传递闭包应取并集。"""
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", _fn),
+            px.TaskSpec("b", _fn, depends_on=("a",)),
+            px.TaskSpec("c", _fn, depends_on=("a",)),
+            px.TaskSpec("d", _fn, depends_on=("c",)),
+        ]
+    )
+    sub = graph.subgraph_with_deps(["b", "d"])
+    assert set(sub.names) == {"a", "b", "c", "d"}
+
+
+def test_subgraph_with_deps_unknown_name() -> None:
+    """未知任务名应抛出 KeyError。"""
+    graph = px.Graph.from_specs([px.TaskSpec("a", _fn)])
+    with pytest.raises(KeyError, match="Unknown task name"):
+        _ = graph.subgraph_with_deps(["missing"])
+
+
+def test_subgraph_with_deps_empty() -> None:
+    """空名称列表应返回空图。"""
+    graph = px.Graph.from_specs([px.TaskSpec("a", _fn)])
+    sub = graph.subgraph_with_deps([])
+    assert len(sub) == 0
+
+
+def test_subgraph_with_deps_preserves_metadata() -> None:
+    """子图应保留原任务的元数据。"""
+    graph = px.Graph.from_specs(
+        [
+            px.TaskSpec("a", _fn, tags=("x",), retry=px.RetryPolicy(max_attempts=3)),
+            px.TaskSpec("b", _fn, depends_on=("a",), tags=("y",)),
+        ]
+    )
+    sub = graph.subgraph_with_deps(["b"])
+    assert sub.spec("a").tags == ("x",)
+    assert sub.spec("a").retry.max_attempts == 3
+
+
 def test_subgraph_by_names_unknown() -> None:
     graph = px.Graph.from_specs([px.TaskSpec("a", _fn)])
     with pytest.raises(KeyError):