DeepSeek API图像与文本分类生产落地避坑指南

简介：面向具备 Python 基础、希望把云端机器学习能力接入实际项目的开发者，这份文档以 DeepSeek API 的图像分类与文本分类调用为主线，系统梳理从注册账号获取密钥、安装 requests 与 Pillow，到构造 POST 请求、设置鉴权头与请求体、解析分类标签和置信度的完整流程。包内共 1 个 docx 文件，约 17KB，内容以步骤说明配合可运行代码示例展开，覆盖图像字节流上传、文本 JSON 传参、响应状态判断、错误信息排查以及图像尺寸预处理等关键环节，并给出文本分类的迁移示例，提示更换请求地址与数据即可适配其他接口。已有 2676 人学习，适合需要快速理解云端 AI 分类服务接入方式、对照示例完成图像或文本分类原型搭建与联调排错的开发团队参考；文中接口地址和第三方工具也提醒以官方文档及项目实际为准。

1. 一张 224×224 的图跑通之后，真正的活才刚开始

某团队做产线物料分拣，第一版分类服务两天就跑通了：一张图 POST 上去，返回label和confidence，前端直接显示。上线第三天挂了两次——一次是密钥轮换后镜像里还是旧值，一次是有人传了张 12MB 的原图，接口直接拒了。这套 DeepSeek API 的图像分类与文本分类调用流程本身只有四步：拿密钥、装requests和Pillow、发 HTTP 请求、解析 JSON 响应。真正决定它能不能进生产的，恰恰是这四步之外的东西：请求体的 MIME 形态、图像体积与色彩模式、状态码分支、批量限速。这篇按我自己落地的顺序拆一遍，适合已经有 Python 基础、准备把云端机器学习服务接进业务系统的开发者。

2. 调用链拆开：鉴权、multipart 请求体与 JSON 响应

这一章先把单次图像分类请求完整走通，重点不在“能跑”，而在每一层为什么这么写。HTTP 调用看起来简单，但请求头、请求体编码、鉴权方式这三处任意一个写错，报出来的错误信息都不指向真正的原因。

2.1 环境与密钥：三个依赖、一个环境变量

先用虚拟环境隔离，依赖只有两个是必须的：requests负责发 HTTP 请求，Pillow负责读图、转码、压体积。第三个可选的是python-dotenv，本地开发时用来从.env读密钥，避免把密钥敲进代码。

# 建虚拟环境，避免和系统里的 requests 版本打架 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate # 两个核心依赖，用于 API 调用与图像处理 pip install requests Pillow pip install python-dotenv # 可选，本地读取 .env

目录我一般固定成这个样子，后面排错时日志和原始图能对上号：

project/ ├── .env # 只放本地密钥，.gitignore 里必须加 ├── client.py # 封装好的调用客户端 ├── assets/ # 待处理的原图 ├── out/ # 预处理后的图与结果 JSON └── logs/ # 每次请求的状态码与耗时

密钥不要硬编码。api_key = 'your_api_key_here'这种写法在示例里没问题，进了仓库就是事故源。读取方式：

import os from dotenv import load_dotenv load_dotenv() # 从 .env 加载，生产环境直接用系统环境变量 API_KEY = os.environ["DEEPSEEK_API_KEY"] if not API_KEY or len(API_KEY) < 16: # 启动即失败，别等到发请求才报 401 raise RuntimeError("API key 未配置或格式可疑，检查环境变量")

这里的关键点是“启动时校验”。很多人把密钥检查放到第一次请求里，结果是服务启动正常、健康检查通过，第一批业务流量打进来才发现全是 401，排查成本直接从秒级变成分钟级。参数上，os.environ[...]用中括号而不是get，是为了缺失时立刻抛KeyError，宁可启动失败也不要带病运行。

2.2 图像分类的请求体：files 和 json 为什么不能混用

图像分类接口走的是文件上传，请求体必须是multipart/form-data。这一点决定了两个参数不能同时出现：files=和json=。json=会把请求体序列化成application/json，图片二进制塞不进 JSON，只能转 base64，体积膨胀三分之一还多；files=则由requests自动生成 boundary 并拼装 multipart 体。

