# flowEngine

**Repository Path**: mq/flow-engine

## Basic Information

- **Project Name**: flowEngine
- **Description**: No description available
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-10-21
- **Last Updated**: 2025-11-03

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Flow Engine Demo

一个基于 JSON 的简易流程引擎，支持条件分支、HTTP 请求与任务分配节点，并提供 FastAPI 管理与编排接口。

## 项目结构（FastAPI 最佳实践）
- `api/` 后端服务模块
  - `app.py`：FastAPI 入口，注册中间件、异常处理与路由，并在 lifespan 启动阶段初始化数据库与导出 OpenAPI
  - `routers/flows_controller.py`：流程管理与运行相关路由（创建/更新/删除/运行/校验/Mermaid 导出）
  - `routers/runs_controller.py`：运行记录相关路由（列表/详情/停止）
  - `routers/dto`：Pydantic DTO（Flow/Run 的请求体与参数，按路由模块拆分）
  - `utils.py`：通用工具（统一返回封装、异常处理注册、Mermaid 生成、默认引擎判断、OpenAPI 导出、运行器注册）
  - `db.py`：数据库模型与 CRUD（SQLite + SQLAlchemy）
- `flow_engine/` 引擎包（IO/Core/Nodes/Conditions/Execution）
- `run_flow.py`：命令行示例入口（可选）
- `openapi.json`：服务启动时自动导出的 OpenAPI 文档



## 快速开始
- 环境要求：Python 3.8+（建议 3.10/3.12）
- 使用 uv 管理依赖（推荐）：
  - 安装项目依赖（基于 `pyproject.toml` 与锁文件）：
    ```bash
    uv sync
    ```
  - 开发模式安装（可编辑安装）：
    ```bash
    uv pip install -e .
    ```
  - 新增依赖（示例：添加 Kafka 客户端）：
    ```bash
    uv add aiokafka>=0.10.0
    ```
  - 移除依赖（示例）：
    ```bash
    uv remove aiokafka
    ```
- 或使用 pip（可选）：
  - 开发模式安装项目依赖：
    ```bash
    python -m pip install -e .
    ```
  - 或仅安装核心依赖：
    ```bash
    python -m pip install fastapi uvicorn sqlalchemy langgraph aiokafka
    ```
- 启动服务（项目根目录，推荐使用 uv）：
  ```bash
  uv run uvicorn api.app:app --host 127.0.0.1 --port 8000 --reload
  ```
  - 或直接脚本运行（亦可用 uv 执行）：
    ```bash
    uv run python api/app.py
    ```
- 访问接口：
  - Swagger UI：http://127.0.0.1:8000/docs

## 接口概览
- Flows（前缀 /flows）：
  - POST `/flows` 创建流程（校验后入库）
  - POST `/flows/list` 列出流程
  - POST `/flows/get` 获取流程详情
  - POST `/flows/update` 更新流程
  - POST `/flows/delete` 删除流程
  - POST `/flows/run` 运行流程（支持 engine: flow/lang/demo）
  - POST `/flows/run_demo` 使用 DemoRunner 运行流程
  - POST `/flows/validate` 检查流程是否可运行（是否有 Start 节点等）
  - POST `/flows/mermaid` 将流程转为 Mermaid 代码（`flowchart TD`）
- Runs（前缀 /runs）：
  - POST `/runs/list` 列出运行记录
  - POST `/runs/get` 获取运行详情（含 trace/context）
  - POST `/runs/stop` 停止正在运行的流程

## 运行引擎与配置
- 引擎选择：
  - `flow`：串行执行（FlowRunner）
  - `lang`：基于 LangGraph 的状态图执行（LangGraphRunner）
  - `demo`：演示用（DemoRunner）
- 默认引擎：可通过环境变量 `FLOW_DEFAULT_ENGINE` 设置（支持 `flow`/`lang`/`demo`，当 `langgraph` 未安装时自动回退为 `flow`）。
- CORS：默认允许所有来源、方法与请求头。
- OpenAPI 导出：服务启动后会生成 `openapi.json` 于项目根目录。

## 数据库说明
- 使用 SQLite，数据库文件：`flow_engine_demo.db`
- 首次启动会根据根目录下 `schema.sql` 初始化数据库结构；若缺失则回退为 ORM 自动创建

