上一篇 下一篇 分享链接 返回 返回顶部

Dify 上线后账单太高?这套降本方案和网关源码可以直接用

发布人:慈云数据-客服中心 发布时间:6 天前 阅读量:31

Dify 如何降低成本|附源码

在企业落地大模型应用的过程中,Dify 是一个非常受欢迎的开源 LLM 应用开发平台。它提供了工作流编排、Agent、知识库、Prompt 管理、API 发布、模型接入等能力,可以帮助团队快速构建智能客服、知识库问答、文本生成、数据分析助手、内部 Copilot 等应用。

但是,很多团队在真正把 Dify 用到生产环境后,会很快遇到一个现实问题:

大模型调用成本越来越高。

尤其是当应用访问量上升、知识库问答频繁、用户输入较长、上下文轮次较多、模型默认使用高规格模型时,Token 消耗会持续增长。如果不做成本优化,一个原本只是内部试点的应用,也可能在短时间内产生较高的 API 费用。

本文将从实际工程角度,系统讲解 Dify 如何降低成本,并提供一套可直接参考的源码示例,帮助你实现:

  • 多模型路由;
  • 低成本模型优先;
  • 高成本模型兜底;
  • Prompt 压缩;
  • 对话上下文裁剪;
  • 知识库检索优化;
  • Token 用量统计;
  • 简单缓存机制;
  • Dify API 调用封装。

一、Dify 成本主要来自哪里?

在优化之前,我们要先知道成本从哪里来。Dify 本身是开源平台,真正持续产生费用的通常是以下几类资源。

1. 大模型 API 调用费用

这是最主要的成本来源。

例如你在 Dify 中接入了:

  • OpenAI GPT-4o / GPT-4.1;
  • Claude;
  • Gemini;
  • DeepSeek;
  • 通义千问;
  • 智谱 GLM;
  • Moonshot;
  • 火山方舟;
  • Azure OpenAI;
  • 私有化大模型接口。

大模型费用通常按 Token 计费,包括:

  • 输入 Token;
  • 输出 Token;
  • 缓存 Token;
  • 推理 Token;
  • 多模态 Token。

如果用户每次问题都带上完整上下文、完整知识库内容、冗长 Prompt,那么一次请求就可能消耗大量 Token。

2. Embedding 向量化费用

如果你使用 Dify 知识库功能,上传文档后需要进行切分和向量化。

成本来自:

  • 文档切片数量;
  • 每个切片的长度;
  • 使用的 Embedding 模型;
  • 文档更新频率。

很多企业在早期没有设计知识库结构,把大量无效文档全部上传,导致切片数量激增,Embedding 成本和检索噪声同时增加。

3. Rerank 重排序费用

Dify 支持检索后重排序能力,Rerank 可以提升知识库问答质量,但部分 Rerank 模型也是按量计费的。

如果每次都对大量候选片段做重排序,成本也会增加。

4. 服务器与数据库成本

自部署 Dify 时,还会涉及:

  • API 服务;
  • Worker;
  • Web 前端;
  • PostgreSQL;
  • Redis;
  • 向量数据库;
  • 对象存储;
  • 日志系统;
  • 监控系统。

如果是中小规模应用,这部分成本一般低于模型成本,但在高并发场景下也需要优化。


二、降低 Dify 成本的核心思路

降低成本并不等于简单地换成便宜模型。真正成熟的策略应该是:

在不明显牺牲用户体验的前提下,把每一次请求分配到最合适、最低成本的资源上。

可以从以下几个方向优化。


三、策略一:模型分层,便宜模型优先

很多场景并不需要每次都调用最强模型。

例如:

  • 用户问“你是谁?”;
  • 用户问“帮我总结这段话”;
  • 用户问“把下面内容翻译成英文”;
  • 用户问“提取 JSON 字段”;
  • 用户问“重写一句营销文案”。

这些任务通常使用中小模型就可以完成。

而以下任务才适合使用更强模型:

  • 复杂推理;
  • 多步骤规划;
  • 高准确率法律、金融、医疗分析;
  • 复杂代码生成;
  • 多工具调用;
  • 高质量长文写作;
  • 高价值客户服务场景。

因此可以设计模型分层:

任务类型 推荐模型 成本
简单问答 小模型
摘要/翻译/改写 中小模型
知识库问答 中模型 + 检索
复杂推理 高级模型
高价值用户请求 高级模型

