这次我们要处理的问题非常具体:一台机器上同时塞进来两个重量级推理任务,一个负责长文本生成,一个负责图像或视频生成。从表面看这只是一次普通的多服务部署,但真正跑起来之后,显存、内存、端口、接口调度全部挤在一起,GPU 显存曲线一路走高,服务差点原地被打崩。这个场景很像一句话:“同一关里塞两个重量级选手,难度直接翻倍。”而这一关要过的不是模型能力,而是资源编排和时间调度。
很多人习惯单服务单显卡跑一个模型,感觉很稳。一旦变成“两个重量级模型同时在线”,所有隐藏问题都会暴露:显存不够用、两个框架互相抢占 CUDA 上下文、API 超时、批量任务卡死、启动脚本互相污染端口。这篇文章就把这类“多模型并发部署”当作一个完整课题,从环境准备、服务启动、并发压测、接口调用、批量任务调度、资源占用观察到问题排查,给出一套可以直接参考的落地流程。
适合阅读这篇文章的读者有三类:一是想在本地一张显卡上同时跑多个 AI 服务的人,二是在做工具链集成时需要把模型打包成 API 的开发者,三是已经遇到过“显存不足 / 服务卡死 / 任务排队混乱”但不知道从哪里下手优化的运维和老手。内容不会回避具体命令,每个环节都给了可复制的脚本和验证方式。
1. 双重量级任务并发部署:核心能力速览
既然是“一关塞两个重量级选手”,先把两个任务同时跑起来的核心能力要求列清楚。下面的表不是某一款软件的功能列表,而是这个部署方案需要满足的能力项。
| 能力项 | 说明 |
|---|---|
| 部署形式 | 两个独立模型服务,分别监听不同端口,可同时调用 |
| 典型负载 | 一个文本生成类模型 + 一个图像/视频生成类模型,或两个同类型模型做 A/B 对比 |
| 推荐硬件 | NVIDIA 独立显卡优先;显存需求取决于具体模型组合,需按实际版本核算 |
| CPU 推理 | 可以,但速度下降明显,适合没有 GPU 的环境做功能验证 |
| 接口 API | 两种服务都可以暴露 HTTP 接口,供外部程序调用 |
| 批量任务 | 通过队列脚本或请求调度,可以串行或分时执行 |
| 显存优化 | 量化、低显存模式、限制 batch、任务排队、单卡按顺序串行 |
| 启动方式 | 命令行前台启动、Docker 隔离、systemd 后台托管 |
| 适合场景 | 本地工具链集成、模型对比测试、离线内容生产、个人工作站多服务压测 |
需要说明的是,显存占用和启动参数不能拍脑袋。两个任务同时跑的时候,显存峰值并不是简单的“模型 A 显存 + 模型 B 显存”,中间还涉及 CUDA context、临时激活值、图像分辨率或文本长度带来的波动。所以下面所有优化手段都以“先单服务验证,再双服务并发测试”为主线。
2. 适用场景与使用边界
这种“两个重量级选手同机并发”的部署方式,适合以下场景。
第一,本地工具链集成。很多人会把文本模型、图像模型、语音模型组合成一个内容生产流水线,比如先让 LLM 生成脚本,再把脚本交给图像模型配图。如果每个模型都开一个容器,单机完全可以用两个端口把服务都拉起来。
第二,模型对比评测。想把两个开源模型放在同一套输入数据下做对比,最直接的方式就是同时启动两个服务,然后向两个接口发送相同的请求,比较输出质量和耗时。
第三,小规模批量生成。图像模型的单次推理时间较长,文本模型处理大量短请求时吞吐要求也不一样。两种负载放在同一台机器上,可以让文本任务利用图像任务的排队空闲时间,提高卡的整体利用率。
但它并不适合所有情况。如果两个模型加起来的需求已经超过单卡显存上线,强行同时跑只会频繁 OOM,这个时候应该改成串行任务队列,或者直接上多卡、云 GPU。如果是面向公网的高并发生产服务,单机双服务也不够稳健,需要完整的负载均衡、容灾和横向扩容方案。
合规方面要特别注意:使用开源模型时要遵守对应 License,模型权重、训练数据、生成内容都可能有版权限制;如果需要处理人脸、声音、隐私数据,必须确认素材来源和授权范围;接口服务如果监听非本地地址,要考虑访问控制和认证,避免变成内部“裸奔”服务。批量生成内容对外发布前,务必做一次人工复核。
3. 环境准备与前置条件
在开始双任务并发之前,先把环境底盘打好。下面是通用检查清单,不同模型框架的细节会不一样,但排查思路是一致的。
硬性条件方面,需要确认以下几点。
- 操作系统:Windows / Linux / macOS 都可以,但 GPU 推理优先推荐 Linux 或者 Windows + WSL2,驱动和 CUDA 环境更容易对齐。
- GPU 驱动与 CUDA:如果走 NVIDIA 显卡,先确认驱动版本支持当前 PyTorch 或 TensorFlow 需要的 CUDA 版本。
nvidia-smi里能看到驱动版本,PyTorch 的torch.version.cuda能看到运行时使用的 CUDA 版本。 - 显存:两个任务的显存需求必须逐个确认。不要只看模型参数文件大小,推理时的显存峰值通常更高,尤其图像生成模型在高分辨率下波动很大。
- 内存:显存不够时系统会借助内存兜底,但速度会暴跌。建议至少给每个模型预留数倍于模型文件体积的系统内存。
- 磁盘空间:模型权重、临时文件、输出结果都会占空间,建议模型目录和输出目录分开,方便清理。
- Python 环境:尽量不要在系统 Python 里直接装依赖,使用 venv 或 conda 创建独立环境,避免两个项目依赖冲突。
- 端口规划:两个服务分别监听不同端口,例如 7860 和 7861。启动前先检查端口是否被占用。
# 查看当前 GPU 状态 nvidia-smi # 查看端口占用情况,Linux 下使用 lsof -i:7860 lsof -i:7861 # Windows 下使用 netstat -ano | findstr 7860 netstat -ano | findstr 7861环境准备的核心原则只有一条:先让两个服务各自能在单机上单独跑通,再考虑同时启动。如果单服务本身就报错,双服务并发只会把问题放大。
4. 安装部署与启动方式
双任务并发部署最常用的方式有三种:命令行双进程、Docker 容器隔离、systemd 后台托管。这里分别给出通用模板,实际操作时把路径、端口、模型名替换成自己项目的真实值。
4.1 命令行双进程模式
如果两个服务都是 Python 项目,最简单的方式是开两个终端,或者在一个脚本里先后启动两个后台进程。单显卡场景下,建议先把显卡指定到第一个服务,再观察显存余量决定第二个服务的参数。
# 进入服务 A 目录,启动文本生成服务,监听 7860 cd /path/to/service_a CUDA_VISIBLE_DEVICES=0 python app.py --port 7860 # 进入服务 B 目录,启动图像生成服务,监听 7861 cd /path/to/service_b CUDA_VISIBLE_DEVICES=0 python app.py --port 7861如果机器有多张显卡,可以分别指定不同的设备编号,避免两个服务抢同一块卡。
# 服务 A 用 0 号卡 CUDA_VISIBLE_DEVICES=0 python app.py --port 7860 # 服务 B 用 1 号卡 CUDA_VISIBLE_DEVICES=1 python app.py --port 78614.2 Docker 容器隔离模式
当两个服务的依赖互相冲突时,Docker 是更干净的隔离方式。通过 NVIDIA Container Toolkit 可以把 GPU 映射进容器,用 docker-compose 同时管理两个服务最方便。
version: "3.9" services: model-a: image: your-service-a-image:latest ports: - "7860:7860" environment: - CUDA_VISIBLE_DEVICES=0 deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu] model-b: image: your-service-b-image:latest ports: - "7861:7861" environment: - CUDA_VISIBLE_DEVICES=0 deploy: resources: reservations: devices: - driver: nvidia count: 1 capabilities: [gpu]上面的配置需要根据实际镜像名、端口映射和显卡策略调整。单卡场景下两个容器可以映射同一块显卡,但要控制显存占用;多卡场景下让容器分别绑到不同卡。
4.3 systemd 后台托管模式
如果要做长时间运行的服务,用 systemd 把两个服务托成后台守护进程,比 nohup 更好管理,日志也更规范。
[Unit] Description=Model A Service After=network.target [Service] User=your_username WorkingDirectory=/path/to/service_a Environment="CUDA_VISIBLE_DEVICES=0" ExecStart=/usr/bin/python3 app.py --port 7860 Restart=on-failure RestartSec=10 [Install] WantedBy=multi-user.target第二个服务写一份同样的 unit,只是路径、端口、描述不同。启动后可以通过systemctl status查看状态,通过journalctl -u model-a查看日志。
无论采用哪种方式,启动后第一件事都是访问 WebUI 或健康检查接口,确认两个服务都已经真正加载完毕。很多模型加载是异步的,终端显示“启动成功”不代表模型已经就绪。
5. 功能测试与效果验证
双服务并发启动之后,不要马上压测,而是按“单服务 → 并发请求 → 资源观察 → 批量任务”的顺序逐步验证。
5.1 单服务基础验证
先分别对每个服务发出一个最小请求,确认接口响应正常。这一步可以暴露模型文件缺失、依赖版本不匹配、显存初始化失败等基础问题。
# 用 curl 测试文本生成服务,接口路径需要按实际项目调整 curl -X POST http://127.0.0.1:7860/api/generate \ -H "Content-Type: application/json" \ -d '{"prompt": "hello", "max_tokens": 32}' # 用 curl 测试图像生成服务 curl -X POST http://127.0.0.1:7861/api/generate \ -H "Content-Type: application/json" \ -d '{"prompt": "a small red cube on white background", "width": 512, "height": 512}'如果两个服务分别返回正常结果,说明它们的单服务能力没问题。接下来才能进入双任务并发测试。
5.2 双服务并发压测
双服务并发测试的重点不是看谁跑得快,而是看显存峰值是否触发 OOM,以及两个任务是否互相拖垮。可以写一个简单的 Python 并发脚本,同时向两个服务发起请求。
import requests from concurrent.futures import ThreadPoolExecutor services = [ { "name": "text-service", "url": "http://127.0.0.1:7860/api/generate", "payload": {"prompt": "写一段关于本地部署的短文", "max_tokens": 256}, }, { "name": "image-service", "url": "http://127.0.0.1:7861/api/generate", "payload": {"prompt": "a mountain landscape, 512x512"}, }, ] def call_service(service): try: response = requests.post( service["url"], json=service["payload"], timeout=300, ) return service["name"], response.status_code, response.elapsed.total_seconds() except Exception as exc: return service["name"], -1, str(exc) with ThreadPoolExecutor(max_workers=2) as executor: futures = [ executor.submit(call_service, service) for service in services ] for future in futures: print(future.result())运行这个脚本的同时,另开一个终端持续观察显存变化:
watch -n 1 nvidia-smi如果显存稳定在可用范围内,两个请求都正常返回,说明当前环境可以支撑双服务并发。如果出现 CUDA out of memory,把图像服务的分辨率调小,或者把文本服务的 batch 降到 1,再做第二轮测试。
5.3 判断成功的标准
双服务并发是否算跑通,建议按下面几条判断。
- 两个接口都返回 HTTP 200 或预期的业务状态码。
- 整个推理过程中显存没有触发 OOM。
- 两个任务的总耗时没有出现异常长尾。
- 并发多次后服务依然能响应,没有出现进程退出或假死。
- 日志中没有 CUDA error、Segmentation fault、端口冲突等关键报错。
其中任何一条不满足,都要回到资源占用和模型参数上做调整。
6. 接口 API 与批量任务
双服务并发部署的一个关键价值就是可以把两个模型能力都暴露成 API,再通过脚本编排成一条批量任务流水线。
6.1 API 调用示例
上面已经给了 curl 示例,下面是 Python 请求模板,适合在批量脚本中直接复用。
import requests import time import json def call_model(url, payload, max_retries=3, timeout=300): for attempt in range(max_retries): try: resp = requests.post(url, json=payload, timeout=timeout) if resp.status_code == 200: return resp.json() else: print(f"attempt {attempt + 1}: status {resp.status_code}") except Exception as exc: print(f"attempt {attempt + 1}: {exc}") time.sleep(5) return None # 示例:先调用文本服务,再用结果调用图像服务 story = call_model( "http://127.0.0.1:7860/api/generate", {"prompt": "生成一句简短的风景描述", "max_tokens": 128}, ) if story: image_result = call_model( "http://127.0.0.1:7861/api/generate", {"prompt": story.get("text", "a beautiful landscape"), "width": 512, "height": 512}, ) print(json.dumps(image_result, ensure_ascii=False, indent=2))接口的字段名一定要以实际服务为准,上面只是通用骨架。开发批处理脚本时先把 payload 打印出来,确认服务端接受的字段,再写完整逻辑。
6.2 批量任务设计
批量任务最怕的是脚本没有超时机制、失败没有日志、中间一个任务卡死后面全部排队。工程化的批处理脚本至少要包含三部分:输入队列、失败重试、结果落盘。
import os import json import time import requests from pathlib import Path INPUT_DIR = Path("./inputs") OUTPUT_DIR = Path("./outputs") OUTPUT_DIR.mkdir(exist_ok=True) TEXT_API = "http://127.0.0.1:7860/api/generate" IMAGE_API = "http://127.0.0.1:7861/api/generate" def process_one(item): task_id = item.get("id", int(time.time() * 1000)) result = {"id": task_id, "status": "failed", "output": None} # 第一步:文本生成 text_payload = {"prompt": item["prompt"], "max_tokens": 256} try: text_resp = requests.post(TEXT_API, json=text_payload, timeout=180) text_resp.raise_for_status() text_output = text_resp.json() except Exception as exc: result["error"] = f"text service error: {exc}" return result # 第二步:图像生成 image_payload = { "prompt": text_output.get("text", item["prompt"]), "width": 512, "height": 512, } try: image_resp = requests.post(IMAGE_API, json=image_payload, timeout=300) image_resp.raise_for_status() result["output"] = image_resp.json() result["status"] = "success" except Exception as exc: result["error"] = f"image service error: {exc}" return result def load_tasks(): tasks = [] for file_path in INPUT_DIR.glob("*.json"): with open(file_path, "r", encoding="utf-8") as f: data = json.load(f) if isinstance(data, list): tasks.extend(data) else: tasks.append(data) return tasks def main(): tasks = load_tasks() for item in tasks: result = process_one(item) output_path = OUTPUT_DIR / f"result_{result['id']}.json" with open(output_path, "w", encoding="utf-8") as f: json.dump(result, f, ensure_ascii=False, indent=2) print(f"task {result['id']}: {result['status']}") if __name__ == "__main__": main()这个脚本代表了批量任务的常见结构:输入先落到本地目录,每一步调用一个 API,失败结果落到独立文件,不会影响后续任务。实际使用时需要按照项目接口返回格式调整字段名。
6.3 失败重试与队列顺序
两个重量级任务同时跑的时候,失败大概率来自资源竞争,而不是模型本身坏了。建议采用“失败重试 + 退避等待”的策略,重试次数控制在 3 次以内,重试间隔逐步增加,避免两个服务在恢复过程中又被并发请求压垮。排队逻辑上,如果是单卡环境,尽量不要让文本和图像任务同时进入推理阶段,而是分批提交,先跑完一类任务再跑另一类。
7. 资源占用与性能观察
资源观察是双任务部署最关键的环节,很多时候问题不是“模型能力不够”,而是“显存和内存已经顶到天花板,但控制台没有暴露出来”。
7.1 显存观察方法
最常用的命令是nvidia-smi,也可以加参数做持续监控。
# 每 1 秒刷新一次 GPU 状态 nvidia-smi -l 1 # 输出到日志文件,用于事后分析 nvidia-smi --query-gpu=timestamp,memory.used,memory.total,utilization.gpu \ --format=csv -l 5 > gpu_monitor.log用--query-gpu把显存占用和 GPU 利用率记录到文件里,批量任务跑完后再分析曲线,可以很清楚地看到双任务并发时显存峰值出现在哪个阶段。
7.2 降低显存占用的手段
如果双服务并发测试中出现 OOM,不要急着换显卡,先从下面这些方向调整。
- 启用量化加载,比如把 FP16 模型换成 8-bit 或 4-bit 版本。量化会带来一定精度损失,但可以显著降低显存占用。
- 降低图像生成分辨率,先以 512x512 测试,不要直接上 1024 甚至更高。
- 把 batch size 固定为 1,避免一次加载多批数据。
- 文本模型缩短 max_tokens,限制生成长度,减少中间激活值的显存峰值。
- 在 PyTorch 环境开启低显存模式,例如 xformers、flash attention,具体能不能用取决于模型框架版本。
- 两个任务串行执行,前台任务结束后再启动另一个任务,用时间换空间。
- 为系统设置足够大的 swap 分区,低于显存需求时系统不至于立刻 OOM,但只能作为兜底,不能当作常规手段。
7.3 CPU 推理与 GPU 推理差异
CPU 推理不是不能用,而是速度差异非常大。同一个模型在 CPU 上推理时,耗时通常是 GPU 的几倍到几十倍,具体取决于模型结构和量化方式。如果双任务都走 CPU,内存会成为新的瓶颈,需要重点观察free -h中内存和 swap 的变化。更稳妥的做法是:GPU 跑图像生成这类重负载任务,CPU 跑文本生成这类相对轻量的任务,或者两者都走 GPU 但严格排队。
7.4 端口与进程清理
两个服务跑久了容易留下残留进程,再次启动时报端口被占用。
# 查看 7860 端口进程 lsof -i:7860 # 按 PID 结束残留进程 kill -9 PID # 确认端口已释放 lsof -i:7860启动脚本里建议加一个自动检测端口的逻辑,发现端口被占用时明确报错并打印占用进程 PID,而不是让 Python 直接抛一个难以理解的异常。
8. 双任务并发部署常见问题与排查方法
双服务部署踩坑是常态,提前把排查清单整理好,可以省很多时间。
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 启动后页面打不开 | 端口被占用或服务未启动成功 | 检查日志和进程列表 | 换端口或重启服务 |
| CUDA out of memory | 两个任务显存需求叠加超过上限 | nvidia-smi观察显存曲线 | 降分辨率、开量化、串行执行 |
| 依赖安装失败 | Python 版本或 CUDA 版本不匹配 | 查看 pip 报错信息 | 新建虚拟环境并固定依赖版本 |
| 模型文件缺失 | 权重下载不完整或路径错误 | 对比文件大小和校验值 | 重新下载并配置正确模型路径 |
| 两个服务抢显卡 | 没有设置 CUDA_VISIBLE_DEVICES | nvidia-smi看进程 PID | 显式指定显卡编号 |
| API 调用超时 | 并发过高或设备推理速度慢 | 打印请求耗时与日志 | 增大 timeout,限制并发数 |
| 批量任务卡住 | 脚本没有超时和失败重试 | 检查任务日志和输出目录 | 每个子任务增加超时和重试 |
| 输出质量不稳定 | 显存优化过度或参数设置不合理 | 对比单服务输出 | 调整量化等级和生成参数 |
| 服务假死但进程还在 | 显存泄漏或死锁 | 观察日志最后输出时间 | 设置定时重启或增加健康检查 |
这里面最容易被忽略的是“批量任务卡住”。很多人的批处理脚本写成了for task in tasks: process(task),一旦某个任务里的 API 一直没有返回,整个脚本就会无限挂起。所有外部调用都要设置 timeout,超时后打入失败列表,继续处理下一个任务。
9. 最佳实践与使用建议
双重量级任务并发部署不是简单的“把两个命令都执行起来”,而是一套资源管理流程。下面是几个工程化的建议。
第一,第一阶段先做小参数验证。双服务启动成功后,不要立刻跑完整数据集,先用最小参数测试一遍,确认显存和内存的峰值都在安全范围内。
第二,保留一套最小可运行配置。把两个服务的启动命令、环境变量、端口、模型路径、量化参数记录下来,形成一个 markdown 或 yaml 文件。以后换机器、重装环境,可以直接照着一键恢复。
第三,目录管理要分明。模型权重、输入素材、输出结果、日志文件分别放在不同目录,避免批处理任务把中间产物和最终结果混在一起。
第四,批量任务必须加日志和重试。日志记录每个任务的开始时间、请求参数、返回状态、耗时和错误信息。重试逻辑要幂等,即同一个任务重复执行不会产生重复结果。
第五,接口服务要限制访问范围。默认监听127.0.0.1,如果一定要对外提供服务,加 token 校验或接入内网网关,不要直接暴露到公网。
第六,涉及人脸、声音、版权素材时,必须确认授权。批量生成、对外发布、商用场景下尤其要谨慎,不能简单认为“模型是开源的,所以所有输出都可以用”。
第七,发布或商用前做效果复核。自动化批量任务只能保证“跑起来了”,不能保证“结果可用”,关键内容必须人工抽检。
10. 总结与下一步
双重量级任务并发部署,最值得尝试的点是把两个原本独立运行的服务,通过端口和 API 整合到一条流水线里,让文本生成和图像生成可以相互协作。这个过程不复杂,但真正做起来,显存和内存的分配才是决定成败的关键。
第一次动手时,建议先验证单服务稳定性,再执行并发脚本,同时用nvidia-smi记录显存曲线。最容易踩的坑有两个:一是显存占用比预期高,两个任务同时启动立刻 OOM;二是批处理脚本没有设置 timeout,一个卡死的请求把整条任务队列拖住。这两点只要提前控制参数和加上超时机制,基本都能避开。
后续可以继续扩展的方向包括:把两个服务容器化并用 compose 统一编排;引入消息队列把输入任务削峰填谷;在多卡机器上把不同任务绑定到不同显卡;甚至把一批模型服务接入统一路由层,按请求类型自动分发到对应端口。这样就等于把“一关塞两个重量级选手”的问题,升级成了一台机器上并调度多类模型的基础设施能力。
如果这篇文章对你有帮助,建议收藏备用。下次需要在本地同时跑两个模型服务时,直接按最小验证流程走一遍,能少踩不少坑。