用 Dify 搭建企业级 AI Agent:从 Demo 到生产环境的全流程

适用读者:AI 应用工程师、后端开发、技术 Leader

技术栈:Dify 0.6+ / Python 3.10+ / Docker / PostgreSQL / Redis

阅读时长:约 15 分钟

---

一、为什么选择 Dify?

在搭建 AI Agent 时,常见选择有:

平台/框架	上手难度	灵活度	适合场景
Coze	低	中	C 端轻量 Bot
Dify	中	高	企业级 Agent 应用
LangChain	高	极高	复杂定制场景
n8n + LLM	中	中	工作流自动化

Dify 的核心优势:

• BaaS + LLMOps 一体化:自带知识库、模型管理、监控、日志
• 可视化编排 + 代码可扩展:非技术人员也能上手,开发者可深入定制
• API 优先:所有能力都暴露为 API,便于嵌入企业系统
• 支持私有化部署:数据不出门,满足合规要求

---

二、Dify 核心概念

┌─────────────────────────────────────────────┐
│                  Dify 工作台                  │
│                                             │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │  应用层   │  │  知识库   │  │  工具集   │  │
│  │ Chatflow  │  │ RAG 数据  │  │ HTTP/    │  │
│  │ Workflow  │  │ 向量索引   │  │ Function │  │
│  └──────────┘  └──────────┘  └──────────┘  │
│                                             │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │  模型管理 │  │  监控日志 │  │  API 网关 │  │
│  │ GPT/Claude│  │ 调用追踪  │  │ 应用密钥  │  │
│  │ 国产模型 │  │ 成本统计   │  │ 限流鉴权  │  │
│  └──────────┘  └──────────┘  └──────────┘  │
└─────────────────────────────────────────────┘

概念	说明
应用	一个独立的 AI 服务,分为 Chatflow(对话型)和 Workflow(自动化型)
知识库	文档上传 → 自动分块 → Embedding → 检索
工具(Tools)	Agent 可调用的外部能力:HTTP 请求、SQL 查询、自定义函数
变量	用户输入、系统参数,可在工作流中传递
节点	工作流的最小单元:LLM、知识检索、条件分支、代码执行等

---

三、环境准备:Docker 部署 Dify

1. 拉取 Dify 源码

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

2. 配置环境变量

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

# 修改关键配置
# 强制使用自带的 PostgreSQL / Redis
sed -i 's/^EXPOSE_NGINX=.*/EXPOSE_NGINX=80/' .env
sed -i 's/^EXPOSE_NGINX_PORT=.*/EXPOSE_NGINX_PORT=80/' .env

3. 启动服务

# 一键启动(包含 API、Web、PostgreSQL、Redis、向量库)
docker compose up -d

# 查看启动状态
docker compose ps

启动后访问 http://localhost/install 完成初始化,创建管理员账号。

---

四、第一个 Agent:智能客服 Demo

场景描述

电商客服 Agent:能查询订单状态、回答常见问题,必要时转人工。

1. 创建应用

进入 Dify 控制台 → 工作室 → 创建空白应用 → Chatflow。

2. 配置 LLM 节点(在可视化界面完成)

用户输入 -> [问题分类器] -> 分支 1:订单查询
                          -> 分支 2:常见问题(走知识库)
                          -> 分支 3:转人工

可视化编排后,Dify 会自动生成可调用 API,我们重点关注后端如何集成。

3. 通过 Python SDK 调用 Dify

# dify_client.py
# 封装 Dify API 调用,方便在业务系统中复用
import requests
import time
from typing import Iterator


class DifyClient:
    # Dify 服务的基础地址
    BASE_URL = "http://localhost/v1"

    def __init__(self, api_key: str, user_id: str = "default-user"):
        # api_key 在 Dify 控制台 -> 应用 -> API 访问 中获取
        self.api_key = api_key
        self.user_id = user_id
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        })

    def chat(self, query: str, conversation_id: str = "",
             inputs: dict = None) -> dict:
        # 发送对话消息(阻塞模式,适合同步业务)
        # 参数说明:
        #   query: 用户输入的问题
        #   conversation_id: 上下文会话 ID(空表示新会话)
        #   inputs: 传入应用的变量(如用户ID、订单号等)
        payload = {
            "inputs": inputs or {},
            "query": query,
            "response_mode": "blocking",   # 阻塞模式,直接返回完整结果
            "conversation_id": conversation_id,
            "user": self.user_id,
        }
        resp = self.session.post(
            f"{self.BASE_URL}/chat-messages",
            json=payload,
            timeout=60,
        )
        resp.raise_for_status()
        return resp.json()

    def chat_stream(self, query: str, conversation_id: str = "",
                    inputs: dict = None) -> Iterator[str]:
        # 流式对话(SSE),适合 Web 前端打字机效果
        payload = {
            "inputs": inputs or {},
            "query": query,
            "response_mode": "streaming",
            "conversation_id": conversation_id,
            "user": self.user_id,
        }
        with self.session.post(
            f"{self.BASE_URL}/chat-messages",
            json=payload,
            stream=True,
            timeout=60,
        ) as resp:
            resp.raise_for_status()
            for line in resp.iter_lines():
                if line:
                    # SSE 格式: data: {...}
                    if line.startswith(b"data:"):
                        yield line.decode("utf-8").lstrip("data: ")