在 Dify 中,你可以为不同应用配置不同模型,也可以通过外部网关统一调用 Dify API,根据请求类型选择不同应用或不同模型。


四、策略二:构建模型路由网关

Dify 本身支持多应用、多模型配置。实际项目中,我更推荐在 Dify 前面加一个轻量级 AI Gateway,统一处理:

  • 用户鉴权;
  • 请求限流;
  • 模型路由;
  • 成本统计;
  • 缓存;
  • 日志;
  • 灰度;
  • 降级;
  • 风控。

整体架构如下:

用户 / 业务系统
      |
      v
AI Gateway 成本控制层
      |
      |-- 简单任务 --> Dify App A:低成本模型
      |
      |-- 知识库问答 --> Dify App B:中成本模型 + 知识库
      |
      |-- 复杂任务 --> Dify App C:高性能模型
      |
      |-- 命中缓存 --> 直接返回

这样的好处是:

  1. 不需要频繁改 Dify 内部代码;
  2. 可以按业务灵活分流;
  3. 可以快速切换模型供应商;
  4. 方便统计每个用户、每个应用、每种任务的成本;
  5. 可以做熔断和兜底。

五、策略三:控制上下文长度

大模型的成本通常和 Token 数量直接相关。很多 Dify 应用成本过高,是因为上下文没有被控制。

常见问题包括:

  • 每一轮都带上完整历史对话;
  • System Prompt 写得过长;
  • 知识库召回片段过多;
  • 每个片段内容过长;
  • 用户上传文档后直接全文塞给模型;
  • 工作流节点之间传递大量无用变量。

优化方式包括:

1. 限制历史轮数

一般客服类应用保留最近 3 到 5 轮即可。

用户最近问题 + 最近若干轮关键上下文 > 完整历史对话

2. 对历史对话做摘要

当对话超过一定长度时,可以把早期对话压缩成摘要。

例如:

历史摘要:
用户是企业 IT 管理员,正在咨询 Dify 私有化部署、知识库问答和成本优化方案。
用户已确认使用 PostgreSQL、Redis 和向量数据库,模型供应商为 DeepSeek。

然后后续对话只带摘要和最近几轮消息。

3. 压缩 Prompt

很多 Prompt 写得像说明书,包含大量重复描述。建议把 Prompt 控制在必要范围内。

例如原始 Prompt:

你是一个非常专业、非常耐心、非常友好、非常具有同理心的客服助手。
你需要根据用户的问题进行回答。
你回答时需要保持礼貌。
你不能编造内容。
你需要基于知识库回答。
如果知识库没有内容,你需要说明不知道。

可以压缩为:

你是企业客服助手。请基于知识库回答,保持礼貌、准确;若无依据,请说明无法确认,不要编造。

这类压缩看似节省不多,但在高频调用下非常有效。


六、策略四:优化知识库检索

知识库问答是 Dify 使用最多的场景之一,也是成本消耗大户。

1. 合理设置切片大小

如果切片太小,召回内容碎片化,模型难以理解;如果切片太大,每次塞给模型的上下文会变长。

建议:

  • 普通知识库:500~800 字/片;
  • 技术文档:800~1200 字/片;
  • FAQ:一问一答作为一个切片;
  • 法规制度:按章节切分。

2. 减少召回数量

很多人喜欢把 Top K 设置为 10 或 20,但实际大多数问答只需要 3~5 个片段。

建议:

Top K = 3~5
Score Threshold = 根据测试设置,例如 0.5~0.7

如果召回片段质量不足,再通过 Rerank 提升,而不是盲目增加数量。

3. 清理无效文档

知识库中不要放:

  • 重复文档;
  • 过期制度;
  • 格式混乱的扫描件;
  • 大量无标题内容;
  • 与业务无关的材料。

知识库质量越高,召回越准,传给模型的 Token 越少。

4. 对长文档做结构化处理

例如把制度文档整理为:

# 报销制度

## 适用范围
...

## 报销标准
...

## 审批流程
...

## 常见问题
...

比直接上传 PDF 扫描件效果更好,成本也更低。


七、策略五:增加缓存机制

