DeepSeek API图像与文本分类调用实战:鉴权、上传与排错

发布时间:2026/10/25 16:29:51
DeepSeek API图像与文本分类调用实战:鉴权、上传与排错

简介:这份面向具备Python基础的开发者与AI集成团队的实践指南,聚焦通过DeepSeek API完成图像分类和文本分类的接入流程。压缩包内仅有1个docx文档,大小约17KB,篇幅精炼,便于快速查阅与随项目归档。文档围绕requests发送HTTP请求、Pillow处理图像、API密钥鉴权、POST请求头与请求体构造展开,并给出图像上传与文本分类两套代码示例;同时说明成功响应中标签与置信度的解析方式,以及无效密钥、负载格式异常等常见错误的排查方向。已有2676人学习,适合希望快速验证云端机器学习服务、搭建分类应用原型的中级开发者参考。其价值在于将接口准备、请求发送、结果处理与图像尺寸预处理串成完整链路,帮助读者理解可复用的调用骨架,并据官方文档替换示例地址与参数,减少接入初期的试错成本。

1. 从一次 401 说起:这份调用指南真正解决的是链路问题

第一次把 DeepSeek 的图像分类接口接进项目,报错不是模型分错,而是 401——密钥被拼进了 URL 而不是请求头。这份《DeepSeek API 调用指南》解决的正是这类问题:它不讲模型结构,只讲怎么用 requests 发出一个 HTTP 请求,把一张图或一段文字 POST 给云端机器学习服务,再拿回 label 和 confidence 两个字段。内容覆盖密钥获取、Pillow 装环境、图像上传、文本分类复用同一套骨架,以及状态码排查。适合已经有 Python 基础、手里有个原型要接分类能力的开发者;如果你的场景是每天几千张图要打标,或者一批文本要自动归类,跟着走一遍能跑通最小闭环。它也明确写了端点只作示例,真实地址要回查官方文档,这点比很多抄来抄去的教程靠谱。

2. 请求链路拆开看:Bearer 鉴权、multipart 上传与 224×224 预处理

把一次分类调用拆开,其实只有四段:拼请求头、准备 body、发出去、解析回来。四段里出错概率最高的是第二段,因为图像和文本的 body 形态完全不同,一个是二进制表单,一个是 JSON。选错形态,服务端要么返回 415,要么解析出一堆乱码。这一章先把链路上每个零件的形状说清楚,后面写代码才不会靠试。

2.1 密钥与请求头:Bearer 的三种写法与唯一推荐

API 密钥在原料里被描述成「注册登录后拿到的一串字符串」,它在 HTTP 层的角色就是身份凭证。常见做法是放进Authorization头,格式为Bearer <key>,注意 Bearer 和 key 之间有一个空格,少一个空格就是 401。有些人图省事把 key 塞进 query 参数,浏览器地址栏、Nginx access log、代理转发记录里都会留下明文,等于把钥匙贴在门上,不建议。

更稳的习惯是从环境变量读密钥,代码里不出现字面量。下面这段用来做连通性自检:发一个轻量请求,能拿到 200 就说明密钥、网络、DNS 三段都通,再去调分类接口就不会把环境问题和业务问题混在一起排查。

import os import requests # 密钥从环境变量读取,不写死在代码和仓库里 API_KEY = os.environ["DEEPSEEK_API_KEY"] headers = { # Bearer 后面必须有一个空格,这是最常见的 401 来源 "Authorization": f"Bearer {API_KEY}", # JSON 请求体时手写这一行;上传文件时千万别手写,见 2.2 "Content-Type": "application/json", } # 端点仅作示例,实际地址以官方文档为准;换成任意轻量接口即可做连通性自检 PROBE_URL = "https://api.deepseek.com/v1/models" try: resp = requests.get(PROBE_URL, headers=headers, timeout=(3.05, 10)) print(resp.status_code) # 只打印前 200 字符,避免日志被大响应撑爆 print(resp.text[:200]) except requests.RequestException as e: print("网络层就失败了,先查 DNS / 出口 / 超时设置:", e)

