跳到主要内容

确定性工作流 AgentFlow 与条件系统

30 秒导读： AgentFlow 是 PraisonAI 的第二条编排范式——开发者把执行顺序写死（用 Python 代码或 YAML），框架照单执行，不需要一个 manager LLM 去决定下一步做什么。它给你六种组合积木（顺序、分支、并行、循环、重试、复用），加一套用正则解析字符串表达式的条件引擎来做路由。

本章讲"确定性 DAG / 循环"这条范式本身。第 04 章的 Task / AgentTeam / 三种 Process 是另一条范式（manager 决策式），本章只在需要对照时提它，不重复它的内容。

1. 这是什么（零基础也能懂）​

一句话定义： AgentFlow 是一条你自己写死顺序的智能体流水线——第一步跑完喂给第二步，中间可以插分支、并行、循环，全部由你在代码里排好，框架只负责忠实执行。

解决什么问题 / 给谁用： 假设你要做一个"写文章 → 编辑 → 发布"的固定流程。你已经知道这三步的先后，不需要让一个 LLM 每次去"想一想现在该干嘛"——那样既慢又不可控。AgentFlow 让你像写普通函数调用链一样，把流程钉死下来。

两条范式的分工（这是理解本章的关键）：

范式谁决定"下一步做什么"适合
AgentTeam + Process（第 04 章）一个 manager LLM 在运行时决策步骤不固定、要智能调度的开放任务
AgentFlow（本章）开发者在代码/YAML 里写死步骤已知、要可复现、要可控成本的流水线

用起来什么样： 最小的顺序流水线——把两个 Agent 依次串起来，run() 一把跑完。

# 示意，非源码（真实 API 见 workflows.py:555 AgentFlow / :998 run）
from praisonaiagents import AgentFlow, Agent

flow = AgentFlow(steps=[
    Agent(instructions="写一段关于 AI 的内容"),   # 第 1 步
    Agent(instructions="把上一步的内容润色"),      # 第 2 步：自动收到上一步输出
])
result = flow.run("写 AI")     # 或 flow.start(...)，两者等价
print(result["output"])       # 最后一步的输出

一句话直觉： 把它当成 Unix 管道 a | b | c——数据从左流到右，每一节是一个智能体或一个函数。只不过这条管道还能长出分支（if）、分叉（parallel）和回环（loop/repeat）。

2. 顶层全景（它大概怎么转）​

2.1 核心部件​

部件干什么在哪
AgentFlow工作流主体，持有 steps 列表，run() 驱动执行workflows.py:555
WorkflowContext传给每一步的只读上下文（input、上一步输出、变量表）workflows.py:174
StepResult每一步返回的结果（output、是否提前终止、要写回的变量）workflows.py:182
六种组合原语Route/Parallel/Loop/Repeat/If/Include——控制流积木workflows.py:197–551
条件引擎evaluate_condition() 把 "{{score}} > 80" 求值成布尔conditions/evaluator.py:129
YAMLWorkflowParser把 YAML 文件解析成一个 AgentFlowworkflows/yaml_parser.py:21

2.2 主循环：一个"边走边认类型"的调度器​

AgentFlow.run()（workflows.py:998）的心脏是一个 while i < len(self.steps) 循环（workflows.py:1092）。它逐个取出 steps 里的元素，先看它是不是某种组合原语，是就交给对应的 _execute_* 处理器；否则当成普通单步（Agent / 函数 / Task）执行。

怎么读下图：从上往下是主循环每一轮的判断顺序，命中一种就分派、然后 i += 1 进入下一轮。

run(input)  ──►  while i < len(steps):  取 step = steps[i]
                     │
                     ├─ 是 Route?     ──► _execute_route     （按关键词路由）
                     ├─ 是 Parallel?  ──► _execute_parallel  （线程池并发）
                     ├─ 是 Loop?      ──► _execute_loop      （遍历列表/CSV）
                     ├─ 是 Repeat?    ──► _execute_repeat    （重复到满足 until）
                     ├─ 是 Include?   ──► _execute_include   （嵌入另一个 recipe）
                     ├─ 是 If?        ──► _execute_if        （表达式真→then 假→else）
                     │
                     └─ 都不是 ──► 普通单步：Agent.chat() / 函数 handler / 临时 Agent
                                     └─ 输出写入 previous_output，并存进变量表

