数据截至 (上游 commit fc74d079a18c)
@flow 步骤、组合子与纯函数
30 秒导读: 在 v3 里,一个"流程"就是一个被
@flow装饰的普通 Python 函数。装饰器在 定义期执行它一次:参数和每次调用的返回值都是Handle(占位符),于是工具调用、模型 调用、条件、循环都被"录"成图步骤。运行期没有解释你的 Python——只有编出来的 IR。 这章讲清你能写什么步骤、组合子家族有哪些,以及旧版$表达式被什么取代了。
本章聚焦编写面。IR 与数据结构见 01;编出来的图怎么被执行见 02;工具本身见 04。
旧版对照(已移除): v1 的 YAML 任务(15 种
kind_步骤)、$前缀表达式、 simpleeval 沙箱(base_evaluate/SimpleEval)、{{}}模板兼容层全部不存在于主干。 本章 §7 专门讲"表达式"被什么取代、为什么。
1. 这是什么(零基础也能懂)
一句话定义: @flow 是一种"定义即构造(define-by-construction)"的流程编写法——
你写的是普通 Python,但函数体在定义时跑一遍,跑的目的不是产生结果,而是产生图。
它解决什么问题。 声明式 YAML(v1)表达能力有限、没有类型检查;命令式 Python 运行期
任意执行,没法冻结、没法校验、没法安全重放。@flow 取中间路线:编写体验是 Python
(IDE 补全、类型、重构工具全可用),产物是数据(可哈希、可校验的 IR)。
一个最小流程长这样(摘自 README.md:55-64,有删节):
# 示意,非源码:@flow 里的每行都是一个图步骤
@flow
def triage(ticket: str) -> dict[str, str]:
hit = lookup_ticket(ticket, retries=2, timeout_s=5) # 工具步(带重试/超时)
prompt = ticket_prompt(hit) # 纯函数步
answer = think(support_reply, prompt, timeout_s=10) # 模型步
return hit | answer # 记录合并(std.merge)
读这段的直觉,记住三点:
- 每个赋值是一个步骤,变量名(单赋值)就是这条边在图里的名字。
- kwargs 是执行策略:
retries=、timeout_s=、name=直接写在调用上,编进Ann。 - 函数返回值必须是 Handle(或 Handle 组合),返回别的会得到带源码位置的报错。
一句话类比: 像 React Hooks——"写的是顺序代码,框架在'渲染'时录下结构";也像 BUILD 文件的 Python 化:Bazel 只执行配置代码来生成依赖图,不执行构建本身。
2. 顶层全景(一行代码怎么变成 IR)
怎么读这张图: 从上到下是编译管道。左列是你写的东西,右列是每步产物。
你写的 产物
──────────────────────────────────────────────────────────
@flow def triage(ticket): ─▶ FlowDef(包装函数 + 源码信息)
hit = tool(h) ─▶ Graph.add_step(TOOL) ┐
prompt = pure(h) ─▶ Graph.add_step(PURE) │ dag.Graph(StepNode 单赋值)
cond/s each/switch ─▶ 嵌套子 Graph ┘
return h1 | h2 ─▶ std.merge 步骤
│ dag.compile(julep/dag.py:306)
▼
Node 树(wire-format IR,11 种 Op)
│ deploy(): freeze + 校验(第 02/04 章)
▼
Deployment(不可变,内容寻址)
三个关键部件:
| 部件 | 干什么 | 在哪 |
|---|---|---|
Handle | 定义期的数据占位符;支持 h1 | h2(merge)、h["key"]/h.key(pluck) | julep/define.py:383 |
_append_step | 所有步骤追加的公共入口:解析 name/retries 等 kwargs、记录绑定 | julep/define.py:946 |
Graph/StepNode | 单赋值中间图,拓扑排序后编 IR | julep/dag.py:90/:61 |
定义期执行是怎么发生的? @flow 包装你的函数;调用时它把参数换成 Handle 压进
构建上下文栈,再执行函数体。函数体里出现的任何注册可调用对象(Tool/Pure/Reasoner
包装器)在"作者期"模式下(apply_if_authoring,julep/define.py:892)不是真调用,而是
_append_step。不在 @flow 里调用它们则照常真执行(所以 @tool 函数可以单测)。
3. Handle:只有五个确定性操作的数据线
Handle(julep/define.py:383)是你在 @flow 里唯一能"摸到"的运行数据。它的操作面
被刻意收窄成五个,其余全是教学式报错:
| 操作 | 编译成 | 说明 |
|---|---|---|
h1 | h2 | std.merge 纯函数步(julep/std.py:14) | 合并两条记录 |
h["key"] / h.key | std.pluck(julep/std.py:34) | 取字段 |
| 作为工具/纯函数/think 的输入 | 对应步骤的输入边 | —— |
作为 cond/each/switch 的 subject | 对应控制结构的输入 | —— |
直接 return h | 图的输出 | —— |
误用会得到"教学式"DefineError——不是晦涩的内部错误,而是告诉你该用哪个组合子:
if h:→ "Handle truthiness is not runtime data; use cond(...)"(julep/define.py:437__bool__);for x in h:→ "use each(...)"(:445__iter__);h1 == h2→ 同样指向 cond(:453__eq__)。
这些报错带 SourceSpan(文件/行号/函数),因为 _SourceMap(julep/define.py:98)
解析了整个函数的 AST 源码——定义期诊断是 API 的一部分(julep/define.py:1 docstring)。
kwargs 绑定规则(值得记):调用步骤时,Handle 值的 kwargs 构成"记录绑定"(先用
std.record(julep/std.py:135)把多个 Handle 与常量拼成一条记录),JSON 常量 kwargs
在纯函数上成为 arr 静态参数(fn(value, **kwargs)),在工具/think 上经 std.bind
(julep/std.py:118)做常量合并进流(julep/define.py:946 起的实现与 julep/define.py:1
的 docstring)。秘密形状的常量会被拒——密钥必须留在环境背书的工具里,不许冻进流程。
4. 步骤类型全表(编写面词汇表)
@flow 里能追加的步骤,与中间图 StepKind(julep/dag.py:37)一一对应:
| 写法 | StepKind | 编成的 IR | 语义一句话 |
|---|---|---|---|
tool(h, retries=2, timeout_s=5) | TOOL | PRIM(CallStep) | 调工具(策略进 Ann) |
think(reasoner, h, timeout_s=10) | THINK | PRIM(ThinkStep) | 一次模型调用 |
pure_fn(h)(注册 @pure) | PURE | PRIM/ARR | 确定性变换 |
cond(pred, h, then=..., orelse=...) | COND | ALT(pure=pred) | 谓词二选一 |
switch(selector, h, cases={...}) | SWITCH | ALT(select+cases) | 多路选择 |
each(body, items, max_parallel=, reducer=) | EACH | EACH(body, bound, reducer) | 遍历(可并行、可折叠) |
reschedule(h, delay_s=...) | PASSTHROUGH | 重排执行计划(值不变) | 延迟/让位 |
mcp_tool(server, tool) 在 flow 里调用 | TOOL | PRIM(CallStep(McpTool)) | MCP 工具引用 |
| 内嵌另一个 @flow / BoundFlow | 内联 | 子图复制进当前图 | 复用 |
补充三点:
cond的谓词必须是注册纯函数(_registered_pure_name,julep/define.py:1436)—— 分支判定是运行期数据依赖,必须可哈希冻结。分支臂是@flow函数或BoundFlow, 臂的参数名必须与 subject 的名字绑定(按名捕获,julep/define.py:662conddocstring)。each的 body 参数是位置参数,名字可以与列表 Handle 不同(julep/define.py:1docstring——与 cond 臂的按名绑定刻意区别)。switch_on(julep/define.py:774)是 switch 的变体:选择器直接产出 case 键。
未用参数是 blocking 错误:@flow 函数有参数没用到会拒绝编译——否则会冻结一个
误导性的 API 并让闭包转换有歧义(julep/define.py:1 docstring 末段)。
5. 低层 DSL 与派生组合子:一套积木,两层 API
5.1 低层组合子(dsl.py,直接产 IR)
julep/dsl.py 提供与 11 种 Op 对应的显式构造器:seq(:176)、par(:181)、
fanout(:186)、alt(:195)、each(:233)、iter_up_to(:254)、stage(:263,
计划先行)、app(:272,agent 循环)、sub(:142,子流程)、call(:103)、
native(:93)/mcp(:98)、think(:117)。@flow 前端最终也降维到这层
("Lowering always goes through julep.dag and the compiler",julep/define.py:1)。
5.2 派生组合子(derived.py,"糖")
julep/derived.py 全部不引入新运行时原语(julep/derived.py:1 docstring),只是
产出带特定 Merge 标记或保留工具的普通 IR:
| 组合子 | 行 | 降成什么 | 语义 |
|---|---|---|---|
race(*flows) | julep/derived.py:123 | 带 Merge(kind="race") 的 par 树 | 先成功者胜,取消其余 |
hedge(*flows, hedge_ms) | :133 | Merge("hedge") | 延迟 hedge_ms 后发援兵 |
quorum(*flows, k) | :141 | Merge("quorum") | 等 k 个成功 |
map_n / map_reduce | :154/:163 | par/seq + reducer 纯函数 | 并行映射/折叠 |
vote(reasoners, agg) | :170 | 多 reasoner think + 聚合纯函数 | 投票 |
review(main, reviewer, k) | :183 | 主流程 + 评审 reasoner | 评审 |
recv(channel) / emit(channel, value) | :197/:210 | 保留工具 __recv__/__emit__ | 通道收/发 |
human_gate(prompt, timeout_s) | :221 | 保留工具 __human_gate__ | 等人审批 |
delay(seconds) | :235 | 保留工具 __sleep__ | 持久定时 |
race 族有硬准入:分支效果必须只读、或被断言幂等——check_race_admission
(julep/derived.py:324)作为部署门之一(第 02 章 §4 的契约代数同源)。理由写在
docstring:输家分支会被取消,所以它的效果必须"要么没发生,要么发生了也安全"。
5.3 std 标准库纯函数
@flow 前端与派生层共享一组具名纯函数(julep/std.py):std.merge(:14)、
std.pluck(:34)、std.init/std.assign、std.collect(:66)、std.pack/std.unpack
(:83/:105)、std.bind(:118)、std.record(:135)、std.branch_predicate(:211)等。
它们和用户纯函数走同一条注册/哈希通道——框架自己不吃小灶。
6. 一个组合示例
下面这个流程同时用到工具步、纯函数步、条件与遍历,并演示 Handle 的三种操作:
# 示意,非源码:一个 @flow 的组合用法
@tool(effect="read", idempotent=True)
def lookup(word: str) -> dict: ... # 查词典
@pure("parse")
def parse(hit: dict) -> dict: ... # 解析成要点列表
@flow
def outline(topic: str) -> dict:
hit = lookup(topic) # ① 工具步
parsed = parse(hit) # ② 纯函数步
answer = think(summarizer, parsed) # ③ 模型步(Reasoner 按名注册)
final = cond(
good_enough, answer, # ④ 谓词必须是注册纯函数
then=keep, orelse=rewrite, # 两臂各是一个 @flow
)
bullets = each(format_one, final["items"], max_parallel=4)
return final | {"bullets": bullets} # ⑤ merge(常量记录也行)
跟着数据走:topic ─lookup─▶ hit ─parse─▶ parsed ─think─▶ answer ─cond─▶ final ─each─▶ bullets ─merge─▶ {…, bullets}。每一条箭头在图里都是一步,编译后全部可见、
全部可静态分析——没有运行期才展开的隐藏控制流。
7. 旧 $ 表达式被什么取代了
这是 v1 → v3 最大的语义替换,值得单独讲清:
| v1(已移除) | v3 | |
|---|---|---|
| 表达式写法 | YAML 里 $ len(_.ideas) > 3($ 前缀 Python 表达式) | 具名注册的纯函数:@pure("parse") def parse(hit): ... |
| 求值器 | simpleeval 沙箱(EvalWithCompoundTypes + 白名单) | 解释器直接调注册函数;远程/bundle 场景可选 wasm 沙箱(julep/execution/wasm_executor.py:156 WasmExecutor) |
| 变量上下文 | _/inputs/outputs/steps 注入变量表 | 函数参数就是数据流边(单赋值) |
| 防护 | 白名单函数/标准库子集 + 1 秒超时 + 输入上限 | 编译期:函数必须注册、源码可哈希(source_hash_of,julep/purity.py);运行期:pure 漂移(PureDriftError)在 worker 启动时校验(verifyPures activity,julep/execution/effects.py:1308) |
| 模板字符串 | 裸字符串包成 f-string | 不存在;需要模板用 dotctx 的 Jinja 渲染器(Reasoner.system_render,julep/dotctx.py:172) |
为什么这么换? $ 表达式是"运行期解析的字符串":不可类型检查、不可内容寻址、每次
求值要重建沙箱。具名纯函数是"定义期就冻结的代码":源码哈希进部署物,重放时可校验"这个
函数就是当时那个函数";verifyPures 在 worker 启动时把注册表与冻结哈希对账,对不上即
PureDriftError——把"表达式安全"从运行期围栏升级成了供应链级完整性。
wasm 执行器(julep[\'wasm\'] extra)则给 bundle 来源的纯函数一个真隔离沙箱(非同语言
执行、无宿主 API),这是 simpleeval 做不到的隔离强度。
8. 巧妙之处(可借鉴的技术)
-
定义即构造。 用"执行用户函数一次"来收集结构,而不是写解析器/新语法——Python 的 类型系统、IDE、调试器全部白拿,产物仍是可冻结的数据。见
julep/define.py:631。 -
教学式报错。
Handle把所有非法操作(真值、迭代、相等)拦截下来并指路到正确 组合子,报错带SourceSpan。定义期诊断是产品特性,不是日志。见julep/define.py:437-470。 -
单赋值命名 + AST 溯源。 步骤名从整个函数的 AST 源码推导,REPL/exec 环境给确定性 兜底名;名字即身份,重命名即重构。见
julep/define.py:98(_SourceMap)。 -
派生组合子零新原语。 race/hedge/quorum 只是"带 Merge 标记的 par";人闸/睡眠/通道 只是"保留名工具"。运行时永远只需理解 11 种 Op。见
julep/derived.py:1。 -
std 库与用户代码同规。 merge/pluck/record 等框架自带纯函数也走注册/哈希通道, 部署物里框架代码和用户代码一视同仁。见
julep/std.py:14。
9. 边界与局限(诚实)
- Handle 表达能力是刻意的五个操作。 想在流程里做复杂的数据变换,必须写成注册纯函数
——不能像 v1 那样随手写
$表达式。这是"可冻结"换来的约束。 - 谓词/选择器/reducer 必须是注册纯函数,闭包、lambda、未注册函数都会在定义期被拒
(
julep/define.py:1436)。 - 捕获必须是 canonical JSON 或 Handle。 非法/秘密形状/超大捕获都会报错
(
_validate_json_value/_secret_path,julep/define.py:1581/:1616)。 - 未用参数、非 Handle 返回、元组解包目标等都是 blocking 错误(
julep/define.py:1docstring 列表)——比一般框架严格,理由是"否则冻结的 API 是误导"。 - 每个臂/body 都会被内联进父图(子图复制,
_copy_graph_with_external_renames,julep/define.py:1169)。复用大段逻辑请用sub(独立部署物/子工作流)而非内联 flow。
10. 代码地图(导航索引)
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| @flow 装饰器 | julep/define.py:631 | flow / FlowDef(:475) / BoundFlow(:598) |
| Handle(数据占位符) | julep/define.py:383 | Handle.__or__ / __getitem__ / __bool__ |
| 步骤追加公共入口 | julep/define.py:946 | _append_step |
| 作者期分派 | julep/define.py:892 | apply_if_authoring |
| 控制组合子 | julep/define.py:662 | cond(:662) / switch(:713) / switch_on(:774) / each(:804) / reschedule(:842) |
| 模型调用步 | julep/define.py:644 | think |
| 源码定位 | julep/define.py:98 | _SourceMap / SourceSpan(julep/ir.py:636) |
| 常量/秘密校验 | julep/define.py:1581 | _validate_json_value / _secret_path(:1616) |
| 低层 DSL | julep/dsl.py:176 | seq(:176) / par(:181) / alt(:195) / each(:233) / iter_up_to(:254) / stage(:263) / app(:272) / sub(:142) / call(:103) |
| 派生组合子 | julep/derived.py:123 | race / hedge(:133) / quorum(:141) / map_reduce(:163) / vote(:170) / review(:183) |
| 通道/人闸/延迟 | julep/derived.py:197 | recv / emit(:210) / human_gate(:221) / delay(:235) |
| race 准入 | julep/derived.py:324 | check_race_admission |
| std 纯函数库 | julep/std.py:14 | std_merge / std_pluck(:34) / std_record(:135) / std_bind(:118) |
| 中间图 | julep/dag.py:37 | StepKind / StepNode(:61) / Graph(:90) / compile(:306) |
| 纯函数注册 | julep/purity.py:62 | pure / register_pure(:87) / Pure(:31) |
| 纯函数漂移校验 | julep/execution/effects.py:1308 | verifyPures |
| wasm 沙箱执行 | julep/execution/wasm_executor.py:156 | WasmExecutor |
| MCP 工具引用步 | julep/mcp_step.py:33 | mcp_tool / McpToolStep(:14) |
相关章节: 图怎么被执行/重试/截断 → 02 任务执行引擎; 工具与 MCP 面怎么冻结 → 04 工具与集成;多轮会话怎么写 → 05 会话与上下文。