跳转至

核心概念 · DSL 构建

Workflow 是编排 DSL 的入口，API 与 LangGraph 对齐。本文逐个讲清它的五个构建方法，以及它们的语义。

1. 总览

from agentframework import Workflow
from agentframework.nodes import Answer, ChatLLM, IfElse

wf = Workflow(name="demo")  # ① 建图
wf.add_node(...)  # ② 加节点
wf.add_edge("llm", "ans")  # ③ 连普通边
wf.add_conditional_edge("judge", {...})  # ④ 连条件边
ans = wf.ref("llm.reply")  # ⑤ 建引用（通常在传参处内联用）
app = wf.compile(store=...)  # ⑥ 编译

Workflow.__init__ 签名：

Workflow(name, *, inputs=None, system=None)
  • name：工作流名（节点日志 / 追踪 appId 用）。
  • inputs / system：阶段 0 预留，一般不用传。

2. add_node(node) — 加节点

传入一个已实例化的内置/自定义节点。节点自带 name、输入输出 schema。节点实例名在工作流内必须唯一，重名会抛 DslError。

wf.add_node(ChatLLM(name="llm", chat_model=model, system="你是客服"))
wf.add_node(
    IfElse(
        name="judge",
        left=wf.ref("llm.reply"),
        condition=lambda inputs, ctx: "转人工" in str(inputs["left"]),
    )
)
wf.add_node(Answer(name="ans", text="好的，为您转人工"))

节点构造的 **params 里，可以是：

  • 常量（system="你是客服"、text="固定话术"）；
  • 引用（left=wf.ref("llm.reply")）；
  • 注入对象（chat_model=model、retriever=retriever、client=httpx_client，便于测试注入 mock）。

3. add_edge(src, dst) — 普通边

src 执行完必执行 dst。dst 可用内置保留节点 "__end__" 表示流程结束。

wf.add_edge("llm", "ans")  # llm → ans
wf.add_edge("ans", "__end__")  # 显式结束（可省略，取决于图）

4. add_conditional_edge(src, mapping) — 条件边

按 src 节点的 branch 输出值，查 mapping 路由到不同下游。src 通常是 IfElse 或 QuestionClassify（它们的输出约定为 branch）。

# 二分支：branch 是 bool
wf.add_conditional_edge("judge", {True: "handoff", False: "auto"})

# 多分支：branch 是 str，查表路由
wf.add_conditional_edge("cls", {"退款": "refund", "物流": "logistics", "其他": "default"})

mapping 的 key 是「分支值」，value 是「目标节点名」（也可为 "__end__"）。分支值未命中 mapping 时回退到结束（LangGraph 默认行为）。

与「条件边」搭配，IfElse 的 condition 有三种形态，QuestionClassify 则用 LLM 分类得到稳定类别名——两者都产出一个可路由的 branch。完整配图见指南 01。

5. wf.ref(expr) — 节点输出引用

引用某节点输出或工作流入口。expr 两种形式：

  • "input"：工作流入口；
  • "节点名.输出键"：例如 "llm.reply"、"kb.text"、"api.json"。
wf.add_node(Answer(name="ans", text=wf.ref("llm.reply")))   # 引用 llm 的 reply 输出
wf.add_node(ChatLLM(name="llm", user_input=wf.ref("input"), ...))  # 引用工作流入口

wf.ref 返回一个可序列化的 Ref 对象，编译期校验引用目标存在，运行期从 state 取值注入。语法非法（如空串、缺输出键）在 Ref.parse 阶段即抛 DslError。

6. compile(store=None) — 编译

把 DSL 静态校验后翻译为 LangGraph StateGraph，返回原生 CompiledStateGraph（不包一层）。

app = wf.compile()  # 内存断点（同进程跨请求可恢复）
app = wf.compile(store=saver)  # 接持久化 saver（跨进程/重启可恢复）

store 的三种模式：

store 取值 含义 典型场景
None（默认） 内存 InMemorySaver，同进程跨请求可暂停恢复 演示 / 单进程服务
官方 saver（如 AsyncSqliteSaver） 直用官方 saver，框架注入 PersistentSerializer 生产持久化
自定义 CheckpointStore 框架适配成 LangGraph saver，业务只管存/读 对接已有存储抽象

compile() 默认就开了内存断点，所以 interactive 节点在单进程内也能「暂停→下次请求恢复」。要跨进程 / 重启不丢，请接持久化，见指南 06。

7. 一个完整的可运行片段

from agentframework import Workflow
from agentframework.nodes import Answer, ChatLLM, IfElse


class EchoModel:  # 本地回显模型
    def invoke(self, messages):
        return f"客服回复: {str(messages[-1].content) if messages else '你好'}"


def build() -> Workflow:
    wf = Workflow(name="demo")
    wf.add_node(ChatLLM(name="llm", chat_model=EchoModel(), output=False))
    wf.add_node(
        IfElse(
            name="judge",
            left=wf.ref("llm.reply"),
            condition=lambda i, ctx: "转人工" in str(i["left"]),
        )
    )
    wf.add_node(Answer(name="handoff", text="好的，为您转人工"))
    wf.add_node(Answer(name="auto", text="已自动处理"))
    wf.add_edge("llm", "judge")
    wf.add_conditional_edge("judge", {True: "handoff", False: "auto"})
    return wf

下一步