## 引擎包概览
- 架构：入口 -> FlowGraph -> FlowRunner/LangGraphRunner/DemoRunner -> 节点执行 -> 返回 context 与 trace
- 主要模块：
  - IO：`flow_engine/io/loader.py`
  - Core：`flow_engine/core/graph.py`, `edge.py`
  - Nodes：`flow_engine/nodes/*`
  - Conditions：`flow_engine/conditions/condition.py`
  - Execution：`flow_engine/execution/runner.py`

更多设计细节请查看 `DESIGN.md` 与测试说明 `TESTING.md`。

## 分层架构（Controller / Service / Manager / DAO）
为便于后续扩展与维护，后端服务遵循更细粒度的分层模式，参考 Java 工程中的 MVC 乃至 Controller/Service/Manager/DAO：

- Controller（FastAPI Routers）：负责 HTTP 路由与入参校验，仅做轻薄的控制与异常转换
  - 位置：`api/routers/flows_controller.py`, `api/routers/runs_controller.py`
  - 行为：解析请求体，调用 Service 层，统一封装返回（`utils.envelope`），将业务错误映射为 HTTP 状态码
- Service（业务编排）：负责具体的业务流程与跨模块编排
  - 位置：`api/services/flows_service.py`, `api/services/runs_service.py`
  - 行为：组合 DAO 与 Manager 完成“创建/更新/删除流程”“运行流程”“查询/停止运行”等业务；做必要的校验（如构建 FlowGraph 验证）
- Manager（资源/运行器管理）：负责与执行器、外部资源交互的中间管理层
  - 位置：`api/managers/engine_manager.py`
  - 行为：根据 `engine` 选择并创建 Runner（Flow/LangGraph/Demo），维护内存中的运行器注册表，提供停止运行的接口
- DAO（数据访问）：封装数据库 CRUD，保证数据访问的一致性与可替换性
  - 位置：`api/dao/flows_dao.py`, `api/dao/runs_dao.py`
  - 行为：调用 `api/db.py` 中的具体实现，将其统一暴露为模块化方法

分层调用路径示例：
Controller -> Service -> (Manager + DAO)

好处：
- 解耦路由与业务逻辑，便于单元测试与后续扩展
- 业务编排集中于 Service，复用 Manager 的运行器管理与 DAO 的数据访问
- 与常见企业级工程实践一致（更清晰的职责划分、可插拔的实现层）

## Kafka 集成与使用
- 依赖：已在 <mcfile name="pyproject.toml" path="e:\work\work_ai\flow-engine\pyproject.toml"></mcfile> 的 dependencies 中加入 `aiokafka>=0.10.0`
- 配置：通过环境变量 `KAFKA_BOOTSTRAP_SERVERS` 指定 Kafka 集群地址，默认 `localhost:9092`
- 模块位置：
  - Manager 层：<mcfile name="kafka_manager.py" path="e:\work\work_ai\flow-engine\api\managers\kafka_manager.py"></mcfile>（提供 KafkaManager 与全局单例 KAFKA_MANAGER）
  - Service 层：<mcfile name="kafka_service.py" path="e:\work\work_ai\flow-engine\api\services\kafka_service.py"></mcfile>（提供便捷服务方法）
- 生产消息（推荐通过 Service 层）：
  - 发送 JSON：`kafka_send_json(topic, obj, key=None, headers=None)`
  - 发送字节：`kafka_send_bytes(topic, value: bytes, key=None, headers=None)`
- 消费消息：
  - 创建消费任务：`kafka_consumer_task(topics: List[str], group_id: str, handler: Callable[[Any], Awaitable[None]], decode_json=True)`
  - 将返回的异步函数交给 FastAPI 的 BackgroundTasks 执行（例如在路由中 `bg.add_task(task)`）
  - 处理函数 `handler` 为异步函数；当 `decode_json=True` 时，消费者会尝试把字节消息按 UTF-8 解码并 `json.loads`
  - 停止消费者：`kafka_stop_consumer(group_id, topics)`
- 生命周期管理：
  - 应用关闭阶段（FastAPI lifespan）会自动调用 `KAFKA_MANAGER.close()`，优雅关闭 Producer/Consumer，避免资源泄露
