大模型 API 接入入门

2026-06-11

基本步骤

获取 API Key：在服务商控制台创建密钥。
选择模型与端点：按需求选择模型与调用方式。
发起请求：构造消息体，处理返回与错误。
控制成本：监控调用量与 token 消耗。

你大概率是这么开始的：注册好账号，控制台里点了个”创建密钥”，复制粘贴到代码里，然后对着一堆报错发呆——不是 401 就是 429，要不就是响应半天不返回，超时报错扔出来一脸懵。这篇就把这几步里真正会卡住你的地方摊开讲，看完你应该能自己排查大部分接入问题，而不是每次都去翻服务商的英文文档。

第一步：拿到 Key 之后先别急着写业务代码

创建密钥这步看起来简单，但两个细节决定了你后面会不会返工：

密钥权限范围：多数服务商的控制台允许给密钥设置权限（比如只读、只调用某几个模型、限定调用来源 IP）。生产环境用的密钥和你本地调试用的密钥，权限最好分开建，出问题排查范围小很多。
密钥存放位置：永远不要把 Key 硬编码进代码仓库。哪怕是个人练手项目，也养成用环境变量或 .env 文件（并加进 .gitignore）的习惯。GitHub 上每天都有扫描机器人在爬公开仓库里泄露的密钥，一旦被扫到，几分钟内就可能被盗刷，你的额度就没了。

拿到密钥后，先用最原始的 curl 跑通一次，别一上来就套 SDK——SDK 出错时你分不清是网络问题、参数问题还是封装问题，裸调用能帮你先确认链路是通的：

curl https://api.example.com/v1/chat/completions \
  -H "Authorization: Bearer $YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "model-name",
    "messages": [{"role": "user", "content": "你好"}]
  }'

如果这一步能拿到正常返回（一段带 choices 字段的 JSON），说明网络、鉴权、端点地址都没问题，后面再往上叠 SDK、框架都好排查。如果这一步就报错，直接看下面的报错对照表，不用去怀疑你上层的业务代码。

第二步：模型与端点怎么选，别凭感觉

“选择模型与端点”这句话说起来轻松，实际决策要看三个维度，我按优先级排：

维度	关注点	踩坑提示
任务类型	对话/推理/代码/多模态，不同模型擅长的场景差很多	拿一个通用对话模型硬撑代码生成任务，效果往往不如专门做过代码训练的模型
上下文长度	你的 prompt + 历史对话 + 期望输出会不会超上限	超限不是报错就是静默截断，具体行为每家服务商不一样，务必先查文档确认
计费方式	按 token 计费还是按调用次数，输入输出是否分开计价	输出 token 通常比输入贵好几倍，长输出任务成本容易失控

端点地址这块也容易踩坑：同一家服务商可能同时提供”兼容 OpenAI 协议”和”原生协议”两套端点，返回的字段结构不一样。如果你用的是通用 SDK（比如 OpenAI 官方 SDK）去接第三方兼容端点，一定先确认对方文档写的是”完全兼容”还是”部分兼容”——很多所谓兼容端点在流式响应、function_call、stop 参数这些细节上会有差异，直接套用容易在边界场景翻车。

第三步：请求怎么发，返回怎么处理

发请求本身不难，难的是把返回值和错误都处理到位。一个稍微像样的调用，至少要包含超时设置、状态码判断、重试逻辑三块。拿 Python 举例：

import requests
import time

def call_api(payload, api_key, max_retries=3):
    url = "https://api.example.com/v1/chat/completions"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    for attempt in range(max_retries):
        try:
            resp = requests.post(url, json=payload, headers=headers, timeout=30)
        except requests.exceptions.Timeout:
            print(f"第 {attempt + 1} 次请求超时，准备重试")
            time.sleep(2 ** attempt)  # 指数退避：1s, 2s, 4s
            continue

        if resp.status_code == 200:
            return resp.json()
        elif resp.status_code == 429:
            # 限流了，按响应头里的建议等待时间来，没有就用退避
            wait = int(resp.headers.get("Retry-After", 2 ** attempt))
            print(f"触发限流，等待 {wait} 秒后重试")
            time.sleep(wait)
        elif resp.status_code == 401:
            raise RuntimeError("密钥无效或已过期，检查 Authorization 头是否正确")
        elif resp.status_code >= 500:
            print(f"服务端错误 {resp.status_code}，重试第 {attempt + 1} 次")
            time.sleep(2 ** attempt)
        else:
            raise RuntimeError(f"请求失败：{resp.status_code} {resp.text}")

    raise RuntimeError("多次重试后仍然失败")

这段代码里几个设计不是随手写的，讲一下为什么这么处理：