很多企业内部问答有大量重复问题,例如:

  • “年假怎么计算?”
  • “VPN 怎么申请?”
  • “报销流程是什么?”
  • “Dify 怎么部署?”
  • “怎么重置密码?”

这些问题不需要每次都调用大模型。可以对用户问题进行归一化后缓存。

缓存命中时直接返回结果,成本几乎为零。

缓存可以分为:

  1. 精确缓存:问题完全相同;
  2. 语义缓存:问题意思相近;
  3. 知识库 FAQ 缓存:高频问题预先生成答案。

简单场景下,精确缓存已经很有用。


八、附源码:Dify 成本控制网关

下面提供一个基于 Python + FastAPI 的简化版 Dify 成本控制网关源码。它实现了:

  • 接收用户请求;
  • 简单任务分类;
  • 低成本与高性能 Dify 应用路由;
  • Redis 缓存;
  • Token 粗略估算;
  • 请求日志;
  • Dify API 调用封装。

注意:以下代码是可运行的工程骨架,实际生产环境需要增加鉴权、限流、异常告警、审计日志、敏感词过滤等能力。


九、项目结构

dify-cost-gateway/
├── main.py
├── config.py
├── dify_client.py
├── router.py
├── cache.py
├── token_counter.py
├── requirements.txt
└── .env.example

十、requirements.txt

fastapi==0.115.0
uvicorn==0.30.6
httpx==0.27.2
python-dotenv==1.0.1
redis==5.0.8
pydantic==2.8.2

十一、.env.example

REDIS_URL=redis://localhost:6379/0

DIFY_BASE_URL=https://api.dify.ai/v1

DIFY_LOW_COST_APP_KEY=app-low-cost-key
DIFY_KB_APP_KEY=app-knowledge-base-key
DIFY_HIGH_QUALITY_APP_KEY=app-high-quality-key

REQUEST_TIMEOUT=60
CACHE_TTL_SECONDS=3600

十二、config.py

import os
from dotenv import load_dotenv

load_dotenv()


class Settings:
    REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")

    DIFY_BASE_URL: str = os.getenv("DIFY_BASE_URL", "https://api.dify.ai/v1")

    DIFY_LOW_COST_APP_KEY: str = os.getenv("DIFY_LOW_COST_APP_KEY", "")
    DIFY_KB_APP_KEY: str = os.getenv("DIFY_KB_APP_KEY", "")
    DIFY_HIGH_QUALITY_APP_KEY: str = os.getenv("DIFY_HIGH_QUALITY_APP_KEY", "")

    REQUEST_TIMEOUT: int = int(os.getenv("REQUEST_TIMEOUT", "60"))
    CACHE_TTL_SECONDS: int = int(os.getenv("CACHE_TTL_SECONDS", "3600"))


settings = Settings()

十三、token_counter.py

这里为了演示,使用非常粗略的 Token 估算方式。生产环境建议使用模型对应的 tokenizer,例如 OpenAI 的 tiktoken,或者供应商提供的 Token 统计工具。

def estimate_tokens(text: str) -> int:
    """
    粗略估算 token 数。
    中文通常可以近似按 1 个汉字约等于 1 个 token,
    英文可以按 4 个字符约等于 1 个 token。
    这里只做简单估算,用于成本控制和日志分析。
    """
    if not text:
        return 0

    chinese_chars = 0
    other_chars = 0

    for ch in text:
        if "\u4e00" <= ch <= "\u9fff":
            chinese_chars += 1
        elif not ch.isspace():
            other_chars += 1

    return chinese_chars + other_chars // 4

十四、cache.py

import hashlib
import redis
from config import settings

redis_client = redis.Redis.from_url(settings.REDIS_URL, decode_responses=True)


def normalize_question(question: str) -> str:
    """
    简单的问题归一化。
    可以根据业务扩展:
    1. 去除多余空格;
    2. 统一大小写;
    3. 去除标点;
    4. 同义词归一;
    5. 接入语义向量缓存。
    """
    return " ".join(question.strip().lower().split())


def build_cache_key(user_id: str, question: str) -> str:
    normalized = normalize_question(question)
    raw = f"{user_id}:{normalized}"
    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()
    return f"dify:answer:{digest}"


def get_cache(user_id: str, question: str) -> str | None:
    key = build_cache_key(user_id, question)
    return redis_client.get(key)


