条件连接

条件连接也是我们在工作流中常用的连接方式，当我们需要根据上一个工作块的输出结果进行下游分支选择时，我们就需要用到条件连接。

在AI应用的实际开发场景中，当我们需要根据模型对用户的输入意图进行判断并选择对应的处理分支时，或当我们需要设计一个agent路由协作模式，让模型根据用户输入进行推理判断选择对应的下游处理agent时，它们都是使用条件连接适当的场景。

Agently Workflow 为开发者提供了一整套条件连接的连接关系表达方法，支持针对超过2个条件的多条件分支连接关系。

条件连接示意图

下图展示了最常见的多分支条件连接流程，通过该示意图，能帮助您更直观地理解条件分支的相关设计思路。

条件连接的三种连接方法

• .if_condition(<condition_func>)：用在条件分支出现时的首次表达，让框架知道发起该方法的工作块（即在.if_condiiton()左侧的对象）需要使用条件连接的方式进行下游连接，同时定义了第一个条件判断函数(condition function)，当条件判断函数输出结果为True，就进入到.if_condition().connect_to()指向的下游工作块
• .elif_condition(<condition_func>)：必须在.if_condition()出现之后表达，允许定义更多的条件判断函数，当条件判断函数输出结果为True，就进入到.elif_condition().connect_to()指向的下游工作块
• .else_condition()：必须在.if_condition()出现之后表达，当上面所有条件判断函数都没有命中时，就进入到.else_condition().connect_to()指向的下游工作块

所以，根据上面的连接方法说明，上面的条件连接示意图对应的样例代码如下：

条件连接示意图对应的样例代码

import Agently
workflow = Agently.Workflow()

@workflow.chunk()
def input_data(inputs, storage):
    return int(input("[Input an interger larger than 0]: "))

@workflow.chunk()
def echo_error(inputs, storage):
    print(inputs["default"], "is equal or less than 0, that's not what I want.")
    return inputs["default"]

@workflow.chunk()
def echo_0_to_10(inputs, storage):
    print(inputs["default"], "is in the range of (0,10]")
    return inputs["default"]

@workflow.chunk()
def echo_10_to_100(inputs, storage):
    print(inputs["default"], "is in the range of (10-100]")
    return inputs["default"]

@workflow.chunk()
def echo_larger_than_100(inputs, storage):
    print(inputs["default"], "is in the range of (10-100]")
    return inputs["default"]

@workflow.chunk()
def conclude(inputs, storage):
    print(inputs["default"], "is a magic number and I love it.")
    return inputs["default"]

# 分两段进行连接关系表达
# 第一段表达条件分支，每个分支末尾都指向"conclude"工作块
(
    workflow
        .connect_to("input_data")
        .if_condition(lambda return_value, storage: return_value <= 0)
            .connect_to("echo_error")
            .connect_to("conclude")
        .elif_condition(lambda return_value, storage: return_value > 0 and return_value <=10)
            .connect_to("echo_0_to_10")
            .connect_to("conclude")
        .elif_condition(lambda return_value, storage: return_value > 10 and return_value <=100)
            .connect_to("echo_10_to_100")
            .connect_to("conclude")
        .else_condition()
            .connect_to("echo_larger_than_100")
            .connect_to("conclude")
)
# 第二段从"conclude"工作块出发，指向"END"块将结果输出
workflow.chunks["conclude"].connect_to("END")

result = workflow.start()
print(result)

运行结果

[Input an interger larger than 0]: 0
0 is equal or less than 0, that's not what I want.
0 is a magic number and I love it.
{'default': 0}

---

[Input an interger larger than 0]: 5
5 is in the range of (0,10]
5 is a magic number and I love it.
{'default': 5}

---

[Input an interger larger than 0]: 21
21 is in the range of (10-100]
21 is a magic number and I love it.
{'default': 21}

---

[Input an interger larger than 0]: 2024
2024 is in the range of (10-100]
2024 is a magic number and I love it.
{'default': 2024}

