GTE+SeqGPT部署教程：Kubernetes集群中GTE+SeqGPT服务化部署方案

GTE+SeqGPT部署教程：Kubernetes集群中GTE+SeqGPT服务化部署方案

1. 引言：从单机脚本到云原生服务

如果你已经尝试过在本地运行GTE和SeqGPT，体验过语义搜索和轻量生成的魅力，那么接下来可能会遇到一个新问题：如何让这个AI能力服务更多人？当你的团队需要调用它，或者你的应用需要稳定、可扩展的AI服务时，单机运行的Python脚本就显得力不从心了。

这正是我们今天要解决的问题。本文将带你一步步，将一个本地AI项目（GTE-Chinese-Large + SeqGPT-560m）打包、容器化，并最终部署到Kubernetes集群中，使其成为一个高可用、可伸缩的云原生服务。你将学到的不只是几条命令，而是一套完整的、可复用的服务化部署方案。

学习目标：

• 理解将AI模型服务化的核心思想与价值。
• 掌握使用Docker将本地项目封装为标准化镜像。
• 学会编写Kubernetes部署文件，将服务发布到集群。
• 了解如何通过Service和Ingress对外提供稳定的API访问。

前置知识：基本的Linux命令操作、对Docker和Kubernetes有概念性了解即可。我们将用最直白的方式讲解每一步。

2. 项目回顾与容器化准备

在开始部署前，我们先快速回顾一下这个AI项目的核心构成，这有助于我们设计合理的服务架构。

2.1 核心组件分析

我们的项目包含两个主要模型和一个演示框架：

1. GTE-Chinese-Large：负责将中文文本转换为语义向量。这是“理解”和“搜索”的核心。
2. SeqGPT-560m：一个轻量级的文本生成模型，负责根据指令生成或改写文本。
3. 演示脚本(main.py,vivid_search.py,vivid_gen.py)：展示了模型的基础调用、语义搜索和文案生成能力。

在单机模式下，我们通过运行Python脚本来调用它们。在服务化场景下，我们需要将它们转变为常驻进程，并通过网络API来提供服务。

2.2 设计服务化架构

一个简单的服务化思路是：

• 模型服务：将GTE和SeqGPT模型分别封装为独立的HTTP服务。例如，一个服务提供“文本转向量”接口，另一个服务提供“文本生成”接口。
• 业务逻辑服务：vivid_search.py和vivid_gen.py中的演示逻辑，可以封装为一个应用服务，它内部调用上述两个模型服务。
• 网关/路由：通过一个统一的入口（如Ingress）来管理外部访问。

为了简化首次部署，我们将采用一种更直接的方案：将整个项目（包括模型和演示逻辑）打包到一个容器中，并通过一个Web框架（如FastAPI）提供统一的API接口。这种方案部署简单，适合中小规模应用。

3. 第一步：创建Docker镜像

容器化是云原生部署的基石。我们将编写一个Dockerfile，来定义如何构建一个包含项目所有依赖和模型的可运行环境。

3.1 编写Dockerfile

在你的项目根目录下，创建一个名为Dockerfile的文件，内容如下：

# 使用一个包含PyTorch的官方Python镜像作为基础，减少自己安装CUDA的麻烦 FROM pytorch/pytorch:2.1.0-cuda11.8-cudnn8-runtime # 设置工作目录 WORKDIR /app # 复制项目依赖文件 COPY requirements.txt . # 安装Python依赖 # 使用清华镜像源加速，并安装requirements.txt中指定的包 RUN pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple && \ pip install --no-cache-dir -r requirements.txt # 安装项目可能缺失的特定依赖（根据开发者笔记） RUN pip install simplejson sortedcontainers # 复制整个项目代码到容器中 COPY . . # 提前下载模型（可选，也可以在启动时下载） # 这里我们选择在启动时下载，以保持镜像轻量，但会延长首次启动时间。 # 你也可以将模型文件直接打包进镜像，但镜像会非常大。 # 暴露服务端口（假设我们的FastAPI服务运行在8000端口） EXPOSE 8000 # 设置容器启动命令 # 这里我们启动一个FastAPI服务，假设入口文件是`api_server.py` CMD ["python", "api_server.py"]

3.2 准备requirements.txt

同样在项目根目录，创建requirements.txt文件，列出所有依赖：

fastapi==0.104.1 uvicorn[standard]==0.24.0 transformers==4.36.0 modelscope==1.20.0 datasets==2.16.0 sentence-transformers==2.2.2 pydantic==2.5.0

3.3 构建Docker镜像

在包含Dockerfile和requirements.txt的目录下，打开终端，执行构建命令：

