数据分析agent_三：docker 拉取embedding镜像部署

Toner_225

116人浏览 · 2026-07-18 00:37:14

Toner_225 · 2026-07-18 00:37:14 发布

 

加上 :/models/bge-m3（容器内挂载路径）

⚠️ 前提条件检查
在执行之前，先确认：

1. 本地模型文件是否存在

powershell
# 检查目录里有没有模型文件
dir G:\ever2026\cache\BAAI\bge-m3
应该能看到 config.json、pytorch_model.bin 或 model.safetensors 等文件。

2. 如果模型不完整，先下载完整模型

powershell
# 设置镜像
$env:HF_ENDPOINT = "https://hf-mirror.com"

# 下载完整模型到指定目录（确保包含所有必要文件）
pip install huggingface-hub
huggingface-cli download BAAI/bge-m3 --local-dir G:/ever2026/cache/BAAI/bge-m3 --local-dir-use-symlinks False
注意：bge-m3 比 bge-small-zh 大得多（约2.2GB），确保磁盘空间充足。


# 1，不需要gpu  cpu版本
     docker run -d --name bge-large-zh-cpu -p 8080:80 -v "E:/笔记2026项目/项目一_问数/day/资料/1_windows_docker安装/embedding/bge-large-zh-v1.5":/models/bge-large-zh-v1.5 ghcr.io/huggingface/text-embeddings-inference:cpu-latest --model-id /models/bge-large-zh-v1.5 --auto-truncate
# 2，Gpu版本

## 本地
    docker run -d --name bge-large-zh -p 8080:80 --gpus all -v "E:/笔记2026项目/项目一_问数/day/资料/1_windows_docker安装/embedding/bge-large-zh-v1.5":/models/bge-large-zh-v1.5 ghcr.io/huggingface/text-embeddings-inference:cuda-1.9 --model-id /models/bge-large-zh-v1.5 --auto-truncate
## 需要下载
    docker run -d --name bge-small-zh -p 8080:80 --gpus all -e HF_ENDPOINT=https://hf-mirror.com ghcr.io/huggingface/text-embeddings-inference:cuda-1.9 --model-id BAAI/bge-small-zh-v1.5 --auto-truncate
如果你要部署 bge-small-zh-v1.5（之前讨论的小模型）：

bash
# 3，下载小模型
    huggingface-cli download BAAI/bge-small-zh-v1.5 --local-dir G:/ever2026/cache/BAAI/bge-small-zh-v1.5 --local-dir-use-symlinks False

# 4，挂载运行
    docker run -d --name bge-small-zh -p 8080:80 --gpus all -v G:/ever2026/cache/BAAI/bge-small-zh-v1.5:/models/bge-small-zh ghcr.io/huggingface/text-embeddings-inference:cuda-1.9 --model-id /models/bge-small-zh --auto-truncate



# 💡 方案一：给Docker容器配置镜像加速（侵入式）
    在运行Docker命令时，通过 -e 参数设置环境变量，让容器内的下载工具自动走镜像源，这是最直接的办法：
    
    bash
    docker run -d --name bge-small-zh -p 8080:80 \
      --gpus all \
      -e HF_ENDPOINT=https://hf-mirror.com \
      ghcr.io/huggingface/text-embeddings-inference:cuda-1.9 \
      --model-id BAAI/bge-small-zh-v1.5 \
      --auto-truncate
    这个 -e HF_ENDPOINT=https://hf-mirror.com 是关键。不过，这个方法是否对所有TEI镜像都有效，有时会因镜像的构建方式而异。如果不行，可以试试下面的“曲线救国”办法。

# 🛠 方案二：先在Windows本地下载，再挂载进容器（稳定可靠）
    先在Windows上用镜像工具把模型下到本地文件夹，然后用Docker挂载进去，完全绕过容器内的下载环节。这个办法非常稳妥，兼容性最好。

## 第1步：在Windows下载模型
    打开Windows的 PowerShell，依次执行：

powershell
### 1. 设置镜像环境变量
    $env:HF_ENDPOINT = "https://hf-mirror.com"

### 2. 安装下载工具
    pip install huggingface-hub

###  3. 下载模型到本地指定目录，例如 D:\bge_model
    huggingface-cli download BAAI/bge-small-zh-v1.5 --local-dir D:\bge_model
    注意：对Windows系统，务必加上 --local-dir 参数指定路径，避免因权限问题导致下载失败。

## 第2步：把本地模型挂载进Docker容器
    下载完成后，用 -v 参数把本地目录挂载到容器内，并告诉TEI直接读取本地模型：
    
    bash
    docker run -d --name bge-small-zh -p 8080:80 \
      --gpus all \
      -v D:\bge_model:/models/bge-small-zh \
      ghcr.io/huggingface/text-embeddings-inference:cuda-1.9 \
      --model-id /models/bge-small-zh \
      --auto-truncate




启动容器
1. 启动命令
如果容器已经创建但停止了，用这个命令启动：

bash
docker start bge-large-zh-cpu
如果是首次创建并运行，直接用之前的 docker run 命令即可。

2. 查看容器状态
bash
# 查看所有容器（包括停止的）
docker ps -a

# 只看运行中的容器
docker ps

# 查看容器详细信息
docker inspect bge-large-zh-cpu
验证部署是否成功
方法一：查看启动日志（最直接）
bash
docker logs bge-large-zh-cpu
成功标志：日志末尾出现类似 Server started at http://0.0.0.0:80 或 Ready 的字样，没有报错。

常见错误排查：

如果出现 No such file or directory：检查模型路径挂载是否正确