依据：分派判断在 workflows.py:1096-1160；普通单步执行在 workflows.py:1162-1483。

三条贯穿全程的暗线（后面各节展开）：

  • 输出即输入。 每步的 output 存进 previous_output，下一步默认自动收到它（workflows.py:1448、变量替换见 _substitute_action_variables workflows.py:107）。
  • 变量表 all_variables 一路累积。 每步结果按 f"{step.name}_output" 或自定义 output_variable 写回（workflows.py:1460），供后续条件/模板引用。
  • 确定性。 走哪条分支、循环几次，只取决于变量值和写定的结构，没有 manager LLM 在中间拍板。

3. 组合原语（六种控制流积木）​

这是本章最核心的部分。AgentFlow 用六个积木拼出任意确定性控制流。每个积木都有两种写法：小写便捷函数（route(...)）和大写数据类（Route(...)）——便捷函数只是薄封装，最终都产出同一个数据类实例。

便捷函数数据类作用类比
route()Route按上一步输出里的关键词跳到不同分支switch/case
parallel()Parallel多步并发跑完再汇总fork/join
loop()Loop对列表 / CSV / 文件逐项执行for-each
repeat()Repeat重复同一步直到条件满足do-while
when() / if_()If求值表达式，真走 then 假走 elseif/else
include()Include把另一个 recipe / workflow 当一步嵌入函数调用

依据：便捷函数 route/parallel/loop/repeat/include 在 workflows.py:361-456；when/if_ 在 workflows.py:506-551；对应数据类在 workflows.py:197/219/250/332/410/461。__init__.py:20-35 把它们全部导出。

3.1 Route —— 按关键词分支​

要解决的小问题： 上一步（通常是个"决策"智能体）输出了一段话，里面含 "approve" 或 "reject"，我想据此走不同后续。

思路： 不做复杂求值，直接在上一步输出文本里搜关键词——但用的是词边界匹配（\bkey\b），避免 "approved" 里的子串误命中。

# 示意，非源码
from praisonaiagents.workflows import route
route({
    "approve": [publish_agent],   # 输出含 "approve" → 走这条
    "reject":  [revise_agent],
    "default": [fallback_agent],  # 都不含 → 兜底
})

真实实现： _execute_route() 遍历 route 键，用 re.search(r'\b'+key+r'\b', prev_lower) 命中即停，没命中走 default（workflows.py:2264-2274）。

关键细节： Route 匹配的是上一步的输出文本，不是变量表里的值——这点和下面的 If 正好相反，别混。

3.2 Parallel —— 并发分叉再汇总​

要解决的小问题： 三个互不依赖的子任务，串行跑太慢。

思路： 用 ThreadPoolExecutor 并发，全部跑完把输出用 \n---\n 拼起来（workflows.py:2442），并存进 parallel_outputs 变量。

两个要注意的设计：

  • 默认限流。 未指定 max_workers 时，并发数 = min(DEFAULT_MAX_PARALLEL_WORKERS, 分支数)，而 DEFAULT_MAX_PARALLEL_WORKERS = 3（workflows.py:42、:2399-2400）——刻意压着，防止 LLM 后端被打到限流。
  • 三种失败策略（on_failure，在 Parallel.__init__ 校验，非法值直接抛错 workflows.py:240-245）：
值语义
partial_ok（默认）某分支失败也继续，把错误当该分支输出
fail_fast首个失败即取消其余分支并抛 WorkflowStepError
fail_all等所有分支跑完，只要有失败就抛

依据：失败分流在 workflows.py:2424-2439。

3.3 Loop —— 逐项遍历​

Loop 对一个列表变量（over="items"）、一个 CSV（from_csv）或文本文件（from_file）逐项执行；可单步也可多步（steps=[...]），可串行也可 parallel=True 并发（workflows.py:250-330）。构造时就校验"不能同时给 step 和 steps、也不能都不给"（workflows.py:313-320）。

