WeLM 调用前先跑通一条最小链路 / 把返回解析和报错分支补上

文章导读
接入 WeLM 时真正容易卡住的环节,通常不是业务逻辑本身,而是请求根本没发出去、或者返回拿到了却不知道怎么解析。建议的顺序是:先用一条不掺业务的最短请求确认链路能通,把原始返回完整打印出来,再固定解析结构、补上超时和限流等失败分支,最后用日志把每次请求的输入长度和耗时记下来。这样出问题时能先看日志判断是网络、鉴权、配额还是解析写错,而不是在业务代码里逐行猜。下面涉及的地址、鉴权字段名、返回字段一
📋 目录
  1. 一 确认可用的调用入口和鉴权方式
  2. 二 用一条最短请求验证连通性
  3. 三 把返回解析成固定结构
  4. 四 补上超时、限流、返回格式异常三类分支
  5. 五 用日志记录每次请求的输入长度和耗时
A A

接入 WeLM 时真正容易卡住的环节,通常不是业务逻辑本身,而是请求根本没发出去、或者返回拿到了却不知道怎么解析。建议的顺序是:先用一条不掺业务的最短请求确认链路能通,把原始返回完整打印出来,再固定解析结构、补上超时和限流等失败分支,最后用日志把每次请求的输入长度和耗时记下来。这样出问题时能先看日志判断是网络、鉴权、配额还是解析写错,而不是在业务代码里逐行猜。下面涉及的地址、鉴权字段名、返回字段一律是占位符,实际以官方文档为准。

如果你刚准备接 WeLM,先把链路跑通再写业务:核对调用地址和鉴权字段(以官方文档为准),用一条最短请求把原始返回打印出来,再补字段解析和超时、限流、格式异常三类分支。验证方式是能在日志里看到请求标识、输入长度、耗时和状态码;边界在于地址、字段名、配额都可能变,未核实前不要写死进业务代码。

确认可用的调用入口和鉴权方式

调用地址、鉴权字段名、请求体字段这类前置条件,建议一次性核对清楚再动手写业务代码。核对对象以官方文档为准,本文示例中的地址、header 名、字段名都是占位符,不能当真实接口直接用。

可以按下面的清单逐项确认,每项后面标上“已确认”或“待核实”,未确认的先别写死:

  • 调用入口:完整 host 与 path、HTTP 方法、是否区分内网和公网地址。
  • 鉴权方式:凭证放 header 还是 query,字段名是 Authorization 还是其他名称,是否需要 Bearer 之类的前缀。
  • 请求体:模型名或版本号的写法、输入字段名、是否支持批量、单次最大输入长度的口径(字符还是 token)。
  • 返回结构:是整段 JSON 还是流式返回,文本字段挂在 result、text 还是 choices 下面。
  • 配额与限流:是否有 QPS 或并发限制,超额时返回什么状态码或错误码。
  • 错误码表:常见错误码含义,以及哪些可以重试。

文档里写得含糊的项,建议用最小请求实测一次并观察原始响应,把它记成待核实项,而不是按经验猜字段名。鉴权凭证不要写进前端代码或提交进仓库,从环境变量读取更稳妥。

用一条最短请求验证连通性

先不写任何业务逻辑,只确认请求能发出、返回能拿到。下面是一段通用骨架,字段名和地址全部是占位符,替换成确认过的值再运行。

WeLM 调用前先跑通一条最小链路 / 把返回解析和报错分支补上
import requests

API_URL = 'https://<api-host>/<path>'      # 以官方文档为准
HEADERS = {
    'Authorization': 'Bearer <your-key>',     # 字段名以官方文档为准
    'Content-Type': 'application/json',
}
payload = {
    'model': '<model-name>',
    'input': '只回答两个字:收到',
}

resp = requests.post(API_URL, headers=HEADERS, json=payload, timeout=10)
print('status:', resp.status_code)
print('headers:', dict(resp.headers))
print('body:', resp.text[:2000])   # 先看原始返回,不要急着 json()

curl 或任意 HTTP 客户端都可以,关键是同一份请求能稳定复现一次。验证点有三个:HTTP 状态码是否属于成功类;返回体是否非空;返回体能否被 json.loads 解析。状态码 401 或 403 多半出在鉴权字段或凭证上;404 通常是 path 写错;429 说明已经触发限流,跟业务代码无关。第一次调用建议把响应 header 也打出来,限流相关信息有时会放在 header 里。

