第一章：第一次 API 调用

我们要造一个旅游 Agent。它能帮你查航班、推荐行程、订酒店、做预算——但这些都是后面的事。这一章只做一件事：发一个请求给大模型，拿到回复。

说白了，不管后面的 Agent 架构多花哨，底层都是一次又一次的 HTTP 请求。把这一次请求摸透了，后面才不会被框架的抽象挡住视线。

---

1.1 环境准备

先把依赖装好。我们需要三样东西：requests 用来手动发 HTTP 请求，openai 是官方 SDK，python-dotenv 管环境变量。

pip install requests openai python-dotenv

然后在项目根目录建一个 .env 文件，把你的 API Key 放进去：

OPENAI_API_KEY=sk-proj-xxxxxxxxxxxxxxxxxxxxxxxx

写一段验证代码，确认环境没问题：

import os
from dotenv import load_dotenv

load_dotenv()

api_key = os.getenv("OPENAI_API_KEY")
if api_key and api_key.startswith("sk-"):
    print("环境准备完成")
else:
    print("请检查 .env 文件中的 OPENAI_API_KEY")

环境准备完成

这段代码做的事情很直接：从 .env 文件加载环境变量，检查 OPENAI_API_KEY 是否存在且格式正确。python-dotenv 的 load_dotenv() 会把 .env 里的键值对注入 os.environ，这样后面的代码就能通过 os.getenv 拿到 Key，而不用把密钥硬编码在源文件里。

1.1.1 Messages 的结构

在发请求之前，先搞清楚一个核心概念：messages。

你可能会觉得，调用大模型不就是"给它一句话，它回一句话"吗？如果只是单轮对话，确实可以这么理解。但实际上 API 的设计比这复杂一点——你发给模型的不是一个字符串，而是一个消息数组。

为什么是数组？因为真实的对话不是一问一答就完事的。用户可能会问："帮我规划杭州一日游"，模型回答之后用户接着问："那住哪个酒店好？"这时候你得把前面的对话历史一起发过去，模型才知道你在聊杭州。所以 messages 必须是数组，每条消息都带着一个 role（角色）和一个 content（内容）。

目前有四种 role：

system — 开发者给模型的指令。用户看不到，但模型会遵守。你可以在这里设定模型的人设、限制它的行为范围、规定输出格式。比如我们的旅行 Agent，可以用 system message 告诉模型"你是一个旅行助手，不要编造实时信息"。

user — 用户说的话。就是最终用户输入的内容。

assistant — 模型的回复。在多轮对话里，你要把模型之前的回复也放进 messages 里，这样模型才有上下文。

tool — 工具返回的结果。这个我们在第二章工具调用的时候详细讲，现在知道有这么个角色就行。

来看一个带 system message 的例子：

messages = [
    {
        "role": "system",
        "content": "你是一个谨慎的旅行助手。不要编造实时信息，如果不确定就说不知道。",
    },
    {
        "role": "user",
        "content": "帮我规划一个轻松的杭州一日游。",
    },
]

这个数组有两条消息。第一条是 system message，定了规矩——"你是旅行助手，别瞎编"。第二条是用户的实际问题。模型在生成回复的时候，会同时参考这两条消息。

用 curl 试一下带 system message 的请求：

curl https://api.openai.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [
      {
        "role": "system",
        "content": "你是一个谨慎的旅行助手。不要编造实时信息，如果不确定就说不知道。"
      },
      {
        "role": "user",
        "content": "帮我规划一个轻松的杭州一日游。"
      }
    ]
  }'

