调用大模型 API：从 requests 到流式响应

调用 LLM 的本质：一个 HTTP 请求

在所有"AI 很神秘"的外衣下面，你与大模型交互的真实动作只有一个——向一个 HTTP 端点发 POST 请求，请求体是 JSON，响应也是 JSON。没有 WebSocket，没有二进制协议，没有特殊 SDK 才能解锁的隐藏能力。你会用 requests，你就会调大模型。

更重要的一件事：主流模型服务商几乎全部兼容 OpenAI 的接口协议。这意味着你学会调通一家之后，切到 DeepSeek、Qwen、Kimi、硅基流动、自己部署的 vLLM，只需要改两个参数：base_url 和 api_key。这是 2023 年以来事实上形成的行业标准，短期不会变。

本篇我们按四个阶段推进：先用最原始的 requests 跑通一次请求看清协议长相，再用 openai SDK 简化代码，然后处理多轮对话和流式输出，最后把密钥管理、错误处理、重试这些工程细节补齐。

准备项目与依赖

沿用上一篇的 ai-hello 目录，先把需要的库装上：

cd ai-hello
source .venv/bin/activate

pip install requests openai python-dotenv
pip freeze > requirements.txt

去 DeepSeek 开放平台 注册账号，领一把 API Key（新用户通常送几块钱，够整个系列用）。不要把 Key 直接写进代码，在项目根目录创建一个 .env 文件：

DEEPSEEK_API_KEY=sk-xxxxxxxxxxxxxxxx

然后把 .env 加入 .gitignore：

.venv/
.env
__pycache__/

这是一个在 AI 项目里会反复出现的模式——密钥放在 .env，代码里通过 os.getenv 读取，.env 不进 Git。养成这个习惯可以避免后面那种"API Key 被扒出来，账单一夜变两千块"的经典事故。

最朴素的一次调用

我们先不使用任何 SDK，用 requests 直接打接口，这样协议最透明：

# hello_raw.py
import os
import requests
from dotenv import load_dotenv

load_dotenv()

resp = requests.post(
    "https://api.deepseek.com/v1/chat/completions",
    headers={
        "Authorization": f"Bearer {os.getenv('DEEPSEEK_API_KEY')}",
        "Content-Type": "application/json",
    },
    json={
        "model": "deepseek-chat",
        "messages": [
            {"role": "user", "content": "用一句话解释什么是闭包"}
        ],
    },
    timeout=30,
)

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

跑一下：

python hello_raw.py

正常情况下你会看到模型对"闭包"的一句话解释。整个流程一共涉及这些内容：

• URL：https://api.deepseek.com/v1/chat/completions，是 OpenAI /v1/chat/completions 的镜像路径
• 请求头 Authorization: Bearer <key>，标准 OAuth2 风格
• 请求体有两个关键字段：model 指定要调用哪个模型，messages 是对话内容
• 响应结构固定：choices[0].message.content 是模型的回答，此外还有 usage 字段记录消耗了多少 Token

这就是所有 LLM 调用的骨架。后面的所有"高级特性"都是在这个骨架上加字段。

理解 messages 协议

messages 是一个列表，每一项是一条消息，每条消息都有两个固定字段：role 和 content。role 只能取几个预定义的值，其中最核心的是 system、user、assistant 这三种。先说它们是怎么来的，再讲各自的含义，这样你记得住。

三种角色的由来

2020 年 GPT-3 最早开放 API 时，并没有对话的概念，接口叫 Completions——你传一段纯文本，模型接着往下续写。想让它"像个聊天机器人"回你，只能自己手动把历史拼成类似 "用户: xxx\n助手:" 的字符串塞进去。问题是模型没有任何结构信号去区分"这句话是谁说的"，经常把你给它的指令和用户的提问混在一起处理，行为很不稳定。

2022 年底 ChatGPT 上线，底层是 gpt-3.5-turbo。为了让这个模型能稳定区分"规则设定""用户提问""模型自己的历史回复"三种不同身份，OpenAI 在训练阶段引入了一套内部标记格式叫 ChatML，把每条消息在训练数据里就已经包成 <|im_start|>role ... <|im_end|> 的形式，让模型从底层就"见过"角色标签。2023 年 3 月推出的 Chat Completions 接口，就是把这套复杂的内部标记对外抽象成了你今天看到的 messages 数组：三种 role 对应训练时模型已经认识的三种身份。

这也是为什么几乎所有后来的模型厂商都直接沿用了同样的三种角色——他们也想兼容这套已经成了事实标准的协议，业务方迁移起来零成本。

system——规则设定者

用来告诉模型它是谁、要遵循什么规则、按什么格式输出。它不是对话内容，而是"游戏规则"。模型在训练时被显式地强化了对 system 内容的遵循权重，因此同样一条指令，写在 system 里通常比写在 user 里更稳定。

{"role": "system", "content": "你是一位资深 Python 工程师。回答只给代码，不附加任何解释文字。"}

