Python 调用 LLM 的基本功：消息、参数、结构与流式

一句话定位：把「调通大模型」变成「调好大模型」——本文按一次 API 调用的解剖顺序，拆消息结构、采样参数、结构化输出、思考模式、流式传输五件基本功，配完整可复现代码。前置：采样与生成参数——temperature、top_p 的机制在那篇，本文只讲工程含义。

先从十行代码的假象说起

第一次用 Python 接大模型的体验通常很美好：装个官方 SDK，十行代码跑通对话，感觉和调一个普通函数没什么两样。

真要接进业务，假象很快碎掉：同一句话两次回答不一样；想让它吐 JSON，它先客气一句「好的，以下是您要的 JSON」再吐；输出到一半用户关了页面，已生成的半截话找不回来；哪天换一家模型供应商，调用代码几乎要重写一遍。

这些坑到底出在 API 的哪个环节？

全落在五件基本功上：消息结构、采样参数、结构化输出、思考模式、流式传输。逐个拆开看，每件都配可运行的代码——这就是本文的全部内容。

全景图：一次调用的解剖

先看全貌。一次 API 调用是「请求三区 → 服务端 → 响应解析」的往返：请求侧装消息数组、参数区与输出约束，响应侧拿到内容、结束原因与用量；流式则是把响应拆成一路 chunk 小水流。五件基本功分别驻守在这条链路的不同位置。

图：一次调用的解剖——请求侧四件（①消息 ②参数 ③约束 ④思考档位），响应侧解析与流式通道（⑤）

消息结构：无状态的接力棒

最容易误解的一点：API 服务端不替你记对话历史。消息数组是接力棒——每次请求都要把 system 设定、全部历史轮次、最新输入完整重放一遍。这带来两个工程推论：账单随对话轮次加速增长（每轮都重付全部历史的钱）；上下文窗口是硬顶，超了要先做上下文管理。

消息按 role 分工：system（Responses API 里叫 developer）定行为与格式，user 带输入，assistant 存历史回答。多模态内容是同一数组的另一种内容块——图像怎么变成 token 那篇讲过原理，在 API 层它就是把 base64 图塞进 user 消息的内容块里。

把两轮对话的请求体并排放在一起，接力棒的样貌就出来了：

图：消息数组接力棒——每轮把 system 与全部历史原样重放，再追加新输入；账单随轮次加速增长，上下文窗口是硬顶

两代 API 的格局要知道（2026-09 查证）：官方已把 Responses API 定为所有新项目的推荐接口，Chat Completions 继续受支持。Responses 把工具调用的 schema 校验默认打开（严格模式）、思考档位成为一等参数、思考过程可以跨轮次保留复用。存量代码不必慌着迁，新项目值得直接用：

两代 API 的官方口径出处见文末参考段；最新模型的部分能力只在 Responses 开放（如新一代工具调用）。另有未系统压测的社区反馈称有状态用法在部分负载下更慢——延迟敏感的场景迁移前先自己压一轮。

参数：把采样旋钮接上工程含义

机制在采样篇拆过：temperature 调分布胖瘦，top_p 剪候选池，T=0 旁路两者直接贪心。工程含义浓缩成三句话：要稳用低温（抽取、分类、格式化，0 到 0.3），要发散用中高温（文案、脑暴，0.7 以上），别迷信精确复现——T=0 也有非确定性来源链，逐字节可复现的活儿不该交给采样。

max_tokens 是输出预算的护栏：截断不报错，只是 finish_reason 变成 length（正常说完整是 stop）。生产代码要检查这个字段——被截断的 JSON 后续必然解析失败，提前拦下能省掉大量事后排查。结束原因字段各家叫法不同（OpenAI 系叫 finish_reason，Anthropic 系叫 stop_reason），适配层统一掉。

结构化输出：别用正则去求模型

SCQA 先立靶：你让模型「输出 JSON」，它确实输出了——但前面带一句寒暄、后面带一句备注，正则截不出来；或者字段名自创、类型乱飘。自由发挥是模型的默认天性，约束要交给接口层而不是祈祷。

约束分三级，能力递增：

1. JSON 模式：只保证「是合法 JSON」，不保证字段。
2. 原生结构化输出：把 JSON Schema 编进解码约束（OpenAI 的严格模式，需显式传 strict），字段与类型由服务端保证。
3. Pydantic 校验：代码里再做最后一道校验，兜住语义错误（值的范围、枚举、业务规则）。

分工一句话：接口层的约束保证「形对」，Pydantic 保证「意对」；走 JSON 模式时形对也归 Pydantic 兜。

