工作流步骤类型与 `$` 表达式求值

30 秒导读: 在 Julep 里,一个"任务(task)"就是一串步骤(steps)。每个步骤是一个带 kind_ 字段的 YAML 对象(evaluate/prompt/if_else/foreach…);步骤里几乎每个值都可以写成 $ 打头的 Python 表达式,由一个受限沙箱(simpleeval)现算。本章讲清两件事:开发者能用哪些步骤,以及**$ 表达式到底怎么被求值**。

本章聚焦 DSL 本身——步骤的语义和表达式的求值规则。任务作为 Temporal 工作流"怎么一步步跑起来、状态怎么流转"是第 2 章的主线;工具调用步骤把参数分发到 function/ integration/api_call/system 的细节留给第 4 章。

1. 这是什么(零基础也能懂)

一句话定义: Julep 的任务 DSL 是一种声明式的工作流语言——你用 YAML 列出一连串步骤, Julep 按顺序执行,步骤之间用 $ 表达式传数据、做判断、搭循环。

它解决什么问题: 假设你要让一个 agent 干一件多步骤的活——"先让模型出主意 → 判断主意够不够多 → 不够就循环补齐 → 最后汇总"。你不想为此写一坨 Python、还得自己管重试和断点续跑。DSL 让你只描述 '做什么',执行、重试、持久化交给底层的 Temporal(见第 2 章)。

一个最小任务长这样(YAML,展示 4 种最常见的步骤):

name: 头脑风暴
main:
  # ① prompt:调模型
  - prompt:
      - role: user
        content: 给"$ _.topic"想 3 个点子,每行一个
    unwrap: true                      # 只要模型吐出的纯文本

  # ② evaluate:把纯文本切成列表
  - evaluate:
      ideas: "$ _.split(NEWLINE)"

  # ③ if_else:点子够不够 3 个?
  - if: "$ len(_.ideas) >= 3"
    then:
      evaluate: { result: "$ _.ideas" }
    else:
      evaluate: { result: "$ _.ideas + ['(需要补充)']" }

  # ④ foreach:逐个加序号
  - foreach:
      in: "$ _.result"
      do:
        evaluate: { line: "$ '- ' + _" }

读这段的直觉,只需记住三点:

每个列表项是一个步骤,步骤类型由它的关键字段决定(prompt / evaluate / if / foreach)。
$ ... 是表达式,里面是真的 Python;不带 $ 的字符串是普通文本(或模板)。
_(下划线)= 上一步的输出,是步骤间传值的主通道。

一句话类比: 把它想成一个"带 $ 公式的 Excel 宏"——每个单元格(步骤字段)要么是常量,要么是一条 $ 公式;公式能引用"上一格的值"(_)和一些内置函数,但被关在沙箱里,不能碰文件系统或网络。

2. 顶层全景(步骤怎么被执行)

Julep 把 15 种步骤分成两大类,分别跑在两个不同的地方。这是理解本章的主骨架:

类别	跑在哪	特点	代表步骤
控制流步骤	Temporal 工作流内(确定性、可回放)	只做编排:判断、分支、循环、传值	`if_else` `switch` `foreach` `map_reduce` `parallel` `sleep` `return` `log` `error` `get` `set` `yield` `wait_for_input`
副作用步骤	Temporal activity里(可失败、可重试)	会碰外部世界:调模型、调工具、写状态	`prompt`(→ activity) `tool_call`(→ activity)

为什么这样分?Temporal 要求工作流代码确定性(能回放),所以任何"碰外部世界"的事(网络、随机、时间)都必须外包给 activity。表达式求值本身也算副作用——它跑成一个叫 base_evaluate 的 activity。