def set_cache(user_id: str, question: str, answer: str, ttl: int | None = None):
    key = build_cache_key(user_id, question)
    redis_client.setex(key, ttl or settings.CACHE_TTL_SECONDS, answer)

十五、router.py

这个文件负责判断请求应该走哪个 Dify 应用。

实际生产中可以做得更复杂,例如:

  • 基于意图分类模型;
  • 基于关键词;
  • 基于用户等级;
  • 基于任务复杂度;
  • 基于预算;
  • 基于模型可用性;
  • 基于响应时延。
from enum import Enum
from token_counter import estimate_tokens


class RouteType(str, Enum):
    LOW_COST = "low_cost"
    KNOWLEDGE_BASE = "knowledge_base"
    HIGH_QUALITY = "high_quality"


LOW_COST_KEYWORDS = [
    "翻译",
    "总结",
    "改写",
    "润色",
    "提取",
    "格式化",
    "生成标题",
    "写摘要",
]

KB_KEYWORDS = [
    "制度",
    "流程",
    "报销",
    "年假",
    "请假",
    "入职",
    "离职",
    "vpn",
    "账号",
    "密码",
    "知识库",
    "文档",
]


COMPLEX_KEYWORDS = [
    "分析",
    "方案",
    "架构",
    "源码",
    "复杂",
    "推理",
    "优化",
    "排查",
    "设计",
    "对比",
]


def route_question(question: str, user_level: str = "normal") -> RouteType:
    """
    简单路由规则:
    1. VIP 用户更容易使用高质量模型;
    2. 知识库关键词走知识库应用;
    3. 简单任务走低成本模型;
    4. 长问题或复杂问题走高质量模型;
    """
    q = question.lower()
    token_count = estimate_tokens(question)

    if user_level == "vip" and token_count > 300:
        return RouteType.HIGH_QUALITY

    if any(keyword in q for keyword in KB_KEYWORDS):
        return RouteType.KNOWLEDGE_BASE

    if any(keyword in q for keyword in LOW_COST_KEYWORDS) and token_count < 800:
        return RouteType.LOW_COST

    if any(keyword in q for keyword in COMPLEX_KEYWORDS):
        return RouteType.HIGH_QUALITY

    if token_count > 1200:
        return RouteType.HIGH_QUALITY

    return RouteType.LOW_COST

十六、dify_client.py

Dify 的 API 通常可以通过 /chat-messages 调用。下面是一个非流式调用示例。

import httpx
from config import settings
from router import RouteType


def get_app_key(route_type: RouteType) -> str:
    if route_type == RouteType.LOW_COST:
        return settings.DIFY_LOW_COST_APP_KEY

    if route_type == RouteType.KNOWLEDGE_BASE:
        return settings.DIFY_KB_APP_KEY

    if route_type == RouteType.HIGH_QUALITY:
        return settings.DIFY_HIGH_QUALITY_APP_KEY

    raise ValueError(f"Unsupported route type: {route_type}")


async def call_dify_chat(
    route_type: RouteType,
    query: str,
    user_id: str,
    conversation_id: str | None = None,
    inputs: dict | None = None,
) -> dict:
    """
    调用 Dify Chat API。
    """
    app_key = get_app_key(route_type)

    if not app_key:
        raise RuntimeError(f"Dify app key is empty for route: {route_type}")

    url = f"{settings.DIFY_BASE_URL}/chat-messages"

    payload = {
        "inputs": inputs or {},
        "query": query,
        "response_mode": "blocking",
        "user": user_id,
    }

    if conversation_id:
        payload["conversation_id"] = conversation_id

    headers = {
        "Authorization": f"Bearer {app_key}",
        "Content-Type": "application/json",
    }

    async with httpx.AsyncClient(timeout=settings.REQUEST_TIMEOUT) as client:
        response = await client.post(url, json=payload, headers=headers)
        response.raise_for_status()
        return response.json()

十七、main.py

import time
from fastapi import FastAPI
from pydantic import BaseModel, Field

from cache import get_cache, set_cache
from dify_client import call_dify_chat
from router import route_question, RouteType
from token_counter import estimate_tokens


app = FastAPI(title="Dify Cost Gateway", version="1.0.0")