timeout=30 一定要显式设置：不设置的话 requests 默认不超时，网络抖动时你的程序会一直卡在那一行，看起来像是”卡死”，其实是在傻等。30 秒是个经验值，短对话场景可以调到 10-15 秒，长文本生成或者带工具调用的场景可能要放宽到 60 秒以上。
指数退避（exponential backoff）而不是固定间隔重试：如果是限流触发的失败，固定间隔重试等于变相把请求压力集中在几个时间点上，容易反复触发限流；指数退避把重试间隔拉开（1秒、2秒、4秒……），既给服务端喘息空间，也降低你自己被拉黑的概率。真正做到生产级别的话，退避时间上还应该加一点随机抖动（jitter），避免大量客户端在同一时刻同时重试造成”惊群效应”。
429 和 5xx 要分开对待：429 是你请求太快了，退避等待通常能解决；5xx 是服务端自己出问题，重试有用但次数别设太多，超过 3-5 次还失败基本就该降级到备用模型或者报错给用户，而不是死等。
401 直接抛异常不重试：密钥错了重试多少次都没用，浪费时间还占着重试预算，不如直接报错让你去查密钥配置。

常见报错怎么查

把这张表贴在手边，遇到问题先对号入座：

现象	大概率原因	排查方法
401 Unauthorized	密钥拼写错、加了多余空格、密钥已被吊销	打印出实际发送的 `Authorization` 头字符串，肉眼核对；去控制台确认密钥状态
429 Too Many Requests	超过 QPS 限制或月度额度用尽	看响应头 `Retry-After`；查控制台用量看是不是额度耗尽
请求一直挂起直到超时	网络链路问题，或者服务商机房临时抖动	换个网络环境测试；同时用 `curl -v` 看 TCP 连接是否建立成功
返回内容是乱码或者半截 JSON	用流式接口却当成普通接口解析，或者字符编码没按 UTF-8 处理	确认接口是不是 `stream: true`，流式响应要按 SSE 格式一行一行解析，不能直接 `json.loads()` 整个响应体
Context length exceeded 或类似报错	输入 + 历史对话 + 期望输出超过了模型上下文窗口	先用 tokenizer 工具估算 token 数（不同模型分词方式不同，字符数估不准），超限就做历史对话裁剪或摘要压缩

流式和非流式，什么时候选哪个

stream: true 和默认的一次性返回，不是纯粹的技术选择，直接影响用户体验和你的代码复杂度：

需要”打字机效果”、用户在等着看结果的场景（聊天界面、实时生成）：选流式。用户能看到内容逐字往外冒，即使模型总耗时没变，感知等待时间会短很多。代价是你要处理 SSE 事件流的解析，出错处理也更麻烦——流式响应中途断开了，你已经吐给用户一半内容，剩下怎么补是个体验设计问题。
后台批处理、要拿到完整结果做二次加工（比如提取结构化数据、写入数据库）的场景：选非流式。逻辑简单，一次拿到完整 JSON，不用自己拼接分片内容。

并发怎么控制

如果你的业务是批量调用（比如给一千条数据逐条生成摘要），不要写个 for 循环顺序跑，太慢；但也不能无脑上百个并发，很容易撞到 QPS 限制被限流。实际做法是用信号量控制并发数：

import asyncio

async def worker(semaphore, payload):
    async with semaphore:
        return await call_api_async(payload)

async def batch_call(payloads, concurrency=5):
    semaphore = asyncio.Semaphore(concurrency)
    tasks = [worker(semaphore, p) for p in payloads]
    return await asyncio.gather(*tasks, return_exceptions=True)

并发数（concurrency）设多少合适，没有万能答案，取决于服务商给你的 QPS 上限。一个稳妥的起步做法是先设成 3-5，观察一段时间有没有触发 429，没有的话再逐步往上加，边测边调，比一上来猜一个数字靠谱。

成本控制不是月底看账单才想起来

第 4 步”控制成本”最容易被新手忽略，等到月底账单一出来才发现超预算。真正有用的做法是把监控做在调用发生的当下：

每次调用后从返回体里读出 usage 字段（大部分服务商都会返回本次请求消耗的输入/输出 token 数），实时累加记录到本地日志或者监控系统里，而不是等服务商控制台的账单出来才知道花了多少。
给关键业务路径设置单次请求的 token 上限（max_tokens），防止模型某次”话痨”生成了一大段远超预期的内容，账单跟着炸。
长对话场景做历史裁剪：不是每轮都把全部历史原样塞进去，超过一定轮次就做摘要压缩，既省 token 也避免撞上下文上限。

把这几件事在接入阶段就搭好架子，比等出问题回头补要省心得多，具体的分销和成本优化打法可以看 /topic/access/ 里聚合的其他教程；如果你想第一时间拿到我们后续上线的国产模型对比和成本测算工具，可以去 /waitlist/ 留个联系方式。

小结

先跑通最小示例，再逐步加上重试、缓存与限流；报错别慌，对着状态码查原因，比瞎猜靠谱得多。