3.4 Repeat —— 重复到满足条件（evaluator-optimizer）​

要解决的小问题： "生成 → 自检 → 不够好就再生成"，最多试 N 次。

思路： 反复跑同一步，每轮后调用 until 回调判断是否收敛；until 是个Python 可调用对象（接收 WorkflowContext 返回 bool），到达 max_iterations（默认 10）无条件停。

# 示意，非源码
from praisonaiagents.workflows import repeat
repeat(generator,
       until=lambda ctx: "done" in ctx.previous_result.lower(),
       max_iterations=5)

真实实现： _execute_repeat() 的 for iteration in range(max_iterations) 循环，每轮跑完构造 WorkflowContext 再调 until（workflows.py:2750-2772）。注意：这里 until 是代码回调，不是字符串表达式——和下面 If 的字符串条件不是一套东西。

3.5 If / when —— 表达式真假分支​

要解决的小问题： "如果分数 > 80 就批准，否则打回"——这次判断依据是变量表里的值，不是文本关键词。

思路： when() 是 if_() 的首选别名（两者完全等价，都造 If 对象，workflows.py:506/533）。它拿一个字符串条件 "{{score}} > 80"，交给条件引擎求值成布尔，真走 then_steps 假走 else_steps。

# 示意，非源码
from praisonaiagents.workflows import when
when(condition="{{score}} > 80",
     then_steps=[approve_agent],
     else_steps=[reject_agent])

真实实现： _execute_if() 先 _evaluate_condition(...) 拿布尔，再选分支执行（workflows.py:2810-2817）。条件引擎是本章第 4 节的主角。

3.6 Include —— 模块化复用​

include("wordpress-publisher") 或 include(workflow=other_flow) 把另一个 recipe / workflow 当一步嵌进来，实现模块化组合；构造时强制"recipe 和 workflow 至少给一个"（workflows.py:438-443）。执行时带环检测：同一执行链里重复 include 同名 recipe 会被拦下报 "Circular include detected"（workflows.py:2894-2901）。

3.7 嵌套与深度上限​

原语可以互相嵌套（if 里放 parallel，loop 里放 route……）。统一入口 _execute_single_step_internal()（workflows.py:2045）在递归进入嵌套原语时把 depth+1，一旦 depth > MAX_NESTING_DEPTH（=5，workflows.py:459）就抛错，防止无限递归爆栈（workflows.py:2074-2078）。

_execute_single_step_internal(step, depth)
   │  depth > 5 ?  ──► ValueError("Maximum nesting depth exceeded")
   ├─ Loop     ─► _execute_loop(...,  depth+1)
   ├─ Parallel ─► _execute_parallel(..., depth+1)
   ├─ Route    ─► _execute_route(...,   depth+1)
   ├─ Repeat   ─► _execute_repeat(...,  depth+1)
   ├─ If       ─► _execute_if(...,      depth+1)
   └─ 普通步   ─► normalize → Agent/handler/临时Agent

依据：嵌套分派 workflows.py:2081-2148。

4. 条件求值系统（字符串表达式怎么被安全求值）​

这是与"确定性 DAG"并列的第二个引擎。praisonaiagents/conditions/ 独立成模块，被 AgentFlow（字符串条件）和 AgentTeam（字典路由）共用，做 DRY 复用。

4.1 为什么不用 eval()​

把 "{{score}} > 80" 变成布尔，最偷懒的写法是 eval()——但那等于让外部/LLM 产生的字符串直接当代码跑，是安全黑洞。PraisonAI 的做法是纯正则解析：先做变量替换，再用几条正则去识别"数值比较 / 字符串相等 / 包含 / 布尔"这几类固定模式，永不执行任意代码。

4.2 求值两步走​

evaluate_condition(condition, variables, previous_output)（conditions/evaluator.py:129）：

第一步——变量替换。 用正则 \{\{([^}]+)\}\} 找出所有 {{var}}，从 variables 取值填进去；支持点号嵌套 {{item.score}}（get_nested_value，evaluator.py:167-175）；缺失变量填空串。