import requests from pathlib import Path API_KEY = os.environ["DEEPSEEK_API_KEY"] # 接口地址以官方文档为准，这里仅作结构演示 URL = "https://api.deepseek.com/v1/classify" # 只写 Authorization，Content-Type 交给 requests 自动生成 headers = {"Authorization": f"Bearer {API_KEY}"} image_path = Path("assets/sample.jpg") # 用二进制模式读，不要用文本模式，否则字节会被破坏 with open(image_path, "rb") as f: files = {"image": (image_path.name, f, "image/jpeg")} resp = requests.post(URL, headers=headers, files=files, timeout=(3.05, 30)) print(resp.status_code) print(resp.text[:300]) # 先看原文，别急着 .json()

逻辑说明：files的字典值是一个三元组(文件名, 文件对象, MIME 类型)，requests会据此组装出Content-Disposition和Content-Type两段头部。最常见的翻车是画蛇添足，在headers里手写"Content-Type": "multipart/form-data"——手写之后 boundary 丢了，服务端解析不出文件分段，返回的往往是一个语焉不详的 400。正确做法是让requests自己设。

参数说明：timeout=(3.05, 30)是连接超时和读取超时的元组。连接超时给 3 秒出头是因为 TCP 重传窗口的常见经验值；读取超时给 30 秒，是因为图像推理在队列拥堵时确实会慢，给太短会误杀。不写timeout是完全不能接受的，requests默认无限等待，批量任务里会整体挂死。

2.3 响应解析与状态码分支：200 之外还有四种情况

成功响应长这样：

{ "status": "success", "data": { "label": "cat", "confidence": 0.98 } }

解析时我坚持两条规则：先判 HTTP 状态码，再判业务status字段。两层都过，才取data.label和data.confidence。因为网关层的 200 不等于业务成功，有些错误会以 200 包一个{"status": "error", "message": "Invalid API key"}返回。

def parse_classify_response(resp): # 第一层：HTTP 状态码 if resp.status_code == 401: raise PermissionError("密钥无效或已过期，检查 Authorization 头") if resp.status_code == 413: raise ValueError("图像体积超限，需要先压缩") if resp.status_code == 429: raise RuntimeError("触发限流，需要退避重试") if resp.status_code >= 500: raise RuntimeError(f"服务端异常 {resp.status_code}，可重试") if resp.status_code != 200: raise RuntimeError(f"未预期状态码 {resp.status_code}: {resp.text[:200]}") # 第二层：业务状态字段 body = resp.json() if body.get("status") != "success": raise RuntimeError(f"业务失败: {body.get('message')}") data = body["data"] label = data["label"] conf = float(data["confidence"]) # 置信度低于阈值时不直接采信，交给上层人工复核 return label, conf, conf >= 0.8

参数说明：0.8这个阈值不是拍脑袋的，后面第 6 章会用回归集扫一遍分桶准确率来确定。resp.text[:200]截断是为了日志不炸，但排查时一定要打全量，我曾经因为截断把真正有用的message字段切掉，白白多查了两小时。resp.json()放在状态码判断之后，是因为错误响应不一定是合法 JSON，提前解析会抛JSONDecodeError掩盖真实错误码。

3. 文本分类只是换个请求体：把一次性脚本改成可复用客户端

图像和文本两个接口的差别只在请求体：一个files=，一个json=。鉴权头、状态码语义、响应结构基本一致。所以第二件事不是再抄一遍脚本，而是抽一个客户端出来，把重试、超时、并发这些东西收敛到一处。

3.1 文本分类的最小闭环

文本分类用json=传参，requests会自动把Content-Type设为application/json并序列化请求体，同样不要手写这个头。

import requests url = "https://api.deepseek.com/v1/text_classify" # 以官方文档为准 headers = {"Authorization": f"Bearer {API_KEY}"} payload = { "text": "Deep learning models are revolutionizing the AI field." } # json= 自动序列化并设置 application/json resp = requests.post(url, headers=headers, json=payload, timeout=(3.05, 15)) if resp.status_code == 200 and resp.json().get("status") == "success": data = resp.json()["data"] print(data["label"], data["confidence"]) # 例如 Technology 0.95

这里的参数选择有几个细节值得说。第一，文本接口的读取超时我压到 15 秒，比图像短，因为文本推理的耗时不随输入线性增长得太夸张，超时太长只会拖慢整批任务。第二，payload里不要塞超长文本，多数接口对单次输入有字符上限，超了返回 400，最好在客户端侧先截断或分片。第三，label的取值集合是模型侧定义的，不要在前端硬编码枚举，用接口返回什么就展示什么，否则模型升级后前端会整片空白。