%%{ init: { 'flowchart': { 'curve': 'linear' }, 'theme': 'neutral' } }%%
%% Rendered By Agently %%
flowchart LR
classDef chunk_style fill:#fbfcdb,stroke:#666,stroke-width:1px,color:#333;
classDef condition_chunk_style fill:#ECECFF,stroke:#9370DB,stroke-width:1px,color:#333;
classDef loop_style fill:#f5f7fa,stroke:#666,stroke-width:1px,color:#333,stroke-dasharray: 5 5
    b311b843-e785-47e0-bbed-11e1deb2a319("START"):::chunk_style -.-> |"* -->-- default"| da2e14b3-5e1e-485d-985a-84ee77f706c3("input_data"):::chunk_style
    da2e14b3-5e1e-485d-985a-84ee77f706c3("input_data"):::chunk_style -.-> |"* -->-- default"| 4fa7195b-3c8e-4d08-b292-5309ff2a15d0{{"Condition"}}:::condition_chunk_style
    4fa7195b-3c8e-4d08-b292-5309ff2a15d0{{"Condition"}}:::condition_chunk_style -.-> |"* -- ◇ -- default"| a50420a9-dbf9-443d-93db-d76a56ef0887("echo_error"):::chunk_style
    a50420a9-dbf9-443d-93db-d76a56ef0887("echo_error"):::chunk_style -.-> |"* -->-- default"| ae63caf2-3358-4eed-9fa2-75ad0bec589e("conclude"):::chunk_style
    4fa7195b-3c8e-4d08-b292-5309ff2a15d0{{"Condition"}}:::condition_chunk_style -.-> |"* -- ◇ -- default"| 0171de7c-b72c-4a5f-9d20-55d9bf7e3e3b("echo_0_to_10"):::chunk_style
    0171de7c-b72c-4a5f-9d20-55d9bf7e3e3b("echo_0_to_10"):::chunk_style -.-> |"* -->-- default"| ae63caf2-3358-4eed-9fa2-75ad0bec589e("conclude"):::chunk_style
    4fa7195b-3c8e-4d08-b292-5309ff2a15d0{{"Condition"}}:::condition_chunk_style -.-> |"* -- ◇ -- default"| e4c78757-381b-4f5d-bd6e-e94390d980a4("echo_10_to_100"):::chunk_style
    e4c78757-381b-4f5d-bd6e-e94390d980a4("echo_10_to_100"):::chunk_style -.-> |"* -->-- default"| ae63caf2-3358-4eed-9fa2-75ad0bec589e("conclude"):::chunk_style
    4fa7195b-3c8e-4d08-b292-5309ff2a15d0{{"Condition"}}:::condition_chunk_style -.-> |"* -- ◇ -- default"| 085c84ca-650c-4ef9-82f9-a5d485dad543("echo_larger_than_100"):::chunk_style
    085c84ca-650c-4ef9-82f9-a5d485dad543("echo_larger_than_100"):::chunk_style -.-> |"* -->-- default"| ae63caf2-3358-4eed-9fa2-75ad0bec589e("conclude"):::chunk_style
    ae63caf2-3358-4eed-9fa2-75ad0bec589e("conclude"):::chunk_style -.-> |"* -->-- default"| ec310479-ce85-4769-bb6a-ab5a4e38d4ff("END"):::chunk_style

条件判断函数

看完上面的样例代码，让我们进一步看看条件判断函数(condition function)的使用方式。

顾名思义，条件判断函数是一个函数，所以您可以使用预先定义的函数或是lambda函数作为条件判断函数。

条件判断函数与工作块执行函数类似，有针对函数输入参数定义和输出结果的规范要求，其规则如下：

• 输入参数
  • return_value: return_value将获取.if_condition()或其他条件连接左侧的发起工作块的return输出结果，作为条件判断函数可使用的条件判断依据数据
  • storage: 您也可以在条件判断函数中使用工作流全局数据存储器storage，将存储器中的数据作为条件判断依据数据
• 输出结果
  • return：条件判断函数的输出结果必须为布尔值True或False，或是它们的等价值

两种条件判断函数示例

# 使用预先定义的函数作为判断函数
def if_int_larger_than_100(return_value, storage):
    if return_value > 100:
        return True
    else:
        return False

(
    workflow.chunks["from_chunk"]
        .if_condition(if_int_larger_than_100)
            .connect_to("to_chunk")
)

# 使用Lambda函数作为判断函数
(
    workflow.chunks["from_chunk"]
        .if_condition(lambda return_value, storage: return_value > 100)
            .connect_to("to_chunk")
)