PaeoniaCommon/SkillsOdea

创建一个流程

所有路径均相对于仓库根目录。包位于 process_package/。

父类的工作方式

process_package/toy_process/process.py 定义了 Process：

• __init__(**overrides) 设置 self.params = {**self.set_parameters(), **overrides}， 然后设置 self.steps = self.set_steps()。
• set_parameters() 和 set_steps() 是抽象方法。缺少其中任何一个的子类都无法实例化。
• run(context=None) 按顺序遍历 self.steps：
  • Process 实例 → step.run(context)，嵌套执行，使用相同的 context 字典
  • 任何其他可调用对象 → step(context)
  • 其他任何内容 → TypeError
• run 返回 context 字典，并打印带缩进的跟踪信息。

不要修改 toy_process/ 或在子类中覆盖 __init__ 或 run。

子类的规则

1. 从 toy_process 导入 Process 并继承它。
2. set_parameters(self) 返回一个包含默认值的 dict。如果没有参数，则使用 {}。
3. set_steps(self) 返回一个 list。每个元素可以是：
   • 一个绑定方法：self.count_words，不带括号，或者
   • 一个 Process 实例：CleanText(lowercase=self.params["lowercase"])。
4. 每个步骤方法具有签名 def name(self, context):。
   • 从 self.params["..."] 读取设置。
   • 在 context["..."] 中读写数据。
   • 不返回任何内容。返回值会被忽略。
5. 在使用子流程的类上方定义该子流程类，或者导入它。

模板

"""流程功能的一行描述。

从 process_package/ 运行：
    python -m examples.<module_name>
"""

from toy_process import Process


class SubTask(Process):
    def set_parameters(self):
        return {"option": True}

    def set_steps(self):
        return [self.do_part]

    def do_part(self, context):
        if self.params["option"]:
            context["value"] = context["value"] * 2


class MainTask(Process):
    def set_parameters(self):
        return {"start": 1, "option": True}

    def set_steps(self):
        return [
            self.load,
            SubTask(option=self.params["option"]),  # 嵌套子流程
            self.finish,
        ]

    def load(self, context):
        context["value"] = self.params["start"]

    def finish(self, context):
        context["result"] = f"value is {context['value']}"


if __name__ == "__main__":
    print(MainTask(start=5).run()["result"])

要创建的文件

1. process_package/examples/<module_name>.py：流程实现，来自模板。
2. process_package/tests/test_<module_name>.py：一个 unittest 测试，调用 .run() 并对返回的 context 中的键进行断言。还要检查覆盖参数是否会改变结果。

import unittest

from examples.<module_name> import MainTask


class MainTaskTests(unittest.TestCase):
    def test_run(self):
        self.assertEqual(MainTask(start=5).run()["result"], "value is 10")

    def test_parameter_override(self):
        self.assertEqual(MainTask(start=5, option=False).run()["result"], "value is 5")


if __name__ == "__main__":
    unittest.main()

然后使用 run-process 技能进行验证。一个完整的可运行示例位于 process_package/examples/word_report.py。