{
  "id": "chatcmpl-CXr2aK9m1NRwzs8tHb3iYfQg5mNeB",
  "object": "chat.completion",
  "created": 1715000100,
  "model": "gpt-4o-mini-2024-07-18",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "好的，这是一个轻松的杭州一日游建议：\n\n上午可以去西湖边走走，沿着白堤或苏堤慢慢散步，不用绕一整圈，走一小段感受一下就够了。中午去河坊街附近吃点杭帮菜或小吃。下午可以去龙井村一带喝茶，找一家茶农的院子坐下来，点一杯龙井慢慢喝。\n\n不过有几点我不太确定，需要你自己核实：\n- 具体的营业时间和价格可能有变化，建议出发前查一下最新信息\n- 节假日人会很多，如果是周末去可能需要早一点出发\n\n需要我再展开某个部分吗？"
      },
      "logprobs": null,
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 52,
    "completion_tokens": 156,
    "total_tokens": 208
  },
  "system_fingerprint": "fp_0ba0d124f1"
}

注意到了吗？因为 system message 说了"不要编造实时信息，如果不确定就说不知道"，模型在回复里主动加了免责说明——"具体的营业时间和价格可能有变化"。这就是 system message 的作用：它不会出现在回复里，但会影响模型的行为方式。

另外看 usage，prompt_tokens 从之前的 38 变成了 52，多出来的就是 system message 消耗的 token。system message 也是要计费的，这点别忘了。

---

1.2 用 curl 发第一个请求

最原始的方式——直接用 curl 往 OpenAI 的 Chat Completions 端点发一个 POST 请求：

curl https://api.openai.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [
      {
        "role": "user",
        "content": "帮我规划一个轻松的杭州一日游。我喜欢茶、本地小吃、不要走太累。"
      }
    ]
  }'

返回的完整 JSON 长这样：

{
  "id": "chatcmpl-BVq8kT3m0RNwzs7rGa2hXePf4lMdA",
  "object": "chat.completion",
  "created": 1715000000,
  "model": "gpt-4o-mini-2024-07-18",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "这是一份适合喜欢茶和本地小吃的轻松杭州一日游安排：\n\n🌅 上午（9:00–11:30）\n\n先去龙井村走走。从龙井问茶的入口进去，沿茶园小道慢慢逛，走累了就找一家茶农自家的院子坐下来，点一杯现炒的龙井新茶。不用赶景点，坐着看茶山发呆就挺好。茶叶可以买一点带回去，但不用在景区门口买，价格虚高。\n\n🍜 午餐（11:30–13:00）\n\n从龙井村打车到河坊街一带。午餐推荐知味观总店，点几样杭帮菜经典：西湖醋鱼、东坡肉、龙井虾仁。如果不想吃正餐，河坊街上也有很多小吃可以边走边吃——葱包桧、定胜糕、吴山酥油饼都是本地老味道。\n\n🌿 下午（13:30–16:00）\n\n沿着南山路慢慢走到太子湾公园或者花港观鱼，不用绕西湖一整圈，走一小段湖边就够了。如果季节对的话太子湾的花很好看。走累了找个湖边的咖啡馆或茶室坐一坐。\n\n🍵 傍晚（16:00–17:30）\n\n去梅家坞喝下午茶。梅家坞整条路都是茶室，随便挑一家坐下来，一壶龙井、几碟茶点，看着远处的茶山，节奏放到最慢。\n\n这个安排全程不赶路，打车加步行为主，不会太累。重点就是两个字：喝茶。"
      },
      "logprobs": null,
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 38,
    "completion_tokens": 312,
    "total_tokens": 350
  },
  "system_fingerprint": "fp_0ba0d124f1"
}

这个 JSON 有不少字段，逐个说清楚。

id — 这次请求的唯一标识。每次调用都会生成一个不同的 ID，格式是 chatcmpl- 加一串随机字符。排查问题的时候靠它定位，比如给 OpenAI 提工单时贴上这个 ID 他们就能查到对应的日志。

object — 固定是 "chat.completion"。这是 OpenAI API 的惯例——每种资源都有个 object 字段标明类型，比如后面讲流式的时候会看到 "chat.completion.chunk"。其实就是个类型标签。

created — Unix 时间戳，秒级。1715000000 大约是 2024 年 5 月 6 日。拿 Python 的 datetime.fromtimestamp() 就能转成人类可读的时间。

