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

加上 :/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 # 2Gpu版本 ## 本地 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_ENDPOINThttps://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_ENDPOINThttps://hf-mirror.com \ ghcr.io/huggingface/text-embeddings-inference:cuda-1.9 \ --model-id BAAI/bge-small-zh-v1.5 \ --auto-truncate 这个 -e HF_ENDPOINThttps://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 8080import requests import numpy as np from typing import List, Optional, Dict, Any from dataclasses import dataclass import time import logging logging.basicConfig(levellogging.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_connections10, pool_maxsize20 )) 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, jsonpayload ) 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, axis1, keepdimsTrue) 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, timeout5) 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( # urlhttp://你的WindowsIP:8080, # 远程地址 urlhttp://127.0.0.1:8080, # 远程地址 batch_size16, max_retries3 ) # 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), keylambda x: x[1], reverseTrue) print(\n相似度排序:) for text, score in ranked: print(f {score:.4f} - {text})✅ 服务连接成功单条向量维度: (1024,)批量向量形状: (3, 1024)相似度: 0.7915相似度排序:0.3683 - 《深度学习》技术书籍0.2914 - 《三体》科幻小说0.2098 - 今天吃什么