- 端点集成建议：
  - 如需对外暴露发送消息或启动/停止消费者的 REST 接口，可在 `api/routers` 中新增路由，并直接调用上述 Service 方法

### Binlog 消费与处理（Canal JSON）
- 启动时自动开启消费者：在 <mcfile name="app.py" path="e:\work\work_ai\flow-engine\api\app.py"></mcfile> 的 lifespan 启动阶段中，基于绑定 "binlog-db" 创建并启动消费者：`kafka_consumer_task_by_binding("binlog-db", binlog_handler)`。
- 处理器位置与逻辑：<mcfile name="binlog_service.py" path="e:\work\work_ai\flow-engine\api\services\binlog_service.py"></mcfile> 提供 `binlog_handler`、`deal_binlog`、`canal_handle_message`、`event_notify_matching`（与 Kafka 通用封装分离）。
  - `binlog_handler`：按 Canal 风格 JSON 反序列化，识别 INSERT/UPDATE/DELETE，构造统一记录并调用 `deal_binlog`。
  - `deal_binlog`：先调用 `canal_handle_message`（可替换为你的业务处理），再调用 `event_notify_matching` 触发事件路由。
- 配置要求：
  - 必填：`KAFKA_BOOTSTRAP_SERVERS`
  - 绑定：默认内置 `binlog-db`（destination=`binlog-db`, content-type=`application/json`, group=`low-engine_test1`），group 可通过 `KAFKA_BINDING_BINLOG_DB_GROUP` 覆盖，或使用 `KAFKA_BINDINGS_JSON` 批量声明。
- 关闭/禁用自动消费：如不希望在启动时自动消费，可在 `app.py` 的 lifespan 启动阶段注释对应代码块。

## Kafka 配置与绑定（对齐 Spring Cloud Stream 思路）

为对齐 Spring Cloud Stream 的 binder/bindings 思路，项目将 Kafka 连接参数与 Topic 绑定抽象为统一配置，并在 Manager/Service 层提供“按绑定名”进行生产与消费的能力。

- 配置模块：`api/config/kafka_config.py`
  - KafkaSettings：包含 `binder`（连接层）与 `bindings`（绑定定义）
  - binder（连接层）参数：
    - `KAFKA_BOOTSTRAP_SERVERS`（必需）：Kafka 集群地址（示例：`10.10.1.52:9092,10.10.1.5:9092,10.10.1.178:9092`）
    - `KAFKA_REPLICATION_FACTOR`（可选，默认 1）：副本因子（参考 spring.cloud.stream.kafka.binder.replication-factor）
    - `KAFKA_AUTO_CREATE_TOPICS`（可选，默认 true）：是否允许自动创建主题（参考 spring.cloud.stream.kafka.binder.auto-create-topics）
    - `KAFKA_ENABLED`（可选，默认 false）：是否启用 Kafka（测试或本地无 Kafka 时可关闭，关联测试自动跳过）
  - bindings（绑定定义）：
    - 支持通过 `KAFKA_BINDINGS_JSON` 批量声明绑定，示例：
      ```json
      {
        "binlog-db": {
          "destination": "binlog-db",
          "group": "low-engine_test1",
          "content_type": "application/json"
        }
      }
      ```
    - 也支持针对单个绑定使用环境变量覆盖：
      - `KAFKA_BINDING_BINLOG_DB_GROUP`（覆盖绑定 `binlog-db` 的 group）
      - `KAFKA_BINDING_BINLOG_DB_CT`（覆盖绑定 `binlog-db` 的 content-type）

- Spring 配置映射关系：
  - `spring.cloud.stream.kafka.binder.brokers` -> `KAFKA_BOOTSTRAP_SERVERS`
  - `spring.cloud.stream.kafka.binder.replication-factor` -> `KAFKA_REPLICATION_FACTOR`
  - `spring.cloud.stream.kafka.binder.auto-create-topics` -> `KAFKA_AUTO_CREATE_TOPICS`
  - `spring.cloud.stream.bindings.<bindingName>.destination` -> `KAFKA_BINDINGS_JSON` 的 `destination` 或默认值
  - `spring.cloud.stream.bindings.<bindingName>.group` -> `KAFKA_BINDINGS_JSON` 的 `group` 或 `KAFKA_BINDING_<NAME>_GROUP`
  - `spring.cloud.stream.bindings.<bindingName>.content-type` -> `KAFKA_BINDINGS_JSON` 的 `content_type` 或 `KAFKA_BINDING_<NAME>_CT`

