01 大模型接入与OpenAI库调用：APIKEY保护、流式输出与多轮对话

一、大模型接入概述

1.1 为什么要用 OpenAI 库接入

大模型部署在云端或本地的服务器上，程序要调用它，本质上是发一个 HTTP 请求过去。如果手写 HTTP 请求，需要自己拼装 JSON 报文、处理鉴权头、再解析流式响应，成本很高。

OpenAI 官方提供的 Python SDK 把这一层全部封装好了，对外只暴露一个 OpenAI 类。关键在于，这套 SDK 并不绑定 OpenAI 自家的模型——只要某个服务实现了”兼容 OpenAI 的接口”，把 base_url 指过去就能用同一份代码调用。阿里云百炼的通义千问、本地部署的 Ollama 都提供了这样的兼容接口，所以云端和本地可以共用一套写法。

后续学习 LangChain 时，它同样是在这一层调用能力之上再封装，所以这里的调用形态是后面所有内容的基础。

1.2 云端与本地两种部署方式

方式	说明	特点
云端（阿里云百炼）	模型由厂商部署，通过网络调用	模型能力强，需要 APIKEY，按量计费
本地（Ollama）	模型下载到本机，由 Ollama 提供兼容接口	免费、数据不出本机，模型规模小、能力有限

Ollama 是本地大模型的运行框架，启动后会监听 11434 端口，并额外暴露一个 /v1 路径用于兼容 OpenAI SDK。课程资料中打包了预下载的 models 文件夹，整个放进 C:\Users\<用户名>\.ollama 覆盖，就能免去下载过程直接使用演示模型。

1.3 调用流程

无论访问云端还是本地，调用都是三步，差别只在参数：

① 创建客户端 OpenAI(base_url=...)      # 决定请求发给谁
        ↓
② 调用 client.chat.completions.create(  # 决定用哪个模型、聊什么
       model=..., messages=[...], stream=...)
        ↓
③ 消费返回结果                         # 非流式取 message.content
                                       # 流式迭代 chunk.choices[0].delta.content

二、OpenAI 库的基础使用

2.1 客户端对象与 base_url

OpenAI 是 SDK 的入口类，构造时主要给两个参数：

参数	作用	说明
`base_url`	请求发往哪个服务	默认是 OpenAI 官方地址，接第三方必须修改
`api_key`	身份凭证	云端服务需要；本地 Ollama 不校验，可以不传

from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:11434/v1"
)

要点：接口形态一致，只有地址不同。更换模型不需要改动代码结构，只改这两个参数。这正是”兼容模式”的意义所在。

2.2 接入地址对比

场景	base_url	model 示例	api_key
阿里云百炼（自定义开通的服务）	`https://ws-u7840hnq50vkil5d.cn-beijing.maas.aliyuncs.com/compatible-mode/v1`	`qwen3.8-flash`	需要
阿里云百炼（官方通义兼容地址）	`https://dashscope.aliyuncs.com/compatible-mode/v1`	`qwen3-max`	需要
本地 Ollama	`http://localhost:11434/v1`	`redule26/huihui_ai_qwen2.5-vl-7b-abliterated`	不需要

两者的 base_url 都以 /compatible-mode/v1 结尾，本地则以 /v1 结尾。Ollama 的这个 /v1 后缀是它为兼容 OpenAI SDK 额外暴露的路径，写漏了会直接 404。

2.3 messages 消息结构

messages 是一个字典列表，每个字典必须包含 role 和 content 两个键。role 决定这条消息由谁发出：

role	含义	常见用法
`system`	系统设定，描述模型的身份和语气	放在最前面，对整个会话全程生效
`user`	用户输入	每轮对话追加一条
`assistant`	模型此前的回复	手动构造历史时才需要

system 的作用非常直接，同一句提问，只改 system 得到的结果完全不同：

messages=[
    {"role":"system","content":"你是一个编程助手,简洁不废话"},
    {"role":"user","content":"python的for循环咋写"}
]

