3.1 接入大模型 API：调用与流式输出

场景：脚本第一次长出嘴

事情起因于一份周报。每周五我都要跑四个脚本，把采集、清洗的结果汇总成文字，手动粘贴到周报模板里。某天加班时我灵光一闪：把这四段结果丢给大模型，让它按模板生成一份通顺的周报初稿，不就省事了吗？

野心再大一点：把这个能力做成工具箱里的一个通用模块 ask_llm，其他脚本需要总结、分类、改写时直接调用。于是第一步就是接入大模型 API。我用的是腾讯云大模型 API（混元，兼容 OpenAI 接口协议），新用户活动页领了额度，跑通整个流程没花一分钱。选它的原因很简单：国内直连稳定、计费透明、Python SDK 用法和市面主流协议一致，换模型时代码几乎不用改。

方案：一个能复用的 ask_llm

需求看起来简单——发请求、收回复——但真要做成稳定模块，得先把三件事想清楚：密钥放哪、超时怎么办、流式还是一次性返回。我的设计是密钥统一走环境变量，超时分级重试，默认流式输出但保留同步开关。

先看最朴素的同步调用：

import os, json, requests

BASE = "https://api.llmA.example/v1"
KEY = os.environ["LLM_API_KEY"]

def ask_llm(prompt, model="standard-v1", stream=False, timeout=30):
    url = f"{BASE}/chat/completions"
    headers = {"Authorization": f"Bearer {KEY}"}
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "stream": stream,
    }
    resp = requests.post(url, headers=headers, json=payload, timeout=timeout)
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

这段代码能跑，但离能上线还差得远。真正的分水岭在流式输出：同步调用要等十几秒才一次性返回，周报这种长文本用户会以为程序卡死；流式则像打字机一样边生成边吐，首字延迟只有一两秒。

流式输出：SSE 解析

大模型 API 的流式返回是 Server-Sent Events 格式，每行 data: {...}，结尾以 data: [DONE] 收尾。解析时有两个细节容易翻车：一是 requests.post(..., stream=True) 得用 iter_lines 逐行读；二是空行要跳过，否则 JSON 解析会炸。

def ask_llm_stream(prompt, model="standard-v1"):
    url = f"{BASE}/chat/completions"
    headers = {"Authorization": f"Bearer {KEY}"}
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "stream": True,
    }
    with requests.post(url, headers=headers, json=payload, stream=True, timeout=60) as r:
        r.raise_for_status()
        for line in r.iter_lines(decode_unicode=True):
            if not line or not line.startswith("data:"):
                continue
            data = line[5:].strip()
            if data == "[DONE]":
                break
            chunk = json.loads(data)
            delta = chunk["choices"][0]["delta"].get("content", "")
            if delta:
                yield delta

把 yield 交给上层拼字符串或者直接打印，体验立刻从死等变成打字机。调试时我习惯加一行 print(delta, end="", flush=True)，先肉眼确认链路通，再接业务代码。

重试与退避

生产环境里不重试的网络调用基本等于没写。我给 ask_llm 包了一层指数退避，专门应付两类瞬时错误：限流 429 和网关 5xx。重试间隔按 1 秒、2 秒、4 秒递增，配上随机抖动避免多个客户端同时重试撞在一起。

import time, random

def ask_with_retry(prompt, max_retries=3, **kw):
    for i in range(max_retries + 1):
        try:
            return ask_llm(prompt, **kw)
        except Exception as e:
            if i == max_retries:
                raise
            wait = (2 ** i) + random.random()
            print(f"第 {i+1} 次重试，{wait:.1f}s 后重发：{e}")
            time.sleep(wait)

注意重试只覆盖幂等的读取类调用——像扣费、下单这种有副作用的操作，重试前必须先查上一笔的状态，否则可能造成重复扣款。 这个原则在后面Function Calling的工具执行里同样成立，工具调用失败时是重试还是放弃，取决于工具本身是否幂等。

调用方式对照

我给周报工具选了流式，给数据批量分类选了同步加并发，两者共用同一个密钥管理模块，这也在后续多轮对话那篇里被直接复用。

踩坑与排查

坑一：密钥写进代码库。 第一版我把 KEY 硬编码在脚本里，提交到内部 Git 后被安全扫描拦下。改用 .env 加 python-dotenv，线上则用环境变量注入。周报里提醒一句：密钥泄漏是这类项目里代价最低、翻车最狠的错误，别省这一步。

坑二：超时设成 30 秒还不够。 长文本生成动辄四五十秒，requests 的 timeout 是连接与读取超时，流式场景下逐块读取时计时会刷新，但同步等待全量返回就会被掐断。我的做法是同步调用 timeout 给到 120，流式读取侧再加一个空闲计时——两个块之间超过 30 秒没动静就主动断开重试。

坑三：429 限流突袭。 批量跑 200 条分类时，模型 API 的并发与每分钟请求数都有限制，一股脑并发过去半数 429。后来改成指数退避重试，参考我之前在定时任务那篇里写的重试逻辑，并加了令牌桶限速。另外注意 raise_for_status() 不会自动重试，得自己包一层。

坑四：流式断流不报错。 网络抖动时 iter_lines 静默结束，上层拿到半截文本还以为是完整结果。排查时发现 transfer-encoding 提前关闭不会抛异常，于是我在生成结束后校验结尾标记：正常流必然以 [DONE] 结束，没收到就视作断流，走重试或降级到同步调用。

坑五：token 计费看不懂。 第一次账单比预估高两倍，查日志才发现 max_tokens 设太大，模型会倾向于把预算用满。计费按输入加输出 token 算，提示词里带的示例和模板都算钱。后来我养成习惯：上线前用 tiktoken 先本地估算，成本控制那篇里有完整的复盘方法。

小结

接入大模型 API 这一步，技术门槛不高，工程门槛不低。核心是把流式解析、超时重试、密钥管理、token 预算四件事一次性做规范，后面的 RAG、Function Calling 都是在这个地基上盖楼。 ask_llm 这个不到五十行的模块，现在是我们工具箱里调用最频繁的函数，没它后面几节都没法展开。如果想让调用跑在云函数上，可以先翻Serverless 部署入门了解运行环境。

延伸阅读

• Python AI 工具箱全书目录
• RAG 检索：本地知识库问答
• 腾讯云大模型官方文档