全中文可视化 AI 工作流编排框架
让每个会聊微信的人，都能用 AI 完成复杂需求
🎮 在线演示 · ✨ 核心特性 · 🚀 快速开始 · GitHub
⚠️ 重要提示：本项目目前仅供作者个人测试使用，距离理想状态还需要迭代多个版本，当前版本不建议在生产环境中使用。API 接口、工作流格式、节点定义等均可能发生破坏性变更。
NodeFlow 是一个全中文可视化的 AI 工作流编排框架。你用自然语言描述需求，AI 生成一张中文标注的流程图——每一步做什么、数据怎么传、分支怎么走，全部可视化地摊在面前。
不需要懂代码，不需要会英文，只要能看懂流程图，就能理解和掌控整个 AI 程序的运行。
💡 核心理念：数字平权 — 不该因为一个人不会英文或不会写代码，就剥夺他用 AI 解决实际问题的能力。
📊 全中文可视化流程图
⬆️ 截图占位：将评分助手工作流的完整流程图截图放在 docs/screenshots/flowchart-demo.png
🎯 实际应用：AI 作品自动评分助手
⬆️ 截图占位：将评分助手生成的评分报告截图放在 docs/screenshots/scoring-demo.png
| 特性 | 说明 |
|---|---|
| 🀄 全中文可视化 | 节点名称用中文描述工作内容，变量名配独立中文名。不懂英文也能看懂 |
| 🤖 自然语言生成 | 内置 nodeflow-builder Skill，用中文描述需求自动生成工作流 |
| 📦 声明式定义 | 节点 + 边 = 完整工作流，零胶水代码。JSON 配置清晰易读 |
| 🔍 可观测可调试 | 结构化执行日志，运行时流程图显示实际路径，对着图就能排查 |
| 🔀 丰富的控制流 | 条件分支、循环、并行（fan_out/fan_in），复杂逻辑轻松搞定 |
| 🧩 节点扩展简单 | 一个 @node 装饰器就能注册自定义节点，业务逻辑高度复用 |
运行以下命令启动交互式演示：
python tests/run_demo.py体验内容：
- 全中文 SVG 流程图可视化
- AI 作品自动评分助手
- 声明式工作流代码展示
pip install -e .python tests/run_demo.pyfrom nodeflow import Flow
flow = Flow.from_json("tests/workflows/scoring_assistant/workflow.json")
flow.save_svg_html("scoring_flow.html")NodeFlow 是一个轻量级工作流引擎。你可以把它理解成一个"流程导演"：
- 你告诉它有哪些步骤（节点）
- 你告诉它步骤之间怎么走（边）
- 它按顺序执行，自动调用大模型、读写文件、执行代码等
比如你可以用它做一个"自动写文章"的流程：输入主题 → 让 AI 写初稿 → 检查字数不够就续写 → 拼接自定义信息 → 输出最终结果。
常见使用场景：
- AI 写作：输入关键词，自动写短文、续写、汇总
- 数据处理：读取文件 → 调用 LLM 分析 → 写入结果文件
- 简单自动化：HTTP 请求 → 判断结果 → 发送通知
- 学习工作流原理：代码量少，结构清晰，适合学习
本项目自带一个 nodeflow-builder Skill（工作流构建助手），支持工作流生成、审查和优化。
Skill 源码放在项目的 skills/ 目录下，你可以要求你的AI IDE读取该文件并添加到技能中。
如果你使用的是支持 Skill 的 AI IDE（如 Trae），安装后直接描述需求即可：
"我想做一个工作流：先读取用户上传的 CSV，调用大模型逐行分析情感，把结果写回新的 CSV，如果某行失败就跳过并记录日志。"
Skill 会结合当前项目已注册的节点，自动生成：
- 工作流设计说明
- 可直接运行的
workflow.json - 必要的自定义节点代码（仅在业务上必须封装时才新建）
- SVG 流程图
这样你不需要先背诵所有节点参数，只需要描述"做什么"，由 Skill 帮你决定"怎么做"。
| 概念 | 大白话 | 例子 |
|---|---|---|
| 节点（Node） | 一个步骤，做一件事 | 调用大模型、执行 Python 代码、读取文件 |
| 边（Edge） | 步骤之间的连线 | A 做完后做 B，或者根据条件选择 B 或 C |
| 状态（State） | 工作流运行时的"公共黑板" | 每个节点都可以读和写上面的数据 |
NodeFlow/
├── src/nodeflow/ # 核心代码
│ ├── core.py # 节点注册、日志、JSON 加载
│ ├── flow.py # 工作流引擎
│ ├── svg_flow.py # SVG 流程图生成
│ ├── graph_layout.py # SVG 布局引擎
│ ├── analysis.py # 工作流分析工具（变量分析等）
│ └── nodes/ # 内置节点
│ ├── input.py # 用户输入、搜索、数据库查询、HTTP、文件读取、正则提取、网页抓取
│ ├── llm.py # 大模型对话
│ ├── condition.py # 条件判断、意图识别（内置循环计数器）
│ ├── operation.py # 代码执行、变量处理、数据库写入
│ └── output.py # 文件写入、消息输出、通知
├── tests/ # 测试和演示
│ ├── run_demo.py # 交互式演示脚本（菜单驱动）
│ ├── test_core.py # 核心功能测试
│ ├── test_flow.py # 工作流引擎测试
│ ├── test_svg.py # SVG 流程图测试
│ └── workflows/ # 所有工作流统一放这里
│ ├── beginner/ # 短文写作循环工作流
│ └── scoring_assistant/ # 自动评分助手工作流
├── skills/ # Skill 源码（随 Git 提交）
│ └── nodeflow-builder/ # 工作流构建助手 Skill
├── docs/ # 文档
├── README.md # 本文件
└── pyproject.toml # 项目配置
pip install -e .如果你想让 LLM 节点真正调用大模型，可用以下三种方式之一（按优先级从高到低）：
# Windows PowerShell
$env:NODEFLOW_LLM_API_KEY = "你的 API Key"
$env:NODEFLOW_LLM_BASE_URL = "https://api.deepseek.com/v1"
$env:NODEFLOW_LLM_MODEL = "deepseek-chat"
# Linux/macOS
export NODEFLOW_LLM_API_KEY="你的 API Key"
export NODEFLOW_LLM_BASE_URL="https://api.deepseek.com/v1"
export NODEFLOW_LLM_MODEL="deepseek-chat"复制配置模板并填入你的 Key：
cp config.json.example config.json编辑 config.json：
{
"llm": {
"api_key": "你的 API Key",
"base_url": "https://api.deepseek.com/v1",
"model": "deepseek-chat"
}
}export NODEFLOW_CONFIG=/path/to/your/config.json三种方式可叠加，优先级：环境变量 A >
NODEFLOW_CONFIG指定文件 C > 当前目录config.jsonB。 如果都不配置，LLM 节点会报错（当前版本已取消 mock 模式）。
通过 Flow.from_json(..., env={...}) 在加载工作流时注入，避免在容器镜像中固化密钥：
flow = Flow.from_json(
"tests/workflows/beginner/workflow.json",
env={
"NODEFLOW_LLM_API_KEY": "动态获取的 Key",
"NODEFLOW_LLM_BASE_URL": "https://api.deepseek.com/v1",
"NODEFLOW_LLM_MODEL": "deepseek-chat",
},
)python tests/run_demo.py这是一个带菜单的交互式脚本，菜单结构：
- 选择工作流并操作 — 列出所有可用工作流，选择后可执行以下操作：
- 执行工作流（统一异步执行）
- 生成静态流程图（不需要运行，直接看结构）
- 生成运行时流程图（根据上次执行结果，画出实际走过的路径）
- 生成变量清单
- 验证工作流合法性
- 查看所有可用节点 — 列出全部内置节点和自定义节点
- 查看指定节点详情 — 看某个节点需要什么参数、返回什么
工作流会自动从 tests/workflows/ 下的子目录中发现，每个子目录如果包含 workflow.json 就是一个工作流。
打开 tests/workflows/beginner/workflow.json，里面详细注释了每个节点和边的作用。
它的执行流程是：
提示输入主题 → 获取主题 → LLM 写短文 → 检查字数
↓
字数 ≤ 500 且循环 < 10 次：续写 → 合并 → 再次检查
↓
字数 > 500 或循环达到 10 次：依次执行 5 个自定义节点 → 汇总输出
from nodeflow import Flow
# 从 JSON 文件加载
flow = Flow.from_json("tests/workflows/beginner/workflow.json")
# 同步运行（命令行脚本用）
result = flow.run(input_state={"topic": "人工智能"})
# 异步运行（FastAPI/Jupyter 等已有事件循环的环境用）
# import asyncio
# result = asyncio.run(flow.arun(input_state={"topic": "人工智能"}))
# 打印结果
print(result)关于异步：NodeFlow 内部完全异步。同步节点自动用
asyncio.to_thread包装，不会阻塞事件循环。
run()— 同步接口，内部用asyncio.run()包装，适合命令行脚本arun()— 异步接口，直接 await，适合 FastAPI、Jupyter 等已有事件循环的环境
errors = flow.validate()
if errors:
print("工作流有问题：", errors)
else:
print("工作流合法")# 静态流程图：只显示结构（SVG 格式）
svg_code = flow.generate_svg()
# 运行时流程图：显示实际执行结果（需要先 run 一次）
flow.run(input_state={"topic": "测试"})
runtime_svg = flow.generate_runtime_svg()
# 保存为 HTML 文件
flow.save_svg_html("my_flow.html")
# 或者用内置方法导出
flow.export_graph("my_flow.html")流程图是纯 SVG + HTML，不依赖任何外部 JS 库，直接用浏览器打开即可查看。
import nodeflow
# 列出所有节点
nodeflow.list_nodes(print_output=True)
# 查看某个节点详情
nodeflow.show_node("llm_chat", print_output=True)from nodeflow import node
@node(
name="my_node",
category="custom",
description="我的自定义节点",
inputs=[{"name": "value", "type": "int", "description": "输入数字"}],
outputs=[{"name": "result", "type": "int", "description": "输出结果"}],
)
def my_node(state, params):
value = params.get("value", 0)
return {"result": value * 2}只要这个文件被 import，节点就注册成功了，工作流 JSON 里可以直接用 "type": "my_node"。
@node 装饰器支持以下可选字段，用于声明节点的外部依赖、临时资源与重试策略：
| 字段 | 类型 | 默认值 | 说明 |
|---|---|---|---|
requires_tools |
List[str] |
[] |
节点依赖的外部命令行工具名（如 ["ffmpeg", "yt-dlp"]），validate() 时会检查 PATH 中是否存在，缺失则生成 warning |
temp_resources |
List[str] |
[] |
节点返回值中的临时资源字段名（如 ["video_path"]），节点成功返回后引擎自动清理并从 state 中移除，避免传递到下游 |
retry |
int |
0 |
节点抛异常时的最大重试次数（不含首次执行）。如 retry=2 表示最多执行 3 次 |
retry_delay |
float |
0.0 |
重试之间的等待秒数 |
@node(
name="download_video",
category="custom",
description="下载视频到临时目录",
requires_tools=["ffmpeg"], # validate() 检查 ffmpeg 是否在 PATH
temp_resources=["video_path"], # 节点返回后自动删除临时视频文件
retry=2, # 失败时最多重试 2 次
retry_delay=1.0, # 每次重试间隔 1 秒
)
def download_video(state, params):
# ... 下载逻辑 ...
return {"video_path": "/tmp/video.mp4", "duration": 60}
# video_path 会被引擎清理，duration 保留传递到下游涉及外部资源（浏览器、临时目录、文件句柄等）的工作流，推荐使用 with Flow() as f: 语法，确保即使中途异常也能执行清理：
from nodeflow import Flow
# 方式一：注册清理回调（with 退出时执行）
with Flow.from_json("workflow.json") as flow:
flow.register_cleanup(lambda state: print(f"清理完成，最终状态键：{list(state.keys())}"))
result = flow.run(input_state={"url": "https://example.com"})
# ... 即使 run() 抛异常，register_cleanup 也会执行
# 方式二：通过 run() 参数传递一次性回调（不使用 with 时）
def my_cleanup(state):
# 关闭浏览器、删除临时目录等
pass
flow = Flow.from_json("workflow.json")
result = flow.run(
input_state={"url": "https://example.com"},
cleanup_handlers=[my_cleanup], # run() 结束后执行（包括异常时）
)注意：
register_cleanup注册的回调仅在with语句退出时执行一次；run(cleanup_handlers=...)传递的回调在每次run()结束时执行。两者可同时使用，互不重复。
默认情况下，节点抛异常会触发工作流的 global_exception_handler。如果只想对特定节点做局部降级，可以为该节点添加 on_error 边：
{
"nodes": [
{"id": "fetch", "type": "http_request", "params": {"url": "{target_url}"}},
{"id": "fallback", "type": "variable_process", "params": {"outputs": {"recovered": true}}}
],
"edges": [
{"from": "fetch", "to": "fallback", "on_error": true},
{"from": "fallback", "to": "END"}
]
}fetch节点抛异常时，引擎会写入_error和_error_node到 state，并路由到fallback节点fallback节点可读取state["_error"]获取错误信息- 若节点同时声明了
retry，会先重试，重试耗尽后才走on_error边 on_error边优先于global_exception_handler
| 节点名 | 类别 | 作用 | 主要参数 |
|---|---|---|---|
user_input |
input | 从控制台读取用户输入 | prompt, variable |
file_read |
input | 读取文件 | file_path |
http_request |
input | 发送 HTTP 请求 | method, url, headers, body |
web_search |
input | 联网搜索（Bing/DuckDuckGo） | query, max_results |
db_query |
input | 数据库查询（mock） | query_type, table, where |
regex_extract |
input | 正则提取 | text, pattern, group |
web_fetch |
input | 网页抓取（提取标题/正文/链接/图片/代码块） | url, max_text_length, timeout |
llm_chat |
llm | 调用大模型 | system_prompt, user_input |
condition |
condition | 条件判断（内置循环计数器） | conditions, logic, max_executions |
intent_recognize |
condition | 意图识别（基于 LLM） | user_input, intent_candidates |
variable_process |
operation | 变量处理（拼接/运算/格式转换） | variables, outputs |
code_exec |
operation | 执行 Python 代码 | code |
db_write |
operation | 数据库写入（mock） | mode, table, data |
file_write |
output | 写入文件 | file_path, content, mode |
message_output |
output | 消息输出 | title, content, channel |
notification |
output | 通知输出（message_output 别名） | title, content |
mock 是什么意思？ 就是这些节点没有真的去调用搜索引擎或数据库，而是返回假数据。后续你可以替换成真实实现。
每个内置节点的完整入参、出参、类型、默认值、详细说明都写在对应源码文件的函数 docstring 里。你也可以通过 AI IDE 的 nodeflow-workflow-studio Skill 直接询问某个节点怎么用。
下面给出各节点的参数速查表，方便你快速定位。
| 节点 | 主要入参 | 主要出参 | 说明 | 源码 |
|---|---|---|---|---|
user_input |
prompt, variable |
{variable}, success, error |
从控制台读取用户输入 | input.py |
file_read |
file_path |
file_content, file_path, file_size, success, error |
真实读取本地文件 | input.py |
http_request |
method, url, headers, body, timeout |
http_response, url, method, success, error |
真实发送 HTTP 请求 | input.py |
web_search |
query, max_results, search_engine |
search_results, success |
真实联网搜索（Bing/DuckDuckGo） | input.py |
db_query |
query_type, table, where, query_params |
query_result, success |
mock 数据库查询 | input.py |
regex_extract |
text, pattern, group |
match, matched, all_matches, success, error |
正则提取 | input.py |
web_fetch |
url, max_text_length, timeout |
title, text, links, images, code_blocks, domain, content_hash, success, error |
网页抓取并解析 HTML | input.py |
| 节点 | 主要入参 | 主要出参 | 说明 | 源码 |
|---|---|---|---|---|
llm_chat |
model, temperature, system_prompt, user_input, api_key, base_url |
content, llm_result, usage, success, error |
调用大语言模型生成文本 | llm.py |
| 节点 | 主要入参 | 主要出参 | 说明 | 源码 |
|---|---|---|---|---|
condition |
conditions, logic, max_executions, counter_var |
_route(true/false), {counter_var} |
多条件 AND/OR 判断，内置循环计数器 | condition.py |
intent_recognize |
user_input, intent_candidates |
intent, confidence, entities, llm_raw, success, error |
基于 LLM 识别用户意图 | llm.py |
| 节点 | 主要入参 | 主要出参 | 说明 | 源码 |
|---|---|---|---|---|
variable_process |
variables, outputs |
outputs 中定义的字段 |
变量处理：模板拼接、简单运算、格式转换 | operation.py |
code_exec |
code, raise_on_error |
code_result, success, error（如果 result 是 dict，字段会展开到 state 顶层） |
执行 Python 代码处理数据或做路由决策 | operation.py |
db_write |
mode, table, data, where |
result, success, error |
mock 数据库增删改（mode: insert/update/delete） |
operation.py |
| 节点 | 主要入参 | 主要出参 | 说明 | 源码 |
|---|---|---|---|---|
file_write |
file_path, content, mode |
written, file_path, content_length, success, error |
将内容写入文件（mode: write/append） |
output.py |
message_output |
title, content, channel |
sent, success, error |
消息输出（当前仅 console 渠道） |
output.py |
notification |
title, content |
sent, success, error |
message_output 别名（兼容旧名称） | output.py |
自定义节点通过 @node 装饰器注册，没有固定参数，完全由你定义。写法见 主要接口 / 注册自定义节点，示例源码见 beginner/nodes.py。
优先使用内置节点。只有以下情况才需要写自定义节点：
- 业务逻辑无法用
code_exec+variable_process组合实现 — 比如需要调用特定 SDK（Playwright、ffmpeg 等） - 需要封装复杂的领域知识 — 比如评分量规解析、特定格式的报告生成
- 同一个逻辑在 3 个以上工作流中重复 — 值得抽象为可复用节点
反例：如果只是"从搜索结果里选一个主题"，直接用 code_exec 写 3 行 Python 即可，不需要新建节点。
判断标准（五问法）：
| 原则 | 判断方法 | 适合升级 | 不适合升级 |
|---|---|---|---|
| 通用性 | 3 个以上不相关场景都能用吗？ | file_read（文件读取）— 任何工作流都可能用 |
pick_hot_topic（选主题）— 只有搜索+写作场景 |
| 无业务耦合 | 是否依赖特定业务领域知识？ | code_exec（执行代码）— 纯通用能力 |
score_refine_rubric（量规细化）— 深度绑定评分 |
| 入参出参规范 | 参数能用基本类型表达吗？ | regex_extract（正则提取）— text/pattern 都是 string |
score_parse_input— 依赖特定输入格式 |
| 可独立测试 | 不依赖外部上下文能跑吗？ | variable_process— 给 state+params 就能跑 |
score_aggregate— 依赖 rubric 结构化数据 |
| 后续工作流通用 | 未来其他工作流会用到吗？ | web_fetch（网页抓取）— 通用场景 |
score_format_report— 只在评分场景 |
参考 LangChain、Airflow 等开源项目，自定义节点的入参出参应遵循：
- 入参用基本类型：string/int/float/bool/list/dict，避免自定义类
- 每个入参必须有 default：
{"name": "url", "type": "string", "default": "", "description": "网页 URL"} - 出参必须包含
success和error：统一错误处理模式 - 出参扁平化：不要返回嵌套过深的 dict
- 幂等性：相同入参应产出相同结果（LLM 节点除外）
| 分类 | 解决的问题 | 对应内置节点 | 设计模式 |
|---|---|---|---|
| 数据获取 | 从外部系统读取数据 | file_read、http_request、web_fetch、web_search、db_query、regex_extract、user_input |
input→output，无状态 |
| 数据处理 | 转换、计算、过滤数据 | code_exec、variable_process |
transform，入参即出参来源 |
| 流程控制 | 条件分支、循环、意图路由 | condition、intent_recognize |
只输出 _route，不修改业务数据 |
| 结果输出 | 写文件、发消息、存数据库 | file_write、message_output、db_write |
副作用型，记录写入路径 |
| 边类型 | 作用 | 例子 |
|---|---|---|
normal |
普通连接 | {"from": "A", "to": "B"} |
conditional |
条件分支 | {"from": "A", "to": "B", "when": "_route = pass"} |
fan_out |
一个节点并行分发到多个 | {"from": "A", "to": "B", "type": "fan_out"} |
fan_in |
多个节点汇聚到一个 | {"from": "A", "to": "C", "type": "fan_in", "mode": -1} |
条件边通过 when 属性指定判断条件，支持以下格式：
-
简单匹配（匹配
_route字段）：{"from": "A", "to": "B", "when": "search"}等价于
_route = search -
完整条件（匹配任意字段）：
{"from": "A", "to": "B", "when": "status = success"} -
AND 组合：
{"from": "A", "to": "B", "when": "count < 10 AND status = active"} -
OR 组合：
{"from": "A", "to": "B", "when": "type = urgent OR priority = high"}
循环不需要专门的循环边类型，只要让条件边指回上游节点即可：
{"from": "count_chars", "to": "continue_writing", "when": "_route = continue"}
{"from": "merge_essay", "to": "count_chars"}注意：在节点中需要维护计数器，避免无限循环。例如在 count_chars 节点中：
loop_count < 10时，设置_route: "continue"（继续循环）char_count >= 500或loop_count >= 10时，设置_route: "done"（退出循环）
每次执行会自动记录日志，默认位置由工作流 JSON 里的 log_path 决定。
例如 workflow.json 里写的是：
"log_path": "tests/logs/beginner.json"打开这个文件，可以看到每次执行的：
- 开始时间、结束时间
- 每个节点的输入参数和输出结果
- 执行顺序
- 最终状态
可以只改 JSON 工作流文件，基本不需要写 Python。但如果要加自定义节点，需要会一点 Python。
放在你自己的项目目录里，只要在工作流运行前被 import 就行。推荐一个文件放多个节点，然后 import 一次。
检查 config.json 是否配置了有效的 API Key。
流程图是纯 SVG + HTML 文件，不依赖任何外部库，直接用浏览器打开即可查看。 如果显示异常，检查浏览器是否支持 SVG（所有现代浏览器都支持）。
- 输入参数：统一显示为
中文解释(变量名) 参数值（如搜索关键词(query) 今日热点新闻） - 输出参数：统一显示为
中文解释(变量名) 变量类型（如是否成功(success) bool） - 变量标签：输入参数中的模板变量引用仍会在
var_labels中查找中文名，显示为中文名(英文名)格式
条件边会显示完整的判断条件，例如：
_route = search：当节点输出的_route等于search时走此边status = success AND count < 10：多条件组合
在工作流 JSON 的 var_labels 字段中定义映射：
{
"var_labels": {
"topic": "主题",
"content": "LLM输出内容",
"essay": "短文内容"
}
}通过条件边指回上游节点实现循环，配合计数器控制终止：
- 在节点中维护
loop_count计数器 - 根据条件设置
_route（如continue或done） - 条件边根据
_route值决定走循环分支还是退出分支
- 运行
python tests/run_demo.py，把每个菜单选项都试一遍 - 打开
tests/workflows/beginner/workflow.json，逐行看注释 - 尝试改一个参数，比如把循环次数从 10 改成 5
- 写一个自己的简单工作流 JSON
MIT