model — 实际执行请求的模型版本。注意这里写的是 gpt-4o-mini-2024-07-18，而我们请求时写的是 gpt-4o-mini。因为 gpt-4o-mini 是一个别名，指向最新的稳定版本。OpenAI 会定期更新这个别名指向的具体版本，所以如果你对输出稳定性有要求，应该在请求里写死完整版本号。

choices — 一个数组。为什么是数组？因为你可以在请求里设置 n=3，让模型一次返回 3 个不同的回复。默认 n=1，所以数组里只有一个元素。每个元素有这几个字段：

• index：这是第几个回复，从 0 开始。
• message.role：回复的角色，这里是 "assistant"。
• message.content：模型生成的文本内容，也就是我们真正要用的东西。
• logprobs：token 级别的对数概率。默认 null，需要在请求里显式开启 logprobs: true 才会返回。做模型评估或不确定性估算的时候有用。
• finish_reason：模型停下来的原因。一共三种可能：
• "stop" — 正常结束，模型自己觉得说完了。
• "length" — 撞到了 max_tokens 上限，被强行截断。如果你看到这个值，说明回复被切了一半，需要调大 max_tokens。
• "tool_calls" — 模型想调用工具。这个我们在第三章工具调用的时候会详细讲。

usage — 这是计费相关的数据。

• prompt_tokens：输入消耗的 token 数。也就是你发给模型的 messages 被 tokenize 之后有多少个 token。这个例子是 38，对应我们那句中文提示。
• completion_tokens：输出消耗的 token 数。模型生成的回复占了 312 个 token。
• total_tokens：就是 prompt_tokens + completion_tokens。

关键是：OpenAI 按 token 计费，而且输入和输出的单价不一样——输出通常是输入的 3-4 倍。为什么输出更贵？这个问题在附录 1A 讲 KV Cache 的时候会解释。

system_fingerprint — 服务端的配置指纹。同一个模型，不同的系统配置（比如量化方式、推理引擎版本）可能给出不同的结果。这个指纹就是用来追踪"到底跑在哪套配置上"的。如果你发现同一个 prompt 今天和昨天的输出风格不一样，对比一下 fingerprint 就能知道是不是后端配置变了。

1.2.1 常用请求参数

到目前为止，我们的请求体只有 model 和 messages 两个字段。其实还有一些参数可以控制模型的行为，挑几个最常用的说一下。

temperature（范围 0-2，默认 1）— 控制输出的随机性。值越低，模型越倾向于选概率最高的 token，输出越确定、越保守；值越高，模型越愿意冒险选概率较低的 token，输出越多样、越"有创意"。

• temperature=0：几乎每次给同样的输入都会得到一样的输出。适合需要稳定性的场景。
• temperature=1：默认值，平衡创造性和一致性。
• temperature>1：更随机，可能说出一些意想不到的东西，但也更容易跑偏。

对于我们的旅行 Agent，建议用 0 到 0.3。规划行程需要靠谱，不需要太多"创意发挥"——你不会希望模型突然推荐你去一个不存在的餐厅。

max_tokens — 模型最多生成多少个 token。不设的话，模型自己决定什么时候停（遇到自然结束点就停）。如果设了，生成到这个数量就强制截断，哪怕话说到一半。被截断的时候，响应里的 finish_reason 会是 "length" 而不是 "stop"。

用处是什么？控制成本和响应时间。如果你只需要一个简短的回答，设一个合理的上限可以省钱。但设太小会导致回复不完整。

top_p（范围 0-1，默认 1）— 核采样。模型从概率最高的 token 里挑，直到累积概率达到 top_p。比如 top_p=0.1 意味着只考虑概率排名前 10% 的 token。

一个重要的原则：temperature 和 top_p 不要同时调。OpenAI 官方文档也这么建议。两个参数都影响采样，同时改容易产生不可预测的效果。一般调一个就够了，大多数情况下用 temperature 就行。

来看一个带这些参数的请求：

curl https://api.openai.com/v1/chat/completions \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [
      {
        "role": "system",
        "content": "你是一个谨慎的旅行助手。"
      },
      {
        "role": "user",
        "content": "用一句话推荐杭州最值得去的地方。"
      }
    ],
    "temperature": 0,
    "max_tokens": 500
  }'