怎么读下面这张图: 从上到下是一个步骤的一生。左边是"算表达式"的公共通道,右边是分类派发。

                     TaskExecutionWorkflow.run()
                     src/.../task_execution/__init__.py:738
                                 │
                                 ▼
             ┌───────────────────────────────────────┐
             │ 这一步是 PromptStep 吗?(STEP_TO_ACTIVITY) │
             └───────────────────────────────────────┘
                   是 │                    │ 否(其余所有步骤)
                     ▼                     ▼
        execute_activity(prompt_step)   eval_step_exprs(step)   :142
        (副作用:调 LLM)                  │  先把该步骤里的 $ 表达式算出来
                     │                     │  (switch/tool_call 在这里就地评估;
                     │                     │   其余委托给 base_evaluate activity)
                     └──────────┬──────────┘
                                ▼
                        handle_step(step)          :718
                        └ 反射到 _handle_<步骤类型名>()
                                ▼
             ┌───────────────────────────────────────┐
             │ _handle_IfElseWorkflowStep / _handle_Foreach │
             │ _handle_MapReduce / _handle_Sleep / …         │
             │ (控制流语义:决定输出 + 下一步去哪)             │
             └───────────────────────────────────────┘
                                ▼
                        transition(...) → 下一步 / 结束

三个关键部件的职责(本章的三个主角):

部件	干什么	在哪个文件
`eval_step_exprs`	每步执行前,把步骤字段里的 `$` 表达式算成实际值	`src/agents-api/agents_api/workflows/task_execution/__init__.py:142` `eval_step_exprs`
`_handle_*` 方法 + `helpers.py`	控制流步骤的语义实现(分支/循环怎么展开)	同上 `handle_step:718`;`.../task_execution/helpers.py`
`base_evaluate` activity	`$` 表达式的真正求值引擎(simpleeval 沙箱)	`src/agents-api/agents_api/activities/task_steps/base_evaluate.py:52` `base_evaluate`

3. 步骤类型全表(DSL 的词汇表)

先给全景,再逐类深入。下表列出全部 15 种步骤:YAML 关键字段、干什么、实现落在哪。

步骤(`kind_`)	YAML 关键字段	语义一句话	实现位置(`_handle_*` / activity)
`evaluate`	`evaluate: {k: $expr}`	算一组表达式,产出一个 dict	`_handle_EvaluateStep:356`
`set`	`set: {k: $expr}`	算表达式并写入用户状态	`_handle_SetStep:540`
`get`	`get: "key"`	从状态读一个 key(见下方局限)	`_handle_GetStep:551`
`log`	`log: "$expr"`	记一条日志,输出透传当前输入	`_handle_LogStep:210`
`return`	`return: {k: $expr}`	结束工作流并返回值	`_handle_ReturnStep:230`
`error`	`error: "msg"`	主动抛错,转 error 态	`_handle_ErrorWorkflowStep:367`
`sleep`	`sleep: {seconds/minutes/hours/days}`	睡一段时间(durable)	`_handle_SleepStep:336`
`if_else`	`if: $cond` / `then` / `else`	条件二选一分支	`_handle_IfElseWorkflowStep:268`
`switch`	`switch: [{case,then}]`	多路 case,命中第一个	`_handle_SwitchStep:245`
`foreach`	`foreach: {in: $list, do: step}`	顺序遍历列表	`_handle_ForeachStep:286`
`map_reduce`	`over` / `map` / `reduce` / `initial` / `parallelism`	map 每个元素再 reduce 汇总	`_handle_MapReduceStep:302`
`parallel`	`parallel: [...]`	并行多个分支	`_handle_ParallelStep:562`(未实现)
`yield`	`workflow` / `arguments`	跳到本任务里另一个子工作流	`_handle_YieldStep:384`
`wait_for_input`	`wait_for_input: {info: $expr}`	暂停,等外部输入喂回	`_handle_WaitForInputStep:419`
`prompt`	`prompt: [...]` / `unwrap` / `auto_run_tools`	调 LLM(副作用)	activity `prompt_step` + `_handle_PromptStep:437`
`tool_call`	`tool: name` / `arguments`	调工具(副作用)	`_handle_ToolCallStep:572`(分发细节见第 4 章)

每个步骤还都有一个可选的 label 字段,用来给这一步命名,方便后面用 steps['标签'].output 引用 (src/agents-api/agents_api/autogen/Tasks.py:352 每个 step 类都带 label)。

下面按"控制流"和"副作用"两类,挑有代表性的深入。

4. 控制流步骤:在工作流里编排(a 类)

