用 Python 调用 BigModel API：模型列表、对话补全与流式输出

BigModel 是一个 AI API 聚合网关，对外提供与 OpenAI 兼容的接口。对 Python 开发者来说，这意味着两件事：既可以用官方 openai SDK 把 base_url 指向 BigModel 的网关地址后直接复用，也可以用最朴素的 requests 手写 HTTP 请求。本文要教会读者的具体事情是：从零把一条 Python 调用链跑通——建 Key、查模型、发对话、开流式、处理报错。如果你只是想快速验证一个 Key 能不能用，或者准备把 BigModel 接进自己的后端服务，这篇的步骤可以直接照做。需要先说明的是，本站也提供 BigModel 相关的工具入口，可以在动手前先看一眼。

整篇教程围绕五个动作展开：创建 API Key、列出账号可用的模型、发送第一次对话补全、切换到流式输出、以及为常见错误写显式处理。每一步都给出可直接运行的代码，最后再串成一个完整示例。文中涉及的价格、免费额度、模型清单与可用性都属于会变动的信息，请以官网当前页面为准。

准备工作

在写第一行代码之前，需要把下面几项准备好。缺任何一项，后面的步骤都会卡住。

• BigModel 账号：在 bigmodel.org 登录，进入控制台（dashboard）或开发者控制台。API Key 的创建、用量日志、余额信息都在控制台里。
• 一个 API Key：在控制台的 API keys（有的界面叫 tokens 或 credentials）区域新建。这个 Key 等同于一个绑定了计费的密码，不要写进源码，也不要提交到版本库。
• Python 3.7 或更高版本，以及 requests 包。如果打算走 SDK 路线，还需要安装 openai 包。
• 一个能保存环境变量的终端：Key 通过环境变量注入，代码里只读 os.environ。

安装依赖的命令如下：

pip install requests
pip install openai

第二条只在你要用 OpenAI SDK 时才需要。两条都装也不冲突，本文后面两种写法都会演示。

操作步骤

第一步：创建 API Key 并放进环境变量

登录 BigModel 控制台，打开 API keys（或 tokens / credentials）区域，新建一个 Key，复制它的值。然后在本机终端里导出为环境变量：

export BIGMODEL_API_KEY="replace-with-your-real-secret"

如果你更习惯用 .env 文件管理配置，也可以把这一行写进 .env，但务必把 .env 加进 .gitignore。Key 泄露的后果不只是别人能调用，而是账单会记在你头上。

验证环境变量是否生效：

python -c "import os; print(bool(os.environ.get('BIGMODEL_API_KEY')))"

输出 True 说明变量已经注入当前 shell。注意环境变量是进程级的，如果你在另一个终端窗口或 IDE 里运行代码，需要在那里重新导出，或者通过 IDE 的运行配置注入。

第二步：列出账号可用的模型

BigModel 的网关使用 /v1 路径和 Bearer Token 认证。模型列表请求一次就能同时验证三件事：Key 是否有效、网关地址是否正确、认证头格式是否写对。所以它适合作为接入的第一个测试点。

import os
import requests

api_key = os.environ["BIGMODEL_API_KEY"]

resp = requests.get(
    "https://bigmodel.org/v1/models",
    headers={"Authorization": f"Bearer {api_key}"},
    timeout=30,
)
resp.raise_for_status()

for model in resp.json().get("data", []):
    print(model.get("id"))

返回体里的 data 是一个数组，每个元素至少带一个 id 字段，这个 id 就是后续所有请求里 model 参数要填的值。

这里有一条很重要的纪律：后续每个请求的模型 ID 都应该来自这次响应，而不是从旧教程里抄一个模型名。 模型是否可用会随账号权限和当前服务配置变化，硬编码一个过时的名字，轻则报 400，重则让你误以为 Key 有问题。把这一步做成脚本里的一个函数，每次启动时刷新一次模型列表，是更稳妥的做法。

第三步：用 requests 发送第一次对话补全

拿到模型 ID 之后，就可以向 /v1/chat/completions 发 POST 请求了。请求体是标准的 OpenAI 风格：model、messages、stream 三个字段。

import os
import requests

api_key = os.environ["BIGMODEL_API_KEY"]

resp = requests.post(
    "https://bigmodel.org/v1/chat/completions",
    headers={
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    },
    json={
        "model": "MODEL_ID_FROM_V1_MODELS",
        "messages": [{"role": "user", "content": "Hello! Tell me about yourself."}],
        "stream": False,
    },
    timeout=60,
)
resp.raise_for_status()

data = resp.json()
print(data["choices"][0]["message"]["content"])

把 MODEL_ID_FROM_V1_MODELS 换成上一步打印出来的某个真实 ID。注意 Content-Type: application/json 这个头不能省，否则服务端可能无法正确解析请求体。