{
  "id": "chatcmpl-DYs3bL0n2OSxzt9uIc4jZgRh6oOfC",
  "object": "chat.completion",
  "created": 1715000200,
  "model": "gpt-4o-mini-2024-07-18",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "西湖是杭州最值得去的地方，湖光山色四季皆美，无论是沿着苏堤漫步还是泛舟湖上，都能感受到这座城市最经典的韵味。"
      },
      "logprobs": null,
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 30,
    "completion_tokens": 42,
    "total_tokens": 72
  },
  "system_fingerprint": "fp_0ba0d124f1"
}

注意 finish_reason 是 "stop"——模型自己说完了，没有被 max_tokens 截断。如果你把 max_tokens 改成 10，回复会在半句话的地方断掉，finish_reason 就会变成 "length"。

1.2.2 Token 到底是什么

前面一直在说 token，但 token 到底是什么？很多人以为一个中文字就是一个 token，或者一个英文单词就是一个 token。实际上都不准确。

Token 是模型的最小处理单位。模型不是按字或按词来理解文本的，而是按 token。每个模型都有自己的 tokenizer（分词器），它会把输入文本切成一个个 token。有些字会被切成一个 token，有些字会被切成两个，有些常见的词组会被合成一个 token。

大致的经验值：

• 中文：1 个字大约 1-2 个 token
• 英文：1 个单词大约 1-1.5 个 token
• 标点符号：通常 1 个 token
• 空格和换行：也占 token

不过这些只是粗略估算。想要准确知道一段文本有多少 token，得用对应模型的 tokenizer 实际数一数。OpenAI 的模型用的 tokenizer 叫 tiktoken，我们可以直接用它：

import tiktoken

enc = tiktoken.encoding_for_model("gpt-4o-mini")

text = "帮我规划一个轻松的杭州一日游"
tokens = enc.encode(text)

print(f"文本: {text}")
print(f"Token 数: {len(tokens)}")
print(f"Token 列表: {tokens}")
print(f"解码回去: {[enc.decode([t]) for t in tokens]}")

文本: 帮我规划一个轻松的杭州一日游
Token 数: 10
Token 列表: [33080, 45895, 83978, 9370, 105043, 21756, 68014, 9370, 33404, 28946]
解码回去: ['帮', '我', '规划', '一个', '轻松', '的', '杭州', '一', '日', '游']

看到了吧？"规划"是一个 token，"杭州"也是一个 token——这些常见的双字词被 tokenizer 合在了一起。而"一日游"被拆成了三个 token："一"、"日"、"游"。总共 14 个中文字，对应 10 个 token。

为什么你需要关心 token 数？两个原因：

1. 计费按 token 算，不按字数算。知道 token 数才能估算成本。
2. 模型有上下文长度限制。比如 gpt-4o-mini 的上下文窗口是 128K token，输入+输出加起来不能超过这个数。如果你要把很长的文档塞进 messages，得先算一下 token 数够不够。

安装 tiktoken 很简单：

pip install tiktoken

Successfully installed tiktoken-0.7.0

---

1.3 用 Python requests 做同样的事

curl 能跑通，但没人会在正式代码里一直调 curl。换成 Python 的 requests 库来做同样的事：

import os
import json
import requests
from dotenv import load_dotenv

load_dotenv()

url = "https://api.openai.com/v1/chat/completions"

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
}

payload = {
    "model": "gpt-4o-mini",
    "messages": [
        {
            "role": "user",
            "content": "帮我规划一个轻松的杭州一日游。我喜欢茶、本地小吃、不要走太累。",
        }
    ],
}

response = requests.post(url, headers=headers, json=payload)
data = response.json()

# 打印完整响应（格式化）
print(json.dumps(data, indent=2, ensure_ascii=False))

# 提取回复文本
reply = data["choices"][0]["message"]["content"]
print("\n--- 模型回复 ---")
print(reply)