if __name__ == "__main__":
    # 初始化客户端(api_key 替换为实际值)
    client = DifyClient(api_key="app-xxxxxxxxxxxxxxxx")

    # 同步调用示例
    result = client.chat(
        query="我的订单 20240701001 状态是什么?",
        inputs={"order_id": "20240701001"},
    )
    print("回答:", result.get("answer"))
    print("会话 ID:", result.get("conversation_id"))

4. 在 Web 后端中集成(以 FastAPI 为例)

# app.py
# 一个最小可用的 AI 客服 API 服务
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from dify_client import DifyClient

app = FastAPI(title="AI 客服 API")

# 应用启动时创建全局 Dify 客户端(单例)
dify = DifyClient(
    api_key="app-xxxxxxxxxxxxxxxx",
    user_id="web-user",
)


class ChatRequest(BaseModel):
    # 用户 ID(用于多租户隔离)
    user_id: str
    # 用户消息
    message: str
    # 上下文会话 ID(可选,用于多轮对话)
    conversation_id: str = ""
    # 透传给 Dify 的变量
    inputs: dict = {}


class ChatResponse(BaseModel):
    answer: str
    conversation_id: str


@app.post("/api/chat", response_model=ChatResponse)
def chat(req: ChatRequest):
    # 1. 调用 Dify
    # 2. 把 user_id 传给 Dify,便于在 Dify 后台按用户追踪用量
    dify.user_id = req.user_id
    try:
        result = dify.chat(
            query=req.message,
            conversation_id=req.conversation_id,
            inputs=req.inputs,
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Dify 调用失败:{e}")

    # 3. 返回给前端
    return ChatResponse(
        answer=result.get("answer", ""),
        conversation_id=result.get("conversation_id", ""),
    )


# 启动:uvicorn app:app --host 0.0.0.0 --port 8000

---

五、给 Agent 添加工具:Function Calling

Dify 支持两种工具接入方式:HTTP 工具和自定义函数。这里演示如何接入企业内部的订单查询 API。

1. 在 Dify 中配置 HTTP 工具

进入 工具 → 自定义工具 → 创建自定义工具,填写:

{
  "name": "query_order",
  "description": "根据订单号查询订单状态、物流信息",
  "method": "POST",
  "url": "http://backend.internal/api/order/query",
  "parameters": [
    {
      "name": "order_id",
      "type": "string",
      "required": true,
      "description": "订单号,例如 20240701001"
    }
  ],
  "headers": [
    {
      "name": "Authorization",
      "value": "Bearer ${INTERNAL_API_KEY}"
    }
  ]
}

2. 在工作流中引用工具

可视化编排中,给 LLM 节点添加 query_order 工具。LLM 会自动判断何时调用该工具。

3. 工具的后端实现示例

# order_api.py
# Dify 调用的订单查询接口(部署在内网)
from fastapi import FastAPI, HTTPException, Header
from pydantic import BaseModel

app = FastAPI()

# 内部服务密钥(与 Dify 工具配置中一致)
INTERNAL_API_KEY = "internal-secret-xxx"


class OrderQuery(BaseModel):
    order_id: str


class OrderInfo(BaseModel):
    order_id: str
    status: str        # 订单状态:已支付/已发货/已签收
    logistics: str     # 物流信息


# 模拟数据库(生产环境换成 SQLAlchemy/MySQL)
MOCK_ORDERS = {
    "20240701001": OrderInfo(
        order_id="20240701001",
        status="已发货",
        logistics="顺丰快递 SF1234567890"
    ),
}


@app.post("/api/order/query", response_model=OrderInfo)
def query_order(
    body: OrderQuery,
    authorization: str = Header(...),
):
    # 1. 鉴权:校验内部服务密钥
    if authorization != f"Bearer {INTERNAL_API_KEY}":
        raise HTTPException(status_code=401, detail="Unauthorized")

    # 2. 查询订单
    order = MOCK_ORDERS.get(body.order_id)
    if not order:
        raise HTTPException(status_code=404, detail="订单不存在")

    return order

生产提示:实际部署时,这个接口应该在内网,且加入 IP 白名单、限流等防护。

---

六、生产环境必做的 5 件事

1. HTTPS + Nginx 反向代理

# /etc/nginx/conf.d/dify.conf
server {
    listen 443 ssl;
    server_name ai.example.com;

    ssl_certificate     /etc/nginx/ssl/cert.pem;
    ssl_certificate_key /etc/nginx/ssl/key.pem;

    # 客户端最大请求体(用于文档上传)
    client_max_body_size 100M;

    location / {
        proxy_pass http://127.0.0.1:80;   # Dify 内部服务
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # SSE 流式响应需要关闭缓冲
        proxy_buffering off;
        proxy_cache off;
    }
}

2. 多租户隔离 + 用户配额

# multi_tenant.py
# 多租户场景下,为每个企业分配独立的 API Key + 配额
from dataclasses import dataclass


@dataclass
class TenantConfig:
    tenant_id: str
    api_key: str
    daily_quota: int        # 每日最大调用次数
    rate_limit_per_min: int  # 每分钟限流


# 配置中心读取(Nacos/Apollo/数据库)
TENANTS = {
    "company_a": TenantConfig(
        tenant_id="company_a",
        api_key="app-xxx-a",
        daily_quota=10000,
        rate_limit_per_min=60,
    ),
    "company_b": TenantConfig(
        tenant_id="company_b",
        api_key="app-xxx-b",
        daily_quota=5000,
        rate_limit_per_min=30,
    ),
}

# 调用时根据租户路由
def get_dify_client(tenant_id: str) -> DifyClient:
    cfg = TENANTS.get(tenant_id)
    if not cfg:
        raise ValueError(f"未知租户:{tenant_id}")
    return DifyClient(api_key=cfg.api_key, user_id=tenant_id)

3. 调用日志与监控

# metrics.py
# 记录每次 Dify 调用的关键指标,接入 Prometheus
import time
from prometheus_client import Counter, Histogram

# 指标定义
dify_requests_total = Counter(
    "dify_requests_total",
    "Dify 请求总数",
    ["tenant", "status"],   # 标签:租户、状态
)

dify_latency_seconds = Histogram(
    "dify_latency_seconds",
    "Dify 调用耗时(秒)",
    ["tenant"],
)


def call_with_metrics(client: DifyClient, query: str, tenant: str):
    # 计时 + 计数,实现可观测性
    start = time.time()
    try:
        result = client.chat(query=query)
        dify_requests_total.labels(tenant=tenant, status="success").inc()
        return result
    except Exception as e:
        dify_requests_total.labels(tenant=tenant, status="error").inc()
        raise
    finally:
        dify_latency_seconds.labels(tenant=tenant).observe(
            time.time() - start
        )

4. 失败重试 + 降级

# retry.py
# LLM 调用不稳定,生产环境必须加重试
import time
from functools import wraps


def retry(max_retries: int = 3, delay: float = 1.0):
    # 装饰器:自动重试,避免瞬时失败影响业务
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_err = None
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_err = e
                    if i < max_retries - 1:
                        time.sleep(delay * (2 ** i))   # 指数退避
            raise last_err
        return wrapper
    return decorator


class RobustDifyClient(DifyClient):
    # 健壮版客户端:重试 + 降级
    @retry(max_retries=3, delay=1.0)
    def chat(self, query: str, **kwargs) -> dict:
        return super().chat(query, **kwargs)

    def chat_with_fallback(self, query: str, **kwargs):
        # 主调用失败时,降级到备用模型
        try:
            return self.chat(query, **kwargs)
        except Exception:
            # 降级:返回预设答案或转人工
            return {
                "answer": "抱歉,系统繁忙,请稍后再试或联系人工客服。",
                "conversation_id": "",
            }

5. 定期备份知识库与配置

# backup.sh
# 每日凌晨备份 Dify 数据库
#!/bin/bash
BACKUP_DIR=/data/dify-backup/$(date +%Y%m%d)
mkdir -p $BACKUP_DIR

# 备份 PostgreSQL(存储应用、知识库、对话记录)
docker exec dify-postgres pg_dump -U postgres dify > $BACKUP_DIR/dify.sql

# 备份上传的原始文档
docker cp dify-api:/app/api/storage $BACKUP_DIR/storage

# 保留 7 天
find /data/dify-backup -mtime +7 -type d | xargs rm -rf

加入 crontab:

# 每天凌晨 3 点执行备份
0 3 * * * /opt/scripts/backup.sh

---

七、6 个常见踩坑

坑 1:忽略知识库分块策略

症状:检索出来的片段不完整,答案断章取义。

解决:在 Dify 的知识库设置中,根据文档类型选择分块模式:

• 短文本(FAQ):通用模式
• 长文档:父子分段(检索父块、返回子块,兼顾上下文)
• 代码/PDF:开启 PDF 转 Markdown 和 QA 切片

坑 2:Prompt 里有太多变量

症状:LLM 拿到一堆变量但不知道用哪个,回答混乱。

解决:

• 变量控制在 5 个以内
• 每个变量必须有清晰说明
• 在 Prompt 模板中显式指定使用时机

用户问题:{query}
用户姓名:{user_name}    # 仅在用户问"我的信息"时使用
订单号:{order_id}        # 仅在用户问订单相关时使用

坑 3:工具描述写得太简单

症状:Agent 该调用工具时不调,不该调时乱调。

解决:工具描述必须包含何时使用、参数含义、返回格式:

## 工具名:query_order
## 描述:当且仅当用户明确询问"我的订单"或提供订单号时调用。
## 参数:
  - order_id: 订单号,通常是 12 位数字
## 返回:JSON,包含 status(订单状态)和 logistics(物流)

坑 4:没做并发限制,被刷爆

症状:上线后 token 费用暴涨,后端响应慢。

解决:

• 在 Dify 控制台 → 设置 → 模型供应商 中开启限流
• 在自己的后端加 Redis 限流(参考第六节)
• 设置单日/单用户配额

坑 5:对话历史无限增长

症状:数据库膨胀,查询变慢。

解决:在 Dify 中开启对话保留策略:

# cleanup_conversations.py
# 定时清理过期对话(超过 30 天)
import requests
from datetime import datetime, timedelta

API_BASE = "http://localhost/v1"
ADMIN_TOKEN = "your-admin-token"   # 管理员 token


def cleanup_old_conversations(days: int = 30):
    # 通过 Dify 管理 API 清理过期对话,释放存储
    cutoff = datetime.now() - timedelta(days=days)
    resp = requests.get(
        f"{API_BASE}/console/api/apps",
        headers={"Authorization": f"Bearer {ADMIN_TOKEN}"},
    )
    for app in resp.json().get("data", []):
        # 删除过期对话(具体接口按 Dify 版本调整)
        requests.delete(
            f"{API_BASE}/console/api/apps/{app['id']}/conversations",
            headers={"Authorization": f"Bearer {ADMIN_TOKEN}"},
            json={"before": cutoff.isoformat()},
        )

坑 6:用 Dify 直接对外提供服务

症状:Dify 自带的 Web 界面被搜索引擎/竞争对手发现,直接套用。

解决:永远不要把 Dify 默认 Web 端暴露给最终用户。

• 通过自己的后端调用 Dify API(参考第四节)
• 在 Nginx 中禁止 /chat /completion 等端点的公网访问
• Dify 仅作为内部 AI 中台,UI 由企业自己定制

---

八、最佳实践总结

维度	建议
部署	Docker Compose 一键起,生产用 Kubernetes + HTTPS
API 集成	自己的后端封装一层,不要让前端直连 Dify
多租户	按企业分配 API Key,后端路由 + Redis 限流
监控	Prometheus 埋点 + Dify 自带日志 + 告警
稳定性	失败重试 + 降级答案 + 限流保护
知识库	选对分块策略,定期清理过期文档
工具描述	写清"何时调用"和"参数含义",Agent 才会用
安全	内部服务加 IP 白名单 + 鉴权 + 审计日志
备份	数据库 + 原始文件每日备份,保留 7~30 天
升级	订阅 Dify Release,先在测试环境验证再升生产

---

九、总结

Dify 把"搭 AI 应用"从写代码变成了配工作流,但从 Demo 到生产,关键不是 Dify 本身,而是:

1. 工程化:API 封装、限流、监控、备份
2. 可观测:调用日志、成本统计、效果评估
3. 稳定性:重试、降级、限流、灾备
4. 安全合规:多租户隔离、私有化部署、审计

掌握这些,你就能把 Dify 从"玩具"变成"生产力工具"。

---

参考文档

• Dify 官方文档:https://docs.dify.ai/
• Dify GitHub:https://github.com/langgenius/dify
• Dify API 参考:https://docs.dify.ai/api-reference/