前言

本文面向已经了解 PLECS XML-RPC 基本原理（见核心原理篇）的读者，通过 5 个实际场景 演示如何编写完整的自动化脚本。每个场景都提供可独立运行的代码模板，只需修改配置部分即可用于你自己的 PLECS 模型。

前提条件：PLECS 已开启服务器模式（Preferences → Enable server mode），端口默认 1080。

---

准备工作：提取通用工具模块

在开始编写各场景脚本之前，先创建一个 plecs_helper.py 文件，将通信层和辅助工具类封装起来，后续所有场景脚本共用。这样做的好处是核心代码只需维护一份，"修改重试次数"这种需求只需改一个地方。

文件：plecs_helper.py

import xmlrpc.client
import numpy as np
import pandas as pd
import os
import math
import time

class PLECSConnector:
    """与 PLECS 的 XML-RPC 连接管理，带重试机制"""

    def __init__(self, url, port, model_path):
        self.url = f"{url}:{port}"
        self.model_path = model_path
        self.model_name = os.path.splitext(os.path.basename(model_path))[0]
        self.proxy = None

    def connect(self):
        print(f"Connecting to PLECS at {self.url}...")
        self.proxy = xmlrpc.client.ServerProxy(self.url)
        self.proxy.plecs.load(self.model_path)
        print(f"Model loaded: {self.model_name}")
        return self.proxy

    def simulate(self, opts=None, max_retries=3, retry_delay=1.0):
        if not self.proxy:
            self.connect()
        for attempt in range(max_retries):
            try:
                if opts:
                    return self.proxy.plecs.simulate(self.model_name, opts)
                else:
                    return self.proxy.plecs.simulate(self.model_name)
            except Exception as e:
                print(f"[Warn] Attempt {attempt+1}/{max_retries}: {e}")
                if attempt < max_retries - 1:
                    time.sleep(retry_delay)
        raise RuntimeError("All simulation attempts failed")

    @staticmethod
    def test_connection(url, port):
        try:
            proxy = xmlrpc.client.ServerProxy(f"{url}:{port}")
            ver = proxy.plecs.version()
            return True, f"Connected. PLECS version: {ver}"
        except Exception as e:
            return False, str(e)


class OptimizationUtils:
    """优化辅助工具：自适应罚函数 + Pareto 前沿提取"""

    @staticmethod
    def calculate_penalty(kpis, constraint_func_str, current_obj_value):
        if not constraint_func_str or constraint_func_str.strip() == "[]":
            return 0.0
        expr_str = constraint_func_str.replace("return ", "").strip()
        try:
            constraints = eval(expr_str, {"np": np, "math": math}, {"kpis": kpis})
            if not isinstance(constraints, list):
                return 0.0
            violation = sum(max(0, g) ** 2 for g in constraints)
            if violation > 0:
                base = abs(current_obj_value) if not math.isinf(current_obj_value) else 1000.0
                penalty = max(base, 1000.0) * 2 + violation * 1e4
                print(f"  [Constraint] Violation! Penalty: +{penalty:.2f}")
                return penalty
        except Exception as e:
            print(f"  [Error] Constraint check failed: {e}")
            return float('inf')
        return 0.0

    @staticmethod
    def get_pareto_front(df, objectives, directions):
        if df.empty or not objectives:
            return df
        data = df[objectives].values.copy()
        data = np.nan_to_num(data, nan=1e12)
        for i, d in enumerate(directions):
            if d == 'maximize':
                data[:, i] = -data[:, i]
        n = data.shape[0]
        is_pareto = np.ones(n, dtype=bool)
        for i in range(n):
            for j in range(n):
                if i == j:
                    continue
                if np.all(data[j] <= data[i]) and np.any(data[j] < data[i]):
                    is_pareto[i] = False
                    break
        return df[is_pareto].copy()


class EarlyStopper:
    """早停机制：连续 patience 次无改善则停止"""

    def __init__(self, patience=20, min_delta=1e-4):
        self.patience = patience
        self.min_delta = min_delta
        self.best_value = float('inf')
        self.wait = 0

    def check(self, current_value):
        if current_value < self.best_value - self.min_delta:
            self.best_value = current_value
            self.wait = 0
            return False
        self.wait += 1
        if self.wait >= self.patience:
            print(f"\n[Early Stopping] No improvement for {self.patience} trials.")
            return True
        return False