如果出现 CUDA 相关错误：CPU 版本不应该有 CUDA 报错，检查镜像是否选对

方法二：健康检查
bash
# 检查容器是否在运行
docker ps | grep bge-large-zh-cpu

# 检查端口是否监听
netstat -ano | findstr 8080

import requests
import numpy as np
from typing import List, Optional, Dict, Any
from dataclasses import dataclass
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class EmbeddingConfig:
    """配置类"""
    url: str = "http://localhost:8080"
    model_name: str = "BAAI/bge-large-zh-v1.5"


    timeout: int = 30
    max_retries: int = 3
    batch_size: int = 32  # 批量处理大小
    dimension: int = 1024  # bge-small-zh 是1024维


class EmbeddingClientManager:
    """Embedding 客户端管理器"""

    def __init__(self, config: EmbeddingConfig = None):
        self.config = config or EmbeddingConfig()
        self.session = requests.Session()
        self.session.timeout = self.config.timeout
        # 连接池配置
        self.session.mount('http://', requests.adapters.HTTPAdapter(
            pool_connections=10,
            pool_maxsize=20
        ))

    def encode(self, texts: List[str]) -> np.ndarray:
        """
        单个或批量编码文本
        用法:
            embeddings = client.encode(["你好", "世界"])
            # 返回 shape: (2, 1024)
        """
        if isinstance(texts, str):
            texts = [texts]

        # 分批处理
        all_embeddings = []
        for i in range(0, len(texts), self.config.batch_size):
            batch = texts[i:i + self.config.batch_size]
            batch_embeddings = self._encode_batch(batch)
            all_embeddings.append(batch_embeddings)

        return np.vstack(all_embeddings) if all_embeddings else np.array([])

    def _encode_batch(self, texts: List[str]) -> np.ndarray:
        """单批次编码（带重试）"""
        payload = {
            "inputs": texts,
            "truncate": True
        }

        for attempt in range(self.config.max_retries):
            try:
                response = self.session.post(
                    f"{self.config.url}/embed",
                    json=payload
                )
                response.raise_for_status()
                data = response.json()

                # TEI返回格式: [[vec1], [vec2], ...]
                if isinstance(data, list):
                    return np.array(data)
                # 某些服务可能返回字典
                elif isinstance(data, dict) and 'embeddings' in data:
                    return np.array(data['embeddings'])
                else:
                    raise ValueError(f"未知返回格式: {type(data)}")

            except Exception as e:
                logger.warning(f"批次请求失败 (尝试 {attempt + 1}/{self.config.max_retries}): {e}")
                if attempt == self.config.max_retries - 1:
                    raise
                time.sleep(2 ** attempt)  # 指数退避

    def encode_single(self, text: str) -> np.ndarray:
        """编码单个文本，返回1D向量"""
        result = self.encode([text])
        return result[0] if len(result) > 0 else np.array([])

    def similarity(self, text1: str, text2: str) -> float:
        """计算两个文本的余弦相似度"""
        vec1 = self.encode_single(text1)
        vec2 = self.encode_single(text2)

        # 归一化
        vec1 = vec1 / np.linalg.norm(vec1)
        vec2 = vec2 / np.linalg.norm(vec2)

        return float(np.dot(vec1, vec2))

    def batch_similarity(self, query: str, candidates: List[str]) -> List[float]:
        """
        计算查询与多个候选文本的相似度
        用法:
            scores = client.batch_similarity("我喜欢猫", ["猫很可爱", "狗很忠诚", "今天天气好"])
            # 返回: [0.89, 0.45, 0.23]
        """
        query_vec = self.encode_single(query)
        candidate_vecs = self.encode(candidates)

        # 归一化
        query_vec = query_vec / np.linalg.norm(query_vec)
        candidate_vecs = candidate_vecs / np.linalg.norm(candidate_vecs, axis=1, keepdims=True)

        similarities = np.dot(candidate_vecs, query_vec)
        return similarities.tolist()

    def health_check(self) -> bool:
        """检查服务是否可用"""
        try:
            response = self.session.get(f"{self.config.url}/health", timeout=5)
            return response.status_code == 200
        except:
            return False

    def close(self):
        """关闭session"""
        self.session.close()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()


# ============ 使用示例 ============

if __name__ == "__main__":
    # 1. 配置连接
    config = EmbeddingConfig(
        # url="http://你的WindowsIP:8080",  # 远程地址
        url="http://127.0.0.1:8080",  # 远程地址
        batch_size=16,
        max_retries=3
    )

    # 2. 创建客户端
    with EmbeddingClientManager(config) as client:
        # 健康检查
        if not client.health_check():
            print("❌ 服务不可用，请检查容器是否启动")
            exit(1)

        print("✅ 服务连接成功")

        # 3. 编码单个文本
        vec = client.encode_single("你好，世界")
        print(f"单条向量维度: {vec.shape}")  # (1024,)

        # 4. 批量编码
        texts = ["我喜欢编程", "今天天气很好", "机器学习很有趣"]
        vecs = client.encode(texts)
        print(f"批量向量形状: {vecs.shape}")  # (3, 1024)

        # 5. 相似度计算
        score = client.similarity("我喜欢猫", "猫很可爱")
        print(f"相似度: {score:.4f}")

        # 6. 批量相似度排序
        query = "推荐一本好书"
        candidates = [
            "《三体》科幻小说",
            "《深度学习》技术书籍",
            "今天吃什么"
        ]
        scores = client.batch_similarity(query, candidates)

        # 按相似度降序
        ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
        print("\n相似度排序:")
        for text, score in ranked:
            print(f"  {score:.4f} - {text}")