简介：面向具备Python基础的开发者与正在集成自动化AI分类能力的开发团队，这份指南围绕DeepSeek提供的图像分类与文本分类API，梳理从注册获取密钥、安装requests与Pillow，到构造POST请求、设置HTTP头部、准备请求主体并解析分类标签与置信度的完整调用链路。文档结合代码实例说明图像上传、文本分类请求及响应处理，并给出密钥失效、请求负载格式异常等常见问题的排查思路，同时提示图像尺寸压缩、接口URL以官方文档为准等实践注意事项，便于读者快速搭建分类应用原型。资源包共1个docx文件，大小约17KB，以文字说明与示例代码为主，结构紧凑、便于随查随用。目前已有2676人学习，适合需要理解API接入流程、对照示例完成图像与文本分类任务的中初级Python开发者参考。

1. 一张图 + 一段文本，先把调用链路跑通

把一个本地图片塞进 HTTP 请求、拿回一个 label 和 confidence，这件事本身只要二十行 requests 代码；真正让人卡住的，是图像分类和文本分类这两条链路在请求体上的差别，以及密钥、超时、重试这些没人愿意写进示例的东西。这份资源给的是一套可直接复现的调用骨架：文本分类走 JSON，图像分类走 multipart，Pillow 负责预处理，requests 负责传输，响应当成结构化数据来处理。它解决的是「云端机器学习服务怎么接进自己的系统」这个落地问题，适合已经写过 Python、手里有一批图片或文本要打标、又不想自己训模型的开发者。下面按接口契约、环境复现、批量工程化、避坑排查四段推进，每一步都能直接抄。

2. 摸清接口契约：图像与文本两条链路的差别

很多人拿到示例代码第一反应是复制粘贴改个 URL，然后发现图像那条链路一直报 415 或者 400，文本那条却一次就通。差别不在密钥，也不在网络，而在请求体的封装方式：一个是 multipart/form-data，一个是 application/json，这两种格式对 HTTP 头、对文件句柄、对重试策略的要求完全不一样。所以动手之前，先把接口契约里最不容易变的那几个常量固定下来，比先写业务逻辑划算得多。

2.1 端点、方法与认证：三个先钉死的常量

无论后面封装多少层，最终发出去的请求只依赖三样东西：地址、方法、凭证。地址由 base_url 加路径拼成，方法固定 POST，凭证放在 Authorization 头里，形式是Bearer <key>。把这三样抽到一个配置文件里，好处是换环境、换账号、换模型版本时只改一处，不用在几十个函数里搜字符串。我一般还会额外加一个 User-Agent，不是接口要求，而是出问题时能在服务端日志里认出是自己的客户端。

这里有个容易被忽略的细节：超时必须写成元组。requests 只给一个数字时，它同时作用于连接阶段和读取阶段，结果就是连接很快但模型推理慢的场景下，请求会在读取阶段被误杀。图像分类的耗时通常比文本分类高一个量级，两者用同一个超时值本身就是隐患，我习惯给图像单独设 60 秒读取上限。

# config.py —— 把会变的东西关进一个文件 import os BASE_URL = os.environ.get("API_BASE_URL", "https://api.example.com/v1") # 换成控制台文档地址 IMAGE_ENDPOINT = "/classify" # 图像分类，示例路径 TEXT_ENDPOINT = "/text_classify" # 文本分类，示例路径 API_KEY = os.environ["API_KEY"] # 缺失时直接 KeyError，比拿到 None 再报 401 好排查 CONNECT_TIMEOUT = 5 READ_TIMEOUT_IMAGE = 60 READ_TIMEOUT_TEXT = 20

这段代码的关键点是os.environ["API_KEY"]用中括号而不是 get。用 get 拿不到会返回 None，程序继续往下跑，最后在服务端换来一个 401，你还得回头猜是密钥错了还是没读到。用中括号会在启动阶段就抛 KeyError，错误位置清晰。

2.2 图像走 multipart，文本走 JSON：请求体构造的两种写法

图像这类二进制内容，标准做法是用files参数交给 requests 自己拼 multipart 边界；文本这类结构化内容，用json=参数让它序列化并自动带上 Content-Type。这两条路各走各的，最忌讳的是「统一封装」——把图像也塞进 json 里做 base64，体积会膨胀三分之一还多，很多服务端还会直接拒收。