- 绑定驱动的生产/消费（Service 层方法）：
  - 生产：
    - 发送 JSON：`kafka_send_by_binding_json("binlog-db", {"id": 1, "op": "update"})`
    - 发送字节：`kafka_send_by_binding_bytes("binlog-db", b"raw-bytes")`
  - 消费（可作为 BackgroundTasks）：
    - 创建绑定消费任务：`task = kafka_consumer_task_by_binding("binlog-db", handler)`
    - 其中 `handler` 为异步函数，若绑定的 `content_type=application/json` 则消费者会自动进行 JSON 解码

- 模块位置（供参考）：
  - 配置：`api/config/kafka_config.py`、`api/config/__init__.py`
  - 管理器：`api/managers/kafka_manager.py`（读取配置，提供按绑定名生产与消费的能力）
  - 服务：`api/services/kafka_service.py`（导出绑定级别的便捷方法）

- 生命周期与清理：
  - 应用关闭时（FastAPI shutdown）会调用 `KAFKA_MANAGER.close()`，统一关闭 Producer/Consumer，避免资源泄露（见 `api/app.py`）

- 可选增强：
  - 如需在启动时自动创建缺失的 Topic 并设置分区/副本，项目已提供 Kafka Admin 客户端的自动创建能力（`KafkaManager._ensure_topic`）。

## Redis 集成与使用
- 依赖：已在 <mcfile name="pyproject.toml" path="e:\work\work_ai\flow-engine\pyproject.toml"></mcfile> 的 dependencies 中加入 `redis>=5.0.0`
- 配置：通过环境变量控制连接参数（`.env` 示例）：
  ```env
  REDIS_ENABLED=true
  REDIS_HOST=r-wz9og1vhqw2witfy9ipd.redis.rds.aliyuncs.com
  REDIS_PORT=16379
  REDIS_PASSWORD=********
  REDIS_DB=0
  REDIS_SSL=false
  ```
- 模块位置：
  - 配置：<mcfile name="redis_config.py" path="e:\work\work_ai\flow-engine\api\config\redis_config.py"></mcfile>
  - 管理器：<mcfile name="redis_manager.py" path="e:\work\work_ai\flow-engine\api\managers\redis_manager.py"></mcfile>
- 使用方式：
  - 获取客户端单例：`from api.managers.redis_manager import REDIS_MANAGER`
  - 基础操作：
    - 连接探测：`await REDIS_MANAGER.ping()`
    - 写入字符串：`await REDIS_MANAGER.set_str("key", "value", ex=60)`
    - 读取字符串：`await REDIS_MANAGER.get_str("key")`
    - 写入 JSON：`await REDIS_MANAGER.set_json("obj:k", {"a": 1}, ex=60)`
    - 读取 JSON：`await REDIS_MANAGER.get_json("obj:k")`
- 注意：
  - 若服务端不支持 TLS 或端口为明文，应设置 `REDIS_SSL=false`（本项目单元测试在开启 TLS 时遇到 `WRONG_VERSION_NUMBER/UNEXPECTED_EOF`，关闭后通过）。
  - 请勿将密码明文提交至版本库，推荐通过 `.env` 与部署平台的密钥管理加载。

### Redis 使用示例（简化封装）

- 模块级便捷方法（同步调用，无需 await）：
  ```python
  from api.managers.redis_manager import (
      redis_set, redis_get, redis_delete, redis_exists, redis_get_or_set,
  )
  ```

- 字符串读写（支持 TTL 秒级过期）：
  ```python
  # 写入字符串，设置 60 秒过期
  redis_set("demo:k1", "hello", ttl_seconds=60)
  # 读取字符串（不存在返回 None）
  val = redis_get("demo:k1")
  ```

- JSON 读写（自动序列化/反序列化，支持 TTL）：
  ```python
  redis_set("demo:user:1", {"id": 1, "name": "Alice"}, ttl_seconds=120)
  obj = redis_get("demo:user:1")  # -> {"id": 1, "name": "Alice"}
  ```

