diff --git a/.trae/docs/iter-12-cli-visualization.md b/.trae/docs/iter-12-cli-visualization.md
new file mode 100644
index 0000000..9d4aa09
--- /dev/null
+++ b/.trae/docs/iter-12-cli-visualization.md
@@ -0,0 +1,65 @@
+# 迭代 12：CLI 可视化
+
+## 本轮目标
+
+1. `pf graph <yaml>` 命令 —— 终端 DAG 可视化
+   - `--format ascii`（默认）：rich Tree 渲染分层结构 + 依赖关系表
+   - `--format mermaid`：输出 Mermaid 语法（可粘贴到 mermaid.live）
+   - `--format list`：仅列出任务名（与 `yamlrun --list` 一致）
+   - `--format deps`：纯依赖关系表
+   - `--orientation`：Mermaid 方向（TD/TB/BT/LR/RL）
+2. `pf yamlrun --progress` —— 启用 rich 进度条（`RichProgressMonitor`）
+
+## 改动文件清单
+
+- `src/pyflowx/cli/pf.py` —— 新增 `pf graph` 命令、`yamlrun --progress` 选项
+- `tests/cli/test_pf_graph.py` —— 新增 CLI 测试
+- `README.md` —— 更新文档
+
+## 关键设计
+
+### 1. `pf graph <yaml>`
+
+**ASCII 默认渲染**：用 `rich.tree.Tree` 按层分组，每层显示任务名。
+再附 `rich.table.Table` 显示每个任务的硬依赖/软依赖/标签。
+
+**Mermaid 输出**：复用 `Graph.to_mermaid()`，直接打印（不加颜色，便于复制）。
+
+**deps 表格**：纯依赖关系表（无分层信息）。
+
+**list**：与 `yamlrun --list` 行为一致。
+
+### 2. `pf yamlrun --progress`
+
+- 添加 `--progress` flag
+- 启用时调用 `run(graph, progress=True)`
+- 与 `--quiet` 互斥（quiet 时无进度条）
+
+### 3. 路由
+
+`pf graph` 作为新的子命令（与 `yamlrun` 同级），不通过 `@px.tool` 注册。
+
+## 验收标准
+
+- `pf graph pipeline.yaml` 默认 ASCII 渲染
+- `pf graph pipeline.yaml --format mermaid` 输出 Mermaid 文本
+- `pf graph pipeline.yaml --format list` 列出任务名
+- `pf graph pipeline.yaml --format deps` 显示依赖表
+- `pf graph pipeline.yaml --orientation LR` 自定义方向
+- `pf yamlrun pipeline.yaml --progress` 启用进度条
+- YAML 加载失败时显示错误信息并返回非零退出码
+- 覆盖率 ≥ 95%
+- ruff / pyrefly / pytest 全部通过
+
+## 验证结果
+
+- **ruff**：All checks passed（含 format）
+- **pyrefly**：0 errors
+- **pytest**：1271 passed（含 14 个新增 CLI 测试）
+- **覆盖率**：97.29%（yaml_loader.py 100%；cli/pf.py 按项目惯例 omit 不计入覆盖率）
+- **冒烟测试**：`pf graph` 三种格式（ASCII/Mermaid/list）+ `pf yamlrun --progress` 均通过
+- **附加修复**：`yaml_loader._safe_load` 包装 `yaml.YAMLError` 为 `ValueError`，统一错误处理
+
+## 遗留事项
+
+- 无
diff --git a/README.md b/README.md
index 51d7639..4069639 100644
--- a/README.md
+++ b/README.md
@@ -38,6 +38,7 @@ PyFlowX 把"任务依赖"这件事做到极致简单：**参数名就是依赖
 - **任务取消** —— `CancelToken` 线程安全取消令牌，支持外部取消与 `KeyboardInterrupt` 优雅停止
 - **子图执行** —— `run(only=[...], tags=[...])` 按名称或标签运行图的子集，自动包含传递依赖
 - **结果序列化** —— `RunReport.to_json()` / `to_dict()` / `to_csv()` 多格式导出运行结果，`from_json()` 支持反序列化重建
