Agent Workflow 学习向：最小工作流，创建图、同步跑通、按 run

上一篇对照 Dify 搭学习仓先让后端能跑起来示例仓库flow-forge本篇讲在已能探活、能连库的后端上第一次把Workflow跑通——用一份图描述步骤、触发一次运行、事后按run_id回看逐步事件。先说结论上一篇只证明「壳子活着」。本篇补的是工作流产品里最短的一条可演示路径能力你能感知到什么持久化图定义用POST /workflows交一份「步骤图」之后能按 id 读回来同步执行用POST .../runs发一次请求服务在同一次请求里跑完 Start → Template → End运行可回看拿回一个run_id再查终态和逐步事件不必开推送长连接一句话本篇证明「一张最小图能被创建、跑完、事后复盘」不做画布、LLM 节点也不做 SSE/队列。跟跑时你主要和两种「文字格式」打交道都不是 Python 专属格式像什么本篇哪里会用到JSON花括号包起来的键值对前后端传数据的常用包装创建图、启动运行时的请求体接口返回HTTP对某个网址发 GET/POST拿回状态码和正文curl跟跑全程Python 只是本仓用来实现这些能力的语言你先会调接口再回头看源码进度更顺。1. 上一篇停在哪本篇补哪块壳子阶段有探活、分层空位、SQLite 连通但还没有什么叫「一份工作流定义」怎么触发「跑一次」跑的过程如何留下可查询记录对照 Dify 一类产品用户真正关心的正是这三块。本仓用 OpenSpec changegraph-runner切出最小可跑切片节点只留三种执行在请求内做完事件先写入数据库给以后「反复查询进度」留句柄。术语是什么本仓怎么用图graph工作流定义有哪些步骤、谁连谁一份 JSON存进数据库节点node图上的一步本阶段只有开始、模板拼接、结束边edge谁连到谁每条边写清「从哪来 / 到哪去」Run某一次执行有稳定编号run_id结果是成功或失败Event这一次运行里的逐步记录例如某节点开始了、成功了、失败了Runner执行器负责按图把节点跑完的程序入口今天在 HTTP 请求里直接跑以后可改成排队事件结构尽量不动2. 功能一用图描述「开始 → 模板 → 结束」2.1 图长什么样先看 JSON不看 Python一份合法图至少有两块字段含义nodes步骤列表每个步骤有自己的id类型写在data.typeedges连线列表用source/target指向节点的id三种节点data.type干什么start入口你这次运行提交的输入从这里进入template把输入填进一句模板例如Hello, {name}!得到一段文字end收尾把最终文字当作本次运行的输出非法图未知节点类型、边指到不存在的节点、两个节点共用同一个 id 等在创建时就会被拒绝不会变成「跑到一半才炸」。最小示例逻辑上就是开始 → 拼一句 Hello → 结束{nodes:[{id:start_1,data:{type:start}},{id:tpl_1,data:{type:template,template:Hello, {name}!}},{id:end_1,data:{type:end}}],edges:[{id:e1,source:start_1,target:tpl_1},{id:e2,source:tpl_1,target:end_1}]}{name}不是神秘语法它表示「这里以后要换成名叫name的输入」。你启动运行时若传name: Forge拼出来就是Hello, Forge!。2.2 接口怎么用动作方法与路径你交出什么你拿到什么创建工作流POST /workflows{ graph: 上面那份图 }工作流id读取工作流GET /workflows/id无 body同一份图2.3 仓库里校验在干什么可选读实现上校验写在graph.py。你暂时只需知道它在做三件事检查每个节点有没有合法的type检查节点id是否重复检查每条边的两端是不是都指向已有节点顺带认一个 Python 词源码里常见的class ...可以先当成「一份带规则的表格模板」——声明图里允许出现哪些字段、不合法时抛出错误。不必先背class语法看到「校验图」就定位到这个文件即可。3. 功能二触发一次运行同请求内跑完整张图创建只是「把菜谱存起来」。真正干活是启动一次Run动作方法与路径请求体示例启动运行POST /workflows/workflow_id/runs{ inputs: { name: Forge } }inputs就是一张「名字 → 值」的对照表。传给模板里的{name}用。服务端按固定顺序做事你无需读代码也能跟新建一条 Run状态进入「执行中」从唯一的start出发沿边走到下一个节点本阶段不支持同时走两条岔路遇到template用inputs以及上游写出的变量填模板缺了需要的名字 → 这次 Run 记为失败但服务进程继续活着遇到end收齐最终输出这次 HTTP 请求返回时运行已经结束正文里带id即run_id、status成功时还有outputs成功时outputs大致是{text:Hello, Forge!}执行器在仓库里怎么走白话版对应文件runner.py。逻辑可以记成一张流程图拿到图和 inputs → 记下「当前节点 start」 → 循环写事件节点开始按类型做事start 几乎空转 / template 填空 / end 收输出成功则写「节点成功」失败则写「节点失败」并结束本次 Run 沿唯一一条出边走到下一节点没有出边就停 → 把 Run 标成 succeeded 或 failed写回数据库顺带认两个 Python 词仍不必会写你在源码里可能瞥到先怎么理解dict/{ name: Forge }「键值对照表」和 JSON 对象很像运行时的变量就放在这类结构里while ...「条件还成立就重复做」这里用来沿着边一节点一节点往下走今天 HTTP 层直接调用执行器以后若改成「先入队、后台再跑」优先换的是谁去调用而不是推倒事件该怎么记——这是本切片故意留下的升级缝。4. 功能三用run_id查终态与逐步事件启动接口返回时你往往已经看到结果了但产品约定仍是以run_id为稳定编号以后查询都认它。接口用途GET /runs/run_id看终态成功还是失败、最终输出、错误信息GET /runs/run_id/events看逐步事件列表按顺序每个节点何时开始、成败如何为什么要先落库、而不是「必须开一条推送流才能看见过程」因为先有可查询的事件记录以后无论是「隔几秒再问一次」轮询还是推送读的都是同一套数据。数据库里对应三张表名字即职责表名存什么workflows图定义菜谱workflow_runs某一次运行的输入、输出、状态workflow_run_events该次运行的逐步事件带序号sequence表结构入口models.py——可以先当「三张表的说明书」不必先学 ORM。5. 跟跑create → run → events按上一篇在api/启动服务。按api/README.md的 curl 示例走三步Windows / macOS 续行符不同README 里有说明步做什么你要记下来的1POST /workflows提交最小图返回的工作流id2POST /workflows/id/runs带上inputs返回的 runid以及outputs3GET /runs/run_id与.../events终态与逐步事件是否对得上想确认仓库自测也绿cdapi uv run pytestpytest是自动跑测试的工具它替你扮演客户端把「创建 → 运行 → 查事件」走一遍。你暂时只需知道「全绿 ≈ 这条主路径没坏」。6. 分层空位怎么被填上上一篇留下的职责地图本篇开始有实活层本篇长出的内容若你想点开文件controllers对外的 HTTP 门口创建图、启动 run、查询workflows.py、runs.pyservices创建/读取时的编排先校验再存盘workflow_service.pycore/workflow图校验规则、真正按边执行、写事件graph.py、runner.pymodels / db三张表启动时建好models.py读仓库的建议顺序仍然可以几乎不读语法先会用 curl 打通三步本节第 5 节再打开controllers对照「哪个网址对应哪段门口代码」最后才进core/workflow对照「填模板 / 写事件」发生在哪不要在路由文件里找「模板字符串怎么替换」——那是core的事。本篇顺带认识的 Python一张表就够词先怎么记和本篇功能的关系.py文件Python 源码文件业务都在api/src/flow_forge/下class「一类带行为的数据结构」的声明图校验、执行器都是 classdict键值对照表inputs、运行中的变量while条件成立就重复执行器沿着边往下走抛错 /raise主动报告「这里不行了」非法图、缺变量时失败并记入 Run系列后续仍按「功能先、语法附注」写每出现绕不开的词就地用一句话钉住不单独开语法长课。和前作怎么接篇补哪一段01 后端壳子依赖、探活、分层空位、库能连本篇最小图、同步执行、run/event 可查询下一篇更可能落在「谁来消费这些 API」例如最小 Web 联调而不是先把节点类型堆满。你可以从这里带走什么工作流最短演示路径是存图 → 跑一次 → 用 run_id 回看不是先做画布。图是定义Run是某一次执行Event是那一次里的时间线三者分开以后才好做轮询或异步。跟跑优先认JSON HTTPPython 是实现语言可以后看。Template 本阶段只做「填空成句」不执行任意代码。读源码按门口 → 编排 → 核心规则的顺序比从上到下背语法更快建立地图。仓库与相关文档GitHubhttps://github.com/jimchou-h/flow-forge跟跑与 curlapi/README.md图校验core/workflow/graph.py执行器core/workflow/runner.pyHTTP工作流controllers/workflows.pyHTTP运行controllers/runs.py欢迎 Star、Issue 和 PR。本文基于 Flow Forgegraph-runner覆盖最小图定义、同步执行与按 run_id 查询事件假定读者不必先会 Python。不包含画布、LLM 节点与 SSE/队列。