class ChatRequest(BaseModel):
    user_id: str = Field(..., description="用户 ID")
    query: str = Field(..., description="用户问题")
    conversation_id: str | None = Field(default=None, description="Dify 会话 ID")
    user_level: str = Field(default="normal", description="用户等级:normal/vip")
    enable_cache: bool = Field(default=True, description="是否启用缓存")


class ChatResponse(BaseModel):
    answer: str
    route_type: str
    cached: bool
    input_tokens_estimated: int
    cost_time_ms: int
    raw: dict | None = None


@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    start = time.time()

    input_tokens = estimate_tokens(req.query)

    # 1. 简单问题优先查缓存
    if req.enable_cache:
        cached_answer = get_cache(req.user_id, req.query)
        if cached_answer:
            return ChatResponse(
                answer=cached_answer,
                route_type="cache",
                cached=True,
                input_tokens_estimated=input_tokens,
                cost_time_ms=int((time.time() - start) * 1000),
                raw=None,
            )

    # 2. 选择路由
    route_type = route_question(req.query, req.user_level)

    # 3. 过长问题可以先降采样或提示用户
    if input_tokens > 6000 and route_type != RouteType.HIGH_QUALITY:
        route_type = RouteType.HIGH_QUALITY

    # 4. 调用 Dify
    result = await call_dify_chat(
        route_type=route_type,
        query=req.query,
        user_id=req.user_id,
        conversation_id=req.conversation_id,
    )

    answer = result.get("answer", "")

    # 5. 低成本与知识库答案可以缓存,高质量复杂推理答案谨慎缓存
    if req.enable_cache and route_type in [RouteType.LOW_COST, RouteType.KNOWLEDGE_BASE]:
        if answer:
            set_cache(req.user_id, req.query, answer)

    return ChatResponse(
        answer=answer,
        route_type=route_type.value,
        cached=False,
        input_tokens_estimated=input_tokens,
        cost_time_ms=int((time.time() - start) * 1000),
        raw=result,
    )


@app.get("/health")
async def health():
    return {"status": "ok"}

十八、启动方式

1. 启动 Redis

docker run -d \
  --name dify-cost-redis \
  -p 6379:6379 \
  redis:7

2. 安装依赖

pip install -r requirements.txt

3. 配置环境变量

cp .env.example .env

然后修改 .env

DIFY_BASE_URL=https://api.dify.ai/v1
DIFY_LOW_COST_APP_KEY=你的低成本应用Key
DIFY_KB_APP_KEY=你的知识库应用Key
DIFY_HIGH_QUALITY_APP_KEY=你的高质量应用Key

4. 启动服务

uvicorn main:app --host 0.0.0.0 --port 8080 --reload

5. 测试请求

curl -X POST http://localhost:8080/chat \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "u001",
    "query": "帮我总结下面这段内容:Dify 是一个开源大模型应用开发平台。",
    "user_level": "normal",
    "enable_cache": true
  }'

返回示例:

{
  "answer": "Dify 是一个用于开发大模型应用的开源平台。",
  "route_type": "low_cost",
  "cached": false,
  "input_tokens_estimated": 33,
  "cost_time_ms": 1280,
  "raw": {
    "event": "message",
    "message_id": "xxx",
    "conversation_id": "xxx",
    "answer": "Dify 是一个用于开发大模型应用的开源平台。"
  }
}

再次请求相同问题时,可能返回:

{
  "answer": "Dify 是一个用于开发大模型应用的开源平台。",
  "route_type": "cache",
  "cached": true,
  "input_tokens_estimated": 33,
  "cost_time_ms": 3,
  "raw": null
}

这说明缓存命中,没有再次调用 Dify 和大模型接口。


十九、进一步优化:加入预算控制

如果你希望控制每个用户每天最多消耗多少预算,可以增加预算模块。

例如:

# budget.py
import redis
from datetime import datetime
from config import settings

redis_client = redis.Redis.from_url(settings.REDIS_URL, decode_responses=True)


def get_budget_key(user_id: str) -> str:
    today = datetime.now().strftime("%Y%m%d")
    return f"dify:budget:{user_id}:{today}"


def increase_usage(user_id: str, tokens: int):
    key = get_budget_key(user_id)
    redis_client.incrby(key, tokens)
    redis_client.expire(key, 86400 * 2)


def get_usage(user_id: str) -> int:
    key = get_budget_key(user_id)
    value = redis_client.get(key)
    return int(value or 0)


