withdrawal_customer_service · 智能客服(离退休提取, 完整状态机)

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

说明与运行

示例: 智能客服·离退休提取流程(完整案例, 含网关). 业务流程(多轮状态机, 由前端在 variables 中携带 currentStage/flag/exitBusiness 驱动): 1. 首个节点为问题分类(QuestionClassify + 真实 DeepSeek): 用户提问离退休提取相关 → 进入离退休提取流程(nextStage=start) 否则 → 给予公积金知识回答(真实 DeepSeek) 2. 离退休提取流程(前端逐阶段传递 currentStage): (1) currentStage=start → 返回 nextStage=confirmSocialSecurityCard (2) currentStage=confirmSocialSecurityCard、flag=true/false → HttpRequest POST http://localhost:9080/test1(application/json, 携带 flag) 响应 {"code":"xxx","data":{"respMsg":"xxx"}}, 成功 code=FBASE0000 成功 → nextStage=withdrawalCard 失败 → 返回第三方报错文案 data.respMsg + errExit=True 前端收到 errExit=True 后携带 exitBusiness=true 走退出流程 (3) currentStage=withdrawalCard → 清空全部变量为空字符串, nextStage=end (4) exitBusiness=true → 清空全部变量为空字符串, nextStage=end (5) currentStage=userInput → UserInput 暂停让用户输入 恢复后按输入文本更新变量 inputStr, 回传前端 LLM 说明: 不 mock, 真实调用第三方 DeepSeek API(OpenAI 兼容 chat/completions, 经 httpx 直连, 不新增框架依赖); 需设置环境变量 DEEPSEEK_API_KEY。 运行(两个终端):

终端 A: 启动模拟第三方服务(可选, 用于演示 HTTP 成功/失败分支)

PYTHONPATH=src python examples/withdrawal_customer_service.py --mock-third-party

终端 B: 启动网关

PYTHONPATH=src python examples/withdrawal_customer_service.py 请求示例(curl, 阶段驱动):

1) 初始提问 → 分类 → 离退休提取, 返回 nextStage=start

curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{ "appId": "app-withdraw", "stream": true, "detail": true, "messages": [{"role":"user","content":"我要办理公积金离退休提取"}], "variables": {}}'

2) currentStage=start → nextStage=confirmSocialSecurityCard

curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{ "appId": "app-withdraw", "stream": true, "messages": [{"role":"user","content":"继续"}], "variables": {"currentStage":"start"}}'

3) 确认社保卡(flag=true) → 第三方成功 → nextStage=withdrawalCard

curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{ "appId": "app-withdraw", "stream": true, "messages": [{"role":"user","content":"确认"}], "variables": {"currentStage":"confirmSocialSecurityCard","flag":true}}'

4) 确认社保卡(flag=false) → 第三方失败 → data.respMsg + errExit=true

curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{ "appId": "app-withdraw", "stream": true, "messages": [{"role":"user","content":"确认"}], "variables": {"currentStage":"confirmSocialSecurityCard","flag":false}}'

5) 前端收到 errExit=true → 退出流程(exitBusiness=true) → 清空变量, nextStage=end

curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{ "appId": "app-withdraw", "stream": true, "messages": [{"role":"user","content":"退出"}], "variables": {"exitBusiness":true}}'

6) 提取完成(currentStage=withdrawalCard) → 清空变量, nextStage=end

curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{ "appId": "app-withdraw", "stream": true, "messages": [{"role":"user","content":"完成"}], "variables": {"currentStage":"withdrawalCard"}}'

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

完整源码

