掘金 人工智能 14小时前
Langchain向量处理长度限制优化
index_new5.html
../../../zaker_core/zaker_tpl_static/wap/tpl_guoji1.html

 

本文分享了在构建知识库时,如何解决Embedding模型上下文长度限制问题的实践经验。通过扩展DashScope Embeddings类,并借鉴OpenAI的实现思路,实现了对超长文本的智能分块处理,从而成功突破了模型长度限制。文章详细介绍了优化过程、架构设计以及需要注意的细节,为类似问题的解决提供了参考。

🤔 知识库构建中,Embedding模型面临上下文长度限制的挑战,超长文本导致调用失败。作者使用百炼的text-embedding-v4模型,当文本块超过8192 tokens时,会遇到错误。

💡 解决方案是扩展DashScopeEmbeddings类,借鉴OpenAI的_get_len_safe_embeddings方法。核心在于使用DashScope的tokenizer对文本进行分词,然后按照embedding_ctx_length进行分块,最后通过加权平均合并结果。

🛠️ 实现过程中,关键在于使用与text-embedding-v4模型匹配的tokenizer(qwen-turbo),以确保token计算准确。同时,embedding_ctx_length参数的设置需要根据实际情况进行调整,建议保守配置,以确保稳定性。

一个知识库构建过程中如何解决 Embedding 模型长度限制问题的实践记录

最近在构建知识库时遇到了一个棘手问题:某些文档的文本块长度超过了 Embedding 模型的上下文限制,导致调用失败。这篇文章记录了我们如何通过扩展 DashScope Embedding 类来解决这个问题的完整过程。

介绍背景

在构建 RAG 应用时,Embedding 模型是不可或缺的组件。然而,市面上的 Embedding 模型都存在一个共同限制:上下文长度限制

模型限制现状

不管是 API 还是开源自己部署的模型,都存在上下文大小限制,一般是 8192 tokens:

实际遇到的问题

在构建知识库时,有些文档确实是一个不可分割的文本块,而这个文本块的长度又超出了 Embedding 模型的长度限制,导致调用 Embedding 会报错。

本次我使用的就是百炼的 text-embedding-v4,使用的 Embedding 类是 langchain_communityDashScopeEmbeddings,遇到一个文本块长度为 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_openaiOpenAIEmbeddings,发现能正常使用:

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_lengthembedding_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 计算不准确,还可能影响最终的嵌入质量。

从技术实现角度,这次优化的核心是:

遗留的坑

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 更有价值。

Fish AI Reader

Fish AI Reader

AI辅助创作,多种专业模板,深度分析,高质量内容生成。从观点提取到深度思考,FishAI为您提供全方位的创作支持。新版本引入自定义参数,让您的创作更加个性化和精准。

FishAI

FishAI

鱼阅,AI 时代的下一个智能信息助手,助你摆脱信息焦虑

联系邮箱 441953276@qq.com

相关标签

Embedding 知识库 Langchain DashScope 文本分块
相关文章