一个周末上线企业知识库问答：Dify 开源平台实战与踩坑全记录

# docker-compose.yaml（精简版，完整版去官方仓库拿）
services:
  api:
    image: langgenius/dify-api:0.15.3
    restart: always
    environment:
      # 模型供应商的密钥在界面里配，这里配的是基础组件
      - DB_PASSWORD=difyai123456
      - REDIS_PASSWORD=difyai123456
      # 向量库可以选：weaviate(默认) / qdrant / milvus
      - VECTOR_STORE=qdrant
      # 文件存储：本地 or S3
      - STORAGE_TYPE=opendal
      - OPENDAL_SCHEME=fs
      - OPENDAL_FS_ROOT=/storage
    depends_on:
      - db
      - redis
      - qdrant

  worker:
    image: langgenius/dify-api:0.15.3
    restart: always
    # 异步任务都在 worker 里跑：文档嵌入、批量导入
    # 知识库大了之后，worker 数量要加
    deploy:
      replicas: 2

  web:
    image: langgenius/dify-web:0.15.3
    restart: always
    ports:
      - "80:3000"

  qdrant:
    image: langgenius/qdrant:v1.7.3
    restart: always
    volumes:
      - ./volumes/qdrant:/qdrant/storage

# HR-IT助手.yml（Dify DSL，有删减）
app:
  name: HR-IT智能助手
  mode: advanced-chat   # chatflow 类型
  version: 0.1.5

model:
  provider: openai_api_compatible
  model: qwen2.5-14b-instruct
  parameters:
    temperature: 0.2     # 制度问答要稳定，温度调低