+- **CLI 可视化** —— `pf graph <yaml>` 终端 DAG 渲染（ASCII 分层树 / Mermaid / 依赖表），`pf yamlrun --progress` rich 进度条
 - **状态后端** —— `MemoryBackend` / `JSONBackend` / `SQLiteBackend`（WAL 模式，适合大规模任务）
 - **YAML 任务编排** —— GitHub Actions 风格 `jobs`/`needs`/`strategy.matrix`/`if` 条件，`pf yamlrun pipeline.yaml` 一键执行
 - **最小依赖** —— `rich` + `typer` + `typing-extensions`（3.13 以下）+ `pyyaml`
@@ -573,6 +574,34 @@ print(restored["double"])  # 任务返回值
 print(restored.result_of("a").status)  # TaskStatus
 ```
 
+### CLI 可视化
+
+`pf graph <yaml>` 在终端可视化任务图 DAG，无需运行即可检查依赖关系：
+
+```bash
+# 默认 ASCII 渲染：分层树 + 依赖关系表
+pf graph pipeline.yaml
+
+# 输出 Mermaid 语法（可粘贴到 mermaid.live）
+pf graph pipeline.yaml --format mermaid
+
+# 自定义方向
+pf graph pipeline.yaml --format mermaid --orientation LR
+
+# 仅列出任务名
+pf graph pipeline.yaml --format list
+
+# 纯依赖关系表
+pf graph pipeline.yaml --format deps
+```
+
+`pf yamlrun` 新增 `--progress` 选项，启用 rich 进度条显示运行中任务、完成
+数与耗时：
+
+```bash
+pf yamlrun pipeline.yaml --progress
+```
+
 ## 开发
 
 ```bash
diff --git a/src/pyflowx/cli/pf.py b/src/pyflowx/cli/pf.py
index be76dac..cb12845 100644
--- a/src/pyflowx/cli/pf.py
+++ b/src/pyflowx/cli/pf.py
@@ -18,11 +18,13 @@ import difflib
 import importlib
 import sys
 from collections.abc import Sequence
+from typing import Any
 
 from rich.console import Console
 from rich.panel import Panel
 from rich.table import Table
 from rich.text import Text
+from rich.tree import Tree
 
 from pyflowx import __version__
 from pyflowx.tools import _TOOL_REGISTRY, run_tool
@@ -149,8 +151,11 @@ class PfApp:
         if first in ("--version", "-V"):
             self._console.print(f"PyFlowX [bold cyan]{__version__}[/bold cyan]")
             return 0
-        if first == "yamlrun":
-            return self._run_yaml(self._argv[1:])
+
+        # 内建子命令（与 @px.tool 注册的 ops 工具同级）
+        builtin = {"yamlrun": self._run_yaml, "graph": self._run_graph}
+        if first in builtin:
+            return builtin[first](self._argv[1:])
 
         rest = self._argv[1:]
         resolved = self._resolve_tool(first)
@@ -276,6 +281,7 @@ 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("--progress", action="store_true", help="显示 rich 进度条")
         _ = parser.add_argument("--only", help="只运行指定任务（逗号分隔）及其依赖")
         _ = parser.add_argument("--tags", help="只运行匹配标签的任务（逗号分隔）及其依赖")
 
@@ -298,9 +304,17 @@ class PfApp:
 
         only = parsed.only.split(",") if parsed.only else None
         tags = parsed.tags.split(",") if parsed.tags else None
+        progress = parsed.progress and not parsed.quiet
 
         try:
-            report = run(graph, strategy=parsed.strategy, verbose=not parsed.quiet, only=only, tags=tags)
+            report = run(
+                graph,
+                strategy=parsed.strategy,
+                verbose=not parsed.quiet,
+                only=only,
+                tags=tags,
+                progress=progress,
+            )
             return CliExitCode.SUCCESS.value if report.success else CliExitCode.FAILURE.value
         except KeyboardInterrupt:
             print("\n操作已取消", file=sys.stderr)
@@ -309,6 +323,85 @@ class PfApp:
             self._err.print(f"[red]错误:[/red] {e}")
             return CliExitCode.FAILURE.value
 