def check_budget(user_id: str, max_tokens_per_day: int) -> bool:
    return get_usage(user_id) < max_tokens_per_day

main.py 中调用:

from budget import check_budget, increase_usage

MAX_TOKENS_PER_DAY = 50000

if not check_budget(req.user_id, MAX_TOKENS_PER_DAY):
    return ChatResponse(
        answer="你今天的 AI 使用额度已达到上限,请明天再试。",
        route_type="budget_limit",
        cached=False,
        input_tokens_estimated=input_tokens,
        cost_time_ms=int((time.time() - start) * 1000),
        raw=None,
    )

# 请求完成后记录用量
increase_usage(req.user_id, input_tokens + estimate_tokens(answer))

预算控制非常适合企业内部场景。例如:

  • 普通员工每天 5 万 Token;
  • 主管每天 20 万 Token;
  • VIP 客户不限制或单独计费;
  • 测试账号限制更低;
  • 非工作时间限制高成本模型调用。

二十、进一步优化:语义缓存

精确缓存只能命中完全相同的问题。语义缓存可以处理类似问题,例如:

报销流程是什么?
怎么申请报销?
费用报销要走什么流程?

它们意思接近,可以复用答案。

语义缓存的基本思路是:

  1. 对用户问题生成向量;
  2. 在缓存向量库中查找相似问题;
  3. 相似度超过阈值时,直接返回历史答案;
  4. 否则调用 Dify,并把问题和答案写入缓存。

伪代码如下:

def semantic_cache_search(question: str):
    query_embedding = embed(question)
    results = vector_db.search(query_embedding, top_k=1)

    if results and results[0].score > 0.92:
        return results[0].answer

    return None


def semantic_cache_save(question: str, answer: str):
    embedding = embed(question)
    vector_db.insert({
        "question": question,
        "answer": answer,
        "embedding": embedding,
    })

语义缓存特别适合:

  • HR 问答;
  • IT 帮助台;
  • 政策制度问答;
  • 产品 FAQ;
  • 售前客服;
  • 售后支持。

不过要注意,语义缓存不适合强时效性、强个性化的问题。例如:

  • “我这个月工资是多少?”
  • “我的订单现在到哪了?”
  • “今天股票应该买吗?”
  • “请根据我刚上传的文件分析。”

这类问题必须实时处理,不能简单复用旧答案。


二十一、进一步优化:工作流节点降本

Dify 工作流非常强大,但如果设计不当,也容易造成重复调用。

常见浪费包括:

  1. 每个节点都调用大模型;
  2. 分类节点使用高端模型;
  3. 提取字段也使用高端模型;
  4. 检索后把所有结果原样传给生成节点;
  5. 没有对中间变量做裁剪;
  6. 每个分支都执行模型节点;
  7. Agent 工具调用循环次数过多。

优化建议:

1. 分类节点使用小模型

意图分类、路由判断、字段抽取通常不需要高级模型。

2. 减少不必要的 LLM 节点

如果规则能解决,就不要调用模型。例如判断文本长度、判断是否包含关键词、判断用户类型,可以由代码节点完成。

3. 限制 Agent 最大迭代次数

Agent 很容易因为工具调用循环导致成本失控。建议设置最大迭代次数,例如 3~5 次。

4. 对工具返回内容做摘要

工具返回大量 JSON、HTML、日志时,不要全部塞给模型。可以先在代码节点中提取关键字段。

5. 输出长度设置上限

很多任务不需要长篇回答。可以在 Prompt 中明确:

请在 300 字以内回答。

或者在模型参数中设置最大输出 Token。


二十二、进一步优化:自部署开源模型

如果调用量很大,可以考虑部分场景使用自部署模型。

例如:

  • 分类;
  • 摘要;
  • 文本改写;
  • 内部知识库问答;
  • JSON 抽取;
  • 敏感词识别;
  • RAG 初步回答。

可以使用:

  • Qwen;
  • DeepSeek;
  • Llama;
  • GLM;
  • Baichuan;
  • Yi;
  • InternLM;
  • MiniCPM。

自部署不一定总是更便宜,因为还要考虑 GPU 成本、运维成本、并发能力和模型效果。但当请求量足够大、任务相对稳定时,自部署小模型通常很划算。

建议采用混合策略:

高频简单任务:自部署小模型
通用问答任务:低价商业模型
复杂高价值任务:高性能商业模型

