跳至主要內容

快速开始

HSXISAWD2026/3/20大约 5 分钟快速入门HsxWorkFlow快速开始

快速开始

本文档将带你快速上手 HsxWorkFlow 框架，从环境准备到第一个工作流的运行。

环境准备

系统要求

  • Python: >= 3.10
  • 操作系统: Windows / Linux / macOS
  • 包管理器: uv（推荐）或 pip

安装依赖

使用 uv（推荐）

# 安装 uv
pip install uv

# 使用 uv 安装 HsxWorkFlow
uv pip install hsxworkflow

使用 pip

pip install hsxworkflow

依赖说明

包名版本用途
fastapi>= 0.109.0Web 框架
uvicorn>= 0.27.0ASGI 服务器
python-socketio>= 5.11.0WebSocket 支持
pydantic-settings>= 2.1.0配置管理
apscheduler>= 3.10.0定时调度
drissionpage>= 4.1.0浏览器自动化
redis>= 5.0.0Redis 客户端
aiomysql>= 0.2.0异步 MySQL

第一个工作流

让我们创建一个简单的工作流，演示 HsxWorkFlow 的基本用法。

工作流代码

创建文件 my_workflow.py：

from hsxworkflow import WorkflowManager, StepActionHandler, WorkRegisterHandler

# 1. 创建工作流注册器
wf = WorkRegisterHandler(
    step_key="demo",           # 唯一标识
    title="演示工作流",         # 显示标题
    desc="三步走的示例流程",    # 描述信息
)

# 2. 定义工作流类
@wf.register_class()
class DemoWorkFlow(StepActionHandler):

    def __init__(self):
        super().__init__()
        self.title = "演示工作流"
        self.desc = "展示基本用法"

    @wf.step(sort=1)
    def prepare(self):
        """准备阶段"""
        self.socket_log("正在初始化资源...", level="i", base=True)
        self.set_global_data("task_id", "TASK-001")
        return self.set_success(message="资源初始化完成")

    @wf.step(sort=2, retry=3, retry_interval=2)
    def process(self):
        """处理阶段（失败自动重试 3 次）"""
        task_id = self.get_global_data("task_id")
        self.socket_log(f"处理任务 {task_id}...", level="i", base=True)

        # 模拟业务逻辑
        result = {"count": 42, "status": "completed"}

        return self.set_success(message="数据处理完成", data=result)

    @wf.step(sort=3)
    def report(self):
        """报告阶段"""
        self.socket_log("生成报告...", level="i", base=True)
        return self.set_success(message="流程结束")

代码解析

  1. WorkRegisterHandler: 工作流注册器，用于收集工作流定义

    • step_key: 工作流的唯一标识
    • title: 显示标题
    • desc: 描述信息

  2. @register_class(): 类装饰器，注册工作流类

    • 必须继承 StepActionHandler
    • 可指定 copy_run 参数控制实例数量

  3. @step(): 步骤装饰器，定义执行步骤

    • sort: 执行顺序（从小到大）
    • retry: 失败重试次数
    • retry_interval: 重试间隔（秒）

  4. 状态设置方法:

    • set_success(): 标记步骤成功
    • set_failure(): 标记步骤失败
    • set_error(): 标记步骤错误

  5. 全局数据方法:

    • set_global_data(): 设置全局共享数据
    • get_global_data(): 获取全局共享数据

  6. socket_log(): 输出日志并推送到 WebSocket

    • level: 日志级别（i/d/w/e/c）
    • base: 是否为顶级日志（控制缩进）

运行工作流

HsxWorkFlow 提供了三种运行方式，你可以根据场景选择合适的方式。

方式一：Web 管理界面（推荐）

这是最推荐的方式，提供可视化的管理界面和实时监控。

if __name__ == "__main__":
    manager = WorkflowManager()
    manager.register_object(wf)
    manager.run_app()  # 访问 http://localhost:5050

访问地址:

  • Web 界面: http://localhost:5050
  • API 文档: http://localhost:5050/docs

自定义配置:

manager.run_app(
    host="0.0.0.0",    # 监听地址
    port=8080,         # 监听端口
    debug=True,        # 调试模式
)

方式二：直接执行

适用于需要程序化控制的场景。

if __name__ == "__main__":
    manager = WorkflowManager()
    manager.register_object(wf)

    # 启动所有工作流
    manager.start_workflow()

    # 等待所有工作流完成
    results = manager.wait_result()

    # 处理结果
    for r in results:
        if r:
            print(f"工作流完成: {r.to_dict()}")
        else:
            print("工作流执行失败")