逻辑说明:先做一次 GET 自检,把「认证失败」和「分类失败」两类问题隔离开。参数说明:timeout=(3.05, 10)是连接和读取两段超时,连接段给 3 秒左右足够覆盖绝大多数机房到公网的首包时间,读取段按接口响应速度给;单写一个数字的话,requests 会把它同时当作两段超时,读大响应时容易误杀。headers里的Content-Type在 GET 上其实可以不带,保留是为了让这段代码和后面的 POST 保持同一份头字典。

还有一种写法是用requests的auth参数,那是 HTTP Basic 认证,用户名密码形式的,和 Bearer Token 不是一回事,套上去只会得到 401 或 403。这一点在原料里没展开,但它是新手最常混淆的地方。

2.2 图像为什么必须走 files 而不是 json

原料给的示例用files=上传图像,而不是把图像塞进 JSON,这个选择不是随意的。二进制数据进 JSON 必须先 base64 编码,体积会涨大约三分之一,服务端还要多做一次解码,两端都亏。业界常见三种上传方式,取舍如下:

方式体积开销服务端处理成本适用场景
multipart/form-data(files=)几乎无膨胀低,流式解析单图、小批量、原型验证
base64 塞进 JSON约 +33%需要解码,内存峰值高网关只允许 JSON 的受限环境
先传对象存储再传 URL无最低,可异步批量任务、大图、需要留档原图

选 multipart 之后有一个必须知道的细节:requests会在检测到files参数时自动生成Content-Type: multipart/form-data; boundary=----xxxx,这个 boundary 是分隔各字段的随机字符串。如果你在headers里手写了一个不带 boundary 的Content-Type: multipart/form-data,requests 就不会再补,服务端拿不到分隔符,直接把整个 body 当成一个字段解析,结果通常是 400 或解析出空文件。所以上传图像时,头字典里只留Authorization。

import requests def upload_once(image_bytes: bytes, api_key: str, url: str): # 三元组顺序是:字段名、文件名、MIME 类型,缺一不可 files = {"image": ("image.jpg", image_bytes, "image/jpeg")} # 附加的表单字段都只能是字符串,数字要自己 str() 转换 data = {"top_k": "3"} headers = {"Authorization": f"Bearer {api_key}"} # 不要加 Content-Type return requests.post( url, headers=headers, files=files, data=data, timeout=(3.05, 60), # 图像推理比文本慢,读取超时给宽一点 )

逻辑说明:files的 value 用三元组而不是裸 bytes,是为了让服务端拿到正确的文件名和 MIME;如果只传{"image": image_bytes},requests 会默认按application/octet-stream发,部分服务端会因此拒绝。参数说明:top_k这类字段在 multipart 里一律是字符串,传3会被 requests 编码成"3",但如果你自己用data={"top_k": 3}再手动拼 body,就可能踩到类型错误;读取超时给到 60 秒是因为图像接口往往要跑一次前向,冷启动时更慢。

2.3 Pillow 预处理:尺寸、色彩模式与字节流复用

原料里提到「模型可能要求 224×224」,这是个假设值,真实尺寸必须回查文档。但无论目标尺寸是多少,预处理要做的事是固定的四件:摆正方向、统一色彩模式、等比缩放加裁剪、控制编码体积。

方向这件事容易被忽略。手机拍的 JPEG 里带 EXIF Orientation 标记,Image.open读出来是原始像素方向,直接缩放再上传,模型看到的是躺倒的猫。ImageOps.exif_transpose会按标记旋转并清掉这个标记,是一行就能解决的血泪经验。色彩模式方面,PNG 可能带 alpha 通道,灰度图只有一个通道,CMYK 的 JPEG 来自印刷物料,这几种直接送进只接受三通道 RGB 的模型都会报错或得到诡异结果,convert("RGB")统一掉。

