interactive_flow · 交互式暂停/恢复

本页收录可运行示例，配套本文档「指南 04」(指南 04)。 代码为随框架交付的示例，已一并收录到本站供直接对照使用。

说明与运行

示例: 交互式客服流程(IfElse 条件分支 + UserSelect 交互). 流程: ChatLLM 回复 → IfElse 判断是否需要转人工 - 需要转人工 → UserSelect 让用户选择 → Answer 回显选择 - 不需要 → Answer 自动回复 运行: cd agentframework PYTHONPATH=src python examples/interactive_flow.py

运行前需在已安装 agentframework 的 Python ≥ 3.11 环境；示例多为自带本地模型，无需真实 API Key；需要外部依赖（如 chromadb / 数据库驱动）会在说明中注明。

完整源码

"""示例: 交互式客服流程(IfElse 条件分支 + UserSelect 交互).

流程: ChatLLM 回复 → IfElse 判断是否需要转人工
  - 需要转人工 → UserSelect 让用户选择 → Answer 回显选择
  - 不需要 → Answer 自动回复

运行:
    cd agentframework
    PYTHONPATH=src python examples/interactive_flow.py
"""

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


class _EchoModel:
    """本地演示模型: 回显用户输入."""

    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

        reply = self.invoke(messages)
        for ch in reply:
            yield AIMessage(content=ch)


def build_interactive_flow() -> Workflow:
    """构建交互式客服工作流."""
    wf = Workflow(name="interactive_service")
    wf.add_node(
        ChatLLM(
            name="llm",
            chat_model=_EchoModel(),
            system="你是一个专业的客服助手",
        )
    )
    wf.add_node(
        IfElse(
            name="judge",
            left=wf.ref("llm.reply"),
            condition=lambda inputs, ctx: "转人工" in str(inputs["left"]),
        )
    )
    wf.add_node(
        UserSelect(
            name="select",
            description="需要转人工, 请选择操作",
            options=[
                {"key": "manual", "value": "转人工客服"},
                {"key": "cancel", "value": "取消"},
            ],
        )
    )
    wf.add_node(Answer(name="handoff", text=wf.ref("select.selected")))
    wf.add_node(Answer(name="auto", text="好的, 已为您自动处理"))
    wf.add_edge("llm", "judge")
    wf.add_conditional_edge("judge", {True: "select", False: "auto"})
    wf.add_edge("select", "handoff")
    return wf


if __name__ == "__main__":
    from agentframework.state.checkpoint import MemoryCheckpointStore

    wf = build_interactive_flow()
    # interactive 必须带 checkpointer 才能跨请求恢复
    app = wf.compile(store=MemoryCheckpointStore())
    print("已编译工作流:", wf.name, "(带 checkpointer, 支持 interactive)")
    serve({"app-interactive": app})

---

回到示例索引。