把返回解析成固定结构

解析的目标是让下游拿到的永远是同一种形状,不要让调用方直接接触原始字符串,也不要在业务里到处写 data['xxx']。下面这段骨架里字段名按文档替换,字段缺失或类型不符时走兜底结构。

import json

def parse_result(raw):
    if not raw:
        return {'ok': False, 'text': '', 'error': 'empty'}
    try:
        data = json.loads(raw)
    except (TypeError, ValueError):
        return {'ok': False, 'text': '', 'error': 'invalid_json'}
    if not isinstance(data, dict):
        return {'ok': False, 'text': '', 'error': 'unexpected_type'}
    text = data.get('result') or data.get('text')   # 字段名以官方文档为准
    if not isinstance(text, str):
        return {'ok': False, 'text': '', 'error': 'text_missing'}
    return {'ok': True, 'text': text, 'error': None}

三个关键点:json.loads 一定要包在 try 里,因为限流或网关错误返回的可能是一段 HTML;取字段用带默认值的方式,不要假定层级一定存在;对 text 做一次类型检查,避免下游把 None 当字符串拼接。如果接口是流式分片返回,解析逻辑要改成逐片拼接后再统一校验,不要每一片都单独判定成功。

WeLM 调用前先跑通一条最小链路 / 把返回解析和报错分支补上

补上超时、限流、返回格式异常三类分支

失败不是异常情况,是常态。建议至少显式处理三类:超时、限流、返回格式异常。每一类都要有明确的判定条件和下一步动作,而不是统一抛一个“调用失败”。

  • 超时:判定条件是客户端抛超时异常,或耗时超过自己设的阈值(例如 10 秒)。动作是可重试 1 到 2 次,采用指数退避加少量随机抖动,避免同一时刻集中重试;仍失败就降级返回。
  • 限流:判定条件常见是 HTTP 429,或错误体里带限流相关错误码。动作是优先按响应里的 Retry-After 等待(如果文档提供该字段),没有就按固定间隔退避并降低并发;限流不建议马上重试。
  • 返回格式异常:判定条件是 json 解析失败、必需字段缺失或类型不符。动作是通常不重试,记录被截断的原始返回便于排查;若怀疑是网关偶发问题,可重试一次再降级。
import time
import random

def backoff(attempt, base=0.5, cap=8.0):
    delay = min(base * (2 ** attempt) + random.random() * 0.2, cap)
    time.sleep(delay)

降级时要返回与成功时同构的内容,例如 {'ok': False, 'text': '', 'error': 'timeout'},让下游不用再写 try/except。错误标识建议区分 timeout、rate_limited、bad_format,不要统一写 fail,这样日志里一眼能看出是哪类问题。重试只对幂等请求做,如果调用本身带副作用,需要结合业务确认能不能重试。

用日志记录每次请求的输入长度和耗时

前面几步做完之后,真正让排查变轻松的是日志。建议每次调用只打一条结构化日志,至少包含这些字段:

  • request_id:每次调用生成,重试时沿用同一个,方便串起多次尝试。
  • input_len:输入长度,按文档口径记字符数或 token 数,两者不要混用。
  • elapsed_ms:从发出请求到拿到响应的耗时;如果包含重试,额外记一个总耗时字段。
  • status:HTTP 状态码;业务层如果有状态,也一并记录。
  • error_code:解析后的错误分类,如 timeout、rate_limited、bad_format,成功时为 null。
  • 可选字段:model 名、retry_count、是否走了降级。
import json
import logging
import time
import uuid

def log_call(resp, payload, t0, error_code=None, retry_count=0):
    record = {
        'request_id': uuid.uuid4().hex,
        'input_len': len(payload.get('input', '')),
        'elapsed_ms': int((time.time() - t0) * 1000),
        'status': getattr(resp, 'status_code', None),
        'error_code': error_code,
        'retry_count': retry_count,
    }
    logging.info(json.dumps(record, ensure_ascii=False))

两个边界要注意:不要把完整输入和完整返回原样打进日志,可能包含敏感内容,建议只记长度和截断后的片段;request_id 要在请求发出前生成并透传到日志里,否则重试之后无法对应。验证方式是各跑一次成功请求和一次故意触发的失败请求,确认两条日志的字段齐全、错误码能区分。做到这一步,回头再查返回解析或超时,通常可以直接从日志定位,而不用靠猜。