3.2 用 Session 封装：超时、连接复用与重试

单次调用用requests.post没问题，批量场景必须换Session。原因很实在：每次requests.post都要重新做 TCP 握手和 TLS 协商，几十次调用下来光握手就吃掉可观的时间，而且连接数飙升容易触发对端限流。

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(api_key: str) -> requests.Session: session = requests.Session() session.headers.update({ "Authorization": f"Bearer {api_key}", "User-Agent": "classify-client/1.0", }) # 只对幂等友好的错误重试：429 限流、500/502/503/504 服务端 retry = Retry( total=3, backoff_factor=0.8, # 0.8s, 1.6s, 3.2s 指数退避 status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["POST"], # 分类请求可安全重放 raise_on_status=False, ) adapter = HTTPAdapter( max_retries=retry, pool_connections=8, # 连接池按 host 缓存 pool_maxsize=16, ) session.mount("https://", adapter) return session

逻辑说明：Retry的backoff_factor决定了退避节奏，第一次重试等 0.8 秒，第二次 1.6 秒，第三次 3.2 秒，配合urllib3自带的抖动，能有效避开限流窗口。allowed_methods里放POST需要一点判断：分类请求是纯查询语义、不带副作用，重放安全；如果接口会扣费或写数据，就必须单独评估。

参数说明：pool_maxsize=16要和后面的并发数对齐，并发 8 的时候池子给 16 有余量；raise_on_status=False让重试耗尽后把响应原样返回，方便上层拿到真实状态码打日志，而不是抛一个信息量很低的RetryError。

3.3 批量任务的分批与并发上限

批量跑图时，我一般把并发压在 4 到 8 之间。这个数字不是性能上限，是安全边界：并发再高，限流一触发，重试风暴反而让整体吞吐下降。

from concurrent.futures import ThreadPoolExecutor, as_completed from pathlib import Path import time def classify_one(session, path: Path): with open(path, "rb") as f: files = {"image": (path.name, f, "image/jpeg")} try: resp = session.post(URL, files=files, timeout=(3.05, 30)) return path.name, resp.status_code, resp.text[:500] except requests.RequestException as e: # 网络层异常单独记，不要和业务错误混在一起 return path.name, -1, repr(e) paths = list(Path("assets").glob("*.jpg")) failed = [] # 每批 100 张，批间 sleep 1 秒，给对端留出恢复窗口 for i in range(0, len(paths), 100): batch = paths[i:i + 100] with ThreadPoolExecutor(max_workers=6) as pool: futures = [pool.submit(classify_one, session, p) for p in batch] for fut in as_completed(futures): name, code, body = fut.result() if code != 200: failed.append((name, code, body)) time.sleep(1) print(f"失败 {len(failed)} 条，待二次重跑")

逻辑说明：分批的目的是把限流风险局部化，一批出问题不会污染全量。failed单独收集而不是当场重试，是为了避免“重试放大”——已经限流了还在原地疯狂重试，只会让情况更糟。批间sleep一秒是保守值，如果对端限流宽松可以去掉。

参数说明：max_workers=6配合前面的pool_maxsize=16是安全的；resp.text[:500]比只存状态码有用得多，很多 400 的根因就写在响应体里。

4. 图像预处理：尺寸、色彩模式与字节体积的三重约束

图像分类的失败案例里，一大半不是接口问题，是图的问题。模型侧对输入有尺寸和格式要求，网关侧对请求体有体积上限，客户端这一层必须在发送前把三件事摆平：方向、色彩模式、字节数。

4.1 尺寸与 EXIF 方向：resize 之前先 transpose

手机拍的图带 EXIF 旋转标记，直接resize会把方向“焊死”在错误状态。比如一张竖拍的人像，读出来是横的，缩到 224×224 之后模型看到的是一个躺着的目标，分类结果自然离谱。

from PIL import Image, ImageOps def load_and_normalize(path: str, size=(224, 224)) -> Image.Image: img = Image.open(path) # 按 EXIF 方向标记纠正，必须在 resize 之前 img = ImageOps.exif_transpose(img) # 统一转 RGB，处理 RGBA / P / CMYK 等模式 if img.mode != "RGB": img = img.convert("RGB") img = img.resize(size, Image.LANCZOS) return img

