在 python 中调用在线大模型 api

你已经在 Python 里能跑通一次大模型请求了，但一放进真实项目就开始出问题：偶发超时、被限流、某个模型挂了整条链路卡住、出了事翻日志什么也查不到。这篇不讲协议也不讲客户端配置，只讲把调用写进工程里的那几步——超时怎么分档、并发怎么压住、失败怎么降级、日志该记哪几个字段、上线前拿什么小评测兜底。

你已经能在 Python 里跑通一次大模型请求了，但只要把它放进真实项目，麻烦就开始冒头：偶发超时、被限流、某个模型挂了整条链路跟着卡住、出了事翻日志什么也查不到。这些问题跟协议无关，也跟具体某家客户端无关，纯粹是工程写法的问题。

这篇不讲原理，也不讲某家 SDK 的配置项，只讲把调用写进项目里的那几步：超时怎么分档、并发怎么压住、失败怎么降级、日志该记哪几个字段、上线前拿什么小评测兜底。每一步都给可执行的动作，遇到需要看数字的地方，会告诉你到站内哪个页面去核对。

先写一个能复用的最小调用函数

很多项目的调用代码是一散到底：Key 写在业务函数里，Base URL 复制了三份，模型名硬编码在五个地方。改一次配置要全局搜。第一步不是优化性能，是先把这些东西收拢到一个地方，让后面所有改造都有抓手。

• 把 Key、Base URL、模型名收进环境变量或一个配置对象，代码里只允许通过配置读取，不准直接写字符串
• 统一走一个 client 函数，requests、httpx 或官方 SDK 三选一，别每个功能各写一遍；选哪个看你要不要流式，不流式的话 requests 够用
• 返回结构收敛成固定形状：正文、用量、耗时、是否命中缓存、错误信息；后面所有业务逻辑都吃这个结构，不再直接碰原始响应
• 先发一次最简单的请求确认连通性，再往里加超时、重试这些复杂度；一次加一层，出问题好定位

超时和重试：分档设置，别用默认值

默认超时通常是「等很久」或者「几乎不等」，两种都坑。要分开设连接超时和读取超时：连接超时管的是握手，几秒就够；读取超时管的是模型生成，得按任务类型分档。短问答、长文生成、流式输出根本不是一档，用一个数字套所有场景，要么误杀要么挂死。

1. 按任务分档设读取超时把任务分成短问答、常规生成、长文/流式三类，各给一个读取超时。判断依据是这类任务正常情况下的耗时分布，宁可略宽也不要宽到没边。
2. 重试只对可恢复错误做连接失败、5xx、429 可以重试；4xx 里的鉴权失败和参数错误重试一百次也是白费，直接抛出去。在 client 里按状态码分流，别写一个 except 包所有。
3. 退避加抖动，并设总耗时上限重试间隔要递增，再加一点随机抖动，避免所有实例同一秒一起打。同时给单次请求设一个总耗时上限，否则重试叠加会把线程占死。
4. 流式单独处理流式响应不能整套重试，断了只能尝试续传或降级成非流式。这条路径要单独写，别混进通用重试逻辑里。

并发怎么压住：线程池、信号量与排队

并发上去以后最常见的两个现象：一是被上游限流，二是自己把线程池打满，导致无关请求也超时。解决办法不是调大线程数，而是给并发加一个明确的闸门。

• 用线程池或异步任务池统一管控并发数，不要在业务代码里裸起线程，散出去的线程没人管得住
• 并发上限按你实际拿到的额度来定，账号档位不同上限不同，去 站内 /price-watch 核对当时的数字和限速说明
• 本地加一层信号量做闸门，上游开始报限流时自动降并发，而不是继续按原速率打
• 批处理任务要能分批加断点续跑，失败项单独收集后重投，别一次全量失败就得从头再来

这里有个常见误区：以为并发数越高吞吐越大。实际上超过上游能承受的点之后，重试和排队会把有效吞吐拉低。真正的做法是先找到不触发限流的那个并发值，然后稳定在那个值附近。

失败降级：换模型、换通道、关功能

