Dify 上线后账单太高?这套降本方案和网关源码可以直接用
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:高性能模型
|
|-- 命中缓存 --> 直接返回
这样的好处是:
- 不需要频繁改 Dify 内部代码;
- 可以按业务灵活分流;
- 可以快速切换模型供应商;
- 方便统计每个用户、每个应用、每种任务的成本;
- 可以做熔断和兜底。
五、策略三:控制上下文长度
大模型的成本通常和 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 怎么部署?”
- “怎么重置密码?”
这些问题不需要每次都调用大模型。可以对用户问题进行归一化后缓存。
缓存命中时直接返回结果,成本几乎为零。
缓存可以分为:
- 精确缓存:问题完全相同;
- 语义缓存:问题意思相近;
- 知识库 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 客户不限制或单独计费;
- 测试账号限制更低;
- 非工作时间限制高成本模型调用。
二十、进一步优化:语义缓存
精确缓存只能命中完全相同的问题。语义缓存可以处理类似问题,例如:
报销流程是什么?
怎么申请报销?
费用报销要走什么流程?
它们意思接近,可以复用答案。
语义缓存的基本思路是:
- 对用户问题生成向量;
- 在缓存向量库中查找相似问题;
- 相似度超过阈值时,直接返回历史答案;
- 否则调用 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 工作流非常强大,但如果设计不当,也容易造成重复调用。
常见浪费包括:
- 每个节点都调用大模型;
- 分类节点使用高端模型;
- 提取字段也使用高端模型;
- 检索后把所有结果原样传给生成节点;
- 没有对中间变量做裁剪;
- 每个分支都执行模型节点;
- 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 本身是一个非常优秀的大模型应用开发平台,但随着业务增长,成本优化一定要提前设计。降低成本的关键不是简单“换便宜模型”,而是构建一套完整的成本控制体系。
核心方法可以总结为:
- 模型分层:简单任务用便宜模型,复杂任务用高级模型;
- 请求路由:通过 AI Gateway 对请求进行智能分流;
- 上下文控制:限制历史轮数、压缩 Prompt、裁剪变量;
- 知识库优化:控制切片大小、Top K、Rerank 和文档质量;
- 缓存机制:重复问题直接返回,减少模型调用;
- 预算管理:限制用户、部门、应用的日消耗;
- 工作流降本:减少不必要的 LLM 节点;
- 混合部署:高频简单任务可用自部署小模型;
- 持续观测:用数据分析高成本请求并持续优化。
如果你正在用 Dify 搭建企业内部 AI 应用,建议不要等账单失控后再优化。最好的做法是在上线初期就增加网关、缓存、预算和日志统计能力,把成本治理作为平台能力的一部分。
本文提供的 FastAPI 网关源码只是一个起点。你可以在此基础上继续扩展:
- 接入 JWT 鉴权;
- 加入 API Key 管理;
- 增加语义缓存;
- 对接 Prometheus 监控;
- 对接企业微信或飞书告警;
- 增加多供应商模型兜底;
- 增加更精准的 Token 计费;
- 加入用户画像和动态路由;
- 支持流式响应;
- 支持部门级预算管理。
最终目标是让 Dify 不只是“能用”,而是能够在企业场景中 稳定、可控、低成本地长期运行。