校验失败怎么反喂最高效？把校验器的报错原文回传——模型知道错在哪个字段，比重发原始请求收敛快——让模型自己修自己：

图：校验反喂重试回路——失败信息原样回喂，预算内回环重试，耗尽抛错给人

反喂有两个讲究：

• 错误原文直接给：Pydantic 的报错自带字段名、期望类型、实际值，就是最好的修正指令，不用人工润色。
• 指令要圈死范围：「只修 JSON、别加解释」，防止模型又开始寒暄。

重试预算一到就抛错——fail fast，反复失败说明 schema 设计得不对，不是重试能解决的。

思考模式：给模型一个「想多久」的档位

推理模型先产出思维链再作答，机制与「多想一会儿换正确率」的取舍在推理模型篇拆过。API 层你要会用三个东西：思考档位（OpenAI 系的 reasoning effort，按档位控制思考 token 预算）、思考内容字段（各家把思维链放在单独字段，分开展示；计费上按输出 token 计价，在用量明细里单列）、跨轮复用（Responses 接口可把上一轮思考项加密回传，续轮不用重新想）。

工程落点是按任务档位化：抽取、改写给低档（想多了白花钱），多步推理、代码、数学给高档。字段名各家不一：

适配层做映射。

流式：SSE、断点与取消

流式走的协议是 SSE（Server-Sent Events）：服务端把回复切成一条条 chunk，每条是 data: 开头的一行 JSON。客户端的活儿只有三件：逐行读、把每条 chunk 的增量拼起来、边拼边吐给前端。注意一个协议细节：标准 SSE 没有结束标记，流以服务端关闭连接收尾；OpenAI 兼容层加了一条 [DONE] 哨兵行，且它只代表「流传完了」——被 max_tokens 截断的流同样以它收尾，完不完整要看 chunk 里的结束原因。

中断与断点是同一件事的两面。用户关页面、网络断掉时，已收到的 chunk 拼接文本就是断点现场——自存自管，没有跨厂商的续传标准。恢复时按业务二选一：把半截 assistant 消息放进上下文请求「继续」，一致性敏感就整段重答。程序要记两个状态：已拼接文本 + 结束原因收到没有（finish_reason 是 stop 才算完整，length 截断要单独处理）。

主动取消靠请求中断：客户端断开连接，服务端感应后停止生成。自建转发层（FastAPI 的 StreamingResponse 配合客户端断连检测）能把上游的取消一路传导下去——别让用户走了模型还在烧钱。

OpenAI Compatible：一套协议接多家

多厂商适配的现实答案是「OpenAI Compatible」事实标准：vLLM 自建服务、Ollama 本地跑、各家云厂商的兼容模式，几乎都实现同一组 /chat/completions 端点与请求体。换供应商的成本被压到改两个配置：base_url 与 api_key。

适配层要管的剩余差异只有三样：

• 模型名映射。
• 少量参数名不同：思考档位、结束原因。
• 个别功能缺失：有的兼容端点不支持严格 schema。

做法是用一层薄适配器收敛这三样，业务代码只面对自家接口。轻量场景用各家 SDK，多厂商路由与降级场景把调用层抽出来——那层的完整设计（重试、降级、看门狗）在模型调用与容错篇已经拆过，本文不重复。

这套「一套协议接多家」的格局，一张图对齐：

图：OpenAI Compatible——薄适配层收敛三样差异，多家服务共用同一组 /chat/completions 端点，换供应商只改 base_url 与 api_key

动手：完整可复现的最小调用栈

先装依赖、设好环境变量（端点指向任何一家 OpenAI 兼容服务均可）：

pip install requests pydantic
export LLM_BASE_URL="https://你的兼容端点/v1" LLM_API_KEY="sk-..." LLM_MODEL="模型名"

下面两段代码构成一个最小但完整的调用栈：带结构化输出与校验反喂重试的请求函数，加一个手解 SSE 的流式循环：

import os, json, requests
from pydantic import BaseModel, ValidationError

BASE, KEY, MODEL = (os.environ[k] for k in ("LLM_BASE_URL", "LLM_API_KEY", "LLM_MODEL"))

class Order(BaseModel):                      # 输出契约：字段、类型、可空性
    item: str
    qty: int
    note: str | None = None

