用 Dify 搭建企业级 AI 工作流：从需求拆解到自动化部署全流程实战

# 1. 克隆仓库
git clone https://github.com/langgenius/dify.git
cd dify/docker

# 2. 复制环境变量模板
cp .env.example .env

# 3.（可选）修改 .env，如关闭匿名遥测、调整端口
# EXPOSE_NGINX_PORT=8080
# CONSOLE_API_URL=http://localhost:8080

# 4. 启动全部服务（包括 api、worker、web、db、redis、weaviate 等）
docker compose up -d

# 5. 查看服务状态，确认均为 healthy / running
docker compose ps

Start(ticket_content)
   └─> Knowledge Retrieval(检索 FAQ 知识库)
          └─> LLM(生成回复 + 置信度评估, 输出结构化 JSON)
                 └─> IF/ELSE(置信度是否 >= 0.7)
                        ├─ YES ─> End(自动回复)
                        └─ NO  ─> HTTP Request(创建人工工单) ─> End(转人工提示)

{
  "app": {
    "name": "智能客服工单处理",
    "mode": "workflow",
    "description": "自动分类工单，命中知识库则自动回复，否则转人工"
  },
  "workflow": {
    "graph": {
      "nodes": [
        {
          "id": "start",
          "type": "custom",
          "data": {
            "type": "start",
            "variables": [
              { "variable": "ticket_content", "type": "text-input", "label": "工单内容" }
            ]
          }
        },
        {
          "id": "knowledge_1",
          "type": "custom",
          "data": {
            "type": "knowledge-retrieval",
            "dataset_ids": ["<your-dataset-id>"],
            "query_variable_selector": ["start", "ticket_content"],
            "retrieval_mode": "multiple",
            "multiple_retrieval_config": {
              "top_k": 3,
              "score_threshold": 0.5,
              "reranking_enable": false
            }
          }
        },
        {
          "id": "llm_1",
          "type": "custom",
          "data": {
            "type": "llm",
            "model": { "provider": "openai", "name": "gpt-4o-mini", "mode": "chat", "completion_params": { "temperature": 0.2 } },
            "prompt_template": [
              {
                "role": "system",
                "text": "你是客服助手。根据检索到的 FAQ 回答用户问题，并严格输出 JSON。\n格式：{\"reply\": \"回复内容\", \"confidence\": 0.0~1.0}\n若 FAQ 足以回答，confidence >= 0.7；若 FAQ 不足，confidence 设为 0。\n\nFAQ：{{#knowledge_1.result#}}\n用户问题：{{#start.ticket_content#}}"
              }
            ],
            "structured_output_enabled": true,
            "structured_output": {
              "type": "object",
              "properties": {
                "reply": { "type": "string" },
                "confidence": { "type": "number" }
              },
              "required": ["reply", "confidence"]
            }
          }
        },
        {
          "id": "if_else_1",
          "type": "custom",
          "data": {
            "type": "if-else",
            "cases": [
              {
                "logical_operator": "and",
                "conditions": [
                  {
                    "variable_selector": ["llm_1", "structured_output", "confidence"],
                    "comparison_operator": ">=",
                    "value": "0.7"
                  }
                ]
              }
            ]
          }
        },
        {
          "id": "http_1",
          "type": "custom",
          "data": {
            "type": "http-request",
            "method": "post",
            "url": "https://your-internal-api.example.com/tickets",
            "headers": "{\n  \"Content-Type\": \"application/json\",\n  \"Authorization\": \"Bearer {{internal_token}}\"\n}",
            "body": "{\n  \"content\": \"{{#start.ticket_content#}}\",\n  \"reason\": \"low_confidence_auto_routed\"\n}"
          }
        },
        {
          "id": "end_auto",
          "type": "custom",
          "data": {
            "type": "end",
            "outputs": [
              { "variable": "reply", "value_selector": ["llm_1", "structured_output", "reply"] },
              { "variable": "routed_to", "value": "auto" }
            ]
          }
        },
        {
          "id": "end_manual",
          "type": "custom",
          "data": {
            "type": "end",
            "outputs": [
              { "variable": "reply", "value": "您的问题较为复杂，已转交人工客服，工单号 {{#http_1.body#}}" },
              { "variable": "routed_to", "value": "manual" }
            ]
          }
        }
      ],
      "edges": [
        { "source": "start", "target": "knowledge_1" },
        { "source": "knowledge_1", "target": "llm_1" },
        { "source": "llm_1", "target": "if_else_1" },
        { "source": "if_else_1", "sourceHandle": "true", "target": "end_auto" },
        { "source": "if_else_1", "sourceHandle": "false", "target": "http_1" },
        { "source": "http_1", "target": "end_manual" }
      ]
    }
  }
}