逻辑说明：exif_transpose读的是 EXIF 的 Orientation 标签，它不会修改像素内容之外的元数据，是纯内存操作。顺序很重要——先转置再缩放，否则缩放后 EXIF 可能被丢弃，方向信息就永久丢了。

参数说明：224×224是原文里的假设值，实际尺寸必须以官方文档为准；Image.LANCZOS是缩小时质量最好的重采样算法，代价是比BILINEAR慢一点，但在 224 这个量级上差异可以忽略。如果你的业务是细粒度分类（比如区分相似零件），不要盲目拉伸，改成等比缩放加灰边填充能保住长宽比。

4.2 RGB 转换与压缩质量：PNG 透明通道的坑

PNG 带 alpha 通道，转 JPEG 保存会直接抛OSError: cannot write mode RGBA as JPEG。就算你绕过去了，透明区域在某些解码器里会变成黑底，分类置信度肉眼可见地掉。

def to_jpeg_bytes(img: Image.Image, quality: int = 85) -> bytes: from io import BytesIO buf = BytesIO() # 透明通道贴白底，再存 JPEG if img.mode == "RGBA": bg = Image.new("RGB", img.size, (255, 255, 255)) bg.paste(img, mask=img.split()[3]) img = bg img.save(buf, format="JPEG", quality=quality, optimize=True) return buf.getvalue() raw = to_jpeg_bytes(load_and_normalize("assets/a.png")) print(f"{len(raw) / 1024:.1f} KB")

逻辑说明：img.split()[3]取的是 alpha 通道作为粘贴蒙版，透明部分自然露出白底。这一步在物料、票据、白底商品图的场景里尤其重要，透明背景换成黑底会显著改变整体色彩分布。

参数说明：quality=85是我默认的起点，视觉效果和体积的平衡点；optimize=True会多花一点 CPU 换更小的体积，批量场景下值得开。

4.3 发送前的体积校验与降级策略

网关的体积上限通常在几 MB 量级，具体值看文档。客户端不该指望“传上去再说”，而是发送前过一遍体积检查，超了就降质量重编，而不是直接报错给用户。

def shrink_until_ok(img: Image.Image, limit_bytes=5 * 1024 * 1024) -> bytes: for q in (85, 70, 55, 40): data = to_jpeg_bytes(img, quality=q) if len(data) <= limit_bytes: return data # 四档都压不下来，说明尺寸本身过大，等比缩一半再来 w, h = img.size return shrink_until_ok(img.resize((w // 2, h // 2)), limit_bytes)

逻辑说明：逐级降质量比一次性压到最低更合理，因为多数图在第一档就过了，画质损失最小。递归缩尺寸是兜底路径，正常业务里很少触发。这段代码要配一个计数器打日志，我一般会统计“降级率”，如果某个来源的图频繁走到第三档，说明上游采集环节该管了。

5. 排查与避坑：五条上线后才会遇到的记录

这一章的每一条都对应真实环境里出现过的现象，按“现象 → 原因 → 解决”写，能对号入座就别重复踩。

5.1 本地跑得通，线上全是 401

现象：同一份代码，开发机返回正常分类结果，部署到容器后所有请求 401，日志里只有Invalid API key。

原因：密钥硬编码在代码里或者写死在镜像的环境变量中，平台上轮换了密钥，镜像没重建。另一种情况是.env被打进了镜像，读取顺序上被系统环境变量覆盖，读到的是空字符串。

解决：密钥只从运行时注入，启动时做一次非空和长度校验；401 单独建告警指标，不要和 400 混在一个错误率里，否则密钥过期会被海量的参数错误淹没。我自己还加了一条：客户端初始化时发一次极小的探活请求，失败就让进程退出，别让一个必然失败的实例接流量。

5.2 手写 Content-Type 导致 400

现象：图像上传稳定返回 400，响应体里提到 multipart 解析失败或者缺少 boundary，但同一个接口用 curl 上传是好的。

原因：headers里同时写了"Content-Type": "multipart/form-data"和用了files=。requests检测到手写的 Content-Type 就不再追加boundary=...，服务端拿不到分段标记，整个请求体解析不出来。

解决：用files=时把 Content-Type 从 headers 里删干净，只留 Authorization。判断方法很直接：打印resp.request.headers，看 Content-Type 尾部有没有boundary=，没有就是这个问题。

5.3 RGBA 图上传后置信度异常低

现象：同一张设计稿，导出 JPG 版本分类正常，导出 PNG 版本置信度从 0.96 掉到 0.4 以下，有时还落到错误类别。