---

场景一：参数扫描 — "我想看参数变化对结果的趋势"

需求

对 DC-DC 变换器中电感 L1 和输出电容 C_out 进行网格扫描，输出效率（efficiency）和电压纹波（ripple），保存为 CSV 供后续分析。

完整脚本

文件：sweep_dcdc.py

import numpy as np
import pandas as pd
from itertools import product
import time
import sys
import os

sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from plecs_helper import PLECSConnector

# ============ 修改这里的配置 ============
CONFIG = {
    "connection": {
        "server_url": "http://localhost",
        "port": "1080",
        "model_path": r"D:/models/dcdc_converter.plecs",
    },
    "parameters": {
        "L1":    {"type": "linspace", "start": 50e-6,  "stop": 200e-6, "num": 10},
        "C_out": {"type": "linspace", "start": 47e-6,  "stop": 470e-6, "num": 10},
    },
    "output_csv": "sweep_results.csv",
}
# ========================================

def get_param_combinations():
    """根据配置生成所有参数组合"""
    ranges = []
    names = []
    for name, cfg in CONFIG["parameters"].items():
        names.append(name)
        if cfg["type"] == "linspace":
            ranges.append(np.linspace(cfg["start"], cfg["stop"], cfg["num"]))
        elif cfg["type"] == "logspace":
            ranges.append(np.logspace(np.log10(cfg["start"]), np.log10(cfg["stop"]), cfg["num"]))
        elif cfg["type"] == "list":
            ranges.append(cfg["values"])
    return names, list(product(*ranges))

def extract_kpis(result):
    """提取 KPI：效率（Outport 0）、纹波（Outport 1）"""
    if not isinstance(result, dict) or "Values" not in result:
        return {"efficiency": np.nan, "ripple": np.nan}
    vals = result["Values"]
    return {
        "efficiency": vals[0][-1] if len(vals) > 0 else np.nan,    # Outport 0: 取稳态值
        "ripple":    vals[1].max() if len(vals) > 1 else np.nan,   # Outport 1: 取最大纹波
    }

if __name__ == "__main__":
    start_time = time.time()
    print("=== PLECS Parameter Sweep ===")

    conn = PLECSConnector(
        CONFIG["connection"]["server_url"],
        CONFIG["connection"]["port"],
        CONFIG["connection"]["model_path"],
    )
    proxy = conn.connect()

    param_names, combinations = get_param_combinations()
    optStructs = [{"ModelVars": dict(zip(param_names, combo))} for combo in combinations]
    print(f"Total simulations: {len(optStructs)}")

    all_results = []
    for i, opt in enumerate(optStructs):
        print(f"Run {i+1}/{len(optStructs)}: {opt['ModelVars']}")
        try:
            result = conn.simulate(opt)
            kpis = extract_kpis(result)
            all_results.append({**opt["ModelVars"], **kpis})
            print(f"  -> efficiency={kpis['efficiency']:.4f}, ripple={kpis['ripple']:.6f}")
        except Exception as e:
            print(f"  [Error] {e}")

    if all_results:
        df = pd.DataFrame(all_results)
        df.to_csv(CONFIG["output_csv"], index=False)
        print(f"\nResults saved to: {CONFIG['output_csv']}")
        print(f"Total time: {time.time() - start_time:.1f}s")

运行方式

# 1. 确保 PLECS 已打开且启用了 Server Mode
# 2. 将模型中的扫参数改为变量名（如 L1, C_out）
# 3. 添加两个 Output Port（Port 0 接 Efficiency，Port 1 接 Ripple）
# 4. 运行脚本
python sweep_dcdc.py

输出

生成的 sweep_results.csv 格式：

L1	C_out	efficiency	ripple
5.0e-05	4.7e-05	0.954	0.000123
5.0e-05	9.4e-05	0.957	0.000098
...	...	...	...

可以直接用 Excel、MATLAB 或 Python 画 2D/3D 曲面图进行趋势分析。

---

场景二：Optuna 参数寻优 — "我想自动找到使 THD 最小的参数"

需求

使用 Optuna 的 TPE（Tree-structured Parzen Estimator）算法，自动搜索使变换器 THD（总谐波失真）最小的控制参数 Final_Vset 和 Final_ps。