"""示例: 智能客服·离退休提取流程(完整案例, 含网关).

业务流程(多轮状态机, 由前端在 variables 中携带 currentStage/flag/exitBusiness 驱动):
    1. 首个节点为问题分类(QuestionClassify + 真实 DeepSeek):
       用户提问离退休提取相关 → 进入离退休提取流程(nextStage=start)
       否则 → 给予公积金知识回答(真实 DeepSeek)
    2. 离退休提取流程(前端逐阶段传递 currentStage):
       (1) currentStage=start             → 返回 nextStage=confirmSocialSecurityCard
       (2) currentStage=confirmSocialSecurityCard、flag=true/false
           → HttpRequest POST http://localhost:9080/test1(application/json, 携带 flag)
             响应 {"code":"xxx","data":{"respMsg":"xxx"}}, 成功 code=FBASE0000
             成功 → nextStage=withdrawalCard
             失败 → 返回第三方报错文案 data.respMsg + errExit=True
             前端收到 errExit=True 后携带 exitBusiness=true 走退出流程
       (3) currentStage=withdrawalCard    → 清空全部变量为空字符串, nextStage=end
       (4) exitBusiness=true              → 清空全部变量为空字符串, nextStage=end
       (5) currentStage=userInput         → UserInput 暂停让用户输入
           恢复后按输入文本更新变量 inputStr, 回传前端

LLM 说明: 不 mock, 真实调用第三方 DeepSeek API(OpenAI 兼容 chat/completions,
经 httpx 直连, 不新增框架依赖); 需设置环境变量 DEEPSEEK_API_KEY。

运行(两个终端):
    # 终端 A: 启动模拟第三方服务(可选, 用于演示 HTTP 成功/失败分支)
    PYTHONPATH=src python examples/withdrawal_customer_service.py --mock-third-party
    # 终端 B: 启动网关
    PYTHONPATH=src python examples/withdrawal_customer_service.py

请求示例(curl, 阶段驱动):
    # 1) 初始提问 → 分类 → 离退休提取, 返回 nextStage=start
    curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{
      "appId": "app-withdraw", "stream": true, "detail": true,
      "messages": [{"role":"user","content":"我要办理公积金离退休提取"}],
      "variables": {}}'
    # 2) currentStage=start → nextStage=confirmSocialSecurityCard
    curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{
      "appId": "app-withdraw", "stream": true,
      "messages": [{"role":"user","content":"继续"}],
      "variables": {"currentStage":"start"}}'
    # 3) 确认社保卡(flag=true) → 第三方成功 → nextStage=withdrawalCard
    curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{
      "appId": "app-withdraw", "stream": true,
      "messages": [{"role":"user","content":"确认"}],
      "variables": {"currentStage":"confirmSocialSecurityCard","flag":true}}'
    # 4) 确认社保卡(flag=false) → 第三方失败 → data.respMsg + errExit=true
    curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{
      "appId": "app-withdraw", "stream": true,
      "messages": [{"role":"user","content":"确认"}],
      "variables": {"currentStage":"confirmSocialSecurityCard","flag":false}}'
    # 5) 前端收到 errExit=true → 退出流程(exitBusiness=true) → 清空变量, nextStage=end
    curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{
      "appId": "app-withdraw", "stream": true,
      "messages": [{"role":"user","content":"退出"}],
      "variables": {"exitBusiness":true}}'
    # 6) 提取完成(currentStage=withdrawalCard) → 清空变量, nextStage=end
    curl -N http://localhost:8080/api/v2/chat/completions -H "Content-Type: application/json" -d '{
      "appId": "app-withdraw", "stream": true,
      "messages": [{"role":"user","content":"完成"}],
      "variables": {"currentStage":"withdrawalCard"}}'
"""

from __future__ import annotations

import json
import os
from typing import Any

import httpx
from langchain_core.messages import AIMessage, BaseMessage, HumanMessage

from agentframework import Workflow, serve
from agentframework.nodes import (
    Answer,
    ChatLLM,
    HttpRequest,
    IfElse,
    QuestionClassify,
    UserInput,
    VariableOutput,
    VariableUpdate,
)
from agentframework.runtime.ctx import RunContext

# 第三方接口: 社保卡校验(模拟服务见 run_mock_third_party)
THIRD_PARTY_URL = "http://localhost:9080/test1"
THIRD_PARTY_SUCCESS_CODE = "FBASE0000"