第二步——按模式匹配求值（evaluator.py:209-278），依次尝试：

条件类别例子识别方式
数值比较90 > 80、50 >= 50正则 numeric_pattern（:222）
字符串相等approved == approved正则 string_eq_pattern（:243）
包含（in）error in some message拆 ' in '（:256）
包含（contains）status contains success拆 ' contains '（:263）
布尔真值true / 非空串兜底真值判断（:272-278）

失败即 False（fail-safe）。 整段包在 try/except 里，任何异常都记 warning 后返回 False（evaluator.py:280-282）；缺变量导致比较式左边为空也直接判 False（:214-219）。设计意图：条件出错宁可不走危险分支。

4.3 三个类 + 一个协议​

模块把两种条件抽象成可互换的实现，用 Protocol 定契约：

符号角色位置
ConditionProtocol最小契约：只要求一个 evaluate(context) -> boolconditions/protocols.py:17
RoutingConditionProtocol扩展契约：再加 get_target(context) -> List[str]（返回路由目标）protocols.py:57
ExpressionCondition字符串表达式实现，包着 evaluate_condition()evaluator.py:17
DictCondition字典路由实现：按 key 取决策值，get_target 做键查找evaluator.py:65

两者都 @runtime_checkable，可用 isinstance 做鸭子类型检查，也方便测试里 mock（protocols.py:16、:56）。AgentFlow 走 ExpressionCondition 那条（表达式→布尔）；AgentTeam 走 DictCondition 那条（决策值→下一批任务）。

4.4 注意：三种"条件"机制并存​

同一个 AgentFlow 里，"条件"其实有三种互不相同的机制，别混：

机制出现处判断依据求值方式
Route 关键词路由route({...})上一步输出文本词边界正则搜索（workflows.py:2268）
If/when 表达式when("{{x}}>80", ...)变量表的值evaluate_condition 正则解析
Repeat.until / Task.should_runrepeat(..., until=fn)任意Python 回调返回 bool（workflows.py:2759、:1186）

5. YAML 工作流解析（CLI/YAML 与 SDK 对等）​

AgentFlow 有两种等价入口：写 Python 或 写 YAML。YAMLWorkflowParser（yaml_parser.py:21）负责把 YAML 翻译成同一套原语对象，因此 YAML 能表达的东西 SDK 都能表达，反之亦然。

解析主线： parse_file() / parse_string()（yaml_parser.py:76、:96）读 YAML → _parse_steps() 遍历 steps: → _parse_single_step() 按键名分派（yaml_parser.py:743-770）。

YAML 键 ↔ SDK 原语的对等表：

YAML 键分派到产出的 SDK 对象
route:_parse_route_step（:945）route(...) → Route
parallel:_parse_parallel_step（:968）parallel(...) → Parallel
loop:_parse_loop_step（:994）loop(...) → Loop
repeat:_parse_repeat_step（:1146）repeat(...) → Repeat
include:_parse_include_stepinclude(...) → Include
if:_parse_if_step（:900）If(condition, then, else)
agent:_parse_agent_step（:772）已注册的 Agent

YAML 里的 if: 块写法直接对应 when() 的三个参数：

steps:
  - if:
      condition: "{{score}} > 80"   # 同一套字符串表达式语法
      then:
        - agent: approver
      else:
        - agent: rejector

依据：_parse_if_step 读取 condition/then/else 造 If（yaml_parser.py:919-942）。

一个 YAML 特有的小工具：repeat 的 until 若写成字符串，会被 _create_condition_from_string() 包成"输出里是否含该子串"的回调函数（yaml_parser.py:1176-1189）——这是给 YAML 用户的便利糖，语义比 SDK 里传 lambda 更弱（只做子串包含）。

6. 两套范式对照：AgentFlow.when() vs AgentTeam 的 Task.condition​

仓库自带一个不跑 LLM 的冒烟测试 examples/smoke_test_condition_syntax.py，专门演示这两套条件语法的差异——因为它们都叫 "condition" 却是两回事，是新手最大的困惑源。

语法对照：

