EmbeddingService 参考

EmbeddingService 负责使用 sentence-transformers 模型生成文本向量，并缓存模型以提升性能。

概览

向量嵌入是语义检索的基础能力。该服务提供：

• 模型延迟加载与缓存
• 单文本与批量嵌入
• 离线模式支持
• 模型健康检查

配置

参数	类型	默认值	说明	环境变量
model	str	paraphrase-multilingual-MiniLM-L12-v2	模型名称	EMBEDDING_MODEL
dim	int	384	期望向量维度	EMBEDDING_DIM
model_dir	str	models/embedding	本地模型目录	EMBEDDING_MODEL_DIR
offline_mode	bool	True	仅使用本地缓存	EMBEDDING_OFFLINE_MODE

超时配置

参数	类型	默认值	说明	环境变量
embedding_model_load_seconds	int	300	模型加载超时	TIMEOUT_EMBEDDING_MODEL_LOAD_SECONDS

API 参考

类：EmbeddingService

class EmbeddingService:
    """Service for generating text embeddings."""

    def __init__(
        self,
        *,
        model_name: Optional[str] = None,
        embedding_dim: Optional[int] = None,
    ) -> None:
        """Initialize the embedding service.

        Args:
            model_name: Model name. Defaults to config.embedding.model
            embedding_dim: Expected embedding dimension. Defaults to config.embedding.dim
        """

方法：embed_text

生成单条文本的向量。

def embed_text(self, text: str) -> list[float]:
    """Generate an embedding for a single text.

    Args:
        text: Text to embed

    Returns:
        Embedding vector as list of floats

    Raises:
        ValueError: If the text is empty
        Exception: If embedding generation fails
    """

性能参考：模型加载完成后，单条约 0.1-0.3 秒。

方法：embed_batch

批量生成文本向量。

def embed_batch(self, texts: list[str]) -> list[list[float]]:
    """Generate embeddings for a batch of texts.

    Args:
        texts: List of texts to embed

    Returns:
        List of embedding vectors

    Raises:
        ValueError: If any text is empty
        Exception: If embedding generation fails
    """

性能参考：10 条文本约 0.1-0.5 秒，批量更高效。

方法：check_readiness

检查模型是否可用。

def check_readiness(self) -> EmbeddingReadinessResult:
    """Check whether the embedding model is ready.

    Loads the model if necessary and runs a test embedding.

    Returns:
        EmbeddingReadinessResult with status and timing info
    """

返回：EmbeddingReadinessResult 包含：

• is_ready：模型是否可用
• was_already_loaded：是否已加载
• model_name：模型名
• embedding_dim：期望维度
• load_time_seconds：加载耗时
• probe_embedding_dim：探测维度
• error_message：失败原因

方法：get_model_info

def get_model_info(self) -> dict:
    """Get information about the embedding model.

    Returns:
        Dict with model_name, embedding_dim, and loaded status
    """

工厂函数：get_embedding_service

def get_embedding_service() -> EmbeddingService:
    """Get or create the singleton embedding service instance."""

使用示例

基础嵌入

from ai_service.services.embedding import get_embedding_service

embedding_service = get_embedding_service()
text = "What is the refund policy?"
embedding = embedding_service.embed_text(text)

print(f"Embedding dimension: {len(embedding)}")
print(f"First 5 values: {embedding[:5]}")

批量嵌入

texts = [
    "What is the refund policy?",
    "How do I track my order?",
    "What are the shipping options?"
]

embeddings = embedding_service.embed_batch(texts)
for i, emb in enumerate(embeddings):
    print(f"Text {i}: {len(emb)} dimensions")

健康检查

result = embedding_service.check_readiness()

if result.is_ready:
    print(f"Model ready: {result.model_name}")
    print(f"Load time: {result.load_time_seconds}s")
    print(f"Dimension: {result.probe_embedding_dim}")
else:
    print(f"Model not ready: {result.error_message}")

指定模型

from ai_service.services.embedding import EmbeddingService

embedding_service = EmbeddingService(
    model_name="sentence-transformers/all-MiniLM-L6-v2",
    embedding_dim=384
)

embedding = embedding_service.embed_text("Hello world")

模型加载

加载策略

1. 延迟加载：首次调用 embed_text 或 embed_batch 时加载。
2. 本地优先：优先使用 models/embedding/{model_name}。
3. 离线模式：启用后仅使用本地缓存。
4. 超时保护：默认 300 秒超时。
5. 内存复用：模型加载后长期驻留内存。

离线模式

当 offline_mode=True 时会设置离线环境变量并强制本地加载：

os.environ["HF_HUB_OFFLINE"] = "1"
os.environ["TRANSFORMERS_OFFLINE"] = "1"

model = SentenceTransformer(model_name, local_files_only=True)

模型可移植性

离线环境建议预先下载模型：

python -c "
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')
model.save('models/embedding/paraphrase-multilingual-MiniLM-L12-v2')
"

详见 Embedding 模型可移植性。

最佳实践

模型选择

模型	维度	速度	质量	适用场景
all-MiniLM-L6-v2	384	快	良好	通用英文
paraphrase-multilingual-MiniLM-L12-v2	384	中	良好	多语言
all-mpnet-base-v2	768	慢	最好	高质量英文

批量优先

# 不推荐
embeddings = [embedding_service.embed_text(t) for t in texts]

# 推荐
embeddings = embedding_service.embed_batch(texts)

错误处理

try:
    embedding = embedding_service.embed_text(text)
except ValueError as e:
    print(f"Invalid input: {e}")
except TimeoutError as e:
    print(f"Timeout: {e}")
except Exception as e:
    print(f"Embedding failed: {e}")

内存管理

• 模型加载后会常驻内存（约 100-500 MB）
• 单例模式避免重复加载
• 无自动清理机制

性能优化

批大小建议

• 小批量 1-10：约 0.1-0.5 秒
• 中批量 10-50：约 0.5-2 秒
• 大批量 50-100：约 2-5 秒

模型预热

embedding_service.check_readiness()  # 启动时预热

硬件建议

• CPU：可用但吞吐较低
• GPU：5-10 倍提速
• 内存：单模型约 100-500 MB

常见问题

问题：模型加载超时

现象：TimeoutError: Embedding model loading timed out after 300s

原因：网络慢、模型大或 Hub 不可用

解决：

1. 提高超时：TIMEOUT_EMBEDDING_MODEL_LOAD_SECONDS=600
2. 预下载模型
3. 设置镜像：HF_ENDPOINT=https://hf-mirror.com

问题：离线模式找不到模型

现象：OSError: Can't load model from local files

原因：本地缓存不存在

解决：先下载模型或关闭离线模式。

问题：维度不匹配

现象：维度不一致警告

原因：EMBEDDING_DIM 与模型输出不一致

解决：将配置维度与模型维度保持一致。

问题：空文本错误

现象：ValueError: Cannot embed empty text

原因：传入空或仅空白文本

解决：嵌入前进行校验。

关联文档

• 架构设计 - 嵌入在流程中的位置
• Embedding 模型可移植性
• IngestionService - 摄取流程
• RAGRetrievalService - 检索流程
• 配置参考 - 嵌入配置项