最近接了一个内部对话系统的活儿,需求本身不复杂:要把大模型能力私有化部署到内网,数据不能出网,接口不能被裸调,而且业务方点名要“通信链路支持国密算法”。聊到后面,模型和接口方案基本定了:用通义千问1.8B-Chat的GPTQ-Int4量化版本做底座,再在模型服务的外层加一层SM4报文加密。整条链路做完后我最大的感受是,轻量模型配合加密层,并没有想象中那么折腾,但里面确实有几个坑值得记下来。
这篇文章会把从量化模型加载、FastAPI服务封装,到SM4加密通信层的完整过程写出来。适合正在做私有化部署、信创环境适配、或者手里只有一张中端显卡的读者参考。你会发现1.8B模型在对话辅助、摘要、指令整理这类真实业务里完全能扛住,而SM4加密层的代码量也就百行左右,性价比非常高。
1. 方案拆解:这套东西到底在解决什么问题
1.1 为什么是1.8B-Chat而不是更大的模型
很多团队一上来就盯着7B、14B甚至更大的开源模型,觉得效果一定更好。我在选型时也算过这笔账:7B模型的FP16权重就要14GB左右,GPTQ-Int4量化后大概4GB,推理时加上KV Cache和激活值,实际显存占用轻松超过10GB。如果手头只有一张12GB显存的消费级显卡,跑7B模型基本是临界状态,并发一高就可能OOM。而14B模型就算量化了,也不是这个预算能碰的。
1.8B-Chat的优势在于:18亿参数在量化为4bit后,权重文件只有1GB出头,推理时整个模型加上临时缓冲区,6GB显存就能跑,12GB显卡已经非常从容。这类轻量Chat模型在指令跟随、文本摘要、格式整理、日常问答上的表现,对大多数内部系统来说已经“够用就好”。影响范围最直接的体现是硬件门槛大幅降低——不需要申请高配GPU,一台普通开发机就能完成部署和测试,这让小团队也能快速验证私有化大模型方案。
1.2 为什么选GPTQ-Int4而不是FP16或AWQ
不做量化当然可以,FP16直接加载1.8B模型也就3.6GB权重,单卡也能跑。但量化的收益不只是省显存,更重要的是减少权重数据在显存带宽上的读取量。模型推理时每个token都要把所有权重过一遍,Int4比FP16少读一半以上的字节,这对生成速度的影响是实打实的。
GPTQ是一种基于二阶误差近似的后训练量化方法,它把权重按组划分,用Hessian信息来做量化误差补偿,所以在4bit位宽下精度损失控制得相当好。市面上还有AWQ、以及那种GGUF配合CPU/llama.cpp跑的路线。我的结论是:如果主路径是Python + GPU + HTTP服务,GPTQ和transformers/AutoGPTQ的生态匹配度最高,加载代码最短,出问题也好搜。GGUF在纯CPU场景很香,但要走web服务还得套一层推理框架,对于这个项目属于绕路。
1.3 模型服务通信为什么要单独做SM4加密
私有化部署经常让人产生一种错觉:模型已经放在内网了,通信就安全了。实际上内网里照样有抓包、有旁路监控、有被攻陷的跳板机。如果模型服务接口是裸HTTP,任何能触达这个端口的内网进程都能直接打过来,用户发给模型的敏感数据等于明文飘在网络上。
标准做法是上HTTPS,但两个现实问题:一是部分等保或信创项目里,验收要求明确写的是“使用国家密码管理部门批准的算法”,标准TLS套件不满足;二是内网自建CA、分发证书、维护信任链,对一个小型模型服务来说成本不低。SM4是国密体系里的对称分组密码算法,密钥128bit、分组128bit,性能开销非常低,特别适合对请求体和响应体做应用层加密。这里要说清楚,应用层SM4加密不等于替代TLS,生产环境我更推荐“TLS + SM4报文加密”两层都上,先保证传输通道可信,再保证具体报文内容只有两端能读懂。
1.4 整条链路的数据流
模型服务和加解密的关系可以理解为:客户端把要发给模型的明文JSON做SM4加密,得到Base64字符串;模型服务收到后先解密成JSON,再做参数校验、调用模型推理,最后把结果JSON加密后返回给客户端。这个设计把加解密层和模型推理层解耦开,后面对接不同的客户端语言、不同的业务系统,都只需要复用一套密钥和报文格式。
2. 部署环境与模型文件准备
2.1 软硬件基线参考
我的测试环境是Linux服务器、12GB显存显卡、32GB内存。如果按最低标准,6GB显存加16GB内存也能跑,但生成速度和并发余量会紧一点。系统层面建议Ubuntu 22.04或Debian 12,Python用3.10或3.11。需要装的Python依赖大致包括:
- torch(CUDA版本要和驱动匹配)
- transformers(4.35以上,推荐4.40+)
- auto_gptq或optimum(加载GPTQ权重用)
- fastapi、uvicorn(模型服务)
- gmssl(SM4国密算法实现)
- requests(客户端调用)
这里有个经验之谈:先在虚拟环境里把torch装好并跑一次CUDA可用性检测,再装其他包。很多后续报错都出在torch和CUDA版本错配上,排查起来比模型本身的问题还麻烦。
2.2 量化模型的目录结构长什么样
从模型托管平台下载GPTQ-Int4版本后,目录里通常是这样:
model/ ├── config.json ├── generation_config.json ├── model.safetensors ├── quant_config.json ├── tokenizer.json ├── tokenizer_config.json └── chat_template.jinja 或者内置在tokenizer_config.json里不要看到model.safetensors就直接开跑。GPTQ模型的config.json里会带quantization_config字段,里面有bits、group_size、desc_act这些关键参数。部署前先确认bits=4、group_size=128,这决定了权重的存储和反量化行为。desc_act如果是True,说明模型按激活值重要性做了排列,有些推理框架不支持,测试时会遇到兼容性报错,这个后面说。
2.3 加载模型并完成第一次对话
加载量化模型最省事的方式是直接用transformers读取本地目录:
from transformers import AutoTokenizer, AutoModelForCausalLM model_dir = "./models/qwen-1.8b-chat-gptq-int4" tokenizer = AutoTokenizer.from_pretrained(model_dir, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained( model_dir, device_map="auto", trust_remote_code=True )如果遇到transformers版本和GPTQ权重不兼容,就用auto_gptq的接口加载:
from auto_gptq import AutoGPTQForCausalLM from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained(model_dir, trust_remote_code=True) model = AutoGPTQForCausalLM.from_quantized( model_dir, device="cuda:0", use_triton=False, trust_remote_code=True )两种写法我在项目里都试过,优先级是先用transformers原生加载,因为代码只要维护一套。auto_gptq作为备选,适合在模型日志里出现quantization_config解析异常时切换。
推理时注意Chat模型要套对话模板。Qwen系模型的通用做法是用tokenizer的apply_chat_template:
import torch def chat(prompt, max_new_tokens=256, temperature=0.7, top_p=0.9): messages = [{"role": "user", "content": prompt}] input_text = tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) inputs = tokenizer(input_text, return_tensors="pt").to(model.device) with torch.no_grad(): outputs = model.generate( **inputs, max_new_tokens=max_new_tokens, temperature=temperature, top_p=top_p, do_sample=True ) answer = tokenizer.decode( outputs[0][inputs.input_ids.shape[-1]:], skip_special_tokens=True ) return answer我第一次跑的时候没套模板,直接把用户文本喂进去,结果是模型答得非常敷衍。Chat模型在预训练时就是按特殊token分隔的角色对话格式学习的,不套模板等于跳过了它的默认指令形态,效果天差地别。实测用老版本模型时,apply_chat_template如果不可用,就手工按模型文档拼下面这种结构:
<|im_start|>system 你是智能助手<|im_end|> <|im_start|>user 你好<|im_end|> <|im_start|>assistant3. 模型服务化:把推理变成HTTP接口
3.1 FastAPI接口设计与明文协议
加密层加进来之前,我习惯先把明文接口跑通。这样做不是走弯路,而是方便分层定位问题:模型输出有问题就不用怀疑加密层,加密层有问题也不用怀疑模型。
明文接口设计如下:
{ "prompt": "帮我整理一下这段工作日志的重点", "max_new_tokens": 256, "temperature": 0.7, "top_p": 0.9 }FastAPI的骨架代码:
from fastapi import FastAPI from pydantic import BaseModel import uvicorn app = FastAPI() class ChatRequest(BaseModel): prompt: str max_new_tokens: int = 256 temperature: float = 0.7 top_p: float = 0.9 @app.post("/api/chat") def chat_api(req: ChatRequest): result = chat(req.prompt, req.max_new_tokens, req.temperature, req.top_p) return {"result": result}启动命令用:
uvicorn main:app --host 127.0.0.1 --port 8000注意host不要直接绑0.0.0.0。模型服务只服务给内网后端,没必要暴露到所有接口。如果部署机上确实有业务方需要跨机器调用,再用防火墙或者云安全组限制来源IP,比裸暴露稳妥得多。
3.2 模型加载生命周期与并发锁
模型是重量级对象,不能在每次请求里加载,也不能在FastAPI的各个路由函数里反复from_pretrained。通行做法是模块级做懒加载:
import threading model = None tokenizer = None inference_lock = threading.Lock() def get_model(): global model, tokenizer if model is None: model_dir = "./models/qwen-1.8b-chat-gptq-int4" tokenizer = AutoTokenizer.from_pretrained(model_dir, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained( model_dir, device_map="auto", trust_remote_code=True ) return model, tokenizer推理时的并发安全必须用锁保护。大模型在Python里不是天然线程安全的,两个请求同时走model.generate,轻则CUDA报错,重则显存溢出把服务搞挂。我把chat函数包一层:
def chat_safe(prompt, max_new_tokens=256, temperature=0.7, top_p=0.9): with inference_lock: return chat(prompt, max_new_tokens, temperature, top_p)这个锁带来的副作用是请求会排队,但对于1.8B模型这样的小模型来说完全可接受。后面我会专门分析吞吐和延迟的取舍。
3.3 为什么不能开多个uvicorn worker
FastAPI和uvicorn默认可以开多个worker进程,很多没有接触过大模型服务的人会习惯性配--workers 4。这个坑我踩过:每个worker进程都会在内存里加载一份完整模型,4个worker意味着4份模型同时占用显存,12GB显卡瞬间见底。
模型服务不适用普通Web服务的横向扩容思路。要么保持单进程单模型,配合前面说的锁来做排队;要么引入vLLM这类专门做高并发推理的服务框架,让模型在进程内共享显存并管理调度。对于1.8B级别的模型,我建议先用单进程,业务量上来了再考虑迁移。
4. SM4国密加密通信层实现
4.1 SM4参数选型:CBC模式、PKCS7填充、随机IV
SM4本身是128bit分组、128bit密钥的分组算法。选型上最需要确定三个参数:分组模式、填充方式、IV策略。
分组模式我建议CBC而不是ECB。ECB模式相同明文会得到相同密文,这个特性在报文加密里很危险,攻击者可以通过观察密文重复推断业务行为。CBC模式让每个明文分组与前一密文分组异或,即使两段明文相同,输出也完全不同。
填充用PKCS7。SM4分组是16字节,业务JSON经过utf-8编码后几乎不可能正好是16的倍数,所以在加密前要填充到边界,解密后再去掉填充。
IV策略上,最简单的是服务端和客户端各保存一份固定IV,但固定IV长期使用会有风险。我的做法是每次加密时随机生成16字节IV,拼在密文前面,一起Base64后传给服务端。IV本身不需要保密,服务端从报文里取出来直接用于解密即可。
4.2 SM4-CBC加密工具类
依赖库用了gmssl,它在Python生态里对国密算法的支持比较成熟。为了不依赖gmssl版本对crypt_cbc的填充处理差异,我直接只用它的ECB块加密原语,自己实现CBC逻辑:
import base64 import os from gmssl.sm4 import CryptSM4, SM4_ENCRYPT, SM4_DECRYPT class SM4CBC: def __init__(self, key: bytes): assert len(key) == 16 self.key = key def _ecb_encrypt(self, block: bytes) -> bytes: sm = CryptSM4() sm.set_key(self.key, SM4_ENCRYPT) return sm.crypt_ecb(block) def _ecb_decrypt(self, block: bytes) -> bytes: sm = CryptSM4() sm.set_key(self.key, SM4_DECRYPT) return sm.crypt_ecb(block) def _pad(self, data: bytes) -> bytes: pad_len = 16 - (len(data) % 16) return data + bytes([pad_len] * pad_len) def _unpad(self, data: bytes) -> bytes: pad_len = data[-1] if pad_len < 1 or pad_len > 16: raise ValueError("invalid padding") if data[-pad_len:] != bytes([pad_len] * pad_len): raise ValueError("invalid padding") return data[:-pad_len] def encrypt(self, plaintext: bytes) -> bytes: iv = os.urandom(16) padded = self._pad(plaintext) result = b"" prev = iv for i in range(0, len(padded), 16): block = padded[i:i + 16] xored = bytes(a ^ b for a, b in zip(block, prev)) encrypted = self._ecb_encrypt(xored) result += encrypted prev = encrypted return iv + result def decrypt(self, raw: bytes) -> bytes: iv = raw[:16] ciphertext = raw[16:] result = b"" prev = iv for i in range(0, len(ciphertext), 16): block = ciphertext[i:i + 16] decrypted = self._ecb_decrypt(block) plain = bytes(a ^ b for a, b in zip(decrypted, prev)) result += plain prev = block return self._unpad(result) def encrypt_to_b64(self, plaintext: str) -> str: raw = self.encrypt(plaintext.encode("utf-8")) return base64.urlsafe_b64encode(raw).decode("utf-8") def decrypt_from_b64(self, b64_text: str) -> str: raw = base64.urlsafe_b64decode(b64_text.encode("utf-8")) return self.decrypt(raw).decode("utf-8")用的时候:
key = b"0123456789abcdef" # 从环境变量读取,不要硬编码 cipher = SM4CBC(key) encrypted = cipher.encrypt_to_b64(json.dumps({"prompt": "你好"}, ensure_ascii=False)) print(encrypted) plaintext = cipher.decrypt_from_b64(encrypted) print(plaintext)这个小工具类的一次性代码量不大,但足够稳定。整个加密层后续不管怎么封装,核心就是这几十行。
4.3 服务端加密路由设计
有了工具类,服务端的加密接口就清晰了。客户端POST过来的请求体是:
{ "cipher": "Base64字符串" }服务端先解密,再解析JSON,然后调用模型推理,最后把结果加密返回。我写的路由大概是:
from fastapi import FastAPI, HTTPException from pydantic import BaseModel import json import os import time app = FastAPI() sm4_key = os.environ["SM4_KEY"].encode("utf-8") # 16字节 cipher_util = SM4CBC(sm4_key) class EncryptedRequest(BaseModel): cipher: str class EncryptedResponse(BaseModel): cipher: str @app.post("/api/chat/encrypted", response_model=EncryptedResponse) def encrypted_chat(req: EncryptedRequest): try: plain_text = cipher_util.decrypt_from_b64(req.cipher) payload = json.loads(plain_text) except Exception: raise HTTPException(status_code=400, detail="invalid cipher") prompt = payload.get("prompt", "") max_new_tokens = int(payload.get("max_new_tokens", 128)) temperature = float(payload.get("temperature", 0.7)) top_p = float(payload.get("top_p", 0.9)) result = chat_safe(prompt, max_new_tokens, temperature, top_p) response_json = json.dumps({"result": result}, ensure_ascii=False) return EncryptedResponse(cipher=cipher_util.encrypt_to_b64(response_json))这里有一个很重要的细节:解密失败时返回的detail必须是统一的“invalid cipher”,千万不要把底层异常堆栈透出去。实际攻击者是可以故意构造错误密文来探测服务的,统一错误信息能减少信息泄露面。
4.4 客户端封装与防重放设计
客户端是Python时,封装很直接:
import requests import json import time def call_model_encrypted(prompt, server_url="http://127.0.0.1:8000"): req_payload = { "prompt": prompt, "max_new_tokens": 128, "temperature": 0.7, "top_p": 0.9, "ts": int(time.time() * 1000), } plain_text = json.dumps(req_payload, ensure_ascii=False) cipher = cipher_util.encrypt_to_b64(plain_text) resp = requests.post( server_url + "/api/chat/encrypted", json={"cipher": cipher}, timeout=60, ) resp.raise_for_status() data = resp.json() decrypted = cipher_util.decrypt_from_b64(data["cipher"]) return json.loads(decrypted)["result"]请求体里的ts字段是时间戳,服务端可以校验当前时间与ts的差值,超过比如30秒就拒绝,这能在一定程度上防止重放攻击。更严格的方案是再加一个request_id,服务端维护一个已处理ID集合做去重,但这种复杂度对内部服务通常没必要。
密钥不要硬编码在代码里。我这边是放在环境变量里:
export SM4_KEY="0123456789abcdef"有密钥管理系统的团队可以放到配置中心,但不管用什么方式,原则是一样的:代码仓库里不能出现生产密钥。
4.5 关于TLS和SM4的关系说明
有些读者可能会问:是不是用了SM4加密,就不需要TLS了?不是。SM4加密的是报文内容,TLS保护的是整个连接通道,包括URL路径、请求头、响应状态码、传输时间特征等元数据。如果服务真的跨公网调用,建议在网关层用支持国密套件的TLS方案,内部再加上这层应用层SM4加密,形成纵深防御。如果只在纯内网跑,并且业务验收明确要求国密加密,那么当前方案已经能覆盖核心诉求。
5. 性能实测与并发优化建议
5.1 各环节耗时拆解
我专门用timeit跑了一组数据,量化每个环节的耗时。测试参数是:请求文本约500个汉字,模型生成80个token左右,结果如下表:
| 环节 | 耗时 | 说明 |
|---|---|---|
| SM4加密请求体 | 1-3ms | 纯Python实现,数据量小 |
| 网络传输(内网) | 1ms左右 | 与机器位置相关 |
| 模型推理生成 | 5-10s | 受显卡和token数影响最大 |
| SM4解密响应体 | 1-3ms | 几乎可忽略 |
| HTTP序列化 | <1ms | pydantic + JSON |
结论很明确:SM4加解密在整个链路里占用不到1%的时间,瓶颈完全在模型推理。所以做性能优化时不要去纠结加密层,把心思花在降低生成token数和提升GPU利用率上更实际。
5.2 并发策略与排队上限
上面提到用threading.Lock做串行化,单实例的并发能力就是“1个请求推理期间,其他请求排队等待”。对于1.8B模型的推理速度,一个请求生成100token大约3-6秒,生成256token大约8-15秒。换算成吞吐,这个方案稳定支撑的QPS在0.5-2之间,适合内部系统低频调用。
如果业务确实需要更高的并发,我建议往两个方向改造:
- 把max_new_tokens限制在64-128,响应更快,减少单请求占用的GPU时间。
- 引入vLLM或者类SGLang的推理框架,对KV Cache做连续内存管理,在模型内部实现请求调度,并发吞吐会有明显提升。但这样SM4加密层的接入方式也要跟着改,通常是把vLLM的OpenAI兼容接口放在后面,加密路由放在前面代理。
5.3 密钥轮换怎么做
预共享密钥最大的痛点是轮换。我采用的是key_version方案:
- 请求体里带上key_version字段。
- 服务端持有多个版本的密钥:K1、K2。
- 发布新密钥K2后,控制一部分客户端先切到K2,服务端保留K1解密旧报文。
- 确信旧客户端全部切换后,从服务端移除K1。
密钥轮换的周期根据业务安全要求来定,我建议至少每90天轮换一次。轮换时不必重启模型服务,密钥存在一个dict里,热加载即可。这个设计虽然只有很小的代码量,但能让密钥生命周期管理正规化,不会出现“密钥泄露但没法换”的尴尬。
6. 常见问题与排障记录
6.1 模型加载与推理问题
问题1:加载模型时报CUDA显存不足。
最直接的原因是显存被其他进程占用。先执行nvidia-smi查可用显存,把模型放到device_map="auto"让transformers自动分配。如果显存只有4-6GB,建议把max_new_tokens调小,同时检查是否在加载时开启了过多的low_memory选项。
问题2:模型输出质量和预期差很多。
十有八九是Prompt格式没走对。确认用apply_chat_template,或者手工套了正确的system/user/assistant角色标记。还有一点容易被忽略:Qwen模型对中文上下文的truncate机制,如果长文本输入超长,后面的内容会被截断,可以先落实max_new_tokens和max_length的关系。
问题3:transformers版本和auto_gptq不兼容。
GPTQ权重的描述文件在不同版本里字段有变化。一个常见做法是先把transformers固定到4.40以上,auto_gptq用0.7.x版本;如果两个库版本不匹配,会出现quantization_config解析报错。遇到这种问题,用前面的auto_gptq加载接口做备选,能规避大部分版本摩擦。
6.2 SM4解密失败问题速查
| 现象 | 原因 | 排查方法 |
|---|---|---|
| 解密后JSON解析失败 | 密钥不一致或IV拼接错误 | 先在两端用同一段明文测试加解密 |
| 解密报padding错误 | 明文长度与PKCS7冲突 | 检查加密前是否使用了utf-8编码 |
| 偶发解密失败 | IV没有随密文传递 | 确认协议里IV=密文前16字节 |
| 密文里有+/=字符,URL传参出错 | Base64编码问题 | 统一用urlsafe_b64encode |
| 服务端返回invalid cipher | 密文被截断或base64多了空格 | 打印cipher长度与日志脱敏比较 |
我在实际调试中踩得最狠的一个坑是:gmssl库不同版本对crypt_cbc的填充行为不一致,同一个密文在一个环境能解密,在另一个环境就报错。所以我在工具类里坚持自己实现PKCS7和解密前的长度对齐,只用库的块加密能力,从此再没出过兼容性问题。
6.3 服务与网络问题
问题1:uvicorn配置了多worker导致显存OOM。
上面已经解释过原因。排除方法就是检查启动参数,确保没有--workers N。FastAPI的app本身不建议在启动时直接load模型到模块全局变量以外的做法,用懒加载更容易控制生命周期。
问题2:客户端请求timeout。
FastAPI的模型推理是同步阻塞的,生成256token可能要十几秒。请求方如果使用默认的requests timeout,很容易提前断开。需要把客户端的timeout设置为60s甚至更长,并且服务端要考虑到连接断开的场景:客户端已经断开,模型还在生成,浪费GPU。一个优化是针对这种情况做任务取消,但内部系统里可以先通过调小max_new_tokens来缓解。
问题3:日志里出现大量明文。
模型服务一旦打印prompt和response的日志,在加解密链路中就相当于泄露了业务数据。我的做法是在日志里只打印request_id、耗时、token数和密文的前16位哈希。如果一定要留明文做审计,就单独接审计系统,并做权限管控。
7. 做完这个方案后的几点体会
整套链路从模型选型到SM4加密,最大的体会是先跑通明文再套加密层。加密会把中间调试的复杂度放大好几倍,如果一开始就直接上SM4,一旦模型输出有问题会很难判断是推理的问题还是解密的问题。我建议大家在复现时也走同样的顺序:明文接口通了,再加加密路由,最后再把明文入口关掉或者加上内部IP白名单。
另外一个值得说的点是,1.8B这类小模型经常被低估。很多人默认大模型必须7B起步,但实际在内部知识整理、客服对话预筛、格式化输出这些相对轻量的任务上,1.8B的GPTQ-Int4量化版本部署成本和效果之间的平衡点非常理想。配合SM4加密层,它能塞进很多之前“不可能上大模型”的预算环境里。
加密层本身不难,难在密钥管理和版本兼容。我强烈建议把密钥轮换、key_version、日志脱敏这些“看不见的工程”当成方案的一部分,而不是可选项。实际生产环境出问题,往往不是SM4算法本身,而是密钥硬编码、轮换没跟上、或者日志把明文打了出去。把这几个点做好,这套模型服务通信方案就可以稳定跑很久。