Dify 工单分诊：Workflow DSL 与 API 集成

这篇做一个工单分诊工作流：输入一段用户反馈，模型输出 {"category": "账单|故障|功能建议|其他", "summary": "..."}。 工作流以 DSL（YAML）文件的形式放进 git，用脚本导入、发布、生成 API Key，再通过公开的 Service API /v1/workflows/run 调用（blocking 和 streaming 两种模式都做）。 另外配一个 8 条标注数据的评测脚本，每次改提示词后先跑一遍。

先看清成品（box 上真实运行，模型是本地 qwen2.5 1.5B）

• 导入状态 completed，发布 success，拿到 app- 开头的 API Key。
• blocking 调用约 1 秒返回，streaming 收到 workflow_started、node_started、text_chunk、node_finished、workflow_finished 五种事件。
• 评测准确率：初版 6/8，加分类定义后反而降到 5/8，加 few-shot 示例后连续两次 8/8。
• pytest 3 项通过。

合规提醒：Dify 采用附加条件的 Apache 2.0 许可证：未经书面授权不得用源码运营多租户服务，也不得移除或修改前端的 LOGO 和版权信息，详见仓库里的 LICENSE。自托管时，数据都留在你自己的机器上，但如果接入了云端模型，提示词和数据会发送给对应的服务商。管理员密码、应用 API Key、外部知识库 Key 都不要提交进仓库。本文不列 Dify Cloud 的价格，以官网为准。