# -t 参数给镜像打标签，格式为 名称:版本 # . 表示使用当前目录下的Dockerfile docker build -t gte-seqgpt-service:1.0 .

这个过程可能会持续几分钟到十几分钟，取决于网络速度和是否需要下载基础镜像。完成后，你可以用docker images命令查看构建好的镜像。

4. 第二步：编写API服务代码

现在我们需要创建api_server.py，这是容器启动后运行的主程序，它使用FastAPI框架提供HTTP接口。

# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import torch from transformers import AutoModel, AutoTokenizer from modelscope import snapshot_download import numpy as np from sentence_transformers.util import cos_sim import asyncio import logging # 配置日志 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="GTE+SeqGPT AI Service", version="1.0") # --- 模型加载（全局变量，在服务启动时加载）--- gte_model = None gte_tokenizer = None seqgpt_model = None seqgpt_tokenizer = None class SearchRequest(BaseModel): query: str candidates: List[str] class GenerateRequest(BaseModel): instruction: str input_text: str @app.on_event("startup") async def startup_event(): """服务启动时，异步加载模型""" logger.info("开始加载AI模型...") await asyncio.gather(load_gte_model(), load_seqgpt_model()) logger.info("所有模型加载完毕，服务准备就绪。") async def load_gte_model(): """加载GTE语义向量模型""" global gte_model, gte_tokenizer try: model_dir = snapshot_download('iic/nlp_gte_sentence-embedding_chinese-large') # 使用transformers原生方式加载，避免modelscope pipeline的兼容性问题 from transformers import AutoModel gte_model = AutoModel.from_pretrained(model_dir, trust_remote_code=True) gte_tokenizer = AutoTokenizer.from_pretrained(model_dir, trust_remote_code=True) gte_model.eval() logger.info("GTE模型加载成功。") except Exception as e: logger.error(f"加载GTE模型失败: {e}") raise async def load_seqgpt_model(): """加载SeqGPT文本生成模型""" global seqgpt_model, seqgpt_tokenizer try: model_dir = snapshot_download('iic/nlp_seqgpt-560m') seqgpt_model = AutoModel.from_pptrained(model_dir, trust_remote_code=True) seqgpt_tokenizer = AutoTokenizer.from_pretrained(model_dir, trust_remote_code=True) seqgpt_model.eval() logger.info("SeqGPT模型加载成功。") except Exception as e: logger.error(f"加载SeqGPT模型失败: {e}") # 注意：SeqGPT非必需，服务可降级运行 pass # --- API端点定义 --- @app.get("/") async def root(): return {"message": "GTE+SeqGPT AI 服务运行中", "status": "healthy"} @app.get("/health") async def health_check(): """健康检查端点，Kubernetes会调用此接口""" if gte_model is None: raise HTTPException(status_code=503, detail="GTE模型未就绪") return {"status": "healthy", "gte_loaded": gte_model is not None, "seqgpt_loaded": seqgpt_model is not None} @app.post("/api/embed") async def get_embeddings(texts: List[str]): """ 将文本列表转换为语义向量。 请求体: {"texts": ["句子1", "句子2"]} """ if gte_model is None: raise HTTPException(status_code=503, detail="GTE模型未加载") try: inputs = gte_tokenizer(texts, padding=True, truncation=True, return_tensors="pt", max_length=512) with torch.no_grad(): outputs = gte_model(**inputs) # 使用mean pooling获取句子向量 embeddings = mean_pooling(outputs, inputs['attention_mask']) # 转换为列表格式返回 return {"embeddings": embeddings.tolist()} except Exception as e: logger.error(f"向量化失败: {e}") raise HTTPException(status_code=500, detail=f"内部错误: {e}") @app.post("/api/semantic_search") async def semantic_search(request: SearchRequest): """ 语义搜索：从候选列表中找出与查询最相似的句子。 请求体: {"query": "你的问题", "candidates": ["答案1", "答案2", ...]} """ if gte_model is None: raise HTTPException(status_code=503, detail="GTE模型未加载") try: # 将查询和所有候选句子一起向量化 all_texts = [request.query] + request.candidates inputs = gte_tokenizer(all_texts, padding=True, truncation=True, return_tensors="pt", max_length=512) with torch.no_grad(): outputs = gte_model(**inputs) embeddings = mean_pooling(outputs, inputs['attention_mask']) # 计算查询向量与每个候选向量的余弦相似度 query_embedding = embeddings[0:1] candidate_embeddings = embeddings[1:] similarities = cos_sim(query_embedding, candidate_embeddings)[0] # 排序并返回结果 results = [] for idx, score in enumerate(similarities): results.append({ "candidate": request.candidates[idx], "score": score.item() }) results.sort(key=lambda x: x["score"], reverse=True) return {"query": request.query, "results": results} except Exception as e: logger.error(f"语义搜索失败: {e}") raise HTTPException(status_code=500, detail=f"内部错误: {e}") @app.post("/api/generate") async def generate_text(request: GenerateRequest): """ 轻量文本生成：根据指令和输入文本生成内容。 请求体: {"instruction": "请写一个标题", "input_text": "关于春天的散文"} """ if seqgpt_model is None: raise HTTPException(status_code=503, detail="SeqGPT模型未加载") try: # 构建符合SeqGPT指令微调格式的Prompt prompt = f"指令：{request.instruction}\n输入：{request.input_text}\n输出：" inputs = seqgpt_tokenizer(prompt, return_tensors="pt", max_length=512, truncation=True) with torch.no_grad(): outputs = seqgpt_model.generate( **inputs, max_new_tokens=100, # 控制生成长度 do_sample=True, temperature=0.7, top_p=0.9 ) generated_text = seqgpt_tokenizer.decode(outputs[0], skip_special_tokens=True) # 从生成的完整文本中提取“输出：”之后的部分 output_part = generated_text.split("输出：")[-1].strip() return {"instruction": request.instruction, "input": request.input_text, "generated_text": output_part} except Exception as e: logger.error(f"文本生成失败: {e}") raise HTTPException(status_code=500, detail=f"内部错误: {e}") def mean_pooling(model_output, attention_mask): """Mean Pooling - 取所有token向量的平均值作为句子向量""" token_embeddings = model_output[0] # 第一个元素包含所有token的向量 input_mask_expanded = attention_mask.unsqueeze(-1).expand(token_embeddings.size()).float() sum_embeddings = torch.sum(token_embeddings * input_mask_expanded, 1) sum_mask = torch.clamp(input_mask_expanded.sum(1), min=1e-9) return sum_embeddings / sum_mask