+    def _run_graph(self, argv: list[str]) -> int:
+        """执行 graph 命令: 终端可视化 YAML 任务图 DAG."""
+        import argparse
+
+        from pyflowx import Graph
+        from pyflowx.errors import PyFlowXError
+        from pyflowx.runner import CliExitCode
+
+        parser = argparse.ArgumentParser(prog="pf graph", description="可视化 YAML 任务图 DAG")
+        _ = parser.add_argument("file", help="YAML 文件路径")
+        _ = parser.add_argument(
+            "--format",
+            choices=["ascii", "mermaid", "list", "deps"],
+            default="ascii",
+            help="输出格式 (默认: %(default)s)",
+        )
+        _ = parser.add_argument(
+            "--orientation",
+            default="TD",
+            help="Mermaid 方向: TD/TB/BT/LR/RL (默认: %(default)s)",
+        )
+
+        parsed = parser.parse_args(argv)
+
+        try:
+            graph = Graph.from_yaml(parsed.file)
+        except (OSError, ValueError, PyFlowXError) as e:
+            self._err.print(f"[red]错误:[/red] 加载 YAML 失败: {e}")
+            return CliExitCode.FAILURE.value
+
+        fmt = parsed.format
+        try:
+            if fmt == "mermaid":
+                # Mermaid 输出不加颜色，便于复制粘贴到 mermaid.live
+                self._console.print(graph.to_mermaid(parsed.orientation), markup=False)
+            elif fmt == "list":
+                for name in graph.names:
+                    self._console.print(name)
+            elif fmt == "deps":
+                self._render_deps_table(graph)
+            else:
+                self._render_ascii(graph)
+        except ValueError as e:
+            self._err.print(f"[red]错误:[/red] {e}")
+            return CliExitCode.FAILURE.value
+        return CliExitCode.SUCCESS.value
+
+    def _render_ascii(self, graph: Any) -> None:
+        """用 rich 渲染分层 DAG 视图与依赖关系表."""
+
+        layers = graph.layers()
+        if not layers:
+            self._console.print("[dim]空图（无任务）[/dim]")
+            return
+
+        tree = Tree(f"[bold cyan]Graph[/bold cyan] ({len(graph)} 任务, {len(layers)} 层)")
+        for i, layer in enumerate(layers, 1):
+            names = ", ".join(f"[cyan]{n}[/cyan]" for n in layer)
+            tree.add(f"[bold]Layer {i}[/bold] ({len(layer)}): {names}")
+        self._console.print(tree)
+        self._console.print()
+        self._render_deps_table(graph)
+
+    def _render_deps_table(self, graph: Any) -> None:
+        """渲染任务依赖关系表."""
+        table = Table(title="任务依赖关系", show_header=True, header_style="bold")
+        table.add_column("任务", style="cyan", no_wrap=True)
+        table.add_column("硬依赖", style="yellow")
+        table.add_column("软依赖", style="dim")
+        table.add_column("标签", style="green")
+
+        for name in graph.names:
+            spec = graph.spec(name)
+            hard = ", ".join(spec.depends_on) if spec.depends_on else "-"
+            soft = ", ".join(spec.soft_depends_on) if spec.soft_depends_on else "-"
+            tags = ", ".join(spec.tags) if spec.tags else "-"
+            table.add_row(name, hard, soft, tags)
+        self._console.print(table)
+
     def _run_tool(self, target: str, argv: list[str]) -> int:
         """运行 @px.tool 工具.
 
diff --git a/src/pyflowx/yaml_loader.py b/src/pyflowx/yaml_loader.py
index 777aede..d54fe89 100644
--- a/src/pyflowx/yaml_loader.py
+++ b/src/pyflowx/yaml_loader.py
@@ -61,6 +61,17 @@ __all__ = ["load_yaml", "parse_yaml_string"]
 _MATRIX_PATTERN = re.compile(r"\$\{\{\s*matrix\.(\w+)\s*\}\}")
 
 
+def _safe_load(text: str) -> Any:
+    """调用 ``yaml.safe_load`` 并将 YAMLError 包装为 ValueError。
+
+    便于调用方统一捕获 ``ValueError`` 处理所有 YAML 解析失败场景。
+    """
+    try:
+        return yaml.safe_load(text)
+    except yaml.YAMLError as exc:
+        raise ValueError(f"YAML 解析失败: {exc}") from exc
+
+
 def parse_yaml_string(text: str) -> Graph:
     """从 YAML 字符串解析任务图。
 
