先跑通最小请求再加业务逻辑——Gemini 4 Carbon 的接入顺序

接入 Gemini 4 Carbon 时，比较稳妥的顺序是：先把密钥和端点用环境变量固定下来并确认程序读到，再发一次最小同步请求验证鉴权、端点和请求体格式，之后才把调用改成流式，接着补上超时与重试，最后把成功的那次请求固化成冒烟测试脚本。这样出问题时，能分清是接入配置错了，还是业务逻辑写错了。反过来，一上手就写完整业务代码，报错会混在一起，排查成本很高。

适用场景：第一次把 Gemini 4 Carbon 接进项目，或更换端点、密钥、模型名之后。操作动作：按环境变量 → 最小同步请求 → 流式 → 超时重试 → 冒烟测试的顺序推进，每一步都独立跑通再往下走。验证方式：看启动日志里读到的配置、看单次请求的状态码与响应字段、看流式分片日志、跑一遍冒烟脚本的退出码。风险边界：如果业务代码里已经混入大量重试和并发，建议先把接入层拆成独立函数再按这个顺序验证。

把密钥和端点放进环境变量再确认加载成功

第一步，要解决的是“程序到底读到了什么”。密钥写死在代码里，一旦换环境或提交到仓库就会出问题；只放到环境变量里但没确认加载，也一样会在后面报鉴权错误时浪费大量时间。命名建议统一前缀，按用途区分：GEMINI_API_KEY、GEMINI_BASE_URL、GEMINI_MODEL。前缀一致，后续换服务商或加多套环境时不容易冲突。

读取时给默认值要谨慎：模型名可以有默认值，密钥和端点建议没有默认值，读到空值就在启动阶段直接失败。日志里只打印“是否读到”和长度、前几位，不打印完整密钥。下面的写法可以放在服务启动入口或配置模块里，跑一次就能确认配置来源：

import os

API_KEY = os.environ.get("GEMINI_API_KEY")
BASE_URL = os.environ.get("GEMINI_BASE_URL", "").rstrip("/")
MODEL = os.environ.get("GEMINI_MODEL", "gemini-4-carbon")

def mask(v):
    if not v:
        return "<empty>"
    return f"{v[:4]}***len={len(v)}"

print("[config] GEMINI_API_KEY =", mask(API_KEY))
print("[config] GEMINI_BASE_URL =", BASE_URL or "<empty>")
print("[config] GEMINI_MODEL =", MODEL)

if not API_KEY or not BASE_URL:
    raise SystemExit("missing GEMINI_API_KEY / GEMINI_BASE_URL")

验证方式很直接：本地跑一次，确认输出里的长度和你在控制台看到的一致；换到容器或 CI 环境再跑一次，确认不是只在本地 shell 里生效。如果长度对不上，通常是被别的环境变量覆盖，或者配置文件加载顺序有问题，先解决这个再往下走。

用最小请求跑通一次同步调用

同步请求阶段只验证三件事：鉴权是否通过、端点是否可达、请求体格式是否被接受。这三件事都成立，才谈得上业务。请求体保持最小，只放一条用户消息，不要带工具调用、系统提示词、多轮历史。端点和字段名以实际接口文档为准，下面给的是通用骨架，路径和字段需要按你的接入方式替换：

import requests

url = f"{BASE_URL}/v1/models/{MODEL}:generateContent"  # 占位路径，按实际端点替换
headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
}
payload = {
    "contents": [{"role": "user", "parts": [{"text": "ping"}]}],
}

resp = requests.post(url, headers=headers, json=payload, timeout=(5, 30))
print("status =", resp.status_code)
print("body =", resp.text[:500])

成功判定不要只看“没抛异常”。建议同时满足两条：状态码是 200 一类，且响应体能解析出非空的文本内容字段。失败时优先看三个位置，按顺序排查效率最高：状态码决定问题类别，响应体里的错误码和错误消息说明具体原因，响应头里的请求标识用于和服务端日志对齐。

401 / 403：先回到上一步，确认密钥读到的值和实际使用的密钥一致，注意有没有多余空格或换行。
404：端点路径或模型名不对，把 BASE_URL 和 MODEL 打印出来对照。
400：请求体字段不匹配，把最小请求体降到一个字段再试，逐步加回。

在同步成功的基础上改为流式读取

流式是增量能力，不是第一版就该上的东西。同步调用能稳定拿到内容之后，再改成流式，此时如果分片解析出错，可以确定是流式读取部分的问题，而不是鉴权或端点的问题。逐块读取的循环骨架大致如下，结束标记的判断方式要按实际协议替换，常见的是空行分隔加一个显式结束信号，或者响应里带结束原因字段：

