在构建基于大语言模型（LLM）的应用时，context\_length\_exceeded 错误是最常见且令人头疼的问题之一。当输入提示（包括对话历史、系统指令和当前查询）加上预期输出 Token 数超过模型的最大上下文窗口时，就会发生这种错误。
接下来将深入剖析 Dify 内部的 Token 计算和上下文管理逻辑。学习如何优雅地解决了这个问题。
👆👆👆欢迎关注，一起进步👆👆👆
一、核心问题：Token 限制
LLM 不直接处理单词或字符，而是处理"Token"。一个 Token 可以是一个单词、单词的一部分或标点符号。每个模型都有一个硬性限制，即单次请求能处理的 Token 数量，这被称为上下文窗口。例如，OpenAI 的 gpt-4-turbo 的上下文窗口为 128,000 个 Token，而 Anthropic 的 Claude 3.5 Sonnet 的上下文窗口为 200,000 个 Token 。
如果应用程序发送的请求过大，提供商的 API 会拒绝该请求。因此，一个健壮的应用程序必须在发送请求之前计算 Token，并智能地管理其上下文以保持在限制范围内。
二、Dify 的 Token 管理流程
Dify 充当用户和 LLM 提供商之间的智能中介。它确保发送给模型的最终负载始终有效。下图展示了这个过程：
逻辑详解
- 配置收集 ：当请求开始时，Dify 加载所选模型的配置。这包括两个关键数字：
- 模型最大上下文（
context\_size） ：模型支持的理论最大 token 数（如 128,000）。 - 用户定义的输出 Token（
max\_tokens） ：开发者在 Dify 界面中设置的值，为模型的 答案 保留一部分上下文窗口。
-
预算计算 ：Dify 使用简单公式计算提示的可用预算：
可用提示 Token 数 = 模型最大上下文 - 用户定义的输出 Token 数 -
Token 化和计数 ：这是实现的核心。Dify 使用 OpenAI 的
tiktoken库进行精确的 Token 计算。 -
上下文截断 ：如果当前上下文的 Token 计数超过
可用提示 Token 数，Dify 会智能地截断它。对于聊天应用，这通常涉及从对话历史中删除最旧的消息，直到总 Token 计数适合预算。 -
最终 API 调用 ：只有在验证和潜在截断之后，Dify 才会将负载发送给 LLM 提供商，确保请求有效并避免
context\_length\_exceeded错误。
三、与提供商模型配置的关系
Dify 的 Token 计算逻辑与提供商模型配置中的参数有着密切的关系，在 Dify 中，每个模型都有以下关键配置参数：
# 模型配置示例
model\_config = {
"context\_size": 128000, # 模型的最大上下文窗口
"max\_tokens": 4096, # 用户设置的最大输出 token 数
"model\_name": "gpt-4-turbo", # 模型名称，用于选择正确的 tokenizer
"provider": "openai" # 提供商，决定使用哪种 token 化方法
}
Dify 的 Token 计算流程
四、代码分析：核心实现
1. 预先计算剩余 Token
在 base\_app\_runner.py 中，get\_pre\_calculate\_rest\_tokens 方法负责计算可用的 Token 预算：
def get\_pre\_calculate\_rest\_tokens(self, app\_record, model\_config, prompt\_template\_entity, inputs, files, query):
# 获取模型实例
model\_instance = ModelInstance(
provider\_model\_bundle=model\_config.provider\_model\_bundle, model=model\_config.model
)
# 获取模型最大上下文 token 数
model\_context\_tokens = model\_config.model\_schema.model\_properties.get(ModelPropertyKey.CONTEXT\_SIZE)
# 获取用户定义的输出 max\_tokens
max\_tokens = 0
for parameter\_rule in model\_config.model\_schema.parameter\_rules:
if parameter\_rule.
name
== "max\_tokens"or (
parameter\_rule.use\_template and parameter\_rule.use\_template == "max\_tokens"
):
max\_tokens = (
model\_config.parameters.get(parameter\_rule.
name
)
or model\_config.parameters.get(parameter\_rule.use\_template or"")
) or0
# 获取不包含记忆和上下文的提示消息
prompt\_messages, stop = self.organize\_prompt\_messages(
app\_record=app\_record,
model\_config=model\_config,
prompt\_template\_entity=prompt\_template\_entity,
inputs=inputs,
files=files,
query=query,
)
# 计算当前提示消息的 token 数
prompt\_tokens = model\_instance.get\_llm\_num\_tokens(prompt\_messages)
# 计算剩余可用 token 数
rest\_tokens: int = model\_context\_tokens - max\_tokens - prompt\_tokens
if rest\_tokens < 0:
raise InvokeBadRequestError(
"Query or prefix prompt is too long, you can reduce the prefix prompt, "
"or shrink the max token, or switch to a llm with a larger token limit size."
)
return rest\_tokens
2. 重新计算最大 Token 数
如果提示 Token 数加上最大输出 Token 数超过模型上下文限制，Dify 会自动调整 max\_tokens 参数：
def recalc\_llm\_max\_tokens(self, model\_config, prompt\_messages):
# 获取模型实例
model\_instance = ModelInstance(
provider\_model\_bundle=model\_config.provider\_model\_bundle, model=model\_config.model
)
# 获取模型最大上下文 token 数
model\_context\_tokens = model\_config.model\_schema.model\_properties.get(ModelPropertyKey.CONTEXT\_SIZE)
# 获取用户定义的输出 max\_tokens
max\_tokens = 0
for parameter\_rule in model\_config.model\_schema.parameter\_rules:
if parameter\_rule.
name
== "max\_tokens"or (
parameter\_rule.use\_template and parameter\_rule.use\_template == "max\_tokens"
):
max\_tokens = (
model\_config.parameters.get(parameter\_rule.
name
)
or model\_config.parameters.get(parameter\_rule.use\_template or"")
) or0
# 计算当前提示消息的 token 数
prompt\_tokens = model\_instance.get\_llm\_num\_tokens(prompt\_messages)
# 如果提示 token 数加上最大输出 token 数超过模型上下文限制，则调整 max\_tokens
if prompt\_tokens + max\_tokens > model\_context\_tokens:
max\_tokens = max(model\_context\_tokens - prompt\_tokens, 16)
for parameter\_rule in model\_config.model\_schema.parameter\_rules:
if parameter\_rule.
name
== "max\_tokens"or (
parameter\_rule.use\_template and parameter\_rule.use\_template == "max\_tokens"
):
model\_config.parameters[parameter\_rule.
name
] = max\_tokens
3. 计算剩余 Token
在 prompt\_transform.py 中，\_calculate\_rest\_token 方法负责计算剩余可用的 Token 数：
def \_calculate\_rest\_token(self, prompt\_messages, model\_config):
rest\_tokens = 2000
model\_context\_tokens = model\_config.model\_schema.model\_properties.get(ModelPropertyKey.CONTEXT\_SIZE)
if model\_context\_tokens:
model\_instance = ModelInstance(
provider\_model\_bundle=model\_config.provider\_model\_bundle, model=model\_config.model
)
curr\_message\_tokens = model\_instance.get\_llm\_num\_tokens(prompt\_messages)
max\_tokens = 0
for parameter\_rule in model\_config.model\_schema.parameter\_rules:
if parameter\_rule.name == "max\_tokens"or (
parameter\_rule.use\_template and parameter\_rule.use\_template == "max\_tokens"
):
max\_tokens = (
model\_config.parameters.get(parameter\_rule.name)
or model\_config.parameters.get(parameter\_rule.use\_template or"")
) or0
rest\_tokens = model\_context\_tokens - max\_tokens - curr\_message\_tokens
rest\_tokens = max(rest\_tokens, 0)
return rest\_tokens
4. 管理对话历史
上下文管理的智能策略，分层截断策略。
在 token\_buffer\_memory.py 中，get\_history\_prompt\_messages 方法负责获取对话历史，并在必要时进行截断：
def get\_history\_prompt\_messages(self, max\_token\_limit=2000, message\_limit=None):
# 获取对话历史消息
messages = query.limit(message\_limit).all()
thread\_messages = extract\_thread\_messages(messages)
messages = list(reversed(thread\_messages))
prompt\_messages = []
# 将消息转换为提示消息
for message in messages:
# 处理消息内容和文件
# ...
# 如果提示消息的 token 数超过最大限制，则进行截断
curr\_message\_tokens = self.model\_instance.get\_llm\_num\_tokens(prompt\_messages)
if curr\_message\_tokens > max\_token\_limit:
pruned\_memory = []
while curr\_message\_tokens > max\_token\_limit and len(prompt\_messages) > 1:
pruned\_memory.append(prompt\_messages.pop(0))
curr\_message\_tokens = self.model\_instance.get\_llm\_num\_tokens(prompt\_messages)
return prompt\_messages
5. Token 计算的核心：tiktoken
Dify 使用 OpenAI 的 tiktoken 库进行精确的 Token 计算。这是一个快速的 BPE（Byte Pair Encoding）tokenizer，专为 OpenAI 模型设计。
在 large\_language\_model.py 中，get\_num\_tokens 方法负责获取提示消息的 Token 数：
def get\_num\_tokens(self, model,
credential
s, prompt\_messages, tools=None):
if dify\_config.PLUGIN\_BASED\_TOKEN\_COUNTING\_ENABLED:
plugin\_model\_manager = PluginModelClient()
return plugin\_model\_manager.get\_llm\_num\_tokens(
tenant\_id=self.tenant\_id,
user\_id="unknown",
plugin\_id=self.plugin\_id,
provider=self.provider\_name,
model\_type=self.model\_type.value,
model=model,
credentials
=
credential
s,
prompt\_messages=prompt\_messages,
tools=tools,
)
return0
五、模型配置与 Token 计算的关系
Dify 中的模型配置参数直接影响 Token 计算和限制处理：
- ModelPropertyKey.CONTEXT_SIZE ：定义模型的最大上下文窗口大小，如 GPT-4 的 128,000 或 Claude 的 200,000。
- max_tokens 参数 ：用户在 Dify 界面中设置的值，为模型的回答保留一部分上下文窗口。
- 模型名称 ：决定使用哪种 tokenizer，因为不同模型使用不同的 token 化规则。
- 提供商 ：不同提供商（如 OpenAI、Anthropic）的模型可能有不同的 token 化方法和限制。
六、总结：Dify 的 Token 管理优势
Dify 的 Token 管理机制提供了几个关键优势：
- 自动化处理 ：开发者无需手动计算 Token 或担心上下文限制。
- 智能截断 ：当上下文过长时，Dify 会自动截断历史消息，确保请求有效。
- 适应性调整 ：如果提示 Token 数加上最大输出 Token 数超过模型上下文限制，Dify 会自动调整
max\_tokens参数。 - 多模型支持 ：Dify 支持多种 LLM 提供商和模型，并为每个模型使用正确的 tokenizer。
- 错误预防 ：通过预先计算和验证 Token 数，Dify 有效防止了
context\_length\_exceeded错误。
通过这种方式，Dify 为开发者提供了一个无缝的体验，使他们能够专注于构建应用程序，而不必担心底层的 Token 管理复杂性。这种抽象层次的提升是 Dify 作为 LLM 应用开发平台的核心优势之一。
参考资料
https://github.com/langgenius/dify (v1.4.1)
七、推荐阅读
- 从零开始学 Dify - 万字详解RAG父子分段模式的原理与实现
- 从零开始学 Dify - Dify 的 RAG 系统如何有效地处理和检索大量文档？
- 智能问数：从模板到指标服务
- 替代 NL2SQL，Agent+业务语义的创新产品设计
- 从零开始学 Dify-详细介绍 Dify 模型运行时的核心架构
- 从零开始学 Dify- 工作流(Workflow)系统架构
- 从零开始学 Dify- 对话系统的关键功能
👆👆👆欢迎关注，一起进步👆👆👆
欢迎留言讨论哈
🧐点赞、分享、推荐 ，一键三连，养成习惯👍