这类步骤不碰外部世界,只做编排。它们的实现是 _handle_* 方法,复杂分支/循环再委托给 helpers.py 里的 execute_* 函数。

4.1 一个共同的套路:先算表达式,再执行语义

大多数步骤走同一条路:eval_step_exprs 先把这一步的"表达式字段"抽出来,交给 base_evaluate activity 算成值,存进 self.outcome;然后 _handle_* 拿这个算好的值去做语义。

看 eval_step_exprs 怎么按步骤类型挑出要算的表达式:

# src/agents-api/agents_api/workflows/task_execution/__init__.py:145  (真实源码,节选)
match step_type:
    case ForeachStep(foreach=ForeachDo(in_=in_)):
        expr = in_                 # foreach 要算的是 in: 后面的列表表达式
    case IfElseWorkflowStep(if_=if_):
        expr = if_                 # if_else 要算的是 if: 的条件
    case ReturnStep(return_=return_):
        expr = return_
    case EvaluateStep(evaluate=evaluate):
        expr = evaluate            # evaluate 要算的是整个 dict
    case MapReduceStep(over=over):
        expr = over
    ...
if expr is not None:
    output = await base_evaluate_activity(expr, self.context)

要点: if_else 到这一步只是把 if: 条件算成一个布尔值;真正"选 then 还是 else 分支"是后面 _handle_IfElseWorkflowStep 的事。switch 和 tool_call 例外——它们的表达式在 eval_step_exprs 里就地循环评估(switch 逐个 case 算到第一个为真,:163-169)。

4.2 `if_else` / `switch`:分支

分支的思路很巧:不在当前工作流里"跳行",而是把选中的分支包成一个临时子工作流,continue-as-child 跳进去。这样每个分支都有独立的执行作用域,和 Temporal 的持久化模型对齐。

# src/agents-api/agents_api/workflows/task_execution/helpers.py:165  execute_if_else_branch(节选)
chosen_branch = then_branch if condition else else_branch
if chosen_branch is None:
    chosen_branch = EvaluateStep(evaluate={"output": "_"})   # else 缺省 = 透传输入
# 把选中的分支塞进一个临时命名的子工作流,再作为子工作流跳进去
if_else_task.workflows = [Workflow(name=if_else_wf_name, steps=[chosen_branch]), *...]
return await continue_as_child(if_else_execution_input, if_else_next_target, current_input, ...)

switch 同理(helpers.py:125 execute_switch_branch),只是分支从 switch 列表里按命中的 index 取(__init__.py:253 _handle_SwitchStep;负 index 报错)。

4.3 `foreach`:顺序循环

foreach 就是对 in: 算出的列表逐个元素跑一遍 do: 步骤,把每次输出收集成列表。

# src/agents-api/agents_api/workflows/task_execution/helpers.py:210  execute_foreach_step(节选)
for i, item in enumerate(items):
    # 每个元素都开一个子工作流,把 item 作为它的输入(_)
    result = await continue_as_child(foreach_execution_input, foreach_next_target, item, ...)
    if result.returned:            # 循环体里 return 会短路整个循环
        return result
    results.append(result.state.output)
return WorkflowResult(state=PartialTransition(output=results))

关键细节: 每次迭代都是一个独立子工作流,item 成为该子工作流的 _;这保证长循环也能被 Temporal 持久化、崩溃可续。foreach 是顺序的——要并行请用 map_reduce 带 parallelism。

4.4 `map_reduce`:map 每项再折叠,含并行变体

map_reduce = 对 over 列表里每个元素跑 map 步骤,再用 reduce 表达式把结果折叠成一个累加值。 _handle_MapReduceStep 按 parallelism 分两条路(__init__.py:309):

`parallelism`	走哪个函数	行为
`None` 或 `1`	`execute_map_reduce_step:256`	顺序:一个个 map,每步做一次 reduce
`> 1`	`execute_map_reduce_step_parallel:312`	分批并行:每批 `asyncio.gather`,批内并发

顺序版的 reduce 默认是"把结果 append 进列表":

# src/agents-api/agents_api/workflows/task_execution/helpers.py:270
reduce = "$ results + [_]" if reduce is None else reduce
#          ^ results = 累加器,_ = 当前 map 输出

