tangwang / SearchEngine

cloud_embedding_usage.md 6.14 KB

云端向量化模块使用说明

概述

本项目新增了基于阿里云 DashScope API 的文本向量化模块，使用 text-embedding-v4 模型生成文本向量。

模块说明

1. CloudTextEncoder (embeddings/cloud_text_encoder.py)

云端文本向量化编码器，使用阿里云 DashScope API。

特性：

  • 单例模式，线程安全
  • 支持单个或批量文本向量化
  • 自动处理 API 调用和错误处理
  • 生成 1024 维向量
  • 支持批处理以避免 API 速率限制

初始化：

from embeddings.cloud_text_encoder import CloudTextEncoder

# 方式1：从环境变量读取 API Key
encoder = CloudTextEncoder()

# 方式2：显式传入 API Key
encoder = CloudTextEncoder(api_key="sk-xxx")

# 方式3：使用新加坡地域
encoder = CloudTextEncoder(
    api_key="sk-xxx",
    base_url="https://dashscope-intl.aliyuncs.com/compatible-mode/v1"
)

使用示例：

# 单个文本向量化
text = "衣服的质量杠杠的"
embedding = encoder.encode(text)
print(embedding.shape)  # (1, 1024)

# 批量文本向量化
texts = ["文本1", "文本2", "文本3"]
embeddings = encoder.encode(texts)
print(embeddings.shape)  # (3, 1024)

# 大批量处理（自动分批）
large_texts = ["文本" + str(i) for i in range(1000)]
embeddings = encoder.encode_batch(large_texts, batch_size=32)
print(embeddings.shape)  # (1000, 1024)

2. 测试脚本 (scripts/test_cloud_embedding.py)

测试云端向量化功能，读取 queries.txt 前100条数据进行测试。

功能：

  • 读取查询文件
  • 逐条发送向量化请求
  • 记录每次请求的发送时间、接收时间和耗时
  • 统计成功率和平均耗时

使用步骤

1. 设置 API Key

首先需要获取阿里云 DashScope API Key：

设置环境变量：

export DASHSCOPE_API_KEY="your-api-key-here"

或者在项目根目录的 .env 文件中添加：

DASHSCOPE_API_KEY=your-api-key-here

2. 安装依赖

确保已安装 OpenAI Python SDK：

pip install openai

或者使用项目的 requirements.txt：

pip install -r requirements.txt

3. 运行测试脚本

# 在项目根目录下运行
python scripts/test_cloud_embedding.py

4. 查看测试结果

测试脚本会输出以下信息：

================================================================================
Cloud Text Embedding Test - Aliyun DashScope API
================================================================================

API Key: sk-xxxxxx...xxxx

Reading queries from: /path/to/queries.txt
Successfully read 100 queries

Initializing CloudTextEncoder...
CloudTextEncoder initialized successfully

================================================================================
Testing 100 queries (one by one)
================================================================================

[  1/100] ✓ SUCCESS
  Query: Bohemian Maxi Dress
  Send Time:    2025-12-05 10:30:45.123
  Receive Time: 2025-12-05 10:30:45.456
  Duration:     0.333s
  Embedding Shape: (1, 1024)

[  2/100] ✓ SUCCESS
  Query: Vintage Denim Jacket
  Send Time:    2025-12-05 10:30:45.789
  Receive Time: 2025-12-05 10:30:46.012
  Duration:     0.223s
  Embedding Shape: (1, 1024)

...

================================================================================
Test Summary
================================================================================
Total Queries:     100
Successful:        100
Failed:            0
Success Rate:      100.0%
Total Time:        35.123s
Total API Time:    32.456s
Average Duration:  0.325s per query
Throughput:        2.85 queries/second
================================================================================

注意事项

  1. API 限制：阿里云 DashScope API 有速率限制，测试脚本在批处理时会添加小延迟以避免触发限制。

  2. 成本：使用云端 API 会产生费用，请注意控制调用量。

  3. 网络要求：需要稳定的网络连接访问阿里云服务。

  4. 错误处理：如果 API 调用失败，编码器会返回零向量作为降级处理，并记录错误日志。

  5. 向量维度：text-embedding-v4 模型生成 1024 维向量，如果需要不同维度，请考虑使用其他模型。

地域选择

  • 北京地域（默认）：https://dashscope.aliyuncs.com/compatible-mode/v1
  • 新加坡地域：https://dashscope-intl.aliyuncs.com/compatible-mode/v1

不同地域使用不同的 API Key，请确保匹配。

集成到项目

在项目中使用云端向量化：

from embeddings.cloud_text_encoder import CloudTextEncoder

# 初始化编码器（全局单例）
encoder = CloudTextEncoder()

# 在搜索、索引等模块中使用
def process_text(text: str):
    embedding = encoder.encode(text)
    # 使用 embedding 进行后续处理
    return embedding

对比本地编码器

特性 CloudTextEncoder BgeEncoder (本地)
部署方式 云端 API 本地服务
初始成本 低（按使用付费） 高（GPU/CPU 资源）
运行成本 按调用量计费 固定资源成本
延迟 较高（网络往返） 低（本地处理）
吞吐量 受 API 限制 受硬件限制
离线使用 不支持 支持
维护成本 低 需要维护服务

故障排除

问题：API Key 未设置

ERROR: DASHSCOPE_API_KEY environment variable is not set!

解决：设置 DASHSCOPE_API_KEY 环境变量。

问题：API 调用失败

Failed to encode texts via DashScope API: ...

解决：

  1. 检查网络连接
  2. 验证 API Key 是否正确
  3. 确认 base_url 与 API Key 地域匹配
  4. 检查账户余额

问题：速率限制

Rate limit exceeded

解决：

  1. 增加批处理之间的延迟
  2. 减小 batch_size
  3. 升级 API 套餐

更多信息