tangwang / SearchEngine

CLOUD_EMBEDDING_README.md 7.63 KB

云端向量化模块 - 更新说明

📝 概述

本次更新为 SearchEngine 项目添加了基于阿里云 DashScope API 的云端文本向量化功能，使用 text-embedding-v4 模型。

🎯 主要功能

  1. CloudTextEncoder - 云端文本向量化编码器

    • 单例模式，线程安全
    • 支持单个/批量文本向量化
    • 自动错误处理和降级
    • 生成 1024 维向量

  2. 测试脚本 - 使用 queries.txt 测试向量化

    • 读取前 100 条查询
    • 记录每次请求的时间和耗时
    • 统计成功率和性能指标

  3. 示例代码 - 展示如何使用模块

    • 单个文本向量化
    • 批量处理
    • 相似度计算

📁 文件结构

SearchEngine/
├── embeddings/
│   ├── cloud_text_encoder.py          # 云端向量化编码器（新增）
│   ├── text_encoder.py                # 本地编码器（现有）
│   └── ...
├── scripts/
│   ├── test_cloud_embedding.py        # 测试脚本（新增）
│   └── ...
├── examples/
│   └── cloud_embedding_example.py     # 示例代码（新增）
├── docs/
│   ├── cloud_embedding_usage.md       # 详细文档（新增）
│   └── cloud_embedding_quickstart.md  # 快速入门（新增）
├── data_crawling/
│   └── queries.txt                    # 测试数据
├── requirements.txt                    # 已添加 openai>=1.0.0
└── CLOUD_EMBEDDING_README.md          # 本文档（新增）

🚀 快速开始

1. 安装依赖

pip install openai

或使用项目 requirements：

pip install -r requirements.txt

2. 设置 API Key

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

获取 API Key：https://help.aliyun.com/zh/model-studio/get-api-key

3. 运行测试

# 测试向量化（使用 queries.txt 前 100 条）
python scripts/test_cloud_embedding.py

# 运行示例代码
python examples/cloud_embedding_example.py

📖 使用方法

基础使用

from embeddings.cloud_text_encoder import CloudTextEncoder

# 初始化编码器
encoder = CloudTextEncoder()

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

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

批量处理

# 大批量自动分批处理
texts = [f"商品 {i}" for i in range(1000)]
embeddings = encoder.encode_batch(texts, batch_size=32)

🧪 测试脚本

测试脚本 scripts/test_cloud_embedding.py 功能：

✅ 读取 data_crawling/queries.txt 前 100 条查询
✅ 逐条发送向量化请求
✅ 记录每次请求的发送时间、接收时间、耗时
✅ 输出向量维度和内容
✅ 统计成功率、平均耗时、吞吐量

测试输出示例

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

[  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)

...

================================================================================
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
================================================================================

📊 性能特点

  • 向量维度：1024
  • 平均延迟：300-400ms/请求
  • 吞吐量：~2-3 queries/秒（单线程）
  • 错误处理：自动降级到零向量
  • 批处理：支持自动分批和速率控制

🔧 接口说明

CloudTextEncoder API

初始化

CloudTextEncoder(api_key=None, base_url=None)

参数：

  • api_key (str, optional): API Key，默认从环境变量读取
  • base_url (str, optional): API 端点，默认北京地域

encode()

encode(sentences, normalize_embeddings=True, device='cpu', batch_size=32)

参数：

  • sentences (str or List[str]): 单个文本或文本列表
  • normalize_embeddings (bool): 是否归一化（API 自动处理）
  • device (str): 设备参数（兼容性参数，云端 API 忽略）
  • batch_size (int): 批处理大小

返回：

  • np.ndarray: 形状为 (n, 1024) 的 numpy 数组

encode_batch()

encode_batch(texts, batch_size=32, device='cpu')

参数：

  • texts (List[str]): 文本列表
  • batch_size (int): 批处理大小
  • device (str): 设备参数（兼容性参数）

返回：

  • np.ndarray: 向量矩阵

📚 文档

  • 快速入门：docs/cloud_embedding_quickstart.md
  • 详细文档：docs/cloud_embedding_usage.md
  • 示例代码：examples/cloud_embedding_example.py

⚠️ 注意事项

  1. API Key 管理：妥善保管 API Key，不要提交到代码仓库
  2. 成本控制：云端 API 按使用量计费，注意控制调用次数
  3. 速率限制：注意 API 速率限制，测试脚本已添加延迟
  4. 网络依赖：需要稳定的网络连接
  5. 错误处理：API 失败时会返回零向量，请检查日志

🆚 对比本地编码器

特性 CloudTextEncoder BgeEncoder (本地)
部署方式 云端 API 本地服务
初始成本 低 高（GPU/CPU）
运行成本 按使用付费 固定
延迟 ~300-400ms
离线使用 ❌ ✅
维护成本 低 需要维护
扩展性 自动扩展 手动扩展

🔄 集成建议

选择使用场景

使用 CloudTextEncoder（云端）：

  • 初期开发和测试
  • 查询量不大的应用
  • 不需要离线支持
  • 希望降低运维成本

使用 BgeEncoder（本地）：

  • 大规模生产环境
  • 需要低延迟
  • 离线使用场景
  • 查询量非常大

混合使用

# 配置文件中选择编码器类型
ENCODER_TYPE = os.getenv("ENCODER_TYPE", "local")  # local or cloud

if ENCODER_TYPE == "cloud":
    from embeddings.cloud_text_encoder import CloudTextEncoder
    encoder = CloudTextEncoder()
else:
    from embeddings.text_encoder import BgeEncoder
    encoder = BgeEncoder()

# 使用统一接口
embeddings = encoder.encode(texts)

🐛 故障排查

问题 1：API Key 未设置

export DASHSCOPE_API_KEY="sk-your-key"

问题 2：网络连接失败

  • 检查网络连接
  • 验证 base_url 是否正确
  • 确认防火墙设置

问题 3：速率限制

  • 减小 batch_size
  • 增加请求间隔
  • 升级 API 套餐

问题 4：返回零向量

  • 检查日志中的错误信息
  • 验证 API Key 是否有效
  • 确认账户余额

🎓 示例代码

查看 examples/cloud_embedding_example.py 了解完整示例：

  • 单个/批量文本向量化
  • 相似度计算
  • 错误处理

📞 支持

✅ 验证清单

完成以下步骤确认模块正常工作：

  • [ ] 安装了 openai 包
  • [ ] 设置了 DASHSCOPE_API_KEY 环境变量
  • [ ] 运行测试脚本成功
  • [ ] 查看了示例代码
  • [ ] 阅读了文档

📅 更新日期

2025-12-05

👨‍💻 维护

如有问题或建议，请联系项目维护者。