维度AgentFlow + when()AgentTeam + Task.condition
条件写法字符串表达式 "{{score}} >= 50"字典 {"approved": ["publish"], "rejected": ["edit"]}
求值产物表达式 → 布尔决策值 → 下一批任务名
谁产生输入变量表已有的值task_type="decision" 让 LLM 吐一个决策词
底层实现ExpressionCondition / evaluate_conditionDictCondition 键查找（evaluator.py:65）
决定路由的是写定的表达式（确定性）LLM 的输出（非确定性）

依据：smoke_test_condition_syntax.py:34-44（AgentFlow 字符串）与 :66-75（Task 字典路由）。测试第 131-175 行的对照框还点名了"同一个词 condition 意思不同""Task 另有 should_run 是第 3 种写法"等困惑点。

该选哪套？

  • 步骤已知、要可复现、想省掉 manager LLM 的开销与不确定性 → AgentFlow（本章）。判断依据来自你能算出的变量值。
  • 需要让 LLM 在运行时决策下一步走向、步骤图更自由 → AgentTeam + Process（第 04 章）。判断依据来自模型输出的决策词。

顺带一提：新版 Task 也支持 when="{{score}} > 80" + then_task/else_task 的统一字符串语法（smoke_test_condition_syntax.py:93-120），底层同样走 evaluate_condition——这是官方在弥合两套语法。但字典 condition 的 LLM 决策路由仍是 AgentTeam 独有。

7. 边界与局限（诚实）​

  • 条件表达式表达力有限。 只认单个二元比较 / in / contains / 真值，不支持 and/or/括号等复合逻辑（evaluator.py:209-278 没有对应分支）。要复合逻辑得拆成嵌套 if 或改用 Repeat.until 回调。
  • Route 匹配是文本搜索，不是语义。 靠关键词词边界命中（workflows.py:2268），上一步输出用词不同就可能漏匹配、落到 default。
  • 嵌套上限硬编码为 5。 超过 MAX_NESTING_DEPTH 直接抛错（workflows.py:2074），深层组合流水线要重构。
  • 并行默认只有 3 个 worker。 为防 LLM 限流刻意压低（workflows.py:42）；大批量并发需显式抬高 max_workers，并自担限流风险（框架会 logger.info 提醒，workflows.py:2394-2398）。
  • 条件求值 fail-safe = 静默走 False。 变量拼错、表达式格式不对，都只记 warning 后判 False（evaluator.py:280-282），不会报错中断——调试时容易被"为什么总走 else"绊住。

8. 代码地图（导航索引）​

主题文件路径符号
工作流主体 / 主循环src/praisonai-agents/praisonaiagents/workflows/workflows.py:555AgentFlow
执行入口（start 为其别名）…/workflows/workflows.py:998 / :3044run / start
向后兼容别名…/workflows/workflows.py:3051Workflow / Pipeline = AgentFlow
步骤上下文 / 结果…/workflows/workflows.py:174 / :182WorkflowContext / StepResult
六原语（数据类）…/workflows/workflows.py:197/219/250/332/410/461Route/Parallel/Loop/Repeat/Include/If
六原语（便捷函数）…/workflows/workflows.py:361-456 / :506/533route/parallel/loop/repeat/include / when/if_
嵌套深度上限…/workflows/workflows.py:459MAX_NESTING_DEPTH
嵌套原语统一分派…/workflows/workflows.py:2045_execute_single_step_internal
各原语执行器…/workflows/workflows.py:2245/2345/2450/2732/2780_execute_route/_parallel/_loop/_repeat/_if
条件引擎（共享函数）src/praisonai-agents/praisonaiagents/conditions/evaluator.py:129evaluate_condition
表达式 / 字典条件类…/conditions/evaluator.py:17 / :65ExpressionCondition / DictCondition
条件协议…/conditions/protocols.py:17 / :57ConditionProtocol / RoutingConditionProtocol
YAML 解析器src/praisonai-agents/praisonaiagents/workflows/yaml_parser.py:21YAMLWorkflowParser
YAML 键→原语分派…/workflows/yaml_parser.py:743_parse_single_step
两套条件语法对照（可运行）src/praisonai-agents/examples/smoke_test_condition_syntax.py—