完整脚本

文件：optimize_dcdc.py

import numpy as np
import pandas as pd
import optuna
import time
import sys
import os

sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from plecs_helper import PLECSConnector, OptimizationUtils, EarlyStopper

# ============ 修改这里的配置 ============
CONFIG = {
    "connection": {
        "server_url": "http://localhost",
        "port": "1080",
        "model_path": r"D:/models/dbsrc_forward.plecs",
    },
    "parameters": [
        {"name": "Final_Vset", "type": "Real", "min": 300.0, "max": 450.0, "scale": "Linear"},
        {"name": "Final_ps",   "type": "Real", "min": 0.05, "max": 0.25,  "scale": "Linear"},
    ],
    "kpis": [
        {"name": "THD", "source": "Outport", "index": 0, "func": "lambda arr: arr[-1]"},
    ],
    "run": {
        "n_trials": 50,
        "direction": "minimize",
        "objective_kpi": "THD",
        "output_csv": "optimize_results.csv",
        # 可选：早停
        "enable_early_stop": True,
        "early_stop_patience": 20,
    },
}
# ========================================

def extract_kpis(result, proxy):
    """从仿真结果提取 KPI"""
    kpi_values = {}
    if not isinstance(result, dict) or "Values" not in result:
        for k in CONFIG["kpis"]:
            kpi_values[k["name"]] = np.nan
        return kpi_values

    all_values = result["Values"]
    for kpi_cfg in CONFIG["kpis"]:
        try:
            if kpi_cfg["source"] == "Outport":
                idx = int(kpi_cfg["index"])
                func = eval(kpi_cfg["func"])
                val = float(func(all_values[idx])) if idx < len(all_values) else np.nan
                kpi_values[kpi_cfg["name"]] = val
        except Exception:
            kpi_values[kpi_cfg["name"]] = np.nan
    return kpi_values


class Objective:
    def __init__(self):
        self.conn = PLECSConnector(
            CONFIG["connection"]["server_url"],
            CONFIG["connection"]["port"],
            CONFIG["connection"]["model_path"],
        )
        self.proxy = self.conn.connect()
        self.log = []
        self.stopper = None
        if CONFIG["run"].get("enable_early_stop"):
            self.stopper = EarlyStopper(CONFIG["run"].get("early_stop_patience", 20))

    def __call__(self, trial):
        # 1. 提议参数
        p = {}
        for pc in CONFIG["parameters"]:
            if pc["type"] == "Real":
                p[pc["name"]] = trial.suggest_float(
                    pc["name"], pc["min"], pc["max"], log=(pc["scale"] == "Log")
                )

        print(f"\n--- Trial {trial.number} ---")
        print(f"  Params: {p}")

        # 2. 运行仿真
        try:
            result = self.conn.simulate({"ModelVars": p})
        except Exception:
            return float("inf")

        # 3. 提取 KPI
        kpis = extract_kpis(result, self.proxy)
        print(f"  KPIs: {kpis}")
        self.log.append({**p, **kpis})

        # 4. 计算目标值
        obj_val = kpis.get(CONFIG["run"]["objective_kpi"], np.nan)
        if np.isnan(obj_val):
            return float("inf")

        direction = CONFIG["run"].get("direction", "minimize")
        score = -obj_val if direction == "maximize" else obj_val
        print(f"  Score: {score:.4f}")

        # 5. 早停检查
        if self.stopper and self.stopper.check(score):
            trial.study.stop()

        return score

    def save(self):
        if self.log:
            pd.DataFrame(self.log).to_csv(CONFIG["run"]["output_csv"], index=False)
            print(f"Results saved to {CONFIG['run']['output_csv']}")


if __name__ == "__main__":
    study = optuna.create_study(
        direction="minimize",
        sampler=optuna.samplers.TPESampler(seed=42),
    )

    obj = Objective()
    try:
        study.optimize(obj, n_trials=CONFIG["run"]["n_trials"])
    except KeyboardInterrupt:
        print("\nInterrupted by user.")
    finally:
        obj.save()

    # 输出最优结果
    print(f"\n=== Optimization Complete ===")
    print(f"Best THD: {study.best_value:.6f}")
    print(f"Best params: {study.best_params}")

运行方式

pip install optuna
python optimize_dcdc.py

输出