约定俗成：一个 messages 列表里通常只放一条 system，而且放在最开头。中途再插一条语法上是合法的，但不同模型的处理行为不一致，不推荐依赖。

user——人类用户的输入

对话里由"人"发出的消息。一个聊天产品里，所有 UI 输入框打进去的文本最终都会被包成 user 角色塞进这个列表。

{"role": "user", "content": "给我写一个斐波那契数列的递归实现"}

assistant——模型之前说过的话

这个最容易被新手忽略。因为 API 本身不保存任何对话历史，如果你想让模型"记得"上一轮聊了什么，必须把它上一轮的回答以 assistant 的身份加回 messages 列表再发过去。这是让多轮对话成立的唯一方式。

{"role": "assistant", "content": "def fib(n): return n if n < 2 else fib(n-1) + fib(n-2)"}

把这三种角色按顺序排列，就构成了一次多轮对话的完整上下文：

messages = [
    {"role": "system", "content": "你是一位简洁、务实的 Python 老师。回答不超过三句话。"},
    {"role": "user", "content": "什么是列表推导式？"},
    {"role": "assistant", "content": "列表推导式是一种用紧凑语法从可迭代对象生成新列表的写法，形如 [表达式 for 元素 in 可迭代对象]。它比 for 循环加 append 更 Pythonic，但嵌套超过两层就该考虑拆成普通循环。"},
    {"role": "user", "content": "再给我一个例子"},
]

顺带一提：后来新增的角色

协议发展过程中又加了两种角色，你先知道它们存在，用到的时候再回来看：

• tool（早期叫 function）：模型调用外部工具之后，工具的返回值以这个角色回填进对话。第 07 篇讲 Function Calling 时会详细用到
• developer：OpenAI 在 o1 系列推理模型之后引入，作用类似 system 但优先级更高。非 OpenAI 家的模型目前大多还是三角色制，可以先不管

无状态的重要推论

这里有一个对新手非常反直觉的点：LLM API 本身是无状态的。模型不会"记住"你们之前聊过什么，每次请求你都必须把完整的历史传进去。表面上看 ChatGPT 能连续聊天，实际上前端每次都在把所有消息重新打包发送。这个事实有两个直接后果：

1. 会话越长，每次请求的 Token 成本越高，因为历史全都要重传
2. 所谓"记忆"功能都是应用层做的——把历史存到数据库、摘要压缩、或者用向量检索提取相关片段

后面的 Agent 和 RAG 实际上都可以理解成"更聪明地决定这个 messages 列表里放什么"。

换用 openai SDK：代码瞬间清爽

原生 requests 的写法每次都要拼 header、处理 JSON、捕获异常，啰嗦。官方有一个 openai Python SDK，虽然名字里有 OpenAI，但它接受自定义的 base_url，所以可以用来调任何兼容协议的服务：

# hello_sdk.py
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

resp = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "system", "content": "你是一位简洁的 Python 老师。"},
        {"role": "user", "content": "什么是装饰器？"},
    ],
)

print(resp.choices[0].message.content)
print(f"消耗 Token：{resp.usage.total_tokens}")

和裸 requests 版本对比，减少了十几行样板代码，还多了类型提示、自动重试和参数校验。整个系列后面都用这种写法。想切换模型后端时，只改 base_url 和 api_key：

# 换到 OpenAI 官方
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))  # base_url 默认就是 OpenAI

# 换到本地 Ollama
client = OpenAI(api_key="ollama", base_url="http://localhost:11434/v1")

# 换到 Qwen
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)

代码主体完全不动。这就是 OpenAI 协议事实标准带来的便利。

流式输出：为什么要流式

你用 ChatGPT 时看到的"一个字一个字往外蹦"的效果叫流式输出（streaming）。它不是装饰效果，有两个实质性原因：

一是体验。LLM 生成一段几百字的回答通常要 3~10 秒，如果用户盯着一个空白页面等 8 秒才看到完整结果，感知上非常慢；而第一个字在 500 毫秒内出现，后续逐字追加，感知延迟会大幅下降。

二是可中断。流式允许前端在读到一半时让用户取消。非流式一旦发出请求就必须等到模型生成完整输出后一次返回。对于 Agent 这类可能跑几十秒甚至更长的场景，流式几乎是必需的。

底层协议是 Server-Sent Events (SSE)：服务器保持 HTTP 连接不断开，持续往里推送 data: {...}\n\n 这样的文本片段，直到发出 data: [DONE] 为止。手写 SSE 解析比较麻烦，SDK 会帮你做好。

用 SDK 实现流式

把 stream=True 打开，返回值就从一个完整的响应对象变成一个可迭代的流：

# hello_stream.py
import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

stream = client.chat.completions.create(
    model="deepseek-chat",
    messages=[
        {"role": "user", "content": "用三段话介绍一下 asyncio 的基本原理"},
    ],
    stream=True,
)

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

注意两个容易踩的点：