import io import os import requests from requests.adapters import HTTPAdapter from config import (BASE_URL, IMAGE_ENDPOINT, TEXT_ENDPOINT, API_KEY, CONNECT_TIMEOUT, READ_TIMEOUT_IMAGE, READ_TIMEOUT_TEXT) def build_session(pool_size: int = 8) -> requests.Session: """复用连接，避免每次请求都重新握手""" s = requests.Session() adapter = HTTPAdapter(pool_connections=pool_size, pool_maxsize=pool_size, max_retries=0) s.mount("https://", adapter) s.headers.update({ "Authorization": f"Bearer {API_KEY}", "User-Agent": "classify-client/1.0", }) return s def classify_image(session, image_bytes: bytes, filename: str = "upload.jpg"): files = {"image": (filename, image_bytes, "image/jpeg")} resp = session.post( BASE_URL + IMAGE_ENDPOINT, files=files, # 不要手写 Content-Type，requests 会带 boundary timeout=(CONNECT_TIMEOUT, READ_TIMEOUT_IMAGE), ) resp.raise_for_status() return resp.json() def classify_text(session, text: str, max_chars: int = 2000): payload = {"text": text[:max_chars]} resp = session.post( BASE_URL + TEXT_ENDPOINT, json=payload, # 自动序列化 + 自动 Content-Type: application/json timeout=(CONNECT_TIMEOUT, READ_TIMEOUT_TEXT), ) resp.raise_for_status() return resp.json()

先看图像这一段。files的值是一个三元组：字段名、文件名、MIME 类型。字段名必须和服务端约定的名字一致，这个值写错会得到一个「缺少 image 字段」的 400，而不是「格式不支持」，排查方向容易跑偏。文件名虽然是字符串，但服务端可能靠后缀判断类型，随便写个blob有风险，保留.jpg更稳。传进去的是 bytes 而不是文件句柄，是为后面的重试铺路——句柄被读一次就到末尾了，重试时发出去的是空内容。

文本这一段更简单，json=会做两件事：序列化字典、设置 Content-Type。唯一需要注意的是长度截断，text[:max_chars]是硬截断，会把句子切一半；如果业务对语义完整敏感，应该按句号切分后再截，或者干脆在入库前就把超长文本拆条。截断值不要凭感觉，先看看服务端文档里写的上限，留 10% 余量。

2.3 响应解析与置信度阈值：把 JSON 变成能用的判断

接口正常时返回的结构大致是一个外层状态加一个数据体，数据体里带label和confidence。示例里那组{"status": "success", "data": {"label": "cat", "confidence": 0.98}}就是典型形状。直接把resp.json()["data"]["label"]写进业务代码是最省事也最危险的做法——一旦服务端改成result或者出错时返回{"status": "error", "message": "..."}，你的代码会抛 KeyError，而这个异常在日志里看起来像代码 bug，实际是数据问题。

def parse_result(payload: dict, threshold: float = 0.6) -> dict: """把接口响应规范化成统一结构，异常情况显式抛出""" if not isinstance(payload, dict): raise ValueError(f"响应不是 JSON 对象: {type(payload)}") if payload.get("status") != "success": raise RuntimeError(f"接口返回失败: {payload.get('message', 'no message')}") data = payload.get("data") or {} label = data.get("label") if not label: raise ValueError(f"响应缺少 label 字段: {payload}") try: conf = float(data.get("confidence", 0.0)) except (TypeError, ValueError): conf = 0.0 return { "label": label, "confidence": conf, "accepted": conf >= threshold, # 低于阈值进人工复核队列 }

这段解析做了三件事。第一，先判status，把业务失败和服务失败分开，前者抛 RuntimeError、后者抛 HTTPError，日志里一眼能区分。第二，data用or {}兜底，防止服务端返回"data": null时None.get崩掉。第三，置信度统一转 float 并给默认值，因为有些实现会返回字符串"0.98"，直接比较大小在 Python 里会抛 TypeError。

阈值 0.6 是起点不是答案。图像分类里常见的猫狗二分类，0.6 以下基本可以当噪声处理；但如果是工业质检那种把「缺陷」当正类的场景，阈值要往 0.85 以上提，宁可多进人工也别漏检。判断阈值合不合适，不能看单张，要拿一批带标注的样本跑一遍，画出准确率和复核率的取舍曲线，这事儿在第 6 章会具体说。

3. 环境与最小复现：requests 与 Pillow 的落地细节

把示例代码贴进编辑器能跑，和把它交给同事也能跑，是两件事。中间差的不是技术难度，是环境隔离、密钥管理、图片兼容性这几个琐碎点。这一章按「装依赖 → 处理图 → 封装请求」的顺序走，每一步都能独立验证，出问题也好定位。

3.1 依赖安装与密钥管理：别把 key 写进代码

依赖只有两个：requests 负责 HTTP，Pillow 负责图像。用虚拟环境装，别往全局环境里怼，否则半年后你会在某个报错里发现是另一个项目的旧版本在捣乱。

python -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate python -m pip install --upgrade pip pip install requests Pillow # 生成依赖清单，锁定当前版本 pip freeze > requirements.txt

密钥管理这块有个血泪经验：只要 key 出现过一次在代码里，它就已经进了 git 历史，后面删掉也没用。正确做法是环境变量或本地.env文件，并且把.env写进.gitignore。

# .env 文件（不要把这一行提交到仓库） API_KEY=sk-xxxxxxxxxxxxxxxx API_BASE_URL=https://api.example.com/v1 # 启动前加载 export $(grep -v '^#' .env | xargs) # 临时用，生产环境交给配置中心

grep -v '^#'过滤注释行，xargs把KEY=VALUE形式拆成 export 的参数。这个写法在值是纯字母数字时没问题，如果 key 里带特殊字符（比如#或空格）会截断，那就老老实实用 python-dotenv 或者容器环境变量注入。团队协作时我一般直接把.env.example提交上去，里面只放占位符，新人复制一份改名即可。

3.2 图像预处理三道关：模式、尺寸、体积

接口文档里写了「224×224」，不代表你传别的尺寸就一定失败，但一定会有某几张图在某些边界条件下失败，而这类失败最难复现。预处理要过三道关：颜色模式、尺寸、字节体积。

import io from PIL import Image, ImageOps def prepare_image(src_path: str, size=(224, 224), max_bytes: int = 1_500_000, min_quality: int = 60) -> tuple[io.BytesIO, str]: with Image.open(src_path) as im: im = ImageOps.exif_transpose(im) # 手机照片方向纠正，缺了这一步图是躺着的 if im.mode != "RGB": im = im.convert("RGB") # RGBA / P / CMYK / L 统一转三通道 im = ImageOps.fit(im, size, method=Image.BICUBIC, centering=(0.5, 0.5)) quality = 88 buf = io.BytesIO() im.save(buf, format="JPEG", quality=quality, optimize=True) while buf.tell() > max_bytes and quality > min_quality: quality -= 8 buf = io.BytesIO() im.save(buf, format="JPEG", quality=quality, optimize=True) buf.seek(0) # 指针必须归零，否则发出去是空内容 return buf, f"upload_q{quality}.jpg"

ImageOps.exif_transpose解决的是手机竖拍照片存成横向像素、靠 EXIF 标记旋转的问题。上传前不纠正，服务端拿到的是躺倒的图，分类结果自然离谱，而且这种错你从文件上根本看不出来。convert("RGB")解决的是 PNG 透明底和灰度图的问题，四通道数据塞进只认三通道的模型，服务端可能直接 422。

ImageOps.fit和resize的区别值得说清楚：resize会把图拉变形，长宽比变了，物体形状跟着变；fit先居中裁剪成正方形再缩放，保持长宽比但会丢边缘。做分类任务我优先用fit，因为形状失真对模型的影响通常比丢一点边缘大。如果主体贴着边缘（比如证件、票据），就得换成 padding 补边的做法。

体积那道循环是兜底。很多接口对单文件有大小上限，超了返回的还是 400，错误信息未必直白。降到最低质量还超限，就说明这张图本身该先缩边长，而不是继续压质量。

3.3 一份能直接跑的封装：超时、重试与日志

单次调用能跑之后，下一步是把「出错怎么办」写进去。接口侧的限流、网络抖动、偶发的 5xx，在批量场景里是必然会遇到的，不处理就是跑到一半崩掉。

import random import time import logging RETRY_STATUS = {429, 500, 502, 503, 504} log = logging.getLogger("classify") def post_with_retry(session, url, *, max_attempts=4, **kwargs): for attempt in range(1, max_attempts + 1): try: resp = session.post(url, **kwargs) except requests.RequestException as exc: if attempt == max_attempts: raise wait = min(2 ** attempt, 8) * (0.5 + random.random()) # 指数退避 + 抖动 log.warning("网络异常 %s，第 %d 次重试，等待 %.1fs", exc.__class__.__name__, attempt, wait) time.sleep(wait) continue if resp.status_code in RETRY_STATUS and attempt < max_attempts: wait = min(2 ** attempt, 8) * (0.5 + random.random()) log.warning("状态码 %s，%.1fs 后重试，第 %d 次", resp.status_code, wait, attempt) time.sleep(wait) continue return resp return resp

三个参数值得逐个交代。max_attempts=4是经验值，第一次 + 三次重试，绝大多数抖动都能扛过去，再多就是在浪费额度。退避用2 ** attempt但上限压到 8 秒，因为批量任务的总时长对业务有感知，不能无限等。random.random()这个抖动很关键——一批任务同时被限流时，如果所有客户端都按同一个间隔重试，下一波还是同时打过去，雪崩。加上抖动后重试时间点被打散，压力自然摊平。

max_retries=0是故意的。urllib3 自带的 Retry 对 POST 默认不重试，行为不够透明；更重要的是，如果files里传的是文件句柄，重试时读到的是空内容，服务端会返回一个「图片格式不支持」的 400，你会以为是图坏了。所以文件一律先读成 bytes 再进重试逻辑，这也是上一节prepare_image返回 BytesIO 而不返回路径的原因之一。

4. 批量与工程化：从单张调用到流水线

单张调通的代码拿去跑一万张图，会遇到三个新问题：太慢、断了要从头再来、跑完不知道花了多少。这三个问题不解决，脚本就只能在本地玩玩，进不了业务。

4.1 分批与并发：线程池大小怎么定

这类调用是典型 IO 密集型，CPU 基本闲着，所以用线程池而不是多进程。并发数不是越大越好，它同时受三方面约束：本地连接池、服务端限流、你的额度。

from concurrent.futures import ThreadPoolExecutor, as_completed def batch_classify_images(session, paths, workers=4, batch_log_every=50): results, errors = [], [] with ThreadPoolExecutor(max_workers=workers) as pool: futures = {pool.submit(classify_image, session, load_bytes(p), p): p for p in paths} for i, fut in enumerate(as_completed(futures), 1): path = futures[fut] try: results.append({"path": path, "raw": fut.result()}) except Exception as exc: errors.append({"path": path, "error": repr(exc)}) if i % batch_log_every == 0: log.info("进度 %d/%d，失败 %d", i, len(paths), len(errors)) return results, errors def load_bytes(path: str) -> bytes: buf, _ = prepare_image(path) return buf.read()

workers=4是保守起点。先按 4 跑 100 张，看平均耗时和失败率；如果全是 429，说明撞了限流，往下调到 2；如果耗时远大于单张乘以数量再除以并发，多半是连接池不够——build_session(pool_size)要跟 workers 对齐，否则线程在等连接。并发调优的正确顺序是：先确认单张耗时稳定，再往上加并发，每次翻倍，观察到失败率开始抬头就退回上一档。

as_completed而不是按提交顺序取结果，是为了让慢任务不阻塞后面的记录。这在图像大小差异很大时尤其明显，一张超大图能拖住整个批次。

4.2 结果落盘与断点续跑：用 jsonl 记录每一条

批量任务跑到 80% 崩掉、重跑又从第一张开始，是最浪费时间的场景。解决办法是每处理完一条就追加写一行 JSON，启动时把已完成的路径读进内存跳过。

import json from pathlib import Path def append_result(fp: Path, record: dict): with fp.open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") def load_done(fp: Path) -> set[str]: if not fp.exists(): return set() done = set() with fp.open(encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: done.add(json.loads(line)["path"]) except (json.JSONDecodeError, KeyError): continue # 半行损坏直接跳过，不影响整体续跑 return done

用 jsonl 而不是 csv，是因为每行独立、追加安全，中途断电最多丢最后半行，load_done里的 try/except 正好把半行吃掉。用 csv 的话，一个引号没转义整份文件就废了。文件名我习惯带上任务标识和日期，重跑时不会覆盖历史。

4.3 一次小规模压测：把耗时和错误率量出来

上线前必须知道三个数：单张平均耗时、P95 耗时、错误率。这三个数决定并发数和重试预算，也决定了要不要做结果缓存。

我一般会在压测脚本里顺手把每次请求的耗时写进 jsonl，跑完直接读出来算分位数，比引第三方压测工具快。有一点要注意：压测要在真实的网络环境下做，本地连公司内网和连公网的 RTT 差几十毫秒，这个差异放大到一万张上就是几分钟。

5. 避坑与排查：五类最常见的翻车现场

下面这几条都是我在实际对接里真踩过的，每条按「现象 → 原因 → 解决」写。看到类似症状可以直接对号入座，不用从头猜。

5.1 401 与 403：密钥没错也可能被拒

现象：本地跑得好好的脚本，换到服务器容器里就返回 401，日志里 key 看起来一模一样。 原因：容器注入的环境变量带了不可见字符（复制粘贴时混进的换行或引号），或者.env文件在 Windows 下保存成了带 BOM 的编码，读出来开头多了几个字节。 解决：拿到 key 先打印长度和前四位，和预期对一下；统一用API_KEY.strip().strip('"').strip("'")清洗。403 则是另一回事，通常是这个 key 没有该模型的权限或者余额不足，别在代码里找问题，去控制台看配额。

5.2 请求头打架：files 和 application/json 不能共存

现象：图像分类返回 415 Unsupported Media Type，或者提示请求体解析失败。 原因：示例代码里在 headers 写了Content-Type: application/json，同时又传了files。requests 用files时会自己生成带 boundary 的 multipart 头，你手写的那一行把它覆盖掉了，服务端按 JSON 解析二进制，必然失败。 解决：用files就把 Content-Type 从 headers 里删掉，只保留 Authorization。同理，用json=时也不要手写 Content-Type。这条区别看着小，却是新手翻车率最高的地方。

5.3 图像看着正常却 400 / 422：通道数、透明底、超大图

现象：文件夹里几百张图，大部分成功，个别一直失败，用看图软件打开完全看不出问题。 原因：三种可能——PNG 带 alpha 通道被转成 RGBA 四通道、CMYK 模式的印刷图、单边超过几千像素的大图超了服务端解码限制。 解决：预处理阶段统一convert("RGB")，尺寸用ImageOps.fit卡死在模型要求的大小，体积循环压到上限以内。做完这三步还失败，就把那张图单独存下来人工看一眼，别再批量重试。

5.4 重试把问题放大：文件流复用与非幂等

现象：加了重试之后失败率反而升高，返回的错误变成「图片内容为空」。 原因：files里传的是文件句柄，第一次请求把句柄读到末尾，重试时再读就是空的。 解决：请求前先read()成 bytes，用 bytes 进重试；同时确认重试只针对 429 和 5xx，400 系列一律不重试，因为参数错误重试一万次还是错。

5.5 200 但字段缺失：把 JSON 当字典裸取的下场

现象：状态码 200，resp.json()也能解析，但data["label"]抛 KeyError 或者返回 None，整批任务中断。 原因：直接用下标取嵌套字段，没有对status、data、label做逐层判断。服务端在降级、限流、模型切换时，返回结构可能变化。 解决：所有解析走一个统一函数（第 2.3 节那份），逐层判空并显式抛错，把「数据异常」和「代码 bug」在日志里分开。另外把失败样本原文存一份，方便事后核对接口文档有没有更新。

6. 上线前跑一遍小样本自检：30 张图 + 60 条文本够不够

我自己接一个新的分类接口，不会一上来就灌全量数据，而是固定跑一个小样本自检：30 张覆盖不同来源的图（手机竖拍、截图、深色底、透明 PNG、超大图各几张），加上 60 条长度和语种分布接近真实业务的文本。这个规模的好处是十几分钟能跑完，坏处是统计意义有限，所以它只用来验证「行为是否符合预期」，不用来评估准确率。

自检清单我习惯做成一张表，跑完逐项打勾，任何一项不过都不往下走。

最后一项是给未来自己留的后悔药：拿 5 条已知答案的样本，把 label 和 confidence 打印出来对一眼。有一次我跑完 3000 张图才发现阈值设成了 0.95，结果 90% 的样本都被判进人工复核队列，白等了两小时。从那以后我每次接新接口，都强制先跑这 30+60 的小样本，确认阈值分布合理再放量。希望帮到你。

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