Agent Workflow 学习向：最小工作流，创建图、同步跑通、按 run_id 查事件

相关推荐

Java面试核心：基础语法、JVM与多线程实战解析

Langfuse本地部署与LLM应用可观测性实践指南

ICM雪崩建模：系统韧性与社会技术建模实战指南

更多资讯

marketing-skills完整安装指南：4种安装方式与10+平台支持清单，Cursor到OpenClaw全覆盖

VoLTE掉话率优化：QCI 1承载的协议栈级保活设计

C# async/await原理与高并发异步编程实战指南

Spring Boot集成SAP RFC全攻略：JCo连接、Shiro权限与Quartz调度

哈夫曼编码文件压缩实战：从建树到解码的完整实现

Java Swing文件管理器：跨平台生产级实现指南

今日更新

Cephadm实战指南：Linux系统工程师的生产级Ceph集群部署与排障

不想再被 Token 额度卡脖子？把免费 API 接入 WorkBuddy 的完整教程

ITK中SpatialObject几何体系解析：坐标系、变换与管道集成

本周精选

交通拥堵预测毕设：基于Python+LSTM的车辆流量预测系统全流程解析

MATLAB/Simulink搭建10机39节点电力系统暂态稳定仿真实战

无人机航拍三维重建全流程：从SfM到网格生成的避坑指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？