关于控制台 API：初始化、装插件、导入 DSL 这几步用的是控制台前端调用的 /console/api/* 接口。它不是官方公开的稳定 API，本文按 1.17.1 的源码确认了路径和字段，升级后可能变化。同样的操作都可以在网页控制台里点完，脚本只是为了让环境可以重复搭建。对外集成请用公开的 /v1 Service API（B 篇）。

运行说明：全部在 box 上用 A 篇的本机 Dify 实际运行。8 条评测数据太少，8/8 不代表线上准确率，只能说明改动方向是对的。

做完之后你会得到什么？

需要准备哪些环境？

A 篇装好的本机 Dify（或者任意一个 Dify 1.17 实例）和 Python 3.10 以上。脚本只用标准库，测试需要 pytest。

文件结构

55-dify-B-src/
├── ticket_triage.yml
├── dify_console.py      # 同 A 篇
├── deploy_app.py
├── update_app.py
├── triage_client.py
├── eval_set.jsonl
├── eval_triage.py
└── test_triage.py

第 1 步：DSL

DSL 的结构参考了 Dify 仓库里的测试样例 api/tests/fixtures/workflow/basic_llm_chat_workflow.yml。模型的 provider 要写完整的插件路径 langgenius/ollama/ollama：

app:
  description: '工单分诊：把用户反馈归类并给出一句话摘要'
  icon: 🎫
  icon_background: '#FFEAD5'
  mode: workflow
  name: ticket-triage
  use_icon_as_answer_icon: false
dependencies: []
kind: app
version: 0.7.0
workflow:
  conversation_variables: []
  environment_variables: []
  features:
    file_upload:
      enabled: false
    opening_statement: ''
    retriever_resource:
      enabled: false
    sensitive_word_avoidance:
      enabled: false
    speech_to_text:
      enabled: false
    suggested_questions: []
    suggested_questions_after_answer:
      enabled: false
    text_to_speech:
      enabled: false
  graph:
    edges:
    - data: {isInIteration: false, isInLoop: false, sourceType: start, targetType: llm}
      id: start-to-llm
      source: start_node
      sourceHandle: source
      target: llm_node
      targetHandle: target
      type: custom
      zIndex: 0
    - data: {isInIteration: false, isInLoop: false, sourceType: llm, targetType: end}
      id: llm-to-end
      source: llm_node
      sourceHandle: source
      target: end_node
      targetHandle: target
      type: custom
      zIndex: 0
    nodes:
    - data:
        desc: ''
        title: 开始
        type: start
        variables:
        - label: 工单内容
          max_length: 2000
          options: []
          required: true
          type: paragraph
          variable: ticket
      height: 90
      id: start_node
      position: {x: 30, y: 227}
      positionAbsolute: {x: 30, y: 227}
      sourcePosition: right
      targetPosition: left
      type: custom
      width: 244
    - data:
        desc: ''
        title: 分诊
        type: llm
        model:
          provider: langgenius/ollama/ollama
          name: qwen2.5-local:1.5b
          mode: chat
          completion_params:
            temperature: 0
        prompt_template:
        - role: system
          text: |-
            你是客服工单分诊助手。只输出一行 JSON，不要输出其他内容。
            格式：{"category": "账单|故障|功能建议|其他", "summary": "不超过30字的中文摘要"}
            分类定义（按顺序判断）：
            - 账单：涉及扣费、退款、价格、续费、发票、订阅金额，即使用户在抱怨也归为账单
            - 故障：已有功能无法正常使用，如闪退、报错、无法登录
            - 功能建议：希望新增或改进功能
            - 其他：以上都不是
            示例：
            输入：我的会员怎么自动扣了年费？ 输出：{"category": "账单", "summary": "会员被自动扣年费"}
            输入：收据上的金额不对。 输出：{"category": "账单", "summary": "收据金额有误"}
            输入：上传图片一直转圈。 输出：{"category": "故障", "summary": "上传图片卡住"}
            输入：建议增加批量删除。 输出：{"category": "功能建议", "summary": "希望支持批量删除"}
        - role: user
          text: '{{#start_node.ticket#}}'
        vision: {enabled: false, configs: {variable_selector: []}}
        memory: {enabled: false, window: {enabled: false, size: 50}}
        context: {enabled: false, variable_selector: []}
        structured_output: {enabled: false}
        retry_config: {enabled: false, max_retries: 1, retry_interval: 1000}
      height: 90
      id: llm_node
      position: {x: 334, y: 227}
      positionAbsolute: {x: 334, y: 227}
      sourcePosition: right
      targetPosition: left
      type: custom
      width: 244
    - data:
        desc: ''
        outputs:
        - value_selector: [llm_node, text]
          value_type: string
          variable: result
        title: 结束
        type: end
      height: 90
      id: end_node
      position: {x: 638, y: 227}
      positionAbsolute: {x: 638, y: 227}
      sourcePosition: right
      targetPosition: left
      type: custom
      width: 244
    viewport: {x: 0, y: 0, zoom: 0.7}

第 2 步：导入、发布、生成 Key

"""导入 DSL -> 发布 -> 生成应用 API Key（写入 .env.app，不打印完整 Key）。"""
import os, pathlib, sys
from dify_console import Console

dsl = pathlib.Path(sys.argv[1] if len(sys.argv) > 1 else "ticket_triage.yml").read_text(encoding="utf-8")
c = Console(); c.login(os.environ["DIFY_EMAIL"], os.environ["DIFY_PASSWORD"])

st, imp = c.call("POST", "/apps/imports", {"mode": "yaml-content", "yaml_content": dsl})
print("导入:", st, imp.get("status"), imp.get("error") or "")
if imp.get("status") == "pending":          # DSL 版本比服务端旧/新时需要二次确认
    st, imp = c.call("POST", f"/apps/imports/{imp['id']}/confirm")
    print("确认导入:", st, imp.get("status"))
if imp.get("status") not in ("completed", "completed-with-warnings"):
    raise SystemExit(f"导入失败: {imp}")
app_id = imp["app_id"]

st, js = c.call("POST", f"/apps/{app_id}/workflows/publish", {"marked_name": "v1", "marked_comment": "from git"})
print("发布:", st, js.get("result") if isinstance(js, dict) else js)

st, key = c.call("POST", f"/apps/{app_id}/api-keys")
print("API Key:", st, key["token"][:8] + "…")
pathlib.Path(".env.app").write_text(f"DIFY_APP_ID={app_id}\nDIFY_APP_KEY={key['token']}\n")
os.chmod(".env.app", 0o600)

"""用同一份 DSL 覆盖已有应用（app_id 来自 .env.app），然后重新发布。API Key 不变。"""
import os, pathlib
from dify_console import Console

env = dict(l.strip().split("=", 1) for l in open(".env.app") if "=" in l)
dsl = pathlib.Path("ticket_triage.yml").read_text(encoding="utf-8")
c = Console(); c.login(os.environ["DIFY_EMAIL"], os.environ["DIFY_PASSWORD"])
st, imp = c.call("POST", "/apps/imports", {"mode": "yaml-content", "yaml_content": dsl, "app_id": env["DIFY_APP_ID"]})
print("覆盖导入:", st, imp.get("status"), imp.get("error") or "")
st, js = c.call("POST", f"/apps/{env['DIFY_APP_ID']}/workflows/publish", {"marked_name": "v2", "marked_comment": "分类定义"})
print("发布:", st, js)

第 3 步：调用 Service API

官方文档说明 /v1/workflows/run 只能运行已发布的工作流，支持 blocking（一次性返回）和 streaming（SSE 事件流）两种模式：

"""调用已发布的 Dify 工作流：/v1/workflows/run（blocking 与 streaming 两种模式）。"""
import json, os, sys, urllib.request, urllib.error

BASE = os.environ.get("DIFY_API_BASE", "http://localhost/v1")
UA = "binbho-tutorial/1.0"

def load_key():
    if os.environ.get("DIFY_APP_KEY"):
        return os.environ["DIFY_APP_KEY"]
    for line in open(".env.app", encoding="utf-8"):
        if line.startswith("DIFY_APP_KEY="):
            return line.split("=", 1)[1].strip()
    raise SystemExit("缺少 DIFY_APP_KEY")

def _req(body, key):
    return urllib.request.Request(BASE + "/workflows/run", data=json.dumps(body).encode(), method="POST",
        headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": UA})

def parse_result(text):
    """模型偶尔会包一层 ```json，先剥掉再解析；失败就返回 None 交给人工。"""
    t = text.strip()
    if t.startswith("```"):
        t = t.strip("`").removeprefix("json").strip()
    try:
        obj = json.loads(t)
    except ValueError:
        return None
    if obj.get("category") not in {"账单", "故障", "功能建议", "其他"}:
        return None
    return obj

def run_blocking(ticket, key, user="demo-user"):
    body = {"inputs": {"ticket": ticket}, "response_mode": "blocking", "user": user}
    try:
        with urllib.request.urlopen(_req(body, key), timeout=180) as r:
            data = json.load(r)["data"]
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode()[:300]}")
    return data

def run_streaming(ticket, key, user="demo-user"):
    body = {"inputs": {"ticket": ticket}, "response_mode": "streaming", "user": user}
    events = []
    with urllib.request.urlopen(_req(body, key), timeout=180) as r:
        for raw in r:
            line = raw.decode("utf-8").strip()
            if not line.startswith("data:"):
                continue
            ev = json.loads(line[5:])
            events.append(ev["event"])
            if ev["event"] == "workflow_finished":
                return events, ev["data"]
    raise RuntimeError(f"流在 workflow_finished 之前结束: {events}")

if __name__ == "__main__":
    key = load_key()
    ticket = sys.argv[1] if len(sys.argv) > 1 else "上个月被重复扣费两次，请退款。"
    d = run_blocking(ticket, key)
    print("blocking:", d["status"], f"{d['elapsed_time']:.1f}s", "tokens", d["total_tokens"])
    print("  原文:", d["outputs"]["result"])
    print("  解析:", parse_result(d["outputs"]["result"]))
    evs, d2 = run_streaming("App 打开就闪退，iOS 18。", key)
    print("streaming 事件:", sorted(set(evs)))
    print("  解析:", parse_result(d2["outputs"]["result"]))

第 4 步：评测与测试

{"ticket": "上个月被重复扣费两次，请退款。", "label": "账单"}
{"ticket": "发票抬头开错了，能重开吗？", "label": "账单"}
{"ticket": "App 打开就闪退，iOS 18。", "label": "故障"}
{"ticket": "登录时一直提示网络错误，但网络是好的。", "label": "故障"}
{"ticket": "希望能支持导出成 Excel。", "label": "功能建议"}
{"ticket": "能不能加一个深色模式？", "label": "功能建议"}
{"ticket": "你们公司地址在哪里？", "label": "其他"}
{"ticket": "续费价格怎么比去年贵了？", "label": "账单"}

"""对已发布工作流跑标注集，打印准确率和错例。temperature=0，但小模型仍可能波动，建议跑 2 次。"""
import json, sys
from triage_client import load_key, run_blocking, parse_result

key = load_key()
rows = [json.loads(l) for l in open("eval_set.jsonl", encoding="utf-8")]
ok = 0
for r in rows:
    out = parse_result(run_blocking(r["ticket"], key)["outputs"]["result"])
    got = out["category"] if out else "解析失败"
    ok += got == r["label"]
    if got != r["label"]:
        print(f"  ✗ {r['ticket']}  期望 {r['label']}  实际 {got}")
print(f"准确率 {ok}/{len(rows)}")
sys.exit(0 if ok / len(rows) >= float(sys.argv[1] if len(sys.argv) > 1 else 0.75) else 1)

from triage_client import parse_result

def test_plain_json():
    assert parse_result('{"category": "账单", "summary": "重复扣费"}')["category"] == "账单"

def test_fenced_json():
    assert parse_result('```json\n{"category": "故障", "summary": "闪退"}\n```')["category"] == "故障"

def test_reject_unknown_category_and_garbage():
    assert parse_result('{"category": "投诉", "summary": "x"}') is None
    assert parse_result("这是账单问题") is None

第 5 步：运行

$ python deploy_app.py
导入: 200 completed-with-warnings      ← 首次运行时 DSL 还写的是 0.3.1，见调试示例；改成 0.7.0 后变为 completed
发布: 200 success
API Key: 201 app-l9Ju…

$ python triage_client.py
blocking: succeeded 1.0s tokens 92
  原文: {"category": "故障", "summary": "上个月被重复扣费两次，请求退款。"}
  解析: {'category': '故障', 'summary': '上个月被重复扣费两次，请求退款。'}
streaming 事件: ['node_finished', 'node_started', 'text_chunk', 'workflow_finished', 'workflow_started']
  解析: {'category': '故障', 'summary': 'App打开闪退，iOS 18'}

（上面是初版提示词的真实输出。重复扣费被分到了“故障”，这就是下面调试示例要解决的问题。）

错误调用的真实返回：

# Key 错误
{"code":"unauthorized","message":"Access token is invalid","status":401}
# 少了必填输入 ticket
{"code":"invalid_param","message":"ticket is required in input form","status":400}

调试示例

分类不准：加了规则反而更差。 初版评测结果是 6/8，两条账单类工单被分错（“重复扣费”被分成故障，“续费变贵”被分成功能建议）。

• 第一次修改：在 system 提示词里加了四类的定义，并强调“抱怨扣费也归为账单”。重新导入、发布后跑了两次，都是 5/8，连“发票抬头开错”也被分成了“其他”。我先确认了改动确实已经生效：发布版本的 prompt_template 里能查到“分类定义”。所以问题不在发布，而是这个 1.5B 的小模型对长篇规则描述的理解有限。
• 第二次修改：保留定义，再加 4 条 few-shot 示例，示例内容特意和评测集不同，避免泄题。连续两次评测都是 8/8。

结论：对小模型来说，给示例比写规则管用。改提示词一定要用评测集验证，不能凭感觉。

导入状态一直是 completed-with-warnings，但 warnings 是空数组。 返回里有 "current_dsl_version": "0.7.0", "imported_dsl_version": "0.3.1"，说明状态来自 DSL 版本差异（我照抄样例写的是 0.3.1）。把 version 改成 0.7.0 后，状态变成 completed。

常见问题

模板规则

• 工作流以 DSL 的形式进 git，导入、发布都用脚本完成，环境之间才能保持一致。
• 模型输出一律先校验再使用，分类值要在白名单里。
• 每次改提示词都先跑评测集，低于阈值就不发布。

项目资源

DSL、发布脚本、评测集都放在一个仓库里，改提示词时有记录，跑完评测才发布。