# 查看 token 用量
usage = data["usage"]
print(f"\n输入 token: {usage['prompt_tokens']}")
print(f"输出 token: {usage['completion_tokens']}")
print(f"总计 token: {usage['total_tokens']}")

{
  "id": "chatcmpl-BVq8kT3m0RNwzs7rGa2hXePf4lMdA",
  "object": "chat.completion",
  "created": 1715000000,
  "model": "gpt-4o-mini-2024-07-18",
  "choices": [
    {
      "index": 0,
      "message": {
        "role": "assistant",
        "content": "这是一份适合喜欢茶和本地小吃的轻松杭州一日游安排：\n\n🌅 上午..."
      },
      "logprobs": null,
      "finish_reason": "stop"
    }
  ],
  "usage": {
    "prompt_tokens": 38,
    "completion_tokens": 312,
    "total_tokens": 350
  },
  "system_fingerprint": "fp_0ba0d124f1"
}

--- 模型回复 ---
这是一份适合喜欢茶和本地小吃的轻松杭州一日游安排：

🌅 上午（9:00–11:30）
先去龙井村走走...

输入 token: 38
输出 token: 312
总计 token: 350

这段代码和 curl 做的事情完全一样——构造 HTTP POST 请求，带上 Authorization 头和 JSON 请求体，发到 https://api.openai.com/v1/chat/completions。区别只是从命令行搬到了 Python 里。requests.post() 的 json=payload 参数会自动帮你做 JSON 序列化并设置 Content-Type（虽然我们已经手动设了）。返回值用 response.json() 就能直接拿到解析好的字典，不用自己 json.loads。

其实跟 curl 对比一下就很清楚：curl 的 -H 对应 headers 字典，-d 对应 payload 字典，response.json() 对应 curl 输出到终端的那堆 JSON。底层是完全相同的 HTTP 请求。

---

1.4 用 OpenAI SDK 简化

手动拼 HTTP 请求当然可以，但 OpenAI 官方提供了 Python SDK，把这些细节都封装好了：

import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

client = OpenAI()  # 自动从环境变量读取 OPENAI_API_KEY

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "user",
            "content": "帮我规划一个轻松的杭州一日游。我喜欢茶、本地小吃、不要走太累。",
        }
    ],
)

# 提取回复
reply = response.choices[0].message.content
print(reply)

# token 用量
print(f"\n输入 token: {response.usage.prompt_tokens}")
print(f"输出 token: {response.usage.completion_tokens}")
print(f"总计 token: {response.usage.total_tokens}")

这是一份适合喜欢茶和本地小吃的轻松杭州一日游安排：

🌅 上午（9:00–11:30）
先去龙井村走走。从龙井问茶的入口进去，沿茶园小道慢慢逛...

🍜 午餐（11:30–13:00）
从龙井村打车到河坊街一带...

🌿 下午（13:30–16:00）
沿着南山路慢慢走到太子湾公园...

🍵 傍晚（16:00–17:30）
去梅家坞喝下午茶...

输入 token: 38
输出 token: 312
总计 token: 350

代码从十几行缩到了六七行。SDK 到底帮你干了什么？

认证管理 — OpenAI() 初始化时自动去环境变量找 OPENAI_API_KEY，不用手动拼 Authorization 头。你也可以显式传 OpenAI(api_key="sk-...")，但不推荐把密钥写在代码里。

序列化和反序列化 — 你传进去的是 Python 字典和字符串，SDK 自动把它们转成 JSON 请求体。返回值也不是原始字典了，而是 Pydantic 模型对象——所以你写 response.choices[0].message.content 而不是 data["choices"][0]["message"]["content"]。对象属性比字典取值的好处是：有类型提示、有自动补全、拼错属性名会直接报 AttributeError 而不是默默返回 None。

重试机制 — 如果遇到 429 Too Many Requests（限速）或 500 Internal Server Error（服务端临时故障），SDK 会自动用指数退避策略重试，默认最多重试 2 次。手动用 requests 的话，这些逻辑要自己写。

