图像与文本分类API接入：请求封装、重试与批量工程化-北京尧图网络

简介面向具备Python基础的开发者与正在集成自动化AI分类能力的开发团队这份指南围绕DeepSeek提供的图像分类与文本分类API梳理从注册获取密钥、安装requests与Pillow到构造POST请求、设置HTTP头部、准备请求主体并解析分类标签与置信度的完整调用链路。文档结合代码实例说明图像上传、文本分类请求及响应处理并给出密钥失效、请求负载格式异常等常见问题的排查思路同时提示图像尺寸压缩、接口URL以官方文档为准等实践注意事项便于读者快速搭建分类应用原型。资源包共1个docx文件大小约17KB以文字说明与示例代码为主结构紧凑、便于随查随用。目前已有2676人学习适合需要理解API接入流程、对照示例完成图像与文本分类任务的中初级Python开发者参考。1. 一张图一段文本先把调用链路跑通把一个本地图片塞进 HTTP 请求、拿回一个 label 和 confidence这件事本身只要二十行 requests 代码真正让人卡住的是图像分类和文本分类这两条链路在请求体上的差别以及密钥、超时、重试这些没人愿意写进示例的东西。这份资源给的是一套可直接复现的调用骨架文本分类走 JSON图像分类走 multipartPillow 负责预处理requests 负责传输响应当成结构化数据来处理。它解决的是「云端机器学习服务怎么接进自己的系统」这个落地问题适合已经写过 Python、手里有一批图片或文本要打标、又不想自己训模型的开发者。下面按接口契约、环境复现、批量工程化、避坑排查四段推进每一步都能直接抄。2. 摸清接口契约图像与文本两条链路的差别很多人拿到示例代码第一反应是复制粘贴改个 URL然后发现图像那条链路一直报 415 或者 400文本那条却一次就通。差别不在密钥也不在网络而在请求体的封装方式一个是 multipart/form-data一个是 application/json这两种格式对 HTTP 头、对文件句柄、对重试策略的要求完全不一样。所以动手之前先把接口契约里最不容易变的那几个常量固定下来比先写业务逻辑划算得多。2.1 端点、方法与认证三个先钉死的常量无论后面封装多少层最终发出去的请求只依赖三样东西地址、方法、凭证。地址由 base_url 加路径拼成方法固定 POST凭证放在 Authorization 头里形式是Bearer key。把这三样抽到一个配置文件里好处是换环境、换账号、换模型版本时只改一处不用在几十个函数里搜字符串。我一般还会额外加一个 User-Agent不是接口要求而是出问题时能在服务端日志里认出是自己的客户端。常量示例值说明BASE_URLhttps://api.example.com/v1以控制台文档给的地址为准示例域名不可直接用IMAGE_ENDPOINT/classify图像分类路径示例值需对照文档TEXT_ENDPOINT/text_classify文本分类路径示例值需对照文档API_KEY环境变量注入形如sk-xxx的字符串只走环境变量TIMEOUT(5, 30)连接 5 秒、读取 30 秒两个值分开设这里有个容易被忽略的细节超时必须写成元组。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_connectionspool_size, pool_maxsizepool_size, max_retries0) s.mount(https://, adapter) s.headers.update({ Authorization: fBearer {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, filesfiles, # 不要手写 Content-Typerequests 会带 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, jsonpayload, # 自动序列化自动 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 负责 HTTPPillow 负责图像。用虚拟环境装别往全局环境里怼否则半年后你会在某个报错里发现是另一个项目的旧版本在捣乱。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_KEYsk-xxxxxxxxxxxxxxxx API_BASE_URLhttps://api.example.com/v1 # 启动前加载 export $(grep -v ^# .env | xargs) # 临时用生产环境交给配置中心grep -v ^#过滤注释行xargs把KEYVALUE形式拆成 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, methodImage.BICUBIC, centering(0.5, 0.5)) quality 88 buf io.BytesIO() im.save(buf, formatJPEG, qualityquality, optimizeTrue) while buf.tell() max_bytes and quality min_quality: quality - 8 buf io.BytesIO() im.save(buf, formatJPEG, qualityquality, optimizeTrue) buf.seek(0) # 指针必须归零否则发出去是空内容 return buf, fupload_q{quality}.jpgImageOps.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_attempts4, **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_attempts4是经验值第一次三次重试绝大多数抖动都能扛过去再多就是在浪费额度。退避用2 ** attempt但上限压到 8 秒因为批量任务的总时长对业务有感知不能无限等。random.random()这个抖动很关键——一批任务同时被限流时如果所有客户端都按同一个间隔重试下一波还是同时打过去雪崩。加上抖动后重试时间点被打散压力自然摊平。max_retries0是故意的。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, workers4, batch_log_every50): results, errors [], [] with ThreadPoolExecutor(max_workersworkers) 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()workers4是保守起点。先按 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, encodingutf-8) as f: f.write(json.dumps(record, ensure_asciiFalse) \n) def load_done(fp: Path) - set[str]: if not fp.exists(): return set() done set() with fp.open(encodingutf-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 耗时、错误率。这三个数决定并发数和重试预算也决定了要不要做结果缓存。指标怎么测观察什么单张平均耗时串行跑 30 张取均值图像明显高于文本分开统计P95 耗时同样 30 条排序取第 28 位远高于均值说明有大图拖尾错误率失败数 / 总数超过 2% 就该先查原因再加并发429 占比单独计数高说明并发过大或没做退避平均响应体积len(resp.content)用于估算流量和日志开销我一般会在压测脚本里顺手把每次请求的耗时写进 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 和 5xx400 系列一律不重试因为参数错误重试一万次还是错。5.5 200 但字段缺失把 JSON 当字典裸取的下场现象状态码 200resp.json()也能解析但data[label]抛 KeyError 或者返回 None整批任务中断。原因直接用下标取嵌套字段没有对status、data、label做逐层判断。服务端在降级、限流、模型切换时返回结构可能变化。解决所有解析走一个统一函数第 2.3 节那份逐层判空并显式抛错把「数据异常」和「代码 bug」在日志里分开。另外把失败样本原文存一份方便事后核对接口文档有没有更新。6. 上线前跑一遍小样本自检30 张图 60 条文本够不够我自己接一个新的分类接口不会一上来就灌全量数据而是固定跑一个小样本自检30 张覆盖不同来源的图手机竖拍、截图、深色底、透明 PNG、超大图各几张加上 60 条长度和语种分布接近真实业务的文本。这个规模的好处是十几分钟能跑完坏处是统计意义有限所以它只用来验证「行为是否符合预期」不用来评估准确率。自检清单我习惯做成一张表跑完逐项打勾任何一项不过都不往下走。检查项通过标准不过怎么办认证全部返回 200查密钥清洗和权限图像格式覆盖五种来源全部成功回到预处理补 convert / fit文本长度边界超长文本不报 4xx调整截断策略错误响应可读故意传坏 key 能拿到明确 message补 parse_result 分支重试行为人为断网后能自愈检查退避与 bytes 化结果落盘中断后重跑跳过已完成检查 load_done最后一项是给未来自己留的后悔药拿 5 条已知答案的样本把 label 和 confidence 打印出来对一眼。有一次我跑完 3000 张图才发现阈值设成了 0.95结果 90% 的样本都被判进人工复核队列白等了两小时。从那以后我每次接新接口都强制先跑这 3060 的小样本确认阈值分布合理再放量。希望帮到你。本文还有配套的精品资源点击获取

图像与文本分类API接入：请求封装、重试与批量工程化

相关推荐

视频异常检测技术演进与实战应用

制造集团数字化转型规划：151页PPT拆解与落地避坑指南

TinyImageNetTorch：微调预训练模型的轻量级压力测试集

更多资讯

基于YOLOv5的绝缘子缺陷检测实战：训练配置、评估指标与避坑指南

AI评测系统谁来评测？元评估与LLM裁判偏差治理实战

双11预售邮件别靠倒计时堆频次：先写清发送状态机

一枚参数叶片，给 AI 绘图一个稳定参照

散步清单可以记忆，不能替人下单

libexif 0.6.21 Windows 构建与部署实战指南

今日更新

防幻觉智能体架构：电源硬件设计中的大模型编排与工具化实践

OpenGL面试核心考点：渲染管线、矩阵变换与性能排查全解析

全国计算机二级C语言真题PDF高效刷题指南

本周精选

防幻觉智能体架构：电源硬件设计中的大模型编排与工具化实践

OpenGL面试核心考点：渲染管线、矩阵变换与性能排查全解析

全国计算机二级C语言真题PDF高效刷题指南

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？