- 简易缓存模式（get_or_set）：
  ```python
  def load_user_42():
      # 这里替换成你的真实数据源读取
      return {"id": 42, "name": "Bob"}

  user = redis_get_or_set("cache:user:42", load_user_42, ttl_seconds=300)
  # 若 key 不存在，执行 factory 计算并写入；若存在，直接返回缓存
  ```

- 检查与删除：
  ```python
  exists = redis_exists("demo:k1")  # -> True/False
  deleted = redis_delete("demo:k1") # -> 删除的键数量（int）
  ```

- 键命名建议：
  - 使用命名空间风格：`app:module:resource:id`，例如：`fe:flows:run:123`。
  - 对缓存类 key 设置合理 TTL，避免长期占用内存；对象类数据建议使用 JSON 存储。
  - 避免缓存敏感信息（密码、token 等）；如需临时缓存，务必设置 TTL。

- 兼容性说明：
  - 仍可使用原有方法：`REDIS_MANAGER.set_str/get_str/set_json/get_json`。
  - 当传入 dict/list 时，`redis_set` 自动序列化为 JSON；其它类型以字符串存储。
  - `redis_get` 会先尝试按 JSON 反序列化，失败则返回原始字符串。
  - 为 JSON 设置 TTL 时，内部通过 `expire(key, seconds)` 实现；字符串写入的 TTL 通过 `ex` 参数完成。


## 测试与验证（使用 uv 管理依赖）

- 安装测试依赖（建议使用 uv 可选依赖机制）：
  - 安装：`uv pip install -e .[test]`
  - 或直接运行：`uv run -m pytest -q`
- Kafka 集成测试（真实连接 Kafka）：
  - 位置：`tests/test_kafka.py`
  - 环境变量对应关系（测试会自动设置）：
    - `KAFKA_BOOTSTRAP_SERVERS=10.10.1.52:9092,10.10.1.5:9092,10.10.1.178:9092`
    - `KAFKA_REPLICATION_FACTOR=1`
    - `KAFKA_AUTO_CREATE_TOPICS=true`
    - `KAFKA_BINDING_BINLOG_DB_GROUP=<unique_group>`
  - 运行示例：
    ```bash
    uv run -m pytest tests/test_kafka.py -q
    ```
  - 测试说明：
    - `test_kafka_producer_send_json_and_consume_real`：真实 Producer 发送 JSON 到绑定 `binlog-db`，真实 Consumer 在同一 group 中消费并断言收到消息
    - `test_kafka_consumer_task_by_binding_real`：通过绑定创建真实消费任务并消费真实 Producer 发送的消息
    - `test_stop_consumer_not_found`：验证停止不存在的消费者时返回 False
  - 跳过策略：
    - 当 `KAFKA_ENABLED=false` 时，整个 `tests/test_kafka.py` 会在模块级跳过，避免无 Kafka 环境下的失败。

- Redis 集成测试（真实连接 Redis）：
  - 位置：`tests/test_redis.py`
  - 运行示例：
    ```bash
    uv run -m pytest tests/test_redis.py -q
    ```
  - 测试说明：
    - `test_redis_connectivity`：验证连接与 `PING`
    - `test_redis_set_get_str`：验证字符串写入/读取
    - `test_redis_set_get_json`：验证 JSON 写入/读取
  - 跳过策略：
    - 当 `REDIS_ENABLED=false` 时测试自动跳过；若启用 TLS 失败，请设置 `REDIS_SSL=false` 并重试。

- 全量测试：
  ```bash
  uv run -m pytest -q
  ```

- 其他：
  - FastAPI 已迁移至 lifespan 事件处理（<mcfile name="app.py" path="e:\work\work_ai\flow-engine\api\app.py"></mcfile>），避免弃用警告；并通过导入字符串 `api.app:app` 支持热重载。

## 执行上下文与 Redis 规范（对齐 dify/n8n/langFlow 思路）

为满足“除 Flow 配置外，还需包含触发入参，并将每个节点的输出写入执行上下文，格式化为 JSON 写入 Redis，供后续节点复用”的诉求，约定如下：