二十三、Dify 配置层面的降本清单

下面是一份可以直接执行的检查清单。

模型配置

  • [ ] 默认模型不要直接使用最贵模型;
  • [ ] 分类、提取、总结任务使用小模型;
  • [ ] 高级模型只用于复杂任务;
  • [ ] 设置合理的最大输出 Token;
  • [ ] 降低不必要的 temperature;
  • [ ] 为不同业务创建不同 Dify 应用。

Prompt 配置

  • [ ] 删除重复说明;
  • [ ] 删除无效背景;
  • [ ] 把长规则整理为简短规则;
  • [ ] 明确回答长度;
  • [ ] 要求无法确认时不要编造;
  • [ ] 避免每次传入超长变量。

知识库配置

  • [ ] 清理重复和过期文档;
  • [ ] 合理切片;
  • [ ] Top K 不宜过大;
  • [ ] 设置 Score Threshold;
  • [ ] 必要时使用 Rerank;
  • [ ] FAQ 类内容单独建库;
  • [ ] 高价值文档优先结构化。

工作流配置

  • [ ] 能用代码节点解决的不用 LLM;
  • [ ] 路由节点使用低成本模型;
  • [ ] 限制 Agent 迭代次数;
  • [ ] 控制工具返回内容长度;
  • [ ] 对长上下文做摘要;
  • [ ] 删除无用变量传递。

运维配置

  • [ ] 接入日志统计;
  • [ ] 统计每个用户 Token;
  • [ ] 统计每个应用 Token;
  • [ ] 设置每日预算;
  • [ ] 设置缓存;
  • [ ] 设置限流;
  • [ ] 设置异常告警;
  • [ ] 定期分析高成本请求。

二十四、成本优化效果如何评估?

不能只看总费用下降,还要看效果是否可接受。建议关注以下指标:

指标 说明
单次请求平均成本 每次请求平均消耗金额
单用户日均成本 每个用户每天消耗
缓存命中率 越高说明重复问题越多
平均输入 Token Prompt 和上下文是否过长
平均输出 Token 回答是否过度冗长
高级模型占比 是否过度使用贵模型
知识库召回命中率 检索质量是否足够
用户满意度 成本下降不能明显伤害体验
响应时延 降本后速度是否提升

一个比较理想的优化结果是:

总成本下降 30%~70%
响应速度提升 20%~50%
用户满意度基本不下降
高成本模型调用占比明显降低
缓存命中率逐步提升

如果你的应用有大量重复问答,加入缓存后成本下降会非常明显。


二十五、总结

Dify 本身是一个非常优秀的大模型应用开发平台,但随着业务增长,成本优化一定要提前设计。降低成本的关键不是简单“换便宜模型”,而是构建一套完整的成本控制体系。

核心方法可以总结为:

  1. 模型分层:简单任务用便宜模型,复杂任务用高级模型;
  2. 请求路由:通过 AI Gateway 对请求进行智能分流;
  3. 上下文控制:限制历史轮数、压缩 Prompt、裁剪变量;
  4. 知识库优化:控制切片大小、Top K、Rerank 和文档质量;
  5. 缓存机制:重复问题直接返回,减少模型调用;
  6. 预算管理:限制用户、部门、应用的日消耗;
  7. 工作流降本:减少不必要的 LLM 节点;
  8. 混合部署:高频简单任务可用自部署小模型;
  9. 持续观测:用数据分析高成本请求并持续优化。

如果你正在用 Dify 搭建企业内部 AI 应用,建议不要等账单失控后再优化。最好的做法是在上线初期就增加网关、缓存、预算和日志统计能力,把成本治理作为平台能力的一部分。

本文提供的 FastAPI 网关源码只是一个起点。你可以在此基础上继续扩展:

  • 接入 JWT 鉴权;
  • 加入 API Key 管理;
  • 增加语义缓存;
  • 对接 Prometheus 监控;
  • 对接企业微信或飞书告警;
  • 增加多供应商模型兜底;
  • 增加更精准的 Token 计费;
  • 加入用户画像和动态路由;
  • 支持流式响应;
  • 支持部门级预算管理。

最终目标是让 Dify 不只是“能用”,而是能够在企业场景中 稳定、可控、低成本地长期运行

目录结构
全文