这个API服务提供了四个主要端点：

1. /和/health: 用于服务状态检查。
2. /api/embed: 文本转向量。
3. /api/semantic_search: 语义搜索。
4. /api/generate: 指令文本生成。

5. 第三步：编写Kubernetes部署文件

现在，我们将定义如何在Kubernetes集群中运行这个容器。通常需要三个YAML文件：Deployment、Service和Ingress。

5.1 Deployment (deployment.yaml)

Deployment定义了要运行什么容器、需要多少副本、资源限制等。

# deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: gte-seqgpt-deployment labels: app: gte-seqgpt spec: replicas: 2 # 运行2个副本，提供基本的高可用 selector: matchLabels: app: gte-seqgpt template: metadata: labels: app: gte-seqgpt spec: containers: - name: ai-service image: gte-seqgpt-service:1.0 # 使用我们本地构建的镜像 # 如果镜像在远程仓库，应写为 your-registry.com/username/gte-seqgpt-service:1.0 ports: - containerPort: 8000 resources: requests: memory: "8Gi" # 模型加载需要较大内存，根据实际情况调整 cpu: "2" limits: memory: "12Gi" cpu: "4" livenessProbe: # 存活探针，检查容器是否健康 httpGet: path: /health port: 8000 initialDelaySeconds: 120 # 首次检查等待时间，给模型加载留足时间 periodSeconds: 30 readinessProbe: # 就绪探针，检查服务是否准备好接收流量 httpGet: path: /health port: 8000 initialDelaySeconds: 120 periodSeconds: 20 # 环境变量示例（如果需要） # env: # - name: MODEL_CACHE_DIR # value: "/app/models" volumeMounts: - name: model-cache mountPath: /root/.cache/modelscope # 将模型缓存挂载出来，避免每次重启重复下载 volumes: - name: model-cache persistentVolumeClaim: claimName: model-cache-pvc # 需要提前创建PVC

5.2 Service (service.yaml)

Service为Deployment中的Pod提供一个稳定的网络访问入口和负载均衡。

# service.yaml apiVersion: v1 kind: Service metadata: name: gte-seqgpt-service spec: selector: app: gte-seqgpt ports: - port: 80 # Service对外暴露的端口 targetPort: 8000 # 容器内应用监听的端口 protocol: TCP type: ClusterIP # 默认类型，仅在集群内部可访问 # 如果需要从集群外直接访问NodePort，可以改为 type: NodePort

5.3 Ingress (ingress.yaml) (可选)

Ingress用于将集群内部的服务暴露到外部网络（互联网），并提供域名路由、SSL终止等功能。这需要集群已安装Ingress Controller（如Nginx Ingress）。

# ingress.yaml apiVersion: networking.k8s.io/v1 kind: Ingress metadata: name: gte-seqgpt-ingress annotations: nginx.ingress.kubernetes.io/rewrite-target: / spec: rules: - host: ai-service.your-domain.com # 替换为你的域名 http: paths: - path: / pathType: Prefix backend: service: name: gte-seqgpt-service port: number: 80