并行版更巧妙:它把 reduce 表达式改写成一个 lambda,再用内置的 reduce() 函数一次折叠一整批 (helpers.py:350-358)。这是为了让"逐项折叠"的语义在"一批结果"上也成立——一个很典型的把标量 reducer 提升到批处理的技巧。

4.5 未实现 / 有局限的控制流步骤(诚实边界)

DSL 里有几个步骤是占位或半成品,写任务时要避开:

步骤	现状	证据
`parallel`	未实现,直接抛错	`__init__.py:562` `_handle_ParallelStep` 抛 `"Not implemented"`;顶部注释 `ParallelStep # ❌`
`get`	只从 Temporal memo 读,`get_value_step` activity 本身是 stub	`__init__.py:551` 读 `workflow.memo_value`;`activities/task_steps/get_value_step.py:16` 抛 `NotImplementedError`
`prompt` 里的非 function 工具调用	integration/api_call/system 类型在 prompt 自动回灌里 `NotImplementedError`	`__init__.py:509-534`

顶部那段"支持步骤"注释(__init__.py:88-105)是一张手写的实现状态表,✅/🟡/❌ 标了各步骤完成度——写任务前值得对照一眼。

5. 副作用步骤:跑成 Temporal activity(b 类)

这类步骤会碰外部世界,所以被外包成 activity(可失败、可重试、有超时和心跳)。它们的代码在 activities/task_steps/ 下。

文件	角色	关键符号
`base_evaluate.py`	表达式求值引擎(所有 `$` 最终跑这)	`base_evaluate:52`
`prompt_step.py`	调 LLM:算 prompt、格式化工具、调 LiteLLM、整理响应	`prompt_step:31`
`tool_call_step.py`	构造 tool_call 对象的纯函数(不含分发)	`generate_call_id:10`、`construct_tool_call:22`
`transition_step.py`	把一次状态转移写进数据库	`transition_step:20`
`raise_complete_async.py`	把步骤挂起为"异步完成",等外部回填	`raise_complete_async:14`
`get_value_step.py`	从上下文取值(stub,未实现)	`get_value_step:12`

三条要点,帮你把这几个文件串起来:

prompt 是唯一进 STEP_TO_ACTIVITY 表的步骤(__init__.py:108)——因为只有它默认就要在执行前跑一个重活(调模型)。其余步骤先在工作流里评估表达式,需要时才临时派发 activity。
tool_call_step.py 只有两个纯函数,不做真正的调用分发。真正把 tool_call 分派到 function/integration/api_call/system 的逻辑在 _handle_ToolCallStep(__init__.py:572)——那属于第 4 章。
raise_complete_async 是 wait_for_input 和 function 工具调用的公共暂停机制:它先写一条 type="wait" 的转移、捕获 Temporal 的 task token,然后调 activity.raise_complete_async() 让这一步 挂起最长 31 天等外部喂回(raise_complete_async.py:22-37,配合 __init__.py:427 的 31 天超时)。

6. `$` 表达式求值:受限 Python 沙箱(本章核心)

现在钻进最底层:一条 $ len(_.ideas) >= 3 到底怎么变成 True。整条链路是 base_evaluate(activity)→ evaluate_expressions → _recursive_evaluate → simpleeval。

6.1 入口:准备变量,递归下钻

base_evaluate activity 先把当前步骤能看到的所有变量装进一个 dict,再交给 evaluate_expressions:

# src/agents-api/agents_api/activities/task_steps/base_evaluate.py:52  base_evaluate(节选)
values = values or {}
if context:
    # 把 _、inputs、outputs、state、steps、secrets 等注入变量表(最多 50 步历史)
    values.update(await context.prepare_for_step(limit=50))
return evaluate_expressions(exprs, values=values, extra_lambda_strs=extra_lambda_strs)

prepare_for_step(common/protocol/tasks.py:300)决定了表达式里能引用哪些变量:

变量名	含义
`_`	当前步骤的输入(上一步输出)
`inputs`	按序号索引的历史各步输入
`outputs`	按序号索引的历史各步输出
`steps`	步骤对象(可用 `label` 命名索引)
`state`	累积的用户状态(`set` 步骤写的)
`secrets`	开发者配置的密钥