from PIL import Image, ImageOps import io def normalize_image(path: str, size=(224, 224), quality: int = 90) -> bytes: with Image.open(path) as im: im = ImageOps.exif_transpose(im) # 按 EXIF 摆正,避免"躺倒的猫" im = im.convert("RGB") # 去 alpha、统一三通道 # fit 是等比缩放后居中裁剪,resize 是硬拉伸,后者会让长图变形 im = ImageOps.fit(im, size, method=Image.BICUBIC) buf = io.BytesIO() im.save(buf, format="JPEG", quality=quality, optimize=True) return buf.getvalue() # 直接拿 bytes,不落盘

逻辑说明:整个函数在内存里完成,返回 bytes 直接喂给 2.2 的上传函数,省掉一次临时文件读写,批量跑的时候差别很明显。参数说明:size按文档改,不要照抄 224;quality=90是体积和画质的折中点,压到 75 以下在细粒度分类任务上会掉点;optimize=True让 Pillow 多花一点 CPU 换更小的文件,对上传耗时是正收益;method=Image.BICUBIC比默认的最近邻更适合缩小,边缘不会出现锯齿。

还有一个常被忽略的点:Image.open是惰性的,文件句柄会一直挂着,批量处理几千张时不关会撞到文件描述符上限。上面用with包住就是为了这个,getvalue()必须在with块内调用,出了块对象就关了。

3. 图像分类调用的完整实现:从读文件到解析 label 与 confidence

第二章把零件备齐了,这一章组装成能跑的东西。判断一段调用代码合不合格,不看它能不能出结果,看它在出错时能不能告诉你错在哪。所以下面的实现里,超时、状态码分支、响应结构校验三样都要有,缺一样在批量场景里都会变成黑匣子。

3.1 最小可运行脚本与逐行参数说明

原料给的主干是requests.post(url, headers=headers, files=files),加一个status_code == 200的分支。这个骨架是对的,但少了三样东西:超时、非 200 分支的细分、JSON 解析的保护。网关返回 HTML 错误页时,response.json()会抛JSONDecodeError,异常信息里看不到状态码,排查时非常难受。

import os import requests from requests import Response API_KEY = os.environ["DEEPSEEK_API_KEY"] # 端点仅作示例,请以官方文档为准 CLASSIFY_URL = "https://api.deepseek.com/v1/classify" def classify_image(image_bytes: bytes, top_k: int = 3) -> dict: files = {"image": ("image.jpg", image_bytes, "image/jpeg")} data = {"top_k": str(top_k)} headers = {"Authorization": f"Bearer {API_KEY}"} try: resp: Response = requests.post( CLASSIFY_URL, headers=headers, files=files, data=data, timeout=(3.05, 60), ) except requests.Timeout: return {"ok": False, "code": "TIMEOUT", "msg": "请求超时,考虑降分辨率或加重试"} except requests.RequestException as e: return {"ok": False, "code": "NETWORK", "msg": str(e)} if resp.status_code != 200: # 保留状态码和响应体前 500 字符,足够定位大多数问题 return { "ok": False, "code": resp.status_code, "msg": resp.text[:500], } try: body = resp.json() except ValueError: return {"ok": False, "code": "BAD_JSON", "msg": resp.text[:500]} # 原料里的响应形如 {"status":"success","data":{"label":...,"confidence":...}} if body.get("status") != "success": return {"ok": False, "code": "API_ERROR", "msg": body.get("message", "")} data_obj = body.get("data") or {} return { "ok": True, "label": data_obj.get("label"), "confidence": float(data_obj.get("confidence") or 0.0), "raw": body, }

逻辑说明:返回统一成{"ok": bool}的字典,调用方只需要判断ok,不用在每个业务点重复写 try/except;失败分支带上code,日志里聚合一下就能看出是密钥问题还是超时问题占多数。参数说明:top_k转成字符串是因为 multipart 表单字段只接受字符串;超时元组里读取段给 60 秒,图像推理冷启动时可能十几秒,给太短会误判成服务不可用;float(... or 0.0)是防御式写法,万一服务端返回字符串"0.98"也能转换,返回 null 时不会抛TypeError。