- 键命名：`#{flowRecordId}_context`，其中 `flowRecordId` 即运行记录 ID（DB: runs.id）。
- 写入时机：
  - 运行开始：写入初始上下文（含触发事件等）。
  - 每个节点执行完成后：写入最新快照（包含该节点的标准化输出）。
  - 运行结束：再次写入最终上下文（含 visited_steps、trace 等持久化到 DB）。
- 上下文 Schema（JSON）：
  ```json
  {
    "user": {"id": "optional"},
    "request": {"source": "optional"},
    "event": {
      "eventCode": "REGISTER_ACCOUNT",
      "eventName": "Partner Registered",
      "eventSource": "fuse pro",
      "data": {"email": "alice@example.com", "name": "Alice"}
    },
    "flow": {"id": 123, "engine": "flow"},
    "nodes": {
      "<nodeId>": {
        "id": "<nodeId>",
        "type": "<nodeType>",
        "input": {"...": "节点输入（可选）"},
        "output": {"...": "节点输出（标准化）"},
        "ts": "YYYY-MM-DD HH:mm:ss"
      }
    },
    // 更新示例：移除 last_http，全量以 nodes.<nodeId>.output 为引用
    {
      "nodes": {
        "n_http_1": {
          "output": { "statusCode": 200, "headers": {"...": "..."}, "body": "..." }
        }
      }
    }
  }
  ```
- 设计理念：
  - 触发上下文：Start 节点支持 `triggerType=event`，需在 `context.event` 中提供与配置匹配的事件；该事件来自 binlog 解析后生成的业务对象（例如 RegisterAccountEvent），在此以 JSON 表示。参考 n8n/Dify 的触发器节点设计。
  - 节点输出标准化：每个节点执行后将其 `input` 与 `output` 写入 `context.nodes[<nodeId>]`，保证后续节点能直接复用。类似于 langFlow 的“state graph”中将消息/变量写入共享状态。
  - Redis 快照：每步将最新 `context` 写入 Redis（键：`#{run_id}_context`），便于其他系统或节点按需读取最新状态（例如下游外部执行器或并行子流程）。

### 最简示例与验证步骤

示例：`examples/http_flow.json`（HTTP 节点将响应写入 `nodes.n_http_1.output`）。

1) 创建并运行 Flow（通过 Service 层，自动生成 run_id 并写入 Redis）：
```python
from flow_engine.io.loader import FlowLoader
from flow_engine.core.graph import FlowGraph
from api.services.flows_service import run_flow_service, create_flow_service
from api.dao.flows_dao import get_flow_by_id

# 1. 从示例加载并入库
flow_def = FlowLoader.load_from_json_file("examples/http_flow.json")
flow_id = create_flow_service("http_flow_demo", flow_def)

# 2. 触发上下文（模拟事件/请求）
context = {
  "event": {"eventCode": "HTTP_DEMO", "eventSource": "local", "data": {"q": "ping"}},
  "request": {"source": "readme"},
}

# 3. 运行（默认 engine=flow），返回 run_id 和执行结果
run_id, result = run_flow_service(flow_id=flow_id, engine="flow", context=context)
print("run_id=", run_id)
print("trace size=", len(result.get("trace", [])))
```

2) 验证 Redis 上下文：
```python
from api.managers.redis_manager import REDIS_MANAGER
ctx = REDIS_MANAGER.get_json(f"{run_id}_context")
print(ctx.get("nodes"))

预期：
- `ctx['event']` 为传入的触发事件。
- `ctx['nodes']` 包含每个运行过节点的 `input/output/ts`，可作为后续节点的输入来源。

### 参考与对齐
- 触发器/上下文传递：对齐 n8n/Dify 的触发器与变量传递理念，节点之间通过共享上下文（state）传递数据。
- 状态图/快照：参考 LangGraph 的状态图思想，每步持久化最新上下文，便于并行与外部编排器读取。

注意：
- 若需要将业务对象（如 RegisterAccountEvent）直接存入上下文，建议序列化为 JSON（可含 `type` 与 `payload` 字段），或在节点层按需转换，避免不可序列化对象导致持久化失败。
- Redis 键 TTL 可在 `REDIS_MANAGER.set_json` 层按需设置（当前默认无过期）。