# ---------------------------------------------------------------------------
# 真实 DeepSeek 客户端(OpenAI 兼容 chat/completions, 经 httpx 直连)
# ---------------------------------------------------------------------------
class DeepSeekClient:
    """真实 DeepSeek API 客户端(不 mock).

    实现 LLMClient 期望的协议形态: 具备 invoke(非流式) 与 stream(流式) 方法,
    接收 LangChain BaseMessage 列表, 输出/产出 AIMessage. 经 httpx 直连
    DeepSeek 的 OpenAI 兼容接口, 不新增框架依赖.
    """

    def __init__(
        self,
        api_key: str,
        model: str = "deepseek-chat",
        base_url: str = "https://api.deepseek.com",
        timeout: float = 60.0,
    ) -> None:
        """初始化 DeepSeek 客户端.

        Args:
            api_key: DeepSeek API 密钥(从 DEEPSEEK_API_KEY 环境变量读取).
            model: 模型名(默认 deepseek-chat).
            base_url: API 基础地址(默认 https://api.deepseek.com).
            timeout: 请求超时秒数.
        """
        self.api_key = api_key
        self.model = model
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout

    def _headers(self) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

    def _openai_messages(self, messages: list[BaseMessage]) -> list[dict[str, str]]:
        """把 LangChain 消息列表转换为 OpenAI 风格消息(role/content)."""
        out: list[dict[str, str]] = []
        for msg in messages:
            mtype = getattr(msg, "type", "user")
            role = "system" if mtype == "system" else "assistant" if mtype == "ai" else "user"
            out.append({"role": role, "content": str(getattr(msg, "content", ""))})
        return out

    def invoke(self, messages: list[BaseMessage]) -> AIMessage:
        """非流式生成: 一次调用返回完整回复(AIMessage)."""
        body = {
            "model": self.model,
            "messages": self._openai_messages(messages),
            "stream": False,
        }
        resp = httpx.post(
            f"{self.base_url}/chat/completions",
            headers=self._headers(),
            json=body,
            timeout=self.timeout,
        )
        resp.raise_for_status()
        content = resp.json()["choices"][0]["message"]["content"]
        return AIMessage(content=content)

    def stream(self, messages: list[BaseMessage]):
        """流式生成: 逐块产出 AIMessage(delta.content), 供 LLMClient 逐块回调."""
        body = {
            "model": self.model,
            "messages": self._openai_messages(messages),
            "stream": True,
        }
        with httpx.stream(
            "POST",
            f"{self.base_url}/chat/completions",
            headers=self._headers(),
            json=body,
            timeout=self.timeout,
        ) as resp:
            resp.raise_for_status()
            for line in resp.iter_lines():
                if not line or not line.startswith("data:"):
                    continue
                payload = line[len("data:") :].strip()
                if payload == "[DONE]":
                    break
                delta = json.loads(payload)["choices"][0]["delta"].get("content")
                if delta:
                    yield AIMessage(content=delta)


# ---------------------------------------------------------------------------
# 流程逻辑(计算函数 / 入口路由 / HTTP 成败判定)
# ---------------------------------------------------------------------------
def _last_user_text(inputs: dict[str, Any], ctx: RunContext) -> str:
    """取对话历史最后一条用户消息文本(供问题分类 query 使用)."""
    last_user = next((m for m in reversed(ctx.messages) if isinstance(m, HumanMessage)), None)
    return str(last_user.content) if last_user else ""


def _flag_value(inputs: dict[str, Any], ctx: RunContext) -> Any:
    """读变量 flag 的值(供 HttpRequest 请求体携带)."""
    return ctx.variables.get("flag")


def _resp_msg(inputs: dict[str, Any], ctx: RunContext) -> Any:
    """从第三方响应取 data.respMsg(供失败文案输出)."""
    data = inputs.get("left")
    if isinstance(data, dict):
        data = data.get("data") or {}
        if isinstance(data, dict):
            return data.get("respMsg")
    return None


def _route_flow(inputs: dict[str, Any], ctx: RunContext) -> str:
    """联合路由: 先分类, 再按 退出/阶段变量 分发.

    优先级: 问题分类结果 > 退出(exitBusiness=true) > 阶段(currentStage).
    只有问题分类为「离退休提取」才进入 退出/阶段 分支, 否则一律 → kb.
    """
    variables = ctx.variables
    # ① 问题分类优先级最高: 非「离退休提取」→ 一律公积金知识回答
    if inputs.get("left") != "离退休提取":
        return "kb"
    # ② 离退休提取内: 退出 > 阶段
    if str(variables.get("exitBusiness", "")).lower() == "true":
        return "exit"
    stage = variables.get("currentStage")
    if stage == "start":
        return "start"
    if stage == "confirmSocialSecurityCard":
        return "confirm"
    if stage == "withdrawalCard":
        return "withdraw"
    if stage == "userInput":
        return "input"
    # ③ 离退休提取且无阶段参数 → 初始化流程: nextStage=start
    return "init"