原因：PNG 带透明通道，直接交给服务端后透明区域被渲染成黑色或随机色，图像整体色彩分布被改掉。模型看到的不是你屏幕上看到的那张图。

解决：发送前统一convert("RGB")，透明通道贴白底（见 4.2）。这条我在设计类、电商类项目里遇到过不止一次，血泪经验是：凡是来源不可控的图，一律先过一遍标准化管线，不要相信“看起来没问题”。

5.4 批量任务跑到一半整体卡死

现象：单张请求几秒返回，批量跑几百张时进程挂住不退出，日志停在某一条上，没有任何超时异常。

原因：requests不写timeout就是无限等待。对端连接建立了但迟迟不返回响应体，客户端就一直挂在读操作上，线程池里的线程被逐个占满，最后整个池子堵死。

解决：所有请求强制走timeout=(连接超时, 读取超时)元组，并且捕获requests.Timeout单独记日志。更进一步，ThreadPoolExecutor外面套一层整体超时，别让单条请求无限期拖住整批。

5.5 限流之后越重试越堵

现象：触发 429 之后错误率不降反升，监控上 QPS 打到一个高位不动，日志里全是重试记录。

原因：重试没有退避，或者退避因子设得过小，加上并发数本来就在临界点，重试流量把仅有的配额又吃掉，形成正反馈。

解决：backoff_factor不要低于 0.5，重试次数封顶 3 次；429 单独识别并触发全局降速，而不是让每个线程各自重试。我一般的做法是给客户端加一个简单的信号量，收到 429 就把并发从 6 降到 3，稳定几分钟后再慢慢升回去。

6. 进阶：用一个离线回归集验证分类结果能不能进生产

接口跑通不等于能上线。真正要回答的问题是：模型的输出在你的业务数据上，准到什么程度，错在哪些类别上，置信度能不能当阈值用。这三件事只能靠一个带人工标注的离线回归集来答。

做法不复杂：从业务数据里抽 200 到 500 张（或条），人工标注真实类别，跑一遍客户端，把结果落成 JSONL。文本分类同理，只是把图换成文本字段。

import json from collections import defaultdict def evaluate(records, threshold=0.8): total = correct = accepted = accepted_correct = 0 buckets = defaultdict(lambda: [0, 0]) # 分桶: [命中, 总数] for r in records: total += 1 hit = (r["label"] == r["true_label"]) correct += hit if r["confidence"] >= threshold: accepted += 1 accepted_correct += hit # 按置信度分桶，看阈值设在哪儿最划算 b = round(r["confidence"] * 10) / 10 buckets[b][1] += 1 buckets[b][0] += hit return { "accuracy": correct / total, "coverage": accepted / total, # 自动通过的比例 "precision_at_threshold": accepted_correct / max(accepted, 1), "buckets": dict(sorted(buckets.items())), }

判断标准我一般看三个数：整体准确率决定这条路走不走得通；coverage决定能自动化多少，剩下的转人工；precision_at_threshold决定自动化那部分的错误率业务能不能忍。如果 0.8 阈值下只覆盖 40% 但精确率 0.98，那是个不错的“高置信自动过、低置信转人工”的分流结构；如果 0.8 阈值下覆盖率 90% 但精确率只有 0.7，说明置信度分布和真实准确率严重不对齐，阈值不能直接用。

分桶表是最有用的产物，扫一遍就知道阈值该往哪挪：

还有一个容易被忽略的技巧：回归集要按类别分层抽样。如果业务数据里某一类占八成，随机抽 200 条会让小类样本只有个位数，那个类别上的准确率就完全没有统计意义，而恰恰是小类最容易出问题。我一般让每个类别至少 20 条，类别不足的先补数据再谈评估。

最后是版本对照。回归集跑出的结果要带客户端版本号和模型标识一起存档，每次改预处理参数、改阈值、换接口版本，都重跑一遍对比。没有 baseline 的优化全是玄学。

这套流程我自己是吃过亏才固化下来的：早先一个项目凭感觉把置信度阈值定在 0.9，上线后自动通过率不到两成，人工复核堆成山，回看才发现 0.75 那一档的准确率其实只差一个百分点。从那以后我每次接新的分类接口，都先花半天把回归集和分桶表跑出来，再谈阈值和并发。希望帮到你。

本文还有配套的精品资源，点击获取