大模型级部署方案
大模型级部署方案
当前大模型应用已从实验阶段转向规模化落地，而部署方案的选择直接决定了服务性能、成本和安全边界。本文将全面解析LLM（大语言模型）、Embedding（文本向量化）、Rerank（语义重排序）三类核心模型的标准化RESTful接口部署方案。
大模型部署技术全景图
大模型部署包含三大技术层级：
- 计算层：GPU选型（消费级卡/专业卡/裸金属集群）
- 框架层：推理引擎（vLLM/Ollama/TEI）
- 服务层：API接口（OpenAI兼容/自定义RESTful）
其中Embedding与Rerank作为RAG（检索增强生成）流水线的核心组件，直接影响语义理解精度。最新研究表明，优化后的Rerank可使问答准确率提升35%+。
以下为三类模型部署方案对比概览：
| 模型类型 | 代表模型 | 典型部署方案 | 接口标准化 |
|------------|------------|----------------|------------|--------------|
| LLM | DeepSeek-R1、GPT-OSS | Ollama本地部署/vLLM/llama.cpp | OpenAI兼容 |
| Embedding | multilingual-e5 | Docker+TEL/Infinity | 自定义RESTful |
| Rerank | bge-reranker | TEI工具链/Infinity | 专用POST接口 |
LLM部署：从本地轻量到云端高性能
方案1：Ollama本地部署（轻量级首选）
适用环境：个人开发/中小企业内部工具
- 部署流程：
# 安装Ollama ollama pull deepseek-coder:6.7b-q4_0 # 量化版仅需4.1GB ollama run deepseek-coder - 优点：
- 数据完全本地化，满足金融/医疗等隐私敏感场景
- 支持断网运行，响应延迟<2秒
- 内存需求低（32B模型仅需32GB内存）
- 缺点：
- 模型能力受限（32B版性能仅为671B满血版的20%）
- 长文本生成可能降至1-2 token/s
实践建议：搭配FastAPI封装OpenAI格式接口：
@app.post("/v1/chat")
def chat_endpoint(request: ChatRequest):
return { "response": ollama.generate(request.prompt) }
- Ollama安装包：https://ollama.com/download
方案2：vLLM
适用环境：高并发服务
vLLM 主要推荐和支持在 Linux 操作系统上安装和运行，Windows只能通过WSL2），在 WSL 的 Ubuntu 终端中，更新软件包列表并安装 。官方文档
pip install vllm
python -m vllm.entrypoints.api_server
–model /path/to/MiniCPM3-4B
–served-model-name MiniCPM3-4B
–host 0.0.0.0
-port 8847
或者：
vllm serve
–model /path/to/MiniCPM3-4B
–served-model-name MiniCPM3-4B
–host 0.0.0.0
-port 8847
vllm serve是 vllm.entrypoints.openai.api_server 的命令行包装（CLI 别名），简化了启动流程。核心功能完全相同：均提供 /v1/chat/completions、/v1/completions 等 OpenAI 标准接口路径，支持流式输出、动态批处理（Token-Level Batching）和分布式部署。
- 优点：
- 动态批处理提升吞吐量3-5倍
- 原生支持OpenAI协议，无缝替换ChatGPT
- 可通过docker部署，只需增加容器副本达到横向扩展
llama.cpp
Embedding模型：向量化服务的容器化实践
Docker+TEI标准化方案
部署步骤：
docker run -d -p 7965:7965 --gpus all \
engchina/embeddings-api:multilingual-e5-large-instruct
- Embedding镜像：
docker pull engchina/embeddings-api
Docker + infinity方案
docker run -itd --gpus ‘“device=8,9”’
-e HF_ENDPOINT=https://hf-mirror.com
-p 7997:7997
swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/michaelf34/infinity:latest
v2
–model-id BAAI/bge-large-zh-v1.5
–model-id BAAI/bge-reranker-large
–port 7997
接口测试：
curl -X POST "http://localhost:7965/v1/embeddings" \
-H "Content-Type: application/json" \
-d '{"input": "大模型部署方案对比"}'
响应特征：
{
"data": [{
"embedding": [0.017, -0.032, ...], // 1024维向量
"index": 0
}],
"model": "text-embedding-3-large"
}
优势：
- 支持多语言文本向量化
- 提供float/int8两种精度格式
- 单容器QPS可达120+（T4 GPU）
Rerank模型：RAG精度提升关键
技术价值
在检索增强生成中，Rerank通过语义重排序将Top1准确率提升40%
HuggingFace TEI部署方案
text-embeddings-router --model-id BAAI/bge-reranker-large --port 8080
调用示例：
import requests
payload = {
"query": "LLM部署方案",
"texts": ["Ollama本地教程...", "vLLM集群方案..."]
}
response = requests.post("http://localhost:8080/rerank", json=payload)
print(response.json()[0]['score']) # 输出相关性分数
Transformers原生部署
Transformers 是Hugging Face开源的Python 库，用于自然语言处理（NLP）和现在多模态模型的 加载、训练和推理库。Transformers 库提供了一套完整的工具链，从模型加载（LLM/Embedding/Reranker）→ 优化（量化/硬件加速）→ 服务封装（API/生产工具）→ 部署（容器化/云平台），全面支持大模型的工业级落地。其核心功能是从 Hugging Face Hub 或本地便捷地 from_pretrained() 加载模型。提供统一的 API（如 pipeline, AutoModel, AutoTokenizer）与成千上万个预训练模型交互，是模型研究、实验、原型开发和简单应用的基石和标准。
- 大型语言模型（LLM）
加载方式：通过 AutoModelForCausalLM（生成任务，如 GPT）或 AutoModelForSeq2SeqLM（如 T5）加载预训练模型，支持本地路径或 Hugging Face Hub 模型 ID。
示例代码:
from transformers import AutoModelForCausalLM, AutoTokenizer
model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-3-8B-Instruct")
tokenizer = AutoTokenizer.from_pretrained(model_name)
# 某些模型也可以使用AutoModle加载，例如：OpenBMB/MiniCPM3-4B
from transformers import AutoModel, AutoTokenizer
model = AutoModel.from_pretrained(model_path, trust_remote_code=True)
tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
import torch
import json
import aiohttp
from PIL import Image
from pathlib import Path
from typing import List, Dict
from transformers import AutoModel, AutoTokenizer
from .base import BaseLLM
from typing import List, Dict
class HuggingFaceServer(BaseLLM):
def __init__(self, model_path, device=None) -> None:
self._model = AutoModel.from_pretrained(model_path, trust_remote_code=True)
self._tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=True)
if device:
self._model.to(device)
elif torch.cuda.is_available():
self._model.cuda()
async def arun(self, messages: List[Dict[str, str]], stream: bool, **kargs):
"""Run the model with messages and return the response.
Args:
messages (List[Dict[str, str]]): List of input messages
Example: [
{"role": "user", "content": ["hello", "/path/to/image.jpg"]},
{"role": "assistant", "content": ["This is a landscape photo"]}
]
stream (bool): Whether to use streaming output
Example: True
Returns:
If stream=False, returns complete response:
Example: {"data": "This is a beautiful landscape photo"}
If stream=True, returns generator with incremental responses:
Example: yield {"data": "This is"}
yield {"data": "a beautiful"}
yield {"data": "landscape photo"}
"""
new_messages = []
for msg in messages:
role, content = msg["role"], msg['content']
if isinstance(content, list):
content = [Image.open(item) if len(item) < 256 and Path(item).exists() else item for item in content]
new_messages.append(dict(role=role, content=content))
if not stream:
answer = self._model.chat(
image=None,
msgs=new_messages,
tokenizer=self._tokenizer,
stream=stream,
sampling=True,
)
return {"data": answer}
else:
return self._stream_response(new_messages)
def _stream_response(self, messages):
answer = self._model.chat(
image=None,
msgs=messages,
tokenizer=self._tokenizer,
stream=True,
sampling=True,
)
for item in answer:
if not item: continue
yield dict(data=item)
def run(self, messages: List[Dict[str, str]], stream: bool, **kargs):
"""Run the model with a list of messages and return the response.
Args:
messages (List[Dict[str, str]]): List of input messages
Each message contains role and content fields:
- role: Message role, can be "user" or "assistant"
- content: Message content, can be a list of text or image paths
Example: [
{"role": "user", "content": ["hello", "/path/to/image.jpg"]},
{"role": "assistant", "content": ["This is a landscape photo"]}
]
stream (bool): Whether to use streaming output
True: Returns a generator that yields responses incrementally
False: Returns complete response at once
Returns:
When stream=False:
Returns complete response dict: {"data": "complete response text"}
When stream=True:
Returns generator yielding responses: {"data": "partial response text"}
"""
def _chat(self, messages, stream=False):
return self._model.chat(
image=None,
msgs=messages,
tokenizer=self._tokenizer,
stream=stream,
sampling=True,
)
def stream_response():
answer = _chat(self, messages, stream)
for item in answer:
if not item: continue
data = dict(data=item)
yield f"data:{data}\n\n"
print(item, end="", flush=True)
if stream:
return stream_response()
else:
answer = _chat(self, messages)
print(answer)
return {"data": answer}
-
文本嵌入模型（Embedding）
加载方式：使用 transformers 或 sentence-transformers 库，适用于语义搜索、聚类等任务。示例代码：
model = AutoModel.from_pretrained(model_name, trust_remote_code=True) tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True) -
重排序模型（Reranker）
加载方式：通过 AutoModelForSequenceClassification 实现，对检索结果进行相关性重排。示例代码：
model = AutoModelForSequenceClassification.from_pretrained("BAAI/bge-reranker-large") tokenizer = AutoTokenizer.from_pretrained(model_name_or_path, cache_dir=cache_dir, trust_remote_code=True) -
使用 FastAPI 或 Flask 封装模型为 RESTful 接口。
示例代码：
import json
import argparse
import traceback
import urllib.parse
from PIL import Image
from pathlib import Path
from loguru import logger
from flask import Flask, request, Response, stream_with_context
from app.ultrarag.modules.llm.huggingface_like import HuggingFaceServer
class MicroServer:
"""A micro server implementation for handling chat requests with HuggingFace models."""
def __init__(self, model_path: str, device: str):
"""
Initialize the server with a HuggingFace model.
Args:
model_path (str): Path to the model files
device (str): Device to run the model on (e.g., 'cpu', 'cuda')
"""
if not Path(model_path).exists():
raise ValueError(f"model path: {model_path} not exist!")
self.llm = HuggingFaceServer(model_path=model_path, device=device)
self.app = Flask(__name__)
self.app.add_url_rule('/chat', view_func=self.process, methods=['get', 'post'])
def process(self):
"""
Process incoming chat requests, handling both text and image inputs.
Returns:
Response: Flask response object containing either streaming or non-streaming chat response
"""
try:
# Parse request data
data = json.loads(request.form.get("data", {}))
messages = data.get('messages')
stream = data.get('stream')
# Process image files if present
name2img = dict()
for item in request.files.values():
if "image" not in item.content_type:
continue
key = urllib.parse.unquote(item.filename)
name2img[key] = Image.open(item)
logger.info(f"images files: {[n for n in name2img.keys()]}")
# Replace image references with actual image objects
if name2img:
img_messages = []
for msg in messages:
role, content = msg["role"], msg['content']
new_content = [name2img.get(item, item) for item in content]
img_messages.append(dict(role=role, content=new_content))
else:
img_messages = messages
logger.info(f"messages: {img_messages}")
resp = self.llm.run(messages=img_messages, stream=stream)
if stream:
return Response(stream_with_context(resp), mimetype='text/event-stream')
else:
return resp
except Exception as e:
err_msg = traceback.format_exc()
return Response(err_msg, status=500)
def run_server(self, host: str, port: int):
"""
Start the Flask server.
Args:
host (str): Server host address
port (int): Server port number
"""
try:
self.app.run(host=host, port=port)
except:
print(traceback.format_exc())
if __name__ == "__main__":
args = argparse.ArgumentParser()
args.add_argument("-host", required=True, type=str, help="server host")
args.add_argument("-port", required=True, type=int, help="server port")
args.add_argument("-model_path", required=True, type=str, help="model file path")
args.add_argument("-device", required=False, default="", type=str, help="model device")
args = args.parse_args()
server = MicroServer(model_path=args.model_path, device=args.device)
server.run_server(host=args.host, port=args.port)
- 对比
Transformers 库提供了一套完整的工具链，从模型加载（LLM/Embedding/Reranker）→ 优化（量化/硬件加速）→ 服务封装（API/生产工具）→ 部署（容器化/云平台），全面支持大模型的工业级落地。开发者可根据场景选择。transformers部署模型与vllm，ollama，infinity_emb等工具部署模型的关系
-
Hugging Face Transformers：
部署灵活性极高：模型研究、实验、原型开发和简单应用的 基石和标准。部署灵活性极高，可以完全控制推理的每一个环节。
性能非最优：其默认的推理方式（尤其是自回归生成）没有对高并发、内存管理、计算优化等生产环境需求做深度优化。例如，在处理大量并发请求时，其 KV Cache 内存管理效率较低。
需要自行搭建服务：需要自己用 FastAPI、Flask 等 Web 框架将 transformers 的推理代码封装成 API 服务，并处理并发、批处理、监控等生产级问题。
简单来说，transformers 解决了“如何正确地调用模型”的问题，但没有完全解决“如何高效、大规模地服务模型”的问题。 -
专业化工具 (vLLM, Ollama, Infinity-emb)：为生产而生的优化
这些工具建立在 transformers 等库的基础之上，针对模型部署和生产服务 的特定痛点进行了深度优化。
-
vLLM： 大规模语言模型（LLM）服务的性能王者
与 Transformers 的关系：可以看作是 transformers 推理功能的一个 高性能替代后端。它重写了模型推理的核心部分，特别是 attention 和内存管理。
核心创新：PagedAttention 算法。它借鉴了操作系统内存分页的思想，来高效管理 KV Cache，几乎完全消除了内存碎片，从而在高并发场景下实现了极高的吞吐量（Tokens per Second）。
极致性能：尤其是在批量处理多个并发请求方面，性能远超原生 transformers。
开源：代码可查，可定制。
API 兼容：其 API Server 通常兼容 OpenAI API 格式，易于集成。
专注 LLM：主要为自回归生成式 LLM 设计。 -
Ollama： 本地模型的用户体验专家
与 Transformers 的关系：Ollama 在底层很可能使用了 transformers 或 ggml 等库来实际运行模型。它的价值不在于底层创新，而在于极致的用户体验和封装。
核心创新：无与伦比的易用性。通过简单的命令行工具（ollama run），自动处理模型的下载、加载、版本管理和运行。它提供了一个非常简单的 REST API 供应用调用。
开箱即用：极大降低了在本地（尤其是个人电脑上）运行大模型的门槛。
强大的社区模型库：轻松运行各种量化后的模型。
本地优先：设计初衷是为了在开发者的笔记本电脑或本地服务器上运行，而非大型数据中心。
轻量级：相比 vLLM，它更侧重于快速启动和易用性，而不是极致的多用户并发性能。 -
Infinity： 嵌入（Embedding）模型的性能专家
与 Transformers 的关系：它专门针对 sentence-transformers（基于 transformers） 这类嵌入模型进行了终极优化。
核心创新：将嵌入模型的推理速度推向了硬件极限。通过极度精细的 GPU 并行化、批处理优化和量化，实现了每秒处理数百万个文本段的超高吞吐量。
专精一地：只做嵌入模型推理，并做到了世界顶尖水平。
为 RAG 优化：是构建检索增强生成（RAG）系统时，用于文档编码（document indexing）和查询编码（query encoding）的绝佳选择。
API 服务：直接提供一个高性能的嵌入向量生成 API。
结语：部署方案没有最好只有最合适
- 个人开发者：首选Ollama+6.7B量化版，成本趋近于零
- 中小企业：Docker Compose编排Embedding/Rerank+云平台LLM
- 大型企业：自建昇腾超节点集群，实现千卡级协同计算
未来趋势：随着MoE架构普及和4位量化技术成熟，消费级设备运行百亿模型将成为可能。但在可预见的未来，混合部署模式（关键业务本地化+通用能力上云）仍是平衡安全与成本的最优解。
愿你我都能在各自的领域里不断成长，勇敢追求梦想，同时也保持对世界的好奇与善意!
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
28
0- 0
已为社区贡献5条内容
所有评论(0)