Python 多线程批量检查 URL 状态码:ThreadPoolExecutor + Session
一个真实场景:data/*.json,每个 JSON 是数组,数组元素里有 imgHttpUrl。想批量检查这些 URL 是不是 403(图片失效),是的话把整个 JSON 文件删掉。文件几千个,单线程要跑一天,加多线程。 单线程版本先跑通 import json from pathlib import Path import requestsfor json_file in Path("data").glob("*.json"): try: data = json.loads(json_file.read_text(encoding="utf-8")) if not data: continue url = data[0].get("imgHttpUrl") if not url: continue r = requests.get(url, stream=True, timeout=10) if r.status_code == 403: print(f"删除 {json_file}") json_file.unlink() except Exception as e: print(json_file, e)要点:只检查数组第一个元素(假设整个文件的图片来自同一源,一个 403 全体 403) stream=True — 只拿响应头,body 不下载,快 用 pathlib.Path.unlink() 删文件,比 os.remove 干净加多线程 IO 密集,直接 ThreadPoolExecutor 就行,不用协程: import json from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed import requestsDATA_DIR = Path("data") MAX_WORKERS = 100session = requests.Session()def check_file(json_file: Path): try: data = json.loads(json_file.read_text(encoding="utf-8")) if not data: return None url = data[0].get("imgHttpUrl") if not url: return None r = session.get(url, stream=True, timeout=10, allow_redirects=True) if r.status_code == 403: json_file.unlink() return json_file except Exception as e: print(f"[错误] {json_file}: {e}") return Nonefiles = list(DATA_DIR.glob("*.json")) deleted = []with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex: futures = {ex.submit(check_file, f): f for f in files} for fut in as_completed(futures): result = fut.result() if result: deleted.append(result)print(f"完成,共删除 {len(deleted)} 个文件")Session 为什么关键 用 requests.Session() 而不是每次 requests.get():复用 TCP 连接 + TLS 握手,快 5~10 倍 连接池自动管理 全局默认 headers / cookies批量请求同一个域名,Session 是必配的。 HEAD 还是 GET? 只判状态码,不需要 body,理论上 HEAD 更快: r = session.head(url, timeout=10, allow_redirects=True)但:CDN / 图片服务器可能不支持 HEAD,返回 405 Method Not Allowed(常见于阿里 OSS 未开该权限、部分反爬策略) 有些服务器 HEAD 返回和 GET 不同的状态码保险起见先 HEAD、失败降级 GET: def check(url): try: r = session.head(url, timeout=10, allow_redirects=True) if r.status_code == 405: r = session.get(url, stream=True, timeout=10, allow_redirects=True) return r.status_code except requests.RequestException: return None线程数怎么调CPU 密集:os.cpu_count() 一般顶 IO 密集(网络请求):可以开几十甚至几百 目标服务器扛得住多少:看它——同源打 100 并发容易被 WAF ban,不同源随便开试起来:从 20 开始,看服务器响应正常、CPU 也没打满、就慢慢加。100 是 Python requests 常见上限。 更快:httpx + asyncio 千 URL 级别 ThreadPoolExecutor 够用。上万级建议改 asyncio: import asyncio, httpx, json from pathlib import Pathasync def check(client, json_file): try: data = json.loads(json_file.read_text(encoding="utf-8")) if not data: return url = data[0].get("imgHttpUrl") if not url: return r = await client.head(url, follow_redirects=True, timeout=10) if r.status_code == 403: json_file.unlink() print(f"删除 {json_file}") except Exception as e: print(json_file, e)async def main(): files = list(Path("data").glob("*.json")) limits = httpx.Limits(max_connections=200) async with httpx.AsyncClient(limits=limits) as client: await asyncio.gather(*(check(client, f) for f in files))asyncio.run(main())asyncio + httpx 可以轻松跑到 1000+ 并发,机器压力更小。 一句话总结 IO 密集批量请求 = ThreadPoolExecutor + Session。开 50-100 线程,先 HEAD 再降级 GET。上万量级换 asyncio + httpx。
SAM3 标注交通牌主体:正负点提示策略
用 SAM3 标注交通标志时,默认分割经常会把灯杆、支架、阴影一起圈进来。关键在于提示点的位置和正负样本的配合。 单点提示(最简方案) 把提示点打在牌面中央,SAM 通常只会分割牌面:牌型 提示点位置圆形限速牌 圆心三角警示牌 三角形中间矩形指示牌 牌面中央points = [[x_center, y_center]] labels = [1] # 1 = foreground正点 + 负点过滤杆子 如果单点还是把杆子带入,加负样本点排除: points = [ [牌面中心x, 牌面中心y], # 正样本 [杆子中间x, 杆子中间y], # 负样本 ] labels = [1, 0] # 1=前景, 0=背景实际调用: masks, scores, logits = predictor.predict( point_coords=np.array(points), point_labels=np.array(labels), multimask_output=True, )# 取得分最高的 mask best_mask = masks[np.argmax(scores)]检测框缩小(配合 Grounding DINO) Grounding DINO 给出的 bbox 通常包含了灯杆,传给 SAM 时手动收缩: def shrink_box(box, ratio=0.85): x1, y1, x2, y2 = box cx, cy = (x1 + x2) / 2, (y1 + y2) / 2 w, h = (x2 - x1) * ratio, (y2 - y1) * ratio return [cx - w/2, cy - h/2, cx + w/2, cy + h/2]tight_box = shrink_box(dino_box, ratio=0.8)masks, scores, _ = predictor.predict( point_coords=None, point_labels=None, box=np.array(tight_box), multimask_output=False, )缩小 15~20% 通常能去掉杆子区域。 mask 面积过滤 有时 SAM 返回多个候选 mask,按面积过滤掉太大或太小的: img_area = image.shape[0] * image.shape[1]valid_masks = [] for mask in masks: area = mask.sum() ratio = area / img_area if 0.01 < ratio < 0.4: # 占图片面积 1%~40% valid_masks.append(mask)交通牌通常不超过画面的 40%,过滤掉背景误判。 完整管线(Grounding DINO + SAM3) from groundingdino.util.inference import load_model, predict from segment_anything import SamPredictor# Step 1: DINO 定位交通牌 boxes, logits, phrases = predict( model=gd_model, image=image, caption="traffic sign", box_threshold=0.35, text_threshold=0.25, )# Step 2: 收缩框 tight_boxes = [shrink_box(b) for b in boxes]# Step 3: SAM 精确分割 predictor.set_image(image) for box in tight_boxes: masks, scores, _ = predictor.predict( box=np.array(box), multimask_output=False, ) # 保存 mask...调参建议 杆子还是出现 → 加负样本点在杆子上,或再缩小 bbox 5% 牌面被切掉 → bbox 缩小太多,调回 ratio=0.9 误检到路面标线 → 提高 DINO 的 box_threshold(0.35 → 0.45) 多块牌聚在一起 → multimask_output=True 后手动选面积最合适的 mask
torch.autocast 混合精度加速:bfloat16 vs float16 怎么选
torch.autocast 是 PyTorch 混合精度(AMP)的核心 API——把部分算子从 FP32 切到 BF16 / FP16,推理速度和训练吞吐都能翻倍,显存也能降 30-50%。 最简用法 推理: import torchmodel = model.cuda().eval()with torch.no_grad(), torch.autocast("cuda", dtype=torch.bfloat16): output = model(input.cuda())训练: model = model.cuda().train() optimizer = torch.optim.AdamW(model.parameters())for x, y in loader: x, y = x.cuda(), y.cuda() optimizer.zero_grad() with torch.autocast("cuda", dtype=torch.bfloat16): pred = model(x) loss = loss_fn(pred, y) loss.backward() optimizer.step()关键:autocast 只包在 forward + loss 计算里,backward 和 optimizer.step() 在外面。 autocast 到底改了什么 进入 autocast 上下文后,PyTorch 会自动选择每个算子的精度:算子类别 会被降精度Linear / Matmul ✅ 用 BF16/FP16Conv2d / Conv3d ✅ 用 BF16/FP16GEMM 类操作 ✅ 用 BF16/FP16Softmax ❌ 保持 FP32(防溢出)LayerNorm ❌ 保持 FP32Loss ❌ 保持 FP32Reduction ❌ 保持 FP32PyTorch 有个内置白名单/黑名单,你不用手动 .half()、不用改模型结构。 bfloat16 vs float16 BF16(bfloat16):✅ 动态范围和 FP32 一样(8 位 exponent),不会像 FP16 那样容易溢出/下溢 ✅ 不需要 GradScaler(训练时不用缩放梯度) ✅ 稳定性接近 FP32,一般不影响 loss ⚠️ 需要 Ampere 及以上(A100、RTX 30 系及以后)——旧卡不支持 ⚠️ 比 FP16 精度低(7 位 mantissa)FP16(float16):✅ 所有现代 GPU 都支持(Volta 起) ✅ 稍快一点(在某些卡上) ⚠️ 容易 NaN(动态范围窄) ⚠️ 必须配 torch.cuda.amp.GradScaler() ⚠️ 大模型容易训崩该选谁:硬件 训练 推理A100 / H100 BF16 BF16RTX 30 / 40 系 BF16 BF16RTX 20 系 / V100 FP16 + GradScaler FP16GTX 10 系及以下 不支持,用 FP32 不支持没有理由用 FP16 训练时选 BF16——除非你的卡不支持。 训练带 GradScaler(FP16 必须) scaler = torch.cuda.amp.GradScaler()for x, y in loader: optimizer.zero_grad() with torch.autocast("cuda", dtype=torch.float16): pred = model(x) loss = loss_fn(pred, y) scaler.scale(loss).backward() scaler.step(optimizer) scaler.update()GradScaler 把 loss 放大再反传,避免小梯度被 FP16 归零。BF16 用了反而干扰训练,别加。 常见坑 1. 输入没上 GPU with torch.autocast("cuda", ...): output = model(input) # input 在 CPU,报错务必: output = model(input.cuda())2. 和 .half() 混用 别这样: model.half() # 全模型转 FP16 with torch.autocast("cuda", dtype=torch.bfloat16): output = model(input)冲突。要么全模型 .half() 不用 autocast,要么用 autocast 不 .half()。混合精度靠 autocast 就够了。 3. CPU 上 autocast 不等价 with torch.autocast("cpu", dtype=torch.bfloat16):CPU 上 autocast 支持有限,别期望和 GPU 一样效果。CPU 推理想加速走 torch.compile 或 ONNX Runtime。 4. 自定义 op 不支持 BF16 第三方 op / native extension 如果没写 BF16 kernel,autocast 会 fallback 到 FP32——不是报错,只是没加速效果。 5. 推理不用 GradScaler # 推理 with torch.no_grad(), torch.autocast("cuda", dtype=torch.bfloat16): output = model(x)torch.no_grad() 和 autocast 一起用最省显存。 加速效果参考 Transformer 类模型(LLaMA 7B、Qwen 8B 之类)BF16 vs FP32:推理速度:1.8 – 2.3x 显存占用:约 50% 精度损失:几乎无(bf16 动态范围和 fp32 一致)CNN 训练(YOLO 之类):训练吞吐:1.3 – 1.7x 显存节省:30 – 45% mAP:几乎无差一句话总结 Ampere 及以上 GPU 就用 torch.autocast("cuda", dtype=torch.bfloat16),无需 GradScaler、稳定性接近 FP32。老卡 / V100 才考虑 FP16 + GradScaler。别和 model.half() 混用。
Python 依赖版本冲突:requires-python 约束与 numpy/torch 的 Python 版本限制
报错解读 × No solution found when resolving dependencies for split (markers: python_full_version == '3.8.*'): Because numpy>=1.26.0 depends on Python>=3.9,<3.13, we can conclude that numpy>=1.26.0 cannot be used on Python 3.8.这不是 numpy 装不上,而是 requires-python 声明的范围与依赖的 Python 要求相矛盾:项目声明支持 Python>=3.8 numpy>=1.26 要求 Python>=3.9,<3.13 求解器为 Python 3.8 找不到满足条件的 numpy 版本,报 No solution found修复方案 收窄 requires-python 范围(推荐): # pyproject.toml [project] requires-python = ">=3.9,<3.13" # 或更保险 requires-python = ">=3.10,<3.13"这是现代 ML/CV 依赖(numpy>=1.26、torch>=2、OpenCV)能稳定运行的范围。 如果必须保留 Python 3.8(不推荐): numpy==1.24.4 # 最后支持 Python 3.8 的 numpy 版本但 torch>=2、transformers、sam 等几乎都不再支持 3.8,维护成本极高。 Python 版本与主要 ML 库支持范围库 最低 Python 最高 Pythonnumpy >= 1.26 3.9 3.12torch >= 2.0 3.8 3.12torch >= 2.3 3.9 3.12transformers >= 4.40 3.8 3.12统一建议:Python 3.10 或 3.11,兼容范围最广,且有长期支持。 triton 只支持 Linux + CUDA × No solution found when resolving dependencies: Because only triton==0.4.1...2.1.0 are available and triton has no wheels for Windowstriton 是 GPU kernel 编译库,官方只发布 Linux + CUDA 的 wheel,在 Windows 上无法通过 pip 安装。 处理方式: # 将 triton 设为可选依赖,或仅在 Linux 下安装 [project.optional-dependencies] gpu = ["triton>=2.0; sys_platform == 'linux'"]或者在 Windows 开发环境使用 WSL2 / Docker。 uv 常用排查命令 # 查看当前 Python 版本 uv python list# 指定 Python 版本安装 uv pip install numpy --python 3.11# 查看依赖树 uv pip show numpy# 锁定依赖版本 uv pip compile pyproject.toml -o requirements.txt
uv 依赖冲突:requires-python 太宽 + Windows 装 triton 的双坑
上一个 CV 项目,uv sync 直接崩: × No solution found when resolving dependencies for split (markers: python_full_version == '3.8.*'): ↳ Because the requested Python version (>=3.8) does not satisfy Python>=3.9.0 and numpy>=1.26.0 depends on Python>=3.9,<3.13, we can conclude that numpy>=1.26.0 cannot be used.看起来是"numpy 装不上",其实是 Python 版本范围和依赖范围打架。 症状识别:requires-python 太宽 pyproject.toml 里如果写了: requires-python = ">=3.8"uv 会尝试为 3.8 / 3.9 / ... / 3.12 全部找一套解。numpy ≥ 1.26 要求 Python ≥ 3.9,Python 3.8 直接被排除——solver 认为"你说要支持 3.8,但 3.8 没解",结论:无解。 解法:收紧 requires-python 改成现代 AI 项目实际支持的范围: requires-python = ">=3.10,<3.13"理由:numpy ≥ 1.26 要求 Python ≥ 3.9 torch 2.x + Windows 的 wheel 覆盖到 3.12 3.13 太新,很多 native wheel 还没出之后 uv sync 通常就过了。 另一坑:Windows 上 triton 无解 同一个项目继续装: uv pip install triton× No solution found: triton<=2.1.0 has no wheels with a matching Python ABI tag (e.g., cp312) triton>=2.2.0 has no wheels with a matching platform tag (win_amd64)不是版本冲突,是 triton 官方根本不出 Windows wheel。Linux 有 manylinux_x86_64,macOS 部分版本有,Windows 是空白。 Windows 上处理 triton 的三条路 方案 A:上 WSL2(推荐) 装个 Ubuntu,在里面: python3.11 -m venv venv source venv/bin/activate pip install triton torchCUDA、torch、triton 生态齐活。 方案 B:直接放弃 triton 只是跑 inference 或 demo,很多模型不强依赖 triton: uv remove triton代码里如果有 import triton,包一层 try/except 或者按平台条件装: [project.optional-dependencies] gpu = ["triton; sys_platform == 'linux'"]方案 C:走 Linux 服务器 有云 GPU 就直接扔上去,从头就没 Windows 这个问题。 uv 常用的调试思路 # 看清依赖树里谁在拉某个包 uv tree | grep numpy# 只解析不装,快速试冲突 uv lock# 指定用某个 Python uv sync --python 3.11# 显式约束某个包 uv add "numpy>=1.26,<2"一句话总结 uv 报 no solution,先看两点:requires-python 范围是不是包了依赖不支持的旧 Python、目标平台是不是根本没轮子。Windows 上装 triton 是无解,别死磕。