除了 system 和用户提问，还可以在 messages 中间插入一条 assistant，例如”好的,有啥想问的,我会简洁回答,绝不废话”。这条不是真实的历史记录，而是预置一次”模型已经这样答过”的上下文，用来强化 system 设定的约束。这实际上已经是提示词工程中”少样本”思路的雏形。

2.4 非流式返回结果

不加 stream 参数时，返回的是一个完整的响应对象，答案位于 choices[0].message.content：

response = client.chat.completions.create(
    model="qwen3-max",
    messages=messages
)

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

三、APIKEY 的保护

3.1 为什么不能硬编码

把 sk-xxx 直接写进代码，一旦代码提交到仓库，就等于把 key 公开了。别人拿到后可以用它调用模型，产生的费用记在账号名下。所以 APIKEY 必须与代码分离。

3.2 三种写法

# ① 硬编码：能用，但绝对不能提交仓库
client = OpenAI(base_url="...", api_key="sk-xxx")

# ② 环境变量：推荐做法
import os
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url="https://ws-u7840hnq50vkil5d.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
)

# ③ 本地 Ollama：不需要 key
client = OpenAI(base_url="http://localhost:11434/v1")

把三种方式并列写成注释留在客户端构造处，切换云端和本地时只改注释开关，改动最小：

client = OpenAI(
    # 如果没有配置环境变量，请用阿里云百炼API Key替换：api_key="sk-xxx"
    # api_key=os.getenv("DASHSCOPE_API_KEY"),
    # base_url="https://ws-u7840hnq50vkil5d.cn-beijing.maas.aliyuncs.com/compatible-mode/v1",
    base_url="http://localhost:11434/v1",
)

3.3 环境变量配置

只对当前终端会话生效：

1	$env:DASHSCOPE_API_KEY="sk-xxx"

写入用户环境变量永久生效，需要重开终端：

1	[Environment]::SetEnvironmentVariable("DASHSCOPE_API_KEY","sk-xxx","User")

Python 中读取回来验证：

1 2	import os print(os.getenv("DASHSCOPE_API_KEY"))

3.4 注意事项

环境变量读不到值时返回 None，不会报错。错误要等到客户端初始化时才抛出，报错信息不容易看懂，所以先 print 一次确认能取到值。
.env 文件配合 python-dotenv 是另一种常见做法，但必须把 .env 写进 .gitignore，否则等于没做保护。
已经提交进仓库的 key，应当立刻去控制台吊销重建，仅仅”后来把文件删了”没有意义——历史提交里仍然存在。

四、流式输出模式

4.1 为什么需要流式输出

非流式调用要等模型把整段话全部生成完，才一次性返回结果。大模型逐字生成，一段长回答可能要十几秒，用户端表现为长时间空白然后突然刷出一整段。

流式输出（stream=True）把生成过程拆开，生成一段就返回一段，程序边接收边打印，用户端立刻能看到文字逐个出现。模型侧的总耗时并没有减少，但体感快得多。

4.2 基本用法

只需要在 create() 里加上 stream=True，然后把取值方式从”读完整消息”改成”遍历碎片”：

response = client.chat.completions.create(
    model='redule26/huihui_ai_qwen2.5-vl-7b-abliterated',
    messages=[
        {"role":"system","content":"你是一个编程助手,简洁不废话"},
        {"role":"user","content":"python的for循环咋写"}
    ],
    stream=True
)

for chunk in response:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content,
              end='',    # 末尾不换行，片段直接拼接
              flush=True # 立即刷新缓冲区
          )

关键点：

stream=True 是 create() 的一个参数，不是单独的方法调用。
打印时必须用 end=''，否则每个片段都会自动换行，输出变成一列。
flush=True 让输出立刻刷到屏幕上。不写的话内容会先攒在缓冲区里，等攒够才一次性显示，效果等同于没开流式。

4.3 两种取值路径