错误类型 — SDK 把不同的 HTTP 错误码映射成了具体的异常类：AuthenticationError（401，Key 无效）、RateLimitError（429，限速）、BadRequestError（400，参数有误）等等。用 try/except 捕获这些异常比检查 response.status_code 要清晰得多。

三种方式放在一起对比：

	curl	requests	OpenAI SDK
认证	手动拼 -H	手动拼 headers 字典	自动读环境变量
请求体	手写 JSON 字符串	Python 字典，自动序列化	函数参数，自动序列化
响应解析	原始 JSON 文本	response.json() → 字典	Pydantic 对象，有类型提示
重试	没有	需要自己写	内置指数退避重试
错误处理	看 HTTP 状态码	看 status_code	具体异常类
适合场景	快速调试、一次性测试	需要自定义底层行为	日常开发，大多数场景

后面的章节我们会一直用 SDK，因为它最省事。但你已经知道 SDK 底下就是一个 HTTP POST 请求，没有任何黑魔法。

1.4.1 错误处理

实际开发中，API 调用不会每次都成功。网络可能抖动，Key 可能过期，请求可能太频繁。来看看常见的错误和怎么处理。

401 Unauthorized — API Key 无效

如果你的 Key 写错了、过期了、或者被撤销了，API 会返回 401：

{
  "error": {
    "message": "Incorrect API key provided: sk-proj-xxxx...xxxx. You can find your API key at https://platform.openai.com/account/api-keys.",
    "type": "invalid_api_key",
    "param": null,
    "code": "invalid_api_key"
  }
}

这个最好排查——检查 .env 文件里的 Key 有没有多余的空格，有没有复制完整。

429 Too Many Requests — 超过速率限制

每个 API Key 都有速率限制（RPM: 每分钟请求数，TPM: 每分钟 token 数）。超了就会收到 429：

{
  "error": {
    "message": "Rate limit reached for gpt-4o-mini in organization org-xxxxx on requests per min (RPM): Limit 500, Used 500, Requested 1. Please try again in 120ms.",
    "type": "requests",
    "param": null,
    "code": "rate_limit_exceeded"
  }
}

响应头里通常会带一个 retry-after 字段，告诉你等多久再重试。

用 SDK 处理这些错误

SDK 已经帮你映射好了异常类型，用 try/except 就能优雅地处理：

from openai import OpenAI, AuthenticationError, RateLimitError, APIError

client = OpenAI()

try:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "user", "content": "你好"}
        ],
    )
    print(response.choices[0].message.content)

except AuthenticationError:
    print("API Key 无效，请检查 .env 文件")

except RateLimitError as e:
    wait_time = e.response.headers.get("retry-after", 60)
    print(f"超过速率限制，{wait_time} 秒后重试")

except APIError as e:
    print(f"API 错误: {e.status_code} - {e.message}")

你好！有什么可以帮你的吗？

正常情况下不会走到 except 分支。但如果你把 Key 改成一个无效值，输出就会变成：

API Key 无效，请检查 .env 文件

前面提过，SDK 有内置的重试机制——遇到 429 或 500 会自动用指数退避重试。所以大多数情况下你甚至不需要自己写重试逻辑。但如果你想自定义重试行为，可以在初始化时设置：

client = OpenAI(
    max_retries=5,       # 最多重试 5 次（默认 2 次）
    timeout=30.0,        # 超时时间 30 秒
)

# 正常输出，和前面一样

一个实用的建议：在开发阶段就把错误处理写好。不要等到线上出了问题才去加 try/except。旅行 Agent 面向终端用户，"API Key 无效"这种技术错误不应该直接暴露给用户，你应该把它转成一句友好的提示。

---

1.5 SSE 与 Streaming

到目前为止，我们发一个请求，等模型把整段文本全部生成完，然后一次性拿回来。这有一个体验上的问题——如果回复比较长（比如一份详细的旅游行程），你可能要干等 3-10 秒才能看到第一个字。

Streaming 就是为了解决这个问题。

1.5.1 SSE 是什么

SSE，全称 Server-Sent Events，是一种让服务器单向推送数据给客户端的协议。它跑在普通的 HTTP 上，不需要额外的协议升级。