降级不是「出错就返回空」，而是提前定义好一条有顺序的退路。顺序要在代码里写死，不要靠运行时临时判断，否则故障来了谁都想不出该怎么办。

1. 定义降级顺序主模型失败换备用模型，备用也失败就走缓存结果或返回兜底文案。每一级都要有明确的触发条件，写进配置。
2. 按错误类型分流限流就退避等待，超时就换源，内容被拒是另一套处理。三种情况混在一个 except 里，处理策略必然错。
3. 给降级路径本身设限制降级也要有超时和次数上限，否则一次故障会在备用通道上放大成雪崩。备用的备用只能是兜底，不能无限套。
4. 把降级信号透出给业务方返回结构里加一个字段标明这次结果是降级来的，让上层自己决定要不要展示、要不要重试、要不要提示用户。

日志该记哪几个字段，出事才复盘得动

很多人日志里只记一句「调用失败」，然后就没有然后了。复盘时最想知道的是：请求发出去没有、卡在哪一段、上游返回了什么错误码。这几个问题对应几个固定字段，缺一个都查不动。

• Prompt 和返回内容按合规要求脱敏，或者只记摘要和哈希，别整段落盘
• 把耗时拆成排队、首字、生成三段，慢在哪一段一眼就能看出来
• 同一批任务用同一个批次号串起来，复盘时能按批次重放
• 日志格式统一成结构化字段，别一半是 JSON 一半是拼接字符串

返回值解析：别假设模型一定听话

你要求返回 JSON，模型大概率会给 JSON，但「大概率」不等于「一定」。多包一层 markdown 代码块、多一句解释、少一个括号，都会让你的解析炸掉。解析这一步要当成不可信输入来处理。

1. 解析必须包 tryJSON 解析失败要么让模型按更严格的提示重来一次，要么走兜底结构。绝对不能让解析异常冒泡到业务层。
2. 先校验再入库结构化输出要校验字段是否齐全、类型是否正确，校验不通过就修或者重试，不要直接把半成品写库。
3. 检查结束原因字段长文本可能被截断，光看内容长度不准。要看响应里的结束原因，标明是被截断的就走续写或者缩短输入。
4. 流式拼接注意分片边界流式返回是按分片来的，半截 JSON 是最常见的坑。要等分片拼完整再解析，别边收边解析。

用量与成本：把统计做进调用层

用量统计如果放在业务层，迟早会漏。正确做法是放在 client 函数里，每次调用自动累加，业务代码不需要关心。分维度记录：按功能、按用户、按天，出问题时能快速定位是谁在烧额度。

• 每次调用把 token 用量累计到统一的地方，别让各业务自己算
• 跑批量任务前先估算总量，站内有成本计算器可以对着一遍当时的单价核算
• 给单个任务设用量上限，超了就中断，避免死循环把额度烧完
• 成本异常时先排查是不是重试次数过多，或者 Prompt 尾部越加越长

# 用量累计示意，字段名按实际接口调整
usage = resp.get('usage') or {}
metrics.add(
    feature=task_name,
    user_id=user_id,
    prompt_tokens=usage.get('prompt_tokens', 0),
    completion_tokens=usage.get('completion_tokens', 0),
    degraded=result.get('degraded', False),
)

上线前的小评测：不用大，要能跑

改 Prompt、换模型、调温度，这些动作都会改变输出，但没人能保证改完还是对的。最省事的办法是固定一组小用例，每次改动都跑一遍，看通过率和失败样例，而不是凭感觉。

1. 准备三类用例正常输入、边界输入、预期报错各准备几条，总量控制在能手动看完的范围，关键是覆盖住典型场景。
2. 跑完记录通过率和失败样例只看一个总分没用，要能看到具体哪条挂了、挂成什么样，这样才知道是改坏了还是用例本身有问题。
3. 换模型或通道后必须重跑能力差异不会自动对齐，A 模型上通过的用例 B 模型未必过。换完先跑评测再上线。
4. 评测结果和日志放一起方便判断是新模型的能力问题，还是你代码里的解析、拼接逻辑出了问题。

调用写对了只是能跑，超时、并发、降级、日志都写对了才算能扛。