• 每个 chunk 的内容在 choices[0].delta.content 里（不是 message.content），而且可能是 None（比如第一个 chunk 只包含 role 信息，没有文本）
• print 必须加 flush=True，否则 Python 的输出缓冲会让你看到的"流式"其实是攒了一批才一起输出，不流畅

跑一下，你能看到明显的逐字输出效果。这就是 ChatGPT 的全部秘密——没有任何魔法，就是 SSE 加一点前端渲染。

异步版本：应对高并发

以上都是同步代码，单个请求没问题。但如果你要同时发起几十个请求（比如批量处理文档、并行调用多个模型比较结果），同步就扛不住了——每个请求要几秒，串行跑完需要几分钟。这时候用 asyncio + AsyncOpenAI：

# hello_async.py
import asyncio
import os
from openai import AsyncOpenAI
from dotenv import load_dotenv

load_dotenv()

client = AsyncOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

async def ask(question: str) -> str:
    resp = await client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": question}],
    )
    return resp.choices[0].message.content

async def main():
    questions = [
        "用一句话解释什么是 GIL",
        "用一句话解释什么是事件循环",
        "用一句话解释什么是协程",
    ]
    results = await asyncio.gather(*[ask(q) for q in questions])
    for q, r in zip(questions, results):
        print(f"Q: {q}\nA: {r}\n")

asyncio.run(main())

三个请求并行发出，总耗时约等于最慢那一个，而不是三个相加。实际工程中处理百万级文档都靠这套模式。如果你还没学过 async/await，现在只要知道它存在，等我们做 RAG 批量向量化的时候再深入。

错误处理与重试

生产环境下你会遇到这几类错误，都要能优雅处理：

• RateLimitError——调用频率超限。应该退避后重试
• APITimeoutError——网络超时。同上
• APIConnectionError——网络断开。同上
• BadRequestError——你传了非法参数。不应该重试，直接报错修代码
• AuthenticationError——API Key 错误。同上不应重试

openai SDK 内部已经做了最多 2 次的自动重试（针对可重试的错误）。生产场景建议在外面再包一层用 tenacity 做的退避逻辑：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import RateLimitError, APITimeoutError, APIConnectionError

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    retry=retry_if_exception_type((RateLimitError, APITimeoutError, APIConnectionError)),
)
def robust_chat(messages):
    return client.chat.completions.create(model="deepseek-chat", messages=messages)

这段配置的含义是：对指定类型的错误最多重试 5 次，间隔按指数退避 2s、4s、8s……最多 30s。对不该重试的错误（参数错、Key 错）让它直接抛出，别浪费重试次数。

一个可复用的封装

把前面的要点合并成一个简洁的模块，后面几篇会直接 from ai import chat 用它：

# ai.py
import os
from typing import Iterable
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

_client = OpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
)

DEFAULT_MODEL = "deepseek-chat"


def chat(messages: list[dict], model: str = DEFAULT_MODEL, **kwargs) -> str:
    """非流式调用，返回完整字符串。"""
    resp = _client.chat.completions.create(model=model, messages=messages, **kwargs)
    return resp.choices[0].message.content


def chat_stream(messages: list[dict], model: str = DEFAULT_MODEL, **kwargs) -> Iterable[str]:
    """流式调用，yield 文本增量。"""
    stream = _client.chat.completions.create(
        model=model, messages=messages, stream=True, **kwargs
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield delta

用起来就非常干净：

from ai import chat, chat_stream

# 一次性拿完整回答
print(chat([{"role": "user", "content": "你好"}]))

# 流式打印
for piece in chat_stream([{"role": "user", "content": "背一首李白的诗"}]):
    print(piece, end="", flush=True)

本篇的几个关键点

• 所有大模型调用本质是一次 HTTP POST，响应结构由 OpenAI 协议规定，已成事实标准
• messages 协议有 system、user、assistant 三种角色，API 本身无状态，每次请求必须传完整历史
• 流式输出基于 SSE，打开 stream=True 后迭代响应，每个 chunk 的文本在 delta.content 里
• 异步调用用 AsyncOpenAI，批处理时能把几十秒压缩成几秒
• 生产环境要处理 RateLimitError 等可恢复错误，SDK 自带重试，建议再叠一层 tenacity 退避
• API Key 一律走 .env，永远不要提交到 Git

下一篇

第 03 篇会进入 Prompt 工程。很多人第一次用 LLM 都会发现：同样一个问题换种问法，效果差很多；问两次得到的回答经常不一致；让它输出 JSON 它偶尔还会塞一些解释文字。这些问题的根源都在 Prompt。下一篇会讲清楚怎么写稳定、结构化、可复用的 Prompt，以及哪些坑不是 Prompt 能解决的——需要用到第 04 篇的 Pydantic 结构化输出。

参考资料

• OpenAI API Reference
• DeepSeek API 文档
• openai-python SDK
• python-dotenv
• tenacity 重试库
• Python asyncio 官方教程
• Server-Sent Events 规范 (MDN)