一个知识库构建过程中如何解决 Embedding 模型长度限制问题的实践记录
最近在构建知识库时遇到了一个棘手问题:某些文档的文本块长度超过了 Embedding 模型的上下文限制,导致调用失败。这篇文章记录了我们如何通过扩展 DashScope Embedding 类来解决这个问题的完整过程。
介绍背景
在构建 RAG 应用时,Embedding 模型是不可或缺的组件。然而,市面上的 Embedding 模型都存在一个共同限制:上下文长度限制。
模型限制现状
不管是 API 还是开源自己部署的模型,都存在上下文大小限制,一般是 8192 tokens:
- OpenAI text-embedding-3-large: 8191 tokens百炼 text-embedding-v4: 8192 tokens其他开源模型: 通常在 512-8192 tokens 之间
实际遇到的问题
在构建知识库时,有些文档确实是一个不可分割的文本块,而这个文本块的长度又超出了 Embedding 模型的长度限制,导致调用 Embedding 会报错。
本次我使用的就是百炼的 text-embedding-v4,使用的 Embedding 类是 langchain_community
的 DashScopeEmbeddings
,遇到一个文本块长度为 30000 多(按字符长度计算的),就报错了:
# 典型的错误场景from langchain_community.embeddings.dashscope import DashScopeEmbeddingsembedding = DashScopeEmbeddings( dashscope_api_key="your-api-key", model="text-embedding-v4")# 长文本(30000+ 字符)long_text = "这是一个超长的文档内容..." * 1000# 调用失败:Range of input length should be [1, 8192]result = embedding.embed_documents([long_text])
错误信息很明确:Range of input length should be [1, 8192]
,文本长度超出了模型的处理范围。
优化过程
2.1 尝试切换为 OpenAI Embeddings
最初尝试切换到 langchain_openai
的 OpenAIEmbeddings
,发现能正常使用:
from langchain_openai import OpenAIEmbeddings# OpenAI 模型可以正常处理长文本openai_embedding = OpenAIEmbeddings( model="text-embedding-3-large", openai_api_key="your-api-key")# 这样可以工作result = openai_embedding.embed_documents([long_text])print(f"成功处理长度为 {len(long_text)} 的文本")
OpenAI 的实现确实能够正常处理长文本,但是这个需要调用 OpenAI 的 tiktoken 来计算 token 长度。我们公司的开发环境比较特殊,访问外部网络都得开白名单,就很麻烦。
2.2 分析 OpenAI Embeddings 的源码
通过查看 OpenAIEmbeddings
的源码,发现他默认开启了 check_embedding_ctx_length
,然后走的是 _get_len_safe_embeddings
方法。
这个方法的核心逻辑是:
def _get_len_safe_embeddings(self, texts: List[str]) -> List[List[float]]: """ OpenAI 的长度安全嵌入实现: 1. 使用 tiktoken 计算 token 长度 2. 按 embedding_ctx_length 分块 3. 分别获取每个块的嵌入 4. 使用加权平均合并结果 """ tokens = [] indices = [] # 使用 tiktoken 进行分词 encoding = tiktoken.encoding_for_model(self.model) for i, text in enumerate(texts): token = encoding.encode(text) # 按上下文长度分块 for j in range(0, len(token), self.embedding_ctx_length): tokens.append(token[j : j + self.embedding_ctx_length]) indices.append(i) # 批量获取嵌入并合并 # ...
关键在于 OpenAI 会自动将超长文本分块处理,然后通过加权平均的方式合并多个块的嵌入向量。
2.3 改造 DashScopeEmbeddingsExt
尝试按照 OpenAIEmbeddings
的做法改造一个 DashScopeEmbeddingsExt
。核心思路是复制 OpenAI 的长度安全处理逻辑,但适配 DashScope 的 API:
class DashScopeEmbeddingsExt(DashScopeEmbeddings): """扩展的DashScope嵌入模型,支持check_embedding_ctx_length和智能分块""" # 新增字段 embedding_ctx_length: int = Field(default=8191, description="嵌入最大token长度") check_embedding_ctx_length: bool = Field(default=True, description="是否检查嵌入上下文长度") tokenizer_model: str = Field(default='qwen-turbo', description="分词器模型") def embed_documents(self, texts: List[str]) -> List[List[float]]: """嵌入文档列表""" if not self.check_embedding_ctx_length: # 不检查长度,使用原始逻辑 return super().embed_documents(texts) # 使用长度安全的嵌入函数 return self._get_len_safe_embeddings(texts)
这里有个细节,计算 token 长度要使用 DashScope 自己的 tokenizer:
def _tokenize(self, texts: List[str], batch_size: int): """使用 DashScope 的 tokenizer 进行分词""" tokens = [] indices = [] try: from dashscope import get_tokenizer tokenizer = get_tokenizer(self.tokenizer_model) for i, text in enumerate(texts): # 使用 DashScope tokenizer 对文本进行标记化 tokenized = tokenizer.encode(text) # 将 tokens 拆分为遵循 embedding_ctx_length 的块 for j in range(0, len(tokenized), self.embedding_ctx_length): token_chunk = tokenized[j : j + self.embedding_ctx_length] # 将 token ID 转换回字符串 chunk_text = tokenizer.decode(token_chunk) tokens.append(chunk_text) indices.append(i) except Exception as e: # 如果 tokenization 失败,回退到字符级分块 logger.warning(f"Tokenization failed, using character-based chunking: {e}") for i, text in enumerate(texts): for j in range(0, len(text), self.embedding_ctx_length): chunk_text = text[j : j + self.embedding_ctx_length] tokens.append(chunk_text) indices.append(i) return range(0, len(tokens), batch_size), tokens, indices
关键是要使用 get_tokenizer(self.tokenizer_model)
来获取与 text-embedding-v4 模型匹配的分词器,这样计算出的 token 长度才准确。
2.4 测试验证
使用以下测试代码进行验证:
if __name__ == "__main__": from dashscope import get_tokenizer embedding = DashScopeEmbeddingsExt( dashscope_api_key="dashscope_api_key", model="text-embedding-v4", check_embedding_ctx_length=True, embedding_ctx_length=8100, # 很小的值,强制分块 ) texts = ["This is a test text that will be split into multiple chunks because it's very long. " * 500] print(f"原始文本长度: {len(texts[0])} 字符") # 计算token数量 tokenizer = get_tokenizer(embedding.tokenizer_model) tokenized = tokenizer.encode(texts[0]) print(f"token数量: {len(tokenized)}") result = embedding.embed_documents(texts) print(f"嵌入结果维度: {len(result)}x{len(result[0]) if result else 0}") print("✓ 成功处理长文本并生成嵌入")
通过调整文本长度(字符串 * 的长度)、check_embedding_ctx_length
和 embedding_ctx_length
,测试发现没问题。
但是虽然 text-embedding-v4 的上下文长度为 8192,但是 embedding_ctx_length
得调到 8100 左右才能不报错,猜测应该还是 tokenizer 的模型对不上(不确定)。
架构设计
整个优化方案的架构如下:
graph TD A["长文本输入<br/>30000+ 字符"] --> B["DashScopeEmbeddingsExt"] B --> C["检查是否开启长度检查"] C -->|关闭| D["直接调用父类方法"] C -->|开启| E["_get_len_safe_embeddings"] E --> F["使用 DashScope tokenizer 分词"] F --> G["按 embedding_ctx_length 分块"] G --> H["分块1<br/>8100 tokens"] G --> I["分块2<br/>8100 tokens"] G --> J["分块3<br/>剩余 tokens"] H --> K["批量调用 API"] I --> K J --> K K --> L["获取各分块嵌入"] L --> M["加权平均合并"] M --> N["L2 归一化"] N --> O["最终嵌入向量"] style A fill:#ffcccc style O fill:#ccffcc style F fill:#ffffcc style M fill:#ccccff
核心流程包括:
- 分词处理:使用 DashScope 的
qwen-turbo
tokenizer 进行准确的 token 计算智能分块:按照 embedding_ctx_length
将长文本拆分为多个块批量嵌入:调用原始 API 获取每个块的嵌入向量加权合并:根据各块的 token 长度进行加权平均向量归一化:确保最终嵌入向量的数学特性结语
这次 Embedding 长度限制的优化让我想起了做架构设计时的一个原则:遇到限制时,不要急于绕过去,而是要深入理解限制背后的原理。
OpenAI 的 _get_len_safe_embeddings
实现给了我们很好的启发。它不是简单粗暴地拒绝长文本,而是通过智能分块和加权合并的方式,在保持语义完整性的前提下突破了长度限制。这种设计思路值得借鉴。
在适配 DashScope 的过程中,最大的收获是对 tokenizer 重要性的认识。不同模型的 tokenizer 差异很大,使用错误的 tokenizer 不仅会导致 token 计算不准确,还可能影响最终的嵌入质量。
从技术实现角度,这次优化的核心是:
- 准确的分词:使用模型匹配的 tokenizer智能分块:基于 token 而非字符进行分块合理合并:通过加权平均保持语义一致性容错处理:提供降级方案确保系统稳定性
遗留的坑
embedding_ctx_length 参数调优问题
这里要特别提示一下:虽然 text-embedding-v4 的官方上下文长度为 8192,但是在实际使用中,embedding_ctx_length
得调到 8100 左右才能不报错。
可能的原因分析:
- 特殊 Token 开销:模型可能会添加特殊的开始/结束 token,占用部分上下文空间Tokenizer 差异:我们使用的
qwen-turbo
tokenizer 与模型内部的 tokenizer 可能存在细微差异API 安全边界:百炼 API 可能设置了安全边界,实际可用长度略小于理论值建议的配置:
# 保守配置,确保稳定性embedding_ctx_length = 8100 # 而不是理论上的 8192# 如果要追求最大利用率,建议逐步测试# embedding_ctx_length = 8150 # 需要在具体环境中验证
这个问题暂时没有找到根本原因,但通过调低 embedding_ctx_length
可以稳定解决。如果有朋友知道具体原因,欢迎交流讨论。
总的来说,这次优化不仅解决了实际问题,也让我们对 Embedding 模型的内部机制有了更深入的理解。在 AI 应用开发中,这种深入技术细节的实践往往比单纯调用 API 更有价值。