def _http_ok(inputs: dict[str, Any], ctx: RunContext) -> str:
    """第三方响应判定: code == FBASE0000 为成功, 否则失败."""
    data = inputs.get("left")
    if isinstance(data, dict) and data.get("code") == THIRD_PARTY_SUCCESS_CODE:
        return "ok"
    return "fail"


# ---------------------------------------------------------------------------
# 工作流构建
# ---------------------------------------------------------------------------
def build_flow(chat_model: Any) -> Workflow:
    """构建智能客服·离退休提取流程(含问题分类 + 阶段状态机)."""
    wf = Workflow(name="withdrawal_customer_service")

    # ① 问题分类(需求第 1 点: 首个决策节点为问题分类; 真实 DeepSeek)
    #    取最后 user 消息: 用 IfElse 计算节点(基础节点)承担
    wf.add_node(IfElse(name="last_user", condition=_last_user_text))
    wf.add_node(
        QuestionClassify(
            name="classify",
            chat_model=chat_model,
            categories=[
                {"name": "离退休提取", "description": "用户咨询公积金离退休提取相关业务"},
                {"name": "其他", "description": "其他公积金问题"},
            ],
            query=wf.ref("last_user.branch"),
            description="判断用户是否咨询公积金离退休提取",
        )
    )

    # ② 联合路由: 先分类, 再按 退出/阶段变量/分类结果 分发
    wf.add_node(IfElse(name="route", left=wf.ref("classify.branch"), condition=_route_flow))
    # 离退休提取 → 初始化流程: nextStage=start
    wf.add_node(VariableUpdate(name="init_flow", key="nextStage", value="start"))
    wf.add_node(VariableOutput(name="init_out", variables={"nextStage": None}))
    # 其他 → 公积金知识回答(真实 DeepSeek, 流式输出)
    wf.add_node(
        ChatLLM(
            name="kb_answer",
            chat_model=chat_model,
            system="你是公积金客服助手, 请针对公积金知识(含离退休提取)给出专业、简洁的回答。",
            output=True,
        )
    )

    # ③ 阶段 start → nextStage=confirmSocialSecurityCard
    wf.add_node(
        VariableUpdate(name="stage_start", key="nextStage", value="confirmSocialSecurityCard")
    )
    wf.add_node(VariableOutput(name="start_out", variables={"nextStage": None}))

    # ④ 阶段 confirmSocialSecurityCard: 携带 flag 调第三方 HTTP
    #    读 flag 变量: 用 IfElse 计算节点(基础节点)承担
    wf.add_node(IfElse(name="flag", condition=_flag_value))
    wf.add_node(
        HttpRequest(
            name="http",
            url=THIRD_PARTY_URL,
            method="POST",
            body={"flag": wf.ref("flag.branch")},
        )
    )
    wf.add_node(IfElse(name="http_judge", left=wf.ref("http.json"), condition=_http_ok))
    # 成功 → nextStage=withdrawalCard
    wf.add_node(VariableUpdate(name="succ_flow", key="nextStage", value="withdrawalCard"))
    wf.add_node(VariableOutput(name="succ_out", variables={"nextStage": None}))
    # 失败 → 返回第三方报错文案(data.respMsg) + errExit=True
    #    取 respMsg: 用 IfElse 计算节点(基础节点)承担
    wf.add_node(IfElse(name="resp", left=wf.ref("http.json"), condition=_resp_msg))
    wf.add_node(VariableUpdate(name="err_flag", key="errExit", value=True))
    wf.add_node(VariableOutput(name="err_out", variables={"errExit": None}))
    wf.add_node(Answer(name="err_ans", text=wf.ref("resp.branch")))

    # ⑤ 阶段 userInput: UserInput 暂停让用户输入 → 更新 inputStr 回传前端
    wf.add_node(UserInput(name="ask_input", description="请输入您的补充信息"))
    wf.add_node(VariableUpdate(name="set_input", key="inputStr", value=wf.ref("ask_input.input")))
    wf.add_node(VariableOutput(name="input_out", variables={"inputStr": None}))

    # ⑥ 结束/退出共用流程: 单节点清空其余变量为空串 + 返回 nextStage=end
    #    显式传值(网关按请求基线 merge_new_variables 只回传变更键),
    #    无需 VariableUpdate 清空链
    wf.add_node(
        VariableOutput(
            name="end_out",
            variables={
                "currentStage": "",
                "flag": "",
                "exitBusiness": "",
                "errExit": "",
                "nextStage": "end",
            },
        )
    )

    # ⑦ 连线
    wf.add_edge("last_user", "classify")  # 问题分类(第一决策节点)
    wf.add_edge("classify", "route")  # 分类结果 → 联合路由
    wf.add_conditional_edge(
        "route",
        {
            "exit": "end_out",  # exitBusiness=true → 清空变量 + nextStage=end
            "start": "stage_start",  # currentStage=start → 阶段推进
            "confirm": "flag",  # currentStage=confirmSocialSecurityCard → 第三方校验
            "withdraw": "end_out",  # currentStage=withdrawalCard → 清空变量 + nextStage=end
            "input": "ask_input",  # currentStage=userInput → 交互输入
            "init": "init_flow",  # 离退休提取(初始) → nextStage=start
            "kb": "kb_answer",  # 其他 → 公积金知识回答
        },
    )
    wf.add_edge("init_flow", "init_out")
    wf.add_edge("init_out", "__end__")
    wf.add_edge("kb_answer", "__end__")

    wf.add_edge("stage_start", "start_out")
    wf.add_edge("start_out", "__end__")

    wf.add_edge("flag", "http")
    wf.add_edge("http", "http_judge")
    wf.add_conditional_edge("http_judge", {"ok": "succ_flow", "fail": "resp"})
    wf.add_edge("succ_flow", "succ_out")
    wf.add_edge("succ_out", "__end__")
    wf.add_edge("resp", "err_flag")
    wf.add_edge("err_flag", "err_out")
    wf.add_edge("err_out", "err_ans")
    wf.add_edge("err_ans", "__end__")

    # userInput 交互链: 用户输入 → 更新 inputStr → 回传前端
    wf.add_edge("ask_input", "set_input")
    wf.add_edge("set_input", "input_out")
    wf.add_edge("input_out", "__end__")

    # 结束/退出共用: 单节点清空变量并返回 nextStage=end
    wf.add_edge("end_out", "__end__")

    return wf