@@ -79,7 +90,7 @@ def parse_yaml_string(text: str) -> Graph:
     ValueError
         YAML 结构不符合 schema 时。
     """
-    data = yaml.safe_load(text)
+    data = _safe_load(text)
     return _build_graph(data)
 
 
@@ -96,7 +107,7 @@ def load_yaml(path: str | Path) -> Graph:
     Graph
         解析后的任务图。
     """
-    data = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
+    data = _safe_load(Path(path).read_text(encoding="utf-8"))
     return _build_graph(data)
 
 
diff --git a/tests/cli/test_pf_graph.py b/tests/cli/test_pf_graph.py
new file mode 100644
index 0000000..b62959b
--- /dev/null
+++ b/tests/cli/test_pf_graph.py
@@ -0,0 +1,282 @@
+"""Tests for ``pf graph`` and ``pf yamlrun --progress`` CLI commands."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from pyflowx.cli.pf import PfApp
+
+
+def _write_yaml(tmp_path: Path, content: str) -> Path:
+    """写入 YAML 文件并返回路径。"""
+    yaml_file = tmp_path / "pipeline.yaml"
+    yaml_file.write_text(content, encoding="utf-8")
+    return yaml_file
+
+
+_SIMPLE_YAML = """
+jobs:
+  extract:
+    cmd: ["echo", "extract"]
+    tags: ["ingest"]
+  transform:
+    cmd: ["echo", "transform"]
+    needs: ["extract"]
+    tags: ["process"]
+  report:
+    cmd: ["echo", "report"]
+    needs: ["transform"]
+    tags: ["report"]
+"""
+
+_SOFT_DEP_YAML = """
+jobs:
+  a:
+    cmd: ["echo", "a"]
+  b:
+    cmd: ["echo", "b"]
+  c:
+    cmd: ["echo", "c"]
+    needs: ["a"]
+    soft-needs: ["b"]
+"""
+
+
+class TestPfGraphCommand:
+    """``pf graph <yaml>`` 命令测试。"""
+
+    def test_graph_ascii_default(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """默认 ASCII 渲染应输出分层视图与依赖表。"""
+        yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
+        app = PfApp(["graph", str(yaml_file)])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "Graph" in out
+        assert "Layer 1" in out
+        assert "Layer 2" in out
+        assert "extract" in out
+        assert "transform" in out
+        assert "report" in out
+        # 依赖关系表也应被渲染
+        assert "硬依赖" in out
+        assert "标签" in out
+
+    def test_graph_mermaid_format(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """--format mermaid 应输出 Mermaid 语法。"""
+        yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
+        app = PfApp(["graph", str(yaml_file), "--format", "mermaid"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "graph TD" in out
+        assert 'extract["extract"]' in out
+        assert "extract --> transform" in out
+        assert "transform --> report" in out
+
+    def test_graph_mermaid_orientation_lr(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """--orientation LR 应改变 Mermaid 方向。"""
+        yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
+        app = PfApp(["graph", str(yaml_file), "--format", "mermaid", "--orientation", "LR"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "graph LR" in out
+
+    def test_graph_mermaid_invalid_orientation(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """非法 orientation 应被 to_mermaid 拒绝（ValueError），CLI 返回非零。"""
+        yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
+        app = PfApp(["graph", str(yaml_file), "--format", "mermaid", "--orientation", "XX"])
+        rc = app.run()
+        assert rc != 0
+        err = capsys.readouterr().err
+        assert "Invalid orientation" in err or "Traceback" in err
+
+    def test_graph_list_format(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """--format list 应仅列出任务名。"""
+        yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
+        app = PfApp(["graph", str(yaml_file), "--format", "list"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        # 任务名按声明顺序输出
+        lines = [line.strip() for line in out.splitlines() if line.strip()]
+        assert "extract" in lines
+        assert "transform" in lines
+        assert "report" in lines
+        # 不应包含 ASCII 装饰
+        assert "Layer" not in out
+        assert "Graph" not in out
+
+    def test_graph_deps_format(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """--format deps 应输出依赖关系表。"""
+        yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
+        app = PfApp(["graph", str(yaml_file), "--format", "deps"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        assert "任务依赖关系" in out
+        assert "硬依赖" in out
+        assert "软依赖" in out
+        assert "标签" in out
+        assert "extract" in out
+        assert "ingest" in out
+
+    def test_graph_soft_deps_displayed(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """软依赖应在依赖表中显示。"""
+        yaml_file = _write_yaml(tmp_path, _SOFT_DEP_YAML)
+        app = PfApp(["graph", str(yaml_file), "--format", "deps"])
+        rc = app.run()
+        assert rc == 0
+        out = capsys.readouterr().out
+        # c 的软依赖 b 应被显示
+        assert "b" in out
+
+    def test_graph_file_not_found(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """不存在的 YAML 文件应返回非零退出码与错误信息。"""
+        missing = tmp_path / "nonexistent.yaml"
+        app = PfApp(["graph", str(missing)])
+        rc = app.run()
+        assert rc != 0
+        err = capsys.readouterr().err
+        assert "加载 YAML 失败" in err
+
+    def test_graph_invalid_yaml(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """无效 YAML 应返回非零退出码与错误信息。"""
+        yaml_file = _write_yaml(tmp_path, "not: valid: yaml: [")
+        app = PfApp(["graph", str(yaml_file)])
+        rc = app.run()
+        assert rc != 0
+        err = capsys.readouterr().err
+        assert "加载 YAML 失败" in err
+
+    def test_graph_empty_yaml(
+        self,
+        tmp_path: Path,
+        capsys: pytest.CaptureFixture[str],
+    ) -> None:
+        """空 jobs（``jobs: {}``）应被 yaml_loader 拒绝并返回错误。"""
+        yaml_file = _write_yaml(tmp_path, "jobs: {}\n")
+        app = PfApp(["graph", str(yaml_file)])
+        rc = app.run()
+        assert rc != 0
+        err = capsys.readouterr().err
+        assert "加载 YAML 失败" in err
+
+    def test_graph_invalid_format_choice(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """--format 非法值应触发 argparse SystemExit(2)。"""
+        yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
+        app = PfApp(["graph", str(yaml_file), "--format", "invalid"])
+        with pytest.raises(SystemExit) as exc_info:
+            app.run()
+        assert exc_info.value.code == 2
+
+
+class TestPfYamlrunProgressFlag:
+    """``pf yamlrun --progress`` 选项测试。"""
+
+    def test_yamlrun_progress_flag_executes_successfully(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """``--progress`` 标志应能成功执行简单图。"""
+        yaml_file = _write_yaml(
+            tmp_path,
+            """
+jobs:
+  hello:
+    cmd: ["echo", "hello"]
+""",
+        )
+        app = PfApp(["yamlrun", str(yaml_file), "--progress", "--quiet"])
+        rc = app.run()
+        assert rc == 0
+
+    def test_yamlrun_progress_with_only_flag(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        """``--progress`` 与 ``--only`` 应能组合使用。"""
+        yaml_file = _write_yaml(tmp_path, _SIMPLE_YAML)
+        app = PfApp(["yamlrun", str(yaml_file), "--progress", "--quiet", "--only", "transform"])
+        rc = app.run()
+        assert rc == 0
+
+    def test_yamlrun_progress_suppressed_by_quiet(
+        self,
+        tmp_path: Path,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """``--quiet`` 与 ``--progress`` 同时存在时，progress 应被禁用。"""
+        yaml_file = _write_yaml(
+            tmp_path,
+            """
+jobs:
+  hello:
+    cmd: ["echo", "hello"]
+""",
+        )
+
+        captured_progress: dict[str, bool] = {}
+
+        def fake_run(*_args: object, **kwargs: object) -> object:
+            captured_progress["progress"] = bool(kwargs.get("progress"))
+            from pyflowx import RunReport
+
+            report = RunReport()
+            report.success = True
+            return report
+
+        # 拦截 pyflowx.run 调用以检查 progress 参数
+        import pyflowx
+
+        monkeypatch.setattr(pyflowx, "run", fake_run)
+        # pf.py 中通过 `from pyflowx import Graph, run` 局部导入，
+        # 所以 monkeypatch 替换 pyflowx.run 后，下次执行 _run_yaml 时会重新导入
+
+        app = PfApp(["yamlrun", str(yaml_file), "--progress", "--quiet"])
+        rc = app.run()
+        assert rc == 0
+        # quiet 优先级高于 progress
+        assert captured_progress["progress"] is False