表达式可以是字符串、列表、或 dict(比如 evaluate: {a: $x, b: $y})。_recursive_evaluate 就负责递归下钻到每个叶子字符串再逐个求值(expressions.py:23,list/dict 分支)。

6.2 三条改写规则:`$` 前缀、`{{}}` 兼容、f-string 包裹

单个字符串怎么被解释,取决于它的形状。这是 DSL 语义最容易踩坑的地方,分三条规则依次判断:

规则一 —— backwards_compatibility(老语法归一化)。 先把历史上的 {{var}} 模板语法翻译成 $ 语法:

# src/agents-api/agents_api/common/utils/task_validation.py:49  backwards_compatibility(节选)
if "{{" in expr:
    return "$ f'''" + expr.replace("{{", "{").replace("}}", "}") + "'''"
#   把 "hello {{name}}" 变成  $ f'''hello {name}'''

规则二 —— $ 前缀 = 当 Python 算。 去掉 $ 后,剩下的原样交给 simpleeval 当表达式求值:

# src/agents-api/agents_api/activities/task_steps/base_evaluate.py:26  _recursive_evaluate(节选)
if isinstance(expr, str) and expr.startswith("$ "):
    expr = expr[1:].strip()          # "$ len(_) > 3"  →  求值 len(_) > 3
else:
    expr = f"f'''{expr}'''"          # 没有 $ 前缀 → 当 f-string 模板
return evaluator.eval(expr)

规则三 —— 没有 $ 前缀 = 当 f-string 模板。 普通字符串被包进 f'''...''' 再求值,所以 "你好 {_.name}" 里的 {...} 会被插值。这就是为什么 prompt 内容里能直接写 {_.topic}。

一句话记忆: $ 开头 → 整体是代码;不带 $ → 整体是模板字符串,只有 {…} 里是代码。

6.3 沙箱:simpleeval + 白名单

真正求值的是 simpleeval 的 EvalWithCompoundTypes。get_evaluator 把"能用什么"钉死成白名单 ——这是安全边界:

# src/agents-api/agents_api/common/utils/evaluator.py:639  get_evaluator(节选)
evaluator = EvalWithCompoundTypes(
    names=names | stdlib | constants,                 # 变量 + 受限标准库 + 常量
    functions=ALLOWED_FUNCTIONS | (extra_functions or {}),   # 只有白名单函数可调
)
evaluator.TIMEOUT = 1.0                               # 单次求值 1 秒超时

沙箱由四层白名单/限制构成:

层	内容	位置
允许的函数	`len/sum/map/zip/range/reduce`、`load_json`、`nlp`、`search_regex`(用 re2 防 ReDoS)…	`evaluator.py:319` `ALLOWED_FUNCTIONS`
受限标准库	`re/json/yaml/math/datetime/base64/random/urllib/csv…` 的安全子集	`evaluator.py:607` `stdlib`
常量	`NEWLINE`、`true`/`false`/`null`	`evaluator.py:622` `constants`
硬限制	变量数 ≤ 10000、字符串 ≤ 1MB、range ≤ 100 万、求值 1 秒超时	`evaluator.py:32-34`、`:635`、`:645`

变量表在传给 simpleeval 前会被包成 Box(expressions.py:96),这样表达式里可以用点号访问嵌套 dict(_.customer.name 等价于 _['customer']['name']),读起来更顺。

6.4 `extra_functions`:并行 reduce 怎么注入自定义 lambda

evaluate_expressions 还能接受额外的 lambda 字符串并注入沙箱——但会先校验它必须以 lambda 开头、且能被 ast.parse 解析(expressions.py:75-90)。§4.4 的并行 map_reduce 正是靠这个把改写后的 reducer_lambda 塞进沙箱、再用内置 reduce() 折叠一整批。

6.5 求值失败会怎样