提示:原料里的响应字段status、data.label、data.confidence是示例结构,接入前先用一次真实请求确认字段路径,再把这几个 key 抽成模块级常量,别散落在业务代码里。

3.2 响应结构解析与置信度阈值的业务用法

confidence这个字段很容易被误用。它是模型对本次预测的打分,不是「这张图有 98% 概率是猫」这种校准过的概率,跨类别、跨批次直接比大小并不严谨。但它作为分流信号是够用的:高分区自动落库,低分区转人工复核,中间灰区打标记后置处理。

置信度区间处理策略说明
≥ 0.90自动入库抽检 1% 做质量监控
0.60 ~ 0.90入待复核队列按天批量人工过一遍
< 0.60转人工或丢弃重采往往是模糊图、多主体图或域外样本
def route(result: dict, high: float = 0.90, low: float = 0.60) -> str: if not result.get("ok"): return "manual" # 调用失败一律进人工,不要静默丢数据 conf = result["confidence"] if conf >= high: return "auto" if conf >= low: return "review" return "manual"

逻辑说明:把阈值写成函数参数而不是硬编码,是因为不同类目的分布差别很大,做一次小样本标注就能把high和low调到位。参数说明:high调高会减少误判但增加人工量,通常先从 0.90 起步;low调低会让灰区变窄,适合人力紧张的团队。这三个档位建议先在历史数据上跑一遍分布,看灰区占比再定,拍脑袋定阈值是后面返工的主要原因。

3.3 批量与并发:Session 复用、超时与重试

单张调用跑通之后,下一步一定是批量。批量场景里两个改动收益最大:复用连接、加受控重试。

requests.post每次都是一次全新的 TCP 加 TLS 握手,批量几百张时握手开销能占到总耗时的一半。用Session复用连接池,配合HTTPAdapter设置池大小,是标准做法。

重试这块有个真实的坑:urllib3的Retry默认allowed_methods不包含 POST,所以你不显式配置的话,POST 请求根本不会重试,代码看起来加了重试其实没生效。反过来,如果你把 POST 加进重试列表,读超时后重试又可能造成重复计费或重复入库。折中做法是只对连接类错误和 429、5xx 重试,读超时不重试。

