一次 LLM API 调用：messages 与流式输出

这篇笔记解决什么问题

不管你最终想做的是聊天机器人、代码助手，还是一个能自主干活的 Agent，最底层永远是同一件事：向大模型发起一次 HTTP 调用，拿回一段生成结果。这一步看起来平平无奇，却藏着两个决定你后面一切设计的关键认知——模型是无状态的，以及所有上下文都装在 messages 数组里。

这篇笔记会把一次 chat completions 调用彻底拆开：核心参数是什么、三种角色各管什么、为什么换一家厂商代码几乎不用改、流式输出底层发生了什么，最后把一次性的脚本升级成一个记得住上下文的多轮对话 CLI。读完你应该能独立写出这个 CLI，并且说清楚它为什么"有记忆"。

一次最小调用长什么样

主流大模型厂商（DeepSeek、Kimi、GLM、通义等）都提供 OpenAI 兼容格式的 HTTP 接口，所以我们可以统一用 OpenAI 的官方 Python SDK 来演示，把接口地址做成可配置项，不绑定任何一家：

import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL"),  # 例如 https://api.deepseek.com/v1
)

completion = client.chat.completions.create(
    model=os.getenv("LLM_MODEL"),  # 例如 deepseek-chat / moonshot-v1-8k / glm-4
    messages=[
        {"role": "system", "content": "你是一名耐心的编程学习助教，回答尽量简短。"},
        {"role": "user", "content": "用一句话解释什么是幂等性。"},
    ],
)

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

这次调用里出场的三个核心参数，几乎每一次 LLM 调用都绕不开：

• model：告诉服务端这次用哪个模型。同一个接口地址下通常挂着多个模型，能力和价格不同。
• messages：本次调用的全部上下文。注意是"全部"——模型能看到的信息，一条不多、一条不少，就是这个数组里的内容。
• stream：要不要流式返回。默认关闭，即等整段回复生成完一次性给你；打开后边生成边推送。

顺带说一下返回值里那个 choices[0]：接口设计上允许一次请求生成多个候选回答（通过 n 参数），返回的是一个候选列表。日常开发几乎都只要一个答案，所以取第 0 个就行——但知道它是个列表，读文档时就不会困惑。

messages：模型的整个世界

这是本篇最重要的一节。

大模型服务端不给你开"会话"，也不替你存任何历史。每次调用，模型看到的世界就是你传过去的那个 messages 数组——它不知道三分钟前你们聊过什么，也不知道你是谁。想让它"知道"，唯一的办法就是把信息写进数组里传回去。

数组里的每一项都是 {"role": ..., "content": ...} 结构，role 只有三种取值：

• system：给模型定规矩的地方。人设、语气、格式要求、禁止事项，都写在这里，一般放在数组第一位，对整段对话生效。后面讲工具调用时，工具的使用守则往往也塞在这。
• user：使用者这一侧的输入。当前的问题、指令、补充材料。
• assistant：模型之前生成过的回答。注意，这个角色的消息不是模型自动记住的，而是由你的代码手动放回数组的。

举个例子，用户第一轮问了个模型答不了的问题，第二轮换了话题，那么第二轮调用时传过去的数组是这样的：

[
  {"role": "system", "content": "你是一名耐心的编程学习助教。"},
  {"role": "user", "content": "现在美元兑人民币的汇率是多少？"},
  {"role": "assistant", "content": "我无法获取实时汇率数据，建议查询银行或财经网站。"},
  {"role": "user", "content": "那给我讲讲浮点数为什么不精确吧。"}
]

看清这个结构，你就理解了 LLM 应用的一条基本定律：

所谓"对话历史"，不是模型的能力，而是客户端的数据结构。 模型无状态，记忆全靠调用方每轮把完整历史重新传一遍。

由此还能推出几个后面会反复用到的结论：

1. 对话越长，每次请求的 token 越多、越贵、越慢——所以才需要上下文压缩、摘要等手段（本系列后面会讲）。
2. 你可以随意"伪造"历史：往数组里手工塞几条 assistant 消息，模型会真诚地以为那是它说过的话。few-shot 示例就是这么做的。
3. Agent 的"工作记忆"本质上也是这个数组，工具调用的结果同样以消息的形式追加进去。

为什么换厂商不用换代码

你可能注意到了：我们连的是国产模型，用的却是 OpenAI 的 SDK。这不是黑魔法，而是行业格局使然。

OpenAI 的 /chat/completions 接口格式公开得早、用的人多，事实上成了行业通用协议。后来的厂商为了降低开发者的迁移成本，纷纷提供兼容这一格式的接口：请求路径、参数名、返回结构全都对齐。SDK 本质上只是个封装好的 HTTP 客户端，它不关心对面是谁——只要你把 base_url 指向兼容接口，api_key 换成对应厂商的密钥，一行业务代码都不用动。

这也是为什么建议把 base_url、api_key、model 三件套全部放进环境变量：换供应商、做多模型对比、生产环境切备用渠道，都只是改配置的事。

当然，兼容是有边界的。基础对话、流式输出、常规 function calling 这些"公约数"能力没问题；但各家的独门功能——比如某些厂商的深度思考模式、原生多模态、上下文缓存——往往要用厂商自己的 SDK 或额外参数才能解锁。写通用代码时守住公约数，需要特殊能力时再单独适配。

流式输出：把等待拆成碎片