任何叶子表达式抛异常,都会被包成 EvaluateError(带上原表达式和当时的变量名),再向上抛 (base_evaluate.py:32-36)。因为 base_evaluate 是 activity,这个错会沿 Temporal 的 ActivityError 链回到工作流,由第 2 章的错误转移逻辑处理成 error 态。helpers.py:68 的 base_evaluate_activity 会先解包嵌套的 ActivityError 再重抛,让真正的错误信息浮上来。

7. 把它们串起来:一个组合示例

下面这个任务同时用到 prompt(副作用)、evaluate、if_else、foreach(控制流),并演示了 §6 的三条表达式规则。注释标出每处 $ 的求值路径。

# 示意任务(简化):从一个主题生成要点并规范化输出
name: outline
main:
  # [prompt 步骤 → prompt_step activity] content 里的 {_.topic} 走 f-string 规则(规则三)
  - prompt:
      - role: user
        content: "给主题 {_.topic} 列出要点,每行一个"
    unwrap: true                         # 只取模型纯文本 → 成为下一步的 _

  # [evaluate 步骤 → base_evaluate] $ 开头 = 当 Python 算(规则二)
  - evaluate:
      points: "$ [p.strip() for p in _.split(NEWLINE) if p.strip()]"

  # [if_else 步骤] if: 先被 eval_step_exprs 算成布尔,再由 _handle_IfElseWorkflowStep 选分支
  - if: "$ len(_.points) > 0"
    then:
      evaluate: { points: "$ _.points" }
    else:
      # error 步骤:主动失败
      error: "模型没有产出任何要点"

  # [foreach 步骤] in: 算出列表,do: 对每个元素跑一次;元素成为子工作流的 _
  - foreach:
      in: "$ _.points"
      do:
        evaluate:
          numbered: "$ '• ' + _"         # 这里的 _ 是单个要点,不是整列表

  # [return 步骤] 结束并返回
  - return:
      outline: "$ _"

跟着数据走一遍:

_.topic ──prompt──▶ 多行文本 ──evaluate──▶ {points:[...]} 
        ──if_else(len>0? then/else)──▶ {points:[...]} 
        ──foreach(每个 point → '• '+_)──▶ [{numbered:...}, ...] 
        ──return──▶ {outline: [...]}

注意三处 $ 语义的差别:prompt 的 {_.topic} 是模板插值(规则三);evaluate/if/foreach 里的 $ ... 是整段 Python(规则二);foreach.do 内的 _ 已经换成了单个元素(§4.3 的子工作局作用域)。

8. 巧妙之处(可借鉴的技术)

表达式求值也是 activity。 把 $ 求值外包成 base_evaluate activity(而非在工作流里直接 eval), 既满足 Temporal 的确定性要求,又让求值有独立超时/重试。base_evaluate.py:50。
分支/循环 = 临时子工作流。 不在当前工作流"跳行",而是把选中的分支包成命名子工作流再 continue-as-child 进去。天然对齐 Temporal 的持久化与作用域模型。helpers.py:165/210。
两套改写让一种字段吃三种写法。 $ 代码 / {{老模板}} / 裸模板,靠 backwards_compatibility
- f-string 包裹统一收敛到 simpleeval。开发者写起来自由,引擎只有一条求值路径。 task_validation.py:30、base_evaluate.py:26-30。
标量 reducer 提升到批。 并行 map_reduce 把 reduce 表达式改写成 lambda 注入沙箱,再用内置 reduce() 一次折叠一批——让顺序语义在并行下依然成立。helpers.py:350-358。
re2 而非 re。 正则函数用 Google 的 re2,天然防 ReDoS(灾难性回溯);这是沙箱安全的一个细节。 evaluator.py:349-350。

9. 边界与局限(诚实)

parallel 步骤未实现——_handle_ParallelStep 直接抛错(__init__.py:562)。要并行只能用 map_reduce 带 parallelism。
get 步骤是半成品——工作流里只从 Temporal memo 读值(__init__.py:557),而 get_value_step activity 本身是 NotImplementedError(get_value_step.py:16)。
prompt 的自动工具回灌只支持 function 类型——integration/api_call/system 类型的自动回灌都抛 NotImplementedError(__init__.py:509-534)。
沙箱有硬上限:变量数 10000、字符串 1MB、range 100 万、单次求值 1 秒(evaluator.py:32-34/645); 历史步骤默认最多注入 50 条(base_evaluate.py:65)。超限即报错,不是静默截断。
map_reduce 的 map 只支持单步——helpers.py:284 有 NOTE: Step needs to be refactored to support multiple steps。