--- Trial 0 ---
  Params: {'Final_Vset': 375.2, 'Final_ps': 0.142}
  KPIs: {'THD': 0.0234}
  Score: 0.0234
...
=== Optimization Complete ===
Best THD: 0.0123
Best params: {'Final_Vset': 400.5, 'Final_ps': 0.118}

补充：使用 SKOptimize (高斯过程)

如果更喜欢 scikit-optimize 的高斯过程代理模型，替换 Objective 类和主逻辑：

from skopt import gp_minimize
from skopt.space import Real, Integer

def objective_function(args):
    p = dict(zip(["Final_Vset", "Final_ps"], args))
    try:
        result = conn.simulate({"ModelVars": p})
        thd = result["Values"][0][-1]
        print(f"  THD = {thd:.6f} @ {p}")
        return thd
    except Exception:
        return float('inf')

space = [Real(300, 450, name="Final_Vset"), Real(0.05, 0.25, name="Final_ps")]
result = gp_minimize(objective_function, space, n_calls=50, random_state=42)

print(f"Best THD: {result.fun:.6f}")
print(f"Best params: {dict(zip(['Final_Vset', 'Final_ps'], result.x))}")

---

场景三：多目标优化 — "我想同时最小化损耗和 THD"

需求

在变换器设计中，损耗和 THD 往往是一对矛盾指标。使用 加权和 或 Pareto 前沿 方法进行多目标优化。

加权和模式（Weighted Sum）

将两个目标合并为一个综合指标：

def calculate_weighted_sum(kpis, objectives):
    """objectives = [{"kpi_name": "loss", "direction": "minimize", "weight": 1.0},
                     {"kpi_name": "thd",  "direction": "minimize", "weight": 1.0}]"""
    score = 0.0
    for obj in objectives:
        val = kpis.get(obj["kpi_name"], np.nan)
        if np.isnan(val):
            return float("inf")
        w = obj.get("weight", 1.0)
        # minimize 时直接累加，maximize 时取负
        score += val * w if obj["direction"] == "minimize" else -val * w
    return score

# 在 Objective.__call__ 中使用
kpis = extract_kpis(result, proxy)
weighted_score = calculate_weighted_sum(kpis, [
    {"kpi_name": "loss", "direction": "minimize", "weight": 1.0},
    {"kpi_name": "thd",  "direction": "minimize", "weight": 0.5},
])
return weighted_score

Pareto 前沿模式

使用 Optuna 的 NSGA-II 采样器直接搜索 Pareto 前沿：

import optuna

def objective_multi(trial):
    p = {
        "Final_Vset": trial.suggest_float("Final_Vset", 300, 450),
        "Final_ps": trial.suggest_float("Final_ps", 0.05, 0.25),
    }
    try:
        result = conn.simulate({"ModelVars": p})
    except Exception:
        return float("inf"), float("inf")

    loss = result["Values"][0][-1]  # Outport 0
    thd  = result["Values"][1][-1]  # Outport 1
    return loss, thd

study = optuna.create_study(
    directions=["minimize", "minimize"],
    sampler=optuna.samplers.NSGAIISampler(),
)
study.optimize(objective_multi, n_trials=100)

# 查看 Pareto 前沿上的解
for i, trial in enumerate(study.best_trials):
    print(f"Solution {i}: loss={trial.values[0]:.4f}, thd={trial.values[1]:.4f}, params={trial.params}")

---

场景四：波形导出 — "我想把仿真波形完整导出供外部分析"

需求

单次仿真，将 Time 向量和所有 Outport 信号的完整波形导出为 CSV，便于在 MATLAB 或 Python 中做 FFT、时域分析等。

完整脚本

文件：export_waveform.py

import xmlrpc.client
import numpy as np
import pandas as pd
import time
import os

# ============ 修改这里的配置 ============
CONFIG = {
    "server_url": "http://localhost",
    "port": "1080",
    "model_path": r"D:/models/dcdc_converter.plecs",
    "output_csv": "waveform_output.csv",
    "signals": [
        {"name": "Vout",      "index": 0},
        {"name": "I_L1",      "index": 1},
        {"name": "Efficiency", "index": 2},
    ],
}
# ========================================