方式三：指定工作流 ID 执行

适用于有多个工作流的场景，可以精确控制执行哪个工作流。

if __name__ == "__main__":
    manager = WorkflowManager()
    manager.register_object(wf)

    # 获取所有工作流 ID
    work_ids = list(manager.get_work_objects().keys())
    print(f"可用工作流 ID: {work_ids}")

    # 启动指定工作流
    if work_ids:
        manager.start_workflow(work_id=work_ids[0])

        # 等待结果
        results = manager.wait_result()
        for r in results:
            print(r.to_dict() if r else "执行失败")

配置文件

HsxWorkFlow 支持通过 .env 文件进行配置，创建 .env 文件：

# 基础配置
WORKFLOW_DEBUG=false
WORKFLOW_ENVIRONMENT=production
WORKFLOW_MAX_WORKERS=10
WORKFLOW_DEFAULT_TIME_INTERVAL=1.0

# Web 服务配置
WEB_HOST=0.0.0.0
WEB_PORT=5050
WEB_DEBUG=false
WEB_DOCS_ENABLED=true

# 日志配置
LOG_LEVEL=INFO
LOG_DIR=logs

# Redis 配置
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=
REDIS_DB=0

# 数据库配置
DB_HOST=localhost
DB_PORT=3306
DB_USER=root
DB_PASSWORD=secret
DB_DATABASE=workflow_db

CLI 命令行工具

HsxWorkFlow 提供了命令行工具，方便快速启动和管理。

启动 Web 服务

# 使用默认配置启动
hsxworkflow serve

# 自定义端口和调试模式
hsxworkflow serve --port 8080 --debug

# 指定监听地址
hsxworkflow serve --host 0.0.0.0 --port 5050

查看版本

hsxworkflow version

系统信息

hsxworkflow info

查看配置

# 输出当前配置
hsxworkflow config

# 以 JSON 格式输出
hsxworkflow config --format json

验证安装

运行以下代码验证安装是否成功：

from hsxworkflow import WorkflowManager, StepActionHandler, WorkRegisterHandler

# 创建简单的工作流
wf = WorkRegisterHandler(step_key="test")

@wf.register_class()
class TestWorkFlow(StepActionHandler):
    @wf.step(sort=1)
    def hello(self):
        self.socket_log("Hello, HsxWorkFlow!", level="i", base=True)
        return self.set_success(message="测试成功")

# 测试运行
if __name__ == "__main__":
    manager = WorkflowManager()
    manager.register_object(wf)
    print("HsxWorkFlow 安装成功！")

    # 可以选择启动 Web 服务
    # manager.run_app()

运行后，如果看到 "HsxWorkFlow 安装成功！" 的输出，说明安装成功。

下一步

现在你已经成功创建了第一个工作流，可以继续学习：

  • 核心概念 - 了解 WorkRegisterHandler、StepActionHandler、WorkflowManager
  • 装饰器 API - 学习 @step、@step_child 等装饰器的使用
  • 执行模式 - 探索 single、traversal、indefinitely 等执行模式
  • Web 管理界面 - 学习如何使用 Web 界面管理工作流

常见问题

Q: 如何查看工作流执行日志？

A: 在 Web 界面中，可以实时查看工作流执行日志。也可以使用 socket_log() 方法将日志输出到控制台和 WebSocket。

Q: 如何停止正在运行的工作流？

A: 可以通过以下方式停止：

  1. Web 界面点击停止按钮
  2. 调用 REST API POST /api/workflow/stop
  3. 通过 WebSocket 发送 stop 事件
  4. 在代码中调用 manager.stop_workflow(work_id)

Q: 如何调试工作流？

A: 可以通过以下方式调试：

  1. 设置 debug=True 启动调试模式
  2. 使用 socket_log() 输出详细日志
  3. 查看 API 文档中的接口说明
  4. 使用 level="d" 输出调试日志

Q: 支持哪些 Python 版本？

A: HsxWorkFlow 支持 Python 3.10 及以上版本，推荐使用 Python 3.11 或 3.12。

Q: 如何升级 HsxWorkFlow？

A: 使用以下命令升级：

# 使用 uv
uv pip install --upgrade hsxworkflow

# 使用 pip
pip install --upgrade hsxworkflow
Default footer
Copyright © 2026 HSXISAWD