curl -X POST 'http://localhost/v1/workflows/run' \
  -H 'Authorization: Bearer app-xxxxxxxxxxxxxxxx' \
  -H 'Content-Type: application/json' \
  -d '{
    "inputs": {
      "ticket_content": "我昨天买的耳机有杂音，能换吗？"
    },
    "response_mode": "blocking",
    "user": "customer-1024"
  }'

{
  "task_id": "9d1dxxxx-xxxx",
  "workflow_run_id": "dj2kxxxx-xxxx",
  "data": {
    "id": "dj2kxxxx",
    "workflow_id": "fldxxxx",
    "status": "succeeded",
    "outputs": {
      "reply": "您好，耳机在 7 天内出现非人为质量问题可申请换新，请提供订单号...",
      "routed_to": "auto"
    },
    "elapsed_time": 3.21,
    "total_tokens": 612,
    "total_steps": 6
  }
}

import json
import requests


class DifyWorkflowClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key

    def run(self, inputs: dict, user: str, response_mode: str = "blocking"):
        """阻塞模式：等待整个工作流跑完一次性返回结果。"""
        url = f"{self.base_url}/v1/workflows/run"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload = {
            "inputs": inputs,
            "response_mode": response_mode,
            "user": user,
        }
        resp = requests.post(url, headers=headers, json=payload, timeout=60)
        resp.raise_for_status()
        return resp.json()

    def run_stream(self, inputs: dict, user: str):
        """流式模式：通过 SSE 逐步接收节点事件，适合长流程或前端实时展示。"""
        url = f"{self.base_url}/v1/workflows/run"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }
        payload = {"inputs": inputs, "response_mode": "streaming", "user": user}
        with requests.post(url, headers=headers, json=payload, stream=True, timeout=60) as resp:
            for line in resp.iter_lines():
                if not line:
                    continue
                chunk = line.decode("utf-8")
                if chunk.startswith("data: "):
                    event = json.loads(chunk[6:])
                    yield event


if __name__ == "__main__":
    client = DifyWorkflowClient(
        base_url="http://localhost",
        api_key="app-xxxxxxxxxxxxxxxx",
    )

    # 阻塞调用
    result = client.run(
        inputs={"ticket_content": "我昨天买的耳机有杂音，能换吗？"},
        user="customer-1024",
    )
    print("回复内容：", result["data"]["outputs"]["reply"])
    print("路由去向：", result["data"]["outputs"]["routed_to"])
    print("耗时：", result["data"]["elapsed_time"], "秒")
    print("Token 消耗：", result["data"]["total_tokens"])

    # 流式调用示例
    for event in client.run_stream(
        inputs={"ticket_content": "订单 8899 还没发货"},
        user="customer-1025",
    ):
        # event["event"] 可能是 workflow_started / node_started / node_finished / text_chunk / workflow_finished
        if event.get("event") == "workflow_finished":
            print("\n最终输出：", event["data"]["outputs"])

pip install dify-client

from dify_client import WorkflowClient

client = WorkflowClient("app-xxxxxxxxxxxxxxxx")
client.base_url = "http://localhost/v1"

response = client.run(
    inputs={"ticket_content": "耳机有杂音想换货"},
    user="customer-1024",
    response_mode="blocking",
)
print(response.text)