import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(pool_size: int = 16) -> requests.Session: session = requests.Session() retry = Retry( total=3, connect=3, # 只重试连接阶段 read=0, # 读超时不重试,避免重复计费 status=3, # 429 / 5xx 重试 status_forcelist=[429, 500, 502, 503, 504], allowed_methods=frozenset(["POST"]), # 默认不含 POST,必须显式加 backoff_factor=0.5, # 退避 0.5s、1s、2s raise_on_status=False, ) adapter = HTTPAdapter(max_retries=retry, pool_connections=pool_size, pool_maxsize=pool_size) session.mount("https://", adapter) return session

逻辑说明:connect和read分开配置是关键,连接失败重试是安全的,读超时重试则可能让同一张图被计费两次。参数说明:pool_maxsize要和你开的并发线程数匹配,线程数大于池大小时多出来的线程会等待连接,表现为「并发上不去」;backoff_factor=0.5产生指数退避,对 429 限流场景很有效;raise_on_status=False让重试耗尽后把原始响应返回给你,而不是抛异常,方便统一处理。

并发执行用ThreadPoolExecutor就够,这类任务是 IO 密集的,协程或线程都行,线程写起来简单。并发数别开太大,先看账号的限流配额,从 4 到 8 起步,撞到 429 就往下调。

4. 文本分类与接口复用:同一套骨架换 payload

原料里给的文本分类示例,和图像分类的差别只有两处:URL 不同、body 从 multipart 换成 JSON。这说明两个接口共享同一套 HTTP 骨架,把它抽出来比复制两份代码划算得多。抽的时候注意 JSON 和 multipart 在请求头上的差异,这是唯一不能共用的地方。

4.1 文本接口的 JSON 载荷与编码细节

requests.post(url, json=data)和requests.post(url, data=json.dumps(data))看起来一样,其实差两点:前者自动设置Content-Type: application/json,后者不会,你需要自己在头里写;前者遇到不可序列化的对象会直接抛TypeError,后者要到dumps阶段才报。统一用json=参数更省心。

import requests def classify_text(text: str, url: str, api_key: str, top_k: int = 3) -> dict: # 超长文本先截断,很多接口对长度有硬限制,超了直接 400 if len(text) > 4000: text = text[:4000] payload = {"text": text, "top_k": top_k} headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", # 用 json= 时 requests 也会自动加 } resp = requests.post(url, headers=headers, json=payload, timeout=(3.05, 15)) resp.raise_for_status() return resp.json()

逻辑说明:json=payload让 requests 用json.dumps序列化并自动设头,中文会被转成\uXXXX转义,这是合法的 JSON,服务端能正确还原,不需要加ensure_ascii=False(那个参数只在你自己dumps时有用)。参数说明:文本截断长度按接口文档调整,不要照抄 4000;raise_for_status()在原型阶段很方便,会让异常带上状态码,但生产代码里建议像第 3 章那样手动分支,因为你想拿到响应体做诊断。

还有一点:如果文本来自网页抓取,先做一次清洗,去掉零宽字符、全角空格、HTML 实体。这些字符在 JSON 里合法,但会让分类结果偏移,属于典型的「模型没问题、数据有脏东西」。

4.2 把图像和文本两类调用抽象成统一客户端

抽象的目标不是少写几行,而是让重试、日志、错误码映射只写一遍。下面的类把公共部分收在_post里,两个公开方法只负责准备 body。

import logging import requests log = logging.getLogger(__name__) class ClassifyClient: def __init__(self, api_key: str, base_url: str, session: requests.Session): self.api_key = api_key self.base_url = base_url.rstrip("/") # session 从外部注入,方便复用连接池和重试策略 self.session = session def _headers(self, with_json: bool = False) -> dict: headers = {"Authorization": f"Bearer {self.api_key}"} if with_json: headers["Content-Type"] = "application/json" return headers def _post(self, path: str, **kwargs) -> dict: url = f"{self.base_url}/{path.lstrip('/')}" resp = self.session.post(url, **kwargs) # 统一把非 200 转成带状态码的字典,调用方不用各自判断 if resp.status_code != 200: log.warning("classify failed status=%s body=%s", resp.status_code, resp.text[:300]) return {"ok": False, "code": resp.status_code, "msg": resp.text[:300]} try: return {"ok": True, **resp.json()} except ValueError: return {"ok": False, "code": "BAD_JSON", "msg": resp.text[:300]} def image(self, image_bytes: bytes, top_k: int = 3) -> dict: return self._post( "/v1/classify", # 示例路径,以文档为准 headers=self._headers(with_json=False), files={"image": ("image.jpg", image_bytes, "image/jpeg")}, data={"top_k": str(top_k)}, timeout=(3.05, 60), ) def text(self, text: str, top_k: int = 3) -> dict: return self._post( "/v1/text_classify", # 示例路径,以文档为准 headers=self._headers(with_json=True), json={"text": text[:4000], "top_k": top_k}, timeout=(3.05, 15), )

逻辑说明:session从构造函数注入,而不是在类里 new 一个,这样测试时可以换成 mock,生产时可以复用第 3 章配好的连接池和重试策略。参数说明:_headers的with_json开关是唯一区分两类请求的地方,multipart 请求不能带Content-Type,这是 2.2 讲过的坑,抽象的时候特别容易统一成「都带上」,然后图像接口集体 415。base_url用rstrip("/")处理,避免路径拼出双斜杠,有些网关对//v1会 404。

4.3 成本与延迟的取舍:本地预处理还是服务端预处理

图像走的链路里,预处理放哪边是个真实的决策。放本地省带宽、省服务端算力,但吃客户端 CPU;放服务端省客户端逻辑,但上传的是原图,体积可能大十倍。

策略上传体积客户端 CPU结果一致性适用
本地缩放到目标尺寸小中高,可控批量任务、移动端上行
本地只压质量不缩放中低中需要服务端统一裁切的场景
原图直传大无取决于服务端实现需要保留细节的细粒度分类

我一般会先在本地压到长边 1024、质量 85 上传,再拿一批样本对比「原图直传」和「本地压缩」的分类一致性。如果一致性差异在可接受范围内,就固定用压缩版本,单张延迟能降下来一大截,尤其是弱网环境。这个对比实验花不了多少时间,但能省掉后面反复怀疑「是不是模型不行」的折腾。

5. 避坑与排查:401、415、413、超时重放这四类翻车现场

前面几章都是顺着成功路径写的,这一章反过来,按我踩过的顺序排。每条都按「现象、原因、解决」写,遇到类似的直接对号入座。

5.1 401 Unauthorized:密钥对了但就是过不去

现象:密钥确认没复制错,用 curl 试也失败,响应体里写着 invalid api key 或 missing authorization。

原因通常有三个。一是Bearer和密钥之间的空格被编辑器吃掉,或者中间混进了全角空格;二是密钥从环境变量读出来时带了尾部的换行,echo复制的时候最容易带上;三是用了另一个产品的密钥,格式看着像但不是同一个认证域。

解决:先打印repr(API_KEY)看首尾字符,把不可见字符暴露出来;如果尾部是\n或者\r,在读取时.strip()。再用 curl 独立验证一次,排除代码层的干扰:

# 用环境变量传密钥,避免密钥进入 shell history curl -s -o /dev/null -w "%{http_code}\n" \ -H "Authorization: Bearer ${DEEPSEEK_API_KEY}" \ "https://api.deepseek.com/v1/models"

返回 200 说明密钥和网络都没问题,回到代码里查头字典;返回 401 就换密钥或确认账号权限。这行命令里的-o /dev/null丢掉响应体,-w "%{http_code}"只打状态码,输出干净。

5.2 415 / 400:multipart 的 boundary 被自己弄丢了

现象:图像上传返回 415 Unsupported Media Type,或者 400 说缺少 file 字段,但字段名明明写了image。

原因:头字典里手写了"Content-Type": "multipart/form-data"。requests 看到你已经设了这个头,就不再补 boundary,服务端拿不到分隔符,把整个 body 解析成一个空字段。这个坑在把文本和图像调用合并成同一个客户端时特别容易犯,因为你会下意识地给两类请求都加上Content-Type。

解决:multipart 请求的头里只留Authorization,让 requests 自己生成带 boundary 的完整头。排查时打印resp.request.headers,看Content-Type里有没有boundary=这个参数,没有就是这个问题。

5.3 413 Payload Too Large:图像没压,或者压了但格式不对

现象:小图能过,大图报 413;或者本地明明缩放过,上传后还是 413。

原因:一是缩放用的resize保留了原始编码格式,PNG 截图缩到 224 之后体积仍然不小;二是只改了尺寸没改质量,JPEG 以 quality=100 保存,一张 224×224 也能到几百 KB;三是服务端限制的是字节数不是像素数,你盯着尺寸看没用。

解决:统一走 2.3 的normalize_image,输出固定为 JPEG、quality 可控、带optimize=True,并在发送前加一道体积闸门:

def guard(image_bytes: bytes, max_bytes: int = 1_500_000) -> bytes: # 超过阈值就降质量重压,最多降三档,避免无限循环 for q in (85, 70, 55): if len(image_bytes) <= max_bytes: break from PIL import Image import io buf = io.BytesIO() Image.open(io.BytesIO(image_bytes)).save( buf, format="JPEG", quality=q, optimize=True) image_bytes = buf.getvalue() return image_bytes

参数说明:max_bytes按接口文档给,1.5 MB 只是个保守起点;三档质量是经验值,再往下压细粒度分类就会掉点。

5.4 超时后重试导致重复计费

现象:账单上的调用次数比实际图片数多,或者结果表里出现重复记录,且这些记录的耗时都集中在某个时间窗口。

原因:读超时被算进了重试条件。服务端其实已经处理完了,只是响应在路上慢了一点,客户端等不及重发,于是同一张图被处理两次。原料里的示例没有涉及重试,但一旦你按 3.3 加了HTTPAdapter,这个问题就会出现。

解决:Retry里把read设为 0,只对连接错误和 429、5xx 重试;如果业务要求严格幂等,就在请求里带一个自己生成的request_id,落库时用它做唯一键,重复的写入直接忽略。这个字段在服务端不支持透传时也没关系,客户端侧做去重同样有效。

5.5 排查顺序:从外到内,别一上来就怀疑模型

遇到分类结果不对,先别改代码。按这个顺序走一遍,八成问题能在前三步定位:先用 curl 验证密钥和端点是否可达;再用固定的一张样本图跑一次,看响应体的原始 JSON 长什么样;然后比对预处理后的图和原图,确认没被裁掉主体;最后才看 label 和 confidence 的分布。把顺序固定下来之后,排查时间能从半小时压到几分钟,这个习惯比任何技巧都值钱。

6. 进阶:把分类调用做成可观测的小服务

跑通脚本和能上线之间差的是可观测性。分类调用是个外部依赖,失败是常态,没有埋点的话,你只能看到「今天结果少了」这种结论,看不到是哪一类错误在涨。

我一般会在客户端外面再包一层,记录四个字段:请求 ID、端点、状态码、耗时。加上 label 和 confidence,就能回答绝大多数运营问题,比如「昨天的低置信度占比为什么翻倍」。表结构大概是这样:

字段类型用途
request_idstring客户端生成的 UUID,重试时复用,用于去重
endpointstringimage 或 text,区分两条链路
status_codeintHTTP 状态码,0 表示网络层失败
latency_msint端到端耗时,含重试
labelstring分类结果,失败时为空
confidencefloat置信度,用于阈值分布监控

埋点代码不用复杂,一个装饰器就够:

import time import uuid import logging log = logging.getLogger("classify") def traced(endpoint: str): def deco(fn): def wrapper(*args, **kwargs): rid = kwargs.pop("request_id", None) or str(uuid.uuid4()) start = time.perf_counter() status = 0 try: result = fn(*args, **kwargs) status = result.get("code", 200) if result.get("ok") else result.get("code", -1) return result finally: cost = int((time.perf_counter() - start) * 1000) # 一行 JSON 日志,方便直接被采集器解析 log.info('{"rid":"%s","ep":"%s","status":%s,"ms":%d}', rid, endpoint, status, cost) return wrapper return deco

逻辑说明:request_id从外部传入时复用,重试链路里就能对齐同一次业务请求;finally保证异常路径也打点,否则失败样本全丢,监控就成了只报喜不报忧。参数说明:耗时用perf_counter而不是time.time,避免系统时间调整带来的负数;状态码在业务失败时取自定义 code,网络层失败统一记 0,方便聚合。

这套东西的另一个用法是回归验证。接口或模型更新后,拿一批固定样本跑一遍,比对 label 变化率和置信度均值偏移,比肉眼看几个案例靠谱得多。我吃过一次亏:换了预处理里的resize为ImageOps.fit,忘了看长图样本,结果一批竖版商品图被裁掉了底部文字,分类整体偏了一档,两天后才发现。从那以后我每次改动预处理参数,都强制走一遍固定样本回归,比对 label 和 confidence 的分布再发布。希望帮到你。

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