# ---------------------------------------------------------------------------
# 模拟第三方服务(演示 HTTP 成功/失败分支, 非业务必需)
# ---------------------------------------------------------------------------
def run_mock_third_party(host: str = "0.0.0.0", port: int = 9080) -> None:
    """启动模拟第三方服务: POST /test1 校验社保卡.

    flag=true  → code=FBASE0000(成功)
    flag=false → code=FBASE9999, data.respMsg=社保卡校验失败(失败)

    Args:
        host/port: 监听地址(默认 0.0.0.0:9080).
    """
    import uvicorn
    from fastapi import FastAPI

    app = FastAPI(title="Mock Third-Party (test1)")

    @app.post("/test1")
    def check_social_security_card(payload: dict[str, Any]) -> dict[str, Any]:
        flag = str(payload.get("flag", "")).lower() == "true"
        if flag:
            return {"code": THIRD_PARTY_SUCCESS_CODE, "data": {"respMsg": "社保卡校验通过"}}
        return {"code": "FBASE9999", "data": {"respMsg": "社保卡校验失败, 请核对信息"}}

    uvicorn.run(app, host=host, port=port)


# ---------------------------------------------------------------------------
# 入口: 启动网关(或模拟第三方)
# ---------------------------------------------------------------------------
def main() -> None:
    """入口: 解析参数后启动网关或模拟第三方服务."""
    import argparse

    parser = argparse.ArgumentParser(description="智能客服·离退休提取流程示例")
    parser.add_argument(
        "--mock-third-party",
        action="store_true",
        help="启动模拟第三方服务(localhost:9080), 而非网关",
    )
    args = parser.parse_args()

    if args.mock_third_party:
        run_mock_third_party()
        return

    api_key = os.environ.get("DEEPSEEK_API_KEY")
    if not api_key:
        raise SystemExit("请设置环境变量 DEEPSEEK_API_KEY 以真实调用 DeepSeek API")

    chat_model = DeepSeekClient(api_key=api_key)
    wf = build_flow(chat_model)
    app = wf.compile()
    print("已编译工作流:", wf.name)
    print("注册应用 appId=app-withdraw, 启动网关 http://0.0.0.0:8080")
    serve({"app-withdraw": app})


if __name__ == "__main__":
    main()

---

回到示例索引。