def main():
    model_name = os.path.splitext(os.path.basename(CONFIG["model_path"]))[0]
    url = f"{CONFIG['server_url']}:{CONFIG['port']}"

    print(f"Connecting to {url}...")
    proxy = xmlrpc.client.ServerProxy(url)
    proxy.plecs.load(CONFIG["model_path"])
    print(f"Model loaded: {model_name}")

    print("Running simulation...")
    result = proxy.plecs.simulate(model_name)

    if "Time" not in result:
        print("[Error] No Time in result")
        return

    df = pd.DataFrame()
    df["Time"] = result["Time"]

    all_values = result.get("Values", [])
    for sig in CONFIG["signals"]:
        idx = sig["index"]
        if idx < len(all_values):
            df[sig["name"]] = all_values[idx]
            print(f"  + Exported: {sig['name']} (index {idx})")
        else:
            print(f"  [Warn] Signal '{sig['name']}' index {idx} out of range")

    df.to_csv(CONFIG["output_csv"], index=False, float_format="%.15g")
    print(f"Waveform saved to: {CONFIG['output_csv']}")

if __name__ == "__main__":
    main()

输出格式

Time,Vout,I_L1,Efficiency
0.0,0.0,0.0,0.0
1e-06,0.12,0.05,0.0
2e-06,0.25,0.12,0.0
...

---

场景五：自动化回归测试 — "我想确保改代码后设计指标没退化"

需求

在电路设计迭代过程中，每次修改参数或拓扑后自动跑一批测试用例，检查关键 KPI 是否在规定的阈值内，生成 Pass/Fail 报告。

完整脚本

文件：regression_test.py

import xmlrpc.client
import numpy as np
import pandas as pd
import time
import sys
import os

sys.path.append(os.path.dirname(os.path.abspath(__file__)))
from plecs_helper import PLECSConnector

# ============ 修改这里的配置 ============
CONFIG = {
    "connection": {
        "server_url": "http://localhost",
        "port": "1080",
        "model_path": r"D:/models/dcdc_converter.plecs",
    },
    # 测试用例：（工况参数，期望阈值）
    "test_cases": [
        {
            "name": "满载效率",
            "params": {"Vin": 400, "Pout": 3000},
            "checks": {
                "efficiency": {"min": 0.95, "max": 1.0},
                "ripple": {"max": 0.01},
            },
        },
        {
            "name": "轻载效率",
            "params": {"Vin": 400, "Pout": 300},
            "checks": {
                "efficiency": {"min": 0.90, "max": 1.0},
                "ripple": {"max": 0.02},
            },
        },
        {
            "name": "低压满载",
            "params": {"Vin": 300, "Pout": 3000},
            "checks": {
                "efficiency": {"min": 0.93, "max": 1.0},
                "ripple": {"max": 0.015},
            },
        },
    ],
    "report_csv": "test_report.csv",
}
# ========================================

def extract_kpis(result):
    """提取效率 (Outport 0 稳态值) 和纹波 (Outport 1 最大值)"""
    if not isinstance(result, dict) or "Values" not in result:
        return {"efficiency": np.nan, "ripple": np.nan}
    vals = result["Values"]
    return {
        "efficiency": vals[0][-1] if len(vals) > 0 else np.nan,
        "ripple": vals[1].max() if len(vals) > 1 else np.nan,
    }

def run_test_case(conn, test_case):
    """执行单个测试用例，返回结果和状态"""
    print(f"\n--- Test: {test_case['name']} ---")
    print(f"  Params: {test_case['params']}")

    status = "PASS"
    messages = []

    try:
        result = conn.simulate({"ModelVars": test_case["params"]})
        kpis = extract_kpis(result)
    except Exception as e:
        return {"name": test_case["name"], "status": "ERROR", "detail": str(e), "kpis": {}}

    for kpi_name, thresholds in test_case["checks"].items():
        actual = kpis.get(kpi_name, np.nan)
        print(f"  {kpi_name}: {actual}")

        if np.isnan(actual):
            status = "FAIL"
            messages.append(f"{kpi_name}: NaN")
            continue

        if "min" in thresholds and actual < thresholds["min"]:
            status = "FAIL"
            messages.append(f"{kpi_name}: {actual:.4f} < {thresholds['min']:.4f}")
        if "max" in thresholds and actual > thresholds["max"]:
            status = "FAIL"
            messages.append(f"{kpi_name}: {actual:.4f} > {thresholds['max']:.4f}")

    detail = "; ".join(messages) if messages else f"All {len(test_case['checks'])} checks passed"
    return {"name": test_case["name"], "status": status, "detail": detail, "kpis": kpis}