workflow:
  graph:
    nodes:
      - id: classify
        type: llm
        data:
          prompt_template:
            - role: system
              content: |
                你是企业内部问题分类器。判断用户问题属于哪一类：
                - hr：考勤、请假、薪酬、福利、入职离职等制度问题
                - it：电脑、网络、账号、软件等技术人员问题
                - other：闲聊或无法判断
                只输出一个词：hr / it / other，不要解释。

      - id: if-branch
        type: if-else
        data:
          conditions:
            - variable_selector: [classify, text]
              comparison_operator: contains
              value: hr

      - id: hr-retrieval
        type: knowledge-retrieval
        data:
          dataset_ids:
            - hr-policy-dataset-id
          retrieval_mode: multiple      # 多路召回
          multiple_retrieval_config:
            top_k: 4
            score_threshold: 0.5        # 相似度阈值，低于0.5的不要
            reranking_enable: true      # 开启重排
            reranking_model:
              model: bge-reranker-large

      - id: hr-answer
        type: llm
        data:
          prompt_template:
            - role: system
              content: |
                你是企业HR制度问答助手。严格依据下方检索到的制度内容回答，
                并在回答末尾注明来源文档和章节。
                如果检索内容无法回答问题，只输出：【NOT_CONFIDENT】
                不要编造制度内容。

                检索到的制度内容：
                {{#hr-retrieval.result#}}

      - id: confidence-check
        type: if-else
        data:
          conditions:
            - variable_selector: [hr-answer, text]
              comparison_operator: not contains
              value: NOT_CONFIDENT

      - id: create-ticket
        type: http-request
        data:
          method: post
          url: https://itsm.internal.example.com/api/v1/tickets
          headers:
            Authorization: Bearer {{#env.ITS_TOKEN#}}
          body:
            type: json
            data: |
              {
                "title": "{{#sys.query#}}",
                "source": "ai_assistant",
                "category": "auto",
                "description": "AI助手无法回答，用户原问题：{{#sys.query#}}"
              }

# Dify Code 节点：加工工单创建结果
# 输入变量：ticket_response (object) - HTTP节点返回的工单系统响应
# 输出变量：reply (string), ticket_id (string)

import json

def main(ticket_response: dict) -> dict:
    """
    把工单系统的原始响应加工成用户能看懂的话
    业务规则：
    1. 创建成功 → 告知单号和预计响应时间
    2. 创建失败 → 降级提示，给出人工渠道
    """
    try:
        code = ticket_response.get("code", -1)
        data = ticket_response.get("data", {})

        if code == 0 and data.get("ticket_id"):
            ticket_id = data["ticket_id"]
            # 工单系统返回的 SLA：P4问题 8 小时内响应
            sla_hours = data.get("sla_hours", 8)

            reply = (
                f"这个问题我暂时没有把握，已经帮你转给专业同事了。\n\n"
                f"📋 工单号：{ticket_id}\n"
                f"⏱️ 预计 {sla_hours} 小时内会有同事联系你。\n"
                f"着急的话也可以直接打 IT 服务台：内线 8888。"
            )
            return {"reply": reply, "ticket_id": ticket_id}
        else:
            # 工单系统异常，降级到人工渠道
            reply = (
                "不好意思，工单系统好像开小差了，没能自动帮你建单。\n"
                "你可以直接打内线 8888 找 IT 服务台，或者发邮件到 "
                "it-help@example.com，说明下问题就行。"
            )
            return {"reply": reply, "ticket_id": ""}

    except Exception as e:
        return {
            "reply": "系统出了点小状况，请联系 IT 服务台（内线 8888）。",
            "ticket_id": ""
        }

import requests
import json

class DifyChatClient:
    """业务系统对接 Dify 应用的客户端"""

    def __init__(self, base_url, api_key):
        # api_key 在 Dify 应用的"访问 API"页面生成
        # 注意：每个应用一个 key，别搞混
        self.url = f"{base_url}/v1/chat-messages"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        # 会话管理：同一个用户的连续对话要用同一个 conversation_id
        self._user_conversations = {}

    def chat(self, user_id: str, message: str, stream: bool = False):
        """
        发送用户消息
        - user_id: 你业务系统里的用户标识，Dify 用它做限流和日志归组
        - message: 用户输入
        """
        payload = {
            "inputs": {},           # 如果工作流定义了开始变量，在这里传
            "query": message,
            "response_mode": "streaming" if stream else "blocking",
            "user": user_id,
        }

        # 关键：带上 conversation_id 才能延续多轮对话
        conv_id = self._user_conversations.get(user_id)
        if conv_id:
            payload["conversation_id"] = conv_id

        if not stream:
            resp = requests.post(self.url, headers=self.headers,
                                 json=payload, timeout=60)
            resp.raise_for_status()
            data = resp.json()

            # 记下 conversation_id，下次带上
            self._user_conversations[user_id] = data["conversation_id"]

            return {
                "answer": data["answer"],
                "conversation_id": data["conversation_id"],
                # 元数据里有引用的知识库片段，前端可以展示"来源"
                "metadata": data.get("metadata", {}),
            }
        else:
            return self._stream_chat(payload)

    def _stream_chat(self, payload):
        """流式响应：打字机效果必备"""
        resp = requests.post(self.url, headers=self.headers,
                             json=payload, stream=True, timeout=120)
        for line in resp.iter_lines():
            if not line:
                continue
            line = line.decode("utf-8")
            if line.startswith("data: "):
                chunk = json.loads(line[6:])
                event = chunk.get("event")

                if event == "message":
                    yield {"type": "text", "content": chunk["answer"]}

                elif event == "message_end":
                    # 对话结束时拿到 conversation_id 和引用
                    yield {
                        "type": "end",
                        "conversation_id": chunk.get("conversation_id"),
                        "metadata": chunk.get("metadata", {}),
                    }

                elif event == "error":
                    yield {"type": "error", "content": chunk.get("message")}


# ===== 企业微信回调里的用法示例 =====
def on_wecom_message(msg):
    client = DifyChatClient(
        base_url="http://dify.internal.example.com",
        api_key="app-xxxxxxxx"   # 应用级 API Key
    )

    result = client.chat(
        user_id=msg["from_user"],   # 用企微 userid 做标识
        message=msg["content"]
    )
    return result["answer"]