本篇对应
app/workflows.py（559 行，全仓库最大模块）与app/safe_eval.py（300 行）。这是平台的"业务流程自动化"核心，也是工程含金量最高的部分：怎么让用户用 JSON 描述一条 AI 流水线，怎么防住"用户能写代码"这个天生的注入风险。
一、问题定义：什么是工作流，为什么是 DAG
运营的真实需求："每周把销售 CSV 丢进去，自动校验数据 → 算汇总 → AI 出报告"。这是多步骤流程，不是一次对话。把流程写成代码？每换一个流程改一次代码，01 篇"助手是配置不是代码"的教训在这里重演——流程也应该是配置。
司南的答案：JSON 定义的 DAG（有向无环图）。七种节点类型：
| 类型 | 职责 | 输出示例 |
|---|---|---|
start |
起点，输出 = 本次运行输入 | {rows: [...], params: {...}} |
llm |
调 Provider 生成文本 | {text, provider, model} |
knowledge_search |
在指定知识库检索 | {results: [...], count} |
http_api |
调外部 HTTP API | {status_code, url, data} |
code |
受限 Python 表达式计算 | {result: ...} |
condition |
布尔判断，出边按 true/false 分支 | {result: true} |
end |
输出运行结果 | {result: ...} |
边（edges）连接节点，可带 condition。为什么必须无环：A→B→A 的环会让执行器死循环。保存时就拒绝环（下文），而不是跑起来后超时。
内置的"销售数据分析"工作流长这样（真实定义摘自 SALES_ANALYSIS_DEFINITION）：
{
"nodes": [
{"id": "start", "type": "start"},
{"id": "check", "type": "condition",
"config": {"expression": "len(rows) > 0", "vars": {"rows": "{{start.rows}}"}}},
{"id": "summary", "type": "code", "config": {"expression": "{...汇总计算...}", "vars": {"rows": "{{start.rows}}"}}},
{"id": "report", "type": "llm",
"config": {"provider": "mock", "prompt": "以下是销售数据汇总JSON，请生成分析报告：\n{{summary.result}}"}},
{"id": "empty_end", "type": "end", "config": {"output": {"error": "CSV 中没有数据行，请检查上传文件"}}},
{"id": "end", "type": "end", "config": {"output": {"summary": "{{summary.result}}", "report": "{{report.text}}"}}}
],
"edges": [
{"from": "start", "to": "check"},
{"from": "check", "to": "summary", "condition": "true"},
{"from": "check", "to": "empty_end", "condition": "false"},
{"from": "summary", "to": "report"},
{"from": "report", "to": "end"}
]
}
读法：CSV 无数据 → check 为 false → 走 empty_end 友好报错；有数据 → 汇总 → AI 报告 → 输出。一个完整的数据分析流水线，纯 JSON 描述。
二、保存期校验：让错误在配置时就暴露
validate_definition(defn) 五道关（保存/执行前都跑）：
- 定义必须是 JSON 对象，
nodes非空； - 每个节点有字符串 id、id 不重复、类型在七种白名单内；
- 有且仅有一个 start；
- id 不得为保留字
input——执行器用context["input"]存运行输入，同名节点会静默覆盖它（这种"看起来能跑但语义被污染"的 bug 最阴险，所以要在保存期点名拒绝）； - 边引用的节点必须存在；Kahn 拓扑排序检测环。
# Kahn 算法：反复摘除入度为 0 的节点；摘不完说明有环
q = deque([i for i in ids if indeg[i] == 0])
seen = 0
while q:
cur = q.popleft()
seen += 1
for nxt in adj[cur]:
indeg[nxt] -= 1
if indeg[nxt] == 0:
q.append(nxt)
if seen != len(ids):
raise WorkflowDefinitionError("工作流存在环，必须是 DAG")
第二遍 validate_expressions：静态检查所有 code/condition 表达式与边上的条件（复用 safe_eval 的 check_expression，只审不求值）。保存时就返回 400 并点名"节点 X 的表达式不安全"——不必跑一遍才发现。执行期的白名单依旧独立生效，双保险：手工写库或历史遗留的坏定义，照样只能得到 failed 运行。
三、模板渲染：{{节点id.路径}} 让数据在节点间流动
TEMPLATE_RE = re.compile(r"\{\{\s*([\w\.\[\]]+)(?:\|([^{}]*?))?\s*\}\}")
每个节点执行前，其 config 做模板渲染，{{路径}} 从 context 取值（context = {"input": 运行输入, "<节点id>": 该节点输出}）。三条精细规则：
- 类型保留：整串恰好是一个模板时（如
"vars": {"rows": "{{start.rows}}"}），返回原始 dict/list——不做 JSON 字符串化，下游code节点才能直接for r in rows； - 字符串插值：
"分析：{{summary.result}}"里嵌入 dict 时自动json.dumps； - 默认值语法
{{路径|默认值}}：路径不存在时用默认值。内置广告工作流靠{{input.params.campaign|双11大促}}实现"不传参数也能跑"。默认值做轻量类型推断：5→int、true→bool、[]→真空列表（不是字符串"[]"）；nan/inf这类词刻意不转浮点（"语义与配置者写下的文本不一致"）。没写默认值且路径缺失 → 抛 KeyError 让节点失败——宁可失败，不静默传空。
四、执行器：BFS 按边推进 + 节点级审计
def execute_workflow(defn: dict, run_input: dict, services: dict, on_node=None) -> dict:
validate_definition(defn)
context: dict = {"input": run_input or {}}
queue = deque([start_id])
while queue:
nid = queue.popleft()
...
try:
config = render_template(raw_config, context)
output = _exec_node(nd, config, context, services)
except Exception as exc:
# 节点失败 → 整个运行 failed，记录错误、停止推进
return {"status": "failed", "output": final_output, "error": f"节点 {nid} 执行失败: {error}", "order": order}
context[nid] = output
if on_node: # 每个节点的输入/输出/状态/起止时间 → node_runs 表
on_node(nid, nd["type"], "success", raw_config, output, "", started, _now())
for edge in edges_by_src.get(nid, []):
if _edge_pass(edge, output, context):
queue.append(edge["to"])
要点：
- 失败语义：任一节点失败 → 整个运行 failed，错误定位到节点 id，后续不再推进。没有"跳过失败节点继续跑"的模糊地带——数据流水线的中间值错了，后面的输出全是垃圾；
- 节点级审计：
on_node回调把每个节点的输入/输出/耗时写进node_runs表，前端逐节点展示实时状态与失败高亮； - 僵尸运行兜底：路由层在
execute_workflow外再包一层 try/except——任何意外（定义被手工改坏、写库失败）都会把workflow_runs落成 failed 并写finished_at，"前端才能安全地结束实时轮询，不会出现永远 running 的僵尸记录"（注释原话）。状态机必须有终态，这是全仓库反复出现的设计纪律。
节点执行的两个安全细节
- http_api 的 SSRF 防护：
validate_http_url校验 scheme 仅 http/https、禁止 URL 内嵌凭证、云元数据地址（169.254.0.0/16、metadata.google.internal）任何情况禁止、私网/回环默认允许（内置 demo 依赖 127.0.0.1 回调本机 mock 路由），生产置AIP_HTTP_API_ALLOW_PRIVATE=0收紧。渲染完模板后的最终 URL 也要再过一遍——防止{{input.params.x}}把请求目标带偏。URL 查询串里像密钥的参数（api_key/token/secret）在执行记录里统一脱敏成***。 - timeout 钳制：配置成超大 timeout 会长期占住线程池线程（同步 def 端点），等效 DoS——钳制在 1~60 秒。
五、code 节点沙箱：本仓库安全设计的巅峰
工作流允许用户写 Python 表达式 = 典型代码注入风险点。直接 eval("os.system('rm -rf /')") 显然不行；禁掉 code 节点又砍掉了"汇总计算"能力。safe_eval.py 的答案是两道防线，全程零 eval：
防线一：AST 白名单静态审查
ALLOWED_NODES = {
ast.Expression, ast.Constant, ast.Name, ast.Load, ast.Store,
ast.BinOp, ast.UnaryOp, ast.BoolOp, ast.Compare, ast.IfExp,
ast.Lambda, ast.arg, ast.arguments,
ast.List, ast.Tuple, ast.Dict, ast.Set, ast.Subscript, ast.Slice, ast.Starred,
ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp, ast.comprehension,
ast.Call, ast.keyword,
# 运算符：算术/一元/布尔/比较全套
...
}
def _check(node: ast.AST) -> None:
if type(node) not in ALLOWED_NODES:
raise UnsafeExpressionError(f"禁止的语法: {type(node).__name__}")
if isinstance(node, ast.Name) and node.id.startswith(("_", "__")):
raise UnsafeExpressionError(f"禁止访问名称: {node.id}")
if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Pow):
# 指数必须是小的常量，防止 9**9**9 资源耗尽
...
不检查就拒绝（default deny）。三个经典 Python 沙箱逃逸链在这里全部被堵死：
().__class__.__bases__[0].__subclasses__()—— 逃逸必经的Attribute节点不在白名单，一切x.y属性访问被禁；__import__('os')—— 下划线开头的名称被禁；open('data/platform.db')/eval(...)/exec(...)——__builtins__被置空，只有len/sum/min/max/round/sorted/range/...白名单函数可用（连range都包了资源上限：迭代量 >100 万直接拒绝）。
防线二：手写 AST 解释器，不把表达式交还给 Python
这是与常见实现最大的不同。多数"安全 eval"做到白名单审查后仍用 eval(code, {"__builtins__": {}}) 执行——白名单有疏漏时照样逃逸。司南不用 eval/exec/compile：_eval_node 逐节点手写求值（短路布尔、链式比较、推导式独立作用域、lambda 参数绑定全套语义对齐 Python），"即使白名单出现疏漏，也不存在把表达式交还给 Python 求值器的瞬间"（docstring 原话）。默认分支拒绝未知语法——纵深防御。
保存期与执行期用同一套规则
check_expression（只审不求值）被 validate_expressions 与 safe_eval 共用——"'保存时能不能存'和'执行时会不会跑'用的是同一套规则，不会出现配置期放行、执行期才炸（或反之）的漂移"。TestSafeEval 参数化覆盖 12+ 种危险表达式全拒、正常业务表达式（求和/推导式/带 key 的 max/字典聚合）全过。
六、两个内置工作流的设计意图
- 销售数据分析：演示
condition双分支 +code汇总 +llm报告。code节点的表达式用安全求值器算出总销售额、月度分解、TOP3 产品、环比增长——一个真实的"表达式能写多复杂"的样板； - 广告数据分析：演示
http_api节点。它调用本服务的/api/mock/ads-data模拟接口（不依赖外网、不另开端口），计算 ACOS/CVR 并由 AI 输出优化建议。params传查询参数由 httpx 正确百分号编码（中文、空格、&、#都能原样到达对端），{{input.params.campaign|双11大促}}演示默认值语法。
动手清单
- 实现
validate_definition（含 Kahn 检环）与execute_workflow，先用一个start → llm → end三节点工作流跑通 - 给
check节点接condition双分支，分别上传"有数据的 CSV"与"空 CSV"，验证两条路径 - 【沙箱实验】依次保存含这些表达式的 code 节点，观察哪些被 400、哪些能跑、哪些运行时报错（对应
tests/test_workflows.py::TestSafeEval的用例清单）： -sum(float(r['sales']) for r in rows)（正常业务 → 通过，参看test_safe_expressions_still_save） -().__class__（属性访问 → 保存即 400） -__import__('os')（下划线名 → 保存即 400） -9**9**9（超大幂 → 保存即 400） -sum(range(10**12))（语法合法但 range 迭代量超限 → 运行期拒绝执行，护栏用例test_safe_eval_range_guard） -1 / 0（合法语法 + 运行期除零 → 运行 failed，节点详情记录错误） - 【自锁实验·进阶】把执行路由改成
async def试试（记得先备份）——理解"同步 def 放线程池，事件循环保持空闲接住回环请求"的自锁规避（02 篇伏笔在此兑现）
常见坑：
- 节点 id 用了 input → 校验报错，这是特性不是 bug
- 模板 {{a.b}} 打进长文本中间但路径不存在 → 降级为空串（整串单模板才是报错），注意两种语义的区别
- 表达式里引用了 vars 没声明的变量 → NameError，节点 failed（先检查 vars 映射）
下一篇把所有领域模块装配成完整的 REST API：路由层的设计、SSE 端点、统计接口，以及"薄路由厚领域"的落地。