5.4 PersistentVolumeClaim (pvc.yaml) (可选，但推荐)

为了持久化存储模型文件，避免每次Pod重启都重新下载，我们需要一个持久化存储卷。

# pvc.yaml apiVersion: v1 kind: PersistentVolumeClaim metadata: name: model-cache-pvc spec: accessModes: - ReadWriteMany # 多个Pod需要同时读写 resources: requests: storage: 20Gi # 根据模型大小调整，GTE+SeqGPT约几个GB # storageClassName: 根据你的集群配置指定，例如 "nfs-client" 或 "standard"

6. 第四步：部署到Kubernetes集群

假设你已经有一个可用的Kubernetes集群（可以是云厂商的，也可以是本地的Minikube/k3s），并且配置好了kubectl命令行工具。

6.1 应用配置

将上述YAML文件保存到同一个目录，然后依次应用：

# 1. 创建持久化存储（如果使用） kubectl apply -f pvc.yaml # 2. 部署应用 kubectl apply -f deployment.yaml # 3. 创建服务 kubectl apply -f service.yaml # 4. 创建Ingress（如果配置了域名和Ingress Controller） kubectl apply -f ingress.yaml

6.2 检查部署状态

使用以下命令检查部署是否成功：

# 查看Deployment状态 kubectl get deployment gte-seqgpt-deployment # 查看Pod状态，应该看到2个Running的Pod kubectl get pods -l app=gte-seqgpt # 查看Pod日志，检查模型加载情况（替换<pod-name>为实际的Pod名称） kubectl logs <pod-name> -f # 查看Service kubectl get service gte-seqgpt-service # 查看Ingress（如果创建了） kubectl get ingress gte-seqgpt-ingress

6.3 测试服务

服务在集群内启动后，你可以通过端口转发在本地测试：

# 将集群内的服务端口8000转发到本地的8080端口 kubectl port-forward service/gte-seqgpt-service 8080:80

然后在浏览器或使用curl访问http://localhost:8080/或测试API端点：

curl -X POST http://localhost:8080/api/semantic_search \ -H "Content-Type: application/json" \ -d '{ "query": "今天天气怎么样", "candidates": ["明天会下雨", "Python是一种编程语言", "气温是25度，晴朗", "我喜欢吃苹果"] }'

7. 总结与进阶建议

7.1 部署回顾

至此，我们已经完成了一个AI项目从本地脚本到Kubernetes云原生服务的完整部署。回顾一下关键步骤：

1. 容器化：通过Dockerfile将项目及其依赖打包成标准镜像。
2. 服务化：使用FastAPI将模型能力封装为HTTP API。
3. 编排部署：利用Kubernetes的Deployment、Service、Ingress等资源对象，定义服务的运行方式、网络访问和资源需求。
4. 持久化：通过PersistentVolumeClaim保存模型文件，提升启动速度。

7.2 生产环境进阶建议

当前方案是一个入门级的服务化部署。对于生产环境，你还可以考虑以下优化：

• 镜像仓库：将构建的Docker镜像推送到Docker Hub、阿里云容器镜像服务等远程仓库，方便在不同环境中拉取。
• 配置管理：使用ConfigMap或Secret来管理API密钥、模型路径等配置信息，而不是硬编码在代码中。
• 模型服务分离：将GTE和SeqGPT拆分为独立的微服务（例如使用ModelServer），实现更细粒度的伸缩和更新。
• 监控与日志：集成Prometheus监控服务指标（请求量、延迟），使用ELK或Loki收集和分析日志。
• 自动伸缩：配置Horizontal Pod Autoscaler (HPA)，根据CPU或内存使用率自动调整Pod副本数。
• GPU支持：如果模型推理需要GPU加速，需要在Deployment中声明GPU资源请求，并确保集群节点有GPU。
• CI/CD流水线：使用GitLab CI、GitHub Actions等工具，实现代码提交后自动构建镜像、运行测试、更新部署。

7.3 核心价值

通过本次部署，你将获得：

• 高可用性：多副本部署确保单个实例故障时服务不中断。
• 易于扩展：只需修改replicas数量或配置HPA，即可轻松应对流量增长。
• 标准化运维：容器化和Kubernetes带来了统一的部署、升级、回滚方式。
• 资源隔离与优化：可以精确控制每个服务实例消耗的CPU和内存。

将AI模型服务化，是将其能力产品化、商业化的关键一步。希望这份教程能为你打开一扇门，让你能更从容地将自己的AI创意，转化为真正可服务大众的稳定应用。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。