def main():
    print("=== PLECS Regression Test Suite ===\n")

    conn = PLECSConnector(
        CONFIG["connection"]["server_url"],
        CONFIG["connection"]["port"],
        CONFIG["connection"]["model_path"],
    )
    conn.connect()

    results = []
    for tc in CONFIG["test_cases"]:
        result = run_test_case(conn, tc)
        results.append(result)
        status_label = f"[{result['status']}]"
        print(f"  {status_label} {result['detail']}")

    # 生成报告
    total = len(results)
    passed = sum(1 for r in results if r["status"] == "PASS")
    failed = sum(1 for r in results if r["status"] == "FAIL")
    errors = sum(1 for r in results if r["status"] == "ERROR")

    print(f"\n=== Summary ===")
    print(f"Total: {total} | PASS: {passed} | FAIL: {failed} | ERROR: {errors}")

    # 写入 CSV 报告
    report_data = []
    for r in results:
        row = {"test_name": r["name"], "status": r["status"], "detail": r["detail"]}
        row.update(r.get("kpis", {}))
        report_data.append(row)

    pd.DataFrame(report_data).to_csv(CONFIG["report_csv"], index=False)
    print(f"Report saved to: {CONFIG['report_csv']}")

    return failed == 0 and errors == 0


if __name__ == "__main__":
    success = main()
    exit(0 if success else 1)

输出示例

=== PLECS Regression Test Suite ===

--- Test: 满载效率 ---
  efficiency: 0.964
  ripple: 0.00085
  [PASS] All 2 checks passed

--- Test: 轻载效率 ---
  efficiency: 0.925
  ripple: 0.0012
  [PASS] All 2 checks passed

--- Test: 低压满载 ---
  efficiency: 0.945
  ripple: 0.0098
  [PASS] All 2 checks passed

=== Summary ===
Total: 3 | PASS: 3 | FAIL: 0 | ERROR: 0

集成到 CI/CD

脚本返回 0（全通过）或 1（有失败），可以集成到任何 CI/CD 流水线：

# GitHub Actions 示例
- name: Run PLECS Regression Tests
  run: python regression_test.py

---

进阶技巧

断点续传

长时间运行的优化任务可能因意外中断。使用 joblib 保存 Optuna Study 的状态：

import joblib

# 每完成一个 trial 保存状态
def save_callback(study, trial):
    joblib.dump(study, "optimize_checkpoint.pkl")

study.optimize(obj, n_trials=100, callbacks=[save_callback])

下次启动时恢复：

try:
    study = joblib.load("optimize_checkpoint.pkl")
    print(f"Resumed with {len(study.trials)} completed trials.")
except FileNotFoundError:
    study = optuna.create_study(direction="minimize")

热启动

用已知的好参数作为初始猜测，加速优化收敛：

study = optuna.create_study(direction="minimize")
study.enqueue_trial({"Final_Vset": 380, "Final_ps": 0.12})  # 热启动
study.optimize(obj, n_trials=50)

增量写入 CSV

防止仿真中途崩溃丢失结果（推荐 SKOptimize 模式使用）：

def append_to_csv(record, csv_path):
    """逐次追加写入，不丢失已完成的结果"""
    header = not os.path.exists(csv_path)
    pd.DataFrame([record]).to_csv(csv_path, mode='a', header=header, index=False)

---

总结

场景	核心技术	适用情况
参数扫描	itertools.product + 串行/并行仿真	探索参数趋势、画 2D/3D 图
Optuna 优化	TPE 贝叶斯 + trial.suggest_float	30~500 次仿真，自动找到最优解
多目标优化	加权和 / NSGA-II Pareto	多个矛盾指标同时优化
波形导出	单次仿真 + 完整时间序列	获取时域波形做 FFT 等分析
回归测试	测试用例 + 阈值检查 + 报告	设计迭代中确保指标不退步

所有脚本的通用模式：连接 PLECS → 定义参数空间 → 循环执行仿真 → 提取 KPI → 保存结果。掌握这一流程后，你可以根据实际需求自由组合和扩展。