import time

start = time.time()
chunks = []

with requests.post(url, headers=headers, json=payload,
                   timeout=(5, 60), stream=True) as resp:
    resp.raise_for_status()
    for i, line in enumerate(resp.iter_lines(decode_unicode=True)):
        if not line:
            continue
        if line.strip() == "data: [DONE]":  # 结束标记按实际协议替换
            break
        chunks.append(line)
        print(f"[chunk] i={i:03d} len={len(line)} t={time.time() - start:.2f}s head={line[:60]}")

print(f"[done] chunks={len(chunks)} total={sum(len(c) for c in chunks)}")

分片日志建议固定成三段信息：序号、本片长度、相对开始的时间，必要时再加前几十个字符。有了这些字段，能直接看出是“一片都没收到”“收到了但内容为空”还是“中途断开”。流式成功判定不能只看连接没断，要看是否收到多个分片，并且最后出现了结束标记；只有结束标记出现，才说明这一轮输出是完整的。

补充超时与重试封装

超时参数放在每一次请求调用的参数里，而不是全局配置里改一个数字，建议连接超时和读取超时分开设置：连接超时短一些，读取超时按输出长度留足余量。流式请求的读取超时管的是单次分片间隔，不是整轮时长，这一点需要在压测或联调时结合实际表现调整。

重试只对“重试有可能成功”的错误有意义。次数不必多，两到三次即可，退避用指数加一点随机抖动，避免多个实例同时重试：

import time, random

RETRYABLE = {500, 502, 503, 504}

def call_with_retry(fn, max_attempts=3, base=0.5):
    for attempt in range(1, max_attempts + 1):
        try:
            r = fn()
            if r.status_code in RETRYABLE and attempt < max_attempts:
                raise IOError(f"status={r.status_code}")
            return r
        except (TimeoutError, ConnectionError, IOError) as e:
            if attempt == max_attempts:
                raise
            sleep = base * (2 ** (attempt - 1)) + random.uniform(0, 0.2)
            print(f"[retry] attempt={attempt} err={e} sleep={sleep:.2f}s")
            time.sleep(sleep)

不该重试的情况要明确写进封装里：400 类请求体错误重试只会重复失败；401 / 403 是鉴权问题，重试没有意义；404 通常是路径或模型名错，需要改配置；限流类错误建议先看服务端是否给了等待提示，是否重试要结合你的调用量和业务容忍度决定，不要默认无脑重试。流式请求已经输出一部分再断开时，重试整轮可能造成内容重复，这种情况建议交给业务层处理，接入层只负责报告断开位置。

把成功请求固化成冒烟测试

最后一步是把上面验证过的最小请求写成脚本，放在仓库里，任何人改完接入层代码都能跑一遍。脚本只做四件事：读取环境变量、发一次最小请求、断言关键字段、输出通过或失败并以退出码体现结果。结构可以照抄下面这份，断言部分按实际响应结构替换：

import os, sys, json, requests

def main() -> int:
    key = os.environ.get("GEMINI_API_KEY")
    base = os.environ.get("GEMINI_BASE_URL", "").rstrip("/")
    model = os.environ.get("GEMINI_MODEL", "gemini-4-carbon")
    if not key or not base:
        print("FAIL: missing GEMINI_API_KEY / GEMINI_BASE_URL")
        return 2

    url = f"{base}/v1/models/{model}:generateContent"
    payload = {"contents": [{"role": "user", "parts": [{"text": "ping"}]}]}

    try:
        resp = requests.post(
            url,
            headers={"Authorization": f"Bearer {key}",
                     "Content-Type": "application/json"},
            json=payload,
            timeout=(5, 30),
        )
    except Exception as e:
        print(f"FAIL: request error {type(e).__name__}: {e}")
        return 1

    if resp.status_code != 200:
        print(f"FAIL: status={resp.status_code} body={resp.text[:300]}")
        return 1

    data = resp.json()
    text = ""  # 按实际响应结构取出文本字段
    if not text:
        print(f"FAIL: empty content body={json.dumps(data, ensure_ascii=False)[:300]}")
        return 1

    print(f"PASS: model={model} len={len(text)}")
    return 0

if __name__ == "__main__":
    sys.exit(main())

用法上，本地改完配置先跑这个脚本，CI 里作为接入层的独立检查项也合适。退出码为 0 才算通过，非 0 时流水线会直接拦下来。它不替代业务测试，只保证“密钥、端点、模型名、请求体格式”这几项没被改坏；业务逻辑的正确性，仍然需要另外的用例覆盖。