和 WebSocket 对比一下就清楚了：

	SSE	WebSocket
方向	服务器→客户端（单向）	双向
协议	普通 HTTP	需要协议升级（Upgrade 头）
重连	内置自动重连	需要手动实现
数据格式	纯文本（data: 行）	二进制或文本

LLM API 用 SSE 而不用 WebSocket，原因很简单：文本生成本质上就是一个单向流——服务器一个 token 一个 token 地往外吐，客户端只管接收。不需要客户端在生成过程中往回发消息，所以双向通信完全是多余的。SSE 用普通 HTTP 就能搞定，部署上也更简单——不用担心代理服务器和 CDN 对 WebSocket 的支持问题。

1.5.2 什么是 Streaming

Streaming 的核心诉求是降低首字延迟（TTFB，Time to First Byte）。

不用 Streaming：客户端发请求 → 服务器生成全部 token → 一次性返回。如果生成 300 个 token 需要 5 秒，你就得等 5 秒才能看到任何内容。

用 Streaming：客户端发请求 → 服务器生成第一个 token 就立刻推送 → 后续 token 陆续到达。第一个 token 通常在 200ms 左右就能到，用户几乎感觉是"实时打字"。

其实总时间差不多——该生成多少 token 还是多少，该花多久还是多久。但用户的体感完全不同。等 5 秒看到一堆文字，和看着文字一个一个冒出来，心理感受差距很大。

1.5.3 手动解析 SSE

我们先用 requests 手动处理 SSE 流，把底层协议看清楚：

import os
import json
import requests
from dotenv import load_dotenv

load_dotenv()

url = "https://api.openai.com/v1/chat/completions"

headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}",
}

payload = {
    "model": "gpt-4o-mini",
    "messages": [
        {
            "role": "user",
            "content": "帮我规划一个轻松的杭州一日游。我喜欢茶、本地小吃、不要走太累。",
        }
    ],
    "stream": True,  # 开启流式
}

response = requests.post(url, headers=headers, json=payload, stream=True)

full_content = ""

for line in response.iter_lines(decode_unicode=True):
    if not line:
        # SSE 协议中，空行是事件分隔符
        continue

    if line.startswith("data: "):
        data_str = line[len("data: "):]

        if data_str == "[DONE]":
            # 流结束信号
            print("\n\n--- 流结束 ---")
            break

        chunk = json.loads(data_str)
        delta = chunk["choices"][0]["delta"]

        if "content" in delta:
            text = delta["content"]
            print(text, end="", flush=True)
            full_content += text

print(f"\n完整回复长度: {len(full_content)} 字符")

这是一份适合喜欢茶和本地小吃的轻松杭州一日游安排：

🌅 上午（9:00–11:30）
先去龙井村走走。从龙井问茶的入口进去，沿茶园小道慢慢逛...
...

--- 流结束 ---
完整回复长度: 587 字符

来看看 SSE 流的原始格式到底长什么样。每一行以 data: 开头，后面跟一个 JSON 对象，事件之间用空行隔开：

data: {"id":"chatcmpl-BVq8kT3m0RNwzs7rGa2hXePf4lMdA","object":"chat.completion.chunk","created":1715000000,"model":"gpt-4o-mini-2024-07-18","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":null}]}

data: {"id":"chatcmpl-BVq8kT3m0RNwzs7rGa2hXePf4lMdA","object":"chat.completion.chunk","created":1715000000,"model":"gpt-4o-mini-2024-07-18","choices":[{"index":0,"delta":{"content":"这"},"finish_reason":null}]}

data: {"id":"chatcmpl-BVq8kT3m0RNwzs7rGa2hXePf4lMdA","object":"chat.completion.chunk","created":1715000000,"model":"gpt-4o-mini-2024-07-18","choices":[{"index":0,"delta":{"content":"是"},"finish_reason":null}]}