怎么读响应。 非流式响应里，生成的文本位于 choices[0].message.content。响应体的其余部分还包含 usage 和模型信息，这些字段适合写进日志，用来做用量核算和排查异常流量。建议在第一次跑通时把整个 JSON 打印出来看一遍，熟悉字段结构，而不是只取 content 就丢掉。

第四步：改用 OpenAI Python SDK

因为网关是 OpenAI 兼容的，官方 openai SDK 只需要改两处就能用：把 base_url 指向 BigModel 的网关地址，把 api_key 换成你的 BigModel Key。其余调用方式与调用 OpenAI 完全一致。

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["BIGMODEL_API_KEY"],
    base_url="https://bigmodel.org/v1",
)

response = client.chat.completions.create(
    model="MODEL_ID_FROM_V1_MODELS",
    messages=[{"role": "user", "content": "Hello!"}],
)

print(response.choices[0].message.content)

SDK 路线的好处是省掉了手写请求头、手写 JSON 序列化、手写响应解析这些样板代码，而且后续如果要加工具调用、多轮对话之类的结构，SDK 的类型定义会更省心。代价是多一个依赖。两种方式选一种即可，不必混用。

第五步：开启流式输出

把 stream 设为 true 之后，API 不再等整段回答生成完才返回，而是边生成边把片段推给客户端。对长回答来说，这能显著改善等待体验——用户能立刻看到内容开始出现，而不是盯着空白页面。

用 OpenAI SDK 消费流：

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.environ["BIGMODEL_API_KEY"],
    base_url="https://bigmodel.org/v1",
)

stream = client.chat.completions.create(
    model="MODEL_ID_FROM_V1_MODELS",
    messages=[{"role": "user", "content": "Write a short haiku about Python."}],
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta.content if chunk.choices else None
    if delta:
        print(delta, end="", flush=True)

这里有两个容易踩的细节。第一，chunk.choices 可能为空数组，直接取 [0] 会抛 IndexError，所以要先用条件表达式兜住。第二，delta.content 也可能是 None（比如某些只带元信息的块），所以要判断非空再打印。flush=True 保证输出立即刷到终端，否则你可能看到内容成批出现，误以为流式没生效。

用 requests 消费流：

走裸 HTTP 时，需要保持连接打开，逐行读取 Server-Sent Events 的 data: 行，直到遇到 [DONE] 标记为止。

import json
import os
import requests

api_key = os.environ["BIGMODEL_API_KEY"]

with requests.post(
    "https://bigmodel.org/v1/chat/completions",
    headers={
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    },
    json={
        "model": "MODEL_ID_FROM_V1_MODELS",
        "messages": [{"role": "user", "content": "Write a short haiku about Python."}],
        "stream": True,
    },
    stream=True,
    timeout=120,
) as resp:
    resp.raise_for_status()
    for line in resp.iter_lines(decode_unicode=True):
        if not line or not line.startswith("data:"):
            continue
        payload = line[len("data:"):].strip()
        if payload == "[DONE]":
            break
        try:
            chunk = json.loads(payload)
        except json.JSONDecodeError:
            continue
        delta = chunk.get("choices", [{}])[0].get("delta", {}).get("content")
        if delta:
            print(delta, end="", flush=True)

这段代码里有几处防御性写法值得保留：跳过空行和非 data: 开头的行；遇到 [DONE] 主动跳出循环；对 json.loads 失败的行直接跳过而不是让整个流崩掉；用 .get() 链式取值避免 KeyError。流式场景下，任何一处没兜住都会让用户看到半截回答然后报错。

第六步：写显式的错误处理

不要指望异常自己变成可读的提示。下面按状态码分类说明该怎么处理。

认证错误（401）。 401 表示请求没有通过认证。依次检查三件事：Key 是否有效（有没有复制时带上了多余空格或换行）、请求头是否严格是 Authorization: Bearer <your-api-key> 这个形式（Bearer 后面一个空格，不能写成 Bearer: ）、以及你调用的确实是 /v1 网关地址。

模型与端点错误（400 / 404）。 重新请求一次 /v1/models，从返回结果里挑一个真实存在的 ID 使用。同时确认请求体里有合法的 model 字段，且 messages 是非空数组。空 messages 或拼错的路径都会落到这两类状态码上。

限流、配额与服务端错误（429 / 5xx）。 这类错误适合用指数退避重试。BigModel 控制台提供用量日志和钱包信息，可以用来核对 token 消耗和余额——如果 429 反复出现，先去看日志确认是流量突增还是余额不足。

网络与超时错误。 requests 在传输层出问题时抛 ConnectionError 和 Timeout。每个请求都要显式传 timeout，并对瞬时故障做小退避重试。

下面是一个带重试的封装：

import os
import time
import requests

api_key = os.environ["BIGMODEL_API_KEY"]

def chat_with_retry(model_id, messages, max_retries=3):
    for attempt in range(max_retries):
        try:
            resp = requests.post(
                "https://bigmodel.org/v1/chat/completions",
                headers={
                    "Authorization": f"Bearer {api_key}",
                    "Content-Type": "application/json",
                },
                json={"model": model_id, "messages": messages, "stream": False},
                timeout=60,
            )
            if resp.status_code == 429 or resp.status_code >= 500:
                raise requests.HTTPError(f"retryable status {resp.status_code}")
            resp.raise_for_status()
            return resp.json()
        except (requests.ConnectionError, requests.Timeout, requests.HTTPError):
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

data = chat_with_retry(
    "MODEL_ID_FROM_V1_MODELS",
    [{"role": "user", "content": "Hello!"}],
)
print(data["choices"][0]["message"]["content"])

这段逻辑的关键在于：只有 429 和 5xx 被主动转成可重试的 HTTPError，其余 4xx（比如 401、400）会由 raise_for_status() 抛出后立即向上冒泡，不会被无意义地重试三次。退避间隔是 2 ** attempt，即 1 秒、2 秒、4 秒。最后一次尝试失败时直接抛出，让调用方知道确实没成功。

一个完整示例

把前面的步骤串起来，下面是一个从查模型到流式输出的最小可运行脚本。它先刷新模型列表，取第一个可用 ID，然后用 SDK 发起流式请求，并带一个简单的重试包装。

import os
import time
import requests
from openai import OpenAI

api_key = os.environ["BIGMODEL_API_KEY"]
base_url = "https://bigmodel.org/v1"

# 1. 取一个当前可用的模型 ID
def pick_model():
    resp = requests.get(
        f"{base_url}/models",
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=30,
    )
    resp.raise_for_status()
    models = resp.json().get("data", [])
    if not models:
        raise RuntimeError("no model available for this account")
    return models[0]["id"]

model_id = pick_model()
print(f"using model: {model_id}")

# 2. 用 SDK 发起流式对话
client = OpenAI(api_key=api_key, base_url=base_url)

stream = client.chat.completions.create(
    model=model_id,
    messages=[
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "Explain what an API gateway does in two sentences."},
    ],
    stream=True,
)

