Agno 工作流编排与条件执行

Agno 工作流编排与条件执行

学习目标

本章将分析 Agno 的工作流系统：

• 串行、并行、条件分支、循环执行四种基本模式
• CEL 表达式引擎用于条件判断
• 人工审核步骤
• 步骤审查与撤销重试

前置知识

建议先阅读：

• 函数式工作流 API — 工作流的函数式编排模式

下文假设你已理解上述概念，直接聚焦 Agno 的具体实现。

---

项目实践

工作流的规模

Agno 的 workflow.py 是项目中最大的单文件（10000+ 行代码），覆盖了完整的工作流引擎。

四种基本执行模式

工作流定义

# 伪代码：Workflow 定义
workflow = Workflow(
    steps=[
        Step(name="research", agent=researcher),
        Step(name="analyze", agent=analyst, condition="previous_result.confidence > 0.8"),
        Step(name="write", agent=writer),
    ],
    db=PostgresDb(db_url="..."),
)

CEL 表达式引擎

Agno 使用 CEL（Common Expression Language）作为条件表达式引擎：

# 伪代码：CEL 条件
Step(
    name="send_alert",
    agent=alert_agent,
    condition="result.severity == 'critical' && result.confidence > 0.9",
)

CEL 支持：

• 布尔运算（&&, ||, !）
• 比较运算（==, !=, >, <, >=, <=）
• 字符串操作（.contains(), .startsWith(), .matches()）
• 列表操作（.size(), in）
• 空值安全（?? 默认值）

人工审核步骤

工作流中可以在任何步骤后插入人工审核：

# 伪代码：工作流中的人工审核
Step(
    name="deploy",
    agent=deploy_agent,
    approval=ApprovalConfig(
        type=ApprovalType.REQUIRED,
        instructions="请审查部署计划，确认无误后批准。",
    ),
)

步骤审查与撤销重试

# 伪代码：步骤审查
Step(
    name="review_step",
    review=True,  # 执行后需要人工审查
    retry_on_revoke=True,  # 如果审查被撤销，自动重试
)

流程：

1. 步骤执行完成
2. 等待人工审查
3. 如果审查通过 → 继续下一步
4. 如果审查被撤销 → 根据 retry_on_revoke 决定是否重试

工作流 Session

工作流有独立的 session 管理：

# 伪代码：WorkflowSession
WorkflowSession(
    workflow_id="wf_123",
    member_run_ids={
        "step_1": ["run_001"],
        "step_2": ["run_002", "run_003"],  # 重试产生的多次 run
    },
    status="completed",
)

工作流的取消会传播到所有成员 run：acleanup_member_runs(workflow_id)。

问题与规避

1. 工作流长度限制

长工作流（>20 步骤）的执行时间和出错概率都增加。

规避：

• 将长工作流拆分为多个子工作流
• 使用条件分支跳过不必要的步骤
• 在关键步骤后插入检查点

2. CEL 表达式的安全性

CEL 表达式由用户配置，可能注入恶意逻辑。

规避：

• CEL 沙箱执行，限制可用的函数和变量
• 只暴露工作流上下文的变量
• 表达式长度和复杂度限制

3. 并行步骤的依赖关系

并行步骤之间可能存在隐式依赖（步骤 B 需要步骤 A 的结果）。

规避：使用串行模式或在并行步骤后插入合并步骤（等待所有并行结果后再执行）。

设计取舍

为什么使用 CEL 而非 Python 表达式？

优势：CEL 是安全的沙箱表达式语言，不会执行任意 Python 代码 代价：表达能力不如 Python 灵活 替代方案：Python lambda 作为条件——更灵活但不安全

为什么工作流是最大的单文件？

工作流引擎需要协调 Agent、Team、条件、循环、审核、取消、重试等多种机制。将这些逻辑分散到多个文件会增加理解成本。

替代方案：将步骤定义、执行引擎、Session 管理分到不同文件——模块化更好但阅读时需要跨文件跳转。

参考来源

• Agno workflow/workflow.py 源码