data: {"id":"chatcmpl-BVq8kT3m0RNwzs7rGa2hXePf4lMdA","object":"chat.completion.chunk","created":1715000000,"model":"gpt-4o-mini-2024-07-18","choices":[{"index":0,"delta":{"content":"一份"},"finish_reason":null}]}

...

data: {"id":"chatcmpl-BVq8kT3m0RNwzs7rGa2hXePf4lMdA","object":"chat.completion.chunk","created":1715000000,"model":"gpt-4o-mini-2024-07-18","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

和非流式响应对比，有几个关键区别：

object 从 "chat.completion" 变成了 "chat.completion.chunk" — 表示这是流的一个片段，不是完整响应。

message 变成了 delta — 非流式里是完整的 message（包含 role 和 content），流式里每个 chunk 只有增量内容 delta。第一个 chunk 的 delta 里会有 role: "assistant"，后面的 chunk 只有 content。

没有 usage 字段 — 因为生成还没结束，不知道最终用了多少 token。如果你需要流式也返回 usage，可以在请求里加 "stream_options": {"include_usage": true}，这样最后一个 chunk 会带上 usage 信息。

最后一个有效 chunk 的 delta 是空对象 {}，finish_reason 是 "stop" — 表示模型正常结束了。然后收到 data: [DONE]，这不是 JSON，是 SSE 层面的结束信号。

把这些区别整理成一张表，方便对比：

	非流式	流式
object	chat.completion	chat.completion.chunk
文本字段	message.content（完整文本）	delta.content（增量片段）
finish_reason	在 choices[0] 里，跟文本一起返回	在最后一个 chunk 的 choices[0] 里
usage	默认包含	默认不包含，需要 stream_options: {"include_usage": true}

代码里的处理逻辑很简单：requests.post() 加了 stream=True 参数，这样它不会等整个响应下载完才返回，而是立即返回一个可以逐行迭代的对象。iter_lines() 按行读取，跳过空行，去掉 data: 前缀，解析 JSON，把 delta.content 拼起来。碰到 [DONE] 就停。

1.5.4 用 SDK 的 stream=True

手动解析 SSE 是为了让你理解底层在发生什么。实际开发中，用 SDK 就行了：

import os
from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

client = OpenAI()

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {
            "role": "user",
            "content": "帮我规划一个轻松的杭州一日游。我喜欢茶、本地小吃、不要走太累。",
        }
    ],
    stream=True,
)

full_content = ""

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

print(f"\n\n完整回复长度: {len(full_content)} 字符")

这是一份适合喜欢茶和本地小吃的轻松杭州一日游安排：

🌅 上午（9:00–11:30）
先去龙井村走走。从龙井问茶的入口进去，沿茶园小道慢慢逛...
...

完整回复长度: 587 字符

对比手动解析的版本，SDK 版省掉了所有 SSE 协议层的处理：不用按行切分、不用去掉 data: 前缀、不用手动 json.loads、不用判断 [DONE]。stream=True 让返回值变成一个可迭代对象，每次迭代拿到的就是一个解析好的 chunk 对象，直接访问 chunk.choices[0].delta.content 就行。

SDK 还帮你处理了一些边界情况——比如网络中断时的重试、不完整行的缓冲等等。这些用 requests 手动做的话都得额外写代码。

---

小结

这一章只做了一件事：给大模型发一个请求，拿到回复。

我们用了三种方式（curl、requests、SDK）做同一件事，从最底层的 HTTP 请求一路到封装好的 SDK 调用。还看了 SSE 流式传输的原始格式和 SDK 封装。

现在你知道了：LLM API 的本质就是一个 HTTP POST 请求。请求体里放消息列表，响应里拿到模型生成的文本。Streaming 不改变结果，只改变交付方式。

但这里有一个明显的问题——我们每次调用都只发了一条消息。如果用户问完杭州行程后接着问"那住哪个酒店好？"，模型不知道前面在聊杭州，因为新的请求里根本没有历史消息。

这就是下一章要解决的事：对话历史。

---

附录

• 附录 1A：KV Cache 与 Prefill — 解释为什么输出 token 比输入 token 贵，以及 Prompt Caching 是怎么回事。