流式和非流式的取值路径完全不同，同一个 response 对象不能按两种方式消费：

模式	参数	取值方式
非流式	默认	`response.choices[0].message.content`
流式	`stream=True`	遍历 `chunk.choices[0].delta.content`

一个是 message（完整消息），一个是 delta（增量片段），写混了会报错。不用的那种写法注释掉留在原地，方便来回对照：

1 2	# 处理非流式返回结果 # print(response.choices[0].message.content)

4.4 深度思考模型：reasoning_content

部分模型（qwen3 系列等）支持深度思考，会先输出一段思维链再给最终答案。两部分不在同一个字段里：

思维链在 delta.reasoning_content
正式答案在 delta.content

在请求中通过 extra_body 打开这一能力：

completion = client.chat.completions.create(
    model="redule26/huihui_ai_qwen2.5-vl-7b-abliterated",
    messages=messages,
    extra_body={"enable_thinking": True},
    stream=True
)

is_answering = False  # 是否进入回复阶段
print("\n" + "=" * 20 + "思考过程" + "=" * 20)

for chunk in completion:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    # 思维链
    if hasattr(delta, "reasoning_content") and delta.reasoning_content is not None:
        if not is_answering:
            print(delta.reasoning_content, end="", flush=True)
    # 正文
    if hasattr(delta, "content") and delta.content:
        if not is_answering:
            print("\n" + "=" * 20 + "完整回复" + "=" * 20)
            is_answering = True
        print(delta.content, end="", flush=True)

处理流程：

1
2
3

逐 chunk 遍历
  ├─ delta.reasoning_content 非空 → 打印思考过程（不切段）
  └─ delta.content 非空        → 先换行打分隔标题，再用 is_answering 锁定，后续只打正文

is_answering 标志位的作用是给输出分段。思维链是分成很多个碎片流式返回的，如果不加这个标志，每个碎片都会重复打印一次”完整回复”的标题。

4.5 易错点

必须先判断 chunk.choices 是否为空。流式过程中会出现 choices 为空的心跳包，不加 if not chunk.choices: continue 会抛 IndexError。
用 hasattr 判断字段是否存在，而不是直接访问 delta.reasoning_content。不支持深度思考的模型没有这个属性，直接访问会抛 AttributeError。
extra_body 是 SDK 的逃生通道，用来传各家自定义的参数。放进这个字典意味着该参数不是 OpenAI 标准参数，换一个不支持的模型可能直接被拒。

五、附带历史消息实现多轮对话

5.1 模型为什么会”失忆”

每次调用都是一次独立的 HTTP 请求，服务端不会为某次对话保留状态，模型本身不知道上一轮说了什么。所谓”多轮对话”，本质是每请求一次，就把之前的全部对话重新发送一遍。

这也意味着对话历史需要由调用方自己维护。

5.2 messages 累加实现

把历史放进 messages，每轮追加而不是重建：

from openai import OpenAI

client = OpenAI(
    base_url='https://ws-u7840hnq50vkil5d.cn-beijing.maas.aliyuncs.com/compatible-mode/v1'
)

# 预设的历史：system 定规则，assistant 给一个示范回答
messages = [
    {"role": "system", "content": "你是一个编程助手,简洁不废话"},
    {"role": "assistant", "content": "好的,有啥想问的,我会简洁回答,绝不废话"},
]

while True:
    userin = input('你:')
    messages.append({"role": "user", "content": userin})

    print('ai:')
    response = client.chat.completions.create(
        model="qwen3.8-flash",
        messages=messages,
        stream=True,
        extra_body={"enable_thinking": True},
    )

    ai = []
    for chunk in response:
	    if(not chunk.choices):continue
        if chunk.choices and chunk.choices[0].delta.content:
            ai.append(chunk.choices[0].delta.content)
            print(chunk.choices[0].delta.content, end='', flush=True)
    print()

    messages.append({"role": "assistant", "content": ''.join(ai)})
    print('\n' + str(messages) + '\n')