for chunk in stream:
    delta = chunk.choices[0].delta.content if chunk.choices else None
    if delta:
        print(delta, end="", flush=True)
print()

运行方式：

export BIGMODEL_API_KEY="replace-with-your-real-secret"
python bigmodel_demo.py

预期看到的行为是：脚本先打印一行 using model: ...，然后回答内容逐字逐段出现在终端上，最后换行结束。如果第一行就报错，问题出在 Key 或网关地址；如果第一行正常但第二段报错，问题多半在模型 ID 或请求体结构。

这个示例刻意把「取模型」和「发请求」分成两个函数，是为了让失败点容易定位。实际项目里，模型 ID 可以缓存一段时间，不必每次请求都查一遍；但缓存要有过期策略，否则服务端模型调整后你的代码会静默失效。

注意事项

Key 只能放在服务端。 不要把 API Key 写进前端代码、移动端包体或公开仓库。从环境变量或密钥管理服务读取，且只在服务端使用。Key 一旦泄露，别人调用产生的费用记在你的账户上。

模型 ID 要动态解析。 每次接入或部署时都从 /v1/models 取当前可用的 ID，不要沿用旧教程里的模型名。账号权限和服务配置都会变，硬编码是接入后最常见的失效原因。

每个请求都要设 timeout。 不设超时的请求在网络抖动时会一直挂着，占住线程或连接池。流式请求的超时应该比非流式更长，因为连接要保持到生成结束。

流式解析要容错。 空行、非 data: 行、无法解析的 JSON、空的 choices、None 的 delta.content，这几种情况在真实流里都会出现，逐条兜住才能保证输出不中断。

用量要盯。 控制台的用量日志可以查看 token 消耗，钱包信息可以看余额。接入初期建议定期翻一下，既能核对计费，也能及早发现异常流量。价格、免费额度与计费口径会调整，具体数字以官网当前信息为准。

重试要有边界。 只对 429、5xx 和传输层错误重试，4xx 里的认证与参数错误重试多少次都不会成功，反而会放大问题。重试次数和退避上限都要设死。

长回答优先用流式。 当回答可能很长时，stream=True 能让用户尽早看到内容。但流式也意味着错误可能在中途才暴露，所以客户端要能处理「已经输出了一部分然后失败」的情况，而不是假设流一定会正常结束。