ifelse_demo · 条件分支完整用法¶
本页收录可运行示例，配套本文档「指南 01」(指南 01)。 代码为随框架交付的示例，已一并收录到本站供直接对照使用。
说明与运行¶
示例: IfElse 条件分支节点完整用法(E4 DSL). 覆盖 5 个点: A. 二分支 —— condition 返回 bool, 条件边 {True:.., False:..} 二选一路由 B. 多分支 —— condition 返回 str, 条件边 {值:..} 查表路由 C. 三种 condition 形态 —— 可调用对象 / 常量 / wf.ref 引用 D. left 输入 —— 把上游输出传进来供 condition 读取(见 A/B) E. 当计算节点用 —— branch 输出被下游 wf.ref 当数据引用(不用于路由) 运行: cd agentframework PYTHONPATH=src python examples/ifelse_demo.py
运行前需在已安装
agentframework的 Python ≥ 3.11 环境；示例多为自带本地模型，无需真实 API Key；需要外部依赖（如chromadb/ 数据库驱动）会在说明中注明。
完整源码¶
"""示例: IfElse 条件分支节点完整用法(E4 DSL).
覆盖 5 个点:
A. 二分支 —— condition 返回 bool, 条件边 {True:.., False:..} 二选一路由
B. 多分支 —— condition 返回 str, 条件边 {值:..} 查表路由
C. 三种 condition 形态 —— 可调用对象 / 常量 / wf.ref 引用
D. left 输入 —— 把上游输出传进来供 condition 读取(见 A/B)
E. 当计算节点用 —— branch 输出被下游 wf.ref 当数据引用(不用于路由)
运行:
cd agentframework
PYTHONPATH=src python examples/ifelse_demo.py
"""
from agentframework import Workflow
from agentframework.nodes import Answer, ChatLLM, IfElse
class _EchoModel:
"""本地演示模型: 回显用户输入, 无需真实 API 密钥."""
def invoke(self, messages):
text = str(messages[-1].content) if messages else ""
return f"客服回复: {text}"
def stream(self, messages):
from langchain_core.messages import AIMessage
for ch in self.invoke(messages):
yield AIMessage(content=ch)
def _cfg(thread_id: str) -> dict:
"""执行配置: checkpointer 常开(compile 默认内存模式), 显式 thread_id."""
return {"configurable": {"thread_id": thread_id}}
# ---------------------------------------------------------------------------
# A. 二分支: condition 返回 bool, 条件边按 True/False 二选一路由
# ---------------------------------------------------------------------------
def build_binary() -> Workflow:
wf = Workflow(name="demo_binary")
wf.add_node(
ChatLLM(
name="llm",
chat_model=_EchoModel(),
system="你是客服",
user_input=wf.ref("input"), # 工作流入口的当前问题
)
)
# left: 把上游 llm.reply 作为「待判断值」传给 IfElse, 供 condition 读取
wf.add_node(
IfElse(
name="judge",
left=wf.ref("llm.reply"),
condition=lambda inputs, ctx: "转人工" in str(inputs["left"]),
)
)
wf.add_node(Answer(name="handoff", text="好的, 为您转接人工客服"))
wf.add_node(Answer(name="auto", text="已为您自动处理"))
wf.add_edge("llm", "judge")
# 条件边: 拿 judge 输出的 branch(True/False) 查表路由
wf.add_conditional_edge("judge", {True: "handoff", False: "auto"})
return wf
# ---------------------------------------------------------------------------
# B. 多分支: condition 返回 str, 条件边查表路由(未命中默认到 __end__)
# ---------------------------------------------------------------------------
def build_multi() -> Workflow:
wf = Workflow(name="demo_multi")
wf.add_node(
ChatLLM(
name="llm",
chat_model=_EchoModel(),
system="你是客服",
user_input=wf.ref("input"),
)
)
wf.add_node(
IfElse(
name="route",
left=wf.ref("llm.reply"),
condition=lambda inputs, ctx: (
"退款"
if "退款" in str(inputs["left"])
else "物流"
if "物流" in str(inputs["left"])
else "其他"
),
)
)
wf.add_node(Answer(name="refund", text="退款流程: 请在订单页申请退款"))
wf.add_node(Answer(name="logistics", text="物流查询: 请提供运单号"))
wf.add_node(Answer(name="default", text="请描述您的问题"))
wf.add_edge("llm", "route")
wf.add_conditional_edge("route", {"退款": "refund", "物流": "logistics", "其他": "default"})
return wf
# ---------------------------------------------------------------------------
# C. 三种 condition 形态对比(各自独立的小工作流)
# ---------------------------------------------------------------------------
def build_condition_forms() -> None:
# 形态① 可调用对象 (inputs, ctx) -> 分支值: 最灵活, 可读 left / ctx.variables
wf = Workflow(name="demo_form_callable")
wf.add_node(
IfElse(
name="judge",
condition=lambda inputs, ctx: (
"vip" if ctx.variables.get("level") == "vip" else "normal"
),
)
)
wf.add_node(Answer(name="vip", text="尊贵的 VIP, 为您优先处理"))
wf.add_node(Answer(name="normal", text="普通用户, 请排队"))
wf.add_conditional_edge("judge", {"vip": "vip", "normal": "normal"})
run_and_print(wf, {"input": {}}, _cfg("c1"), "C1 callable(读 ctx.variables)")
# 形态② 常量值: branch 固定(常用于暂时固定走某条路)
wf = Workflow(name="demo_form_const")
wf.add_node(IfElse(name="judge", condition=True))
wf.add_node(Answer(name="a", text="走了 True 分支"))
wf.add_node(Answer(name="b", text="走了 False 分支"))
wf.add_conditional_edge("judge", {True: "a", False: "b"})
run_and_print(wf, {"input": {}}, _cfg("c2"), "C2 常量 condition=True")
# 形态③ wf.ref 引用: 转发其他节点的 branch 值(复用已有判定)
wf = Workflow(name="demo_form_ref")
wf.add_node(IfElse(name="first", condition=True))
wf.add_node(IfElse(name="forward", condition=wf.ref("first.branch")))
wf.add_node(Answer(name="a", text="forward 收到了 True"))
wf.add_node(Answer(name="b", text="forward 收到了 False"))
wf.add_edge("first", "forward")
wf.add_conditional_edge("forward", {True: "a", False: "b"})
run_and_print(wf, {"input": {}}, _cfg("c3"), "C3 condition=wf.ref 转发")
# ---------------------------------------------------------------------------
# E. 当计算节点用: branch 不参与路由, 被下游当数据引用
# ---------------------------------------------------------------------------
def extract_order_no(text: str) -> str:
"""从回复文本里抠订单号(演示用: 取 '订单号:xxx' 后的部分)."""
if "订单号:" in text:
return text.split("订单号:", 1)[1].strip()
return "未找到订单号"
def build_compute() -> Workflow:
wf = Workflow(name="demo_compute")
wf.add_node(
ChatLLM(
name="llm",
chat_model=_EchoModel(),
system="你是客服",
user_input=wf.ref("input"),
)
)
# IfElse 当计算节点: 从 llm.reply 提取订单号 → 存进 extract.branch
wf.add_node(
IfElse(
name="extract",
left=wf.ref("llm.reply"),
condition=lambda inputs, ctx: extract_order_no(str(inputs["left"])),
)
)
# 下游直接用 wf.ref 引用 branch 的值(注意: 这里没有条件边!)
wf.add_node(Answer(name="out", text=wf.ref("extract.branch")))
wf.add_edge("llm", "extract")
wf.add_edge("extract", "out")
return wf
# ---------------------------------------------------------------------------
# 运行辅助: 编译 + invoke + 打印结果
# ---------------------------------------------------------------------------
def run_and_print(wf: Workflow, inputs: dict, config: dict, label: str) -> None:
app = wf.compile()
result = app.invoke(inputs, config=config)
print(f"--- {label} ---")
for k, v in result.items():
if k != "_ctx":
print(f" {k} = {v!r}")
print()
def main() -> None:
# A. 二分支: 输入含「转人工」→ 走 handoff; 否则走 auto
run_and_print(build_binary(), {"input": "我要转人工"}, _cfg("a1"), "A 二分支(转人工)")
run_and_print(build_binary(), {"input": "谢谢"}, _cfg("a2"), "A 二分支(普通)")
# B. 多分支
run_and_print(build_multi(), {"input": "我要退款"}, _cfg("b1"), "B 多分支(退款)")
# C. 三种 condition 形态
build_condition_forms()
# E. 计算节点
run_and_print(build_compute(), {"input": "我的订单号: 9527"}, _cfg("e1"), "E 计算节点")
if __name__ == "__main__":
main()
回到示例索引。