10. 代码地图(导航索引)

用符号名 grep 比行号更抗漂移。

主题	文件路径	符号名
工作流主循环 / 步骤派发	`src/agents-api/agents_api/workflows/task_execution/__init__.py`	`TaskExecutionWorkflow.run`、`handle_step`
步骤类型 → activity 映射	同上	`STEP_TO_ACTIVITY`
执行前评估步骤表达式	同上	`eval_step_exprs`
各控制流步骤语义	同上	`_handle_IfElseWorkflowStep`、`_handle_ForeachStep`、`_handle_MapReduceStep`、`_handle_SwitchStep`、`_handle_ParallelStep`
分支/循环展开	`src/agents-api/agents_api/workflows/task_execution/helpers.py`	`execute_if_else_branch`、`execute_switch_branch`、`execute_foreach_step`、`execute_map_reduce_step`、`execute_map_reduce_step_parallel`
表达式求值 activity	`src/agents-api/agents_api/activities/task_steps/base_evaluate.py`	`base_evaluate`、`_recursive_evaluate`
表达式引擎(递归 + Box)	`src/agents-api/agents_api/common/utils/expressions.py`	`evaluate_expressions`、`_recursive_evaluate`
simpleeval 沙箱与白名单	`src/agents-api/agents_api/common/utils/evaluator.py`	`get_evaluator`、`ALLOWED_FUNCTIONS`、`stdlib`、`constants`
老语法归一化 / 校验	`src/agents-api/agents_api/common/utils/task_validation.py`	`backwards_compatibility`、`validate_py_expression`
表达式可见变量的准备	`src/agents-api/agents_api/common/protocol/tasks.py`	`StepContext.prepare_for_step`
LLM 调用步骤	`src/agents-api/agents_api/activities/task_steps/prompt_step.py`	`prompt_step`
tool_call 对象构造	`src/agents-api/agents_api/activities/task_steps/tool_call_step.py`	`construct_tool_call`、`generate_call_id`
异步暂停(等外部输入)	`src/agents-api/agents_api/activities/task_steps/raise_complete_async.py`	`raise_complete_async`
状态转移写库	`src/agents-api/agents_api/activities/task_steps/transition_step.py`	`transition_step`、`original_transition_step`
步骤类型 Pydantic 模型	`src/agents-api/agents_api/autogen/Tasks.py`	`EvaluateStep`、`IfElseWorkflowStep`、`ForeachDo`、`SwitchStep`、`CaseThen`

相关章节: 步骤如何被调度/持久化/状态流转 → 02 任务执行引擎; tool_call 参数如何分发到 function/integration/api_call/system → 04 工具与集成; 表达式里能读到的记忆/文档从哪来 → 05 记忆与混合检索。

1. 这是什么(零基础也能懂)​

2. 顶层全景(步骤怎么被执行)​

3. 步骤类型全表(DSL 的词汇表)​

4. 控制流步骤:在工作流里编排(a 类)​

4.1 一个共同的套路:先算表达式,再执行语义​

4.2 if_else / switch:分支​

4.3 foreach:顺序循环​

4.4 map_reduce:map 每项再折叠,含并行变体​

4.5 未实现 / 有局限的控制流步骤(诚实边界)​

5. 副作用步骤:跑成 Temporal activity(b 类)​

6. $ 表达式求值:受限 Python 沙箱(本章核心)​

6.1 入口:准备变量,递归下钻​

6.2 三条改写规则:$ 前缀、{{}} 兼容、f-string 包裹​

6.3 沙箱:simpleeval + 白名单​

6.4 extra_functions:并行 reduce 怎么注入自定义 lambda​

6.5 求值失败会怎样​

7. 把它们串起来:一个组合示例​

8. 巧妙之处(可借鉴的技术)​

9. 边界与局限(诚实)​

10. 代码地图(导航索引)​