def chat(messages, retries=2):               # 结构化输出 + 校验反喂重试
    for attempt in range(retries + 1):
        r = requests.post(f"{BASE}/chat/completions", timeout=60, headers={
            "Authorization": f"Bearer {KEY}"},
            json={"model": MODEL, "messages": messages,
                  "response_format": {"type": "json_object"}})
        r.raise_for_status()                 # HTTP 层错误直接抛，不重试
        msg = r.json()["choices"][0]["message"]
        try:
            return Order.model_validate(json.loads(msg["content"]))
        except (json.JSONDecodeError, ValidationError) as e:
            if attempt == retries:
                raise                        # 预算耗尽抛给人，绝不吞
            messages = messages + [          # 重建新列表，不动调用方的原列表
                {"role": "assistant", "content": msg["content"]},
                {"role": "user", "content": f"未通过校验：{e}。只修 JSON 重出，别解释。"}]

call = [{"role": "system", "content": "以 JSON 对象输出订单：item 字符串、qty 整数、note 可空。"},   # json_object 模式要求上下文出现 JSON 字样
        {"role": "user", "content": "记一笔：买了 3 箱 A4 纸，备注用完再补"}]
print(chat(call))

buffer, finish = [], None                    # 流式：手解 SSE，chunk 累积即断点现场
with requests.post(f"{BASE}/chat/completions", stream=True, timeout=120,
                   headers={"Authorization": f"Bearer {KEY}"},
                   json={"model": MODEL, "messages": [{"role": "user", "content": "讲个冷笑话"}],
                         "stream": True}) as r:
    for line in r.iter_lines(chunk_size=1):  # 逐字节读，防转发场景缓冲延迟
        if not line.startswith(b"data: "):
            continue
        payload = line[6:]
        if payload == b"[DONE]":
            break                            # 兼容层流结束哨兵：不区分 stop/length
        chunk = json.loads(payload)
        choices = chunk.get("choices")
        if not choices:
            continue                         # 部分端点会发无 choices 的用量 chunk
        finish = choices[0].get("finish_reason") or finish
        delta = choices[0]["delta"].get("content")
        if delta:
            buffer.append(delta)             # 生产版在这里逐段推给前端并落盘断点
print("".join(buffer), "| finish_reason:", finish)   # finish 才是完整性的判据

对着代码核对五件基本功：

1. 消息数组全量重放（messages 一路带着历史长）。
2. 参数区本文从简、采样篇已拆。
3. json_object 声明 JSON 模式 + Pydantic 严校验——示例选 JSON 模式是取兼容面最广的一级（严格 schema 模式是官方首选，部分兼容端点不支持），形错意错都由校验兜住。
4. 思考档位在请求体加 reasoning_effort 字段即可（结构不变，OpenAI 系参数名）。
5. SSE 循环里 buffer 加 finish 就是完整断点现场，落盘即断点保存。

把环境变量指向任何一家兼容端点（云厂商兼容模式、vLLM、Ollama），这套代码原样能跑——这就是事实标准的意义。

边界与取舍

什么时候别用结构化输出

开放式任务（写文案、自由问答）硬套 schema 是自缚手脚——约束只在「输出要进程序」时才值得付（解析失败重试的延迟与成本）。先问一句：这份输出是人读的还是机器读的？人读就放开。

两代 API 怎么选

新项目、要用最新模型、要多轮思考复用，直接 Responses；存量 Chat Completions 不必恐慌性迁移（官方继续支持），但延迟敏感场景迁移前压测有状态用法的开销。中间态做一个薄适配层，把「未来切换」的成本锁在一处。

思考模式不是默认开

档位是延迟与正确率的汇率：低档任务开高档思考，钱花在没人看的地方。给业务分类定档位表，别让调用方随手选。

自建转发层才考虑断点保存

直接调官方 API 的场景，中断重问通常够用；只有你在做面向终端用户的长回答产品，断点续答才值得做——记住自存自管，没有标准可依赖。

自测

合上文章先自己答，再展开对照：

参考与延伸

• OpenAI 官方迁移指南：Responses API 成为新项目推荐、与 Chat Completions 的差异清单（2026-09 查证） — developers.openai.com/api/docs/guides/migrate-to-responses
• OpenAI 结构化输出指南：JSON Schema 约束与严格模式 — developers.openai.com/api/docs/guides/structured-outputs
• OpenAI 推理模型指南：思考档位与思考项跨轮复用 — developers.openai.com/api/docs/guides/reasoning
• MDN「Server-sent events」：SSE 协议的权威参考 — developer.mozilla.org/en-US/docs/Web/API/Server-sent_events
• FastAPI 响应参考：StreamingResponse 与转发流式响应 — fastapi.tiangolo.com/reference/responses/

下一站：微调（LoRA 等）实操——模型出厂的通用能力不贴身时，怎么用最小的代价把它掰弯。