上面的脚本有个体验问题：请求发出后，程序会静止不动，直到模型把整段话写完才一次性打印。回答一长，用户就要对着空屏幕干等十几秒。

流式输出解决的就是这个问题。原理上，普通响应是服务器攒齐全部内容再返回；流式响应则借助 HTTP 分块传输（chunked transfer），服务器每生成一小段就立刻推一段，连接保持打开，直到全部内容发完。你在各家官方聊天页面看到的"打字机效果"，底层就是它。

在 SDK 里开启只需要一个参数，但读取方式随之改变：

stream = client.chat.completions.create(
    model=os.getenv("LLM_MODEL"),
    messages=[
        {"role": "system", "content": "你是一名耐心的编程学习助教。"},
        {"role": "user", "content": "分三点说明为什么要写单元测试。"},
    ],
    stream=True,
)

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

和非流式版本对比，有四个容易踩坑的细节：

1. 返回值变成了迭代器。不再是一个完整的 response 对象，而是要用 for 循环逐个消费的流，每次迭代吐出一个 chunk。
2. 字段从 message 变成 delta。delta 的语义是"相比上一块新增的内容"，通常只有几个字符，需要你自己拼接成完整回答。
3. delta.content 可能为 None，必须判空。流的首尾往往有不携带正文的 chunk（比如只带角色信息的开头块、标记结束的收尾块），不判空直接拼接会报错。
4. 打印要加 end="" 和 flush=True。前者防止每个碎片各占一行，后者强制立刻刷新输出缓冲区——少了 flush，终端可能攒一大段才显示，流式就白开了。

一句话总结两种模式的取舍：非流式代码简单、适合后台批处理；流式体验好、适合一切有人盯着屏幕的场景。

升级：一个记得住上下文的对话 CLI

把前面的认知组装起来，就能写出一个真正可用的多轮对话程序。核心机制只有一句话：每一轮结束后，把用户的提问和模型的完整回答都追加进历史数组，下一轮原样传回去。

# repl.py — 一个带记忆的流式对话 CLI
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("LLM_API_KEY"),
    base_url=os.getenv("LLM_BASE_URL"),
)
MODEL = os.getenv("LLM_MODEL", "deepseek-chat")


def stream_reply(history: list[dict]) -> str:
    """发起一次流式调用：边打印边收集，最后返回完整回答。"""
    stream = client.chat.completions.create(
        model=MODEL,
        messages=history,
        stream=True,
    )
    parts = []
    for event in stream:
        piece = event.choices[0].delta.content
        if piece:
            print(piece, end="", flush=True)
            parts.append(piece)
    print()
    return "".join(parts)


def main() -> None:
    history = [
        {"role": "system", "content": "你是一名耐心的编程学习助教，回答尽量简短。"},
    ]
    print("对话已就绪，输入 /bye 退出。")
    while True:
        try:
            question = input("\n> ").strip()
        except (EOFError, KeyboardInterrupt):
            break
        if question == "/bye":
            break
        if not question:
            continue

        history.append({"role": "user", "content": question})
        answer = stream_reply(history)
        history.append({"role": "assistant", "content": answer})


if __name__ == "__main__":
    main()

跑起来之后可以做个自证实验：先告诉它"我正在学 Rust"，下一轮问"我在学什么语言"，它答得出来。然后把 history.append({"role": "assistant", ...}) 那一行注释掉再试——它立刻"失忆"。这个小实验比任何文字都更能说明：记忆不在模型那边，在你的数组里。

另外留意 stream_reply 里 parts 列表的作用：流式模式下回答是碎片化到达的，屏幕上打印归打印，你还必须自己把碎片拼成完整字符串存回历史。漏掉这一步，下一轮模型看到的历史就是残缺的。

实践要点与常见坑

• 401 认证失败：十有八九是 key 的问题——环境变量没设置、复制时带上了引号或首尾空格、key 已过期，或者 key 和 base_url 不是同一家的（拿 A 厂的 key 打 B 厂的接口）。
• Connection error / 超时：先检查 base_url 拼写。不少厂商的兼容接口要求带版本路径（如结尾的 /v1），少一段就连不上；其次排查本地代理和网络。
• 长时间无输出：如果是非流式调用，长回答本来就要等；切到流式能立刻看出模型到底是在生成还是卡死了。这也是调试时优先用流式的一个理由。
• 回答越来越慢、账单越来越贵：历史数组只增不减的自然结果。演示程序可以不管，生产环境必须设计历史截断或摘要策略。
• 忘记回填 assistant 消息：症状是模型每轮都像第一次见你。检查你的循环里是不是只 append 了 user 那一条。

动手清单

1. 把上面的 CLI 跑通，然后换一家 OpenAI 兼容厂商的 base_url 和 key 再跑一遍，体会"改配置不改代码"。
2. 在流式循环里临时加一行 print(event)，观察原始 chunk 的结构，找出哪些块的 delta.content 是空的、最后一块长什么样。
3. 做"失忆实验"：注释掉回填 assistant 消息的那行，验证多轮记忆立刻失效，再恢复。
4. 给 CLI 增加一个 /reset 命令，把 history 重置为只剩 system 消息，实现"清空对话"。
5. 把 system 消息改成一个极端人设（比如"只能用反问句回答"），感受 system 角色对整段对话的约束力。

*本文为学习笔记，知识框架参考自叶小钗「生产级 Agent」系列课程；文字与代码示例为本站原创整理。*